SQS FIFO Lambda: SQS FIFO + Lambda: Ordered Agent Job Queues Without Double-Applies

// ❌ Parallel Lambda on standard SQS — race on same workspace
export async function handler(event: SQSEvent) {
 for (const rec of event.Records) {
 const job = JSON.parse(rec.body);
 await applyPatch(job.workspaceId, job.diff); // overlapping applies corrupt git
 }
}

import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
import { createHash } from "crypto";

const sqs = new SQSClient({});
const QUEUE_URL = process.env.AGENT_JOBS_FIFO_URL!;

export type AgentJob = {
 jobId: string;
 workspaceId: string;
 tenantId: string;
 type: "apply_patch" | "run_tests" | "open_pr";
 payload: Record<string, unknown>;
};

export async function enqueueAgentJob(job: AgentJob) {
 // ✅ Group by workspace — order preserved per repo sandbox
 const groupId = `ws#${job.workspaceId}`;

 // ✅ Stable dedup for producer retries (same jobId)
 const dedupId = createHash("sha256")
 .update(`${job.jobId}:${job.type}`)
 .digest("hex")
 .slice(0, 128);

 await sqs.send(
 new SendMessageCommand({
 QueueUrl: QUEUE_URL,
 MessageBody: JSON.stringify(job),
 MessageGroupId: groupId,
 MessageDeduplicationId: dedupId,
 MessageAttributes: {
 tenantId: { DataType: "String", StringValue: job.tenantId },
 type: { DataType: "String", StringValue: job.type },
 },
 })
 );
}

// CDK-ish
const fn = new lambda.Function(this, "AgentJobWorker", {
 runtime: lambda.Runtime.NODEJS_20_X,
 handler: "worker.handler",
 timeout: cdk.Duration.minutes(2),
 reservedConcurrentExecutions: 50, // total, not per group
 code: lambda.Code.fromAsset("dist"),
});

fn.addEventSource(
 new sources.SqsEventSource(fifoQueue, {
 batchSize: 5, // FIFO still respects group ordering within batch semantics
 reportBatchItemFailures: true, // ✅ critical
 maxConcurrency: 20, // scaling config for the ESM
 })
);

import type { SQSEvent, SQSBatchResponse } from "aws-lambda";

export async function handler(event: SQSEvent): Promise<SQSBatchResponse> {
 const failures: { itemIdentifier: string }[] = [];

 // Optional: group records by MessageGroupId for clarity
 for (const rec of event.Records) {
 try {
 const job = JSON.parse(rec.body) as AgentJob;
 await processJobOnce(job, rec.messageId);
 } catch (e) {
 console.error("job_failed", rec.messageId, e);
 failures.push({ itemIdentifier: rec.messageId });
 }
 }
 return { batchItemFailures: failures };
}

import {
 DynamoDBClient,
 PutItemCommand,
 GetItemCommand,
} from "@aws-sdk/client-dynamodb";

const ddb = new DynamoDBClient({});
const TABLE = process.env.JOB_IDEMPOTENCY_TABLE!;

async function processJobOnce(job: AgentJob, sqsMessageId: string) {
 const pk = `job#${job.jobId}`;

 const prior = await ddb.send(
 new GetItemCommand({ TableName: TABLE, Key: { pk: { S: pk } } })
 );
 if (prior.Item?.status?.S === "SUCCEEDED") {
 return; // ✅ replay no-op
 }

 try {
 await ddb.send(
 new PutItemCommand({
 TableName: TABLE,
 Item: {
 pk: { S: pk },
 status: { S: "IN_FLIGHT" },
 sqsMessageId: { S: sqsMessageId },
 workspaceId: { S: job.workspaceId },
 ttl: { N: String(Math.floor(Date.now() / 1000) + 7 * 86400) },
 },
 ConditionExpression: "attribute_not_exists(pk) OR #s <> :ok",
 ExpressionAttributeNames: { "#s": "status" },
 ExpressionAttributeValues: { ":ok": { S: "SUCCEEDED" } },
 })
 );
 } catch {
 return; // lost race — other worker owns it
 }

 switch (job.type) {
 case "apply_patch":
 await applyPatch(job.workspaceId, job.payload);
 break;
 case "run_tests":
 await runTests(job.workspaceId, job.payload);
 break;
 case "open_pr":
 await openPr(job.workspaceId, job.payload);
 break;
 }

 await ddb.send(
 new PutItemCommand({
 TableName: TABLE,
 Item: {
 pk: { S: pk },
 status: { S: "SUCCEEDED" },
 sqsMessageId: { S: sqsMessageId },
 workspaceId: { S: job.workspaceId },
 ttl: { N: String(Math.floor(Date.now() / 1000) + 30 * 86400) },
 },
 })
 );
}

async function applyPatch(workspaceId: string, payload: Record<string, unknown>) {
 /* git apply — workspace-scoped lock optional */
}
async function runTests(workspaceId: string, payload: Record<string, unknown>) {}
async function openPr(workspaceId: string, payload: Record<string, unknown>) {}