diff --git a/packages/amqp/lib/AbstractAmqpPublisher.ts b/packages/amqp/lib/AbstractAmqpPublisher.ts index 2f1a478bf..dcef3f2db 100644 --- a/packages/amqp/lib/AbstractAmqpPublisher.ts +++ b/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 diff --git a/packages/amqp/test/publishers/AmqpPermissionPublisher.spec.ts b/packages/amqp/test/publishers/AmqpPermissionPublisher.spec.ts index 7962d7e68..b3ac7d615 100644 --- a/packages/amqp/test/publishers/AmqpPermissionPublisher.spec.ts +++ b/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() diff --git a/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts b/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts index 8462cb97c..51734e0f6 100644 --- a/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts +++ b/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts @@ -67,8 +67,14 @@ export abstract class AbstractPubSubPublisher 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 { diff --git a/packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts b/packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts index 29d3c8734..0e5cad8ca 100644 --- a/packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts +++ b/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', diff --git a/packages/sns/lib/sns/AbstractSnsPublisher.ts b/packages/sns/lib/sns/AbstractSnsPublisher.ts index 44bdaab42..8b666c16c 100644 --- a/packages/sns/lib/sns/AbstractSnsPublisher.ts +++ b/packages/sns/lib/sns/AbstractSnsPublisher.ts @@ -116,8 +116,14 @@ export abstract class AbstractSnsPublisher 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 { diff --git a/packages/sns/test/publishers/SnsPermissionPublisher.spec.ts b/packages/sns/test/publishers/SnsPermissionPublisher.spec.ts index def3e75f8..567001daf 100644 --- a/packages/sns/test/publishers/SnsPermissionPublisher.spec.ts +++ b/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() + }) }) }) diff --git a/packages/sqs/lib/sqs/AbstractSqsPublisher.ts b/packages/sqs/lib/sqs/AbstractSqsPublisher.ts index 728362961..959d38195 100644 --- a/packages/sqs/lib/sqs/AbstractSqsPublisher.ts +++ b/packages/sqs/lib/sqs/AbstractSqsPublisher.ts @@ -110,8 +110,14 @@ export abstract class AbstractSqsPublisher 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 { diff --git a/packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts b/packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts index af4243335..69464e176 100644 --- a/packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts +++ b/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() + }) }) })