From 864fbaaeb7b4106d3c3e4281e63aee217c7e845d Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Mon, 24 Aug 2026 20:19:05 +0530 Subject: [PATCH 1/3] fix(aws-sqs): precreate self-referential queue identity Precreate the queue URL and ARN before resolving self-referential policies, and grant the bootstrap calls needed to discover that immutable identity. Original-Commits: 210799615 8ae19a190 Upstream-Candidate: SQS precreated identity --- packages/alchemy/src/AWS/SQS/Queue.ts | 111 ++++++++++++++++++++ packages/alchemy/test/AWS/SQS/Queue.test.ts | 42 ++++++++ 2 files changed, 153 insertions(+) diff --git a/packages/alchemy/src/AWS/SQS/Queue.ts b/packages/alchemy/src/AWS/SQS/Queue.ts index 0c801f3726..527006658b 100644 --- a/packages/alchemy/src/AWS/SQS/Queue.ts +++ b/packages/alchemy/src/AWS/SQS/Queue.ts @@ -20,6 +20,18 @@ export type QueueName = string; export type QueueArn = `arn:aws:sqs:${RegionID}:${AccountID}:${QueueName}`; export type QueueUrl = string; +export class QueuePrecreateConflict extends Data.TaggedError( + "QueuePrecreateConflict", +)<{ + readonly id: string; + readonly queueName: string; + readonly reason: string; +}> { + override get message() { + return `Cannot pre-create SQS queue '${this.id}' as '${this.queueName}': ${this.reason}`; + } +} + export type QueueProps = { /** * Name of the queue. @@ -486,6 +498,105 @@ export const QueueProvider = () => } // Return undefined to allow update function to be called for other attribute changes }), + precreate: Effect.fn(function* ({ id, news = {}, session }) { + // Queue policies commonly name the queue's own ARN. That makes the + // binding graph self-referential on a greenfield deploy: the policy + // cannot resolve until the queue has an identity, while reconcile + // cannot receive the resolved policy until that identity exists. + // + // Create only the stable physical shell here. Reconcile remains the + // single authority for mutable attributes, user tags, and binding- + // contributed policy once every dependency can be evaluated. + const identity = { + queueName: news.queueName, + fifo: news.fifo, + }; + if (!isResolved(identity)) { + return yield* Effect.die( + new Error( + `SQS Queue '${id}' cannot be pre-created because queueName or fifo depends on an unresolved resource output`, + ), + ); + } + + const { accountId, region } = yield* AWSEnvironment.current; + const queueName = yield* createQueueName(id, identity); + const queueArn = + `arn:aws:sqs:${region}:${accountId}:${queueName}` as const; + const internalTags = yield* createInternalTags(id); + let queueUrl = yield* sqs.getQueueUrl({ QueueName: queueName }).pipe( + Effect.map((result) => result.QueueUrl), + Effect.catchTag("QueueDoesNotExist", () => + Effect.succeed(undefined), + ), + ); + let requiresOwnershipCheck = queueUrl !== undefined; + + if (queueUrl === undefined) { + const observed = yield* sqs + .createQueue({ + QueueName: queueName, + ...(identity.fifo ? { Attributes: { FifoQueue: "true" } } : {}), + tags: internalTags, + }) + .pipe( + Effect.retry({ + while: (error) => error._tag === "QueueDeletedRecently", + schedule: Schedule.fixed(1000).pipe( + Schedule.tap(({ attempt }) => + session.note( + `Queue was deleted recently, retrying... ${attempt}s`, + ), + ), + ), + }), + Effect.map((result) => ({ + queueUrl: result.QueueUrl!, + raced: false, + })), + Effect.catchTag("QueueNameExists", () => + sqs.getQueueUrl({ QueueName: queueName }).pipe( + Effect.map((result) => ({ + queueUrl: result.QueueUrl!, + raced: true, + })), + ), + ), + ); + queueUrl = observed.queueUrl; + requiresOwnershipCheck = observed.raced; + } + + if (requiresOwnershipCheck) { + const [tags, attributes] = yield* Effect.all([ + sqs + .listQueueTags({ QueueUrl: queueUrl }) + .pipe(Effect.map((result) => result.Tags ?? {})), + sqs.getQueueAttributes({ + QueueUrl: queueUrl, + AttributeNames: ["FifoQueue"], + }), + ]); + if (!(yield* hasAlchemyTags(id, tags))) { + return yield* new QueuePrecreateConflict({ + id, + queueName, + reason: + "an existing queue with that name is not owned by this stack", + }); + } + const observedFifo = attributes.Attributes?.FifoQueue === "true"; + if (observedFifo !== (identity.fifo ?? false)) { + return yield* new QueuePrecreateConflict({ + id, + queueName, + reason: `the existing queue is ${observedFifo ? "FIFO" : "standard"}, but the desired queue is ${identity.fifo ? "FIFO" : "standard"}`, + }); + } + } + + return { queueName, queueUrl, queueArn }; + }), reconcile: Effect.fn(function* ({ id, news = {}, diff --git a/packages/alchemy/test/AWS/SQS/Queue.test.ts b/packages/alchemy/test/AWS/SQS/Queue.test.ts index 102a9841c5..5e52058317 100644 --- a/packages/alchemy/test/AWS/SQS/Queue.test.ts +++ b/packages/alchemy/test/AWS/SQS/Queue.test.ts @@ -54,6 +54,48 @@ provider("create and delete queue with default props", (stack) => }), ); +provider( + "fresh create resolves a queue policy that references its own ARN", + (stack) => + Effect.gen(function* () { + yield* stack.destroy(); + + const queue = yield* stack.deploy( + Effect.gen(function* () { + const queue = yield* Queue("SelfPolicyQueue"); + yield* queue.bind`Allow(${queue}, AWS.SQS.SendMessage(${queue}))`({ + policyStatements: [ + { + Effect: "Allow", + Principal: { Service: "sns.amazonaws.com" }, + Action: ["sqs:SendMessage"], + Resource: queue.queueArn, + }, + ], + }); + return queue; + }), + ); + + const attributes = yield* SQS.getQueueAttributes({ + QueueUrl: queue.queueUrl, + AttributeNames: ["Policy"], + }); + const policy = JSON.parse(attributes.Attributes?.Policy ?? "null") as { + readonly Statement?: ReadonlyArray<{ readonly Resource?: string }>; + } | null; + + expect( + policy?.Statement?.some( + (statement) => statement.Resource === queue.queueArn, + ), + ).toBe(true); + + yield* stack.destroy(); + yield* assertQueueDeleted(queue.queueUrl); + }), +); + provider("create, update, delete standard queue", (stack) => Effect.gen(function* () { yield* stack.destroy(); From 3cf502e3e142abf165e2cbdea8d45c37a9da9b18 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 25 Aug 2026 14:19:50 +0530 Subject: [PATCH 2/3] fix(aws-sqs): bound precreate queue retry --- packages/alchemy/src/AWS/SQS/Queue.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/alchemy/src/AWS/SQS/Queue.ts b/packages/alchemy/src/AWS/SQS/Queue.ts index 527006658b..bccc3fdcbb 100644 --- a/packages/alchemy/src/AWS/SQS/Queue.ts +++ b/packages/alchemy/src/AWS/SQS/Queue.ts @@ -542,7 +542,10 @@ export const QueueProvider = () => .pipe( Effect.retry({ while: (error) => error._tag === "QueueDeletedRecently", - schedule: Schedule.fixed(1000).pipe( + schedule: Schedule.max([ + Schedule.fixed(1000), + Schedule.recurs(30), + ]).pipe( Schedule.tap(({ attempt }) => session.note( `Queue was deleted recently, retrying... ${attempt}s`, From 6c6d3f351544018bf51afb896d918ef65440d900 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 25 Aug 2026 16:21:11 +0530 Subject: [PATCH 3/3] refactor(aws-sqs): share queue identity creation logic --- packages/alchemy/src/AWS/SQS/Queue.ts | 305 ++++++++++++++------------ 1 file changed, 170 insertions(+), 135 deletions(-) diff --git a/packages/alchemy/src/AWS/SQS/Queue.ts b/packages/alchemy/src/AWS/SQS/Queue.ts index bccc3fdcbb..c4d1f27976 100644 --- a/packages/alchemy/src/AWS/SQS/Queue.ts +++ b/packages/alchemy/src/AWS/SQS/Queue.ts @@ -5,6 +5,7 @@ import * as Effect from "effect/Effect"; import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; import { Unowned } from "../../AdoptPolicy.ts"; +import type { ScopedPlanStatusSession } from "../../Cli/Cli.ts"; import { isResolved } from "../../Diff.ts"; import { createPhysicalName } from "../../PhysicalName.ts"; import * as Provider from "../../Provider.ts"; @@ -417,6 +418,136 @@ export const QueueProvider = () => return baseAttributes; }; + const resolveQueueIdentity = Effect.fn(function* ( + id: string, + props: { + queueName?: string | undefined; + fifo?: boolean | undefined; + }, + ) { + const identity = { + queueName: props.queueName, + fifo: props.fifo, + }; + if (!isResolved(identity)) { + return yield* Effect.die( + new Error( + `SQS Queue '${id}' cannot be pre-created because queueName or fifo depends on an unresolved resource output`, + ), + ); + } + + const { accountId, region } = yield* AWSEnvironment.current; + const queueName = yield* createQueueName(id, identity); + return { + queueName, + fifo: identity.fifo ?? false, + queueArn: `arn:aws:sqs:${region}:${accountId}:${queueName}` as const, + }; + }); + const getQueueUrl = (queueName: string) => + sqs.getQueueUrl({ QueueName: queueName }).pipe( + Effect.map((result) => result.QueueUrl), + Effect.catchTag("QueueDoesNotExist", () => Effect.succeed(undefined)), + ); + const createQueueShell = Effect.fn(function* ({ + queueName, + attributes, + tags, + session, + retryInvalidParameter, + }: { + queueName: string; + attributes: Record; + tags: Record; + session: Pick; + retryInvalidParameter?: boolean; + }) { + const create = sqs + .createQueue({ + QueueName: queueName, + Attributes: attributes, + tags, + }) + .pipe( + Effect.retry({ + while: (error) => error._tag === "QueueDeletedRecently", + schedule: Schedule.max([ + Schedule.fixed(1000), + Schedule.recurs(30), + ]).pipe( + Schedule.tap(({ attempt }) => + session.note( + `Queue was deleted recently, retrying... ${attempt}s`, + ), + ), + ), + }), + ); + const createEffect = retryInvalidParameter + ? create.pipe( + Effect.retry({ + while: (error) => + error._tag === "InvalidParameterValueException", + schedule: Schedule.max([ + Schedule.fixed(1000), + Schedule.recurs(30), + ]), + }), + ) + : create; + return yield* createEffect.pipe( + Effect.map((result) => ({ + queueUrl: result.QueueUrl!, + raced: false, + })), + Effect.catchTag("QueueNameExists", () => + sqs.getQueueUrl({ QueueName: queueName }).pipe( + Effect.map((result) => ({ + queueUrl: result.QueueUrl!, + raced: true, + })), + ), + ), + ); + }); + const verifyQueueOwnership = Effect.fn(function* ({ + id, + queueName, + queueUrl, + fifo, + }: { + id: string; + queueName: string; + queueUrl: string; + fifo: boolean; + }) { + const [tags, attributes] = yield* Effect.all([ + sqs + .listQueueTags({ QueueUrl: queueUrl }) + .pipe(Effect.map((result) => result.Tags ?? {})), + sqs.getQueueAttributes({ + QueueUrl: queueUrl, + AttributeNames: ["FifoQueue"], + }), + ]); + if (!(yield* hasAlchemyTags(id, tags))) { + return yield* new QueuePrecreateConflict({ + id, + queueName, + reason: + "an existing queue with that name is not owned by this stack", + }); + } + const observedFifo = attributes.Attributes?.FifoQueue === "true"; + if (observedFifo !== fifo) { + return yield* new QueuePrecreateConflict({ + id, + queueName, + reason: `the existing queue is ${observedFifo ? "FIFO" : "standard"}, but the desired queue is ${fifo ? "FIFO" : "standard"}`, + }); + } + }); return Queue.Provider.of({ stables: ["queueName", "queueUrl", "queueArn"], // Enumerate every queue in the ambient account/region. `listQueues` @@ -507,98 +638,29 @@ export const QueueProvider = () => // Create only the stable physical shell here. Reconcile remains the // single authority for mutable attributes, user tags, and binding- // contributed policy once every dependency can be evaluated. - const identity = { - queueName: news.queueName, - fifo: news.fifo, - }; - if (!isResolved(identity)) { - return yield* Effect.die( - new Error( - `SQS Queue '${id}' cannot be pre-created because queueName or fifo depends on an unresolved resource output`, - ), - ); - } - - const { accountId, region } = yield* AWSEnvironment.current; - const queueName = yield* createQueueName(id, identity); - const queueArn = - `arn:aws:sqs:${region}:${accountId}:${queueName}` as const; + const identity = yield* resolveQueueIdentity(id, news); + const { queueName, queueArn } = identity; const internalTags = yield* createInternalTags(id); - let queueUrl = yield* sqs.getQueueUrl({ QueueName: queueName }).pipe( - Effect.map((result) => result.QueueUrl), - Effect.catchTag("QueueDoesNotExist", () => - Effect.succeed(undefined), - ), - ); - let requiresOwnershipCheck = queueUrl !== undefined; - - if (queueUrl === undefined) { - const observed = yield* sqs - .createQueue({ - QueueName: queueName, - ...(identity.fifo ? { Attributes: { FifoQueue: "true" } } : {}), - tags: internalTags, - }) - .pipe( - Effect.retry({ - while: (error) => error._tag === "QueueDeletedRecently", - schedule: Schedule.max([ - Schedule.fixed(1000), - Schedule.recurs(30), - ]).pipe( - Schedule.tap(({ attempt }) => - session.note( - `Queue was deleted recently, retrying... ${attempt}s`, - ), - ), - ), - }), - Effect.map((result) => ({ - queueUrl: result.QueueUrl!, - raced: false, - })), - Effect.catchTag("QueueNameExists", () => - sqs.getQueueUrl({ QueueName: queueName }).pipe( - Effect.map((result) => ({ - queueUrl: result.QueueUrl!, - raced: true, - })), - ), - ), - ); - queueUrl = observed.queueUrl; - requiresOwnershipCheck = observed.raced; - } - - if (requiresOwnershipCheck) { - const [tags, attributes] = yield* Effect.all([ - sqs - .listQueueTags({ QueueUrl: queueUrl }) - .pipe(Effect.map((result) => result.Tags ?? {})), - sqs.getQueueAttributes({ - QueueUrl: queueUrl, - AttributeNames: ["FifoQueue"], - }), - ]); - if (!(yield* hasAlchemyTags(id, tags))) { - return yield* new QueuePrecreateConflict({ - id, - queueName, - reason: - "an existing queue with that name is not owned by this stack", - }); - } - const observedFifo = attributes.Attributes?.FifoQueue === "true"; - if (observedFifo !== (identity.fifo ?? false)) { - return yield* new QueuePrecreateConflict({ - id, - queueName, - reason: `the existing queue is ${observedFifo ? "FIFO" : "standard"}, but the desired queue is ${identity.fifo ? "FIFO" : "standard"}`, - }); - } + const existingUrl = yield* getQueueUrl(queueName); + const observed = + existingUrl !== undefined + ? { queueUrl: existingUrl, raced: true } + : yield* createQueueShell({ + queueName, + attributes: identity.fifo ? { FifoQueue: "true" } : {}, + tags: internalTags, + session, + }); + if (observed.raced) { + yield* verifyQueueOwnership({ + id, + queueName, + queueUrl: observed.queueUrl, + fifo: identity.fifo, + }); } - return { queueName, queueUrl, queueArn }; + return { queueName, queueUrl: observed.queueUrl, queueArn }; }), reconcile: Effect.fn(function* ({ id, @@ -608,12 +670,11 @@ export const QueueProvider = () => bindings, }) { yield* validateEncryption(news); - const { accountId, region } = yield* AWSEnvironment.current; - const queueName = - output?.queueName ?? (yield* createQueueName(id, news)); - const queueArn = - output?.queueArn ?? - (`arn:aws:sqs:${region}:${accountId}:${queueName}` as const); + const identity = output + ? output + : yield* resolveQueueIdentity(id, news); + const queueName = identity.queueName; + const queueArn = identity.queueArn; const desiredAttributes = createAttributes(news, bindings); const internalTags = yield* createInternalTags(id); @@ -623,12 +684,7 @@ export const QueueProvider = () => // deleted out-of-band, downstream API calls fail with // `QueueDoesNotExist` and we recreate. This keeps the reconciler // convergent regardless of the starting cloud state. - let queueUrl = yield* sqs.getQueueUrl({ QueueName: queueName }).pipe( - Effect.map((r) => r.QueueUrl!), - Effect.catchTag("QueueDoesNotExist", () => - Effect.succeed(undefined), - ), - ); + let queueUrl = yield* getQueueUrl(queueName); if (queueUrl === undefined) { // `createQueue` is idempotent for identical params; with different @@ -643,40 +699,19 @@ export const QueueProvider = () => if (value === undefined || value === "") continue; createAttrs[key] = value; } - queueUrl = yield* sqs - .createQueue({ - QueueName: queueName, - Attributes: createAttrs, - tags: { ...internalTags, ...news.tags }, - }) - .pipe( - Effect.retry({ - while: (e) => e._tag === "QueueDeletedRecently", - schedule: Schedule.fixed(1000).pipe( - Schedule.tap(({ attempt }) => - session.note( - `Queue was deleted recently, retrying... ${attempt}s`, - ), - ), - ), - }), - // A `RedrivePolicy` referencing a just-created dead-letter - // queue is transiently rejected with - // `InvalidParameterValueException` until that DLQ's ARN is - // visible to SQS. It's an eventual-consistency race, not a - // genuine validation failure, so retry on a bounded schedule. - Effect.retry({ - while: (e) => e._tag === "InvalidParameterValueException", - schedule: Schedule.max([ - Schedule.fixed(1000), - Schedule.recurs(30), - ]), - }), - Effect.catchTag("QueueNameExists", () => - sqs.getQueueUrl({ QueueName: queueName }), - ), - Effect.map((r) => r.QueueUrl!), - ); + const observed = yield* createQueueShell({ + queueName, + attributes: createAttrs, + tags: { ...internalTags, ...news.tags }, + session, + // A `RedrivePolicy` referencing a just-created dead-letter + // queue is transiently rejected with + // `InvalidParameterValueException` until that DLQ's ARN is + // visible to SQS. It's an eventual-consistency race, not a + // genuine validation failure, so retry on a bounded schedule. + retryInvalidParameter: true, + }); + queueUrl = observed.queueUrl; } // Sync attributes — diff observed cloud state against desired and