fix: clear the lazy init promise on publisher close

packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts

@@ -173,4 +173,11 @@ export abstract class AbstractPubSubPublisher<MessagePayloadType extends object>
   protected override resolveSchema(message: MessagePayloadType) {
     return this.messageSchemaContainer.resolveSchema(message)
   }
+
+  override async close(): Promise<void> {
+    // Dropped so a publish after close() starts a fresh lazy init, rather than awaiting an attempt
+    // that began before the close. Matches AbstractAmqpPublisher.
+    this.initPromise = undefined
+    await super.close()
+  }
 }

packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts

@@ -81,6 +81,30 @@ describe('PubSubPermissionPublisher', () => {
       await newPublisher.close()
     })
 
+    it('starts a fresh lazy init when publishing after close', async () => {
+      const newPublisher = new PubSubPermissionPublisher(diContainer.cradle)
+      // The first attempt never settles, as if it were still in flight when close() ran
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockImplementationOnce(() => new Promise(() => {}))
+      const message = (id: string) => ({
+        id,
+        messageType: 'add' as const,
+        timestamp: new Date().toISOString(),
+        userIds: ['userId1'],
+      })
+
+      void newPublisher.publish(message('close-reinit-1'))
+      await newPublisher.close()
+      await newPublisher.publish(message('close-reinit-2'))
+
+      const spy = await newPublisher.handlerSpy.waitForMessageWithId('close-reinit-2', 'published')
+      expect(spy.processingResult).toEqual({ status: 'published' })
+      expect(initSpy).toHaveBeenCalledTimes(2)
+
+      await newPublisher.close()
+    })
+
     it('publishes a message to topic', async () => {
       const message = {
         id: '1',

packages/sns/lib/sns/AbstractSnsPublisher.ts

@@ -301,4 +301,11 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
     return resolvedOptions
   }
   /* c8 ignore stop */
+
+  override async close(): Promise<void> {
+    // Dropped so a publish after close() starts a fresh lazy init, rather than awaiting an attempt
+    // that began before the close. Matches AbstractAmqpPublisher.
+    this.initPromise = undefined
+    await super.close()
+  }
 }

packages/sns/test/publishers/SnsPermissionPublisher.spec.ts

@@ -408,5 +408,23 @@ describe('SnsPermissionPublisher', () => {
 
       await newPublisher.close()
     })
+
+    it('starts a fresh lazy init when publishing after close', async () => {
+      const newPublisher = new SnsPermissionPublisher(diContainer.cradle)
+      // The first attempt never settles, as if it were still in flight when close() ran
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockImplementationOnce(() => new Promise(() => {}))
+
+      void newPublisher.publish({ id: '1', messageType: 'add' })
+      await newPublisher.close()
+      await newPublisher.publish({ id: '2', messageType: 'add' })
+
+      const spy = await newPublisher.handlerSpy.waitForMessageWithId('2', 'published')
+      expect(spy.processingResult).toEqual({ status: 'published' })
+      expect(initSpy).toHaveBeenCalledTimes(2)
+
+      await newPublisher.close()
+    })
   })
 })

packages/sqs/lib/sqs/AbstractSqsPublisher.ts

@@ -290,4 +290,11 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
 
     return resolvedOptions
   }
+
+  override async close(): Promise<void> {
+    // Dropped so a publish after close() starts a fresh lazy init, rather than awaiting an attempt
+    // that began before the close. Matches AbstractAmqpPublisher.
+    this.initPromise = undefined
+    await super.close()
+  }
 }

packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts

@@ -567,5 +567,23 @@ describe('SqsPermissionPublisher', () => {
 
       await newPublisher.close()
     })
+
+    it('starts a fresh lazy init when publishing after close', async () => {
+      const newPublisher = new SqsPermissionPublisher(diContainer.cradle)
+      // The first attempt never settles, as if it were still in flight when close() ran
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockImplementationOnce(() => new Promise(() => {}))
+
+      void newPublisher.publish({ id: '1', messageType: 'add' })
+      await newPublisher.close()
+      await newPublisher.publish({ id: '2', messageType: 'add' })
+
+      const spy = await newPublisher.handlerSpy.waitForMessageWithId('2', 'published')
+      expect(spy.processingResult).toEqual({ status: 'published' })
+      expect(initSpy).toHaveBeenCalledTimes(2)
+
+      await newPublisher.close()
+    })
   })
 })