Implement polling for topics#388
Conversation
📝 WalkthroughWalkthroughAdds startup resource polling support: a core polling utility, new types/constants, wiring into SNS/SQS init flows (blocking and non-blocking modes with callbacks), type and API surface extensions, tests, and README documentation (duplicated section present). Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Consumer as Consumer / Service init
participant InitUtil as initSns / initSqs
participant PollUtil as waitForResource (core)
participant AWS as AWS SNS/SQS
participant Logger as Logger / ErrorReporter
Consumer->>InitUtil: start init(locatorConfig, extraParams)
alt startupResourcePolling enabled
InitUtil->>PollUtil: waitForResource(checkFn, config, resourceName, callbacks)
loop polling
PollUtil->>AWS: getAttributes / head call
AWS-->>PollUtil: not_found / found / error
alt found
PollUtil-->>InitUtil: resource result
alt non-blocking
InitUtil-->>Consumer: return resourcesReady:false
PollUtil->>InitUtil: invoke onResourceAvailable -> trigger on*Ready callback
else blocking
InitUtil-->>Consumer: return resourcesReady:true
end
else not_found
PollUtil-->>Logger: log progress/status
end
end
alt timeout
PollUtil-->>Logger: throw or report StartupResourcePollingTimeoutError
PollUtil--xInitUtil: throw or continue based on throwOnTimeout
end
else polling disabled
InitUtil->>AWS: single getAttributes
AWS-->>InitUtil: found / not found (throw)
InitUtil-->>Consumer: return resourcesReady:true
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
📜 Recent review detailsConfiguration used: defaults Review profile: CHILL Plan: Pro 📒 Files selected for processing (4)
🚧 Files skipped from review as they are similar to previous changes (1)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (19)
🔇 Additional comments (8)
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 Fix all issues with AI agents
In @packages/core/lib/types/queueOptionsTypes.ts:
- Around line 170-191: The JSDoc for ResourceAvailabilityConfig.enabled claims a
default of false but the type requires callers to provide it; make the type and
runtime consistent by changing enabled to optional (enabled?: boolean) and
ensure any code that constructs/consumes ResourceAvailabilityConfig applies a
false default when undefined, or alternatively remove the "Default: false" text
from the JSDoc; update ResourceAvailabilityConfig and the consuming
initialization logic (places that read enabled) so the type and documentation
match.
In @packages/sns/lib/utils/snsInitter.ts:
- Around line 145-147: The error thrown when topicCheckResult?.error ===
'not_found' uses the wrong variable locatorConfig.topicArn which can be
undefined when the topic was resolved by name; update the throw to reference the
resolved subscriptionTopicArn (the variable set around lines where
subscriptionTopicArn is computed) so the error message always contains the
actual ARN and gives correct context.
- Around line 282-284: The error throw in the checkResult.error === 'not_found'
branch uses the wrong variable: replace locatorConfig.topicArn with the resolved
topicArn variable (the one computed on lines ~260–261) so the thrown message
reads the actual resolved ARN; mirror the same fix applied in initSnsSqs and
update the template string in that throw (refer to checkResult,
locatorConfig.topicArn, and topicArn).
In @packages/sns/lib/utils/snsUtils.ts:
- Around line 41-43: Remove the incorrect comment about "NotFoundException" and
update the inline comment to state that SNS returns 'NotFound' for non-existent
topics; then simplify the conditional to only check err.name === 'NotFound'
(replace the current check that also tests 'NotFoundException') while keeping
the surrounding error-handling logic intact (look for the err.name check in
snsUtils.ts).
In @packages/sqs/lib/utils/sqsInitter.ts:
- Around line 106-108: The throw uses locatorConfig.queueUrl which can be
undefined if the queue was resolved by name; replace that with the resolved
queueUrl variable (the one assigned around line 81) so the error message always
shows the actual URL. Update the throw in the block checking checkResult.error
=== 'not_found' to reference queueUrl (not locatorConfig.queueUrl) and keep the
same message format to stay consistent with the waiting-enabled path.
In @README.md:
- Around line 247-322: The README contains a duplicate "Resource Availability
Polling (Eventual Consistency Mode)" section; locate the repeated block that
documents resourceAvailabilityConfig, the example consumer init usage (e.g.,
consumer.init()), and the ResourceAvailabilityTimeoutError example, then remove
the redundant copy so only one comprehensive section describing
resourceAvailabilityConfig, pollingIntervalMs, timeoutMs, environment examples,
and the ResourceAvailabilityTimeoutError handling remains.
🧹 Nitpick comments (2)
packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts (2)
19-71: Consider using imported types for test consumer options.The test consumer duplicates the structure of
locatorConfig,creationConfig, andresourceAvailabilityConfigrather than importing the actual types. This could lead to drift if the real types change.♻️ Suggested improvement
+import type { ResourceAvailabilityConfig } from '@message-queue-toolkit/core' +import type { SNSSQSCreationConfig, SNSSQSQueueLocatorType } from '../../lib/sns/AbstractSnsSqsConsumer.ts' class TestResourceAvailabilityConsumer extends AbstractSnsSqsConsumer< PERMISSIONS_ADD_MESSAGE_TYPE, undefined, undefined > { constructor( dependencies: SNSSQSConsumerDependencies, options: { - locatorConfig?: { - topicArn?: string - topicName?: string - queueUrl?: string - queueName?: string - subscriptionArn?: string - } - creationConfig?: { - queue?: { QueueName: string } - topic?: { Name: string } - } - resourceAvailabilityConfig?: { - enabled: boolean - timeoutMs?: number - pollingIntervalMs?: number - } + locatorConfig?: SNSSQSQueueLocatorType + creationConfig?: SNSSQSCreationConfig + resourceAvailabilityConfig?: ResourceAvailabilityConfig }, ) {
107-140: Test timing may be fragile in slow environments.The test waits 300ms before creating the topic, relying on specific timing. In slow CI environments or under load, this could cause flakiness if the 300ms delay is insufficient for the polling loop to start.
Consider either:
- Increasing the delay to be more conservative (e.g., 500ms or 1000ms)
- Adding logging or assertions to verify polling occurred multiple times before the resource appeared
- Using a more deterministic synchronization mechanism if the test framework supports it
However, the current approach is reasonable for integration tests with external services.
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (15)
README.mdpackages/core/lib/index.tspackages/core/lib/queues/AbstractQueueService.tspackages/core/lib/types/queueOptionsTypes.tspackages/core/lib/utils/resourceAvailabilityUtils.spec.tspackages/core/lib/utils/resourceAvailabilityUtils.tspackages/sns/lib/index.tspackages/sns/lib/sns/AbstractSnsSqsConsumer.tspackages/sns/lib/utils/snsInitter.tspackages/sns/lib/utils/snsUtils.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.tspackages/sqs/lib/index.tspackages/sqs/lib/sqs/AbstractSqsService.tspackages/sqs/lib/utils/sqsInitter.tspackages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts
🧰 Additional context used
🧬 Code graph analysis (4)
packages/core/lib/queues/AbstractQueueService.ts (1)
packages/core/lib/types/queueOptionsTypes.ts (1)
ResourceAvailabilityConfig(170-191)
packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts (3)
packages/sns/lib/sns/AbstractSnsSqsConsumer.ts (1)
SNSSQSConsumerDependencies(16-19)packages/sns/lib/utils/snsUtils.ts (2)
deleteTopic(144-158)assertTopic(79-142)packages/core/lib/utils/resourceAvailabilityUtils.ts (1)
ResourceAvailabilityTimeoutError(40-53)
packages/core/lib/utils/resourceAvailabilityUtils.ts (1)
packages/core/lib/types/queueOptionsTypes.ts (1)
ResourceAvailabilityConfig(170-191)
packages/sqs/lib/utils/sqsInitter.ts (3)
packages/core/lib/types/queueOptionsTypes.ts (1)
ResourceAvailabilityConfig(170-191)packages/sqs/lib/utils/sqsUtils.ts (2)
resolveQueueUrlFromLocatorConfig(90-109)getQueueAttributes(62-88)packages/core/lib/utils/resourceAvailabilityUtils.ts (2)
isResourceAvailabilityWaitingEnabled(160-164)waitForResource(101-155)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (19)
- GitHub Check: general (20.x, @message-queue-toolkit/metrics) / build
- GitHub Check: general (22.x, @message-queue-toolkit/metrics) / build
- GitHub Check: general (20.x, @message-queue-toolkit/outbox-core) / build
- GitHub Check: general (20.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (24.x, @message-queue-toolkit/gcs-payload-store) / build
- GitHub Check: general (22.x, @message-queue-toolkit/redis-message-deduplication-store) / build
- GitHub Check: general (20.x, @message-queue-toolkit/core) / build
- GitHub Check: general (20.x, @message-queue-toolkit/gcs-payload-store) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (20.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (22.x, @message-queue-toolkit/s3-payload-store) / build
- GitHub Check: general (20.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (24.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (22.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sqs) / build
- GitHub Check: kafka (24.x) / build
- GitHub Check: kafka (22.x) / build
🔇 Additional comments (17)
packages/sqs/lib/index.ts (1)
38-43: LGTM!The new exports for
initSqsandInitSqsExtraParamsproperly expose the resource availability polling functionality through the public API.packages/sns/lib/sns/AbstractSnsSqsConsumer.ts (1)
105-108: LGTM!The addition of
resourceAvailabilityConfigto theinitSnsSqsoptions enables eventual consistency polling during SNS+SQS initialization, which aligns with the PR's goal of handling cross-service deployment dependencies.packages/sns/lib/utils/snsUtils.ts (1)
68-70: Same verification needed for subscription error handling.This location has the same dual error name check as the topic attributes function above. The verification requested for Line 41-43 applies here as well.
packages/sqs/lib/sqs/AbstractSqsService.ts (1)
111-114: LGTM!The extraParams object properly passes
resourceAvailabilityConfigandloggertoinitSqs, enabling resource availability polling with appropriate logging during queue initialization.packages/sns/lib/index.ts (1)
28-33: LGTM! Clean type export extension.The addition of
InitSnsExtraParamsandInitSnsSqsExtraParamsto the public API is well-structured, using inline type modifiers appropriately.packages/core/lib/types/queueOptionsTypes.ts (1)
121-127: LGTM! Well-documented optional configuration.The addition of
resourceAvailabilityConfigtoCommonQueueOptionsis backward-compatible and clearly documents the eventual consistency use case for handling cross-service deployment dependencies.packages/core/lib/queues/AbstractQueueService.ts (1)
49-49: LGTM! Consistent configuration plumbing.The
resourceAvailabilityConfigfield is properly imported, declared asprotected readonly, and initialized from options. This follows the same pattern as other configuration fields inAbstractQueueService.Also applies to: 129-129, 165-165
packages/core/lib/index.ts (1)
108-114: LGTM! Appropriate public API extension.The resource availability utilities are cleanly exported with correct type annotations, making the polling functionality available to package consumers.
packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts (2)
90-98: Defensive cleanup pattern is solid.Both
beforeEachandafterEachdelete the queue and topic, ensuring test isolation even if a test fails mid-execution. This redundancy is intentional and prevents state leakage between tests.
200-242: LGTM! Backward compatibility tests are solid.These tests verify that when
resourceAvailabilityConfig.enabled = false, the consumer fails immediately rather than polling. This ensures backward compatibility with existing behavior.The error message assertions using
/does not exist/are appropriate for integration tests where exact error text may vary.packages/core/lib/utils/resourceAvailabilityUtils.spec.ts (1)
1-180: LGTM! Comprehensive test coverage for the resource availability utilities.The test suite thoroughly covers:
isResourceAvailabilityWaitingEnabledwith all config stateswaitForResourcesuccess, polling, timeout, error propagation, and logging paths- Indefinite polling scenarios (both
timeoutMsundefined andtimeoutMs: 0)ResourceAvailabilityTimeoutErrormessage compositionpackages/sqs/test/consumers/SqsPermissionConsumer.resourceAvailability.spec.ts (1)
1-181: LGTM! Well-structured integration tests for SQS resource availability.The tests effectively validate:
- Polling until queue becomes available
- Timeout behavior with
ResourceAvailabilityTimeoutError- Indefinite polling when
timeoutMsis not set- Backward compatibility (immediate failure when config is disabled or not provided)
The
TestResourceAvailabilityConsumersubclass cleanly exposesqueuePropsfor assertions.packages/sqs/lib/utils/sqsInitter.ts (1)
17-20: Clean integration of resource availability polling.The
InitSqsExtraParamstype and conditional polling logic are well-implemented. ThecheckFncorrectly handles thenot_founderror fromgetQueueAttributesand extracts theQueueArnfrom successful results.Also applies to: 72-101
packages/sns/lib/utils/snsInitter.ts (1)
21-27: Clean integration of resource availability polling for SNS.The new types and conditional polling logic are well-implemented. Sequential waiting for topic and queue (rather than parallel) is reasonable as it provides clearer error messages and resource-specific logging.
Also applies to: 103-130, 266-278
packages/core/lib/utils/resourceAvailabilityUtils.ts (3)
117-154: Well-implemented polling loop with proper termination conditions.The
while(true)loop safely terminates via:
- Return when resource is available (line 126)
- Throw
ResourceAvailabilityTimeoutErrorwhen timeout exceeded (line 119 → line 74)- Rethrow unexpected errors from
checkFn(line 149)The logging strategy (first attempt + every ~minute at 5s intervals) is pragmatic for long-running waits.
40-53: Well-designed error class with actionable message.The
ResourceAvailabilityTimeoutErrorincludes bothresourceNameandtimeoutMsas properties, making it easy for consumers to handle programmatically. The error message is clear and suggests potential causes.
6-13: Clean type definitions with proper type guard.The
ResourceAvailabilityCheckResult<T>discriminated union andWaitForResourceOptions<T>generic type are well-designed. TheisResourceAvailabilityWaitingEnabledtype guard properly narrows the config type for downstream usage.Also applies to: 15-38, 160-163
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In @packages/sns/lib/utils/snsInitter.ts:
- Around line 153-154: The thrown error in snsInitter.ts references
locatorConfig.topicArn which may be undefined when the topic was resolved by
name; change the error to include the resolved subscriptionTopicArn variable
instead (use subscriptionTopicArn in the message), so the thrown Error uses the
actual ARN that was checked (reference symbols: topicCheckResult,
locatorConfig.topicArn, subscriptionTopicArn).
- Around line 290-291: The thrown Error uses locatorConfig.topicArn which may be
undefined when the topic was resolved by name; change the throw in the
checkResult.error === 'not_found' branch to use the resolved topicArn variable
(the one set earlier when locating the topic) instead of locatorConfig.topicArn
so the error message always contains the actual ARN value.
In @packages/sqs/lib/utils/sqsInitter.ts:
- Around line 110-111: The throw in the not-found branch of initSqs/whatever in
sqsInitter.ts uses locatorConfig.queueUrl which can be undefined when the queue
was resolved by queueName; replace that reference with the resolved queueUrl
variable (the one assigned around line 85) so the Error message reads the actual
URL (e.g., throw new Error(`Queue with queueUrl ${queueUrl} does not exist.`) in
the block where checkResult.error === 'not_found').
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
packages/core/lib/utils/resourceAvailabilityUtils.tspackages/sns/lib/index.tspackages/sns/lib/utils/snsInitter.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.tspackages/sqs/lib/index.tspackages/sqs/lib/utils/sqsInitter.ts
🚧 Files skipped from review as they are similar to previous changes (2)
- packages/sqs/lib/index.ts
- packages/core/lib/utils/resourceAvailabilityUtils.ts
🧰 Additional context used
🧬 Code graph analysis (1)
packages/sqs/lib/utils/sqsInitter.ts (3)
packages/core/lib/types/queueOptionsTypes.ts (1)
ResourceAvailabilityConfig(170-191)packages/sqs/lib/utils/sqsUtils.ts (2)
resolveQueueUrlFromLocatorConfig(90-109)getQueueAttributes(62-88)packages/core/lib/utils/resourceAvailabilityUtils.ts (2)
isResourceAvailabilityWaitingEnabled(161-165)waitForResource(102-156)
🔇 Additional comments (13)
packages/sqs/lib/utils/sqsInitter.ts (3)
21-24: LGTM!The
InitSqsExtraParamstype definition is clean and follows the same pattern as the SNS initter types.
92-105: LGTM!The waiting-enabled path correctly uses
waitForResourcewith a checkFn that polls for queue availability viagetQueueAttributes. The implementation is consistent with the SNS initter pattern.
81-81: LGTM!The optional
extraParamsparameter extends the public API while maintaining backward compatibility.packages/sns/lib/index.ts (1)
28-33: LGTM!The new type exports (
InitSnsExtraParamsandInitSnsSqsExtraParams) appropriately extend the public API surface to support resourceAvailabilityConfig configuration.packages/sns/test/consumers/SnsSqsPermissionConsumer.resourceAvailability.spec.ts (5)
19-71: LGTM!The
TestResourceAvailabilityConsumeris well-designed for testing resource availability behavior. ThesubscriptionPropsgetter provides convenient access to initialization results for assertions.
83-104: LGTM!The test setup and teardown logic properly manages resource lifecycle, ensuring clean state for each test and proper disposal of the DI container.
107-140: LGTM!This test effectively validates that the consumer waits for topic availability when polling is enabled. The pattern of starting initialization in the background and creating the resource after a delay properly exercises the polling logic.
174-197: LGTM!The timeout test correctly validates that
ResourceAvailabilityTimeoutErroris thrown when resources don't become available within the configured timeout period.
200-242: LGTM!The disabled configuration tests properly validate fail-fast behavior, ensuring that when polling is disabled, missing resources cause immediate failures.
packages/sns/lib/utils/snsInitter.ts (4)
29-35: LGTM!The
InitSnsSqsExtraParamsandInitSnsExtraParamstype definitions are consistent with the pattern established in the SQS initter and properly extend the baseExtraParamstype.
108-124: LGTM!The topic waiting logic correctly uses
waitForResourceto poll for SNS topic availability when resource availability config is enabled.
126-138: LGTM!The queue waiting logic properly polls for SQS queue availability after the topic check completes, maintaining consistency with the resource availability pattern.
274-286: LGTM!The waiting-enabled path for
initSnscorrectly polls for topic availability using the same pattern as the other init functions.
| // Wait for topic to become available | ||
| await waitForResource({ | ||
| config: resourceAvailabilityConfig, | ||
| resourceName: `SNS topic ${subscriptionTopicArn}`, | ||
| logger: extraParams?.logger, | ||
| checkFn: async () => { | ||
| const result = await getTopicAttributes(snsClient, subscriptionTopicArn) | ||
| if (result.error === 'not_found') { | ||
| return { isAvailable: false } | ||
| } | ||
| return { isAvailable: true, result: result.result } | ||
| }, | ||
| }) |
There was a problem hiding this comment.
If the resource is not available within the given timeout, we throw a ResourceAvailabilityTimeoutError. What is the expected behaviour of the app in this case? Should it fail and restart until the resource becomes available?
Wondering if we could let the app continue running and report the error via the ErrorReporter for visibility, but not entering in a crash loop
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In
@packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts:
- Around line 94-186: Add an integration test that covers NO_TIMEOUT indefinite
polling: in SnsSqsPermissionConsumer.startupResourcePolling.spec.ts add a new
it() similar to the provided suggested test that creates the queue first,
constructs TestStartupResourcePollingConsumer with
locatorConfig.startupResourcePolling.timeoutMs set to NO_TIMEOUT (and
pollingIntervalMs small), starts init() in background, waits briefly, then
creates the topic and awaits initPromise; assert subscriptionProps.topicArn and
queueName as in other tests; also import NO_TIMEOUT from
'@message-queue-toolkit/core' alongside MessageHandlerConfigBuilder and
StartupResourcePollingTimeoutError.
🧹 Nitpick comments (3)
packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts (1)
103-104: Consider extracting the hard-coded subscription ARN.The subscription ARN
'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395'is repeated throughout the test file. Extracting it as a constant would improve maintainability.♻️ Suggested refactor
Add near the top of the test suite (after line 64):
const queueUrl = `http://sqs.eu-west-1.localstack:4566/000000000000/${queueName}` +const testSubscriptionArn = 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395'Then replace all occurrences:
subscriptionArn: - 'arn:aws:sns:eu-west-1:000000000000:dummy:bdf640a2-bedf-475a-98b8-758b88c87395', + testSubscriptionArn,Also applies to: 138-139, 170-171, 199-200, 220-221
packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts (1)
119-143: Consider adding a timeout guard for the NO_TIMEOUT test.While the test verifies indefinite polling works, if the queue creation fails or the test regresses, this test could hang indefinitely in CI environments.
♻️ Optional improvement
Consider wrapping the test with Vitest's timeout option to prevent CI hangs:
- it('polls indefinitely when NO_TIMEOUT is used', async () => { + it('polls indefinitely when NO_TIMEOUT is used', { timeout: 10000 }, async () => {packages/sns/lib/utils/snsInitter.ts (1)
100-148: Consider parallelizing resource polling for faster startup.Currently, the topic and queue are polled sequentially. Since they are independent resources, polling them in parallel could reduce startup time when both resources become available at different times.
♻️ Proposed parallel polling
if (isStartupResourcePollingEnabled(startupResourcePolling)) { - // Wait for topic to become available - await waitForResource({ + // Wait for both resources in parallel + await Promise.all([ + waitForResource({ config: startupResourcePolling, resourceName: `SNS topic ${subscriptionTopicArn}`, logger: extraParams?.logger, checkFn: async () => { const result = await getTopicAttributes(snsClient, subscriptionTopicArn) if (result.error === 'not_found') { return { isAvailable: false } } return { isAvailable: true, result: result.result } }, - }) - - // Wait for queue to become available - await waitForResource({ + }), + waitForResource({ config: startupResourcePolling, resourceName: `SQS queue ${queueUrl}`, logger: extraParams?.logger, checkFn: async () => { const result = await getQueueAttributes(sqsClient, queueUrl) if (result.error === 'not_found') { return { isAvailable: false } } return { isAvailable: true, result: result.result } }, - }) + }), + ]) }
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (11)
README.mdpackages/core/lib/index.tspackages/core/lib/types/queueOptionsTypes.tspackages/core/lib/utils/startupResourcePollingUtils.spec.tspackages/core/lib/utils/startupResourcePollingUtils.tspackages/sns/lib/sns/AbstractSnsService.tspackages/sns/lib/utils/snsInitter.tspackages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.tspackages/sqs/lib/sqs/AbstractSqsService.tspackages/sqs/lib/utils/sqsInitter.tspackages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/core/lib/index.ts
🧰 Additional context used
🧬 Code graph analysis (3)
packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts (5)
packages/sqs/lib/index.ts (1)
AbstractSqsConsumer(5-5)packages/core/lib/index.ts (2)
MessageHandlerConfigBuilder(63-63)StartupResourcePollingTimeoutError(111-111)examples/sns-sqs/lib/common/Dependencies.ts (1)
sqsClient(17-17)packages/core/lib/utils/startupResourcePollingUtils.ts (1)
StartupResourcePollingTimeoutError(41-54)packages/core/lib/types/queueOptionsTypes.ts (1)
NO_TIMEOUT(148-148)
packages/sqs/lib/sqs/AbstractSqsService.ts (1)
packages/core/lib/types/queueOptionsTypes.ts (1)
BaseQueueLocatorType(222-229)
packages/sns/lib/sns/AbstractSnsService.ts (2)
packages/sns/lib/index.ts (1)
SNSTopicLocatorType(9-9)packages/core/lib/types/queueOptionsTypes.ts (1)
BaseQueueLocatorType(222-229)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (20)
- GitHub Check: general (20.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (20.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (24.x, @message-queue-toolkit/redis-message-deduplication-store) / build
- GitHub Check: general (24.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (22.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (22.x, @message-queue-toolkit/schemas) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (20.x, @message-queue-toolkit/metrics) / build
- GitHub Check: general (20.x, @message-queue-toolkit/amqp) / build
- GitHub Check: general (24.x, @message-queue-toolkit/schemas) / build
- GitHub Check: general (24.x, @message-queue-toolkit/outbox-core) / build
- GitHub Check: general (20.x, @message-queue-toolkit/schemas) / build
- GitHub Check: general (24.x, @message-queue-toolkit/core) / build
- GitHub Check: general (22.x, @message-queue-toolkit/redis-message-deduplication-store) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (20.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sqs) / build
- GitHub Check: kafka (24.x) / build
- GitHub Check: kafka (22.x) / build
🔇 Additional comments (28)
packages/sns/test/consumers/SnsSqsPermissionConsumer.startupResourcePolling.spec.ts (3)
1-59: LGTM! Clean test setup.The test consumer class is well-structured and minimal, exposing only the necessary properties for validation. The imports are appropriate for the polling integration tests.
61-93: LGTM! Proper test lifecycle management.The test setup correctly manages resource lifecycle with appropriate cleanup in beforeEach/afterEach hooks, and proper container disposal in afterAll.
188-232: LGTM! Disabled polling behavior is correctly tested.The tests properly verify that when
startupResourcePolling.enabledisfalse, the consumer fails immediately rather than polling for resource availability.README.md (1)
247-343: LGTM! Comprehensive and well-structured documentation.The documentation clearly explains:
- The motivation and use case for startup resource polling
- Configuration options with appropriate defaults
- Environment-specific patterns (production with timeout, dev/staging with NO_TIMEOUT)
- Error handling with concrete code examples
The integration into
locatorConfigaddresses the past review feedback, and the section is well-positioned in the README.packages/sqs/lib/sqs/AbstractSqsService.ts (2)
2-6: LGTM! Proper type integration.The
SQSQueueLocatorTypecorrectly extendsBaseQueueLocatorTypewhile preserving the existing discriminated union forqueueUrl/queueName. This enables startup resource polling configuration without breaking existing code.Also applies to: 45-55
117-117: LGTM! Logger propagation supports polling progress logging.Passing the logger to
initSqsenables progress logging during startup resource polling, consistent with the pattern used inAbstractSnsService.packages/core/lib/utils/startupResourcePollingUtils.spec.ts (3)
10-29: LGTM! Complete coverage of enabled state checking.The tests verify all relevant scenarios: enabled true/false, missing enabled field, and undefined config. This ensures the utility function behaves correctly in all cases.
31-159: LGTM! Thorough test coverage of polling behavior.The test suite comprehensively covers:
- Immediate resource availability
- Multi-attempt polling until success
- Timeout behavior with proper error
- Error propagation from check function
- Default configuration values
- Logger integration
- NO_TIMEOUT indefinite polling
All scenarios are well-tested with appropriate assertions.
161-171: LGTM! Proper error structure validation.The test verifies that
StartupResourcePollingTimeoutErrorincludes all necessary information (resource name, timeout) in both the message and as properties, enabling effective debugging and error handling.packages/sns/lib/sns/AbstractSnsService.ts (2)
3-7: LGTM! Consistent type integration with SQS.The
SNSTopicLocatorTypeproperly extendsBaseQueueLocatorType, enabling startup resource polling for SNS topics. The pattern is consistent with the SQS implementation.Also applies to: 47-50
89-89: LGTM! Logger propagation matches SQS pattern.Passing the logger to
initSnsenables startup resource polling progress logging, maintaining consistency with the SQS implementation.packages/core/lib/types/queueOptionsTypes.ts (3)
133-148: LGTM! Well-documented NO_TIMEOUT symbol.The symbol provides a clear, type-safe way to indicate indefinite polling. The JSDoc example demonstrates proper usage.
150-216: LGTM! Comprehensive type definition with excellent documentation.The
StartupResourcePollingConfigtype is well-designed:
enabledis now optional (per past review feedback)timeoutMsis required, ensuring callers explicitly choose between a timeout orNO_TIMEOUTpollingIntervalMsis optional with a sensible default- Multiple examples cover common use cases
218-229: LGTM! Clean base type for locator composition.
BaseQueueLocatorTypeprovides a clean extension point for protocol-specific locator types to inherit startup polling configuration.packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts (4)
1-55: LGTM! Well-structured test consumer and imports.The
TestStartupResourcePollingConsumerclass provides a minimal test harness that properly exposes queue properties for assertions.
57-74: LGTM! Proper test setup and teardown.The test setup correctly:
- Registers DI dependencies
- Cleans up the queue before and after each test
- Disposes the container properly
76-117: LGTM! Good coverage of enabled polling scenarios.Tests correctly verify:
- Successful initialization after delayed queue creation
StartupResourcePollingTimeoutErrorwhen timeout is reached- Proper timing of queue creation vs. init promise resolution
145-173: LGTM! Backward compatibility tests are essential.These tests ensure the default behavior (immediate failure) is preserved when
startupResourcePollingis disabled or not provided, maintaining backward compatibility.packages/sqs/lib/utils/sqsInitter.ts (2)
21-23: LGTM! Clean type definition.
InitSqsExtraParamsprovides a clear extension point for passing logger context to the initialization flow.
75-114: LGTM! Well-structured polling integration.The implementation cleanly separates the polling and non-polling paths:
- Uses
isStartupResourcePollingEnabledfor type-safe guardwaitForResourcereceives properly configuredcheckFnthat returns QueueArn- Error message now correctly uses the resolved
queueUrlvariable (addressing past review)- Both paths properly assign
queueArnfor subsequent validationpackages/sns/lib/utils/snsInitter.ts (2)
25-27: LGTM! Type aliases maintain consistency.Using type aliases for
InitSnsSqsExtraParamsandInitSnsExtraParamskeeps the API surface clean while allowing future extension.
246-285: LGTM! Clean polling integration for SNS-only path.The
initSnsfunction properly integrates polling with a type-safe guard and correct error handling. The error message correctly uses the resolvedtopicArnvariable.packages/core/lib/utils/startupResourcePollingUtils.ts (6)
1-5: LGTM! Clean imports and sensible default.The 5-second default polling interval is a reasonable balance between responsiveness and avoiding excessive API calls.
7-39: LGTM! Well-designed type definitions.The discriminated union for
StartupResourcePollingCheckResultprovides type-safe handling of both available and unavailable states. TheWaitForResourceOptionstype is thoroughly documented.
41-54: LGTM! Proper custom error implementation.The
StartupResourcePollingTimeoutErrorcorrectly:
- Extends
Error- Sets
this.namefor proper error identification- Includes contextual properties (
resourceName,timeoutMs)- Provides a descriptive message
56-92: LGTM! Clean helper functions.The helper functions properly encapsulate timeout checking and success logging, keeping the main polling loop focused.
118-155: Timeout check timing allows one extra check after deadline.The timeout is checked at the start of each iteration, meaning if a check takes a long time, the actual elapsed time could exceed
timeoutMsbefore the timeout is triggered. This is generally acceptable for startup polling, but worth noting.Additionally, the throttled logging (
attemptCount % 12 === 0) assumes the default 5s interval for the "every minute" comment, but if a custom interval is used, the logging frequency will differ.These are minor behavioral notes rather than issues:
- The timeout behavior (check-before-poll) is standard and acceptable
- The throttled logging comment could be clarified to note it's based on the default interval
No changes required unless you prefer to document or adjust this behavior.
158-165: LGTM! Type guard enables safe config access.The
isStartupResourcePollingEnabledfunction is a proper type guard that narrowsconfigtoStartupResourcePollingConfigwhen it returnstrue, ensuring type-safe access in the polling branches.
There was a problem hiding this comment.
Actionable comments posted: 0
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/utils/snsInitter.ts (1)
42-50: Return type signature missingresourcesReadyfield.The return type on line 50 specifies
{ subscriptionArn: string; topicArn: string; queueUrl: string; queueName: string }, but the actual returns at lines 103, 151, 187, and 224 all include an additionalresourcesReady: booleanfield. This causes a type mismatch.🐛 Proposed fix
export async function initSnsSqs( sqsClient: SQSClient, snsClient: SNSClient, stsClient: STSClient, locatorConfig?: SNSSQSQueueLocatorType, creationConfig?: SNSCreationConfig & SQSCreationConfig, subscriptionConfig?: SNSSubscriptionOptions, extraParams?: InitSnsSqsExtraParams, -): Promise<{ subscriptionArn: string; topicArn: string; queueUrl: string; queueName: string }> { +): Promise<{ subscriptionArn: string; topicArn: string; queueUrl: string; queueName: string; resourcesReady: boolean }> {
🧹 Nitpick comments (1)
packages/core/lib/utils/startupResourcePollingUtils.ts (1)
297-309: Consider adding error handler to outer promise chain.In non-blocking mode, the background polling is initiated with
setTimeout().then(...). While the innerpollForResourcehas a.catch()handler (lines 302-308), the outersetTimeoutpromise chain itself lacks one. ThoughsetTimeoutrarely fails, adding a catch handler would be more defensive.♻️ Suggested improvement
// Fire and forget - start polling in background setTimeout(pollingIntervalMs).then(() => { pollForResource(options, pollingIntervalMs, hasTimeout, timeoutMs, throwOnTimeout, 1) .then((result) => { onResourceAvailable?.(result) }) .catch((error) => { logger?.error({ message: `Background polling for resource "${resourceName}" failed`, resourceName, error, }) }) - }) + }).catch((error) => { + logger?.error({ + message: `Failed to schedule background polling for resource "${resourceName}"`, + resourceName, + error, + }) + })
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
README.mdpackages/core/lib/types/MessageQueueTypes.tspackages/core/lib/types/queueOptionsTypes.tspackages/core/lib/utils/startupResourcePollingUtils.spec.tspackages/core/lib/utils/startupResourcePollingUtils.tspackages/sns/lib/utils/snsInitter.tspackages/sqs/lib/utils/sqsInitter.ts
✅ Files skipped from review due to trivial changes (1)
- packages/core/lib/utils/startupResourcePollingUtils.spec.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- README.md
🧰 Additional context used
🧬 Code graph analysis (1)
packages/core/lib/utils/startupResourcePollingUtils.ts (2)
packages/core/lib/index.ts (5)
StartupResourcePollingCheckResult(110-110)WaitForResourceOptions(112-112)StartupResourcePollingTimeoutError(111-111)waitForResource(113-113)isStartupResourcePollingEnabled(109-109)packages/core/lib/types/queueOptionsTypes.ts (2)
StartupResourcePollingConfig(198-238)NO_TIMEOUT(148-148)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: general (20.x, @message-queue-toolkit/amqp) / build
- GitHub Check: kafka (22.x) / build
- GitHub Check: kafka (24.x) / build
🔇 Additional comments (5)
packages/core/lib/types/MessageQueueTypes.ts (1)
1-5: LGTM - Clean addition of error reporting support.The addition of
ErrorReportertoExtraParamscleanly enables error reporting throughout the startup resource polling flows without breaking existing code.Also applies to: 45-45
packages/core/lib/types/queueOptionsTypes.ts (1)
133-251: LGTM - Well-documented polling configuration types.The new types (
NO_TIMEOUT,StartupResourcePollingConfig,BaseQueueLocatorType) are well-designed with comprehensive documentation and clear examples. The use of a Symbol forNO_TIMEOUTis appropriate for a sentinel value, and the optionalenabledfield correctly reflects the JSDoc description.packages/sqs/lib/utils/sqsInitter.ts (1)
87-148: LGTM - Polling logic correctly integrated.The startup resource polling implementation is correct:
- The polling flow properly uses
waitForResourcewhen enabled- Non-blocking mode is handled correctly with the
onQueueReadycallback- Fallback to original immediate-check behavior is preserved
- Error messages correctly use the resolved
queueUrlvariable- Return type matches the actual returns
packages/core/lib/utils/startupResourcePollingUtils.ts (1)
170-241: LGTM - Solid polling implementation.The core polling logic is well-structured:
- First synchronous check optimizes for immediate availability
- Polling loop correctly handles timeout, logging, and error propagation
- Non-blocking mode properly returns
undefinedand schedules background polling- Timeout reset (line 206) correctly supports the "report and continue" mode
- Error handling appropriately catches, logs, and rethrows unexpected errors from
checkFnAlso applies to: 252-316
packages/sns/lib/utils/snsInitter.ts (1)
116-226: LGTM - Polling logic correctly implemented.The startup resource polling implementation follows the same pattern as SQS and is correctly structured:
- Sequential polling for topic and queue in
initSnsSqs- Non-blocking mode returns early with
resourcesReady: false- Callbacks (
onResourcesReady,onTopicReady) are invoked when resources become available- Fallback to immediate checks when polling is disabled
- Error messages use resolved ARN/URL variables
Also applies to: 326-360
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In @packages/sns/lib/utils/snsInitter.ts:
- Around line 122-195: The onResourcesReady callback is being called twice from
the per-resource onResourceAvailable handlers (when nonBlocking is true) even
though it should fire only once after both topic and queue are available; modify
the logic around waitForResource (the topic and queue wait blocks using
getTopicAttributes/getQueueAttributes) to stop invoking
extraParams?.onResourcesReady inside each onResourceAvailable, and instead track
availability with local flags (e.g., topicAvailable and queueAvailable) and
invoke extraParams?.onResourcesReady({ topicArn: subscriptionTopicArn, queueUrl
}) exactly once when both topicResult and queueResult indicate availability (and
only when nonBlocking), making sure you still return early with resourcesReady:
false if a non-blocking wait returned undefined for either resource.
In @packages/sqs/lib/sqs/AbstractSqsService.ts:
- Line 98: The field queueArn in AbstractSqsService was changed to string |
undefined causing potential undefined access when non-blocking startup (initSqs)
returns undefined; update any code that reads this.queueArn (including
subclasses and methods in AbstractSqsService) to guard against undefined before
use (e.g., conditional checks or early returns) and add appropriate
fallback/error logging that includes the undefined case so initialization and
callbacks don’t throw; specifically audit initSqs, any constructors,
initialization callbacks, and methods referencing queueArn to ensure they handle
the undefined value safely.
🧹 Nitpick comments (1)
packages/sns/lib/utils/snsInitter.ts (1)
148-150: Consider extracting queue name extraction logic.The logic to extract the queue name from the queue URL is duplicated three times in this function (lines 148-150, 184-186, and 216-219). Consider extracting this into a helper function to improve maintainability.
♻️ Proposed refactor
function extractQueueNameFromUrl(queueUrl: string): string { const splitUrl = queueUrl.split('/') // biome-ignore lint/style/noNonNullAssertion: It's ok return splitUrl[splitUrl.length - 1]! }Then replace the duplicated code:
// If non-blocking and topic wasn't immediately available, return early if (nonBlocking && topicResult === undefined) { - const splitUrl = queueUrl.split('/') - // biome-ignore lint/style/noNonNullAssertion: It's ok - const queueName = splitUrl[splitUrl.length - 1]! + const queueName = extractQueueNameFromUrl(queueUrl) return { subscriptionArn: locatorConfig.subscriptionArn, topicArn: subscriptionTopicArn, queueUrl, queueName, resourcesReady: false, } }(And similar changes for the other occurrences.)
Also applies to: 184-186, 216-219
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
README.mdpackages/sns/lib/utils/snsInitter.tspackages/sqs/lib/sqs/AbstractSqsService.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- README.md
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (20)
- GitHub Check: general (20.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (20.x, @message-queue-toolkit/schemas) / build
- GitHub Check: general (24.x, @message-queue-toolkit/redis-message-deduplication-store) / build
- GitHub Check: general (24.x, @message-queue-toolkit/gcs-payload-store) / build
- GitHub Check: general (22.x, @message-queue-toolkit/metrics) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (20.x, @message-queue-toolkit/outbox-core) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (20.x, @message-queue-toolkit/amqp) / build
- GitHub Check: general (20.x, @message-queue-toolkit/redis-message-deduplication-store) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (24.x, @message-queue-toolkit/schemas) / build
- GitHub Check: general (20.x, @message-queue-toolkit/metrics) / build
- GitHub Check: general (24.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (22.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (20.x, @message-queue-toolkit/sns) / build
- GitHub Check: kafka (24.x) / build
- GitHub Check: general (20.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: kafka (22.x) / build
🔇 Additional comments (6)
packages/sqs/lib/sqs/AbstractSqsService.ts (3)
2-6: LGTM!The addition of
BaseQueueLocatorTypeto the imports is necessary for the updatedSQSQueueLocatorTypedefinition and aligns with the polling feature implementation.
116-116: Good addition for observability.Passing the logger context to
initSqsenables better debugging and monitoring during queue initialization, which is especially valuable for the new polling feature.
45-55: No breaking changes introduced by this intersection.The
BaseQueueLocatorTypeadds only an optionalstartupResourcePollingfield. Existing consumers can continue to constructSQSQueueLocatorTypewith justqueueUrlorqueueNamewithout providing this field, so no modifications are required.packages/sns/lib/utils/snsInitter.ts (3)
25-39: LGTM!The type definitions are well-documented and clearly express when the callbacks will be invoked. The optional nature of the callbacks allows backward compatibility.
312-365: LGTM!The polling logic for the single topic resource is correctly implemented. The callback is invoked appropriately when the topic becomes available in non-blocking mode, and early returns are handled properly.
122-144: Verify the crash-loop behavior is intentional.Responding to CarlosGamero's question: When resources aren't available within the configured timeout,
waitForResourcethrows aResourceAvailabilityTimeoutError, which will cause the application to crash and restart (crash-loop behavior).This may be the intended design (fail-fast), but it's worth considering whether a more resilient approach would be better for production deployments:
- Continue running and report the error via
ErrorReporterfor visibility- Allow the application to remain available for other operations
- Retry resource checks periodically without full restart overhead
Please confirm whether the crash-loop behavior is the desired failure mode, or whether the error should be reported but allow the application to continue.
| protected queueUrl: string | ||
| // @ts-expect-error | ||
| protected queueArn: string | ||
| protected queueArn: string | undefined |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find the directory structure
echo "=== Finding AbstractSqsService and related files ==="
find . -name "AbstractSqsService.ts" -o -name "*SqsService*.ts" | head -20
echo ""
echo "=== Finding all classes extending AbstractSqsService ==="
rg -l "extends AbstractSqsService" --type ts | grep -v node_modules
echo ""
echo "=== Checking subclasses and their queueArn usage ==="
rg "extends AbstractSqsService" --type ts -A 50 | grep -E "(class|queueArn)"
echo ""
echo "=== All references to queueArn in codebase ==="
rg "queueArn" --type ts -B 2 -A 2 | grep -v node_modules | grep -v ".test.ts"
echo ""
echo "=== Checking AbstractSqsService.ts for queueArn definition and usage ==="
cat packages/sqs/lib/sqs/AbstractSqsService.ts | head -150Repository: kibertoad/message-queue-toolkit
Length of output: 20066
Type signature change makes queueArn nullable in non-blocking startup mode.
The change from string to string | undefined is intentional: initSqs() explicitly returns undefined when using non-blocking resource polling. While this is a type-level breaking change requiring null checks in strict TypeScript code, the actual runtime impact appears limited since current usage occurs primarily in error logging and non-critical paths.
If non-blocking startup polling is enabled in your setup, verify that any subclass logic accessing this.queueArn can tolerate undefined values, particularly in initialization or callback sequences.
🤖 Prompt for AI Agents
In @packages/sqs/lib/sqs/AbstractSqsService.ts at line 98, The field queueArn in
AbstractSqsService was changed to string | undefined causing potential undefined
access when non-blocking startup (initSqs) returns undefined; update any code
that reads this.queueArn (including subclasses and methods in
AbstractSqsService) to guard against undefined before use (e.g., conditional
checks or early returns) and add appropriate fallback/error logging that
includes the undefined case so initialization and callbacks don’t throw;
specifically audit initSqs, any constructors, initialization callbacks, and
methods referencing queueArn to ensure they handle the undefined value safely.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In
@packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts:
- Around line 77-101: Tests use fixed setTimeout delays (e.g., in
SqsPermissionConsumer.startupResourcePolling.spec.ts around the
TestStartupResourcePollingConsumer init flow) which can be flaky; replace the
ad-hoc sleeps with a deterministic wait helper that polls the resource until
available (e.g., poll assertQueue or describeQueue until success with a short
interval and overall timeout) and use that helper before awaiting
consumer.init(), or at minimum increase timeouts; update all occurrences
referencing setTimeout(300/500/200) and the
TestStartupResourcePollingConsumer.init() test flows (also lines 119-142 and
218-252) to use the polling helper or retries instead of fixed delays.
🧹 Nitpick comments (1)
packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts (1)
218-252: Testing internal implementation directly.This test imports and calls the internal
initSqsfunction directly to verify theonQueueReadycallback mechanism. While the comment acknowledges thatAbstractSqsConsumerdoesn't expose this callback, testing internal implementation details can make tests brittle and tightly coupled to the implementation.Consider whether the callback mechanism is important enough to expose as a protected or public API on the consumer, which would allow testing through the intended abstraction.
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (20)
- GitHub Check: general (22.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (20.x, @message-queue-toolkit/metrics) / build
- GitHub Check: general (20.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (22.x, @message-queue-toolkit/schemas) / build
- GitHub Check: general (20.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (22.x, @message-queue-toolkit/core) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (24.x, @message-queue-toolkit/outbox-core) / build
- GitHub Check: general (24.x, @message-queue-toolkit/s3-payload-store) / build
- GitHub Check: general (22.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (22.x, @message-queue-toolkit/gcs-payload-store) / build
- GitHub Check: general (24.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: general (24.x, @message-queue-toolkit/metrics) / build
- GitHub Check: general (22.x, @message-queue-toolkit/amqp) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sqs) / build
- GitHub Check: general (20.x, @message-queue-toolkit/redis-message-deduplication-store) / build
- GitHub Check: general (24.x, @message-queue-toolkit/sns) / build
- GitHub Check: general (20.x, @message-queue-toolkit/gcp-pubsub) / build
- GitHub Check: kafka (22.x) / build
- GitHub Check: kafka (24.x) / build
| it('waits for queue to become available and initializes successfully', async () => { | ||
| const consumer = new TestStartupResourcePollingConsumer(diContainer.cradle, { | ||
| locatorConfig: { | ||
| queueUrl, | ||
| startupResourcePolling: { | ||
| enabled: true, | ||
| pollingIntervalMs: 100, | ||
| timeoutMs: 5000, | ||
| }, | ||
| }, | ||
| }) | ||
|
|
||
| // Start init in background | ||
| const initPromise = consumer.init() | ||
|
|
||
| // Wait a bit then create the queue | ||
| await setTimeout(300) | ||
| await assertQueue(sqsClient, { QueueName: queueName }) | ||
|
|
||
| // Init should complete successfully | ||
| await initPromise | ||
|
|
||
| expect(consumer.queueProps.url).toBe(queueUrl) | ||
| expect(consumer.queueProps.name).toBe(queueName) | ||
| }) |
There was a problem hiding this comment.
Tests use fixed delays that may be timing-sensitive.
Multiple tests use setTimeout with fixed delays (300ms, 500ms, 200ms) to simulate asynchronous queue creation. While these delays are reasonable for LocalStack integration tests, they could potentially be flaky on slow CI systems or under heavy load.
If these tests become flaky in CI, consider:
- Increasing the delays slightly
- Using more robust polling/waiting patterns
- Adding retry logic for transient failures
For now, the current approach is acceptable for LocalStack-based integration tests.
Also applies to: 119-142, 218-252
🤖 Prompt for AI Agents
In
@packages/sqs/test/consumers/SqsPermissionConsumer.startupResourcePolling.spec.ts
around lines 77 - 101, Tests use fixed setTimeout delays (e.g., in
SqsPermissionConsumer.startupResourcePolling.spec.ts around the
TestStartupResourcePollingConsumer init flow) which can be flaky; replace the
ad-hoc sleeps with a deterministic wait helper that polls the resource until
available (e.g., poll assertQueue or describeQueue until success with a short
interval and overall timeout) and use that helper before awaiting
consumer.init(), or at minimum increase timeouts; update all occurrences
referencing setTimeout(300/500/200) and the
TestStartupResourcePollingConsumer.init() test flows (also lines 119-142 and
218-252) to use the polling helper or retries instead of fixed delays.
3 participants
Original problem statement:
Cross dependency between services
Lack of granularity when enabling consumers on AP side
If AWS config is supplied, AP enables all consumers, meaning it requires all of the resources (mostly topics) to be present on AWS. We often face a situation when we only want to have a connection between a few of the services only (example AP and CTENG, but not Expert). Currently, the only workaround is to find all of the unnecessary consumers and disable them manually (via amending the code).
In CTENG, we have an env flag that allows to enable/disable specific consumers (for example AP-related ones), which gives us more flexibility. I know it's been a case of AP previously, but some time ago it got changed and now works as described in 2.a. Wondering if we should also provide more granular control over the consumers.
Summary by CodeRabbit
New Features
Documentation
Bug Fixes
Tests
✏️ Tip: You can customize this high-level summary in your review settings.