Skip to content

fix(aws-sqs): precreate self-referential queue identity #1356

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aryasaatvik wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(aws-sqs): precreate self-referential queue identity #1356

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
864fbaa
fix(aws-sqs): precreate self-referential queue identity
aryasaatvik Aug 24, 2026
3cf502e
fix(aws-sqs): bound precreate queue retry
aryasaatvik Aug 25, 2026
6c6d3f3
refactor(aws-sqs): share queue identity creation logic
aryasaatvik Aug 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
241 changes: 195 additions & 46 deletions packages/alchemy/src/AWS/SQS/Queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -20,6 +21,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.
Expand Down Expand Up @@ -405,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<string, string>;
tags: Record<string, string>;
session: Pick<ScopedPlanStatusSession, "note">;
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`
Expand Down Expand Up @@ -486,6 +629,39 @@ export const QueueProvider = () =>
}
// Return undefined to allow update function to be called for other attribute changes
}),
precreate: Effect.fn(function* ({ id, news = {}, session }) {

@sam-goodwin sam-goodwin Aug 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we share any logic with create? Looks a bit redundant.

@aryasaatvik aryasaatvik Aug 25, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I pulled the shared queue logic into helpers; precreate still only creates the shell. Pushed in 6c6d3f3

// 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 = yield* resolveQueueIdentity(id, news);
const { queueName, queueArn } = identity;
const internalTags = yield* createInternalTags(id);
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: observed.queueUrl, queueArn };
}),
reconcile: Effect.fn(function* ({
id,
news = {},
Expand All @@ -494,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);

Expand All @@ -509,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
Expand All @@ -529,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
Expand Down
42 changes: 42 additions & 0 deletions packages/alchemy/test/AWS/SQS/Queue.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down