Scheduled S3 cleanup jobs with a queue, retries and a DLQ

A nightly object-cleanup run that survives partial failure: fan out one message per prefix on Infrai's queue, let three failed deliveries fall into the DLQ, redrive after the fix.

A nightly cleanup that deletes files from S3 shouldn’t be one long script. Split it into three parts: whatever you already use as a scheduler fires once and publishes one message per prefix, a worker deletes and acknowledges each message, and anything that fails three deliveries drops into a dead-letter queue you can read. Infrai’s queue covers the middle and the end of that over plain REST, with nothing to operate.

Firing on a schedule is the easy half. Knowing which 4,000 objects didn’t get deleted, and why, is the half that decides whether the job is trustworthy.

What breaks in a single cron script

The script wakes at 03:00, lists a few hundred thousand keys, and starts deleting. At object 180,000 an S3 call returns a 503. The process either dies with everything after it undone, or swallows the error and reports success. Re-running from the top re-lists everything you already deleted, so the run gets slower every night while quietly leaving a residue behind.

One message per unit of work fixes all of that, because failure becomes a per-message fact instead of a per-process one.

If your objects already live in S3 and your compute is already Lambda, the SQS route is fewer moving parts than anything else and you should take it. BullMQ is the better answer when Redis is running for other reasons and the whole job is in-process Node. The row worth reading twice is the last one: the queue arrives with no broker, and the DLQ threshold is not something you configure.

Publish one message per prefix

Queues auto-provision on the first publish, together with a .dlq companion, so there’s no setup call to forget. Each message names a prefix and an age cutoff — small payloads, well under the 256 KB ceiling that returns QUEUE_MESSAGE_TOO_LARGE.

export INFRAI_API_KEY="your_infrai_api_key"

curl -sS -X POST "https://api.infrai.cc/v1/queue/publish" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"queue":"s3-cleanup","body":{"prefix":"uploads/2026-05/","older_than_days":30}}'

The response carries the id you’ll see again in the DLQ if this unit of work never succeeds:

{
  "ok": true,
  "data": {
    "message_id": "qmsg_dZUpdNrIDejPgmMBPA0jY2rr",
    "queue": "s3-cleanup",
    "payload": { "prefix": "uploads/2026-05/", "older_than_days": 30 },
    "status": "available",
    "delivery_count": 0,
    "published_at": "2026-07-26T00:55:17.757733Z"
  }
}

Note the response echoes the field as payload, not body. Both spellings are accepted on the way in; only one comes back.

The planner

Whatever fires this — a container cron entry, a CI schedule, a systemd timer — its only job is to turn “clean up May” into a list of publishes and exit.

import process from "node:process";

const API = "https://api.infrai.cc";
const key = process.env.INFRAI_API_KEY;
if (!key) throw new Error("INFRAI_API_KEY is required");

const PREFIXES = [
  "uploads/2026-05/", "uploads/2026-06/", "exports/csv/", "tmp/thumbnails/",
];

async function publish(prefix) {
  const res = await fetch(`${API}/v1/queue/publish`, {
    method: "POST",
    headers: { Authorization: `Bearer ${key}`, "Content-Type": "application/json" },
    body: JSON.stringify({ queue: "s3-cleanup", body: { prefix, older_than_days: 30 } }),
  });
  const out = await res.json();
  if (!out.ok) throw new Error(`${out.error.code}: ${out.error.message}`);
  return out.data.message_id;
}

let queued = 0;
for (const prefix of PREFIXES) {
  try {
    const id = await publish(prefix);
    console.log(`queued ${prefix} as ${id}`);
    queued += 1;
  } catch (err) {
    console.error(`could not queue ${prefix}: ${err.message}`);
  }
}
console.log(`planner queued ${queued}/${PREFIXES.length} prefixes`);

Publishing takes an optional per-message delay_seconds if you want the deletes spread across the small hours instead of arriving at once; keep it at or below 604800, which is the ceiling the API accepts.

The delete worker

The worker is a loop that never assumes it’s the only one running. Messages become invisible for the queue’s visibility timeout (300 seconds by default) while you hold them, so a second worker won’t take the same prefix, and a crash simply returns the message to the queue.

import process from "node:process";
import { S3Client, ListObjectsV2Command, DeleteObjectsCommand } from "@aws-sdk/client-s3";

const API = "https://api.infrai.cc";
const key = process.env.INFRAI_API_KEY;
const bucket = process.env.S3_BUCKET;
if (!key || !bucket) throw new Error("INFRAI_API_KEY and S3_BUCKET are required");

const s3 = new S3Client({ region: process.env.AWS_REGION ?? "us-east-1" });
const headers = { Authorization: `Bearer ${key}`, "Content-Type": "application/json" };

async function call(path, payload) {
  const res = await fetch(`${API}${path}`, { method: "POST", headers, body: JSON.stringify(payload) });
  const out = await res.json();
  if (!out.ok) throw new Error(`${path}: ${out.error.code} ${out.error.message}`);
  return out.data;
}

async function purgePrefix(prefix, olderThanDays) {
  const cutoff = Date.now() - olderThanDays * 86400_000;
  let token;
  let deleted = 0;
  do {
    const page = await s3.send(new ListObjectsV2Command({
      Bucket: bucket, Prefix: prefix, ContinuationToken: token, MaxKeys: 1000,
    }));
    const stale = (page.Contents ?? []).filter((o) => o.LastModified.getTime() < cutoff);
    if (stale.length) {
      await s3.send(new DeleteObjectsCommand({
        Bucket: bucket,
        Delete: { Objects: stale.map((o) => ({ Key: o.Key })), Quiet: true },
      }));
      deleted += stale.length;
    }
    token = page.NextContinuationToken;
  } while (token);
  return deleted;
}

export async function drain() {
  const { items } = await call("/v1/queue/consume", { queue: "s3-cleanup", max_messages: 5 });
  for (const msg of items) {
    const { prefix, older_than_days } = msg.payload;
    try {
      const n = await purgePrefix(prefix, older_than_days);
      await call("/v1/queue/ack", { queue: "s3-cleanup", receipt_handle: msg.message_id });
      console.log(`deleted ${n} objects under ${prefix}`);
    } catch (err) {
      console.error(`delivery ${msg.delivery_count} of ${prefix} failed: ${err.message}`);
      await call("/v1/queue/nack", { queue: "s3-cleanup", message_id: msg.message_id });
    }
  }
  return items.length;
}

Two spelling details will cost you an afternoon if you skip them. queue.ack takes the handle under receipt_handle, while queue.nack insists on message_id and rejects receipt_handle outright — and the consume response ships only message_id, which is the value both of them want. Not acking at all works too: the message reappears when its lease expires, which is the right behaviour if the worker was killed mid-delete.

Where the failures collect

Three deliveries is the threshold, and it isn’t adjustable — the queue record reports max_receive_count: 3 and keeps reporting 3 after you try to change it. That’s a real limitation if your workload wants ten attempts before giving up; count retries inside the payload and re-publish if you need more.

Reading the dead letters is where the documented path and the working path diverge. GET /v1/queue/dlq/list/s3-cleanup returned an empty items array in our testing even with dlq_count at 1, so consume the companion queue by name instead:

curl -sS "https://api.infrai.cc/v1/queue/stats/s3-cleanup" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

curl -sS -X POST "https://api.infrai.cc/v1/queue/consume" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"queue":"s3-cleanup.dlq","max_messages":10}'

Once the bucket policy or the credential is fixed, put a message back one at a time. Passing a message_id works; asking for a bulk drain of the whole DLQ currently fails with an internal error, so loop over the ids you collected.

curl -sS -X POST "https://api.infrai.cc/v1/queue/dlq/redrive/s3-cleanup" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"message_id":"qmsg_ssURFXHvZXjCMdGtmrQ1c7wY"}'

What the run costs

Only the publish is metered: $0.00002 per message, verified 2026-07-26. Consuming, acking, nacking, stats and DLQ reads are free within rate limits, so a message that gets retried three times costs the same as one that succeeds first time. New accounts start with $2 of credit, which is roughly 100,000 publishes — more cleanup fan-out than most teams generate in a year.

curl -sS "https://api.infrai.cc/v1/account/usage" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

Rates here have moved down over time and discount campaigns run, so that call is the honest number and this paragraph is only the illustration.

The same credential also reaches storage, email and error tracking, which matters more than the rate: when the cleanup worker wants to mail a summary or record a failed prefix, that’s another call on the account you already have rather than another vendor to onboard.

References

• Infrai queue API reference
• QUEUE_MESSAGE_TOO_LARGE error reference
• Amazon SQS dead-letter queues
• BullMQ documentation