fix: subscribe timeout rejection

packages/core/src/controllers/relayer.ts

@@ -167,6 +167,12 @@ export class Relayer extends IRelayer {
     if (opts?.transportType === "relay") {
       await this.toEstablishConnection();
     }
+    // throw unless explicitly set to false
+    const shouldThrowOnFailure =
+      typeof opts?.internal?.throwOnFailedPublish === "undefined"
+        ? true
+        : opts?.internal?.throwOnFailedPublish;
+
     let id = this.subscriber.topicMap.get(topic)?.[0] || "";
     let resolvePromise: () => void;
     const onSubCreated = (subscription: SubscriberTypes.Active) => {
@@ -181,8 +187,19 @@ export class Relayer extends IRelayer {
         resolvePromise = resolve;
         this.subscriber.on(SUBSCRIBER_EVENTS.created, onSubCreated);
       }),
-      new Promise<void>(async (resolve) => {
-        const result = await this.subscriber.subscribe(topic, opts);
+      new Promise<void>(async (resolve, reject) => {
+        const result = await this.subscriber
+          .subscribe(topic, {
+            internal: {
+              throwOnFailedPublish: shouldThrowOnFailure,
+            },
+            ...opts,
+          })
+          .catch((error) => {
+            if (shouldThrowOnFailure) {
+              reject(error);
+            }
+          });
         id = result || id;
         resolve();
       }),

packages/core/src/controllers/subscriber.ts

@@ -101,7 +101,7 @@
       const relay = getRelayProtocolName(opts);
       const params = { topic, relay, transportType: opts?.transportType };
       this.pending.set(topic, params);
-      const id = await this.rpcSubscribe(topic, relay, opts?.transportType);
+      const id = await this.rpcSubscribe(topic, relay, opts);
       if (typeof id === "string") {
         this.onSubscribe(id, params);
         this.logger.debug(`Successfully Subscribed Topic`);
@@ -222,9 +222,9 @@
   private async rpcSubscribe(
     topic: string,
     relay: RelayerTypes.ProtocolOptions,
-    transportType: RelayerTypes.TransportType = TRANSPORT_TYPES.relay,
+    opts?: RelayerTypes.SubscribeOptions,
   ) {
-    if (transportType === TRANSPORT_TYPES.relay) {
+    if (opts?.transportType === TRANSPORT_TYPES.relay) {
       await this.restartToComplete();
     }
     const api = getRelayProtocolApi(relay.protocol);
@@ -239,26 +239,28 @@
     try {
       const subId = hashMessage(topic + this.clientId);
       // in link mode, allow the app to update its network state (i.e. active airplane mode) with small delay before attempting to subscribe
-      if (transportType === TRANSPORT_TYPES.link_mode) {
+      if (opts?.transportType === TRANSPORT_TYPES.link_mode) {
         setTimeout(() => {
           if (this.relayer.connected || this.relayer.connecting) {
             this.relayer.request(request).catch((e) => this.logger.warn(e));
           }
         }, toMiliseconds(ONE_SECOND));
         return subId;
       }
-
-      const subscribe = await createExpiringPromise(
+      const subscribe = createExpiringPromise(
         this.relayer.request(request).catch((e) => this.logger.warn(e)),
         this.subscribeTimeout,
+        `Subscribing to ${topic} failed, please try again`,
       );
       const result = await subscribe;
-
       // return null to indicate that the subscription failed
       return result ? subId : null;
     } catch (err) {
       this.logger.debug(`Outgoing Relay Subscribe Payload stalled`);
       this.relayer.events.emit(RELAYER_EVENTS.connection_stalled);
+      if (opts?.internal?.throwOnFailedPublish) {
+        throw err;
+      }
     }
     return null;
   }
@@ -266,7 +268,7 @@
   private async rpcBatchSubscribe(subscriptions: SubscriberTypes.Params[]) {
     if (!subscriptions.length) return;
     const relay = subscriptions[0].relay;
     const api = getRelayProtocolApi(relay!.protocol);
     const request: RequestArguments<RelayJsonRpc.BatchSubscribeParams> = {
       method: api.batchSubscribe,
       params: {
@@ -289,7 +291,7 @@
   private async rpcBatchFetchMessages(subscriptions: SubscriberTypes.Params[]) {
     if (!subscriptions.length) return;
     const relay = subscriptions[0].relay;
     const api = getRelayProtocolApi(relay!.protocol);
     const request: RequestArguments<RelayJsonRpc.BatchFetchMessagesParams> = {
       method: api.batchFetchMessages,
       params: {

packages/core/test/relayer.spec.ts

@@ -1,4 +1,4 @@
-import { expect, describe, it, beforeEach, afterEach } from "vitest";
+import { expect, describe, it, beforeEach, afterEach, vi } from "vitest";
 import { getDefaultLoggerOptions, pino } from "@walletconnect/logger";
 import { JsonRpcProvider } from "@walletconnect/jsonrpc-provider";
 
@@ -125,16 +125,20 @@ describe("Relayer", () => {
         projectId: TEST_CORE_OPTIONS.projectId,
       });
       await relayer.init();
+      await relayer.transportOpen();
     });
     afterEach(async () => {
       await disconnectSocket(relayer);
     });
 
     it("returns the id provided by calling `subscriber.subscribe` with the passed topic", async () => {
-      const spy = Sinon.spy((topic) => {
-        relayer.subscriber.events.emit(SUBSCRIBER_EVENTS.created, { topic });
-        return topic;
-      });
+      const spy = Sinon.spy(
+        (topic) =>
+          new Promise((resolve) => {
+            relayer.subscriber.events.emit(SUBSCRIBER_EVENTS.created, { topic });
+            resolve(topic);
+          }),
+      );
       relayer.subscriber.subscribe = spy;
 
       const testTopic = "abc123";
@@ -149,10 +153,13 @@ describe("Relayer", () => {
     });
 
     it("should subscribe multiple topics", async () => {
-      const spy = Sinon.spy((topic) => {
-        relayer.subscriber.events.emit(SUBSCRIBER_EVENTS.created, { topic });
-        return topic;
-      });
+      const spy = Sinon.spy(
+        (topic) =>
+          new Promise((resolve) => {
+            relayer.subscriber.events.emit(SUBSCRIBER_EVENTS.created, { topic });
+            resolve(topic);
+          }),
+      );
       relayer.subscriber.subscribe = spy;
       const subscriber = relayer.subscriber as ISubscriber;
       // record the number of listeners before subscribing
@@ -164,6 +171,23 @@ describe("Relayer", () => {
       expect(subscriber.events.listenerCount(SUBSCRIBER_EVENTS.created)).to.eq(startNumListeners);
     });
 
+    it("should throw when subscribe reaches a publish timeout", async () => {
+      await relayer.transportOpen();
+      await relayer.toEstablishConnection();
+      relayer.subscriber.subscribeTimeout = 5_000;
+      relayer.request = () => {
+        return new Promise<void>((_, reject) => {
+          setTimeout(() => {
+            reject(new Error("Subscription timeout"));
+          }, 100_000);
+        });
+      };
+      const topic = generateRandomBytes32();
+      await expect(relayer.subscribe(topic)).rejects.toThrow(
+        `Subscribing to ${topic} failed, please try again`,
+      );
+    });
+
     it("should be able to resubscribe on topic that already exists", async () => {
       const topic = generateRandomBytes32();
       const id = await relayer.subscribe(topic);

packages/types/src/core/relayer.ts

@@ -29,6 +29,9 @@ export declare namespace RelayerTypes {
   export interface SubscribeOptions {
     relay?: ProtocolOptions;
     transportType?: TransportType;
+    internal?: {
+      throwOnFailedPublish?: boolean;
+    };
   }
 
   export interface UnsubscribeOptions {