From 3de03c5581fea5594196f46d2719b68ebe65c156 Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Tue, 29 Sep 2026 20:30:51 +0200 Subject: [PATCH 1/9] feat: adding subscription locateOnly mode --- packages/sns/lib/index.ts | 5 +- .../sns/lib/sns/AbstractSnsSqsConsumer.ts | 9 ++- packages/sns/lib/utils/snsInitter.spec.ts | 2 +- packages/sns/lib/utils/snsInitter.ts | 75 ++++++++++++------- packages/sns/lib/utils/snsSubscriber.ts | 73 +++++++++++++----- 5 files changed, 112 insertions(+), 52 deletions(-) diff --git a/packages/sns/lib/index.ts b/packages/sns/lib/index.ts index dce81d925..ce91fd42a 100644 --- a/packages/sns/lib/index.ts +++ b/packages/sns/lib/index.ts @@ -34,7 +34,10 @@ export { } from './utils/snsInitter.ts' export { deserializeSNSMessage } from './utils/snsMessageDeserializer.ts' export { readSnsMessage } from './utils/snsMessageReader.ts' -export { type SNSSubscriptionOptions, subscribeToTopic } from './utils/snsSubscriber.ts' +export { + type SNSSubscriptionOptions, + subscribeToTopic, +} from './utils/snsSubscriber.ts' export { assertTopic, calculateOutgoingMessageSize, diff --git a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts index 3a75a5f4b..a66cfa1cc 100644 --- a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts +++ b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts @@ -132,7 +132,12 @@ export abstract class AbstractSnsSqsConsumer< } override async init(): Promise { - if (this.deletionConfig && this.creationConfig && this.subscriptionConfig) { + if ( + this.deletionConfig && + this.creationConfig && + this.subscriptionConfig && + !this.subscriptionConfig.locateOnly + ) { await deleteSnsSqs( this.sqsClient, this.snsClient, @@ -144,7 +149,7 @@ export abstract class AbstractSnsSqsConsumer< undefined, this.locatorConfig, ) - } else if (this.deletionConfig && this.creationConfig) { + } else if (this.deletionConfig && this.creationConfig && !this.subscriptionConfig?.locateOnly) { await deleteSqs(this.sqsClient, this.deletionConfig, this.creationConfig) } diff --git a/packages/sns/lib/utils/snsInitter.spec.ts b/packages/sns/lib/utils/snsInitter.spec.ts index a6bc9778b..31b71ad93 100644 --- a/packages/sns/lib/utils/snsInitter.spec.ts +++ b/packages/sns/lib/utils/snsInitter.spec.ts @@ -226,7 +226,7 @@ describe('snsInitter', () => { { queue: { QueueName: queueName } }, ), ).rejects.toThrow( - 'If creationConfig.queue is specified, subscriptionConfig is mandatory, as the subscription of a queue being created cannot be located', + 'If creationConfig.queue is specified, a subscriptionConfig without locateOnly is mandatory, as the subscription of a queue being created cannot be located', ) }) }) diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index 56d8d99ab..600d6b30f 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -23,8 +23,13 @@ import { import type { SNSCreationConfig, SNSTopicLocatorType } from '../sns/AbstractSnsService.ts' import type { SNSSQSQueueLocatorType } from '../sns/AbstractSnsSqsConsumer.ts' import { isCreateTopicCommand } from '../types/TopicTypes.ts' -import type { SNSSubscriptionOptions } from './snsSubscriber.ts' -import { assertSubscription, subscribeToTopic } from './snsSubscriber.ts' +import { + assertSubscription, + type SNSSubscriptionCreationOptions, + type SNSSubscriptionOptions, + setSubscriptionAttributes, + subscribeToTopic, +} from './snsSubscriber.ts' import { assertTopic, deleteSubscription, @@ -239,8 +244,11 @@ async function findConfirmedSubscriptionArn( } /** - * Subscription is used as is when its ARN is given, created (or updated) when subscriptionConfig is - * given, and otherwise located by looking up the queue's subscription on the topic. + * Subscription is created (or updated) when a creation subscriptionConfig is given. Otherwise, it is located: + * used as is when its ARN is given, or looked up on the topic, and the attributes of a locate-only + * subscriptionConfig are applied to it. + * + * Note: a creation subscriptionConfig is never given along with the subscription ARN, as initSnsSqs discards it. */ async function resolveSubscriptionArn( snsClient: SNSClient, @@ -250,9 +258,7 @@ async function resolveSubscriptionArn( subscriptionConfig: SNSSubscriptionOptions | undefined, options: ResourceResolutionOptions, ): Promise { - if (locatorConfig?.subscriptionArn) return locatorConfig.subscriptionArn - - if (subscriptionConfig) { + if (subscriptionConfig && !subscriptionConfig.locateOnly) { const subscriptionArn = await assertSubscription( snsClient, topicArn, @@ -266,20 +272,28 @@ async function resolveSubscriptionArn( return subscriptionArn } - return await waitForOrCheckResource( - `SNS subscription of SQS queue ${queue.queueArn} to topic ${topicArn}`, - `Subscription of queue ${queue.queueArn} to topic ${topicArn} does not exist.`, - async () => { - const subscriptionArn = await findConfirmedSubscriptionArn( - snsClient, - topicArn, - queue.queueArn, - ) - if (!subscriptionArn) return { isAvailable: false } - return { isAvailable: true, result: subscriptionArn } - }, - options, - ) + const subscriptionArn = + locatorConfig?.subscriptionArn ?? + (await waitForOrCheckResource( + `SNS subscription of SQS queue ${queue.queueArn} to topic ${topicArn}`, + `Subscription of queue ${queue.queueArn} to topic ${topicArn} does not exist.`, + async () => { + const subscriptionArn = await findConfirmedSubscriptionArn( + snsClient, + topicArn, + queue.queueArn, + ) + if (!subscriptionArn) return { isAvailable: false } + return { isAvailable: true, result: subscriptionArn } + }, + options, + )) + + if (subscriptionConfig?.locateOnly && subscriptionConfig?.Attributes) { + await setSubscriptionAttributes(snsClient, subscriptionArn, subscriptionConfig.Attributes) + } + + return subscriptionArn } async function resolveSnsSqsResources( @@ -335,9 +349,9 @@ function validateInitSnsSqsConfig( 'If locatorConfig.subscriptionArn is not specified, creationConfig.queue.QueueName parameter is mandatory, as there will be an attempt to create the missing queue', ) } - if (!subscriptionConfig) { + if (!subscriptionConfig || subscriptionConfig.locateOnly) { throw new Error( - 'If creationConfig.queue is specified, subscriptionConfig is mandatory, as the subscription of a queue being created cannot be located', + 'If creationConfig.queue is specified, a subscriptionConfig without locateOnly is mandatory, as the subscription of a queue being created cannot be located', ) } } @@ -351,10 +365,12 @@ function validateInitSnsSqsConfig( * - Queue: located when `locatorConfig.queueUrl` or `locatorConfig.queueName` is given, otherwise * created from `creationConfig.queue`. * - Subscription: created (or updated) when `subscriptionConfig` is given, otherwise located by - * looking up the queue's subscription on the topic. + * looking up the queue's subscription on the topic. With `subscriptionConfig.locateOnly`, it is + * always located and its `Attributes` (e.g. `FilterPolicy`) are applied to it. * * When `locatorConfig.subscriptionArn` is given, every resource is located and `creationConfig` and - * `subscriptionConfig` are ignored. + * `subscriptionConfig` are ignored, except for a locate-only `subscriptionConfig`, whose attributes + * are still applied. * * Located resources are waited for when `locatorConfig.startupResourcePolling` is enabled. In * non-blocking mode, `undefined` is returned if they are not immediately available, and @@ -371,7 +387,10 @@ export async function initSnsSqs( ): Promise { const isFullyLocated = !!locatorConfig?.subscriptionArn const resolvedCreationConfig = isFullyLocated ? undefined : creationConfig - const resolvedSubscriptionConfig = isFullyLocated ? undefined : subscriptionConfig + // A locate-only subscriptionConfig doesn't create anything, so it is kept to apply its attributes + const resolvedSubscriptionConfig = + isFullyLocated && !subscriptionConfig?.locateOnly ? undefined : subscriptionConfig + const createsSubscription = !!resolvedSubscriptionConfig && !resolvedSubscriptionConfig.locateOnly validateInitSnsSqsConfig(locatorConfig, resolvedCreationConfig, resolvedSubscriptionConfig) @@ -388,7 +407,7 @@ export async function initSnsSqs( const startupResourcePolling = locatorConfig?.startupResourcePolling if (!isStartupResourcePollingEnabled(startupResourcePolling)) { - return await resolve({ checkLocatedTopic: !resolvedSubscriptionConfig }) + return await resolve({ checkLocatedTopic: !createsSubscription }) } if (startupResourcePolling.nonBlocking !== true) { return await resolve({ polling: startupResourcePolling, checkLocatedTopic: true }) @@ -429,7 +448,7 @@ export async function deleteSnsSqs( deletionConfig: DeletionConfig, queueConfiguration: CreateQueueCommandInput, topicConfiguration: CreateTopicCommandInput | undefined, - subscriptionConfiguration: SNSSubscriptionOptions, + subscriptionConfiguration: SNSSubscriptionCreationOptions, extraParams?: ExtraParams, topicLocator?: SNSTopicLocatorType, ) { diff --git a/packages/sns/lib/utils/snsSubscriber.ts b/packages/sns/lib/utils/snsSubscriber.ts index 025e963cb..56df6d3f9 100644 --- a/packages/sns/lib/utils/snsSubscriber.ts +++ b/packages/sns/lib/utils/snsSubscriber.ts @@ -20,10 +20,31 @@ import { import { assertTopic, findSubscriptionByTopicAndQueue } from './snsUtils.ts' import { buildTopicArn } from './stsUtils.ts' -export type SNSSubscriptionOptions = Omit< - SubscribeCommandInput, - 'TopicArn' | 'Endpoint' | 'Protocol' | 'ReturnSubscriptionArn' -> & { updateAttributesIfExists: boolean } +/** + * Options for the subscription of a queue to a topic. + * + * There are two modes for this config: + * 1. Creation (default): the subscription is created, or updated if it already exists. + * 2. Locate only: for subscriptions managed externally. The subscription is only located, never created, + * and the given attributes (e.g. `FilterPolicy`) are applied to it. + */ +export type SNSSubscriptionOptions = + /** Creation */ + | (Omit & { + updateAttributesIfExists: boolean + /** Should not be present when the subscription is created */ + locateOnly?: never + }) + /** Locate only */ + | { + /** Marks the subscription as located only */ + locateOnly: true + /** Attributes applied to the located subscription */ + Attributes?: SubscribeCommandInput['Attributes'] + } + +/** Subscription options that create the subscription, required when subscribing */ +export type SNSSubscriptionCreationOptions = Exclude async function resolveTopicArnToSubscribeTo( snsClient: SNSClient, @@ -61,7 +82,7 @@ export async function subscribeToTopic( stsClient: STSClient, queueConfiguration: CreateQueueCommandInput, topicConfiguration: TopicResolutionOptions, - subscriptionConfiguration: SNSSubscriptionOptions, + subscriptionConfiguration: SNSSubscriptionCreationOptions, extraParams?: ExtraSNSCreationParams & ExtraSQSCreationParams & ExtraParams, ) { const topicArn = await resolveTopicArnToSubscribeTo( @@ -103,7 +124,7 @@ export async function assertSubscription( snsClient: SNSClient, topicArn: string, queueArn: string, - subscriptionConfiguration: SNSSubscriptionOptions, + subscriptionConfiguration: SNSSubscriptionCreationOptions, errorContext: { queueName?: string; topicName?: string }, logger?: CommonLogger, ): Promise { @@ -156,26 +177,38 @@ async function tryToUpdateSubscription( snsClient: SNSClient, topicArn: string, queueArn: string, - subscriptionConfiguration: SNSSubscriptionOptions, + subscriptionConfiguration: SNSSubscriptionCreationOptions, ) { const subscription = await findSubscriptionByTopicAndQueue(snsClient, topicArn, queueArn) - if (!subscription || !subscriptionConfiguration.Attributes) { + if (!subscription?.SubscriptionArn || !subscriptionConfiguration.Attributes) { return undefined } - const setSubscriptionAttributesCommands = Object.entries( + await setSubscriptionAttributes( + snsClient, + subscription.SubscriptionArn, subscriptionConfiguration.Attributes, - ).map(([key, value]) => { - return new SetSubscriptionAttributesCommand({ - SubscriptionArn: subscription.SubscriptionArn, - AttributeName: key, - AttributeValue: value, - }) - }) - - for (const command of setSubscriptionAttributesCommands) { - await snsClient.send(command) - } + ) return subscription } + +/** + * Applies the given attributes to an existing subscription, one by one as SNS only allows setting a single + * attribute per call. + */ +export async function setSubscriptionAttributes( + snsClient: SNSClient, + subscriptionArn: string, + attributes: NonNullable, +): Promise { + for (const [key, value] of Object.entries(attributes)) { + await snsClient.send( + new SetSubscriptionAttributesCommand({ + SubscriptionArn: subscriptionArn, + AttributeName: key, + AttributeValue: value, + }), + ) + } +} From 82dcc75788a2c7e9342c63f892f07e279dc44a45 Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Tue, 29 Sep 2026 20:42:39 +0200 Subject: [PATCH 2/9] Adding deprecation notice --- packages/sns/lib/sns/AbstractSnsSqsConsumer.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts index a66cfa1cc..1718354d4 100644 --- a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts +++ b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts @@ -28,6 +28,11 @@ export type SNSSQSCreationConfig = Omit & SNS export type SNSSQSQueueLocatorType = Partial & SNSTopicLocatorType & { + /** + * @deprecated The subscription is located from the topic and queue locators, so its ARN is no longer needed. + * To avoid creating it, omit `subscriptionConfig` or use `subscriptionConfig.locateOnly`. It will be removed in + * the next major version. + */ subscriptionArn?: string } From 4be0f4d919de27bf1c6792c9dc15b8e06f3a2327 Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Tue, 29 Sep 2026 20:45:00 +0200 Subject: [PATCH 3/9] Improve validation to avoid locate only subscription when creating topic and queue (not possible) --- packages/sns/lib/utils/snsInitter.spec.ts | 54 ++++++++++++++++++++++- packages/sns/lib/utils/snsInitter.ts | 12 ++++- 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/packages/sns/lib/utils/snsInitter.spec.ts b/packages/sns/lib/utils/snsInitter.spec.ts index 31b71ad93..4040ba8e1 100644 --- a/packages/sns/lib/utils/snsInitter.spec.ts +++ b/packages/sns/lib/utils/snsInitter.spec.ts @@ -226,9 +226,61 @@ describe('snsInitter', () => { { queue: { QueueName: queueName } }, ), ).rejects.toThrow( - 'If creationConfig.queue is specified, a subscriptionConfig without locateOnly is mandatory, as the subscription of a queue being created cannot be located', + 'If creationConfig.queue is specified, subscriptionConfig is mandatory, as the subscription of a queue being created cannot be located', ) }) + + it('throws when queue creation config is provided with a locate-only subscription config', async () => { + await expect( + initSnsSqs( + sqsClient, + snsClient, + stsClient, + { topicName }, + { queue: { QueueName: queueName } }, + { locateOnly: true }, + ), + ).rejects.toThrow( + 'If subscriptionConfig.locateOnly is specified, both the topic and the queue must be located', + ) + }) + + it('throws when topic creation config is provided with a locate-only subscription config', async () => { + await expect( + initSnsSqs( + sqsClient, + snsClient, + stsClient, + { queueName }, + { topic: { Name: topicName }, queue: { QueueName: queueName } }, + { locateOnly: true }, + ), + ).rejects.toThrow( + 'If subscriptionConfig.locateOnly is specified, both the topic and the queue must be located', + ) + }) + + it('ignores creation config when topic and queue are located with a locate-only subscription config', async () => { + const topicArn = await testAdmin.createTopic(topicName) + const { queueArn } = await testAdmin.createQueue(queueName) + const subscriptionArn = await testAdmin.createSubscription(topicArn, queueArn) + const snsSpy = vi.spyOn(snsClient, 'send') + const sqsSpy = vi.spyOn(sqsClient, 'send') + + const result = await initSnsSqs( + sqsClient, + snsClient, + stsClient, + { topicName, queueName }, + { topic: { Name: topicName }, queue: { QueueName: queueName } }, + { locateOnly: true }, + ) + + expect(result).toEqual({ topicArn, queueUrl, queueArn, queueName, subscriptionArn }) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(CreateTopicCommand)) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(SubscribeCommand)) + expect(sqsSpy).not.toHaveBeenCalledWith(expect.any(CreateQueueCommand)) + }) }) describe('non-blocking startup resource polling', () => { diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index 600d6b30f..c629626ee 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -343,15 +343,23 @@ function validateInitSnsSqsConfig( } // Queue creation config is ignored when the queue is located const isQueueLocated = !!(locatorConfig?.queueUrl || locatorConfig?.queueName) + const isTopicLocated = !!(locatorConfig?.topicArn || locatorConfig?.topicName) + + // A subscription managed externally can only belong to a topic and a queue managed externally too + if (subscriptionConfig?.locateOnly && (!isTopicLocated || !isQueueLocated)) { + throw new Error( + 'If subscriptionConfig.locateOnly is specified, both the topic and the queue must be located', + ) + } if (!isQueueLocated && creationConfig?.queue) { if (!creationConfig.queue.QueueName) { throw new Error( 'If locatorConfig.subscriptionArn is not specified, creationConfig.queue.QueueName parameter is mandatory, as there will be an attempt to create the missing queue', ) } - if (!subscriptionConfig || subscriptionConfig.locateOnly) { + if (!subscriptionConfig) { throw new Error( - 'If creationConfig.queue is specified, a subscriptionConfig without locateOnly is mandatory, as the subscription of a queue being created cannot be located', + 'If creationConfig.queue is specified, subscriptionConfig is mandatory, as the subscription of a queue being created cannot be located', ) } } From af49f84c445996dd70f8cdd9d095ce1567b71052 Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Tue, 29 Sep 2026 20:45:17 +0200 Subject: [PATCH 4/9] Adding tests to cover changes --- packages/sns/lib/utils/snsInitter.spec.ts | 136 ++++++++++++++++++ .../SnsSqsPermissionConsumer.spec.ts | 33 ++++- packages/sqs/lib/sqs/AbstractSqsConsumer.ts | 2 +- 3 files changed, 168 insertions(+), 3 deletions(-) diff --git a/packages/sns/lib/utils/snsInitter.spec.ts b/packages/sns/lib/utils/snsInitter.spec.ts index 4040ba8e1..0ecce14a1 100644 --- a/packages/sns/lib/utils/snsInitter.spec.ts +++ b/packages/sns/lib/utils/snsInitter.spec.ts @@ -1,8 +1,10 @@ import { setTimeout } from 'node:timers/promises' import { CreateTopicCommand, + GetSubscriptionAttributesCommand, GetTopicAttributesCommand, ListSubscriptionsByTopicCommand, + SetSubscriptionAttributesCommand, type SNSClient, SubscribeCommand, } from '@aws-sdk/client-sns' @@ -182,6 +184,140 @@ describe('snsInitter', () => { }) }) + describe('locate-only subscription', () => { + const filterPolicy = JSON.stringify({ type: ['entity.created'] }) + const getFilterPolicy = async (subscriptionArn: string) => { + const { Attributes } = await snsClient.send( + new GetSubscriptionAttributesCommand({ SubscriptionArn: subscriptionArn }), + ) + return Attributes?.FilterPolicy + } + + it('locates existing subscription and applies its attributes without subscribing', async () => { + const topicArn = await testAdmin.createTopic(topicName) + const { queueArn } = await testAdmin.createQueue(queueName) + const subscriptionArn = await testAdmin.createSubscription(topicArn, queueArn) + const snsSpy = vi.spyOn(snsClient, 'send') + + const result = await initSnsSqs( + sqsClient, + snsClient, + stsClient, + { topicName, queueName }, + undefined, + { locateOnly: true, Attributes: { FilterPolicy: filterPolicy } }, + ) + + expect(result).toEqual({ topicArn, queueUrl, queueArn, queueName, subscriptionArn }) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(SubscribeCommand)) + expect(await getFilterPolicy(subscriptionArn)).toBe(filterPolicy) + }) + + it('does not update the subscription when no attributes are provided', async () => { + const topicArn = await testAdmin.createTopic(topicName) + const { queueArn } = await testAdmin.createQueue(queueName) + const subscriptionArn = await testAdmin.createSubscription(topicArn, queueArn) + const snsSpy = vi.spyOn(snsClient, 'send') + + const result = await initSnsSqs( + sqsClient, + snsClient, + stsClient, + { topicName, queueName }, + undefined, + { locateOnly: true }, + ) + + expect(result?.subscriptionArn).toBe(subscriptionArn) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(SubscribeCommand)) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(SetSubscriptionAttributesCommand)) + }) + + it('applies attributes to the subscription when its ARN is provided', async () => { + const topicArn = await testAdmin.createTopic(topicName) + const { queueArn } = await testAdmin.createQueue(queueName) + const subscriptionArn = await testAdmin.createSubscription(topicArn, queueArn) + const snsSpy = vi.spyOn(snsClient, 'send') + + const result = await initSnsSqs( + sqsClient, + snsClient, + stsClient, + { topicName, queueName, subscriptionArn }, + undefined, + { locateOnly: true, Attributes: { FilterPolicy: filterPolicy } }, + ) + + expect(result?.subscriptionArn).toBe(subscriptionArn) + // The subscription is not looked up, as its ARN is provided + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(ListSubscriptionsByTopicCommand)) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(SubscribeCommand)) + expect(await getFilterPolicy(subscriptionArn)).toBe(filterPolicy) + }) + + it('throws without creating the subscription when it does not exist', async () => { + const topicArn = await testAdmin.createTopic(topicName) + const { queueArn } = await testAdmin.createQueue(queueName) + + await expect( + initSnsSqs(sqsClient, snsClient, stsClient, { topicName, queueName }, undefined, { + locateOnly: true, + Attributes: { FilterPolicy: filterPolicy }, + }), + ).rejects.toThrow(/Subscription of queue .* to topic .* does not exist/) + expect(await findSubscriptionByTopicAndQueue(snsClient, topicArn, queueArn)).toBeUndefined() + }) + + it('throws when the located topic does not exist', async () => { + await testAdmin.createQueue(queueName) + + await expect( + initSnsSqs(sqsClient, snsClient, stsClient, { topicName, queueName }, undefined, { + locateOnly: true, + }), + ).rejects.toThrow(/Topic with topicArn .* does not exist/) + }) + + it('applies attributes once the subscription becomes available in non-blocking mode', async () => { + const topicArn = await testAdmin.createTopic(topicName) + const { queueArn } = await testAdmin.createQueue(queueName) + const onResourcesReady = vi.fn() + + const result = await initSnsSqs( + sqsClient, + snsClient, + stsClient, + { + topicName, + queueName, + startupResourcePolling: { + enabled: true, + pollingIntervalMs: 50, + timeoutMs: 5000, + nonBlocking: true, + }, + }, + undefined, + { locateOnly: true, Attributes: { FilterPolicy: filterPolicy } }, + { onResourcesReady }, + ) + + expect(result).toBeUndefined() + + const subscriptionArn = await testAdmin.createSubscription(topicArn, queueArn) + + await waitAndRetry(() => onResourcesReady.mock.calls.length > 0, 50, 40) + expect(onResourcesReady).toHaveBeenCalledWith({ + topicArn, + queueUrl, + queueArn, + queueName, + subscriptionArn, + }) + expect(await getFilterPolicy(subscriptionArn)).toBe(filterPolicy) + }) + }) + describe('config validation', () => { it('throws when neither topic creation config nor topic locator is provided', async () => { await expect( diff --git a/packages/sns/test/consumers/SnsSqsPermissionConsumer.spec.ts b/packages/sns/test/consumers/SnsSqsPermissionConsumer.spec.ts index ca12d2eae..399873867 100644 --- a/packages/sns/test/consumers/SnsSqsPermissionConsumer.spec.ts +++ b/packages/sns/test/consumers/SnsSqsPermissionConsumer.spec.ts @@ -1,6 +1,12 @@ import { setTimeout } from 'node:timers/promises' -import { ListTagsForResourceCommand, type SNSClient, SubscribeCommand } from '@aws-sdk/client-sns' -import { ListQueueTagsCommand, type SQSClient } from '@aws-sdk/client-sqs' +import { + DeleteTopicCommand, + ListTagsForResourceCommand, + type SNSClient, + SubscribeCommand, + UnsubscribeCommand, +} from '@aws-sdk/client-sns' +import { DeleteQueueCommand, ListQueueTagsCommand, type SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' import { waitAndRetry } from '@lokalise/node-core' import { getQueueAttributes } from '@message-queue-toolkit/sqs' @@ -161,6 +167,29 @@ describe('SnsSqsPermissionConsumer', () => { expect(snsSpy).not.toHaveBeenCalledWith(expect.any(SubscribeCommand)) }) + it('does not delete nor create any resource with a locate-only subscription config', async () => { + const topicArn = await testAdmin.createTopic(topicName) + const { queueArn } = await testAdmin.createQueue(queueName) + const subscriptionArn = await testAdmin.createSubscription(topicArn, queueArn) + const snsSpy = vi.spyOn(snsClient, 'send') + const sqsSpy = vi.spyOn(sqsClient, 'send') + + const newConsumer = new SnsSqsPermissionConsumer(diContainer.cradle, { + locatorConfig: { topicName, queueName }, + creationConfig: { queue: { QueueName: queueName }, topic: { Name: topicName } }, + deletionConfig: { deleteIfExists: true }, + subscriptionConfig: { locateOnly: true }, + }) + + await newConsumer.init() + expect(newConsumer.subscriptionProps.subscriptionArn).toBe(subscriptionArn) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(SubscribeCommand)) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(UnsubscribeCommand)) + expect(snsSpy).not.toHaveBeenCalledWith(expect.any(DeleteTopicCommand)) + expect(sqsSpy).not.toHaveBeenCalledWith(expect.any(DeleteQueueCommand)) + expect(await findSubscriptionByTopicAndQueue(snsClient, topicArn, queueArn)).toBeDefined() + }) + describe('tags update', () => { const getQueueTags = (queueUrl: string) => sqsClient.send(new ListQueueTagsCommand({ QueueUrl: queueUrl })) diff --git a/packages/sqs/lib/sqs/AbstractSqsConsumer.ts b/packages/sqs/lib/sqs/AbstractSqsConsumer.ts index c2799aad0..516850a11 100644 --- a/packages/sqs/lib/sqs/AbstractSqsConsumer.ts +++ b/packages/sqs/lib/sqs/AbstractSqsConsumer.ts @@ -78,7 +78,7 @@ type SQSDeadLetterQueueOptions = | { /** * A located DLQ can omit the redrive policy, in which case the redrive policy of the source queue is left - * untouched (e.g. when both queues are managed externally) + * untouched */ locatorConfig: SQSQueueLocatorType creationConfig?: never From 203a18035ff26a5fff4024bb6e3c6cc372c0767b Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Tue, 29 Sep 2026 20:48:22 +0200 Subject: [PATCH 5/9] readme update --- packages/sns/README.md | 88 ++++++++++++++++++++++++++++-------------- 1 file changed, 58 insertions(+), 30 deletions(-) diff --git a/packages/sns/README.md b/packages/sns/README.md index d175c1566..10c23b512 100644 --- a/packages/sns/README.md +++ b/packages/sns/README.md @@ -488,9 +488,9 @@ When using `locatorConfig`, you connect to an existing topic without creating it // or // queueName: 'my-queue', - // Optional: Existing subscription ARN. When omitted and no `subscriptionConfig` is given, the subscription of - // the queue to the topic is looked up instead (see Resource Resolution) - subscriptionArn: 'arn:aws:sns:us-east-1:123456789012:my-topic:uuid', + // Deprecated: the subscription is located from the topic and the queue (see Resource Resolution), so its ARN is + // no longer needed. It will be removed in the next major version + // subscriptionArn: 'arn:aws:sns:us-east-1:123456789012:my-topic:uuid', }, } ``` @@ -500,16 +500,25 @@ When using `locatorConfig`, you connect to an existing topic without creating it Consumers resolve the topic, the queue and the subscription independently, so each of them can either be managed by your application or by external tooling (e.g. Terraform): -| Resource | Located when | Created when | -|--------------|------------------------------------------------|----------------------------------------| -| Topic | `locatorConfig.topicArn` or `topicName` is set | otherwise, from `creationConfig.topic` | -| Queue | `locatorConfig.queueUrl` or `queueName` is set | otherwise, from `creationConfig.queue` | -| Subscription | `subscriptionConfig` is not set | `subscriptionConfig` is set | +| Resource | Located when | Created when | +|--------------|------------------------------------------------|-------------------------------------------| +| Topic | `locatorConfig.topicArn` or `topicName` is set | otherwise, from `creationConfig.topic` | +| Queue | `locatorConfig.queueUrl` or `queueName` is set | otherwise, from `creationConfig.queue` | +| Subscription | no `subscriptionConfig`, or `locateOnly: true` | `subscriptionConfig` without `locateOnly` | A subscription that already exists is reused, and its attributes are updated when they differ and `subscriptionConfig.updateAttributesIfExists` is enabled. When both a locator and a creation config are given for the -same resource, the locator takes precedence and the creation config is ignored. When `locatorConfig.subscriptionArn` -is set, every resource is located and both `creationConfig` and `subscriptionConfig` are ignored. +same resource, the locator takes precedence and the creation config is ignored. + +With `subscriptionConfig: { locateOnly: true, Attributes }` the subscription is located, never created, and the given +`Attributes` (e.g. `FilterPolicy`) are applied to it on startup. This lets external tooling own the subscription while +the application keeps its filter policy in sync with the consumer handlers. A locate-only subscription requires both +the topic and the queue to be located, and none of the located resources are deleted when `deletionConfig` is set. + +> **Deprecated**: `locatorConfig.subscriptionArn` is no longer needed, as the subscription is located from the topic +> and the queue. When it is set, every resource is located and `creationConfig` and `subscriptionConfig` are ignored, +> except for a locate-only `subscriptionConfig`, whose attributes are still applied. It will be removed in the next +> major version. ```typescript // Application manages everything @@ -535,14 +544,25 @@ is set, every resource is located and both `creationConfig` and `subscriptionCon { locatorConfig: { topicName: 'my-topic', queueName: 'my-queue' }, } + +// Everything managed externally, the application only manages the subscription filter policy +{ + locatorConfig: { topicName: 'my-topic', queueName: 'my-queue' }, + subscriptionConfig: { + locateOnly: true, + Attributes: { FilterPolicy: JSON.stringify({ messageType: ['user.created'] }) }, + }, +} ``` Some things to keep in mind when resources are managed externally: - **Queue policy**: when the queue is located, the application does not set its policy. The external tooling must allow the topic to send messages to the queue, otherwise the subscription exists but messages are never delivered. -- **Filter policy**: when the subscription is located, its filter policy is not derived from the consumer handlers. - The external tooling must keep it in sync with the message types the consumer handles. +- **Filter policy**: when the subscription is located without `subscriptionConfig`, its filter policy is not derived + from the consumer handlers, so the external tooling must keep it in sync with the message types the consumer + handles. Use `subscriptionConfig.locateOnly` with `Attributes.FilterPolicy` to keep managing it from the application + (requires `sns:SetSubscriptionAttributes`, and the external tooling should ignore changes to it). - **Permissions**: locating a subscription requires `sns:ListSubscriptionsByTopic`. Subscriptions pending confirmation (e.g. cross-account) are not considered until they are confirmed. - A missing located resource makes `init()` fail, unless startup resource polling is enabled (see below). @@ -556,7 +576,6 @@ When your SNS topic or SQS queue may not exist at startup (e.g., created by anot locatorConfig: { topicName: 'my-topic', queueUrl: 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue', - subscriptionArn: 'arn:aws:sns:us-east-1:123456789012:my-topic:uuid', // Enable startup resource polling startupResourcePolling: { @@ -582,7 +601,6 @@ const consumer = new MyConsumer(deps, { locatorConfig: { topicName: 'my-topic', queueUrl: 'https://sqs...', - subscriptionArn: 'arn:aws:sns:...', startupResourcePolling: { enabled: true, pollingIntervalMs: 5000, @@ -606,7 +624,6 @@ const consumer = new MyConsumer(deps, { locatorConfig: { topicName: 'my-topic', queueUrl: 'https://sqs...', - subscriptionArn: 'arn:aws:sns:...', startupResourcePolling: { enabled: true, pollingIntervalMs: 5000, @@ -680,13 +697,13 @@ if (result.resourcesReady) { #### Subscription Creation Mode -When you want to **create** a subscription (no existing `subscriptionArn`), startup resource polling will wait for the topic to exist before attempting to subscribe: +When you want to **create** a subscription (`subscriptionConfig` without `locateOnly`), startup resource polling will +wait for the topic to exist before attempting to subscribe: ```typescript const consumer = new MyConsumer(deps, { locatorConfig: { topicName: 'my-topic', // Topic created by another service - // No subscriptionArn - we'll create the subscription startupResourcePolling: { enabled: true, pollingIntervalMs: 5000, @@ -696,6 +713,7 @@ const consumer = new MyConsumer(deps, { creationConfig: { queue: { QueueName: 'my-consumer-queue' }, // Queue will be created }, + subscriptionConfig: { updateAttributesIfExists: true }, // Subscription will be created }) // This will: @@ -708,8 +726,8 @@ await consumer.start() #### Subscription Locate Mode -When the subscription is managed externally (no `subscriptionConfig`), startup resource polling will also wait for the -subscription of the queue to the topic to exist: +When the subscription is managed externally (no `subscriptionConfig`, or `subscriptionConfig.locateOnly`), startup +resource polling will also wait for the subscription of the queue to the topic to exist: ```typescript const consumer = new MyConsumer(deps, { @@ -722,13 +740,15 @@ const consumer = new MyConsumer(deps, { timeoutMs: 60000, }, }, - // No subscriptionConfig - the subscription is located, never created + // No subscriptionConfig - the subscription is located, never created. Alternatively, keep managing its filter policy: + // subscriptionConfig: { locateOnly: true, Attributes: { FilterPolicy: '...' } }, }) // This will: // 1. Poll until the topic and the queue exist // 2. Poll until the queue is subscribed to the topic -// 3. Start consuming +// 3. Apply the subscription attributes, if a locate-only subscriptionConfig is given +// 4. Start consuming await consumer.start() ``` @@ -799,7 +819,6 @@ SNS consumers use the same options as SQS consumers, plus SNS-specific subscript locatorConfig: { topicArn: 'arn:aws:sns:...', queueUrl: 'https://sqs...', - subscriptionArn: 'arn:aws:sns:...', }, // SNS-Specific - Subscription Configuration @@ -819,6 +838,8 @@ SNS consumers use the same options as SQS consumers, plus SNS-specific subscript deadLetterTargetArn: 'arn:aws:sqs:us-east-1:123456789012:my-dlq', }, }, + // or, for a subscription managed externally (requires topic and queue locators) + // subscriptionConfig: { locateOnly: true, Attributes: { FilterPolicy: '...' } }, // Optional - FIFO Configuration fifoQueue: false, @@ -1382,14 +1403,21 @@ type SNSSQSConsumerDependencies = SNSDependencies & SQSDependencies & { } // Subscription options -type SNSSubscriptionOptions = { - updateAttributesIfExists?: boolean - filterPolicy?: Record - rawMessageDelivery?: boolean - redrivePolicy?: { - deadLetterTargetArn: string - } -} +type SNSSubscriptionOptions = + // Creation: the subscription is created, or updated if it already exists + | { + updateAttributesIfExists?: boolean + filterPolicy?: Record + rawMessageDelivery?: boolean + redrivePolicy?: { + deadLetterTargetArn: string + } + } + // Locate only: the subscription is located, never created, and the given attributes are applied to it + | { + locateOnly: true + Attributes?: Record + } ``` ### Utility Functions From 6874e2bf0d634ccbafd78cbd35be13fcff283daf Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Tue, 29 Sep 2026 20:49:34 +0200 Subject: [PATCH 6/9] readme fix --- README.md | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 5215b2322..6f440adfd 100644 --- a/README.md +++ b/README.md @@ -110,7 +110,7 @@ Multi-schema consumers support multiple message types via handler configs. They * `queueName`; (for SNS publishers this is a misnomer which actually refers to a topic name) * `locatorConfig` - configuration for resolving existing queue and/or topic. Should not be specified together with the `creationConfig`. * `creationConfig` - configuration for queue and/or topic to create, if one does not exist. Should not be specified together with the `locatorConfig`. - * `subscriptionConfig` - SNS SQS consumer only - configuration for SNS -> SQS subscription to create, if one doesn't exist. + * `subscriptionConfig` - SNS SQS consumer only - configuration for SNS -> SQS subscription to create, if one doesn't exist. With `locateOnly: true`, the subscription is only located (never created) and its `Attributes` (e.g. `FilterPolicy`) are applied to it, which requires the topic and the queue to be located too (see the [SNS README](packages/sns/README.md#resource-resolution)). * `policyConfig` - SQS only - configuration for queue access policies (see [SQS Policy Configuration](#sqs-policy-configuration) for more information); * `deletionConfig` - automatic cleanup of resources; * `consumerOverrides` – available only for SQS consumers; @@ -337,7 +337,6 @@ const consumer = new MySnsSqsConsumer(dependencies, { locatorConfig: { topicArn: 'arn:aws:sns:...', queueUrl: 'https://sqs...', - subscriptionArn: '...', // Enable eventual consistency mode startupResourcePolling: { enabled: true, // Enable polling for resource availability @@ -474,7 +473,6 @@ const result = await initSnsSqs( { topicArn: '...', queueUrl: '...', - subscriptionArn: '...', startupResourcePolling: { enabled: true, timeoutMs: 5 * 60 * 1000, @@ -485,7 +483,7 @@ const result = await initSnsSqs( undefined, { onResourcesReady: ({ topicArn, queueUrl }) => { - // Called only when BOTH topic and queue are available + // Called only when the topic, the queue and the subscription are all available console.log(`Resources ready: topic=${topicArn}, queue=${queueUrl}`) }, }, From cd6a94f724f5cf45d1bba11bd02460f2cf01158f Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Fri, 2 Oct 2026 10:43:43 +0300 Subject: [PATCH 7/9] feat(sns): read-first subscription reconciliation with typed managedAttributes (#618) * feat(sns): check subscription attributes before writing them assertSubscription now looks up an existing subscription and reads its attributes before doing anything. When they already match the config, no Subscribe or SetSubscriptionAttributes call is made. Otherwise only the differing attributes are written (or an error is thrown when updateAttributesIfExists is off). The locateOnly path gets the same read-first behaviour through setSubscriptionAttributes. Adds subscriptionConfig.manageOnlyFilterPolicy, which limits checking and writing to FilterPolicy and FilterPolicyScope. * feat(sns): replace manageOnlyFilterPolicy with typed managedAttributes subscriptionConfig.managedAttributes lists the subscription attributes the application owns, defaulting to FilterPolicy, FilterPolicyScope, RawMessageDelivery and RedrivePolicy. A managed attribute missing from Attributes is now reset on the existing subscription, unmanaged ones are never read or written, and configuring an unmanaged one throws. The consumer DLQ RedrivePolicy write goes through the read-first setSubscriptionAttributes and RedrivePolicy is excluded from the managed set when reuseConsumerDeadLetterQueue is on, so the two never fight over it. * test(sns): skip redrive policy removal on localstack LocalStack rejects any RedrivePolicy that is not a policy with a valid deadLetterTargetArn, so it cannot remove one. The reset test no longer sets RedrivePolicy, and its removal is covered by a separate test that runs on fauxqs only. * fix(sns): remove RedrivePolicy by omitting its value Real AWS rejects SetSubscriptionAttributes with an empty RedrivePolicy; the attribute is removed by leaving AttributeValue out, which is what the Terraform AWS provider does. FilterPolicy is reset with "{}" as in the SNS docs. --- packages/sns/README.md | 38 ++- packages/sns/lib/index.ts | 2 + .../sns/lib/sns/AbstractSnsSqsConsumer.ts | 49 +++- packages/sns/lib/utils/snsInitter.ts | 23 +- packages/sns/lib/utils/snsSubscriber.spec.ts | 267 +++++++++++++++++- packages/sns/lib/utils/snsSubscriber.ts | 259 +++++++++++++---- packages/sns/lib/utils/snsUtils.ts | 12 + packages/sns/test/utils/fauxqsInstance.ts | 2 +- 8 files changed, 558 insertions(+), 94 deletions(-) diff --git a/packages/sns/README.md b/packages/sns/README.md index 10c23b512..b138961e4 100644 --- a/packages/sns/README.md +++ b/packages/sns/README.md @@ -506,12 +506,33 @@ your application or by external tooling (e.g. Terraform): | Queue | `locatorConfig.queueUrl` or `queueName` is set | otherwise, from `creationConfig.queue` | | Subscription | no `subscriptionConfig`, or `locateOnly: true` | `subscriptionConfig` without `locateOnly` | -A subscription that already exists is reused, and its attributes are updated when they differ and -`subscriptionConfig.updateAttributesIfExists` is enabled. When both a locator and a creation config are given for the -same resource, the locator takes precedence and the creation config is ignored. +A subscription that already exists is reused. Its attributes are read first and compared with the configured ones +(JSON policies structurally), so a subscription that is already up to date receives no writes at all. Differing +attributes are written one by one when `subscriptionConfig.updateAttributesIfExists` is enabled, and an error is thrown +otherwise. When both a locator and a creation config are given for the same resource, the locator takes precedence and +the creation config is ignored. + +`subscriptionConfig.managedAttributes` lists the subscription attributes the application owns. It defaults to all of +`FilterPolicy`, `FilterPolicyScope`, `RawMessageDelivery` and `RedrivePolicy`. A managed attribute that is missing from +`Attributes` is reset on the existing subscription: `FilterPolicy` and `RedrivePolicy` are removed, `FilterPolicyScope` +goes back to `MessageAttributes` and `RawMessageDelivery` to `false`. Attributes left out of `managedAttributes` are +neither checked nor written, so external tooling can own them, and setting one of them in `Attributes` is a +configuration error. + +```typescript +// Filter policy owned by the application, everything else by Terraform +subscriptionConfig: { + locateOnly: true, + managedAttributes: ['FilterPolicy', 'FilterPolicyScope'], + Attributes: { FilterPolicy: JSON.stringify({ type: ['entity.created'] }) }, +} +``` + +With `subscriptionDeadLetterQueue.reuseConsumerDeadLetterQueue`, `RedrivePolicy` is set from the consumer DLQ and is +excluded from the managed attributes. With `subscriptionConfig: { locateOnly: true, Attributes }` the subscription is located, never created, and the given -`Attributes` (e.g. `FilterPolicy`) are applied to it on startup. This lets external tooling own the subscription while +managed attributes (e.g. `FilterPolicy`) are applied to it on startup when they differ from the current ones. This lets external tooling own the subscription while the application keeps its filter policy in sync with the consumer handlers. A locate-only subscription requires both the topic and the queue to be located, and none of the located resources are deleted when `deletionConfig` is set. @@ -824,6 +845,7 @@ SNS consumers use the same options as SQS consumers, plus SNS-specific subscript // SNS-Specific - Subscription Configuration subscriptionConfig: { updateAttributesIfExists: false, // Update subscription attributes if exists + managedAttributes: ['FilterPolicy', 'FilterPolicyScope'], // Defaults to all supported attributes // Optional: Message filtering filterPolicy: { @@ -1407,6 +1429,7 @@ type SNSSubscriptionOptions = // Creation: the subscription is created, or updated if it already exists | { updateAttributesIfExists?: boolean + managedAttributes?: SubscriptionManagedAttributeName[] filterPolicy?: Record rawMessageDelivery?: boolean redrivePolicy?: { @@ -1416,8 +1439,15 @@ type SNSSubscriptionOptions = // Locate only: the subscription is located, never created, and the given attributes are applied to it | { locateOnly: true + managedAttributes?: SubscriptionManagedAttributeName[] Attributes?: Record } + +type SubscriptionManagedAttributeName = + | 'FilterPolicy' + | 'FilterPolicyScope' + | 'RawMessageDelivery' + | 'RedrivePolicy' ``` ### Utility Functions diff --git a/packages/sns/lib/index.ts b/packages/sns/lib/index.ts index ce91fd42a..784d01956 100644 --- a/packages/sns/lib/index.ts +++ b/packages/sns/lib/index.ts @@ -36,6 +36,8 @@ export { deserializeSNSMessage } from './utils/snsMessageDeserializer.ts' export { readSnsMessage } from './utils/snsMessageReader.ts' export { type SNSSubscriptionOptions, + SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES, + type SubscriptionManagedAttributeName, subscribeToTopic, } from './utils/snsSubscriber.ts' export { diff --git a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts index 1718354d4..c1710f0a4 100644 --- a/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts +++ b/packages/sns/lib/sns/AbstractSnsSqsConsumer.ts @@ -1,5 +1,4 @@ import type { SNSClient } from '@aws-sdk/client-sns' -import { SetSubscriptionAttributesCommand } from '@aws-sdk/client-sns' import type { STSClient } from '@aws-sdk/client-sts' import { type Either, InternalError } from '@lokalise/node-core' import type { @@ -17,7 +16,11 @@ import type { import { AbstractSqsConsumer, deleteSqs } from '@message-queue-toolkit/sqs' import { deleteSnsSqs, initSnsSqs } from '../utils/snsInitter.ts' import { readSnsMessage } from '../utils/snsMessageReader.ts' -import type { SNSSubscriptionOptions } from '../utils/snsSubscriber.ts' +import { + type SNSSubscriptionOptions, + SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES, + setSubscriptionAttributes, +} from '../utils/snsSubscriber.ts' import type { SNSCreationConfig, SNSOptions, SNSTopicLocatorType } from './AbstractSnsService.ts' export type SNSSQSConsumerDependencies = SQSConsumerDependencies & { @@ -120,9 +123,11 @@ export abstract class AbstractSnsSqsConsumer< ) { super(dependencies, { ...options }, executionContext) - this.subscriptionConfig = options.subscriptionConfig this.reuseConsumerDeadLetterQueueForSubscription = !!options.subscriptionDeadLetterQueue?.reuseConsumerDeadLetterQueue + this.subscriptionConfig = this.reuseConsumerDeadLetterQueueForSubscription + ? withoutManagedRedrivePolicy(options.subscriptionConfig) + : options.subscriptionConfig if (this.reuseConsumerDeadLetterQueueForSubscription && !options.deadLetterQueue) { throw new InternalError({ @@ -262,12 +267,11 @@ export abstract class AbstractSnsSqsConsumer< const dlq = this.deadLetterQueue if (!dlq) return - await this.snsClient.send( - new SetSubscriptionAttributesCommand({ - SubscriptionArn: this.subscription.subscriptionArn, - AttributeName: 'RedrivePolicy', - AttributeValue: JSON.stringify({ deadLetterTargetArn: dlq.arn }), - }), + await setSubscriptionAttributes( + this.snsClient, + this.subscription.subscriptionArn, + { RedrivePolicy: JSON.stringify({ deadLetterTargetArn: dlq.arn }) }, + ['RedrivePolicy'], ) } @@ -328,3 +332,30 @@ export abstract class AbstractSnsSqsConsumer< return this._messageSchemaContainer.resolveSchema(messagePayload) } } + +/** + * With `subscriptionDeadLetterQueue.reuseConsumerDeadLetterQueue`, the redrive policy is set once the + * DLQ is resolved, so the subscription config must not manage (and reset) it on its own. + */ +function withoutManagedRedrivePolicy( + subscriptionConfig: SNSSubscriptionOptions | undefined, +): SNSSubscriptionOptions | undefined { + if (!subscriptionConfig) return undefined + if ( + subscriptionConfig.managedAttributes?.includes('RedrivePolicy') || + subscriptionConfig.Attributes?.RedrivePolicy + ) { + throw new InternalError({ + errorCode: 'invalid_subscription_dlq_configuration', + message: + 'subscriptionDeadLetterQueue.reuseConsumerDeadLetterQueue sets the subscription RedrivePolicy, so subscriptionConfig must not manage it', + }) + } + + return { + ...subscriptionConfig, + managedAttributes: ( + subscriptionConfig.managedAttributes ?? SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES + ).filter((name) => name !== 'RedrivePolicy'), + } +} diff --git a/packages/sns/lib/utils/snsInitter.ts b/packages/sns/lib/utils/snsInitter.ts index c629626ee..d31ad175c 100644 --- a/packages/sns/lib/utils/snsInitter.ts +++ b/packages/sns/lib/utils/snsInitter.ts @@ -34,7 +34,7 @@ import { assertTopic, deleteSubscription, deleteTopic, - findSubscriptionByTopicAndQueue, + findConfirmedSubscriptionArn, getTopicAttributes, } from './snsUtils.ts' import { buildTopicArn } from './stsUtils.ts' @@ -231,18 +231,6 @@ async function resolveQueue( ) } -async function findConfirmedSubscriptionArn( - snsClient: SNSClient, - topicArn: string, - queueArn: string, -): Promise { - const subscription = await findSubscriptionByTopicAndQueue(snsClient, topicArn, queueArn) - const subscriptionArn = subscription?.SubscriptionArn - - // Unconfirmed subscriptions are listed with a status placeholder instead of an ARN - return subscriptionArn?.startsWith('arn:') ? subscriptionArn : undefined -} - /** * Subscription is created (or updated) when a creation subscriptionConfig is given. Otherwise, it is located: * used as is when its ARN is given, or looked up on the topic, and the attributes of a locate-only @@ -289,8 +277,13 @@ async function resolveSubscriptionArn( options, )) - if (subscriptionConfig?.locateOnly && subscriptionConfig?.Attributes) { - await setSubscriptionAttributes(snsClient, subscriptionArn, subscriptionConfig.Attributes) + if (subscriptionConfig?.locateOnly) { + await setSubscriptionAttributes( + snsClient, + subscriptionArn, + subscriptionConfig.Attributes, + subscriptionConfig.managedAttributes, + ) } return subscriptionArn diff --git a/packages/sns/lib/utils/snsSubscriber.spec.ts b/packages/sns/lib/utils/snsSubscriber.spec.ts index 57560c202..181319be8 100644 --- a/packages/sns/lib/utils/snsSubscriber.spec.ts +++ b/packages/sns/lib/utils/snsSubscriber.spec.ts @@ -1,13 +1,18 @@ -import type { SNSClient } from '@aws-sdk/client-sns' +import { + SetSubscriptionAttributesCommand, + type SNSClient, + SubscribeCommand, +} from '@aws-sdk/client-sns' import type { SQSClient } from '@aws-sdk/client-sqs' import type { STSClient } from '@aws-sdk/client-sts' import type { AwilixContainer } from 'awilix' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { FakeLogger } from '../../test/fakes/FakeLogger.ts' +import { isLocalstack } from '../../test/utils/fauxqsInstance.ts' import type { TestAwsResourceAdmin } from '../../test/utils/testAdmin.ts' import type { Dependencies } from '../../test/utils/testContext.ts' import { registerDependencies } from '../../test/utils/testContext.ts' -import { assertSubscription, subscribeToTopic } from './snsSubscriber.ts' +import { assertSubscription, setSubscriptionAttributes, subscribeToTopic } from './snsSubscriber.ts' import { findSubscriptionByTopicAndQueue, getSubscriptionAttributes } from './snsUtils.ts' const TOPIC_NAME = 'topic' @@ -81,13 +86,11 @@ describe('snsSubscriber', () => { logger, }, ), - ).rejects.toThrow( - /Invalid parameter: Attributes Reason: Subscription already exists with different attributes/, - ) + ).rejects.toThrow(/Subscription already exists with different attributes/) expect(logger.loggedErrors).toHaveLength(1) expect(logger.loggedErrors[0]).toBe( - 'Error while creating subscription for queue "queue", topic "topic": Invalid parameter: Attributes Reason: Subscription already exists with different attributes', + 'Error while creating subscription for queue "queue", topic "topic": Subscription already exists with different attributes: FilterPolicy, FilterPolicyScope', ) }) @@ -234,6 +237,231 @@ describe('snsSubscriber', () => { }) }) + it('does not write to an existing subscription when its attributes are unchanged', async () => { + const topicArn = await testAdmin.createTopic(TOPIC_NAME) + const { queueArn } = await testAdmin.createQueue(QUEUE_NAME) + const subscriptionConfig = { + Attributes: { + FilterPolicy: `{"type":["add","remove"]}`, + FilterPolicyScope: 'MessageBody', + RawMessageDelivery: 'true', + }, + updateAttributesIfExists: true, + } + const existingSubscriptionArn = await assertSubscription( + snsClient, + topicArn, + queueArn, + subscriptionConfig, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + const sendSpy = vi.spyOn(snsClient, 'send') + + const subscriptionArn = await assertSubscription( + snsClient, + topicArn, + queueArn, + { + ...subscriptionConfig, + Attributes: { + ...subscriptionConfig.Attributes, + FilterPolicy: `{ "type": [ "add", "remove" ] }`, + }, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + + expect(subscriptionArn).toBe(existingSubscriptionArn) + const sentCommands = sendSpy.mock.calls.map(([command]) => command) + expect(sentCommands.some((command) => command instanceof SubscribeCommand)).toBe(false) + expect( + sentCommands.some((command) => command instanceof SetSubscriptionAttributesCommand), + ).toBe(false) + }) + + it('writes only the attributes that changed', async () => { + const topicArn = await testAdmin.createTopic(TOPIC_NAME) + const { queueArn } = await testAdmin.createQueue(QUEUE_NAME) + const subscriptionArn = await assertSubscription( + snsClient, + topicArn, + queueArn, + { + Attributes: { FilterPolicy: `{"type":["remove"]}`, RawMessageDelivery: 'true' }, + updateAttributesIfExists: false, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + const sendSpy = vi.spyOn(snsClient, 'send') + + await assertSubscription( + snsClient, + topicArn, + queueArn, + { + Attributes: { FilterPolicy: `{"type":["add"]}`, RawMessageDelivery: 'true' }, + updateAttributesIfExists: true, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + new FakeLogger(), + ) + + const writtenAttributes = sendSpy.mock.calls + .map(([command]) => command) + .filter((command) => command instanceof SetSubscriptionAttributesCommand) + .map((command) => command.input.AttributeName) + expect(writtenAttributes).toEqual(['FilterPolicy']) + const subscriptionAttributes = await getSubscriptionAttributes(snsClient, subscriptionArn!) + expect(subscriptionAttributes.result?.attributes?.FilterPolicy).toBe(`{"type":["add"]}`) + }) + + it('neither checks nor writes attributes that are not managed', async () => { + const topicArn = await testAdmin.createTopic(TOPIC_NAME) + const { queueArn } = await testAdmin.createQueue(QUEUE_NAME) + const subscriptionArn = await assertSubscription( + snsClient, + topicArn, + queueArn, + { + Attributes: { FilterPolicy: `{"type":["add"]}`, RawMessageDelivery: 'true' }, + updateAttributesIfExists: false, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + const sendSpy = vi.spyOn(snsClient, 'send') + + await assertSubscription( + snsClient, + topicArn, + queueArn, + { + Attributes: { FilterPolicy: `{"type":["add"]}` }, + managedAttributes: ['FilterPolicy', 'FilterPolicyScope'], + updateAttributesIfExists: false, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + + expect( + sendSpy.mock.calls.some(([command]) => command instanceof SetSubscriptionAttributesCommand), + ).toBe(false) + const subscriptionAttributes = await getSubscriptionAttributes(snsClient, subscriptionArn!) + expect(subscriptionAttributes.result?.attributes?.RawMessageDelivery).toBe('true') + }) + + it('resets managed attributes that were removed from the config', async () => { + const topicArn = await testAdmin.createTopic(TOPIC_NAME) + const { queueArn } = await testAdmin.createQueue(QUEUE_NAME) + const subscriptionArn = await assertSubscription( + snsClient, + topicArn, + queueArn, + { + Attributes: { + FilterPolicy: `{"type":["add"]}`, + FilterPolicyScope: 'MessageBody', + RawMessageDelivery: 'true', + }, + updateAttributesIfExists: false, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + + await assertSubscription( + snsClient, + topicArn, + queueArn, + { Attributes: { FilterPolicy: `{"type":["add"]}` }, updateAttributesIfExists: true }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + new FakeLogger(), + ) + + const subscriptionAttributes = await getSubscriptionAttributes(snsClient, subscriptionArn!) + expect(subscriptionAttributes.result?.attributes).toMatchObject({ + FilterPolicy: `{"type":["add"]}`, + FilterPolicyScope: 'MessageAttributes', + RawMessageDelivery: 'false', + }) + + // A subscription that is already reset is not written to again + const sendSpy = vi.spyOn(snsClient, 'send') + await assertSubscription( + snsClient, + topicArn, + queueArn, + { Attributes: { FilterPolicy: `{"type":["add"]}` }, updateAttributesIfExists: true }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + expect( + sendSpy.mock.calls.some(([command]) => command instanceof SetSubscriptionAttributesCommand), + ).toBe(false) + }) + + // LocalStack cannot remove a RedrivePolicy: it rejects both an empty and an omitted value + it.skipIf(isLocalstack)( + 'removes a managed redrive policy missing from the config', + async () => { + const topicArn = await testAdmin.createTopic(TOPIC_NAME) + const { queueArn } = await testAdmin.createQueue(QUEUE_NAME) + const subscriptionArn = await assertSubscription( + snsClient, + topicArn, + queueArn, + { + Attributes: { RedrivePolicy: JSON.stringify({ deadLetterTargetArn: queueArn }) }, + updateAttributesIfExists: false, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ) + const sendSpy = vi.spyOn(snsClient, 'send') + + await assertSubscription( + snsClient, + topicArn, + queueArn, + { updateAttributesIfExists: true }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + new FakeLogger(), + ) + + const redrivePolicyWrite = sendSpy.mock.calls + .map(([command]) => command) + .find( + (command) => + command instanceof SetSubscriptionAttributesCommand && + command.input.AttributeName === 'RedrivePolicy', + ) + expect(redrivePolicyWrite?.input).toEqual({ + SubscriptionArn: subscriptionArn, + AttributeName: 'RedrivePolicy', + AttributeValue: undefined, + }) + const subscriptionAttributes = await getSubscriptionAttributes(snsClient, subscriptionArn!) + expect( + subscriptionAttributes.result?.attributes?.RedrivePolicy || undefined, + ).toBeUndefined() + }, + ) + + it('throws when a configured attribute is not managed', async () => { + const topicArn = await testAdmin.createTopic(TOPIC_NAME) + const { queueArn } = await testAdmin.createQueue(QUEUE_NAME) + + await expect( + assertSubscription( + snsClient, + topicArn, + queueArn, + { + Attributes: { FilterPolicy: `{"type":["add"]}`, RawMessageDelivery: 'true' }, + managedAttributes: ['FilterPolicy'], + updateAttributesIfExists: true, + }, + { queueName: QUEUE_NAME, topicName: TOPIC_NAME }, + ), + ).rejects.toThrow(/RawMessageDelivery are set, but not listed in managedAttributes/) + }) + it('throws when attributes are different and update is disabled', async () => { const logger = new FakeLogger() const topicArn = await testAdmin.createTopic(TOPIC_NAME) @@ -270,4 +498,29 @@ describe('snsSubscriber', () => { ) }) }) + + describe('setSubscriptionAttributes', () => { + it('writes only the attributes that differ from the current ones', async () => { + const topicArn = await testAdmin.createTopic(TOPIC_NAME) + const { queueArn } = await testAdmin.createQueue(QUEUE_NAME) + const subscriptionArn = await testAdmin.createSubscription(topicArn, queueArn) + await setSubscriptionAttributes(snsClient, subscriptionArn, { + FilterPolicy: `{"type":["add"]}`, + }) + const sendSpy = vi.spyOn(snsClient, 'send') + + await setSubscriptionAttributes( + snsClient, + subscriptionArn, + { FilterPolicy: `{ "type": ["add"] }`, RawMessageDelivery: 'true' }, + ['FilterPolicy', 'RawMessageDelivery'], + ) + + const writtenAttributes = sendSpy.mock.calls + .map(([command]) => command) + .filter((command) => command instanceof SetSubscriptionAttributesCommand) + .map((command) => command.input.AttributeName) + expect(writtenAttributes).toEqual(['RawMessageDelivery']) + }) + }) }) diff --git a/packages/sns/lib/utils/snsSubscriber.ts b/packages/sns/lib/utils/snsSubscriber.ts index 56df6d3f9..a2e86a29b 100644 --- a/packages/sns/lib/utils/snsSubscriber.ts +++ b/packages/sns/lib/utils/snsSubscriber.ts @@ -1,3 +1,4 @@ +import { isDeepStrictEqual } from 'node:util' import type { SNSClient, SubscribeCommandInput } from '@aws-sdk/client-sns' import { SetSubscriptionAttributesCommand, SubscribeCommand } from '@aws-sdk/client-sns' import type { CreateQueueCommandInput, SQSClient } from '@aws-sdk/client-sqs' @@ -17,31 +18,55 @@ import { isSNSTopicLocatorType, type TopicResolutionOptions, } from '../types/TopicTypes.ts' -import { assertTopic, findSubscriptionByTopicAndQueue } from './snsUtils.ts' +import { assertTopic, findConfirmedSubscriptionArn, getSubscriptionAttributes } from './snsUtils.ts' import { buildTopicArn } from './stsUtils.ts' +/** Subscription attributes that can be managed for an SQS subscription */ +export const SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES = [ + 'FilterPolicy', + 'FilterPolicyScope', + 'RawMessageDelivery', + 'RedrivePolicy', +] as const + +export type SubscriptionManagedAttributeName = (typeof SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES)[number] + +type SubscriptionAttributeManagementOptions = { + /** + * Attributes whose value is owned by this config. Defaults to all of + * {@link SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES}. + * + * A managed attribute that is missing from `Attributes` is reset on the existing subscription + * (`FilterPolicy` and `RedrivePolicy` are removed, the others go back to their defaults). + * Attributes that are not managed are neither checked nor written, so external tooling such as + * Terraform can own them, and must not be present in `Attributes`. + */ + managedAttributes?: readonly SubscriptionManagedAttributeName[] +} + /** * Options for the subscription of a queue to a topic. * * There are two modes for this config: * 1. Creation (default): the subscription is created, or updated if it already exists. * 2. Locate only: for subscriptions managed externally. The subscription is only located, never created, - * and the given attributes (e.g. `FilterPolicy`) are applied to it. + * and its managed attributes (e.g. `FilterPolicy`) are applied to it. */ export type SNSSubscriptionOptions = /** Creation */ - | (Omit & { - updateAttributesIfExists: boolean - /** Should not be present when the subscription is created */ - locateOnly?: never - }) + | (Omit & + SubscriptionAttributeManagementOptions & { + updateAttributesIfExists: boolean + /** Should not be present when the subscription is created */ + locateOnly?: never + }) /** Locate only */ - | { + | (SubscriptionAttributeManagementOptions & { /** Marks the subscription as located only */ locateOnly: true /** Attributes applied to the located subscription */ Attributes?: SubscribeCommandInput['Attributes'] - } + }) /** Subscription options that create the subscription, required when subscribing */ export type SNSSubscriptionCreationOptions = Exclude @@ -116,9 +141,10 @@ export async function subscribeToTopic( } /** - * Subscribes an existing SQS queue to an existing SNS topic. Subscribing is idempotent, so an - * existing subscription is returned as is, or has its attributes updated when they differ and - * `updateAttributesIfExists` is enabled. + * Subscribes an existing SQS queue to an existing SNS topic. An existing subscription is checked + * with read-only calls first and is only written to when its managed attributes differ from the + * configured ones and `updateAttributesIfExists` is enabled. Only the differing attributes are + * written. */ export async function assertSubscription( snsClient: SNSClient, @@ -128,42 +154,68 @@ export async function assertSubscription( errorContext: { queueName?: string; topicName?: string }, logger?: CommonLogger, ): Promise { - const subscribeCommand = new SubscribeCommand({ - TopicArn: topicArn, - Endpoint: queueArn, - Protocol: 'sqs', - ReturnSubscriptionArn: true, - ...subscriptionConfiguration, - }) + const { updateAttributesIfExists, managedAttributes, locateOnly, ...subscribeInput } = + subscriptionConfiguration + const resolvedManagedAttributes = resolveManagedAttributes( + subscribeInput.Attributes, + managedAttributes, + ) + const resolvedLogger = logger ?? console + const errMessagePrefix = `Error while creating subscription for queue "${errorContext.queueName}", topic "${errorContext.topicName}"` + + const reconcileExistingSubscription = async (subscriptionArn: string) => { + const changedAttributes = await findChangedSubscriptionAttributes( + snsClient, + subscriptionArn, + subscribeInput.Attributes, + resolvedManagedAttributes, + ) + if (Object.keys(changedAttributes).length === 0) return subscriptionArn + + const errMessage = `${errMessagePrefix}: Subscription already exists with different attributes: ${Object.keys(changedAttributes).join(', ')}` + if (!updateAttributesIfExists) { + resolvedLogger.error(errMessage) + throw new InternalError({ + errorCode: 'sns_subscription_creation_failed', + message: errMessage, + details: { queueName: errorContext.queueName, topicArn }, + }) + } + + resolvedLogger.warn(`${errMessage}. Updating subscription`) + await writeSubscriptionAttributes(snsClient, subscriptionArn, changedAttributes) + return subscriptionArn + } + + const existingSubscriptionArn = await findConfirmedSubscriptionArn(snsClient, topicArn, queueArn) + if (existingSubscriptionArn) return reconcileExistingSubscription(existingSubscriptionArn) try { - const subscriptionResult = await snsClient.send(subscribeCommand) + const subscriptionResult = await snsClient.send( + new SubscribeCommand({ + ...subscribeInput, + TopicArn: topicArn, + Endpoint: queueArn, + Protocol: 'sqs', + ReturnSubscriptionArn: true, + }), + ) return subscriptionResult.SubscriptionArn } catch (err) { if (!isError(err)) throw err - const resolvedLogger = logger ?? console - const errMessage = `Error while creating subscription for queue "${errorContext.queueName}", topic "${errorContext.topicName}": ${err.message}` - - if ( - subscriptionConfiguration.updateAttributesIfExists && - err.message.includes('Subscription already exists with different attributes') - ) { - resolvedLogger.warn(`${errMessage}. Trying to update subscription`) - - const result = await tryToUpdateSubscription( + // Another instance may have created the subscription after the lookup above + if (err.message.includes('Subscription already exists with different attributes')) { + const concurrentSubscriptionArn = await findConfirmedSubscriptionArn( snsClient, topicArn, queueArn, - subscriptionConfiguration, ) - if (result) return result.SubscriptionArn - - resolvedLogger.error('Failed to update subscription') - } else { - resolvedLogger.error(errMessage) + if (concurrentSubscriptionArn) return reconcileExistingSubscription(concurrentSubscriptionArn) } + const errMessage = `${errMessagePrefix}: ${err.message}` + resolvedLogger.error(errMessage) throw new InternalError({ errorCode: 'sns_subscription_creation_failed', message: errMessage, @@ -173,42 +225,133 @@ export async function assertSubscription( } } -async function tryToUpdateSubscription( +/** + * Applies the given attributes to an existing subscription, resetting managed attributes that are + * missing from them. Current attributes are read first, and only the ones that differ are written. + */ +export async function setSubscriptionAttributes( snsClient: SNSClient, - topicArn: string, - queueArn: string, - subscriptionConfiguration: SNSSubscriptionCreationOptions, -) { - const subscription = await findSubscriptionByTopicAndQueue(snsClient, topicArn, queueArn) - if (!subscription?.SubscriptionArn || !subscriptionConfiguration.Attributes) { - return undefined - } - - await setSubscriptionAttributes( + subscriptionArn: string, + attributes: SubscribeCommandInput['Attributes'], + managedAttributes?: readonly SubscriptionManagedAttributeName[], +): Promise { + const changedAttributes = await findChangedSubscriptionAttributes( snsClient, - subscription.SubscriptionArn, - subscriptionConfiguration.Attributes, + subscriptionArn, + attributes, + resolveManagedAttributes(attributes, managedAttributes), ) - - return subscription + await writeSubscriptionAttributes(snsClient, subscriptionArn, changedAttributes) } -/** - * Applies the given attributes to an existing subscription, one by one as SNS only allows setting a single - * attribute per call. - */ -export async function setSubscriptionAttributes( +// SNS only allows setting a single attribute per call +async function writeSubscriptionAttributes( snsClient: SNSClient, subscriptionArn: string, - attributes: NonNullable, + attributes: Record, ): Promise { for (const [key, value] of Object.entries(attributes)) { await snsClient.send( new SetSubscriptionAttributesCommand({ SubscriptionArn: subscriptionArn, AttributeName: key, - AttributeValue: value, + // AWS rejects an empty RedrivePolicy, omitting the value is what removes it + AttributeValue: key === 'RedrivePolicy' && value === '' ? undefined : value, }), ) } } + +// Values that reset an attribute to the state of a subscription where it was never set +const SUBSCRIPTION_ATTRIBUTE_UNSET_VALUES: Record = { + FilterPolicy: '{}', + FilterPolicyScope: 'MessageAttributes', + RawMessageDelivery: 'false', + RedrivePolicy: '', +} + +function isManagedAttributeName(name: string): name is SubscriptionManagedAttributeName { + return (SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES as readonly string[]).includes(name) +} + +function resolveManagedAttributes( + attributes: Record | undefined, + managedAttributes: readonly SubscriptionManagedAttributeName[] | undefined, +): readonly SubscriptionManagedAttributeName[] { + const resolvedManagedAttributes = managedAttributes ?? SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES + const unmanagedAttributes = Object.keys(attributes ?? {}).filter( + (name) => isManagedAttributeName(name) && !resolvedManagedAttributes.includes(name), + ) + if (unmanagedAttributes.length > 0) { + throw new InternalError({ + errorCode: 'invalid_subscription_configuration', + message: `Subscription attributes ${unmanagedAttributes.join(', ')} are set, but not listed in managedAttributes`, + }) + } + + return resolvedManagedAttributes +} + +/** + * Compares the current attributes of the subscription with the expected ones. Every managed + * attribute is compared, falling back to its unset value when it is not configured. Configured + * attributes outside {@link SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES} are compared, but never reset. + */ +async function findChangedSubscriptionAttributes( + snsClient: SNSClient, + subscriptionArn: string, + attributes: Record | undefined, + managedAttributes: readonly SubscriptionManagedAttributeName[], +): Promise> { + const expectedAttributes: Record = {} + // Iterating in the declared order writes FilterPolicy before FilterPolicyScope + for (const name of SUBSCRIPTION_MANAGED_ATTRIBUTE_NAMES) { + if (managedAttributes.includes(name)) { + expectedAttributes[name] = attributes?.[name] ?? SUBSCRIPTION_ATTRIBUTE_UNSET_VALUES[name] + } + } + for (const [name, value] of Object.entries(attributes ?? {})) { + if (!isManagedAttributeName(name)) expectedAttributes[name] = value + } + if (Object.keys(expectedAttributes).length === 0) return {} + + const currentAttributesResult = await getSubscriptionAttributes(snsClient, subscriptionArn) + const currentAttributes = currentAttributesResult.result?.attributes ?? {} + + const changedAttributes = Object.fromEntries( + Object.entries(expectedAttributes).filter( + ([name, value]) => !isSameAttributeValue(name, value, currentAttributes[name]), + ), + ) + + // Filter policy scope means nothing without a filter policy, so it is not reset on its own + const resultingFilterPolicy = expectedAttributes.FilterPolicy ?? currentAttributes.FilterPolicy + if ( + isUnsetAttributeValue('FilterPolicy', resultingFilterPolicy) && + isUnsetAttributeValue('FilterPolicyScope', changedAttributes.FilterPolicyScope) + ) { + delete changedAttributes.FilterPolicyScope + } + + return changedAttributes +} + +function isUnsetAttributeValue(name: string, value: string | undefined): boolean { + if (value === undefined || value === '') return true + if (name === 'FilterPolicy' && value === '{}') return true + + return isManagedAttributeName(name) && SUBSCRIPTION_ATTRIBUTE_UNSET_VALUES[name] === value +} + +// JSON attributes are compared structurally, as SNS does not preserve their formatting +function isSameAttributeValue(name: string, expected: string, current: string | undefined) { + if (expected === current) return true + if (isUnsetAttributeValue(name, expected) && isUnsetAttributeValue(name, current)) return true + if (current === undefined || !name.endsWith('Policy')) return false + + try { + return isDeepStrictEqual(JSON.parse(expected), JSON.parse(current)) + } catch { + return false + } +} diff --git a/packages/sns/lib/utils/snsUtils.ts b/packages/sns/lib/utils/snsUtils.ts index 5d7c8b827..0960bd644 100644 --- a/packages/sns/lib/utils/snsUtils.ts +++ b/packages/sns/lib/utils/snsUtils.ts @@ -190,6 +190,18 @@ export async function findSubscriptionByTopicAndQueue( return undefined } +export async function findConfirmedSubscriptionArn( + snsClient: SNSClient, + topicArn: string, + queueArn: string, +): Promise { + const subscription = await findSubscriptionByTopicAndQueue(snsClient, topicArn, queueArn) + const subscriptionArn = subscription?.SubscriptionArn + + // Unconfirmed subscriptions are listed with a status placeholder instead of an ARN + return subscriptionArn?.startsWith('arn:') ? subscriptionArn : undefined +} + /** * Calculates the size of an outgoing SNS message. * diff --git a/packages/sns/test/utils/fauxqsInstance.ts b/packages/sns/test/utils/fauxqsInstance.ts index c2dafa5f0..4a6366098 100644 --- a/packages/sns/test/utils/fauxqsInstance.ts +++ b/packages/sns/test/utils/fauxqsInstance.ts @@ -1,7 +1,7 @@ /** biome-ignore-all lint/suspicious/noConsole: test **/ import { type FauxqsServer, startFauxqs } from 'fauxqs' -const isLocalstack = process.env.QUEUE_BACKEND === 'localstack' +export const isLocalstack = process.env.QUEUE_BACKEND === 'localstack' const LOCALSTACK_PORT = 4566 const LOCALSTACK_HOST = 'localstack' const FAUXQS_PORT = 4567 From afd03d29dc1698d3ed6e264287a5ea86eb422a4f Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Fri, 2 Oct 2026 10:11:52 +0200 Subject: [PATCH 8/9] chore: fauxqs bump --- packages/s3-payload-store/package.json | 2 +- packages/sns/package.json | 2 +- packages/sqs/package.json | 2 +- pnpm-lock.yaml | 18 +++++++++--------- pnpm-workspace.yaml | 1 + 5 files changed, 13 insertions(+), 12 deletions(-) diff --git a/packages/s3-payload-store/package.json b/packages/s3-payload-store/package.json index 4bc14a5f9..863cc6cd1 100644 --- a/packages/s3-payload-store/package.json +++ b/packages/s3-payload-store/package.json @@ -41,7 +41,7 @@ "@message-queue-toolkit/core": "workspace:*", "@types/node": "^26.6.1", "@vitest/coverage-v8": "^5.0.1", - "fauxqs": "^2.12.0", + "fauxqs": "^2.12.1", "rimraf": "^6.1.3", "typescript": "^7.0.2", "vitest": "^5.0.1" diff --git a/packages/sns/package.json b/packages/sns/package.json index 7ffdf12fd..b8514f926 100644 --- a/packages/sns/package.json +++ b/packages/sns/package.json @@ -59,7 +59,7 @@ "@vitest/coverage-v8": "^5.0.1", "awilix": "^13.0.5", "awilix-manager": "^7.0.1", - "fauxqs": "^2.12.0", + "fauxqs": "^2.12.1", "ioredis": "^5.11.1", "rimraf": "^6.1.3", "typescript": "^7.0.2", diff --git a/packages/sqs/package.json b/packages/sqs/package.json index 97dbf42da..1e0db9b72 100644 --- a/packages/sqs/package.json +++ b/packages/sqs/package.json @@ -54,7 +54,7 @@ "@vitest/coverage-v8": "^5.0.1", "awilix": "^13.0.5", "awilix-manager": "^7.0.1", - "fauxqs": "^2.12.0", + "fauxqs": "^2.12.1", "ioredis": "^5.11.1", "rimraf": "^6.1.3", "typescript": "^7.0.2", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8410d7d35..241405e9e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -424,8 +424,8 @@ importers: specifier: ^5.0.1 version: 5.0.1(vitest@5.0.1) fauxqs: - specifier: ^2.12.0 - version: 2.12.0(typescript@7.0.2) + specifier: ^2.12.1 + version: 2.12.1(typescript@7.0.2) rimraf: specifier: ^6.1.3 version: 6.1.3 @@ -524,8 +524,8 @@ importers: specifier: ^7.0.1 version: 7.0.1(awilix@13.0.5) fauxqs: - specifier: ^2.12.0 - version: 2.12.0(typescript@7.0.2) + specifier: ^2.12.1 + version: 2.12.1(typescript@7.0.2) ioredis: specifier: ^5.11.1 version: 5.11.1(supports-color@7.2.0) @@ -591,8 +591,8 @@ importers: specifier: ^7.0.1 version: 7.0.1(awilix@13.0.5) fauxqs: - specifier: ^2.12.0 - version: 2.12.0(typescript@7.0.2) + specifier: ^2.12.1 + version: 2.12.1(typescript@7.0.2) ioredis: specifier: ^5.11.1 version: 5.11.1(supports-color@7.2.0) @@ -1704,8 +1704,8 @@ packages: fastq@1.20.3: resolution: {integrity: sha512-XKv5nnLs6nLF71NgiKJLIZFLkPyIEuOselLG7ujZnGrRfQK8HpvY+WqKhAJUAdLomwVHErVS4LfxFlPq0/FTAw==} - fauxqs@2.12.0: - resolution: {integrity: sha512-8Q+xlQbIBkwGxrMI66oXewnTJYmlONirFb2wV9TkZTy/WLePrKiJj/F4zZjtwhVUtt4QIveCia9tdE58PxybaQ==} + fauxqs@2.12.1: + resolution: {integrity: sha512-KrlSflFgpwIxkp2h83vf+g6GEWckTjepQnjo8VimMHoJo5UhxX2HdXA9+DycO5mPlqxJIKzKYqChCdWuKTpMVQ==} engines: {node: '>=22.5.0', pnpm: '>=11'} hasBin: true @@ -3617,7 +3617,7 @@ snapshots: dependencies: reusify: 1.1.0 - fauxqs@2.12.0(typescript@7.0.2): + fauxqs@2.12.1(typescript@7.0.2): dependencies: '@aws-sdk/client-s3': 3.1134.0 '@aws-sdk/client-sns': 3.1134.0 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index d1a24c524..8613589bd 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -24,6 +24,7 @@ minimumReleaseAgeExclude: - '@aws-sdk/client-s3@3.1106.0' - '@aws-sdk/client-sns@3.1106.0' - '@aws-sdk/client-sqs@3.1106.0' + - fauxqs@2.12.1 # Hoist all dependencies to the workspace root to preserve npm-compatible # resolution behavior for transitive deps used directly in test files. From f9170f51f65f557b6459e167b6142035929bb7af Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Fri, 2 Oct 2026 10:56:36 +0200 Subject: [PATCH 9/9] readme fix --- packages/sns/README.md | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/packages/sns/README.md b/packages/sns/README.md index b138961e4..e3f9acb15 100644 --- a/packages/sns/README.md +++ b/packages/sns/README.md @@ -531,10 +531,13 @@ subscriptionConfig: { With `subscriptionDeadLetterQueue.reuseConsumerDeadLetterQueue`, `RedrivePolicy` is set from the consumer DLQ and is excluded from the managed attributes. -With `subscriptionConfig: { locateOnly: true, Attributes }` the subscription is located, never created, and the given -managed attributes (e.g. `FilterPolicy`) are applied to it on startup when they differ from the current ones. This lets external tooling own the subscription while -the application keeps its filter policy in sync with the consumer handlers. A locate-only subscription requires both -the topic and the queue to be located, and none of the located resources are deleted when `deletionConfig` is set. +With `subscriptionConfig: { locateOnly: true, Attributes }` the subscription is located, never created, and its +managed attributes are applied to it on startup when they differ from the current ones. Managed attributes missing +from `Attributes` are reset as well, so set `managedAttributes` to the ones the application owns (usually +`['FilterPolicy', 'FilterPolicyScope']`), otherwise attributes set by the external tooling, such as `RedrivePolicy`, +are removed. This lets external tooling own the subscription while the application keeps its filter policy in sync +with the consumer handlers. A locate-only subscription requires both the topic and the queue to be located, and none +of the located resources are deleted when `deletionConfig` is set. > **Deprecated**: `locatorConfig.subscriptionArn` is no longer needed, as the subscription is located from the topic > and the queue. When it is set, every resource is located and `creationConfig` and `subscriptionConfig` are ignored, @@ -571,6 +574,7 @@ the topic and the queue to be located, and none of the located resources are del locatorConfig: { topicName: 'my-topic', queueName: 'my-queue' }, subscriptionConfig: { locateOnly: true, + managedAttributes: ['FilterPolicy', 'FilterPolicyScope'], Attributes: { FilterPolicy: JSON.stringify({ messageType: ['user.created'] }) }, }, } @@ -762,7 +766,11 @@ const consumer = new MyConsumer(deps, { }, }, // No subscriptionConfig - the subscription is located, never created. Alternatively, keep managing its filter policy: - // subscriptionConfig: { locateOnly: true, Attributes: { FilterPolicy: '...' } }, + // subscriptionConfig: { + // locateOnly: true, + // managedAttributes: ['FilterPolicy', 'FilterPolicyScope'], + // Attributes: { FilterPolicy: '...' }, + // }, }) // This will: @@ -861,7 +869,11 @@ SNS consumers use the same options as SQS consumers, plus SNS-specific subscript }, }, // or, for a subscription managed externally (requires topic and queue locators) - // subscriptionConfig: { locateOnly: true, Attributes: { FilterPolicy: '...' } }, + // subscriptionConfig: { + // locateOnly: true, + // managedAttributes: ['FilterPolicy', 'FilterPolicyScope'], + // Attributes: { FilterPolicy: '...' }, + // }, // Optional - FIFO Configuration fifoQueue: false,