SQS

SQS is a message queue. A producer puts a message on a Queue and moves on. A consumer takes messages off later, does the work, and deletes them. The two never wait for each other, so a slow job doesn’t slow down the request that asked for it, and a burst of work waits in the Queue instead of overwhelming whatever does it.

Consumers take messages in any order

A standard Queue doesn’t keep messages in a line. It spreads them across many servers, and each receive returns whichever messages it finds, up to 10 at a time. So consumers get messages roughly, but not exactly, in the order they were sent:

Worker 1 has job 5 and Worker 2 has job 2, while job 1 still waits. The messages they hold stay in the Queue, hidden from everyone else.

A message is received, then deleted

A message stays on the Queue until a consumer deletes it. Receiving it doesn’t remove it. It only hides it from other consumers for the visibility timeout, 30 seconds by default.

Receiving a message hides it. Deleting it removes it. A message nobody deletes comes back, so the same job can run twice.

If the consumer finishes, it deletes the message and it’s gone. If the consumer crashes or times out, the message becomes visible again and another consumer receives it. So a message is never lost to a failure, but it can be delivered more than once. Write consumers so that doing a job twice is safe: check whether it’s done already, or make the work idempotent.

Declare a Queue

A Queue is a declaration. Put it in its own module, so the producer and the consumer can both import it:

src/jobs.ts

import * as AWS from "@alchemy.run/aws";

export const Jobs = AWS.SQS.Queue("Jobs");

Messages are kept for four days by default. Set messageRetentionPeriod to keep them longer, up to 14 days, and set visibilityTimeout longer than your consumer takes to finish one batch:

src/jobs.ts

export const Jobs = AWS.SQS.Queue("Jobs", { visibilityTimeout: "2 minutes" });

Send messages

Bind SQS.SendMessage in the producer. Its role gets sqs:SendMessage on this Queue only:

src/api.ts

const sendMessage = yield* AWS.SQS.SendMessage(Jobs);

A message body is a string, up to 256 KB. Send JSON:

src/api.ts

yield* sendMessage({ MessageBody: JSON.stringify({ type: "thumbnail", key }) });

Provide AWS.SQS.SendMessageHttp to the producer. Sending returns as soon as SQS has stored the message, so the request handler can respond while the work waits.

Consume messages

A Lambda Function consumes a Queue by subscribing to it from Construction. Lambda polls the Queue for you and invokes the Function with batches of messages:

src/worker.ts

export default class Worker extends AWS.Lambda.Function<Worker>()(
  "Worker",
  { main: import.meta.url, functionUrl: false },
  Effect.gen(function* () {
    yield* AWS.SQS.consumeQueueMessages(Jobs, { batchSize: 10 }, (event) =>
      Effect.forEach(event.Records, (record) => Console.log(record.body), { discard: true }),
    );
    return {};
  }).pipe(Effect.provide(AWS.Lambda.QueueEventSource)),
) {}

Deploy creates the event source mapping and grants the Function sqs:ReceiveMessage, sqs:DeleteMessage and sqs:GetQueueAttributes on the Queue. When the handler succeeds, Lambda deletes the whole batch. When it fails, the whole batch becomes visible again and is retried.

A process on ECS or EC2 consumes the same way, with its own event source layer: it long-polls the Queue in a loop instead of being invoked.

Retry only the messages that failed

Failing a whole batch for one bad message repeats the work that succeeded. Instead, return the ids of the messages that failed. Lambda deletes the rest and redelivers only those:

src/worker.ts

yield* AWS.SQS.consumeQueueMessages(Jobs, { batchSize: 10 }, (event) =>
  Effect.filter(
    event.Records,
    (record) =>
      handleJob(record.body).pipe(
        Effect.as(false),
        Effect.catchCause(() => Effect.succeed(true)),
      ),
    { concurrency: 4 },
  ).pipe(
    Effect.map((failed) => ({
      batchItemFailures: failed.map((record) => ({ itemIdentifier: record.messageId })),
    })),
  ),
);

Messages in a batch are processed four at a time here, and one failure no longer drags the others back onto the Queue.

Move poison messages aside

Some messages fail every time: malformed input, a bug, a record that no longer exists. Without a limit, they’re retried until they expire. A dead-letter queue catches them instead. After a message has been received maxReceiveCount times, SQS moves it there:

alchemy.run.ts

const deadLetters = yield* AWS.SQS.Queue("JobsDeadLetters", { messageRetentionPeriod: "14 days" });
const jobs = yield* AWS.SQS.Queue("Jobs", {
  redrivePolicy: { deadLetterTargetArn: deadLetters.queueArn, maxReceiveCount: 5 },
});

The work stops failing in a loop, and the failed messages wait for you to look at them. Put an alarm on the dead-letter queue’s depth so you know when one arrives.

Keep order with a FIFO queue

A standard Queue delivers messages roughly in order, at least once. A FIFO queue delivers them exactly in order and exactly once, within each message group:

src/ledger.ts

export const Ledger = AWS.SQS.Queue("Ledger", { fifo: true, contentBasedDeduplication: true });

Every message names its group, like an account id. Messages in one group are delivered one after another, and different groups are processed in parallel:

src/api.ts

yield* sendMessage({ MessageBody: body, MessageGroupId: accountId });

Each group is a line of its own. While ana #1 is in flight, SQS hands out nothing else from acct_ana, so her entries run in order. acct_sam moves at the same time, on Worker 2.

FIFO queues trade throughput for that guarantee. Use one when the order of messages for the same thing matters, like the entries in one account’s ledger.

Send many messages at once

To forward a Stream of messages, like a table’s changes or a file’s lines, bind SQS.QueueSink. It batches messages ten at a time into sqs:SendMessageBatch calls:

src/api.ts

const sink = yield* AWS.SQS.QueueSink(Jobs);
yield* Stream.fromIterable(keys).pipe(
  Stream.map((key) => JSON.stringify({ type: "thumbnail", key })),
  Stream.run(sink),
);

QueueSinkHttp is built on SendMessageBatchHttp, so provide both: QueueSinkHttp with Layer.provideMerge(AWS.SQS.SendMessageBatchHttp) beneath it.

When one event should reach several queues or functions, publish it to an SNS Topic instead, and subscribe each Queue to the Topic.