From 1ec03783fe01dd847cd6b964418a62b55613fef0 Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 12:42:45 +0200 Subject: [PATCH 01/11] Implement polling for topics --- README.md | 77 ++++++ packages/core/lib/index.ts | 7 + .../core/lib/queues/AbstractQueueService.ts | 3 + packages/core/lib/types/queueOptionsTypes.ts | 60 +++++ .../utils/resourceAvailabilityUtils.spec.ts | 180 +++++++++++++ .../lib/utils/resourceAvailabilityUtils.ts | 164 ++++++++++++ packages/sns/lib/index.ts | 7 +- .../sns/lib/sns/AbstractSnsSqsConsumer.ts | 5 +- packages/sns/lib/utils/snsInitter.ts | 97 +++++-- packages/sns/lib/utils/snsUtils.ts | 6 +- ...ssionConsumer.resourceAvailability.spec.ts | 243 ++++++++++++++++++ packages/sqs/lib/index.ts | 7 +- packages/sqs/lib/sqs/AbstractSqsService.ts | 4 + packages/sqs/lib/utils/sqsInitter.ts | 44 +++- ...ssionConsumer.resourceAvailability.spec.ts | 181 +++++++++++++ 15 files changed, 1056 insertions(+), 29 deletions(-) create mode 100644 packages/core/lib/utils/resourceAvailabilityUtils.spec.ts create mode 100644 packages/core/lib/utils/resourceAvailabilityUtils.ts create mode 100644 packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts create mode 100644 packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts diff --git a/README.md b/README.md index a3886060..564e6fae 100644 --- a/README.md +++ b/README.md @@ -244,6 +244,83 @@ Both publishers and consumers accept a queue name and configuration as parameter If you do not want to create a new queue/topic, you can set `queueLocator` field for `queueConfiguration`. In that case `message-queue-toolkit` will not attempt to create a new queue or topic, and instead throw an error if they don't already exist. +## Resource Availability Polling (Eventual Consistency Mode) + +When using `locatorConfig` to reference existing queues or topics, the default behavior is to fail immediately if the resource doesn't exist. However, in some deployment scenarios (especially with cross-service dependencies), resources may not be available at startup time. + +The `resourceAvailabilityConfig` option enables "eventual consistency mode" where consumers poll for resources to become available instead of failing immediately. + +### Use Case: Cross-Service Dependencies + +Consider two services that need to subscribe to each other's topics: +- **Service A** needs to subscribe to **Service B's** topic +- **Service B** needs to subscribe to **Service A's** topic + +Without eventual consistency mode, neither service can deploy first because the other's topic doesn't exist yet. With `resourceAvailabilityConfig`, both services can start simultaneously and wait for the other's resources to appear. + +### Configuration + +```typescript +const consumer = new MySnsSqsConsumer(dependencies, { + locatorConfig: { + topicArn: 'arn:aws:sns:...', + queueUrl: 'https://sqs...', + subscriptionArn: '...' + }, + // Enable eventual consistency mode + resourceAvailabilityConfig: { + enabled: true, // Enable polling for resource availability + pollingIntervalMs: 5000, // Check every 5 seconds (default: 5000) + timeoutMs: 300000, // Fail after 5 minutes (optional, default: no timeout) + } +}) +``` + +### Configuration Options + +| Option | Type | Default | Description | +|--------|------|---------|-------------| +| `enabled` | `boolean` | `false` | Enable eventual consistency mode | +| `pollingIntervalMs` | `number` | `5000` | Interval between availability checks (ms) | +| `timeoutMs` | `number` | `undefined` | Maximum wait time before throwing `ResourceAvailabilityTimeoutError`. If not set, polls indefinitely | + +### Environment-Specific Configuration + +```typescript +// Development/Staging - poll indefinitely +{ + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 5000, + } +} + +// Production - poll with timeout to catch misconfigurations +{ + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 10000, + timeoutMs: 5 * 60 * 1000, // 5 minutes + } +} +``` + +### Error Handling + +When `timeoutMs` is configured and the resource doesn't become available within the specified time, a `ResourceAvailabilityTimeoutError` is thrown: + +```typescript +import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core' + +try { + await consumer.init() +} catch (error) { + if (error instanceof ResourceAvailabilityTimeoutError) { + console.error(`Resource ${error.resourceName} not available after ${error.timeoutMs}ms`) + } +} +``` + ## SQS Policy Configuration SQS queues can be configured with access policies to control who can send messages to and receive messages from the queue. The `policyConfig` parameter allows you to define these policies when creating or updating SQS queues. diff --git a/packages/core/lib/index.ts b/packages/core/lib/index.ts index 4fd464cd..bc16151a 100644 --- a/packages/core/lib/index.ts +++ b/packages/core/lib/index.ts @@ -105,5 +105,12 @@ export { isProduction, reloadConfig } from './utils/envUtils.ts' export { isShallowSubset, objectMatches } from './utils/matchUtils.ts' export { type ParseMessageResult, parseMessage } from './utils/parseUtils.ts' export { objectToBuffer } from './utils/queueUtils.ts' +export { + isResourceAvailabilityWaitingEnabled, + type ResourceAvailabilityCheckResult, + ResourceAvailabilityTimeoutError, + type WaitForResourceOptions, + waitForResource, +} from './utils/resourceAvailabilityUtils.ts' export { toDatePreprocessor } from './utils/toDateProcessor.ts' export { waitAndRetry } from './utils/waitUtils.ts' diff --git a/packages/core/lib/queues/AbstractQueueService.ts b/packages/core/lib/queues/AbstractQueueService.ts index 4a9e6a39..04a88b6b 100644 --- a/packages/core/lib/queues/AbstractQueueService.ts +++ b/packages/core/lib/queues/AbstractQueueService.ts @@ -46,6 +46,7 @@ import type { ProcessedMessageMetadata, QueueDependencies, QueueOptions, + ResourceAvailabilityConfig, } from '../types/queueOptionsTypes.ts' import { isRetryDateExceeded } from '../utils/dateUtils.ts' import { streamWithKnownSizeToString } from '../utils/streamUtils.ts' @@ -125,6 +126,7 @@ export abstract class AbstractQueueService< protected readonly creationConfig?: QueueConfiguration protected readonly locatorConfig?: QueueLocatorType protected readonly deletionConfig?: DeletionConfig + protected readonly resourceAvailabilityConfig?: ResourceAvailabilityConfig protected readonly payloadStoreConfig?: | MakeRequired | MakeRequired @@ -160,6 +162,7 @@ export abstract class AbstractQueueService< this.creationConfig = options.creationConfig this.locatorConfig = options.locatorConfig this.deletionConfig = options.deletionConfig + this.resourceAvailabilityConfig = options.resourceAvailabilityConfig this.payloadStoreConfig = options.payloadStoreConfig ? { serializer: jsonStreamStringifySerializer, diff --git a/packages/core/lib/types/queueOptionsTypes.ts b/packages/core/lib/types/queueOptionsTypes.ts index 9100af14..8ffcb6d5 100644 --- a/packages/core/lib/types/queueOptionsTypes.ts +++ b/packages/core/lib/types/queueOptionsTypes.ts @@ -118,6 +118,13 @@ export type CommonQueueOptions = { deletionConfig?: DeletionConfig payloadStoreConfig?: PayloadStoreConfig messageDeduplicationConfig?: MessageDeduplicationConfig + /** + * Configuration for eventual consistency mode. + * When enabled, the consumer will poll for the topic/queue to become available + * instead of failing immediately when using locatorConfig. + * This is useful for handling cross-service dependencies during deployment. + */ + resourceAvailabilityConfig?: ResourceAvailabilityConfig } export type CommonCreationConfigType = { @@ -130,6 +137,59 @@ export type DeletionConfig = { forceDeleteInProduction?: boolean } +/** + * Configuration for eventual consistency mode when resources may not exist at startup. + * + * This is useful in scenarios where services have cross-dependencies: + * - Service A needs to subscribe to Service B's topic + * - Service B needs to subscribe to Service A's topic + * - Neither can deploy first without the other's topic existing + * + * With `resourceAvailabilityConfig.enabled = true`, the consumer will poll for the + * topic/queue to become available instead of failing immediately. + * + * @example + * // Development/staging - poll indefinitely + * { + * resourceAvailabilityConfig: { + * enabled: true, + * pollingIntervalMs: 5000, + * } + * } + * + * @example + * // Production - poll with timeout to catch misconfigurations + * { + * resourceAvailabilityConfig: { + * enabled: true, + * timeoutMs: 5 * 60 * 1000, // 5 minutes + * pollingIntervalMs: 10000, + * } + * } + */ +export type ResourceAvailabilityConfig = { + /** + * If true, the consumer will poll for the topic/queue to become available + * instead of failing immediately when using locatorConfig. + * Default: false (fail immediately for backwards compatibility) + */ + enabled: boolean + + /** + * Maximum time in milliseconds to wait for the resource to become available. + * If not set or set to 0, will poll indefinitely (useful for dev/staging environments). + * For production, it's recommended to set a reasonable timeout (e.g., 5 minutes). + * Default: undefined (no timeout - poll indefinitely) + */ + timeoutMs?: number + + /** + * Interval in milliseconds between polling attempts. + * Default: 5000 (5 seconds) + */ + pollingIntervalMs?: number +} + type NewQueueOptions = { creationConfig?: CreationConfigType } diff --git a/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts b/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts new file mode 100644 index 00000000..f6bea58d --- /dev/null +++ b/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts @@ -0,0 +1,180 @@ +import { describe, expect, it, vi } from 'vitest' +import { + isResourceAvailabilityWaitingEnabled, + ResourceAvailabilityTimeoutError, + waitForResource, +} from './resourceAvailabilityUtils.ts' + +describe('resourceAvailabilityUtils', () => { + describe('isResourceAvailabilityWaitingEnabled', () => { + it('returns true when enabled is true', () => { + expect(isResourceAvailabilityWaitingEnabled({ enabled: true })).toBe(true) + }) + + it('returns false when enabled is false', () => { + expect(isResourceAvailabilityWaitingEnabled({ enabled: false })).toBe(false) + }) + + it('returns false when config is undefined', () => { + expect(isResourceAvailabilityWaitingEnabled(undefined)).toBe(false) + }) + }) + + describe('waitForResource', () => { + it('returns immediately when resource is available on first check', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' }) + + const result = await waitForResource({ + config: { enabled: true, pollingIntervalMs: 100 }, + checkFn, + resourceName: 'test-resource', + }) + + expect(result).toBe('test-result') + expect(checkFn).toHaveBeenCalledTimes(1) + }) + + it('polls until resource becomes available', async () => { + let callCount = 0 + const checkFn = vi.fn().mockImplementation(() => { + callCount++ + if (callCount < 3) { + return Promise.resolve({ isAvailable: false }) + } + return Promise.resolve({ isAvailable: true, result: 'test-result' }) + }) + + const result = await waitForResource({ + config: { enabled: true, pollingIntervalMs: 10 }, + checkFn, + resourceName: 'test-resource', + }) + + expect(result).toBe('test-result') + expect(checkFn).toHaveBeenCalledTimes(3) + }) + + it('throws ResourceAvailabilityTimeoutError when timeout is reached', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: false }) + + await expect( + waitForResource({ + config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 50 }, + checkFn, + resourceName: 'test-resource', + }), + ).rejects.toThrow(ResourceAvailabilityTimeoutError) + + // Should have made at least a few attempts + expect(checkFn.mock.calls.length).toBeGreaterThan(0) + }) + + it('throws the original error when checkFn throws', async () => { + const checkFn = vi.fn().mockRejectedValue(new Error('Unexpected error')) + + await expect( + waitForResource({ + config: { enabled: true, pollingIntervalMs: 10 }, + checkFn, + resourceName: 'test-resource', + }), + ).rejects.toThrow('Unexpected error') + + expect(checkFn).toHaveBeenCalledTimes(1) + }) + + it('uses default polling interval when not specified', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' }) + + const result = await waitForResource({ + config: { enabled: true }, + checkFn, + resourceName: 'test-resource', + }) + + expect(result).toBe('test-result') + }) + + it('logs progress when logger is provided', async () => { + const logger = { + info: vi.fn(), + debug: vi.fn(), + error: vi.fn(), + } + const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' }) + + await waitForResource({ + config: { enabled: true, pollingIntervalMs: 10 }, + checkFn, + resourceName: 'test-resource', + // @ts-expect-error - partial logger for testing + logger, + }) + + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + message: expect.stringContaining('Waiting for resource'), + resourceName: 'test-resource', + }), + ) + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + message: expect.stringContaining('is now available'), + resourceName: 'test-resource', + }), + ) + }) + + it('polls indefinitely when timeoutMs is not set', async () => { + let callCount = 0 + const checkFn = vi.fn().mockImplementation(() => { + callCount++ + if (callCount < 10) { + return Promise.resolve({ isAvailable: false }) + } + return Promise.resolve({ isAvailable: true, result: 'test-result' }) + }) + + const result = await waitForResource({ + config: { enabled: true, pollingIntervalMs: 1 }, + checkFn, + resourceName: 'test-resource', + }) + + expect(result).toBe('test-result') + expect(checkFn).toHaveBeenCalledTimes(10) + }) + + it('polls indefinitely when timeoutMs is 0', async () => { + let callCount = 0 + const checkFn = vi.fn().mockImplementation(() => { + callCount++ + if (callCount < 5) { + return Promise.resolve({ isAvailable: false }) + } + return Promise.resolve({ isAvailable: true, result: 'test-result' }) + }) + + const result = await waitForResource({ + config: { enabled: true, pollingIntervalMs: 1, timeoutMs: 0 }, + checkFn, + resourceName: 'test-resource', + }) + + expect(result).toBe('test-result') + expect(checkFn).toHaveBeenCalledTimes(5) + }) + }) + + describe('ResourceAvailabilityTimeoutError', () => { + it('includes resource name and timeout in message', () => { + const error = new ResourceAvailabilityTimeoutError('my-queue', 5000) + + expect(error.message).toContain('my-queue') + expect(error.message).toContain('5000') + expect(error.resourceName).toBe('my-queue') + expect(error.timeoutMs).toBe(5000) + expect(error.name).toBe('ResourceAvailabilityTimeoutError') + }) + }) +}) diff --git a/packages/core/lib/utils/resourceAvailabilityUtils.ts b/packages/core/lib/utils/resourceAvailabilityUtils.ts new file mode 100644 index 00000000..8ea7bfc9 --- /dev/null +++ b/packages/core/lib/utils/resourceAvailabilityUtils.ts @@ -0,0 +1,164 @@ +import type { CommonLogger } from '@lokalise/node-core' +import type { ResourceAvailabilityConfig } from '../types/queueOptionsTypes.ts' + +const DEFAULT_POLLING_INTERVAL_MS = 5000 + +export type ResourceAvailabilityCheckResult = + | { + isAvailable: true + result: T + } + | { + isAvailable: false + } + +export type WaitForResourceOptions = { + /** + * Resource availability configuration + */ + config: ResourceAvailabilityConfig + + /** + * Function that checks if the resource is available. + * Should return { isAvailable: true, result: T } when resource exists, + * or { isAvailable: false } when resource doesn't exist. + * Should throw on unexpected errors. + */ + checkFn: () => Promise> + + /** + * Human-readable name of the resource for logging + */ + resourceName: string + + /** + * Logger instance for progress logging + */ + logger?: CommonLogger +} + +export class ResourceAvailabilityTimeoutError extends Error { + public readonly resourceName: string + public readonly timeoutMs: number + + constructor(resourceName: string, timeoutMs: number) { + super( + `Timeout waiting for resource "${resourceName}" to become available after ${timeoutMs}ms. ` + + 'The resource may not exist or there may be a configuration issue.', + ) + this.name = 'ResourceAvailabilityTimeoutError' + this.resourceName = resourceName + this.timeoutMs = timeoutMs + } +} + +function checkTimeoutExceeded( + hasTimeout: boolean, + timeoutMs: number | undefined, + startTime: number, + resourceName: string, + attemptCount: number, + logger?: CommonLogger, +): void { + if (!hasTimeout || timeoutMs === undefined) return + + const elapsedMs = Date.now() - startTime + if (elapsedMs >= timeoutMs) { + logger?.error({ + message: `Timeout waiting for resource "${resourceName}" to become available`, + resourceName, + timeoutMs, + attemptCount, + elapsedMs, + }) + throw new ResourceAvailabilityTimeoutError(resourceName, timeoutMs) + } +} + +function logResourceAvailable( + resourceName: string, + attemptCount: number, + startTime: number, + logger?: CommonLogger, +): void { + const elapsedMs = Date.now() - startTime + logger?.info({ + message: `Resource "${resourceName}" is now available`, + resourceName, + attemptCount, + elapsedMs, + }) +} + +/** + * Waits for a resource to become available by polling. + * This is used for eventual consistency mode where resources may not exist at startup. + * + * @param options - Configuration and check function + * @returns The result from the check function when resource becomes available + * @throws ResourceAvailabilityTimeoutError if timeout is reached + */ +export async function waitForResource(options: WaitForResourceOptions): Promise { + const { config, checkFn, resourceName, logger } = options + const pollingIntervalMs = config.pollingIntervalMs ?? DEFAULT_POLLING_INTERVAL_MS + const timeoutMs = config.timeoutMs + const hasTimeout = timeoutMs !== undefined && timeoutMs > 0 + + const startTime = Date.now() + let attemptCount = 0 + + logger?.info({ + message: `Waiting for resource "${resourceName}" to become available`, + resourceName, + pollingIntervalMs, + timeoutMs: timeoutMs ?? 'indefinite', + }) + + while (true) { + attemptCount++ + checkTimeoutExceeded(hasTimeout, timeoutMs, startTime, resourceName, attemptCount, logger) + + try { + const result = await checkFn() + + if (result.isAvailable) { + logResourceAvailable(resourceName, attemptCount, startTime, logger) + return result.result + } + + // Resource not available yet, log and wait + if (attemptCount === 1 || attemptCount % 12 === 0) { + // Log on first attempt and then every minute (assuming 5s interval) + const elapsedMs = Date.now() - startTime + logger?.debug({ + message: `Resource "${resourceName}" not available yet, will retry`, + resourceName, + attemptCount, + elapsedMs, + nextRetryInMs: pollingIntervalMs, + }) + } + } catch (error) { + // Unexpected error during check - log and rethrow + logger?.error({ + message: `Error checking resource availability for "${resourceName}"`, + resourceName, + error, + attemptCount, + }) + throw error + } + + // Wait before next attempt + await new Promise((resolve) => setTimeout(resolve, pollingIntervalMs)) + } +} + +/** + * Helper to check if resource availability waiting is enabled + */ +export function isResourceAvailabilityWaitingEnabled( + config?: ResourceAvailabilityConfig, +): config is ResourceAvailabilityConfig { + return config?.enabled === true +} diff --git a/packages/sns/lib/index.ts b/packages/sns/lib/index.ts index 25c476d1..33321ec1 100644 --- a/packages/sns/lib/index.ts +++ b/packages/sns/lib/index.ts @@ -25,7 +25,12 @@ export { generateFilterAttributes, generateTopicSubscriptionPolicy, } from './utils/snsAttributeUtils.ts' -export { initSns, initSnsSqs } from './utils/snsInitter.ts' +export { + initSns, + initSnsSqs, + type InitSnsExtraParams, + type InitSnsSqsExtraParams, +} from './utils/snsInitter.ts' export { deserializeSNSMessage } from './utils/snsMessageDeserializer.ts' export { readSnsMessage } from './utils/snsMessageReader.ts' export { subscribeToTopic } from './utils/snsSubscriber.ts' diff --git a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts index 7aa6a81f..01e7983b 100644 --- a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts +++ b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts @@ -102,7 +102,10 @@ export abstract class AbstractSnsSqsConsumer< this.locatorConfig, this.creationConfig, this.subscriptionConfig, - { logger: this.logger }, + { + logger: this.logger, + resourceAvailabilityConfig: this.resourceAvailabilityConfig, + }, ) this.queueName = initSnsSqsResult.queueName this.queueUrl = initSnsSqsResult.queueUrl diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index 5c1c9b40..b3a49c79 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -2,8 +2,8 @@ import type { CreateTopicCommandInput, SNSClient } from '@aws-sdk/client-sns' import type { CreateQueueCommandInput, SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' import type { Either } from '@lokalise/node-core' -import type { DeletionConfig, ExtraParams } from '@message-queue-toolkit/core' -import { isProduction } from '@message-queue-toolkit/core' +import type { DeletionConfig, ExtraParams, ResourceAvailabilityConfig } from '@message-queue-toolkit/core' +import { isProduction, isResourceAvailabilityWaitingEnabled, waitForResource } from '@message-queue-toolkit/core' import { deleteQueue, getQueueAttributes, @@ -18,6 +18,14 @@ import { subscribeToTopic } from './snsSubscriber.ts' import { assertTopic, deleteSubscription, deleteTopic, getTopicAttributes } from './snsUtils.ts' import { buildTopicArn } from './stsUtils.ts' +export type InitSnsSqsExtraParams = ExtraParams & { + resourceAvailabilityConfig?: ResourceAvailabilityConfig +} + +export type InitSnsExtraParams = ExtraParams & { + resourceAvailabilityConfig?: ResourceAvailabilityConfig +} + // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: fixme export async function initSnsSqs( sqsClient: SQSClient, @@ -26,7 +34,7 @@ export async function initSnsSqs( locatorConfig?: SNSSQSQueueLocatorType, creationConfig?: SNSCreationConfig & SQSCreationConfig, subscriptionConfig?: SNSSubscriptionOptions, - extraParams?: ExtraParams, + extraParams?: InitSnsSqsExtraParams, ): Promise<{ subscriptionArn: string; topicArn: string; queueUrl: string; queueName: string }> { if (!locatorConfig?.subscriptionArn) { if (!creationConfig?.topic && !locatorConfig?.topicArn && !locatorConfig?.topicName) { @@ -85,23 +93,58 @@ export async function initSnsSqs( const queueUrl = await resolveQueueUrlFromLocatorConfig(sqsClient, locatorConfig) - const checkPromises: Promise>[] = [] // Check for existing resources, using the locators const subscriptionTopicArn = locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? '')) - const topicPromise = getTopicAttributes(snsClient, subscriptionTopicArn) - checkPromises.push(topicPromise) - const queuePromise = getQueueAttributes(sqsClient, queueUrl) - checkPromises.push(queuePromise) + const resourceAvailabilityConfig = extraParams?.resourceAvailabilityConfig - const [topicCheckResult, queueCheckResult] = await Promise.all(checkPromises) + // If resource availability waiting is enabled, poll for resources to become available + if (isResourceAvailabilityWaitingEnabled(resourceAvailabilityConfig)) { + // Wait for topic to become available + await waitForResource({ + config: resourceAvailabilityConfig, + resourceName: `SNS topic ${subscriptionTopicArn}`, + logger: extraParams?.logger, + checkFn: async () => { + const result = await getTopicAttributes(snsClient, subscriptionTopicArn) + if (result.error === 'not_found') { + return { isAvailable: false } + } + return { isAvailable: true, result: result.result } + }, + }) - if (queueCheckResult?.error === 'not_found') { - throw new Error(`Queue with queueUrl ${queueUrl} does not exist.`) - } - if (topicCheckResult?.error === 'not_found') { - throw new Error(`Topic with topicArn ${locatorConfig.topicArn} does not exist.`) + // Wait for queue to become available + await waitForResource({ + config: resourceAvailabilityConfig, + resourceName: `SQS queue ${queueUrl}`, + logger: extraParams?.logger, + checkFn: async () => { + const result = await getQueueAttributes(sqsClient, queueUrl) + if (result.error === 'not_found') { + return { isAvailable: false } + } + return { isAvailable: true, result: result.result } + }, + }) + } else { + // Original behavior: check resources once and fail immediately if not found + const checkPromises: Promise>[] = [] + const topicPromise = getTopicAttributes(snsClient, subscriptionTopicArn) + checkPromises.push(topicPromise) + + const queuePromise = getQueueAttributes(sqsClient, queueUrl) + checkPromises.push(queuePromise) + + const [topicCheckResult, queueCheckResult] = await Promise.all(checkPromises) + + if (queueCheckResult?.error === 'not_found') { + throw new Error(`Queue with queueUrl ${queueUrl} does not exist.`) + } + if (topicCheckResult?.error === 'not_found') { + throw new Error(`Topic with topicArn ${locatorConfig.topicArn} does not exist.`) + } } let queueName: string @@ -205,6 +248,7 @@ export async function initSns( stsClient: STSClient, locatorConfig?: SNSTopicLocatorType, creationConfig?: SNSCreationConfig, + extraParams?: InitSnsExtraParams, ) { if (locatorConfig) { if (!locatorConfig.topicArn && !locatorConfig.topicName) { @@ -216,9 +260,28 @@ export async function initSns( const topicArn = locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? '')) - const checkResult = await getTopicAttributes(snsClient, topicArn) - if (checkResult.error === 'not_found') { - throw new Error(`Topic with topicArn ${locatorConfig.topicArn} does not exist.`) + const resourceAvailabilityConfig = extraParams?.resourceAvailabilityConfig + + // If resource availability waiting is enabled, poll for topic to become available + if (isResourceAvailabilityWaitingEnabled(resourceAvailabilityConfig)) { + await waitForResource({ + config: resourceAvailabilityConfig, + resourceName: `SNS topic ${topicArn}`, + logger: extraParams?.logger, + checkFn: async () => { + const result = await getTopicAttributes(snsClient, topicArn) + if (result.error === 'not_found') { + return { isAvailable: false } + } + return { isAvailable: true, result: result.result } + }, + }) + } else { + // Original behavior: check once and fail immediately if not found + const checkResult = await getTopicAttributes(snsClient, topicArn) + if (checkResult.error === 'not_found') { + throw new Error(`Topic with topicArn ${locatorConfig.topicArn} does not exist.`) + } } return { topicArn } diff --git a/packages/sns/lib/utils/snsUtils.ts b/packages/sns/lib/utils/snsUtils.ts index 866899a0..7f541e38 100644 --- a/packages/sns/lib/utils/snsUtils.ts +++ b/packages/sns/lib/utils/snsUtils.ts @@ -38,8 +38,9 @@ export async function getTopicAttributes( }, } } catch (err) { + // SNS returns NotFoundException for non-existent topics // @ts-expect-error - if (err.Code === 'AWS.SimpleQueueService.NonExistentQueue') { + if (err.name === 'NotFoundException' || err.name === 'NotFound') { return { error: 'not_found', } @@ -64,8 +65,9 @@ export async function getSubscriptionAttributes( }, } } catch (err) { + // SNS returns NotFoundException for non-existent subscriptions // @ts-expect-error - if (err.Code === 'AWS.SimpleQueueService.NonExistentQueue') { + if (err.name === 'NotFoundException' || err.name === 'NotFound') { return { error: 'not_found', } diff --git a/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts b/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts new file mode 100644 index 00000000..2e892f7f --- /dev/null +++ b/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts @@ -0,0 +1,243 @@ +import { setTimeout } from 'node:timers/promises' +import type { SNSClient } from '@aws-sdk/client-sns' +import type { SQSClient } from '@aws-sdk/client-sqs' +import type { STSClient } from '@aws-sdk/client-sts' +import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core' +import { assertQueue, deleteQueue } from '@message-queue-toolkit/sqs' +import type { AwilixContainer } from 'awilix' +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest' +import { AbstractSnsSqsConsumer } from '../../lib/sns/AbstractSnsSqsConsumer.ts' +import type { SNSSQSConsumerDependencies } from '../../lib/sns/AbstractSnsSqsConsumer.ts' +import { assertTopic, deleteTopic } from '../../lib/utils/snsUtils.ts' +import type { Dependencies } from '../utils/testContext.ts' +import { registerDependencies } from '../utils/testContext.ts' +import { + PERMISSIONS_ADD_MESSAGE_SCHEMA, + type PERMISSIONS_ADD_MESSAGE_TYPE, +} from './userConsumerSchemas.ts' + +// Simple consumer for testing resource availability +class TestResourceAvailabilityConsumer extends AbstractSnsSqsConsumer< + PERMISSIONS_ADD_MESSAGE_TYPE, + undefined, + undefined +> { + constructor( + dependencies: SNSSQSConsumerDependencies, + options: { + locatorConfig?: { + topicArn?: string + topicName?: string + queueUrl?: string + queueName?: string + subscriptionArn?: string + } + creationConfig?: { + queue?: { QueueName: string } + topic?: { Name: string } + } + resourceAvailabilityConfig?: { + enabled: boolean + timeoutMs?: number + pollingIntervalMs?: number + } + }, + ) { + super( + dependencies, + { + handlers: [ + { + schema: PERMISSIONS_ADD_MESSAGE_SCHEMA, + handler: () => Promise.resolve({ result: 'success' }), + }, + ], + messageTypeResolver: { messageTypePath: 'messageType' }, + subscriptionConfig: { updateAttributesIfExists: false }, + ...options, + }, + undefined, + ) + } + + get subscriptionProps() { + return { + topicArn: this.topicArn, + queueUrl: this.queueUrl, + queueName: this.queueName, + subscriptionArn: this.subscriptionArn, + } + } +} + +describe('SnsSqsPermissionConsumer - resourceAvailabilityConfig', () => { + const queueName = 'resource-availability-test-queue' + const topicName = 'resource-availability-test-topic' + const queueUrl = `http://sqs.eu-west-1.localstack:4566/000000000000/${queueName}` + + let diContainer: AwilixContainer + let sqsClient: SQSClient + let snsClient: SNSClient + let stsClient: STSClient + + beforeAll(async () => { + diContainer = await registerDependencies({}, false) + sqsClient = diContainer.cradle.sqsClient + snsClient = diContainer.cradle.snsClient + stsClient = diContainer.cradle.stsClient + }) + + beforeEach(async () => { + await deleteQueue(sqsClient, queueName) + await deleteTopic(snsClient, stsClient, topicName) + }) + + afterEach(async () => { + await deleteQueue(sqsClient, queueName) + await deleteTopic(snsClient, stsClient, topicName) + }) + + afterAll(async () => { + const { awilixManager } = diContainer.cradle + await awilixManager.executeDispose() + await diContainer.dispose() + }) + + describe('when resourceAvailabilityConfig is enabled', () => { + it('waits for topic to become available and initializes successfully', async () => { + // Create queue first, but not the topic + await assertQueue(sqsClient, { QueueName: queueName }) + + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + topicName, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + }, + creationConfig: { + queue: { QueueName: queueName }, + }, + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + }, + }) + + // Start init in background + const initPromise = consumer.init() + + // Wait a bit then create the topic + await setTimeout(300) + const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) + + // Init should complete successfully + await initPromise + + expect(consumer.subscriptionProps.topicArn).toBe(topicArn) + expect(consumer.subscriptionProps.queueName).toBe(queueName) + }) + + it('waits for queue to become available and initializes successfully', async () => { + // Create topic first, but not the queue + const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) + + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + topicArn, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + }, + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + }, + }) + + // Start init in background + const initPromise = consumer.init() + + // Wait a bit then create the queue + await setTimeout(300) + await assertQueue(sqsClient, { QueueName: queueName }) + + // Init should complete successfully + await initPromise + + expect(consumer.subscriptionProps.topicArn).toBe(topicArn) + expect(consumer.subscriptionProps.queueUrl).toBe(queueUrl) + }) + + it('throws ResourceAvailabilityTimeoutError when timeout is reached', async () => { + // Create queue but not topic + await assertQueue(sqsClient, { QueueName: queueName }) + + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + topicName, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + }, + creationConfig: { + queue: { QueueName: queueName }, + }, + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: 200, // Short timeout + }, + }) + + // Should throw timeout error since topic never appears + await expect(consumer.init()).rejects.toThrow(ResourceAvailabilityTimeoutError) + }) + }) + + describe('when resourceAvailabilityConfig is disabled', () => { + it('throws immediately when topic does not exist', async () => { + // Create queue but not topic + await assertQueue(sqsClient, { QueueName: queueName }) + + const topicArn = 'arn:aws:sns:eu-west-1:000000000000:non-existent-topic' + + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + topicArn, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + }, + resourceAvailabilityConfig: { + enabled: false, + }, + }) + + // Should throw immediately + await expect(consumer.init()).rejects.toThrow(/does not exist/) + }) + + it('throws immediately when queue does not exist', async () => { + // Create topic but not queue + const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) + + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + topicArn, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + }, + resourceAvailabilityConfig: { + enabled: false, + }, + }) + + // Should throw immediately + await expect(consumer.init()).rejects.toThrow(/does not exist/) + }) + }) +}) diff --git a/packages/sqs/lib/index.ts b/packages/sqs/lib/index.ts index 1d397f43..d20c82c3 100644 --- a/packages/sqs/lib/index.ts +++ b/packages/sqs/lib/index.ts @@ -35,7 +35,12 @@ export { generateWildcardSnsArn, generateWildcardSqsArn, } from './utils/sqsAttributeUtils.ts' -export { deleteSqs, updateQueueAttributes } from './utils/sqsInitter.ts' +export { + deleteSqs, + initSqs, + updateQueueAttributes, + type InitSqsExtraParams, +} from './utils/sqsInitter.ts' export { deserializeSQSMessage } from './utils/sqsMessageDeserializer.ts' export { createEventBridgeResolverWithMapping, diff --git a/packages/sqs/lib/sqs/AbstractSqsService.ts b/packages/sqs/lib/sqs/AbstractSqsService.ts index 45aa0fc7..6990d99e 100644 --- a/packages/sqs/lib/sqs/AbstractSqsService.ts +++ b/packages/sqs/lib/sqs/AbstractSqsService.ts @@ -108,6 +108,10 @@ export abstract class AbstractSqsService< this.locatorConfig, this.creationConfig, this.isFifoQueue, + { + resourceAvailabilityConfig: this.resourceAvailabilityConfig, + logger: this.logger, + }, ) this.queueName = queueName this.queueUrl = queueUrl diff --git a/packages/sqs/lib/utils/sqsInitter.ts b/packages/sqs/lib/utils/sqsInitter.ts index b9272e97..a03eaf03 100644 --- a/packages/sqs/lib/utils/sqsInitter.ts +++ b/packages/sqs/lib/utils/sqsInitter.ts @@ -1,7 +1,8 @@ import type { QueueAttributeName, SQSClient } from '@aws-sdk/client-sqs' import { SetQueueAttributesCommand, TagQueueCommand } from '@aws-sdk/client-sqs' -import type { DeletionConfig } from '@message-queue-toolkit/core' -import { isProduction } from '@message-queue-toolkit/core' +import type { CommonLogger } from '@lokalise/node-core' +import type { DeletionConfig, ResourceAvailabilityConfig } from '@message-queue-toolkit/core' +import { isProduction, isResourceAvailabilityWaitingEnabled, waitForResource } from '@message-queue-toolkit/core' import type { SQSCreationConfig, SQSQueueLocatorType } from '../sqs/AbstractSqsService.ts' @@ -13,6 +14,11 @@ import { validateFifoQueueName, } from './sqsUtils.ts' +export type InitSqsExtraParams = { + resourceAvailabilityConfig?: ResourceAvailabilityConfig + logger?: CommonLogger +} + export async function deleteSqs( sqsClient: SQSClient, deletionConfig: DeletionConfig, @@ -68,18 +74,42 @@ export async function initSqs( locatorConfig?: Partial, creationConfig?: SQSCreationConfig, isFifoQueue?: boolean, + extraParams?: InitSqsExtraParams, ) { // reuse existing queue only if (locatorConfig) { const queueUrl = await resolveQueueUrlFromLocatorConfig(sqsClient, locatorConfig) - const checkResult = await getQueueAttributes(sqsClient, queueUrl, ['QueueArn']) - - if (checkResult.error === 'not_found') { - throw new Error(`Queue with queueUrl ${locatorConfig.queueUrl} does not exist.`) + const resourceAvailabilityConfig = extraParams?.resourceAvailabilityConfig + + let queueArn: string | undefined + + // If resource availability waiting is enabled, poll for queue to become available + if (isResourceAvailabilityWaitingEnabled(resourceAvailabilityConfig)) { + const result = await waitForResource({ + config: resourceAvailabilityConfig, + resourceName: `SQS queue ${queueUrl}`, + logger: extraParams?.logger, + checkFn: async () => { + const checkResult = await getQueueAttributes(sqsClient, queueUrl, ['QueueArn']) + if (checkResult.error === 'not_found') { + return { isAvailable: false } + } + return { isAvailable: true, result: checkResult.result?.attributes?.QueueArn } + }, + }) + queueArn = result + } else { + // Original behavior: check once and fail immediately if not found + const checkResult = await getQueueAttributes(sqsClient, queueUrl, ['QueueArn']) + + if (checkResult.error === 'not_found') { + throw new Error(`Queue with queueUrl ${locatorConfig.queueUrl} does not exist.`) + } + + queueArn = checkResult.result?.attributes?.QueueArn } - const queueArn = checkResult.result?.attributes?.QueueArn if (!queueArn) { throw new Error('Queue ARN was not set') } diff --git a/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts b/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts new file mode 100644 index 00000000..8ff7cde5 --- /dev/null +++ b/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts @@ -0,0 +1,181 @@ +import { setTimeout } from 'node:timers/promises' +import type { SQSClient } from '@aws-sdk/client-sqs' +import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core' +import type { AwilixContainer } from 'awilix' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' + +import type { SQSConsumerDependencies } from '../../lib/sqs/AbstractSqsConsumer.ts' +import { AbstractSqsConsumer } from '../../lib/sqs/AbstractSqsConsumer.ts' +import { assertQueue, deleteQueue } from '../../lib/utils/sqsUtils.ts' +import type { Dependencies } from '../utils/testContext.ts' +import { registerDependencies } from '../utils/testContext.ts' +import { + PERMISSIONS_ADD_MESSAGE_SCHEMA, + type PERMISSIONS_ADD_MESSAGE_TYPE, +} from './userConsumerSchemas.ts' + +// Simple consumer for testing resource availability +class TestResourceAvailabilityConsumer extends AbstractSqsConsumer< + PERMISSIONS_ADD_MESSAGE_TYPE, + undefined, + undefined +> { + constructor( + dependencies: SQSConsumerDependencies, + options: { + locatorConfig?: { + queueUrl?: string + queueName?: string + } + creationConfig?: { + queue?: { QueueName: string } + } + resourceAvailabilityConfig?: { + enabled: boolean + timeoutMs?: number + pollingIntervalMs?: number + } + }, + ) { + super( + dependencies, + { + handlers: [ + { + schema: PERMISSIONS_ADD_MESSAGE_SCHEMA, + handler: () => Promise.resolve({ result: 'success' }), + }, + ], + messageTypeResolver: { messageTypePath: 'messageType' }, + ...options, + }, + undefined, + ) + } + + get queueProps() { + return { + url: this.queueUrl, + name: this.queueName, + arn: this.queueArn, + } + } +} + +describe('SqsPermissionConsumer - resourceAvailabilityConfig', () => { + const queueName = 'resource-availability-test-queue' + const queueUrl = `http://sqs.eu-west-1.localstack:4566/000000000000/${queueName}` + + let diContainer: AwilixContainer + let sqsClient: SQSClient + + beforeEach(async () => { + diContainer = await registerDependencies() + sqsClient = diContainer.cradle.sqsClient + await deleteQueue(sqsClient, queueName) + }) + + afterEach(async () => { + await deleteQueue(sqsClient, queueName) + await diContainer.cradle.awilixManager.executeDispose() + await diContainer.dispose() + }) + + describe('when resourceAvailabilityConfig is enabled', () => { + it('waits for queue to become available and initializes successfully', async () => { + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + queueUrl, + }, + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + }, + }) + + // Start init in background + const initPromise = consumer.init() + + // Wait a bit then create the queue + await setTimeout(300) + await assertQueue(sqsClient, { QueueName: queueName }) + + // Init should complete successfully + await initPromise + + expect(consumer.queueProps.url).toBe(queueUrl) + expect(consumer.queueProps.name).toBe(queueName) + }) + + it('throws ResourceAvailabilityTimeoutError when timeout is reached', async () => { + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + queueUrl, + }, + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: 200, // Short timeout + }, + }) + + // Should throw timeout error since queue never appears + await expect(consumer.init()).rejects.toThrow(ResourceAvailabilityTimeoutError) + }) + + it('polls indefinitely when timeoutMs is not set', async () => { + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + queueUrl, + }, + resourceAvailabilityConfig: { + enabled: true, + pollingIntervalMs: 50, + // No timeout - polls indefinitely + }, + }) + + // Start init in background + const initPromise = consumer.init() + + // Wait a bit then create the queue + await setTimeout(500) + await assertQueue(sqsClient, { QueueName: queueName }) + + // Init should complete successfully + await initPromise + + expect(consumer.queueProps.url).toBe(queueUrl) + }) + }) + + describe('when resourceAvailabilityConfig is disabled', () => { + it('throws immediately when queue does not exist', async () => { + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + queueUrl, + }, + resourceAvailabilityConfig: { + enabled: false, + }, + }) + + // Should throw immediately + await expect(consumer.init()).rejects.toThrow(/does not exist/) + }) + }) + + describe('when resourceAvailabilityConfig is not provided', () => { + it('throws immediately when queue does not exist (default behavior)', async () => { + const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + locatorConfig: { + queueUrl, + }, + }) + + // Should throw immediately (backwards compatible behavior) + await expect(consumer.init()).rejects.toThrow(/does not exist/) + }) + }) +}) From 4579bde7cf8bed5ecd3c276f7ac5cb7492db947a Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 12:46:48 +0200 Subject: [PATCH 02/11] Cleanup --- packages/core/lib/utils/resourceAvailabilityUtils.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/core/lib/utils/resourceAvailabilityUtils.ts b/packages/core/lib/utils/resourceAvailabilityUtils.ts index 8ea7bfc9..66527ed7 100644 --- a/packages/core/lib/utils/resourceAvailabilityUtils.ts +++ b/packages/core/lib/utils/resourceAvailabilityUtils.ts @@ -1,3 +1,4 @@ +import { setTimeout } from 'node:timers/promises' import type { CommonLogger } from '@lokalise/node-core' import type { ResourceAvailabilityConfig } from '../types/queueOptionsTypes.ts' @@ -150,7 +151,7 @@ export async function waitForResource(options: WaitForResourceOptions): Pr } // Wait before next attempt - await new Promise((resolve) => setTimeout(resolve, pollingIntervalMs)) + await setTimeout(pollingIntervalMs) } } From 2b833a80d0e178d6258ac527b86eaa4973894184 Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 12:47:17 +0200 Subject: [PATCH 03/11] Cleanup --- packages/sns/lib/index.ts | 4 ++-- packages/sns/lib/utils/snsInitter.ts | 12 ++++++++++-- ...qsPermissionConsumer.resourceAvailability.spec.ts | 2 +- packages/sqs/lib/index.ts | 2 +- packages/sqs/lib/utils/sqsInitter.ts | 6 +++++- 5 files changed, 19 insertions(+), 7 deletions(-) diff --git a/packages/sns/lib/index.ts b/packages/sns/lib/index.ts index 33321ec1..2c7981a4 100644 --- a/packages/sns/lib/index.ts +++ b/packages/sns/lib/index.ts @@ -26,10 +26,10 @@ export { generateTopicSubscriptionPolicy, } from './utils/snsAttributeUtils.ts' export { - initSns, - initSnsSqs, type InitSnsExtraParams, type InitSnsSqsExtraParams, + initSns, + initSnsSqs, } from './utils/snsInitter.ts' export { deserializeSNSMessage } from './utils/snsMessageDeserializer.ts' export { readSnsMessage } from './utils/snsMessageReader.ts' diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index b3a49c79..31bb6f01 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -2,8 +2,16 @@ import type { CreateTopicCommandInput, SNSClient } from '@aws-sdk/client-sns' import type { CreateQueueCommandInput, SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' import type { Either } from '@lokalise/node-core' -import type { DeletionConfig, ExtraParams, ResourceAvailabilityConfig } from '@message-queue-toolkit/core' -import { isProduction, isResourceAvailabilityWaitingEnabled, waitForResource } from '@message-queue-toolkit/core' +import type { + DeletionConfig, + ExtraParams, + ResourceAvailabilityConfig, +} from '@message-queue-toolkit/core' +import { + isProduction, + isResourceAvailabilityWaitingEnabled, + waitForResource, +} from '@message-queue-toolkit/core' import { deleteQueue, getQueueAttributes, diff --git a/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts b/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts index 2e892f7f..8894d6c3 100644 --- a/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts +++ b/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts @@ -6,8 +6,8 @@ import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core' import { assertQueue, deleteQueue } from '@message-queue-toolkit/sqs' import type { AwilixContainer } from 'awilix' import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest' -import { AbstractSnsSqsConsumer } from '../../lib/sns/AbstractSnsSqsConsumer.ts' import type { SNSSQSConsumerDependencies } from '../../lib/sns/AbstractSnsSqsConsumer.ts' +import { AbstractSnsSqsConsumer } from '../../lib/sns/AbstractSnsSqsConsumer.ts' import { assertTopic, deleteTopic } from '../../lib/utils/snsUtils.ts' import type { Dependencies } from '../utils/testContext.ts' import { registerDependencies } from '../utils/testContext.ts' diff --git a/packages/sqs/lib/index.ts b/packages/sqs/lib/index.ts index d20c82c3..49d3803c 100644 --- a/packages/sqs/lib/index.ts +++ b/packages/sqs/lib/index.ts @@ -37,9 +37,9 @@ export { } from './utils/sqsAttributeUtils.ts' export { deleteSqs, + type InitSqsExtraParams, initSqs, updateQueueAttributes, - type InitSqsExtraParams, } from './utils/sqsInitter.ts' export { deserializeSQSMessage } from './utils/sqsMessageDeserializer.ts' export { diff --git a/packages/sqs/lib/utils/sqsInitter.ts b/packages/sqs/lib/utils/sqsInitter.ts index a03eaf03..b7cafb2d 100644 --- a/packages/sqs/lib/utils/sqsInitter.ts +++ b/packages/sqs/lib/utils/sqsInitter.ts @@ -2,7 +2,11 @@ import type { QueueAttributeName, SQSClient } from '@aws-sdk/client-sqs' import { SetQueueAttributesCommand, TagQueueCommand } from '@aws-sdk/client-sqs' import type { CommonLogger } from '@lokalise/node-core' import type { DeletionConfig, ResourceAvailabilityConfig } from '@message-queue-toolkit/core' -import { isProduction, isResourceAvailabilityWaitingEnabled, waitForResource } from '@message-queue-toolkit/core' +import { + isProduction, + isResourceAvailabilityWaitingEnabled, + waitForResource, +} from '@message-queue-toolkit/core' import type { SQSCreationConfig, SQSQueueLocatorType } from '../sqs/AbstractSqsService.ts' From dbc10e1a857e0ec477e652e20da41ee17c466eab Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 12:59:56 +0200 Subject: [PATCH 04/11] Address review comments --- packages/core/lib/types/queueOptionsTypes.ts | 21 +++++--- .../utils/resourceAvailabilityUtils.spec.ts | 6 +++ .../lib/utils/resourceAvailabilityUtils.ts | 5 +- packages/sns/lib/utils/snsInitter.ts | 4 +- ...ssionConsumer.resourceAvailability.spec.ts | 48 +++++++------------ packages/sqs/lib/utils/sqsInitter.ts | 2 +- ...ssionConsumer.resourceAvailability.spec.ts | 45 +++++++---------- 7 files changed, 62 insertions(+), 69 deletions(-) diff --git a/packages/core/lib/types/queueOptionsTypes.ts b/packages/core/lib/types/queueOptionsTypes.ts index 8ffcb6d5..839b990e 100644 --- a/packages/core/lib/types/queueOptionsTypes.ts +++ b/packages/core/lib/types/queueOptionsTypes.ts @@ -145,14 +145,13 @@ export type DeletionConfig = { * - Service B needs to subscribe to Service A's topic * - Neither can deploy first without the other's topic existing * - * With `resourceAvailabilityConfig.enabled = true`, the consumer will poll for the + * When `resourceAvailabilityConfig` is provided, the consumer will poll for the * topic/queue to become available instead of failing immediately. * * @example * // Development/staging - poll indefinitely * { * resourceAvailabilityConfig: { - * enabled: true, * pollingIntervalMs: 5000, * } * } @@ -161,19 +160,27 @@ export type DeletionConfig = { * // Production - poll with timeout to catch misconfigurations * { * resourceAvailabilityConfig: { - * enabled: true, * timeoutMs: 5 * 60 * 1000, // 5 minutes * pollingIntervalMs: 10000, * } * } + * + * @example + * // Temporarily disable without removing config + * { + * resourceAvailabilityConfig: { + * enabled: false, + * timeoutMs: 5 * 60 * 1000, + * } + * } */ export type ResourceAvailabilityConfig = { /** - * If true, the consumer will poll for the topic/queue to become available - * instead of failing immediately when using locatorConfig. - * Default: false (fail immediately for backwards compatibility) + * Controls whether polling is enabled. + * Default: true (when resourceAvailabilityConfig is provided) + * Set to false to temporarily disable polling without removing the config. */ - enabled: boolean + enabled?: boolean /** * Maximum time in milliseconds to wait for the resource to become available. diff --git a/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts b/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts index f6bea58d..527ac7c8 100644 --- a/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts +++ b/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts @@ -11,6 +11,12 @@ describe('resourceAvailabilityUtils', () => { expect(isResourceAvailabilityWaitingEnabled({ enabled: true })).toBe(true) }) + it('returns true when enabled is not specified (defaults to true)', () => { + expect(isResourceAvailabilityWaitingEnabled({})).toBe(true) + expect(isResourceAvailabilityWaitingEnabled({ pollingIntervalMs: 1000 })).toBe(true) + expect(isResourceAvailabilityWaitingEnabled({ timeoutMs: 5000 })).toBe(true) + }) + it('returns false when enabled is false', () => { expect(isResourceAvailabilityWaitingEnabled({ enabled: false })).toBe(false) }) diff --git a/packages/core/lib/utils/resourceAvailabilityUtils.ts b/packages/core/lib/utils/resourceAvailabilityUtils.ts index 66527ed7..602564e9 100644 --- a/packages/core/lib/utils/resourceAvailabilityUtils.ts +++ b/packages/core/lib/utils/resourceAvailabilityUtils.ts @@ -156,10 +156,11 @@ export async function waitForResource(options: WaitForResourceOptions): Pr } /** - * Helper to check if resource availability waiting is enabled + * Helper to check if resource availability waiting is enabled. + * Returns true when config is provided and enabled is not explicitly false. */ export function isResourceAvailabilityWaitingEnabled( config?: ResourceAvailabilityConfig, ): config is ResourceAvailabilityConfig { - return config?.enabled === true + return config != null && config.enabled !== false } diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index 31bb6f01..abdea964 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -151,7 +151,7 @@ export async function initSnsSqs( throw new Error(`Queue with queueUrl ${queueUrl} does not exist.`) } if (topicCheckResult?.error === 'not_found') { - throw new Error(`Topic with topicArn ${locatorConfig.topicArn} does not exist.`) + throw new Error(`Topic with topicArn ${subscriptionTopicArn} does not exist.`) } } @@ -288,7 +288,7 @@ export async function initSns( // Original behavior: check once and fail immediately if not found const checkResult = await getTopicAttributes(snsClient, topicArn) if (checkResult.error === 'not_found') { - throw new Error(`Topic with topicArn ${locatorConfig.topicArn} does not exist.`) + throw new Error(`Topic with topicArn ${topicArn} does not exist.`) } } diff --git a/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts b/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts index 8894d6c3..b55be3ab 100644 --- a/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts +++ b/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts @@ -2,12 +2,18 @@ import { setTimeout } from 'node:timers/promises' import type { SNSClient } from '@aws-sdk/client-sns' import type { SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' -import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core' +import { + MessageHandlerConfigBuilder, + ResourceAvailabilityTimeoutError, +} from '@message-queue-toolkit/core' import { assertQueue, deleteQueue } from '@message-queue-toolkit/sqs' import type { AwilixContainer } from 'awilix' import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest' -import type { SNSSQSConsumerDependencies } from '../../lib/sns/AbstractSnsSqsConsumer.ts' -import { AbstractSnsSqsConsumer } from '../../lib/sns/AbstractSnsSqsConsumer.ts' +import { + AbstractSnsSqsConsumer, + type SNSSQSConsumerDependencies, + type SNSSQSConsumerOptions, +} from '../../lib/sns/AbstractSnsSqsConsumer.ts' import { assertTopic, deleteTopic } from '../../lib/utils/snsUtils.ts' import type { Dependencies } from '../utils/testContext.ts' import { registerDependencies } from '../utils/testContext.ts' @@ -16,42 +22,24 @@ import { type PERMISSIONS_ADD_MESSAGE_TYPE, } from './userConsumerSchemas.ts' +type TestConsumerOptions = Pick< + SNSSQSConsumerOptions, + 'locatorConfig' | 'creationConfig' | 'resourceAvailabilityConfig' +> + // Simple consumer for testing resource availability class TestResourceAvailabilityConsumer extends AbstractSnsSqsConsumer< PERMISSIONS_ADD_MESSAGE_TYPE, undefined, undefined > { - constructor( - dependencies: SNSSQSConsumerDependencies, - options: { - locatorConfig?: { - topicArn?: string - topicName?: string - queueUrl?: string - queueName?: string - subscriptionArn?: string - } - creationConfig?: { - queue?: { QueueName: string } - topic?: { Name: string } - } - resourceAvailabilityConfig?: { - enabled: boolean - timeoutMs?: number - pollingIntervalMs?: number - } - }, - ) { + constructor(dependencies: SNSSQSConsumerDependencies, options: TestConsumerOptions) { super( dependencies, { - handlers: [ - { - schema: PERMISSIONS_ADD_MESSAGE_SCHEMA, - handler: () => Promise.resolve({ result: 'success' }), - }, - ], + handlers: new MessageHandlerConfigBuilder() + .addConfig(PERMISSIONS_ADD_MESSAGE_SCHEMA, () => Promise.resolve({ result: 'success' })) + .build(), messageTypeResolver: { messageTypePath: 'messageType' }, subscriptionConfig: { updateAttributesIfExists: false }, ...options, diff --git a/packages/sqs/lib/utils/sqsInitter.ts b/packages/sqs/lib/utils/sqsInitter.ts index b7cafb2d..f0e2ba30 100644 --- a/packages/sqs/lib/utils/sqsInitter.ts +++ b/packages/sqs/lib/utils/sqsInitter.ts @@ -108,7 +108,7 @@ export async function initSqs( const checkResult = await getQueueAttributes(sqsClient, queueUrl, ['QueueArn']) if (checkResult.error === 'not_found') { - throw new Error(`Queue with queueUrl ${locatorConfig.queueUrl} does not exist.`) + throw new Error(`Queue with queueUrl ${queueUrl} does not exist.`) } queueArn = checkResult.result?.attributes?.QueueArn diff --git a/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts b/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts index 8ff7cde5..3ed6e84c 100644 --- a/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts +++ b/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts @@ -1,11 +1,16 @@ import { setTimeout } from 'node:timers/promises' import type { SQSClient } from '@aws-sdk/client-sqs' -import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core' +import { + MessageHandlerConfigBuilder, + ResourceAvailabilityTimeoutError, +} from '@message-queue-toolkit/core' import type { AwilixContainer } from 'awilix' import { afterEach, beforeEach, describe, expect, it } from 'vitest' - -import type { SQSConsumerDependencies } from '../../lib/sqs/AbstractSqsConsumer.ts' -import { AbstractSqsConsumer } from '../../lib/sqs/AbstractSqsConsumer.ts' +import { + AbstractSqsConsumer, + type SQSConsumerDependencies, + type SQSConsumerOptions, +} from '../../lib/sqs/AbstractSqsConsumer.ts' import { assertQueue, deleteQueue } from '../../lib/utils/sqsUtils.ts' import type { Dependencies } from '../utils/testContext.ts' import { registerDependencies } from '../utils/testContext.ts' @@ -14,38 +19,24 @@ import { type PERMISSIONS_ADD_MESSAGE_TYPE, } from './userConsumerSchemas.ts' +type TestConsumerOptions = Pick< + SQSConsumerOptions, + 'locatorConfig' | 'creationConfig' | 'resourceAvailabilityConfig' +> + // Simple consumer for testing resource availability class TestResourceAvailabilityConsumer extends AbstractSqsConsumer< PERMISSIONS_ADD_MESSAGE_TYPE, undefined, undefined > { - constructor( - dependencies: SQSConsumerDependencies, - options: { - locatorConfig?: { - queueUrl?: string - queueName?: string - } - creationConfig?: { - queue?: { QueueName: string } - } - resourceAvailabilityConfig?: { - enabled: boolean - timeoutMs?: number - pollingIntervalMs?: number - } - }, - ) { + constructor(dependencies: SQSConsumerDependencies, options: TestConsumerOptions) { super( dependencies, { - handlers: [ - { - schema: PERMISSIONS_ADD_MESSAGE_SCHEMA, - handler: () => Promise.resolve({ result: 'success' }), - }, - ], + handlers: new MessageHandlerConfigBuilder() + .addConfig(PERMISSIONS_ADD_MESSAGE_SCHEMA, () => Promise.resolve({ result: 'success' })) + .build(), messageTypeResolver: { messageTypePath: 'messageType' }, ...options, }, From 8970771d38caf70b6864f201c818563c3d89d96b Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 14:33:15 +0200 Subject: [PATCH 05/11] Address review comments --- README.md | 67 +++++++++----- packages/core/lib/index.ts | 8 +- .../core/lib/queues/AbstractQueueService.ts | 3 - packages/core/lib/types/queueOptionsTypes.ts | 87 +++++++++++++------ ...ts => startupResourcePollingUtils.spec.ts} | 74 +++++++--------- ...tils.ts => startupResourcePollingUtils.ts} | 42 ++++----- packages/sns/lib/sns/AbstractSnsService.ts | 9 +- .../sns/lib/sns/AbstractSnsSqsConsumer.ts | 5 +- packages/sns/lib/utils/snsInitter.ts | 34 +++----- ...onConsumer.startupResourcePolling.spec.ts} | 76 ++++++++-------- packages/sqs/lib/sqs/AbstractSqsService.ts | 31 ++++--- packages/sqs/lib/utils/sqsInitter.ts | 13 ++- ...onConsumer.startupResourcePolling.spec.ts} | 72 +++++++-------- 13 files changed, 278 insertions(+), 243 deletions(-) rename packages/core/lib/utils/{resourceAvailabilityUtils.spec.ts => startupResourcePollingUtils.spec.ts} (64%) rename packages/core/lib/utils/{resourceAvailabilityUtils.ts => startupResourcePollingUtils.ts} (75%) rename packages/sns/test/consumers/{SnsSqsPermissionConsumer.resourceAvailability.spec.ts => SnsSqsPermissionConsumer.startupResourcePolling.spec.ts} (77%) rename packages/sqs/test/consumers/{SqsPermissionConsumer.resourceAvailability.spec.ts => SqsPermissionConsumer.startupResourcePolling.spec.ts} (67%) diff --git a/README.md b/README.md index 564e6fae..3074e865 100644 --- a/README.md +++ b/README.md @@ -244,11 +244,11 @@ Both publishers and consumers accept a queue name and configuration as parameter If you do not want to create a new queue/topic, you can set `queueLocator` field for `queueConfiguration`. In that case `message-queue-toolkit` will not attempt to create a new queue or topic, and instead throw an error if they don't already exist. -## Resource Availability Polling (Eventual Consistency Mode) +## Startup Resource Polling (Eventual Consistency Mode) When using `locatorConfig` to reference existing queues or topics, the default behavior is to fail immediately if the resource doesn't exist. However, in some deployment scenarios (especially with cross-service dependencies), resources may not be available at startup time. -The `resourceAvailabilityConfig` option enables "eventual consistency mode" where consumers poll for resources to become available instead of failing immediately. +The `startupResourcePolling` option within `locatorConfig` enables "eventual consistency mode" where consumers poll for resources to become available instead of failing immediately. ### Use Case: Cross-Service Dependencies @@ -256,22 +256,24 @@ Consider two services that need to subscribe to each other's topics: - **Service A** needs to subscribe to **Service B's** topic - **Service B** needs to subscribe to **Service A's** topic -Without eventual consistency mode, neither service can deploy first because the other's topic doesn't exist yet. With `resourceAvailabilityConfig`, both services can start simultaneously and wait for the other's resources to appear. +Without eventual consistency mode, neither service can deploy first because the other's topic doesn't exist yet. With `startupResourcePolling`, both services can start simultaneously and wait for the other's resources to appear. ### Configuration ```typescript +import { NO_TIMEOUT } from '@message-queue-toolkit/core' + const consumer = new MySnsSqsConsumer(dependencies, { locatorConfig: { topicArn: 'arn:aws:sns:...', queueUrl: 'https://sqs...', - subscriptionArn: '...' - }, - // Enable eventual consistency mode - resourceAvailabilityConfig: { - enabled: true, // Enable polling for resource availability - pollingIntervalMs: 5000, // Check every 5 seconds (default: 5000) - timeoutMs: 300000, // Fail after 5 minutes (optional, default: no timeout) + subscriptionArn: '...', + // Enable eventual consistency mode + startupResourcePolling: { + enabled: true, // Enable polling for resource availability + pollingIntervalMs: 5000, // Check every 5 seconds (default: 5000) + timeoutMs: 300000, // Fail after 5 minutes (required) + } } }) ``` @@ -280,42 +282,61 @@ const consumer = new MySnsSqsConsumer(dependencies, { | Option | Type | Default | Description | |--------|------|---------|-------------| -| `enabled` | `boolean` | `false` | Enable eventual consistency mode | +| `enabled` | `boolean` | - | Must be set to `true` to enable polling | | `pollingIntervalMs` | `number` | `5000` | Interval between availability checks (ms) | -| `timeoutMs` | `number` | `undefined` | Maximum wait time before throwing `ResourceAvailabilityTimeoutError`. If not set, polls indefinitely | +| `timeoutMs` | `number \| NO_TIMEOUT` | - (required) | Maximum wait time before throwing `StartupResourcePollingTimeoutError`. Use `NO_TIMEOUT` to poll indefinitely | ### Environment-Specific Configuration ```typescript +import { NO_TIMEOUT } from '@message-queue-toolkit/core' + +// Production - with timeout +{ + locatorConfig: { + queueUrl: '...', + startupResourcePolling: { + enabled: true, + timeoutMs: 5 * 60 * 1000, // 5 minutes + } + } +} + // Development/Staging - poll indefinitely { - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 5000, + locatorConfig: { + queueUrl: '...', + startupResourcePolling: { + enabled: true, + timeoutMs: NO_TIMEOUT, + } } } -// Production - poll with timeout to catch misconfigurations +// Custom timeout and interval { - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 10000, - timeoutMs: 5 * 60 * 1000, // 5 minutes + locatorConfig: { + queueUrl: '...', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 10000, + timeoutMs: 10 * 60 * 1000, // 10 minutes + } } } ``` ### Error Handling -When `timeoutMs` is configured and the resource doesn't become available within the specified time, a `ResourceAvailabilityTimeoutError` is thrown: +When `timeoutMs` is configured and the resource doesn't become available within the specified time, a `StartupResourcePollingTimeoutError` is thrown: ```typescript -import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core' +import { StartupResourcePollingTimeoutError } from '@message-queue-toolkit/core' try { await consumer.init() } catch (error) { - if (error instanceof ResourceAvailabilityTimeoutError) { + if (error instanceof StartupResourcePollingTimeoutError) { console.error(`Resource ${error.resourceName} not available after ${error.timeoutMs}ms`) } } diff --git a/packages/core/lib/index.ts b/packages/core/lib/index.ts index bc16151a..406712da 100644 --- a/packages/core/lib/index.ts +++ b/packages/core/lib/index.ts @@ -106,11 +106,11 @@ export { isShallowSubset, objectMatches } from './utils/matchUtils.ts' export { type ParseMessageResult, parseMessage } from './utils/parseUtils.ts' export { objectToBuffer } from './utils/queueUtils.ts' export { - isResourceAvailabilityWaitingEnabled, - type ResourceAvailabilityCheckResult, - ResourceAvailabilityTimeoutError, + isStartupResourcePollingEnabled, + type StartupResourcePollingCheckResult, + StartupResourcePollingTimeoutError, type WaitForResourceOptions, waitForResource, -} from './utils/resourceAvailabilityUtils.ts' +} from './utils/startupResourcePollingUtils.ts' export { toDatePreprocessor } from './utils/toDateProcessor.ts' export { waitAndRetry } from './utils/waitUtils.ts' diff --git a/packages/core/lib/queues/AbstractQueueService.ts b/packages/core/lib/queues/AbstractQueueService.ts index 04a88b6b..4a9e6a39 100644 --- a/packages/core/lib/queues/AbstractQueueService.ts +++ b/packages/core/lib/queues/AbstractQueueService.ts @@ -46,7 +46,6 @@ import type { ProcessedMessageMetadata, QueueDependencies, QueueOptions, - ResourceAvailabilityConfig, } from '../types/queueOptionsTypes.ts' import { isRetryDateExceeded } from '../utils/dateUtils.ts' import { streamWithKnownSizeToString } from '../utils/streamUtils.ts' @@ -126,7 +125,6 @@ export abstract class AbstractQueueService< protected readonly creationConfig?: QueueConfiguration protected readonly locatorConfig?: QueueLocatorType protected readonly deletionConfig?: DeletionConfig - protected readonly resourceAvailabilityConfig?: ResourceAvailabilityConfig protected readonly payloadStoreConfig?: | MakeRequired | MakeRequired @@ -162,7 +160,6 @@ export abstract class AbstractQueueService< this.creationConfig = options.creationConfig this.locatorConfig = options.locatorConfig this.deletionConfig = options.deletionConfig - this.resourceAvailabilityConfig = options.resourceAvailabilityConfig this.payloadStoreConfig = options.payloadStoreConfig ? { serializer: jsonStreamStringifySerializer, diff --git a/packages/core/lib/types/queueOptionsTypes.ts b/packages/core/lib/types/queueOptionsTypes.ts index 839b990e..6570d0ff 100644 --- a/packages/core/lib/types/queueOptionsTypes.ts +++ b/packages/core/lib/types/queueOptionsTypes.ts @@ -118,13 +118,6 @@ export type CommonQueueOptions = { deletionConfig?: DeletionConfig payloadStoreConfig?: PayloadStoreConfig messageDeduplicationConfig?: MessageDeduplicationConfig - /** - * Configuration for eventual consistency mode. - * When enabled, the consumer will poll for the topic/queue to become available - * instead of failing immediately when using locatorConfig. - * This is useful for handling cross-service dependencies during deployment. - */ - resourceAvailabilityConfig?: ResourceAvailabilityConfig } export type CommonCreationConfigType = { @@ -138,57 +131,82 @@ export type DeletionConfig = { } /** - * Configuration for eventual consistency mode when resources may not exist at startup. + * Symbol to indicate no timeout - polling will continue indefinitely. + * Use this for dev/staging environments where you want to wait indefinitely for resources. + * + * @example + * { + * locatorConfig: { + * queueUrl: '...', + * startupResourcePolling: { + * enabled: true, + * timeoutMs: NO_TIMEOUT, + * } + * } + * } + */ +export const NO_TIMEOUT = Symbol('NO_TIMEOUT') + +/** + * Configuration for startup resource polling mode when resources may not exist at startup. * * This is useful in scenarios where services have cross-dependencies: * - Service A needs to subscribe to Service B's topic * - Service B needs to subscribe to Service A's topic * - Neither can deploy first without the other's topic existing * - * When `resourceAvailabilityConfig` is provided, the consumer will poll for the - * topic/queue to become available instead of failing immediately. + * When `startupResourcePolling` is provided with `enabled: true` inside `locatorConfig`, + * the consumer will poll for the topic/queue to become available instead of failing immediately. * * @example - * // Development/staging - poll indefinitely + * // Enable with 5 minute timeout * { - * resourceAvailabilityConfig: { - * pollingIntervalMs: 5000, + * locatorConfig: { + * queueUrl: '...', + * startupResourcePolling: { + * enabled: true, + * timeoutMs: 5 * 60 * 1000, + * } * } * } * * @example - * // Production - poll with timeout to catch misconfigurations + * // Poll indefinitely (useful for dev/staging environments) * { - * resourceAvailabilityConfig: { - * timeoutMs: 5 * 60 * 1000, // 5 minutes - * pollingIntervalMs: 10000, + * locatorConfig: { + * queueUrl: '...', + * startupResourcePolling: { + * enabled: true, + * timeoutMs: NO_TIMEOUT, + * } * } * } * * @example - * // Temporarily disable without removing config + * // Custom timeout and interval * { - * resourceAvailabilityConfig: { - * enabled: false, - * timeoutMs: 5 * 60 * 1000, + * locatorConfig: { + * queueUrl: '...', + * startupResourcePolling: { + * enabled: true, + * timeoutMs: 10 * 60 * 1000, // 10 minutes + * pollingIntervalMs: 10000, + * } * } * } */ -export type ResourceAvailabilityConfig = { +export type StartupResourcePollingConfig = { /** * Controls whether polling is enabled. - * Default: true (when resourceAvailabilityConfig is provided) - * Set to false to temporarily disable polling without removing the config. + * Must be set to true to enable polling. */ enabled?: boolean /** * Maximum time in milliseconds to wait for the resource to become available. - * If not set or set to 0, will poll indefinitely (useful for dev/staging environments). - * For production, it's recommended to set a reasonable timeout (e.g., 5 minutes). - * Default: undefined (no timeout - poll indefinitely) + * Use `NO_TIMEOUT` to disable timeout and poll indefinitely. */ - timeoutMs?: number + timeoutMs: number | typeof NO_TIMEOUT /** * Interval in milliseconds between polling attempts. @@ -197,6 +215,19 @@ export type ResourceAvailabilityConfig = { pollingIntervalMs?: number } +/** + * Base type for queue locator configurations that includes startup resource polling. + * Protocol-specific locator types should extend this. + */ +export type BaseQueueLocatorType = { + /** + * Configuration for startup resource polling mode. + * When enabled, the consumer will poll for the resource to become available + * instead of failing immediately. + */ + startupResourcePolling?: StartupResourcePollingConfig +} + type NewQueueOptions = { creationConfig?: CreationConfigType } diff --git a/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts b/packages/core/lib/utils/startupResourcePollingUtils.spec.ts similarity index 64% rename from packages/core/lib/utils/resourceAvailabilityUtils.spec.ts rename to packages/core/lib/utils/startupResourcePollingUtils.spec.ts index 527ac7c8..17edd9fa 100644 --- a/packages/core/lib/utils/resourceAvailabilityUtils.spec.ts +++ b/packages/core/lib/utils/startupResourcePollingUtils.spec.ts @@ -1,28 +1,30 @@ import { describe, expect, it, vi } from 'vitest' +import { NO_TIMEOUT } from '../types/queueOptionsTypes.ts' import { - isResourceAvailabilityWaitingEnabled, - ResourceAvailabilityTimeoutError, + isStartupResourcePollingEnabled, + StartupResourcePollingTimeoutError, waitForResource, -} from './resourceAvailabilityUtils.ts' +} from './startupResourcePollingUtils.ts' -describe('resourceAvailabilityUtils', () => { - describe('isResourceAvailabilityWaitingEnabled', () => { +describe('startupResourcePollingUtils', () => { + describe('isStartupResourcePollingEnabled', () => { it('returns true when enabled is true', () => { - expect(isResourceAvailabilityWaitingEnabled({ enabled: true })).toBe(true) + expect(isStartupResourcePollingEnabled({ enabled: true, timeoutMs: 5000 })).toBe(true) }) - it('returns true when enabled is not specified (defaults to true)', () => { - expect(isResourceAvailabilityWaitingEnabled({})).toBe(true) - expect(isResourceAvailabilityWaitingEnabled({ pollingIntervalMs: 1000 })).toBe(true) - expect(isResourceAvailabilityWaitingEnabled({ timeoutMs: 5000 })).toBe(true) + it('returns false when enabled is not specified', () => { + expect(isStartupResourcePollingEnabled({ timeoutMs: 5000 })).toBe(false) + expect(isStartupResourcePollingEnabled({ pollingIntervalMs: 1000, timeoutMs: 5000 })).toBe( + false, + ) }) it('returns false when enabled is false', () => { - expect(isResourceAvailabilityWaitingEnabled({ enabled: false })).toBe(false) + expect(isStartupResourcePollingEnabled({ enabled: false, timeoutMs: 5000 })).toBe(false) }) it('returns false when config is undefined', () => { - expect(isResourceAvailabilityWaitingEnabled(undefined)).toBe(false) + expect(isStartupResourcePollingEnabled(undefined)).toBe(false) }) }) @@ -31,7 +33,7 @@ describe('resourceAvailabilityUtils', () => { const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' }) const result = await waitForResource({ - config: { enabled: true, pollingIntervalMs: 100 }, + config: { enabled: true, pollingIntervalMs: 100, timeoutMs: 5000 }, checkFn, resourceName: 'test-resource', }) @@ -51,7 +53,7 @@ describe('resourceAvailabilityUtils', () => { }) const result = await waitForResource({ - config: { enabled: true, pollingIntervalMs: 10 }, + config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 5000 }, checkFn, resourceName: 'test-resource', }) @@ -60,7 +62,7 @@ describe('resourceAvailabilityUtils', () => { expect(checkFn).toHaveBeenCalledTimes(3) }) - it('throws ResourceAvailabilityTimeoutError when timeout is reached', async () => { + it('throws StartupResourcePollingTimeoutError when timeout is reached', async () => { const checkFn = vi.fn().mockResolvedValue({ isAvailable: false }) await expect( @@ -69,7 +71,7 @@ describe('resourceAvailabilityUtils', () => { checkFn, resourceName: 'test-resource', }), - ).rejects.toThrow(ResourceAvailabilityTimeoutError) + ).rejects.toThrow(StartupResourcePollingTimeoutError) // Should have made at least a few attempts expect(checkFn.mock.calls.length).toBeGreaterThan(0) @@ -80,7 +82,7 @@ describe('resourceAvailabilityUtils', () => { await expect( waitForResource({ - config: { enabled: true, pollingIntervalMs: 10 }, + config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 5000 }, checkFn, resourceName: 'test-resource', }), @@ -93,7 +95,7 @@ describe('resourceAvailabilityUtils', () => { const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' }) const result = await waitForResource({ - config: { enabled: true }, + config: { enabled: true, timeoutMs: 5000 }, checkFn, resourceName: 'test-resource', }) @@ -110,7 +112,7 @@ describe('resourceAvailabilityUtils', () => { const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' }) await waitForResource({ - config: { enabled: true, pollingIntervalMs: 10 }, + config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 5000 }, checkFn, resourceName: 'test-resource', // @ts-expect-error - partial logger for testing @@ -131,27 +133,7 @@ describe('resourceAvailabilityUtils', () => { ) }) - it('polls indefinitely when timeoutMs is not set', async () => { - let callCount = 0 - const checkFn = vi.fn().mockImplementation(() => { - callCount++ - if (callCount < 10) { - return Promise.resolve({ isAvailable: false }) - } - return Promise.resolve({ isAvailable: true, result: 'test-result' }) - }) - - const result = await waitForResource({ - config: { enabled: true, pollingIntervalMs: 1 }, - checkFn, - resourceName: 'test-resource', - }) - - expect(result).toBe('test-result') - expect(checkFn).toHaveBeenCalledTimes(10) - }) - - it('polls indefinitely when timeoutMs is 0', async () => { + it('polls indefinitely when NO_TIMEOUT is used', async () => { let callCount = 0 const checkFn = vi.fn().mockImplementation(() => { callCount++ @@ -162,7 +144,11 @@ describe('resourceAvailabilityUtils', () => { }) const result = await waitForResource({ - config: { enabled: true, pollingIntervalMs: 1, timeoutMs: 0 }, + config: { + enabled: true, + pollingIntervalMs: 1, + timeoutMs: NO_TIMEOUT, + }, checkFn, resourceName: 'test-resource', }) @@ -172,15 +158,15 @@ describe('resourceAvailabilityUtils', () => { }) }) - describe('ResourceAvailabilityTimeoutError', () => { + describe('StartupResourcePollingTimeoutError', () => { it('includes resource name and timeout in message', () => { - const error = new ResourceAvailabilityTimeoutError('my-queue', 5000) + const error = new StartupResourcePollingTimeoutError('my-queue', 5000) expect(error.message).toContain('my-queue') expect(error.message).toContain('5000') expect(error.resourceName).toBe('my-queue') expect(error.timeoutMs).toBe(5000) - expect(error.name).toBe('ResourceAvailabilityTimeoutError') + expect(error.name).toBe('StartupResourcePollingTimeoutError') }) }) }) diff --git a/packages/core/lib/utils/resourceAvailabilityUtils.ts b/packages/core/lib/utils/startupResourcePollingUtils.ts similarity index 75% rename from packages/core/lib/utils/resourceAvailabilityUtils.ts rename to packages/core/lib/utils/startupResourcePollingUtils.ts index 602564e9..8726613d 100644 --- a/packages/core/lib/utils/resourceAvailabilityUtils.ts +++ b/packages/core/lib/utils/startupResourcePollingUtils.ts @@ -1,10 +1,10 @@ import { setTimeout } from 'node:timers/promises' import type { CommonLogger } from '@lokalise/node-core' -import type { ResourceAvailabilityConfig } from '../types/queueOptionsTypes.ts' +import { NO_TIMEOUT, type StartupResourcePollingConfig } from '../types/queueOptionsTypes.ts' const DEFAULT_POLLING_INTERVAL_MS = 5000 -export type ResourceAvailabilityCheckResult = +export type StartupResourcePollingCheckResult = | { isAvailable: true result: T @@ -15,9 +15,9 @@ export type ResourceAvailabilityCheckResult = export type WaitForResourceOptions = { /** - * Resource availability configuration + * Startup resource polling configuration */ - config: ResourceAvailabilityConfig + config: StartupResourcePollingConfig /** * Function that checks if the resource is available. @@ -25,7 +25,7 @@ export type WaitForResourceOptions = { * or { isAvailable: false } when resource doesn't exist. * Should throw on unexpected errors. */ - checkFn: () => Promise> + checkFn: () => Promise> /** * Human-readable name of the resource for logging @@ -38,7 +38,7 @@ export type WaitForResourceOptions = { logger?: CommonLogger } -export class ResourceAvailabilityTimeoutError extends Error { +export class StartupResourcePollingTimeoutError extends Error { public readonly resourceName: string public readonly timeoutMs: number @@ -47,7 +47,7 @@ export class ResourceAvailabilityTimeoutError extends Error { `Timeout waiting for resource "${resourceName}" to become available after ${timeoutMs}ms. ` + 'The resource may not exist or there may be a configuration issue.', ) - this.name = 'ResourceAvailabilityTimeoutError' + this.name = 'StartupResourcePollingTimeoutError' this.resourceName = resourceName this.timeoutMs = timeoutMs } @@ -55,13 +55,13 @@ export class ResourceAvailabilityTimeoutError extends Error { function checkTimeoutExceeded( hasTimeout: boolean, - timeoutMs: number | undefined, + timeoutMs: number, startTime: number, resourceName: string, attemptCount: number, logger?: CommonLogger, ): void { - if (!hasTimeout || timeoutMs === undefined) return + if (!hasTimeout) return const elapsedMs = Date.now() - startTime if (elapsedMs >= timeoutMs) { @@ -72,7 +72,7 @@ function checkTimeoutExceeded( attemptCount, elapsedMs, }) - throw new ResourceAvailabilityTimeoutError(resourceName, timeoutMs) + throw new StartupResourcePollingTimeoutError(resourceName, timeoutMs) } } @@ -93,17 +93,17 @@ function logResourceAvailable( /** * Waits for a resource to become available by polling. - * This is used for eventual consistency mode where resources may not exist at startup. + * This is used for startup resource polling mode where resources may not exist at startup. * * @param options - Configuration and check function * @returns The result from the check function when resource becomes available - * @throws ResourceAvailabilityTimeoutError if timeout is reached + * @throws StartupResourcePollingTimeoutError if timeout is reached */ export async function waitForResource(options: WaitForResourceOptions): Promise { const { config, checkFn, resourceName, logger } = options const pollingIntervalMs = config.pollingIntervalMs ?? DEFAULT_POLLING_INTERVAL_MS - const timeoutMs = config.timeoutMs - const hasTimeout = timeoutMs !== undefined && timeoutMs > 0 + const hasTimeout = config.timeoutMs !== NO_TIMEOUT + const timeoutMs = hasTimeout ? (config.timeoutMs as number) : 0 const startTime = Date.now() let attemptCount = 0 @@ -112,7 +112,7 @@ export async function waitForResource(options: WaitForResourceOptions): Pr message: `Waiting for resource "${resourceName}" to become available`, resourceName, pollingIntervalMs, - timeoutMs: timeoutMs ?? 'indefinite', + timeoutMs: hasTimeout ? timeoutMs : 'NO_TIMEOUT', }) while (true) { @@ -156,11 +156,11 @@ export async function waitForResource(options: WaitForResourceOptions): Pr } /** - * Helper to check if resource availability waiting is enabled. - * Returns true when config is provided and enabled is not explicitly false. + * Helper to check if startup resource polling is enabled. + * Returns true only when config is provided and enabled is explicitly true. */ -export function isResourceAvailabilityWaitingEnabled( - config?: ResourceAvailabilityConfig, -): config is ResourceAvailabilityConfig { - return config != null && config.enabled !== false +export function isStartupResourcePollingEnabled( + config?: StartupResourcePollingConfig, +): config is StartupResourcePollingConfig { + return config?.enabled === true } diff --git a/packages/sns/lib/sns/AbstractSnsService.ts b/packages/sns/lib/sns/AbstractSnsService.ts index 638aa9d8..e6b0d08b 100644 --- a/packages/sns/lib/sns/AbstractSnsService.ts +++ b/packages/sns/lib/sns/AbstractSnsService.ts @@ -1,6 +1,10 @@ import type { CreateTopicCommandInput, SNSClient, Tag } from '@aws-sdk/client-sns' import type { STSClient } from '@aws-sdk/client-sts' -import type { QueueDependencies, QueueOptions } from '@message-queue-toolkit/core' +import type { + BaseQueueLocatorType, + QueueDependencies, + QueueOptions, +} from '@message-queue-toolkit/core' import { AbstractQueueService } from '@message-queue-toolkit/core' import type { SNS_MESSAGE_BODY_TYPE } from '../types/MessageTypes.ts' import { deleteSns, initSns } from '../utils/snsInitter.ts' @@ -40,7 +44,7 @@ export type SNSCreationConfig = { updateAttributesIfExists?: boolean } & ExtraSNSCreationParams -export type SNSTopicLocatorType = { +export type SNSTopicLocatorType = BaseQueueLocatorType & { topicArn?: string topicName?: string } @@ -82,6 +86,7 @@ export abstract class AbstractSnsService< this.stsClient, this.locatorConfig, this.creationConfig, + { logger: this.logger }, ) this.topicArn = initResult.topicArn this.isInitted = true diff --git a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts index 01e7983b..7aa6a81f 100644 --- a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts +++ b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts @@ -102,10 +102,7 @@ export abstract class AbstractSnsSqsConsumer< this.locatorConfig, this.creationConfig, this.subscriptionConfig, - { - logger: this.logger, - resourceAvailabilityConfig: this.resourceAvailabilityConfig, - }, + { logger: this.logger }, ) this.queueName = initSnsSqsResult.queueName this.queueUrl = initSnsSqsResult.queueUrl diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index abdea964..50b7e4e1 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -2,14 +2,10 @@ import type { CreateTopicCommandInput, SNSClient } from '@aws-sdk/client-sns' import type { CreateQueueCommandInput, SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' import type { Either } from '@lokalise/node-core' -import type { - DeletionConfig, - ExtraParams, - ResourceAvailabilityConfig, -} from '@message-queue-toolkit/core' +import type { DeletionConfig, ExtraParams } from '@message-queue-toolkit/core' import { isProduction, - isResourceAvailabilityWaitingEnabled, + isStartupResourcePollingEnabled, waitForResource, } from '@message-queue-toolkit/core' import { @@ -26,13 +22,9 @@ import { subscribeToTopic } from './snsSubscriber.ts' import { assertTopic, deleteSubscription, deleteTopic, getTopicAttributes } from './snsUtils.ts' import { buildTopicArn } from './stsUtils.ts' -export type InitSnsSqsExtraParams = ExtraParams & { - resourceAvailabilityConfig?: ResourceAvailabilityConfig -} +export type InitSnsSqsExtraParams = ExtraParams -export type InitSnsExtraParams = ExtraParams & { - resourceAvailabilityConfig?: ResourceAvailabilityConfig -} +export type InitSnsExtraParams = ExtraParams // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: fixme export async function initSnsSqs( @@ -105,13 +97,13 @@ export async function initSnsSqs( const subscriptionTopicArn = locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? '')) - const resourceAvailabilityConfig = extraParams?.resourceAvailabilityConfig + const startupResourcePolling = locatorConfig.startupResourcePolling - // If resource availability waiting is enabled, poll for resources to become available - if (isResourceAvailabilityWaitingEnabled(resourceAvailabilityConfig)) { + // If startup resource polling is enabled, poll for resources to become available + if (isStartupResourcePollingEnabled(startupResourcePolling)) { // Wait for topic to become available await waitForResource({ - config: resourceAvailabilityConfig, + config: startupResourcePolling, resourceName: `SNS topic ${subscriptionTopicArn}`, logger: extraParams?.logger, checkFn: async () => { @@ -125,7 +117,7 @@ export async function initSnsSqs( // Wait for queue to become available await waitForResource({ - config: resourceAvailabilityConfig, + config: startupResourcePolling, resourceName: `SQS queue ${queueUrl}`, logger: extraParams?.logger, checkFn: async () => { @@ -268,12 +260,12 @@ export async function initSns( const topicArn = locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? '')) - const resourceAvailabilityConfig = extraParams?.resourceAvailabilityConfig + const startupResourcePolling = locatorConfig.startupResourcePolling - // If resource availability waiting is enabled, poll for topic to become available - if (isResourceAvailabilityWaitingEnabled(resourceAvailabilityConfig)) { + // If startup resource polling is enabled, poll for topic to become available + if (isStartupResourcePollingEnabled(startupResourcePolling)) { await waitForResource({ - config: resourceAvailabilityConfig, + config: startupResourcePolling, resourceName: `SNS topic ${topicArn}`, logger: extraParams?.logger, checkFn: async () => { diff --git a/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts b/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts similarity index 77% rename from packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts rename to packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts index b55be3ab..00824c09 100644 --- a/packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts +++ b/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts @@ -4,7 +4,7 @@ import type { SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' import { MessageHandlerConfigBuilder, - ResourceAvailabilityTimeoutError, + StartupResourcePollingTimeoutError, } from '@message-queue-toolkit/core' import { assertQueue, deleteQueue } from '@message-queue-toolkit/sqs' import type { AwilixContainer } from 'awilix' @@ -24,11 +24,11 @@ import { type TestConsumerOptions = Pick< SNSSQSConsumerOptions, - 'locatorConfig' | 'creationConfig' | 'resourceAvailabilityConfig' + 'locatorConfig' | 'creationConfig' > -// Simple consumer for testing resource availability -class TestResourceAvailabilityConsumer extends AbstractSnsSqsConsumer< +// Simple consumer for testing startup resource polling +class TestStartupResourcePollingConsumer extends AbstractSnsSqsConsumer< PERMISSIONS_ADD_MESSAGE_TYPE, undefined, undefined @@ -58,9 +58,9 @@ class TestResourceAvailabilityConsumer extends AbstractSnsSqsConsumer< } } -describe('SnsSqsPermissionConsumer - resourceAvailabilityConfig', () => { - const queueName = 'resource-availability-test-queue' - const topicName = 'resource-availability-test-topic' +describe('SnsSqsPermissionConsumer - startupResourcePollingConfig', () => { + const queueName = 'startup-resource-polling-test-queue' + const topicName = 'startup-resource-polling-test-topic' const queueUrl = `http://sqs.eu-west-1.localstack:4566/000000000000/${queueName}` let diContainer: AwilixContainer @@ -91,26 +91,26 @@ describe('SnsSqsPermissionConsumer - resourceAvailabilityConfig', () => { await diContainer.dispose() }) - describe('when resourceAvailabilityConfig is enabled', () => { + describe('when startupResourcePolling is enabled', () => { it('waits for topic to become available and initializes successfully', async () => { // Create queue first, but not the topic await assertQueue(sqsClient, { QueueName: queueName }) - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { topicName, queueUrl, subscriptionArn: 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + }, }, creationConfig: { queue: { QueueName: queueName }, }, - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 100, - timeoutMs: 5000, - }, }) // Start init in background @@ -131,17 +131,17 @@ describe('SnsSqsPermissionConsumer - resourceAvailabilityConfig', () => { // Create topic first, but not the queue const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { topicArn, queueUrl, subscriptionArn: 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', - }, - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 100, - timeoutMs: 5000, + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + }, }, }) @@ -159,48 +159,49 @@ describe('SnsSqsPermissionConsumer - resourceAvailabilityConfig', () => { expect(consumer.subscriptionProps.queueUrl).toBe(queueUrl) }) - it('throws ResourceAvailabilityTimeoutError when timeout is reached', async () => { + it('throws StartupResourcePollingTimeoutError when timeout is reached', async () => { // Create queue but not topic await assertQueue(sqsClient, { QueueName: queueName }) - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { topicName, queueUrl, subscriptionArn: 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: 200, // Short timeout + }, }, creationConfig: { queue: { QueueName: queueName }, }, - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 50, - timeoutMs: 200, // Short timeout - }, }) // Should throw timeout error since topic never appears - await expect(consumer.init()).rejects.toThrow(ResourceAvailabilityTimeoutError) + await expect(consumer.init()).rejects.toThrow(StartupResourcePollingTimeoutError) }) }) - describe('when resourceAvailabilityConfig is disabled', () => { + describe('when startupResourcePolling is disabled', () => { it('throws immediately when topic does not exist', async () => { // Create queue but not topic await assertQueue(sqsClient, { QueueName: queueName }) const topicArn = 'arn:aws:sns:eu-west-1:000000000000:non-existent-topic' - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { topicArn, queueUrl, subscriptionArn: 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', - }, - resourceAvailabilityConfig: { - enabled: false, + startupResourcePolling: { + enabled: false, + timeoutMs: 5000, + }, }, }) @@ -212,15 +213,16 @@ describe('SnsSqsPermissionConsumer - resourceAvailabilityConfig', () => { // Create topic but not queue const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { topicArn, queueUrl, subscriptionArn: 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', - }, - resourceAvailabilityConfig: { - enabled: false, + startupResourcePolling: { + enabled: false, + timeoutMs: 5000, + }, }, }) diff --git a/packages/sqs/lib/sqs/AbstractSqsService.ts b/packages/sqs/lib/sqs/AbstractSqsService.ts index 6990d99e..a8b160be 100644 --- a/packages/sqs/lib/sqs/AbstractSqsService.ts +++ b/packages/sqs/lib/sqs/AbstractSqsService.ts @@ -1,5 +1,9 @@ import type { CreateQueueRequest, SQSClient } from '@aws-sdk/client-sqs' -import type { QueueDependencies, QueueOptions } from '@message-queue-toolkit/core' +import type { + BaseQueueLocatorType, + QueueDependencies, + QueueOptions, +} from '@message-queue-toolkit/core' import { AbstractQueueService } from '@message-queue-toolkit/core' import type { SQSMessage } from '../types/MessageTypes.ts' import { deleteSqs, initSqs } from '../utils/sqsInitter.ts' @@ -38,15 +42,17 @@ export type SQSCreationConfig = { policyConfig?: SQSPolicyConfig } & ExtraSQSCreationParams -export type SQSQueueLocatorType = - | { - queueUrl: string - queueName?: never - } - | { - queueName: string - queueUrl?: never - } +export type SQSQueueLocatorType = BaseQueueLocatorType & + ( + | { + queueUrl: string + queueName?: never + } + | { + queueName: string + queueUrl?: never + } + ) export type SQSQueueConfig = { /** @@ -108,10 +114,7 @@ export abstract class AbstractSqsService< this.locatorConfig, this.creationConfig, this.isFifoQueue, - { - resourceAvailabilityConfig: this.resourceAvailabilityConfig, - logger: this.logger, - }, + { logger: this.logger }, ) this.queueName = queueName this.queueUrl = queueUrl diff --git a/packages/sqs/lib/utils/sqsInitter.ts b/packages/sqs/lib/utils/sqsInitter.ts index f0e2ba30..cff257c5 100644 --- a/packages/sqs/lib/utils/sqsInitter.ts +++ b/packages/sqs/lib/utils/sqsInitter.ts @@ -1,10 +1,10 @@ import type { QueueAttributeName, SQSClient } from '@aws-sdk/client-sqs' import { SetQueueAttributesCommand, TagQueueCommand } from '@aws-sdk/client-sqs' import type { CommonLogger } from '@lokalise/node-core' -import type { DeletionConfig, ResourceAvailabilityConfig } from '@message-queue-toolkit/core' +import type { DeletionConfig } from '@message-queue-toolkit/core' import { isProduction, - isResourceAvailabilityWaitingEnabled, + isStartupResourcePollingEnabled, waitForResource, } from '@message-queue-toolkit/core' @@ -19,7 +19,6 @@ import { } from './sqsUtils.ts' export type InitSqsExtraParams = { - resourceAvailabilityConfig?: ResourceAvailabilityConfig logger?: CommonLogger } @@ -84,14 +83,14 @@ export async function initSqs( if (locatorConfig) { const queueUrl = await resolveQueueUrlFromLocatorConfig(sqsClient, locatorConfig) - const resourceAvailabilityConfig = extraParams?.resourceAvailabilityConfig + const startupResourcePolling = locatorConfig.startupResourcePolling let queueArn: string | undefined - // If resource availability waiting is enabled, poll for queue to become available - if (isResourceAvailabilityWaitingEnabled(resourceAvailabilityConfig)) { + // If startup resource polling is enabled, poll for queue to become available + if (isStartupResourcePollingEnabled(startupResourcePolling)) { const result = await waitForResource({ - config: resourceAvailabilityConfig, + config: startupResourcePolling, resourceName: `SQS queue ${queueUrl}`, logger: extraParams?.logger, checkFn: async () => { diff --git a/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts b/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts similarity index 67% rename from packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts rename to packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts index 3ed6e84c..e52b415c 100644 --- a/packages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts +++ b/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts @@ -2,7 +2,8 @@ import { setTimeout } from 'node:timers/promises' import type { SQSClient } from '@aws-sdk/client-sqs' import { MessageHandlerConfigBuilder, - ResourceAvailabilityTimeoutError, + NO_TIMEOUT, + StartupResourcePollingTimeoutError, } from '@message-queue-toolkit/core' import type { AwilixContainer } from 'awilix' import { afterEach, beforeEach, describe, expect, it } from 'vitest' @@ -21,11 +22,11 @@ import { type TestConsumerOptions = Pick< SQSConsumerOptions, - 'locatorConfig' | 'creationConfig' | 'resourceAvailabilityConfig' + 'locatorConfig' | 'creationConfig' > -// Simple consumer for testing resource availability -class TestResourceAvailabilityConsumer extends AbstractSqsConsumer< +// Simple consumer for testing startup resource polling +class TestStartupResourcePollingConsumer extends AbstractSqsConsumer< PERMISSIONS_ADD_MESSAGE_TYPE, undefined, undefined @@ -53,8 +54,8 @@ class TestResourceAvailabilityConsumer extends AbstractSqsConsumer< } } -describe('SqsPermissionConsumer - resourceAvailabilityConfig', () => { - const queueName = 'resource-availability-test-queue' +describe('SqsPermissionConsumer - startupResourcePollingConfig', () => { + const queueName = 'startup-resource-polling-test-queue' const queueUrl = `http://sqs.eu-west-1.localstack:4566/000000000000/${queueName}` let diContainer: AwilixContainer @@ -72,16 +73,16 @@ describe('SqsPermissionConsumer - resourceAvailabilityConfig', () => { await diContainer.dispose() }) - describe('when resourceAvailabilityConfig is enabled', () => { + describe('when startupResourcePolling is enabled', () => { it('waits for queue to become available and initializes successfully', async () => { - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { queueUrl, - }, - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 100, - timeoutMs: 5000, + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + }, }, }) @@ -99,31 +100,31 @@ describe('SqsPermissionConsumer - resourceAvailabilityConfig', () => { expect(consumer.queueProps.name).toBe(queueName) }) - it('throws ResourceAvailabilityTimeoutError when timeout is reached', async () => { - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + it('throws StartupResourcePollingTimeoutError when timeout is reached', async () => { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { queueUrl, - }, - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 50, - timeoutMs: 200, // Short timeout + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: 200, // Short timeout + }, }, }) // Should throw timeout error since queue never appears - await expect(consumer.init()).rejects.toThrow(ResourceAvailabilityTimeoutError) + await expect(consumer.init()).rejects.toThrow(StartupResourcePollingTimeoutError) }) - it('polls indefinitely when timeoutMs is not set', async () => { - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + it('polls indefinitely when NO_TIMEOUT is used', async () => { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { queueUrl, - }, - resourceAvailabilityConfig: { - enabled: true, - pollingIntervalMs: 50, - // No timeout - polls indefinitely + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: NO_TIMEOUT, + }, }, }) @@ -141,14 +142,15 @@ describe('SqsPermissionConsumer - resourceAvailabilityConfig', () => { }) }) - describe('when resourceAvailabilityConfig is disabled', () => { + describe('when startupResourcePolling is disabled', () => { it('throws immediately when queue does not exist', async () => { - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { queueUrl, - }, - resourceAvailabilityConfig: { - enabled: false, + startupResourcePolling: { + enabled: false, + timeoutMs: 5000, + }, }, }) @@ -157,9 +159,9 @@ describe('SqsPermissionConsumer - resourceAvailabilityConfig', () => { }) }) - describe('when resourceAvailabilityConfig is not provided', () => { + describe('when startupResourcePolling is not provided', () => { it('throws immediately when queue does not exist (default behavior)', async () => { - const consumer = new TestResourceAvailabilityConsumer(diContainer.cradle, { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { locatorConfig: { queueUrl, }, From cf023cc08e51099f5c45a117ed5a696a162a365b Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 14:42:22 +0200 Subject: [PATCH 06/11] Add missing test --- ...ionConsumer.startupResourcePolling.spec.ts | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts b/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts index 00824c09..3feb030f 100644 --- a/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts +++ b/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts @@ -4,6 +4,7 @@ import type { SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' import { MessageHandlerConfigBuilder, + NO_TIMEOUT, StartupResourcePollingTimeoutError, } from '@message-queue-toolkit/core' import { assertQueue, deleteQueue } from '@message-queue-toolkit/sqs' @@ -183,6 +184,41 @@ describe('SnsSqsPermissionConsumer - startupResourcePollingConfig', () => { // Should throw timeout error since topic never appears await expect(consumer.init()).rejects.toThrow(StartupResourcePollingTimeoutError) }) + + it('polls indefinitely when NO_TIMEOUT is used', async () => { + // Create queue first, but not the topic + await assertQueue(sqsClient, { QueueName: queueName }) + + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { + locatorConfig: { + topicName, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: NO_TIMEOUT, + }, + }, + creationConfig: { + queue: { QueueName: queueName }, + }, + }) + + // Start init in background + const initPromise = consumer.init() + + // Wait a bit then create the topic + await setTimeout(500) + const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) + + // Init should complete successfully + await initPromise + + expect(consumer.subscriptionProps.topicArn).toBe(topicArn) + expect(consumer.subscriptionProps.queueName).toBe(queueName) + }) }) describe('when startupResourcePolling is disabled', () => { From bbed22666af8953792c56dc1957a0734fc3a1f39 Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 15:44:40 +0200 Subject: [PATCH 07/11] Add non-blocking mode --- README.md | 16 +- packages/core/lib/types/MessageQueueTypes.ts | 7 +- packages/core/lib/types/queueOptionsTypes.ts | 22 ++ .../utils/startupResourcePollingUtils.spec.ts | 228 ++++++++++++++++ .../lib/utils/startupResourcePollingUtils.ts | 252 ++++++++++++++---- packages/sns/lib/utils/snsInitter.ts | 87 +++++- packages/sqs/lib/utils/sqsInitter.ts | 25 +- 7 files changed, 579 insertions(+), 58 deletions(-) diff --git a/README.md b/README.md index 3074e865..099b2726 100644 --- a/README.md +++ b/README.md @@ -285,13 +285,14 @@ const consumer = new MySnsSqsConsumer(dependencies, { | `enabled` | `boolean` | - | Must be set to `true` to enable polling | | `pollingIntervalMs` | `number` | `5000` | Interval between availability checks (ms) | | `timeoutMs` | `number \| NO_TIMEOUT` | - (required) | Maximum wait time before throwing `StartupResourcePollingTimeoutError`. Use `NO_TIMEOUT` to poll indefinitely | +| `throwOnTimeout` | `boolean` | `true` | When `true`, throws error on timeout. When `false`, reports error via errorReporter, resets timeout counter, and continues polling | ### Environment-Specific Configuration ```typescript import { NO_TIMEOUT } from '@message-queue-toolkit/core' -// Production - with timeout +// Production - with timeout (throws error when timeout reached) { locatorConfig: { queueUrl: '...', @@ -302,6 +303,19 @@ import { NO_TIMEOUT } from '@message-queue-toolkit/core' } } +// Production - report timeout but keep trying +// Useful when you want visibility into prolonged unavailability without failing +{ + locatorConfig: { + queueUrl: '...', + startupResourcePolling: { + enabled: true, + timeoutMs: 5 * 60 * 1000, // Report every 5 minutes + throwOnTimeout: false, // Don't throw, just report and continue + } + } +} + // Development/Staging - poll indefinitely { locatorConfig: { diff --git a/packages/core/lib/types/MessageQueueTypes.ts b/packages/core/lib/types/MessageQueueTypes.ts index c008dcc5..0e33f9ef 100644 --- a/packages/core/lib/types/MessageQueueTypes.ts +++ b/packages/core/lib/types/MessageQueueTypes.ts @@ -1,4 +1,8 @@ -import type { CommonLogger, TransactionObservabilityManager } from '@lokalise/node-core' +import type { + CommonLogger, + ErrorReporter, + TransactionObservabilityManager, +} from '@lokalise/node-core' import type { ZodSchema } from 'zod/v4' import type { PublicHandlerSpy } from '../queues/HandlerSpy.ts' @@ -38,6 +42,7 @@ export type { TransactionObservabilityManager } export type ExtraParams = { logger?: CommonLogger + errorReporter?: ErrorReporter } export type SchemaMap = Record< diff --git a/packages/core/lib/types/queueOptionsTypes.ts b/packages/core/lib/types/queueOptionsTypes.ts index 6570d0ff..10507448 100644 --- a/packages/core/lib/types/queueOptionsTypes.ts +++ b/packages/core/lib/types/queueOptionsTypes.ts @@ -213,6 +213,28 @@ export type StartupResourcePollingConfig = { * Default: 5000 (5 seconds) */ pollingIntervalMs?: number + + /** + * Whether to throw an error when timeout is reached. + * - `true` (default): Throws `StartupResourcePollingTimeoutError` when timeout is reached + * - `false`: Reports the error via errorReporter, resets the timeout counter, and continues polling + * + * Use `false` when you want to be notified about prolonged unavailability but don't want to fail. + * Default: true + */ + throwOnTimeout?: boolean + + /** + * Whether to run polling in non-blocking mode. + * - `false` (default): init() waits for the resource to become available before resolving + * - `true`: If resource is not immediately available, init() resolves immediately and + * polling continues in the background. When the resource becomes available, + * the `onResourceAvailable` callback is invoked. + * + * Use `true` when you want the service to start quickly without waiting for dependencies. + * Default: false + */ + nonBlocking?: boolean } /** diff --git a/packages/core/lib/utils/startupResourcePollingUtils.spec.ts b/packages/core/lib/utils/startupResourcePollingUtils.spec.ts index 17edd9fa..47f35ab7 100644 --- a/packages/core/lib/utils/startupResourcePollingUtils.spec.ts +++ b/packages/core/lib/utils/startupResourcePollingUtils.spec.ts @@ -156,6 +156,234 @@ describe('startupResourcePollingUtils', () => { expect(result).toBe('test-result') expect(checkFn).toHaveBeenCalledTimes(5) }) + + it('throws by default when throwOnTimeout is not specified', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: false }) + + await expect( + waitForResource({ + config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 50 }, + checkFn, + resourceName: 'test-resource', + }), + ).rejects.toThrow(StartupResourcePollingTimeoutError) + }) + + it('throws when throwOnTimeout is true', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: false }) + + await expect( + waitForResource({ + config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 50, throwOnTimeout: true }, + checkFn, + resourceName: 'test-resource', + }), + ).rejects.toThrow(StartupResourcePollingTimeoutError) + }) + + it('reports error and continues polling when throwOnTimeout is false', async () => { + let callCount = 0 + const checkFn = vi.fn().mockImplementation(() => { + callCount++ + // Become available after what would be 2 timeout cycles + if (callCount < 8) { + return Promise.resolve({ isAvailable: false }) + } + return Promise.resolve({ isAvailable: true, result: 'test-result' }) + }) + + const errorReporter = { + report: vi.fn(), + } + + const result = await waitForResource({ + config: { + enabled: true, + pollingIntervalMs: 10, + timeoutMs: 35, // Will timeout around every 3-4 calls + throwOnTimeout: false, + }, + checkFn, + resourceName: 'test-resource', + errorReporter, + }) + + expect(result).toBe('test-result') + // Should have reported at least one timeout error + expect(errorReporter.report).toHaveBeenCalled() + expect(errorReporter.report).toHaveBeenCalledWith( + expect.objectContaining({ + error: expect.any(StartupResourcePollingTimeoutError), + context: expect.objectContaining({ + resourceName: 'test-resource', + timeoutMs: 35, + }), + }), + ) + }) + + it('logs warning when throwOnTimeout is false and timeout is reached', async () => { + let callCount = 0 + const checkFn = vi.fn().mockImplementation(() => { + callCount++ + if (callCount < 5) { + return Promise.resolve({ isAvailable: false }) + } + return Promise.resolve({ isAvailable: true, result: 'test-result' }) + }) + + const logger = { + info: vi.fn(), + debug: vi.fn(), + error: vi.fn(), + warn: vi.fn(), + } + + await waitForResource({ + config: { + enabled: true, + pollingIntervalMs: 10, + timeoutMs: 25, + throwOnTimeout: false, + }, + checkFn, + resourceName: 'test-resource', + // @ts-expect-error - partial logger for testing + logger, + }) + + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + message: expect.stringContaining('resetting timeout counter'), + resourceName: 'test-resource', + }), + ) + }) + + it('tracks timeout count when throwOnTimeout is false', async () => { + let callCount = 0 + const checkFn = vi.fn().mockImplementation(() => { + callCount++ + // Become available after multiple timeout cycles + if (callCount < 12) { + return Promise.resolve({ isAvailable: false }) + } + return Promise.resolve({ isAvailable: true, result: 'test-result' }) + }) + + const errorReporter = { + report: vi.fn(), + } + + await waitForResource({ + config: { + enabled: true, + pollingIntervalMs: 5, + timeoutMs: 20, // Will timeout every ~4 calls + throwOnTimeout: false, + }, + checkFn, + resourceName: 'test-resource', + errorReporter, + }) + + // Should have reported multiple timeout errors with increasing timeoutCount + expect(errorReporter.report.mock.calls.length).toBeGreaterThan(1) + + // Verify timeoutCount increments + const firstCall = errorReporter.report.mock.calls[0]?.[0] as { + context: { timeoutCount: number } + } + const secondCall = errorReporter.report.mock.calls[1]?.[0] as { + context: { timeoutCount: number } + } + expect(firstCall.context.timeoutCount).toBe(1) + expect(secondCall.context.timeoutCount).toBe(2) + }) + + describe('nonBlocking mode', () => { + it('returns result immediately when resource is available on first check', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' }) + const onResourceAvailable = vi.fn() + + const result = await waitForResource({ + config: { enabled: true, pollingIntervalMs: 100, timeoutMs: 5000, nonBlocking: true }, + checkFn, + resourceName: 'test-resource', + onResourceAvailable, + }) + + expect(result).toBe('test-result') + expect(checkFn).toHaveBeenCalledTimes(1) + expect(onResourceAvailable).not.toHaveBeenCalled() + }) + + it('returns undefined immediately when resource is not available and starts background polling', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: false }) + const onResourceAvailable = vi.fn() + + const result = await waitForResource({ + config: { enabled: true, pollingIntervalMs: 100, timeoutMs: 5000, nonBlocking: true }, + checkFn, + resourceName: 'test-resource', + onResourceAvailable, + }) + + expect(result).toBeUndefined() + expect(checkFn).toHaveBeenCalledTimes(1) + }) + + it('calls onResourceAvailable when resource becomes available in background', async () => { + let callCount = 0 + const checkFn = vi.fn().mockImplementation(() => { + callCount++ + if (callCount < 3) { + return Promise.resolve({ isAvailable: false }) + } + return Promise.resolve({ isAvailable: true, result: 'test-result' }) + }) + const onResourceAvailable = vi.fn() + + const result = await waitForResource({ + config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 5000, nonBlocking: true }, + checkFn, + resourceName: 'test-resource', + onResourceAvailable, + }) + + expect(result).toBeUndefined() + + // Wait for background polling to complete + await new Promise((resolve) => globalThis.setTimeout(resolve, 100)) + + expect(onResourceAvailable).toHaveBeenCalledWith('test-result') + }) + + it('logs info message when starting background polling', async () => { + const checkFn = vi.fn().mockResolvedValue({ isAvailable: false }) + const logger = { + info: vi.fn(), + debug: vi.fn(), + error: vi.fn(), + warn: vi.fn(), + } + + await waitForResource({ + config: { enabled: true, pollingIntervalMs: 100, timeoutMs: 5000, nonBlocking: true }, + checkFn, + resourceName: 'test-resource', + // @ts-expect-error - partial logger for testing + logger, + }) + + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + message: expect.stringContaining('starting background polling'), + resourceName: 'test-resource', + }), + ) + }) + }) }) describe('StartupResourcePollingTimeoutError', () => { diff --git a/packages/core/lib/utils/startupResourcePollingUtils.ts b/packages/core/lib/utils/startupResourcePollingUtils.ts index 8726613d..cd37764f 100644 --- a/packages/core/lib/utils/startupResourcePollingUtils.ts +++ b/packages/core/lib/utils/startupResourcePollingUtils.ts @@ -1,5 +1,5 @@ import { setTimeout } from 'node:timers/promises' -import type { CommonLogger } from '@lokalise/node-core' +import type { CommonLogger, ErrorReporter } from '@lokalise/node-core' import { NO_TIMEOUT, type StartupResourcePollingConfig } from '../types/queueOptionsTypes.ts' const DEFAULT_POLLING_INTERVAL_MS = 5000 @@ -36,6 +36,18 @@ export type WaitForResourceOptions = { * Logger instance for progress logging */ logger?: CommonLogger + + /** + * Error reporter for reporting timeout errors when throwOnTimeout is false. + * Required when config.throwOnTimeout is false. + */ + errorReporter?: ErrorReporter + + /** + * Callback invoked when resource becomes available in non-blocking mode. + * Only used when config.nonBlocking is true and the resource was not immediately available. + */ + onResourceAvailable?: (result: T) => void } export class StartupResourcePollingTimeoutError extends Error { @@ -53,6 +65,10 @@ export class StartupResourcePollingTimeoutError extends Error { } } +type TimeoutCheckResult = + | { exceeded: false } + | { exceeded: true; error: StartupResourcePollingTimeoutError } + function checkTimeoutExceeded( hasTimeout: boolean, timeoutMs: number, @@ -60,8 +76,8 @@ function checkTimeoutExceeded( resourceName: string, attemptCount: number, logger?: CommonLogger, -): void { - if (!hasTimeout) return +): TimeoutCheckResult { + if (!hasTimeout) return { exceeded: false } const elapsedMs = Date.now() - startTime if (elapsedMs >= timeoutMs) { @@ -72,8 +88,64 @@ function checkTimeoutExceeded( attemptCount, elapsedMs, }) - throw new StartupResourcePollingTimeoutError(resourceName, timeoutMs) + return { + exceeded: true, + error: new StartupResourcePollingTimeoutError(resourceName, timeoutMs), + } + } + return { exceeded: false } +} + +type HandleTimeoutParams = { + timeoutResult: TimeoutCheckResult & { exceeded: true } + throwOnTimeout: boolean + resourceName: string + timeoutMs: number + attemptCount: number + timeoutCount: number + logger?: CommonLogger + errorReporter?: ErrorReporter +} + +/** + * Handles timeout condition - either throws or reports and returns new timeout count. + * @returns Updated timeout count after handling + */ +function handleTimeout(params: HandleTimeoutParams): number { + const { + timeoutResult, + throwOnTimeout, + resourceName, + timeoutMs, + attemptCount, + timeoutCount, + logger, + errorReporter, + } = params + + if (throwOnTimeout) { + throw timeoutResult.error } + + // Report error and reset timeout counter + const newTimeoutCount = timeoutCount + 1 + errorReporter?.report({ + error: timeoutResult.error, + context: { + resourceName, + timeoutMs, + attemptCount, + timeoutCount: newTimeoutCount, + }, + }) + logger?.warn({ + message: `Timeout waiting for resource "${resourceName}", resetting timeout counter and continuing`, + resourceName, + timeoutMs, + attemptCount, + timeoutCount: newTimeoutCount, + }) + return newTimeoutCount } function logResourceAvailable( @@ -91,68 +163,156 @@ function logResourceAvailable( }) } +/** + * Internal polling loop that waits for a resource to become available. + * Separated from main function to support non-blocking mode. + */ +async function pollForResource( + options: WaitForResourceOptions, + pollingIntervalMs: number, + hasTimeout: boolean, + timeoutMs: number, + throwOnTimeout: boolean, + initialAttemptCount: number, +): Promise { + const { checkFn, resourceName, logger, errorReporter } = options + + let startTime = Date.now() + let attemptCount = initialAttemptCount + let timeoutCount = 0 + + while (true) { + attemptCount++ + const timeoutResult = checkTimeoutExceeded( + hasTimeout, + timeoutMs, + startTime, + resourceName, + attemptCount, + logger, + ) + + if (timeoutResult.exceeded) { + timeoutCount = handleTimeout({ + timeoutResult, + throwOnTimeout, + resourceName, + timeoutMs, + attemptCount, + timeoutCount, + logger, + errorReporter, + }) + startTime = Date.now() + } + + const result = await checkFn().catch((error) => { + // Unexpected error during check - log and rethrow + logger?.error({ + message: `Error checking resource availability for "${resourceName}"`, + resourceName, + error, + attemptCount, + }) + throw error + }) + + if (result.isAvailable) { + logResourceAvailable(resourceName, attemptCount, startTime, logger) + return result.result + } + + // Resource not available yet, log and wait + if (attemptCount === 1 || attemptCount % 12 === 0) { + // Log on first attempt and then every minute (assuming 5s interval) + const elapsedMs = Date.now() - startTime + logger?.debug({ + message: `Resource "${resourceName}" not available yet, will retry`, + resourceName, + attemptCount, + elapsedMs, + nextRetryInMs: pollingIntervalMs, + }) + } + + // Wait before next attempt + await setTimeout(pollingIntervalMs) + } +} + /** * Waits for a resource to become available by polling. * This is used for startup resource polling mode where resources may not exist at startup. * * @param options - Configuration and check function - * @returns The result from the check function when resource becomes available - * @throws StartupResourcePollingTimeoutError if timeout is reached + * @returns The result from the check function when resource becomes available, + * or undefined if nonBlocking is true and resource was not immediately available + * @throws StartupResourcePollingTimeoutError if timeout is reached and throwOnTimeout is true (default) */ -export async function waitForResource(options: WaitForResourceOptions): Promise { - const { config, checkFn, resourceName, logger } = options +export async function waitForResource( + options: WaitForResourceOptions, +): Promise { + const { config, checkFn, resourceName, logger, onResourceAvailable } = options const pollingIntervalMs = config.pollingIntervalMs ?? DEFAULT_POLLING_INTERVAL_MS const hasTimeout = config.timeoutMs !== NO_TIMEOUT const timeoutMs = hasTimeout ? (config.timeoutMs as number) : 0 - - const startTime = Date.now() - let attemptCount = 0 + const throwOnTimeout = config.throwOnTimeout !== false + const nonBlocking = config.nonBlocking === true logger?.info({ message: `Waiting for resource "${resourceName}" to become available`, resourceName, pollingIntervalMs, timeoutMs: hasTimeout ? timeoutMs : 'NO_TIMEOUT', + throwOnTimeout, + nonBlocking, }) - while (true) { - attemptCount++ - checkTimeoutExceeded(hasTimeout, timeoutMs, startTime, resourceName, attemptCount, logger) - - try { - const result = await checkFn() - - if (result.isAvailable) { - logResourceAvailable(resourceName, attemptCount, startTime, logger) - return result.result - } - - // Resource not available yet, log and wait - if (attemptCount === 1 || attemptCount % 12 === 0) { - // Log on first attempt and then every minute (assuming 5s interval) - const elapsedMs = Date.now() - startTime - logger?.debug({ - message: `Resource "${resourceName}" not available yet, will retry`, - resourceName, - attemptCount, - elapsedMs, - nextRetryInMs: pollingIntervalMs, + // First check - always done synchronously + const initialResult = await checkFn().catch((error) => { + logger?.error({ + message: `Error checking resource availability for "${resourceName}"`, + resourceName, + error, + attemptCount: 1, + }) + throw error + }) + + if (initialResult.isAvailable) { + logResourceAvailable(resourceName, 1, Date.now(), logger) + return initialResult.result + } + + // Resource not available on first check + if (nonBlocking) { + // Start background polling and return immediately + logger?.info({ + message: `Resource "${resourceName}" not immediately available, starting background polling`, + resourceName, + pollingIntervalMs, + }) + + // Fire and forget - start polling in background + setTimeout(pollingIntervalMs).then(() => { + pollForResource(options, pollingIntervalMs, hasTimeout, timeoutMs, throwOnTimeout, 1) + .then((result) => { + onResourceAvailable?.(result) }) - } - } catch (error) { - // Unexpected error during check - log and rethrow - logger?.error({ - message: `Error checking resource availability for "${resourceName}"`, - resourceName, - error, - attemptCount, - }) - throw error - } + .catch((error) => { + logger?.error({ + message: `Background polling for resource "${resourceName}" failed`, + resourceName, + error, + }) + }) + }) - // Wait before next attempt - await setTimeout(pollingIntervalMs) + return undefined } + + // Blocking mode - continue polling and wait for result + return pollForResource(options, pollingIntervalMs, hasTimeout, timeoutMs, throwOnTimeout, 1) } /** diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index 50b7e4e1..89275361 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -22,9 +22,21 @@ import { subscribeToTopic } from './snsSubscriber.ts' import { assertTopic, deleteSubscription, deleteTopic, getTopicAttributes } from './snsUtils.ts' import { buildTopicArn } from './stsUtils.ts' -export type InitSnsSqsExtraParams = ExtraParams +export type InitSnsSqsExtraParams = ExtraParams & { + /** + * Callback invoked when resources become available in non-blocking mode. + * Only called when startupResourcePolling.nonBlocking is true and resources were not immediately available. + */ + onResourcesReady?: (result: { topicArn: string; queueUrl: string }) => void +} -export type InitSnsExtraParams = ExtraParams +export type InitSnsExtraParams = ExtraParams & { + /** + * Callback invoked when topic becomes available in non-blocking mode. + * Only called when startupResourcePolling.nonBlocking is true and topic was not immediately available. + */ + onTopicReady?: (result: { topicArn: string }) => void +} // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: fixme export async function initSnsSqs( @@ -88,6 +100,7 @@ export async function initSnsSqs( topicArn, queueName: creationConfig.queue.QueueName, queueUrl, + resourcesReady: true, } } @@ -101,11 +114,20 @@ export async function initSnsSqs( // If startup resource polling is enabled, poll for resources to become available if (isStartupResourcePollingEnabled(startupResourcePolling)) { + const nonBlocking = startupResourcePolling.nonBlocking === true + // Wait for topic to become available - await waitForResource({ + const topicResult = await waitForResource({ config: startupResourcePolling, resourceName: `SNS topic ${subscriptionTopicArn}`, logger: extraParams?.logger, + errorReporter: extraParams?.errorReporter, + onResourceAvailable: () => { + // In non-blocking mode, when topic becomes available, notify caller + if (nonBlocking) { + extraParams?.onResourcesReady?.({ topicArn: subscriptionTopicArn, queueUrl }) + } + }, checkFn: async () => { const result = await getTopicAttributes(snsClient, subscriptionTopicArn) if (result.error === 'not_found') { @@ -115,11 +137,33 @@ export async function initSnsSqs( }, }) + // If non-blocking and topic wasn't immediately available, return early + if (nonBlocking && topicResult === undefined) { + const splitUrl = queueUrl.split('/') + // biome-ignore lint/style/noNonNullAssertion: It's ok + const queueName = splitUrl[splitUrl.length - 1]! + + return { + subscriptionArn: locatorConfig.subscriptionArn, + topicArn: subscriptionTopicArn, + queueUrl, + queueName, + resourcesReady: false, + } + } + // Wait for queue to become available - await waitForResource({ + const queueResult = await waitForResource({ config: startupResourcePolling, resourceName: `SQS queue ${queueUrl}`, logger: extraParams?.logger, + errorReporter: extraParams?.errorReporter, + onResourceAvailable: () => { + // In non-blocking mode, when queue becomes available, notify caller + if (nonBlocking) { + extraParams?.onResourcesReady?.({ topicArn: subscriptionTopicArn, queueUrl }) + } + }, checkFn: async () => { const result = await getQueueAttributes(sqsClient, queueUrl) if (result.error === 'not_found') { @@ -128,6 +172,21 @@ export async function initSnsSqs( return { isAvailable: true, result: result.result } }, }) + + // If non-blocking and queue wasn't immediately available, return early + if (nonBlocking && queueResult === undefined) { + const splitUrl = queueUrl.split('/') + // biome-ignore lint/style/noNonNullAssertion: It's ok + const queueName = splitUrl[splitUrl.length - 1]! + + return { + subscriptionArn: locatorConfig.subscriptionArn, + topicArn: subscriptionTopicArn, + queueUrl, + queueName, + resourcesReady: false, + } + } } else { // Original behavior: check resources once and fail immediately if not found const checkPromises: Promise>[] = [] @@ -162,6 +221,7 @@ export async function initSnsSqs( topicArn: subscriptionTopicArn, queueUrl, queueName, + resourcesReady: true, } } @@ -264,10 +324,18 @@ export async function initSns( // If startup resource polling is enabled, poll for topic to become available if (isStartupResourcePollingEnabled(startupResourcePolling)) { - await waitForResource({ + const nonBlocking = startupResourcePolling.nonBlocking === true + + const topicResult = await waitForResource({ config: startupResourcePolling, resourceName: `SNS topic ${topicArn}`, logger: extraParams?.logger, + errorReporter: extraParams?.errorReporter, + onResourceAvailable: () => { + if (nonBlocking) { + extraParams?.onTopicReady?.({ topicArn }) + } + }, checkFn: async () => { const result = await getTopicAttributes(snsClient, topicArn) if (result.error === 'not_found') { @@ -276,6 +344,11 @@ export async function initSns( return { isAvailable: true, result: result.result } }, }) + + // In non-blocking mode, return early if topic wasn't immediately available + if (nonBlocking && topicResult === undefined) { + return { topicArn, resourcesReady: false } + } } else { // Original behavior: check once and fail immediately if not found const checkResult = await getTopicAttributes(snsClient, topicArn) @@ -284,7 +357,7 @@ export async function initSns( } } - return { topicArn } + return { topicArn, resourcesReady: true } } // create new topic if it does not exist @@ -300,5 +373,5 @@ export async function initSns( forceTagUpdate: creationConfig.forceTagUpdate, }) - return { topicArn } + return { topicArn, resourcesReady: true } } diff --git a/packages/sqs/lib/utils/sqsInitter.ts b/packages/sqs/lib/utils/sqsInitter.ts index cff257c5..de0ee15d 100644 --- a/packages/sqs/lib/utils/sqsInitter.ts +++ b/packages/sqs/lib/utils/sqsInitter.ts @@ -1,6 +1,6 @@ import type { QueueAttributeName, SQSClient } from '@aws-sdk/client-sqs' import { SetQueueAttributesCommand, TagQueueCommand } from '@aws-sdk/client-sqs' -import type { CommonLogger } from '@lokalise/node-core' +import type { CommonLogger, ErrorReporter } from '@lokalise/node-core' import type { DeletionConfig } from '@message-queue-toolkit/core' import { isProduction, @@ -18,8 +18,20 @@ import { validateFifoQueueName, } from './sqsUtils.ts' +export type InitSqsResult = { + queueArn: string | undefined + queueUrl: string + queueName: string +} + export type InitSqsExtraParams = { logger?: CommonLogger + errorReporter?: ErrorReporter + /** + * Callback invoked when queue becomes available in non-blocking mode. + * Only called when startupResourcePolling.nonBlocking is true and queue was not immediately available. + */ + onQueueReady?: (result: { queueArn: string }) => void } export async function deleteSqs( @@ -78,7 +90,7 @@ export async function initSqs( creationConfig?: SQSCreationConfig, isFifoQueue?: boolean, extraParams?: InitSqsExtraParams, -) { +): Promise { // reuse existing queue only if (locatorConfig) { const queueUrl = await resolveQueueUrlFromLocatorConfig(sqsClient, locatorConfig) @@ -93,6 +105,12 @@ export async function initSqs( config: startupResourcePolling, resourceName: `SQS queue ${queueUrl}`, logger: extraParams?.logger, + errorReporter: extraParams?.errorReporter, + onResourceAvailable: (arn) => { + if (arn) { + extraParams?.onQueueReady?.({ queueArn: arn }) + } + }, checkFn: async () => { const checkResult = await getQueueAttributes(sqsClient, queueUrl, ['QueueArn']) if (checkResult.error === 'not_found') { @@ -113,7 +131,8 @@ export async function initSqs( queueArn = checkResult.result?.attributes?.QueueArn } - if (!queueArn) { + // In non-blocking mode, queueArn may be undefined if resource wasn't immediately available + if (!queueArn && !startupResourcePolling?.nonBlocking) { throw new Error('Queue ARN was not set') } From d24e84b28aa1f646d9e0e3412a1b1a9e9e1b66d5 Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 15:51:34 +0200 Subject: [PATCH 08/11] Cleanup --- README.md | 14 +++ packages/sns/lib/utils/snsInitter.ts | 107 ++++++++++++--------- packages/sqs/lib/sqs/AbstractSqsService.ts | 3 +- 3 files changed, 76 insertions(+), 48 deletions(-) diff --git a/README.md b/README.md index 099b2726..3ecf7991 100644 --- a/README.md +++ b/README.md @@ -286,6 +286,7 @@ const consumer = new MySnsSqsConsumer(dependencies, { | `pollingIntervalMs` | `number` | `5000` | Interval between availability checks (ms) | | `timeoutMs` | `number \| NO_TIMEOUT` | - (required) | Maximum wait time before throwing `StartupResourcePollingTimeoutError`. Use `NO_TIMEOUT` to poll indefinitely | | `throwOnTimeout` | `boolean` | `true` | When `true`, throws error on timeout. When `false`, reports error via errorReporter, resets timeout counter, and continues polling | +| `nonBlocking` | `boolean` | `false` | When `true`, `init()` returns immediately if resource is not available, and polling continues in the background. A callback is invoked when the resource becomes available | ### Environment-Specific Configuration @@ -338,6 +339,19 @@ import { NO_TIMEOUT } from '@message-queue-toolkit/core' } } } + +// Non-blocking mode - service starts immediately, polling continues in background +// Useful when you want the service to be available even if dependencies are not ready +{ + locatorConfig: { + queueUrl: '...', + startupResourcePolling: { + enabled: true, + timeoutMs: 5 * 60 * 1000, + nonBlocking: true, // init() resolves immediately + } + } +} ``` ### Error Handling diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index 89275361..e4cc81ab 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -47,7 +47,13 @@ export async function initSnsSqs( creationConfig?: SNSCreationConfig & SQSCreationConfig, subscriptionConfig?: SNSSubscriptionOptions, extraParams?: InitSnsSqsExtraParams, -): Promise<{ subscriptionArn: string; topicArn: string; queueUrl: string; queueName: string }> { +): Promise<{ + subscriptionArn: string + topicArn: string + queueUrl: string + queueName: string + resourcesReady: boolean +}> { if (!locatorConfig?.subscriptionArn) { if (!creationConfig?.topic && !locatorConfig?.topicArn && !locatorConfig?.topicName) { throw new Error( @@ -303,61 +309,70 @@ export async function deleteSns( await deleteTopic(snsClient, stsClient, creationConfig.topic.Name) } -export async function initSns( +async function initSnsWithLocator( snsClient: SNSClient, stsClient: STSClient, - locatorConfig?: SNSTopicLocatorType, - creationConfig?: SNSCreationConfig, + locatorConfig: SNSTopicLocatorType, extraParams?: InitSnsExtraParams, ) { - if (locatorConfig) { - if (!locatorConfig.topicArn && !locatorConfig.topicName) { - throw new Error( - 'When locatorConfig for the topic is specified, either topicArn or topicName must be specified', - ) - } + if (!locatorConfig.topicArn && !locatorConfig.topicName) { + throw new Error( + 'When locatorConfig for the topic is specified, either topicArn or topicName must be specified', + ) + } - const topicArn = - locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? '')) + const topicArn = + locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? '')) - const startupResourcePolling = locatorConfig.startupResourcePolling + const startupResourcePolling = locatorConfig.startupResourcePolling - // If startup resource polling is enabled, poll for topic to become available - if (isStartupResourcePollingEnabled(startupResourcePolling)) { - const nonBlocking = startupResourcePolling.nonBlocking === true + // If startup resource polling is enabled, poll for topic to become available + if (isStartupResourcePollingEnabled(startupResourcePolling)) { + const nonBlocking = startupResourcePolling.nonBlocking === true - const topicResult = await waitForResource({ - config: startupResourcePolling, - resourceName: `SNS topic ${topicArn}`, - logger: extraParams?.logger, - errorReporter: extraParams?.errorReporter, - onResourceAvailable: () => { - if (nonBlocking) { - extraParams?.onTopicReady?.({ topicArn }) - } - }, - checkFn: async () => { - const result = await getTopicAttributes(snsClient, topicArn) - if (result.error === 'not_found') { - return { isAvailable: false } - } - return { isAvailable: true, result: result.result } - }, - }) - - // In non-blocking mode, return early if topic wasn't immediately available - if (nonBlocking && topicResult === undefined) { - return { topicArn, resourcesReady: false } - } - } else { - // Original behavior: check once and fail immediately if not found - const checkResult = await getTopicAttributes(snsClient, topicArn) - if (checkResult.error === 'not_found') { - throw new Error(`Topic with topicArn ${topicArn} does not exist.`) - } + const topicResult = await waitForResource({ + config: startupResourcePolling, + resourceName: `SNS topic ${topicArn}`, + logger: extraParams?.logger, + errorReporter: extraParams?.errorReporter, + onResourceAvailable: () => { + if (nonBlocking) { + extraParams?.onTopicReady?.({ topicArn }) + } + }, + checkFn: async () => { + const result = await getTopicAttributes(snsClient, topicArn) + if (result.error === 'not_found') { + return { isAvailable: false } + } + return { isAvailable: true, result: result.result } + }, + }) + + // In non-blocking mode, return early if topic wasn't immediately available + if (nonBlocking && topicResult === undefined) { + return { topicArn, resourcesReady: false } + } + } else { + // Original behavior: check once and fail immediately if not found + const checkResult = await getTopicAttributes(snsClient, topicArn) + if (checkResult.error === 'not_found') { + throw new Error(`Topic with topicArn ${topicArn} does not exist.`) } + } + + return { topicArn, resourcesReady: true } +} - return { topicArn, resourcesReady: true } +export async function initSns( + snsClient: SNSClient, + stsClient: STSClient, + locatorConfig?: SNSTopicLocatorType, + creationConfig?: SNSCreationConfig, + extraParams?: InitSnsExtraParams, +) { + if (locatorConfig) { + return initSnsWithLocator(snsClient, stsClient, locatorConfig, extraParams) } // create new topic if it does not exist diff --git a/packages/sqs/lib/sqs/AbstractSqsService.ts b/packages/sqs/lib/sqs/AbstractSqsService.ts index a8b160be..c1b2d0d4 100644 --- a/packages/sqs/lib/sqs/AbstractSqsService.ts +++ b/packages/sqs/lib/sqs/AbstractSqsService.ts @@ -95,8 +95,7 @@ export abstract class AbstractSqsService< protected queueName: string // @ts-expect-error protected queueUrl: string - // @ts-expect-error - protected queueArn: string + protected queueArn: string | undefined protected readonly isFifoQueue: boolean constructor(dependencies: DependenciesType, options: SQSOptionsType) { From fcada62b89fde56c724c5b944acf79ff7b0e086d Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 16:03:52 +0200 Subject: [PATCH 09/11] Add extra tests --- ...ionConsumer.startupResourcePolling.spec.ts | 80 +++++++++++++++++++ 1 file changed, 80 insertions(+) diff --git a/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts b/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts index e52b415c..d262524b 100644 --- a/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts +++ b/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts @@ -171,4 +171,84 @@ describe('SqsPermissionConsumer - startupResourcePollingConfig', () => { await expect(consumer.init()).rejects.toThrow(/does not exist/) }) }) + + describe('when nonBlocking mode is enabled', () => { + it('returns immediately when resource is available on first check', async () => { + // Create queue first + await assertQueue(sqsClient, { QueueName: queueName }) + + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { + locatorConfig: { + queueUrl, + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + nonBlocking: true, + }, + }, + }) + + await consumer.init() + + expect(consumer.queueProps.url).toBe(queueUrl) + expect(consumer.queueProps.arn).toBeDefined() + }) + + it('returns immediately when resource is not available and queueArn is undefined', async () => { + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { + locatorConfig: { + queueUrl, + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + nonBlocking: true, + }, + }, + }) + + // Init should complete immediately even though queue doesn't exist + await consumer.init() + + expect(consumer.queueProps.url).toBe(queueUrl) + expect(consumer.queueProps.arn).toBeUndefined() + }) + + it('invokes onQueueReady callback when resource becomes available in background', async () => { + // We need to test the callback at the initter level since AbstractSqsConsumer + // doesn't expose the onQueueReady callback directly + const { initSqs } = await import('../../lib/utils/sqsInitter.ts') + + let callbackInvoked = false + let callbackArn: string | undefined + + // Start init without queue existing + const initPromise = initSqs( + sqsClient, + { queueUrl, startupResourcePolling: { enabled: true, pollingIntervalMs: 50, timeoutMs: 5000, nonBlocking: true } }, + undefined, + undefined, + { + onQueueReady: (result) => { + callbackInvoked = true + callbackArn = result.queueArn + }, + }, + ) + + // Init should return immediately + const result = await initPromise + expect(result.queueArn).toBeUndefined() + + // Create queue after init returns + await assertQueue(sqsClient, { QueueName: queueName }) + + // Wait for background polling to detect the queue + await setTimeout(200) + + expect(callbackInvoked).toBe(true) + expect(callbackArn).toBeDefined() + }) + }) }) From d03af6288d16560f7c8a4b6162a8dc362e560821 Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 16:09:34 +0200 Subject: [PATCH 10/11] Address code review comments --- README.md | 74 +++++++++++++++++++ packages/sns/lib/utils/snsInitter.ts | 58 +++++++++++++-- ...ionConsumer.startupResourcePolling.spec.ts | 10 ++- 3 files changed, 133 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 3ecf7991..13690df4 100644 --- a/README.md +++ b/README.md @@ -354,6 +354,80 @@ import { NO_TIMEOUT } from '@message-queue-toolkit/core' } ``` +### Non-Blocking Mode + +When `nonBlocking: true` is set, `init()` returns immediately if the resource is not available on the first check. Polling continues in the background, and you can use callbacks to be notified when resources become available. + +**Important behaviors:** +- If the resource IS immediately available, `init()` returns normally with the resource ready +- If the resource is NOT immediately available, `init()` returns with `resourcesReady: false` (for SNS/SQS) or `queueArn: undefined` (for SQS) +- Background polling continues and invokes the callback when the resource becomes available +- For SNS+SQS consumers, the callback is only invoked when BOTH topic and queue are available + +```typescript +// SQS non-blocking with callback +import { initSqs } from '@message-queue-toolkit/sqs' + +const result = await initSqs( + sqsClient, + { + queueUrl: '...', + startupResourcePolling: { + enabled: true, + timeoutMs: 5 * 60 * 1000, + nonBlocking: true, + }, + }, + undefined, + undefined, + { + onQueueReady: ({ queueArn }) => { + console.log(`Queue is now available: ${queueArn}`) + // Start consuming messages, update health checks, etc. + }, + }, +) + +if (result.queueArn) { + console.log('Queue was immediately available') +} else { + console.log('Queue not yet available, will be notified via callback') +} + +// SNS+SQS non-blocking with callback +import { initSnsSqs } from '@message-queue-toolkit/sns' + +const result = await initSnsSqs( + sqsClient, + snsClient, + stsClient, + { + topicArn: '...', + queueUrl: '...', + subscriptionArn: '...', + startupResourcePolling: { + enabled: true, + timeoutMs: 5 * 60 * 1000, + nonBlocking: true, + }, + }, + undefined, + undefined, + { + onResourcesReady: ({ topicArn, queueUrl }) => { + // Called only when BOTH topic and queue are available + console.log(`Resources ready: topic=${topicArn}, queue=${queueUrl}`) + }, + }, +) + +if (result.resourcesReady) { + console.log('All resources were immediately available') +} else { + console.log('Resources not yet available, will be notified via callback') +} +``` + ### Error Handling When `timeoutMs` is configured and the resource doesn't become available within the specified time, a `StartupResourcePollingTimeoutError` is thrown: diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index e4cc81ab..9c44b812 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -122,6 +122,16 @@ export async function initSnsSqs( if (isStartupResourcePollingEnabled(startupResourcePolling)) { const nonBlocking = startupResourcePolling.nonBlocking === true + // Track availability for non-blocking mode coordination + let topicAvailable = false + let queueAvailable = false + + const notifyIfBothReady = () => { + if (nonBlocking && topicAvailable && queueAvailable) { + extraParams?.onResourcesReady?.({ topicArn: subscriptionTopicArn, queueUrl }) + } + } + // Wait for topic to become available const topicResult = await waitForResource({ config: startupResourcePolling, @@ -129,10 +139,8 @@ export async function initSnsSqs( logger: extraParams?.logger, errorReporter: extraParams?.errorReporter, onResourceAvailable: () => { - // In non-blocking mode, when topic becomes available, notify caller - if (nonBlocking) { - extraParams?.onResourcesReady?.({ topicArn: subscriptionTopicArn, queueUrl }) - } + topicAvailable = true + notifyIfBothReady() }, checkFn: async () => { const result = await getTopicAttributes(snsClient, subscriptionTopicArn) @@ -143,12 +151,43 @@ export async function initSnsSqs( }, }) + // If topic was immediately available, mark it + if (topicResult !== undefined) { + topicAvailable = true + } + // If non-blocking and topic wasn't immediately available, return early + // Background polling will continue and call notifyIfBothReady when topic is available if (nonBlocking && topicResult === undefined) { const splitUrl = queueUrl.split('/') // biome-ignore lint/style/noNonNullAssertion: It's ok const queueName = splitUrl[splitUrl.length - 1]! + // Also start polling for queue in background so we can notify when both are ready + waitForResource({ + config: startupResourcePolling, + resourceName: `SQS queue ${queueUrl}`, + logger: extraParams?.logger, + errorReporter: extraParams?.errorReporter, + onResourceAvailable: () => { + queueAvailable = true + notifyIfBothReady() + }, + checkFn: async () => { + const result = await getQueueAttributes(sqsClient, queueUrl) + if (result.error === 'not_found') { + return { isAvailable: false } + } + return { isAvailable: true, result: result.result } + }, + }).catch((error) => { + extraParams?.logger?.error({ + message: 'Background polling for SQS queue failed', + queueUrl, + error, + }) + }) + return { subscriptionArn: locatorConfig.subscriptionArn, topicArn: subscriptionTopicArn, @@ -165,10 +204,8 @@ export async function initSnsSqs( logger: extraParams?.logger, errorReporter: extraParams?.errorReporter, onResourceAvailable: () => { - // In non-blocking mode, when queue becomes available, notify caller - if (nonBlocking) { - extraParams?.onResourcesReady?.({ topicArn: subscriptionTopicArn, queueUrl }) - } + queueAvailable = true + notifyIfBothReady() }, checkFn: async () => { const result = await getQueueAttributes(sqsClient, queueUrl) @@ -179,6 +216,11 @@ export async function initSnsSqs( }, }) + // If queue was immediately available, mark it + if (queueResult !== undefined) { + queueAvailable = true + } + // If non-blocking and queue wasn't immediately available, return early if (nonBlocking && queueResult === undefined) { const splitUrl = queueUrl.split('/') diff --git a/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts b/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts index d262524b..b668033d 100644 --- a/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts +++ b/packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts @@ -226,7 +226,15 @@ describe('SqsPermissionConsumer - startupResourcePollingConfig', () => { // Start init without queue existing const initPromise = initSqs( sqsClient, - { queueUrl, startupResourcePolling: { enabled: true, pollingIntervalMs: 50, timeoutMs: 5000, nonBlocking: true } }, + { + queueUrl, + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: 5000, + nonBlocking: true, + }, + }, undefined, undefined, { From f88b9e588b06c177c984d7dbc00c1d87195cc88e Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Wed, 7 Jan 2026 16:33:27 +0200 Subject: [PATCH 11/11] add more tests --- packages/sns/lib/utils/snsInitter.ts | 21 ++- ...ionConsumer.startupResourcePolling.spec.ts | 139 +++++++++++++++++- 2 files changed, 153 insertions(+), 7 deletions(-) diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index 9c44b812..91a43909 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -180,13 +180,22 @@ export async function initSnsSqs( } return { isAvailable: true, result: result.result } }, - }).catch((error) => { - extraParams?.logger?.error({ - message: 'Background polling for SQS queue failed', - queueUrl, - error, - }) }) + .then((result) => { + // If queue was immediately available, waitForResource returns the result + // but doesn't call onResourceAvailable, so we handle it here + if (result !== undefined) { + queueAvailable = true + notifyIfBothReady() + } + }) + .catch((error) => { + extraParams?.logger?.error({ + message: 'Background polling for SQS queue failed', + queueUrl, + error, + }) + }) return { subscriptionArn: locatorConfig.subscriptionArn, diff --git a/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts b/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts index 3feb030f..b47cc412 100644 --- a/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts +++ b/packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts @@ -9,7 +9,7 @@ import { } from '@message-queue-toolkit/core' import { assertQueue, deleteQueue } from '@message-queue-toolkit/sqs' import type { AwilixContainer } from 'awilix' -import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest' +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' import { AbstractSnsSqsConsumer, type SNSSQSConsumerDependencies, @@ -221,6 +221,143 @@ describe('SnsSqsPermissionConsumer - startupResourcePollingConfig', () => { }) }) + describe('when nonBlocking mode is enabled', () => { + it('returns immediately when both resources are available on first check', async () => { + // Create queue first, then topic (order matters for LocalStack) + await assertQueue(sqsClient, { QueueName: queueName }) + const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) + + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { + locatorConfig: { + topicArn, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + nonBlocking: true, + }, + }, + }) + + await consumer.init() + + expect(consumer.subscriptionProps.topicArn).toBe(topicArn) + expect(consumer.subscriptionProps.queueUrl).toBe(queueUrl) + }) + + it('returns immediately when topic is not available', async () => { + // Create queue but not topic + await assertQueue(sqsClient, { QueueName: queueName }) + + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { + locatorConfig: { + topicName, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + nonBlocking: true, + }, + }, + creationConfig: { + queue: { QueueName: queueName }, + }, + }) + + // Init should return immediately even though topic doesn't exist + await consumer.init() + + // queueName should still be set + expect(consumer.subscriptionProps.queueName).toBe(queueName) + }) + + it('returns immediately when queue is not available', async () => { + // Create topic but not queue + const topicArn = await assertTopic(snsClient, stsClient, { Name: topicName }) + + const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { + locatorConfig: { + topicArn, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 100, + timeoutMs: 5000, + nonBlocking: true, + }, + }, + }) + + // Init should return immediately even though queue doesn't exist + await consumer.init() + + expect(consumer.subscriptionProps.topicArn).toBe(topicArn) + }) + + it('invokes onResourcesReady callback when both resources become available in background', async () => { + // Test at the initter level since AbstractSnsSqsConsumer doesn't expose the callback + const { initSnsSqs } = await import('../../lib/utils/snsInitter.ts') + + let callbackInvoked = false + let callbackTopicArn: string | undefined + let callbackQueueUrl: string | undefined + + // Create queue but not topic + await assertQueue(sqsClient, { QueueName: queueName }) + + const result = await initSnsSqs( + sqsClient, + snsClient, + stsClient, + { + topicName, + queueUrl, + subscriptionArn: + 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: 5000, + nonBlocking: true, + }, + }, + undefined, + undefined, + { + onResourcesReady: ({ topicArn, queueUrl: url }) => { + callbackInvoked = true + callbackTopicArn = topicArn + callbackQueueUrl = url + }, + }, + ) + + expect(result.resourcesReady).toBe(false) + + // Create topic after init returns + await assertTopic(snsClient, stsClient, { Name: topicName }) + + // Wait for callback to be invoked using vi.waitFor (more reliable than fixed timeout) + await vi.waitFor( + () => { + expect(callbackInvoked).toBe(true) + }, + { timeout: 2000, interval: 50 }, + ) + + expect(callbackTopicArn).toBeDefined() + expect(callbackQueueUrl).toBe(queueUrl) + }) + }) + describe('when startupResourcePolling is disabled', () => { it('throws immediately when topic does not exist', async () => { // Create queue but not topic