Lease-Based ECS Leadership: Singleton Cron That Survives Rolling Deploys

// lease/ddb-lease.ts
export async function tryAcquire(job: string, owner: string, ttlSec = 30) {
 const now = Math.floor(Date.now() / 1000);
 try {
 const out = await ddb.update({
 Key: { pk: `lease#${job}` },
 UpdateExpression:
 "SET #o = :o, #exp = :exp, fence = if_not_exists(fence, :z) + :one",
 ConditionExpression: "attribute_not_exists(pk) OR #exp < :now",
 ExpressionAttributeNames: { "#o": "owner", "#exp": "expiresAt" },
 ExpressionAttributeValues: {
 ":o": owner,
 ":exp": now + ttlSec,
 ":now": now,
 ":z": 0,
 ":one": 1,
 },
 ReturnValues: "ALL_NEW",
 });
 return { ok: true as const, fence: out.Attributes!.fence as number };
 } catch (e: any) {
 if (e.name === "ConditionalCheckFailedException") return { ok: false as const };
 throw e;
 }
}

export async function heartbeat(job: string, owner: string, fence: number, ttlSec = 30) {
 const now = Math.floor(Date.now() / 1000);
 await ddb.update({
 Key: { pk: `lease#${job}` },
 UpdateExpression: "SET expiresAt = :exp",
 ConditionExpression: "owner = :o AND fence = :f",
 ExpressionAttributeValues: {
 ":exp": now + ttlSec,
 ":o": owner,
 ":f": fence,
 },
 }); // ✅ stale leader cannot extend
}

export function installResign(job: string, owner: string, fence: number) {
 process.on("SIGTERM", async () => {
 try {
 await ddb.update({
 Key: { pk: `lease#${job}` },
 UpdateExpression: "SET expiresAt = :z",
 ConditionExpression: "owner = :o AND fence = :f",
 ExpressionAttributeValues: { ":z": 0, ":o": owner, ":f": fence },
 });
 } finally {
 process.exit(0);
 }
 });
}

export async function runSingletonTick(job: string, fence: number, work: () => Promise<void>) {
 await ddb.put({
 Item: {
 pk: `jobrun#${job}#${slotId()}`,
 fence,
 status: "started",
 },
 ConditionExpression: "attribute_not_exists(pk)",
 });
 await work();
}

export async function applySideEffect(evt: { fence: number; payload: unknown }) {
 const lease = await ddb.get({ Key: { pk: `lease#billing-nightly` } });
 if (!lease.Item || evt.fence < lease.Item.fence) {
 // ❌ stale leader after deploy — fail closed
 throw new Error("stale_fence");
 }
 await sink.write(evt.payload);
}