fix: connection recovery

package.json

@@ -72,7 +72,7 @@
     "rollup-plugin-esbuild": "4.9.3",
     "sinon": "14.0.0",
     "typescript": "4.7.4",
-    "vitest": "0.21.1",
+    "vitest": "^0.22.1",
     "lerna": "5.4.2"
   }
 }

packages/core/package.json

@@ -31,7 +31,7 @@
   },
   "dependencies": {
     "@walletconnect/heartbeat": "1.2.0",
-    "@walletconnect/jsonrpc-provider": "1.0.9",
+    "@walletconnect/jsonrpc-provider": "1.0.10",
     "@walletconnect/jsonrpc-utils": "^1.0.4",
     "@walletconnect/jsonrpc-ws-connection": "1.0.10",
     "@walletconnect/keyvaluestorage": "^1.0.2",

packages/core/src/controllers/publisher.ts

@@ -11,6 +11,7 @@ import {
   createExpiringPromise,
 } from "@walletconnect/utils";
 import { EventEmitter } from "events";
+
 import { PUBLISHER_CONTEXT, PUBLISHER_DEFAULT_TTL, RELAYER_EVENTS } from "../constants";
 
 export class Publisher extends IPublisher {
@@ -104,7 +105,7 @@ export class Publisher extends IPublisher {
     if (isUndefined(request.params?.tag)) delete request.params?.tag;
     this.logger.debug(`Outgoing Relay Payload`);
     this.logger.trace({ type: "message", direction: "outgoing", request });
-    return this.relayer.provider.request(request);
+    return this.relayer.request(request);
   }
 
   private onPublish(hash: string, _params: PublisherTypes.Params) {

packages/core/src/controllers/relayer.ts

@@ -7,6 +7,7 @@ import {
   isJsonRpcRequest,
   JsonRpcPayload,
   JsonRpcRequest,
+  RequestArguments,
 } from "@walletconnect/jsonrpc-utils";
 import WsConnection from "@walletconnect/jsonrpc-ws-connection";
 import {
@@ -27,7 +28,7 @@ import {
   RelayerTypes,
   SubscriberTypes,
 } from "@walletconnect/types";
-import { formatRelayRpcUrl, getInternalError } from "@walletconnect/utils";
+import { createExpiringPromise, formatRelayRpcUrl, getInternalError } from "@walletconnect/utils";
 
 import {
   RELAYER_SDK_VERSION,
@@ -43,7 +44,6 @@ import {
 import { MessageTracker } from "./messages";
 import { Publisher } from "./publisher";
 import { Subscriber } from "./subscriber";
-
 export class Relayer extends IRelayer {
   public protocol = "wc";
   public version = 2;
@@ -62,6 +62,7 @@ export class Relayer extends IRelayer {
   private reconnecting = false;
   private relayUrl: string;
   private projectId: string | undefined;
+  private connectionStatusPollingInterval = 20;
   constructor(opts: RelayerOptions) {
     super(opts);
     this.core = opts.core;
@@ -82,7 +83,7 @@ export class Relayer extends IRelayer {
 
   public async init() {
     this.logger.trace(`Initialized`);
-    this.provider = await this.createProvider();
+    await this.createProvider();
     await Promise.all([this.messages.init(), this.transportOpen(), this.subscriber.init()]);
     this.registerEventListeners();
     this.initialized = true;
@@ -130,6 +131,18 @@ export class Relayer extends IRelayer {
     return id;
   }
 
+  public request = async (request: RequestArguments<RelayJsonRpc.SubscribeParams>) => {
+    this.logger.debug(`Publishing Request Payload`);
+    try {
+      await this.toEstablishConnection();
+      return await this.provider.request(request);
+    } catch (e) {
+      this.logger.debug(`Failed to Publish Request`);
+      this.logger.error(e as any);
+      throw e;
+    }
+  };
+
   public async unsubscribe(topic: string, opts?: RelayerTypes.UnsubscribeOptions) {
     this.isInitialized();
     await this.subscriber.unsubscribe(topic, opts);
@@ -168,15 +181,14 @@ export class Relayer extends IRelayer {
       await Promise.all([
         new Promise<void>((resolve) => {
           if (!this.initialized) resolve();
-
           // wait for the subscriber to finish resubscribing to its topics
           this.subscriber.once(SUBSCRIBER_EVENTS.resubscribed, () => {
             resolve();
           });
         }),
         await Promise.race([
           new Promise<void>(async (resolve) => {
-            await this.provider.connect();
+            await createExpiringPromise(this.provider.connect(), 5_000, "socket hang up");
             this.removeListener(RELAYER_EVENTS.transport_closed, this.rejectTransportOpen);
             resolve();
           }),
@@ -188,25 +200,23 @@ export class Relayer extends IRelayer {
         ]),
       ]);
     } catch (e: unknown | Error) {
+      this.logger.error(e);
       const error = e as Error;
       if (!/socket hang up/i.test(error.message)) {
         throw e;
       }
-      this.logger.error(e);
       this.events.emit(RELAYER_EVENTS.transport_closed);
     } finally {
       this.reconnecting = false;
     }
   }
 
   public async restartTransport(relayUrl?: string) {
-    if (this.transportExplicitlyClosed) {
-      return;
-    }
-
+    if (this.transportExplicitlyClosed) return;
+    this.relayUrl = relayUrl || this.relayUrl;
     await this.transportClose();
-    await new Promise<void>((resolve) => setTimeout(resolve, RELAYER_RECONNECT_TIMEOUT));
-    await this.transportOpen(relayUrl);
+    await this.createProvider();
+    await this.transportOpen();
   }
 
   // ---------- Private ----------------------------------------------- //
@@ -217,7 +227,7 @@ export class Relayer extends IRelayer {
 
   private async createProvider() {
     const auth = await this.core.crypto.signJWT(this.relayUrl);
-    return new JsonRpcProvider(
+    this.provider = new JsonRpcProvider(
       new WsConnection(
         formatRelayRpcUrl({
           sdkVersion: RELAYER_SDK_VERSION,
@@ -230,6 +240,7 @@ export class Relayer extends IRelayer {
         }),
       ),
     );
+    this.registerProviderListeners();
   }
 
   private async recordMessageEvent(messageEvent: RelayerTypes.MessageEvent) {
@@ -271,36 +282,41 @@ export class Relayer extends IRelayer {
     await this.provider.connection.send(response);
   }
 
-  private registerEventListeners() {
+  private registerProviderListeners() {
     this.provider.on(RELAYER_PROVIDER_EVENTS.payload, (payload: JsonRpcPayload) =>
       this.onProviderPayload(payload),
     );
     this.provider.on(RELAYER_PROVIDER_EVENTS.connect, () => {
       this.events.emit(RELAYER_EVENTS.connect);
     });
     this.provider.on(RELAYER_PROVIDER_EVENTS.disconnect, () => {
-      this.events.emit(RELAYER_EVENTS.disconnect);
-
-      this.attemptToReconnect();
+      this.onProviderDisconnect();
     });
     this.provider.on(RELAYER_PROVIDER_EVENTS.error, (err: unknown) => {
       this.logger.error(err);
       this.events.emit(RELAYER_EVENTS.error, err);
     });
+  }
 
+  private registerEventListeners() {
     this.events.on(RELAYER_EVENTS.connection_stalled, async () => {
       await this.restartTransport();
     });
   }
 
+  private onProviderDisconnect() {
+    this.events.emit(RELAYER_EVENTS.disconnect);
+    this.attemptToReconnect();
+  }
+
   private attemptToReconnect() {
     if (this.transportExplicitlyClosed) {
       return;
     }
 
     // Attempt reconnection after one second.
     setTimeout(async () => {
-      await this.transportOpen();
+      await this.restartTransport();
     }, toMiliseconds(RELAYER_RECONNECT_TIMEOUT));
   }
 
@@ -310,4 +326,19 @@ export class Relayer extends IRelayer {
       throw new Error(message);
     }
   }
+
+  private async toEstablishConnection() {
+    if (this.connected) return;
+    if (this.connecting) {
+      return await new Promise<void>((resolve) => {
+        const interval = setInterval(() => {
+          if (this.connected) {
+            clearInterval(interval);
+            resolve();
+          }
+        }, this.connectionStatusPollingInterval);
+      });
+    }
+    await this.restartTransport();
+  }
 }

packages/core/src/controllers/subscriber.ts

@@ -216,12 +216,12 @@ export class Subscriber extends ISubscriber {
     this.logger.trace({ type: "payload", direction: "outgoing", request });
     try {
       const subscribe = await createExpiringPromise(
-        this.relayer.provider.request(request),
+        this.relayer.request(request),
         this.subscribeTimeout,
       );
       await subscribe;
     } catch (err) {
-      this.logger.debug(`Outgoing Relay Payload stalled`);
+      this.logger.debug(`Outgoing Relay Subscribe Payload stalled`);
       this.relayer.events.emit(RELAYER_EVENTS.connection_stalled);
     }
     return hashMessage(topic + this.clientId);
@@ -241,7 +241,7 @@ export class Subscriber extends ISubscriber {
     this.logger.trace({ type: "payload", direction: "outgoing", request });
     try {
       const subscribe = await createExpiringPromise(
-        this.relayer.provider.request(request),
+        this.relayer.request(request),
         this.subscribeTimeout,
       );
       return await subscribe;
@@ -262,7 +262,7 @@ export class Subscriber extends ISubscriber {
     };
     this.logger.debug(`Outgoing Relay Payload`);
     this.logger.trace({ type: "payload", direction: "outgoing", request });
-    return this.relayer.provider.request(request);
+    return this.relayer.request(request);
   }
 
   private onSubscribe(id: string, params: SubscriberTypes.Params) {
@@ -384,6 +384,7 @@ export class Subscriber extends ISubscriber {
   private async batchSubscribe(subscriptions: SubscriberTypes.Params[]) {
     if (!subscriptions.length) return;
     const result = (await this.rpcBatchSubscribe(subscriptions)) as string[];
+    if (!result) return;
     this.onBatchSubscribe(result.map((id, i) => ({ ...subscriptions[i], id })));
   }