Scheduled Postgres log cleanup in Node: cron plans, queue batches, safe replays

Deleting old log rows on a schedule without locking the table: a planner that pins an absolute cutoff, fixed-size batches on a queue, and a worker safe to replay.

A DELETE FROM logs WHERE created_at < now() - interval '90 days' is correct and still the wrong thing to run at midnight, because on a big table it takes one long transaction, holds one long lock, and gives you nothing to resume from when it dies at minute forty. Break it into fixed-size chunks, put each chunk on a queue, and let a worker grind through them. Infrai’s queue routes make that a publish and a consume loop — no broker to run, and only the publish is metered.

The subtle part isn’t the deleting. It’s that a queue delivers at least once, so every chunk message has to be safe to process twice, and a relative cutoff quietly isn’t.

The delete is idempotent; the plan is not

DELETE ... WHERE id BETWEEN 400000 AND 405000 run twice deletes nothing the second time. Fine. But a message that says “delete anything older than 90 days” means something different at 00:05 than it does when it’s redelivered at 06:30 after a worker crash, and the second reading eats six hours of rows nobody asked you to remove.

So the planner resolves time once and writes the resolved value into every message.

Plan at a fixed cutoff

import process from "node:process";
import { parseArgs } from "node:util";
import pg from "pg";

const { values } = parseArgs({ options: { days: { type: "string", default: "90" }, chunk: { type: "string", default: "5000" } } });
const KEY = process.env.INFRAI_API_KEY;
const DSN = process.env.DATABASE_URL;
if (!KEY || !DSN) throw new Error("set INFRAI_API_KEY and DATABASE_URL");

const cutoff = new Date(Date.now() - Number(values.days) * 86_400_000).toISOString();
const chunkSize = Number(values.chunk);

async function callInfrai(path, payload) {
  const res = await fetch(`https://api.infrai.cc${path}`, {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify(payload),
  });
  const parsed = await res.json();
  if (parsed.ok !== true) throw new Error(`${path} failed: ${parsed.error?.code} ${parsed.error?.message}`);
  return parsed.data;
}

const db = new pg.Client({ connectionString: DSN });
await db.connect();
try {
  const { rows } = await db.query("SELECT min(id) AS lo, max(id) AS hi FROM app_logs WHERE created_at < $1", [cutoff]);
  if (rows[0].lo === null) { console.log("nothing older than the cutoff"); process.exit(0); }
  let planned = 0;
  for (let from = Number(rows[0].lo); from <= Number(rows[0].hi); from += chunkSize) {
    await callInfrai("/v1/queue/publish", {
      queue: "log-cleanup",
      body: { cutoff, id_from: from, id_to: from + chunkSize - 1 },
    });
    planned++;
  }
  console.log(`planned ${planned} chunks of ${chunkSize} up to cutoff ${cutoff}`);
} finally {
  await db.end();
}

One publish per chunk, and the chunk boundaries are pure arithmetic over the id range — no counting query, no OFFSET, no second scan of a table you’re about to shrink.

Here’s the same publish as a shell one-liner, for when you’re testing the shape by hand:

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":"log-cleanup","body":{"cutoff":"2026-04-27T00:00:00Z","id_from":400000,"id_to":404999}}'

{
  "ok": true,
  "data": {
    "message_id": "qmsg_MREXMkwF9erDIvvcxVbQ3eWr",
    "queue": "log-cleanup",
    "payload": { "cutoff": "2026-04-27T00:00:00Z", "id_from": 400000, "id_to": 404999 },
    "status": "available",
    "delivery_count": 0,
    "published_at": "2026-07-26T00:33:41.568872Z"
  }
}

The statement each chunk runs

Bound it twice — by id range and by the pinned cutoff — so a bad range can never outrun the retention policy.

DELETE FROM app_logs
WHERE id >= $1
  AND id <= $2
  AND created_at < $3;

The worker

import process from "node:process";
import { setTimeout as pause } from "node:timers/promises";
import pg from "pg";

const KEY = process.env.INFRAI_API_KEY;
const DSN = process.env.DATABASE_URL;
if (!KEY || !DSN) throw new Error("set INFRAI_API_KEY and DATABASE_URL");

async function callInfrai(path, payload) {
  const res = await fetch(`https://api.infrai.cc${path}`, {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify(payload),
  });
  const parsed = await res.json();
  if (parsed.ok !== true) throw new Error(`${path} failed: ${parsed.error?.code} ${parsed.error?.message}`);
  return parsed.data;
}

const db = new pg.Client({ connectionString: DSN });
await db.connect();

let emptyPolls = 0;
while (emptyPolls < 10) {
  const { items } = await callInfrai("/v1/queue/consume", { queue: "log-cleanup", max_messages: 5 });
  if (items.length === 0) { emptyPolls++; await pause(2000); continue; }
  emptyPolls = 0;

  for (const msg of items) {
    const { cutoff, id_from, id_to } = msg.payload;
    try {
      const started = Date.now();
      const out = await db.query(
        "DELETE FROM app_logs WHERE id >= $1 AND id <= $2 AND created_at < $3",
        [id_from, id_to, cutoff],
      );
      await callInfrai("/v1/queue/ack", { queue: "log-cleanup", receipt_handle: msg.message_id });
      console.log(`chunk ${id_from}-${id_to}: ${out.rowCount} rows in ${Date.now() - started}ms`);
    } catch (e) {
      console.error(`chunk ${id_from}-${id_to} failed on delivery ${msg.delivery_count}: ${e.message}`);
    }
    await pause(250); // give autovacuum and your read replicas some air
  }
}
await db.end();

The 250ms pause is doing quiet work. Deleting five thousand rows a second sustained will out-run autovacuum on a busy table and leave you with bloat that costs more disk than the logs did.

Watching it shrink

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

{
  "ok": true,
  "data": {
    "queue": "log-cleanup",
    "message_count": 14,
    "available_count": 9,
    "in_flight_count": 5,
    "delayed_count": 0,
    "dlq_count": 0,
    "oldest_message_age_seconds": 41
  }
}

If dlq_count moves off zero, a chunk has failed three deliveries — usually a statement timeout on a range that turned out to be much denser than its neighbours. Halve the chunk size and replan.

What the sweep costs

curl -sS "https://api.infrai.cc/v1/discovery" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  | jq '.capabilities[] | select(.id == "queue.publish") | .billing'

Publishing costs $0.00002 per message, verified 2026-07-26; consume, ack and stats are free and rate-limited rather than metered. A nightly cleanup of 20 million rows in 5,000-row chunks is 4,000 messages, so about $0.08 a night. New accounts start with $2 of free credit, which covers a lot of nights. Read the rate live rather than from this page — these numbers drift downward and discounts run.

Where a different tool is the better answer

If your whole cleanup is one statement and your table is small enough that it finishes in seconds, pg_cron keeps it inside the database and you’d be better off skipping the queue entirely — no worker, no credential, nothing to deploy. If you’re a Python shop already running celery with a beat scheduler, that stack does chunked cleanup perfectly well and adding a second system buys you nothing. sidekiq is the same argument for Ruby.

The honest limitations on this route: max_messages caps at 10 per consume call, messages cap at 256 KB, and a queue-level delivery_delay_seconds had no effect when we tried it — so a “run this chunk in an hour” delay isn’t something to build on. What you do get is the second question for free: the same key that carries these chunk messages also sends the completion email, records the error when a chunk fails, and reports the storage you just reclaimed.

References

• Infrai queue API reference
• QUEUE_MESSAGE_NOT_IN_FLIGHT error reference
• Citus Data: pg_cron, periodic jobs in Postgres
• Sequin: time-based retention strategies in Postgres