fix: retry lazy publisher init after a failed attempt

packages/amqp/lib/AbstractAmqpPublisher.ts

@@ -83,12 +83,21 @@ export abstract class AbstractAmqpPublisher<
        * it is intentional that this promise is not awaited, that's how we keep the method invocation synchronous
        * RabbitMQ publish by itself doesn't guarantee that your message is delivered successfully, so this kind of fire-and-forget is not strongly different from how amqp-lib behaves in the first place.
        */
-      this.initPromise
-        .then(() => {
-          this.publish(message, options)
-        })
+      const initPromise = this.initPromise
+      initPromise
+        .then(
+          () => {
+            this.publish(message, options)
+          },
+          (err) => {
+            // Init failures only, not errors from the publish above: cleared so the next publish
+            // retries the init instead of re-awaiting the same rejection. Only if still ours: an
+            // earlier failed publish may already have started a retry.
+            if (this.initPromise === initPromise) this.initPromise = undefined
+            throw err
+          },
+        )
         .catch((err) => {
-          /* v8 ignore next */
           this.handleError(err)
         })
       return

packages/amqp/test/publishers/AmqpPermissionPublisher.spec.ts

@@ -4,7 +4,7 @@ import type { Channel } from 'amqplib'
 import type { AwilixContainer } from 'awilix'
 import { asClass, asFunction, Lifetime } from 'awilix'
 import { asMockFunction } from 'awilix-manager'
-import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest'
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import { ZodError } from 'zod/v4'
 import { AbstractAmqpQueuePublisher } from '../../lib/AbstractAmqpQueuePublisher.ts'
 import { deserializeAmqpMessage } from '../../lib/amqpMessageDeserializer.ts'
@@ -342,6 +342,50 @@ describe('PermissionPublisher', () => {
       })
     })
 
+    it('retries lazy init on the next publish after a failed one', async () => {
+      await permissionConsumer.close()
+      const newPublisher = new AmqpPermissionPublisher(diContainer.cradle)
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockRejectedValueOnce(new Error('Transient init failure'))
+      const handleErrorSpy = vi.spyOn(newPublisher as any, 'handleError')
+
+      const message = {
+        id: '1',
+        messageType: 'add',
+        userIds: [1],
+        permissions: ['100'],
+        timestamp: new Date(),
+      } satisfies PERMISSIONS_MESSAGE_TYPE
+
+      newPublisher.publish(message)
+      await vi.waitFor(() => {
+        expect(handleErrorSpy).toHaveBeenCalledWith(new Error('Transient init failure'))
+      })
+
+      newPublisher.publish(message)
+      await vi.waitFor(() => {
+        expect(initSpy).toHaveBeenCalledTimes(2)
+      })
+
+      let receivedMessage: unknown
+      await channel.consume(AmqpPermissionPublisher.QUEUE_NAME, (message) => {
+        if (message === null) {
+          return
+        }
+        receivedMessage = deserializeAmqpMessage(
+          message,
+          PERMISSIONS_MESSAGE_SCHEMA,
+          new FakeConsumerErrorResolver(),
+        ).result
+      })
+
+      await waitAndRetry(() => !!receivedMessage)
+      expect(receivedMessage).toMatchObject({ parsedMessage: { id: '1', messageType: 'add' } })
+
+      await newPublisher.close()
+    })
+
     it('publishes a message auto-filling internal properties', async () => {
       await permissionConsumer.close()
 

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

@@ -67,8 +67,14 @@ export abstract class AbstractPubSubPublisher<MessagePayloadType extends object>
       if (!this.initPromise) {
         this.initPromise = this.init()
       }
-      await this.initPromise
-      this.initPromise = undefined
+      const initPromise = this.initPromise
+      try {
+        await initPromise
+      } finally {
+        // Cleared on failure too, so the next publish retries the init instead of re-awaiting the
+        // same rejection. Only if still ours: a concurrent caller may already have started a retry.
+        if (this.initPromise === initPromise) this.initPromise = undefined
+      }
     }
 
     try {

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

@@ -1,5 +1,5 @@
 import type { PubSub } from '@google-cloud/pubsub'
-import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
 
 import { deletePubSubTopic } from '../utils/cleanupPubSub.ts'
 import { registerDependencies } from '../utils/testContext.ts'
@@ -58,6 +58,29 @@ describe('PubSubPermissionPublisher', () => {
   })
 
   describe('publish', () => {
+    it('retries lazy init on the next publish after a failed one', async () => {
+      const newPublisher = new PubSubPermissionPublisher(diContainer.cradle)
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockRejectedValueOnce(new Error('Transient init failure'))
+
+      const message = {
+        id: 'lazy-init-retry',
+        messageType: 'add' as const,
+        timestamp: new Date().toISOString(),
+        userIds: ['userId1'],
+      }
+
+      await expect(newPublisher.publish(message)).rejects.toThrow('Transient init failure')
+      await newPublisher.publish(message)
+
+      const spy = await newPublisher.handlerSpy.waitForMessageWithId('lazy-init-retry', '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

@@ -116,8 +116,14 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
       if (!this.initPromise) {
         this.initPromise = this.init()
       }
-      await this.initPromise
-      this.initPromise = undefined
+      const initPromise = this.initPromise
+      try {
+        await initPromise
+      } finally {
+        // Cleared on failure too, so the next publish retries the init instead of re-awaiting the
+        // same rejection. Only if still ours: a concurrent caller may already have started a retry.
+        if (this.initPromise === initPromise) this.initPromise = undefined
+      }
     }
 
     try {

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

@@ -387,5 +387,26 @@ describe('SnsPermissionPublisher', () => {
       const res = await newPublisher.handlerSpy.waitForMessageWithId('1', 'published')
       expect(res.message).toEqual(message)
     })
+
+    it('retries lazy init on the next publish after a failed one', async () => {
+      const newPublisher = new SnsPermissionPublisher(diContainer.cradle)
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockRejectedValueOnce(new Error('Transient init failure'))
+
+      const message = {
+        id: '1',
+        messageType: 'add',
+      } satisfies PERMISSIONS_ADD_MESSAGE_TYPE
+
+      await expect(newPublisher.publish(message)).rejects.toThrow('Transient init failure')
+      await newPublisher.publish(message)
+
+      const res = await newPublisher.handlerSpy.waitForMessageWithId('1', 'published')
+      expect(res.message).toEqual(message)
+      expect(initSpy).toHaveBeenCalledTimes(2)
+
+      await newPublisher.close()
+    })
   })
 })

packages/sqs/lib/sqs/AbstractSqsPublisher.ts

@@ -110,8 +110,14 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
       if (!this.initPromise) {
         this.initPromise = this.init()
       }
-      await this.initPromise
-      this.initPromise = undefined
+      const initPromise = this.initPromise
+      try {
+        await initPromise
+      } finally {
+        // Cleared on failure too, so the next publish retries the init instead of re-awaiting the
+        // same rejection. Only if still ours: a concurrent caller may already have started a retry.
+        if (this.initPromise === initPromise) this.initPromise = undefined
+      }
     }
 
     try {

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

@@ -502,5 +502,70 @@ describe('SqsPermissionPublisher', () => {
       expect(spy.message).toEqual(message)
       expect(spy.processingResult).toEqual({ status: 'published' })
     })
+
+    it('retries lazy init on the next publish after a failed one', async () => {
+      const newPublisher = new SqsPermissionPublisher(diContainer.cradle)
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockRejectedValueOnce(new Error('Transient init failure'))
+
+      const message = {
+        id: '1',
+        messageType: 'add',
+      } satisfies PERMISSIONS_ADD_MESSAGE_TYPE
+
+      await expect(newPublisher.publish(message)).rejects.toThrow('Transient init failure')
+      await newPublisher.publish(message)
+
+      const spy = await newPublisher.handlerSpy.waitForMessageWithId('1', 'published')
+      expect(spy.processingResult).toEqual({ status: 'published' })
+      expect(initSpy).toHaveBeenCalledTimes(2)
+
+      await newPublisher.close()
+    })
+
+    it('retries lazy init once the missing queue exists', async () => {
+      const queueName = 'lazy_init_retry_queue'
+      await diContainer.cradle.testAdmin.deleteQueues(queueName)
+      const newPublisher = new SqsPermissionPublisher(diContainer.cradle, {
+        locatorConfig: { queueName },
+      })
+
+      const message = {
+        id: '1',
+        messageType: 'add',
+      } satisfies PERMISSIONS_ADD_MESSAGE_TYPE
+
+      await expect(newPublisher.publish(message)).rejects.toThrow(/does not exist/)
+
+      await diContainer.cradle.testAdmin.createQueue(queueName)
+      await newPublisher.publish(message)
+
+      const spy = await newPublisher.handlerSpy.waitForMessageWithId('1', 'published')
+      expect(spy.processingResult).toEqual({ status: 'published' })
+
+      await newPublisher.close()
+      await diContainer.cradle.testAdmin.deleteQueues(queueName)
+    })
+
+    it('shares one failed lazy init between concurrent publishes', async () => {
+      const newPublisher = new SqsPermissionPublisher(diContainer.cradle)
+      const initSpy = vi
+        .spyOn(newPublisher, 'init')
+        .mockRejectedValueOnce(new Error('Transient init failure'))
+
+      const results = await Promise.allSettled([
+        newPublisher.publish({ id: '1', messageType: 'add' }),
+        newPublisher.publish({ id: '2', messageType: 'add' }),
+      ])
+
+      expect(results).toEqual([
+        { status: 'rejected', reason: new Error('Transient init failure') },
+        { status: 'rejected', reason: new Error('Transient init failure') },
+      ])
+      expect(initSpy).toHaveBeenCalledTimes(1)
+
+      await newPublisher.close()
+    })
   })
 })