Repository navigation
chore: SNS support for SQS 26 - #445
Conversation
📝 WalkthroughWalkthroughThis PR refactors SNS/SQS resource management and non-blocking initialization. Internal ARN/URL/name fields are encapsulated as private state objects ( ChangesResource encapsulation and readiness gating
Implementation and test updates to use new resource contracts
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 ESLint
ESLint skipped: no ESLint configuration detected in root package.json. To enable, add Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
kibertoad
left a comment
kibertoad
left a comment
There was a problem hiding this comment.
Drive-by review of the SQS-26 mirroring work. Two P0 correctness issues called out inline (close() leak, queueArn! assertion), plus a few P1/P2 follow-ups. A combined UPGRADING.md migration note for the SQS-26 + SNS protected-field rename would also be worth adding before merge.
| * callback). Subclasses read via the {@link subscription} getter so the | ||
| * ARNs are impossible to access in an "uninitialised" state. | ||
| */ | ||
| private _subscription?: SubscriptionResource |
There was a problem hiding this comment.
P0 — close() leaks this field, defeating the stated invariant.
The class inherits AbstractSqsConsumer.close() → AbstractSqsService.close(), which only resets _queue and isInitted. _subscription (added here) is never cleared, nor are resourcesReady or startRequested. So immediately after close():
consumer.subscription.topicArnstill returns the stale ARN — the docstring above says the ARN should be "impossible to access in an 'uninitialised' state", but it is.resourcesReady === truecan mislead a re-init code path.startRequestedmay incorrectly carry across init cycles.
Suggested fix — add an override:
public override async close(abort?: boolean): Promise<void> {
this._subscription = undefined
this.resourcesReady = false
this.startRequested = false
await super.close(abort)
}There was a problem hiding this comment.
Thanks for catching this — totally right. Added a close() override in 4d9d523 that resets _subscription, resourcesReady and startRequested before delegating to super.close(abort).
| topicArn: subscriptionTopicArn, | ||
| queueUrl, | ||
| // biome-ignore lint/style/noNonNullAssertion: queueAvailable is checked above | ||
| queueArn: queueArn!, |
There was a problem hiding this comment.
P0 — queueArn! is unsafe.
markQueueAvailable (just below) sets queueAvailable = true whenever getQueueAttributes doesn't throw, but attrs.result?.attributes?.QueueArn can be undefined (the Attributes map only contains requested names, and SDK responses with an empty map have been seen in the wild). When that happens, onResourcesReady fires with queueArn: undefined, setQueueResource happily stores it, and any later this.queue.arn access returns undefined — silently.
Mirror the explicit check used on the synchronous path at the bottom of initSnsSqs:
if (!queueArn) {
extraParams?.onResourcesError?.(
new Error(`Could not resolve QueueArn for queue at ${queueUrl}`),
{ isFinal: true },
)
return
}
queueAvailable = true…then the non-null assertion here goes away.
There was a problem hiding this comment.
Thanks, good spot. Fixed in 4d9d523: markQueueAvailable now reports via onResourcesError (final) when the QueueArn fetch returns an empty Attributes map, instead of silently flipping the queue to "ready". Took the chance to drop the separate queueAvailable flag and use queueArn truthiness as the ready signal, so the ! assertion in notifyIfBothReady goes away too.
|
|
||
| const markQueueAvailable = async () => { | ||
| try { | ||
| const attrs = await getQueueAttributes(sqsClient, queueUrl, ['QueueArn']) |
There was a problem hiding this comment.
P2 — extra GetQueueAttributes round-trip per init.
pollForQueue's checkFn already calls getQueueAttributes(queueUrl) with default ['All'] (which includes QueueArn) but throws the result away because pollForQueue is typed Promise<unknown | undefined>. Same applies to the synchronous getQueueAttributes(..., ['QueueArn']) at L482.
Threading the result through (change pollForQueue return type to Promise<QueueAttributesResult | undefined>) lets you reuse it and eliminates both extra round-trips. Modest gain per startup but it compounds with the new background polling path.
There was a problem hiding this comment.
Thanks. Done in c4e60ab:
QueueAttributesResultis now exported from@message-queue-toolkit/sqs.pollForQueueis typedPromise<QueueAttributesResult | undefined>and threads the attrs throughonResourceAvailable(attrs).markQueueAvailableis now a sync helper that readsQueueArnfrom the attrs it receives — no separate fetch.- The non-polling branch (
Promise.all) capturesqueueCheckResult.resultinto a sharedqueueAttributesvariable. - The L482 final-return fetch is gone — it reads
QueueArnfrom the capturedqueueAttributesinstead.
Net: 2 fewer round-trips per init in the locator-only polling path, 1 fewer in the non-polling path.
| @@ -152,8 +165,8 @@ export abstract class AbstractSnsSqsConsumer< | |||
| if (this.startRequested) { | |||
| this.logger.info({ | |||
| message: 'Resources now ready, starting consumers', | |||
There was a problem hiding this comment.
P1 — log loses context for locator-only setups.
When the consumer was wired with locatorConfig.topicName (not topicArn), the resolved ARN isn't reachable here yet (_subscription is set in the same callback, but the log is built before this.subscription is safe to read — and even after, you can't read it for the startRequested branch at L229).
Same issue at L229: topicArn: this.locatorConfig?.topicArn will be undefined for any locator using topicName, or for a locator with only subscriptionArn.
Prefer reading from result.topicArn here (already in scope), and at L229 either resolve and cache the topic ARN once via buildTopicArn(...) for diagnostics, or omit the field when unavailable rather than emitting undefined.
There was a problem hiding this comment.
Thanks. Fixed in 2dd1123:
- L167 callback log now reads
queueNamefromresulttoo (same source of truth asresult.topicArnalready was). - L229
start()log uses a newbuildConfiguredResourceLogContext()helper that emits only the identifiers we can derive from config —queueName, plustopicArnortopicNameas fallback — and omits the field entirely when unknown (so locator-only-with-subscriptionArnno longer logstopicArn: undefined).
Skipped the buildTopicArn cache path for now — felt heavier than the diagnostic warrants.
| return this._subscription | ||
| } | ||
|
|
||
| protected setSubscriptionResource(resource: SubscriptionResource): void { |
There was a problem hiding this comment.
P3 — asymmetry: setQueueResource and setTopicResource flip isInitted, this one doesn't.
It's correct today because setSubscriptionResource is always paired with setQueueResource in the flow, but the next maintainer who reaches for this from a custom subclass will silently leave isInitted = false. Either:
- add a one-line comment explaining the asymmetry, or
- make it set
isInitted = truefor symmetry (cheap idempotent set).
There was a problem hiding this comment.
Thanks. Done in 9f2328a — setSubscriptionResource now flips isInitted = true for symmetry with setQueueResource / setTopicResource. Idempotent at the existing call sites (always paired with setQueueResource).
| deadLetterQueueUrl: this.deadLetterQueueUrl, | ||
| // topicArn / subscriptionArn / _queue are populated together by init; | ||
| // if the queue resource isn't resolved yet, none of the others are either. | ||
| try { |
There was a problem hiding this comment.
P1 — this try/catch pattern is duplicated across 5 consumer files and conflates two failure modes.
Same block appears in SnsSqsPermissionConsumer.ts, SnsSqsPermissionConsumerFifo.ts, SnsSqsEntityConsumer.ts, CreateLocateConfigMixConsumer.ts, and the polling spec's TestStartupResourcePollingConsumer. Two suggestions:
- Lift it into
AbstractSnsSqsConsumer(e.g.,protected getResourceHandles(): { … } | undefinedor anisResourcesReadygetter) so test consumers don't each catch errors. - Inside the helper, prefer probing private state (
if (!this._subscription || !this._queue) return undefined) over try/catch — that way unexpected getter exceptions still bubble up instead of being silently swallowed into an all-undefinedshape that looks like "not initialised yet".
There was a problem hiding this comment.
Thanks. Done in 3efb46e:
- Added
protected get areResourcesReadyonAbstractSnsSqsConsumer(exposes the existing privateresourcesReadyflag). - All five test consumers'
subscriptionPropsgetters now short-circuit on!this.areResourcesReadyand read the throwingqueue/subscriptiongetters directly when ready. No more try/catch, so unexpected exceptions propagate as you suggested instead of being flattened into an all-undefinedshape that looks like "not initialised yet".
- AbstractSnsSqsConsumer.close() now resets _subscription, resourcesReady and startRequested so the subscription getter does not return a stale ARN after close and a re-init does not observe leftover state. - initSnsSqs non-blocking path: markQueueAvailable now reports via onResourcesError when QueueArn cannot be resolved instead of advertising the queue as ready with an undefined ARN. notifyIfBothReady gates on queueArn truthiness so the non-null assertion is no longer needed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
When a consumer is wired with locator-only and only `topicName` (or just `subscriptionArn`) is provided, `this.locatorConfig?.topicArn` was `undefined` in the "Start requested but resources not ready yet" log, producing a noisy `topicArn: undefined` line with no usable identifier. Replace the inline log payload with `buildConfiguredResourceLogContext()` which only emits the identifiers we actually have from config (queueName, topicArn or topicName as fallback). The earlier `Resources now ready` log inside `onResourcesReady` now sources `queueName` from `result` for the same reason — `result` already carries the resolved identifiers. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Five test consumers reproduced the same try/catch around the throwing `queue` / `subscription` getters to coerce a "not ready yet" failure into an all-undefined shape, which also swallowed any unexpected exceptions. Expose `protected get areResourcesReady` on `AbstractSnsSqsConsumer` so callers that legitimately run before resources are populated can short-circuit explicitly. Unexpected getter exceptions now propagate instead of being silently flattened. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…ore/sns_support_sqs_26
…trips `pollForQueue.checkFn` already calls `getQueueAttributes(queueUrl)` (which includes `QueueArn` in the default `['All']` response) but the helper was typed `Promise<unknown | undefined>` so callers couldn't reuse the result. `markQueueAvailable` then re-fetched the ARN, and the synchronous path at the bottom of `initSnsSqs` re-fetched it a third time. - Type `pollForQueue` as `Promise<QueueAttributesResult | undefined>` and pass the attrs through the `onResourceAvailable` callback. - `markQueueAvailable` is now a pure synchronous helper that reads the ARN from the attrs it was given. - The non-polling branch captures `queueCheckResult.result` into a shared `queueAttributes` variable. - The final return reads `QueueArn` from `queueAttributes` — no extra GetQueueAttributes call. Net: 2 fewer GetQueueAttributes round-trips per init in the locator-only polling path, 1 fewer in the non-polling path. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Matches setQueueResource / setTopicResource so a maintainer reaching for setSubscriptionResource from a custom subclass doesn't silently leave isInitted = false. Idempotent — when paired with setQueueResource (the existing call sites) it just flips the flag once. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/sns/lib/sns/AbstractSnsSqsConsumer.ts (1)
156-179:⚠️ Potential issue | 🟠 Major | ⚡ Quick winDLQ init failure is swallowed in non-blocking mode, but blocks in synchronous mode.
The promise chain structure causes consumers to start even when
initDeadLetterQueue()fails:this.initDeadLetterQueue() .catch((err) => { /* logs but doesn't rethrow */ }) .then(() => { /* runs anyway → starts consumers */ })In contrast, the synchronous path at line 203 (
await this.initDeadLetterQueue()) propagates failures. This inconsistency means non-blocking consumers may run without DLQ protection, risking message loss on processing failures.Suggested fix: propagate DLQ failure or gate consumer start
this.initDeadLetterQueue() - .catch((err) => { - this.logger.error({ - message: 'Failed to initialize dead letter queue after resources became ready', - error: err, - }) - }) .then(() => { // If start() was called while resources weren't ready, start consumers now if (this.startRequested) { this.logger.info({ message: 'Resources now ready, starting consumers', queueName: result.queueName, topicArn: result.topicArn, }) return this.startConsumers() } }) .catch((err) => { this.logger.error({ - message: 'Failed to start consumers after resources became ready', + message: 'Failed to initialize DLQ or start consumers after resources became ready', error: err, }) })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/sns/lib/sns/AbstractSnsSqsConsumer.ts` around lines 156 - 179, The current non-blocking path swallows initDeadLetterQueue() failures because the catch handler on initDeadLetterQueue() does not rethrow, allowing then() to run and call startConsumers() even if DLQ initialization failed; change the promise chain so that the catch either rethrows the error or prevents the then()-branch from executing (e.g., rethrow inside the catch or move the startConsumers() call into the try/then after a successful init), ensuring initDeadLetterQueue() failures propagate and gate startConsumers() when startRequested is true; update the handlers around initDeadLetterQueue(), startConsumers(), startRequested and logger.error calls accordingly so behavior matches the synchronous await path.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts`:
- Around line 361-369: Tests that call consumer.init() in non-blocking mode
return early and leave the consumer's background polling running; update the
affected specs that perform early-return assertions (the ones asserting
subscriptionProps fields after init) to ensure the consumer is closed afterwards
by calling await consumer.close(), ideally placed in a finally block so
background polling is always stopped even if assertions fail; locate the
instances using the consumer.init() and consumer.subscriptionProps checks and
add the finally + await consumer.close() cleanup around each test.
---
Outside diff comments:
In `@packages/sns/lib/sns/AbstractSnsSqsConsumer.ts`:
- Around line 156-179: The current non-blocking path swallows
initDeadLetterQueue() failures because the catch handler on
initDeadLetterQueue() does not rethrow, allowing then() to run and call
startConsumers() even if DLQ initialization failed; change the promise chain so
that the catch either rethrows the error or prevents the then()-branch from
executing (e.g., rethrow inside the catch or move the startConsumers() call into
the try/then after a successful init), ensuring initDeadLetterQueue() failures
propagate and gate startConsumers() when startRequested is true; update the
handlers around initDeadLetterQueue(), startConsumers(), startRequested and
logger.error calls accordingly so behavior matches the synchronous await path.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 81a175d2-8570-4f35-a7b2-fbd7c729e00f
📒 Files selected for processing (17)
UPGRADING.mdpackages/sns/lib/sns/AbstractSnsPublisher.tspackages/sns/lib/sns/AbstractSnsService.tspackages/sns/lib/sns/AbstractSnsSqsConsumer.tspackages/sns/lib/sns/fakes/TestSnsPublisher.tspackages/sns/lib/utils/snsInitter.tspackages/sns/package.jsonpackages/sns/test/consumers/CreateLocateConfigMixConsumer.tspackages/sns/test/consumers/SnsSqsEntityConsumer.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.codec.spec.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.deadLetterQueue.spec.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.spec.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.tspackages/sns/test/consumers/SnsSqsPermissionConsumerFifo.tspackages/sns/test/publishers/SnsPermissionPublisher.tspackages/sns/test/publishers/SnsPermissionPublisherFifo.ts
| // Init should return immediately even though topic doesn't exist; | ||
| // resources are populated together via the background callback, so | ||
| // none of subscriptionProps' fields are set yet. | ||
| await consumer.init() | ||
|
|
||
| // queueName should still be set | ||
| expect(consumer.subscriptionProps.queueName).toBe(queueName) | ||
| expect(consumer.subscriptionProps.queueName).toBeUndefined() | ||
| expect(consumer.subscriptionProps.queueUrl).toBeUndefined() | ||
| expect(consumer.subscriptionProps.topicArn).toBeUndefined() | ||
| }) |
There was a problem hiding this comment.
Close non-blocking consumers after early-return assertions.
On Line 364 and Line 393, consumer.init() returns before readiness in non-blocking mode, but these tests do not close the consumer. That can leave background polling active and create test flakiness/interference across cases. Add await consumer.close() (preferably in finally) in these early-return tests.
Proposed test cleanup pattern
- await consumer.init()
-
- expect(consumer.subscriptionProps.queueName).toBeUndefined()
- expect(consumer.subscriptionProps.queueUrl).toBeUndefined()
- expect(consumer.subscriptionProps.topicArn).toBeUndefined()
+ try {
+ await consumer.init()
+ expect(consumer.subscriptionProps.queueName).toBeUndefined()
+ expect(consumer.subscriptionProps.queueUrl).toBeUndefined()
+ expect(consumer.subscriptionProps.topicArn).toBeUndefined()
+ } finally {
+ await consumer.close()
+ }Also applies to: 390-397
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts`
around lines 361 - 369, Tests that call consumer.init() in non-blocking mode
return early and leave the consumer's background polling running; update the
affected specs that perform early-return assertions (the ones asserting
subscriptionProps fields after init) to ensure the consumer is closed afterwards
by calling await consumer.close(), ideally placed in a finally block so
background polling is always stopped even if assertions fail; locate the
instances using the consumer.init() and consumer.subscriptionProps checks and
add the finally + await consumer.close() cleanup around each test.
2 participants
Summary by CodeRabbit
Release Notes
Breaking Changes
areResourcesReadygetter for checking resource initialization status.>=26.0.0.Documentation