STREAM-1362: skip retry requests when not yet connected

Author: Aidan Zimmermann

src/notifications.ts

@@ -30,6 +30,7 @@ export class Notifications implements StreamingClientExtension {
 
   private internalSubscriptions: string[];
   private pendingBulkSubscribeRetry?: RetryPromise<any>;
+  private pendingConnectedWait?: { cancel: (reason?: any) => void };
 
   constructor (client, options?: IClientOptions) {
     this.subscriptions = {};
@@ -210,16 +211,43 @@ export class Notifications implements StreamingClientExtension {
     return keptTopics;
   }
 
-  makeBulkSubscribeRequest (topics: string[], options): Promise<AxiosResponse<ChannelTopicsEntityListing>> {
-    // Cancel any in-flight retry from a previous bulk subscribe. If the topic list has changed
-    // (subscribe/unsubscribe called between retries), the stale request would send an outdated
-    // topic list. Since bulk subscribes with replace=true use PUT (full replacement), only the
-    // most recent request matters.
+  async makeBulkSubscribeRequest (topics: string[], options): Promise<AxiosResponse<ChannelTopicsEntityListing>> {
+    // Cancel any in-flight retry or connected-wait from a previous bulk subscribe.
+    // If the topic list has changed (subscribe/unsubscribe called between retries),
+    // the stale request would send an outdated topic list. Since bulk subscribes with
+    // replace=true use PUT (full replacement), only the most recent request matters.
+    if (this.pendingConnectedWait) {
+      this.pendingConnectedWait.cancel(new Error('Superseded by newer bulk subscribe request'));
+      this.pendingConnectedWait = undefined;
+    }
     if (this.pendingBulkSubscribeRetry && !this.pendingBulkSubscribeRetry.hasCompleted()) {
       this.client.logger.info('Cancelling previous bulk subscribe retry — new request supersedes it');
       this.pendingBulkSubscribeRetry.cancel(new Error('Superseded by newer bulk subscribe request'));
     }
 
+    // If the client is not connected, wait for reconnection before attempting the request.
+    // This prevents firing requests against a dead/stale channel ID which would result in
+    // 400 errors (notification.unable.to.get.channel.id). After reconnection, the stanza
+    // instance will have the fresh channel ID from the new connection.
+    if (!this.client.connected) {
+      this.client.logger.info('Client not connected, waiting for reconnection before bulk subscribe', {
+        topicCount: topics.length
+      });
+      await new Promise<void>((resolve, reject) => {
+        const onConnected = () => {
+          this.pendingConnectedWait = undefined;
+          resolve();
+        };
+        this.client.once('connected', onConnected);
+        this.pendingConnectedWait = {
+          cancel: (reason?: any) => {
+            this.client.off('connected', onConnected);
+            reject(reason);
+          }
+        };
+      });
+    }
+
     const requestOptions: RequestApiOptions = {
       method: options.replace ? 'put' : 'post',
       host: this.client.config.apiHost,

test/unit/notifications.test.ts

@@ -140,6 +140,7 @@ describe('Notifications', () => {
         channelId,
         apiHost: 'example.com',
       });
+      client.connected = true;
       bulkSubscribeUrl = `https://api.example.com/api/v2/notifications/channels/${channelId}/subscriptions`;
       axiosMock.onPut(bulkSubscribeUrl).reply((config) => {
         const topics = JSON.parse(config.data).map((topic) => topic.id);
@@ -738,6 +739,7 @@ describe('Notifications', () => {
         apiHost: 'example.com',
         channelId: 'notification-test-channel'
       });
+      client.connected = true;
       const notification = new Notifications(client);
       notification.stanzaInstance = fakeStanza;
       const xmppSubscribeSpy = jest.fn().mockResolvedValue({});
@@ -1135,6 +1137,7 @@ describe('Notifications', () => {
         apiHost: 'example.com',
         channelId: 'notification-test-channel'
       });
+      client.connected = true;
       notification = new Notifications(client);
       notification.stanzaInstance = getFakeStanzaClient();
     });
@@ -1233,5 +1236,92 @@ describe('Notifications', () => {
       await expect(notification.bulkSubscribe(['topicA'], { replace: true }))
         .rejects.toBeNull();
     });
+
+    it('should wait for connected event before making bulk subscribe request when disconnected', async () => {
+      client.connected = false;
+      const requestSpy = jest.spyOn(client.http, 'requestApiWithRetry').mockReturnValue({
+        promise: Promise.resolve({ data: { entities: [] } }),
+        cancel: jest.fn(),
+        complete: jest.fn(),
+        hasCompleted: () => false,
+        _id: 'test-id'
+      } as any);
+
+      // Start the request — it should not fire immediately
+      const promise = notification.makeBulkSubscribeRequest(['topicA'], { replace: true });
+      expect(requestSpy).not.toHaveBeenCalled();
+
+      // Simulate reconnection
+      client.connected = true;
+      (client as any).emit('connected');
+
+      await promise;
+      expect(requestSpy).toHaveBeenCalledTimes(1);
+    });
+
+    it('should use the fresh channel ID after waiting for reconnection', async () => {
+      client.connected = false;
+      const oldChannelId = notification.stanzaInstance!.channelId;
+
+      const requestSpy = jest.spyOn(client.http, 'requestApiWithRetry').mockReturnValue({
+        promise: Promise.resolve({ data: { entities: [] } }),
+        cancel: jest.fn(),
+        complete: jest.fn(),
+        hasCompleted: () => false,
+        _id: 'test-id'
+      } as any);
+
+      const promise = notification.makeBulkSubscribeRequest(['topicA'], { replace: true });
+
+      // Simulate reconnection with a new stanza instance (new channel)
+      const newStanza = getFakeStanzaClient();
+      newStanza.channelId = 'brand-new-channel';
+      notification.stanzaInstance = newStanza;
+      client.connected = true;
+      (client as any).emit('connected');
+
+      await promise;
+
+      // Verify the request used the NEW channel ID, not the old one
+      const calledPath = requestSpy.mock.calls[0][0];
+      expect(calledPath).toContain('brand-new-channel');
+      expect(calledPath).not.toContain(oldChannelId);
+    });
+
+    it('should cancel pending connected-wait when a new bulk subscribe supersedes it', async () => {
+      client.connected = false;
+
+      // Start first request — will wait for connected
+      const firstPromise = notification.makeBulkSubscribeRequest(['topicA'], { replace: true });
+
+      // Second request while first is waiting — should cancel the first
+      client.connected = true;
+      const requestSpy = jest.spyOn(client.http, 'requestApiWithRetry').mockReturnValue({
+        promise: Promise.resolve({ data: { entities: [] } }),
+        cancel: jest.fn(),
+        complete: jest.fn(),
+        hasCompleted: () => false,
+        _id: 'test-id'
+      } as any);
+
+      await notification.makeBulkSubscribeRequest(['topicA', 'topicB'], { replace: true });
+
+      // The first promise should reject with supersession error
+      await expect(firstPromise).rejects.toThrow('Superseded by newer bulk subscribe request');
+    });
+
+    it('should proceed immediately when client is already connected', async () => {
+      client.connected = true;
+      const requestSpy = jest.spyOn(client.http, 'requestApiWithRetry').mockReturnValue({
+        promise: Promise.resolve({ data: { entities: [] } }),
+        cancel: jest.fn(),
+        complete: jest.fn(),
+        hasCompleted: () => false,
+        _id: 'test-id'
+      } as any);
+
+      await notification.makeBulkSubscribeRequest(['topicA'], { replace: true });
+      expect(requestSpy).toHaveBeenCalledTimes(1);
+    });
   });
 });