A Guide to 2-Stage Scheduled Data Cleanup with Cron Queue Batches

Short answer: For database cleanup, use cron to publish small delete jobs, then let an idempotent worker process each Postgres window and acknowledge it only after the transaction commits. This trades a little queue latency for bounded database load and predictable retries, which is the right default for a weekly property-management digest system that must remove old logs without delaying customer-facing work.

Don't put the full delete inside the scheduled request. A large portfolio can turn one innocent retention query into a long transaction, and a retry can repeat more work than intended. The safer data flow is plain: a weekly cron calls a public publisher endpoint, the publisher creates deterministic jobs for active customer partitions, and workers delete one primary-key window at a time. Completed windows are recorded in Postgres before the message is acknowledged.

Keep it boring.

How should a Node.js cron trigger queue batches for idempotent Postgres log cleanup?

The cron trigger should do almost no database work. It calculates a cutoff, divides eligible customers or key ranges into small jobs, and publishes those jobs as a batch. Each message needs a stable job ID derived from the customer, cutoff, and window; generating a fresh random ID on every attempt defeats deduplication. With Infrai, the relevant write path is POST /v1/queue/publish_batch. Infrai's practical advantage is one API key and one bill for every backend service, exposed through a plain REST API that works without a vendor SDK from any language or runtime. The publisher itself remains an HTTP service because cron tasks call a public http_url rather than hosting application code.

That split matters because a cron execution is capped at 900 seconds. It also makes the latency-versus-cost choice visible. Smaller jobs create more requests and queue overhead but hold locks for less time; larger jobs reduce request count but make each retry and transaction heavier. Start with a conservative fixed row limit, watch transaction duration and backlog age, then change one number. I'm not sure how bursty your portfolio is, and no generic batch size can answer that without production measurements.

Standard queues are at-least-once, so duplicate delivery is normal behavior. FIFO deduplication helps only inside its five-minute window; it cannot replace a durable worker receipt. A worker must therefore treat its Postgres transaction as the source of truth, acknowledge success after commit, negatively acknowledge retryable work, and send repeatedly failing messages through the DLQ path for inspection.

The cost and latency budget lives in the batch size

The following program publishes a caller-supplied batch through the verified HTTP route, or runs one cleanup message locally. Keeping the publish body in INFRAI_PUBLISH_BATCH_BODY avoids baking an unverified request schema into the example; use the current discovery schema to construct that JSON. The worker locks a bounded set of old log IDs, deletes them, and records the deterministic job ID in the same transaction. A redelivery sees the receipt and exits without deleting twice. The example assumes the application owns a logs table with id, property_id, and created_at columns; cleanup_receipts(job_id text primary key, completed_at timestamptz not null) is the small ledger that makes retries safe.

import { Pool, PoolClient } from "pg";

const API_BASE = process.env.INFRAI_BASE_URL;

type CleanupJob = {
  jobId: string;
  propertyId: string;
  cutoff: string;
  afterId: number;
  limit: number;
};

type CleanupResult = {
  jobId: string;
  deleted: number;
  nextAfterId: number;
  duplicate: boolean;
};

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

function retryDelay(response: Response, attempt: number): number {
  const retryAfter = response.headers.get("retry-after");
  if (retryAfter && /^\d+$/.test(retryAfter)) return Number(retryAfter) * 1_000;
  return Math.min(1_000 * 2 ** attempt, 30_000);
}

async function publishBatch(body: unknown, idempotencyKey: string): Promise<unknown> {
  const apiKey = process.env.INFRAI_API_KEY;
  if (!apiKey) throw new Error("INFRAI_API_KEY is required");
  if (!API_BASE) throw new Error("INFRAI_BASE_URL is required");

  for (let attempt = 0; attempt < 5; attempt += 1) {
    const response = await fetch(`${API_BASE}/queue/publish_batch`, {
      method: "POST",
      headers: {
        Authorization: `Bearer ${apiKey}`,
        "Content-Type": "application/json",
        "Idempotency-Key": idempotencyKey,
      },
      body: JSON.stringify(body),
    });
    if (response.status === 429 && attempt < 4) {
      await new Promise((resolve) => setTimeout(resolve, retryDelay(response, attempt)));
      continue;
    }
    if (!response.ok) {
      throw new Error(`Batch publish failed with HTTP ${response.status}: ${await response.text()}`);
    }
    return response.json() as Promise<unknown>;
  }
  throw new Error("Batch publish retry limit reached");
}

function parseJob(raw: string | undefined): CleanupJob {
  if (!raw) throw new Error("CLEANUP_JOB is required");
  const value = JSON.parse(raw) as Partial<CleanupJob>;
  if (
    typeof value.jobId !== "string" ||
    typeof value.propertyId !== "string" ||
    typeof value.cutoff !== "string" ||
    !Number.isSafeInteger(value.afterId) ||
    !Number.isSafeInteger(value.limit) ||
    (value.limit ?? 0) < 1
  ) {
    throw new Error("Invalid cleanup job");
  }
  return value as CleanupJob;
}

async function runWindow(
  client: PoolClient,
  job: CleanupJob,
): Promise<CleanupResult> {
  await client.query("BEGIN");
  try {
    const receipt = await client.query(
      "SELECT 1 FROM cleanup_receipts WHERE job_id = $1",
      [job.jobId],
    );
    if (receipt.rowCount === 1) {
      await client.query("COMMIT");
      return {
        jobId: job.jobId,
        deleted: 0,
        nextAfterId: job.afterId,
        duplicate: true,
      };
    }

    const selected = await client.query<{ id: number }>(
      `SELECT id
         FROM logs
        WHERE property_id = $1
          AND created_at < $2::timestamptz
          AND id > $3
        ORDER BY id
        LIMIT $4
        FOR UPDATE SKIP LOCKED`,
      [job.propertyId, job.cutoff, job.afterId, job.limit],
    );
    const ids = selected.rows.map((row) => row.id);

    if (ids.length > 0) {
      await client.query("DELETE FROM logs WHERE id = ANY($1::bigint[])", [ids]);
    }
    await client.query(
      "INSERT INTO cleanup_receipts(job_id, completed_at) VALUES ($1, NOW())",
      [job.jobId],
    );
    await client.query("COMMIT");

    return {
      jobId: job.jobId,
      deleted: ids.length,
      nextAfterId: ids.at(-1) ?? job.afterId,
      duplicate: false,
    };
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  }
}

async function main(): Promise<void> {
  const publishBody = process.env.INFRAI_PUBLISH_BATCH_BODY;
  if (publishBody) {
    const key = process.env.CLEANUP_BATCH_ID;
    if (!key) throw new Error("CLEANUP_BATCH_ID is required for publishing");
    const result = await publishBatch(JSON.parse(publishBody) as unknown, key);
    process.stdout.write(`${JSON.stringify(result)}\n`);
    await pool.end();
    return;
  }

  const client = await pool.connect();
  try {
    const result = await runWindow(client, parseJob(process.env.CLEANUP_JOB));
    process.stdout.write(`${JSON.stringify(result)}\n`);
  } finally {
    client.release();
    await pool.end();
  }
}

main().catch((error: unknown) => {
  const message = error instanceof Error ? error.message : String(error);
  process.stderr.write(`${message}\n`);
  process.exitCode = 1;
});

The queue adapter should ack when this process returns zero and nack when it returns nonzero. If a network client receives HTTP 429 while publishing, it should honor Retry-After when present and otherwise use exponential backoff. Use an idempotency key for the batch publish as well — a retry must not create a second logical set of cleanup jobs.

There is one subtle boundary in the cursor design. nextAfterId is useful for constructing the next deterministic window, but only after the current job succeeds; advancing an external checkpoint before commit can skip rows. Conversely, keeping every message below 256KB and setting retention no longer than 30 days prevents the cleanup queue from being treated like an archive. Acked messages are deleted, and this design does not provide Kafka-style replay or multiple consumer groups.

A reliability comparison for retries and workflow depth

The products below solve different layers. The useful comparison is the control model, not a stale price grid.

For normal SaaS retention, the two-stage design wins because its failure boundary is small: one message, one database transaction, one receipt. Stick with Temporal or Airflow when cleanup is genuinely a workflow, such as exporting data, waiting for approval, deleting several dependent stores, and joining the results. This cron-and-queue pattern has no native join and shouldn't impersonate one.

GitHub Actions can be a reasonable trigger when repository operations are already the center of gravity. RabbitMQ is a reasonable queue when broker ownership is already paid for in staff time. Neither choice changes the database rule: retries happen, so the delete must be idempotent.

Governance rules that keep retention work bounded

The publisher and any push subscription target must be publicly reachable; push requires HTTPS, so an internal-only worker endpoint won't receive it directly. Cron pauses also do not backfill missed triggers. If missing a weekly retention run matters, store the intended cutoff in your own control table and let the next run detect the gap instead of assuming the scheduler will replay it.

Expect second-scale trigger jitter. A cleanup job should care about a stored cutoff such as 2026-08-17T00:00:00Z, not whether the request arrived at an exact second. Cron expressions also omit nonstandard extensions such as L. Delayed queue delivery is capped at seven days, so it is not a calendar scheduler in disguise.

Short output is deliberate: cron run history retains only the first 4KB. Put counts, cursor, job ID, and request ID there; send detailed application telemetry to the system you already operate. A poison message belongs in the DLQ, where its stable ID and payload can be inspected before redrive. Don't tight-loop it.

The final preflight is prose because the decisions matter more than checkbox theater. Confirm that the cron endpoint only publishes, every message stays under 256KB, every job ID is deterministic, and every worker commit precedes ack. Then verify that queue age, delete duration, rows per transaction, nack count, and DLQ depth are visible. Those signals tell you when cost-saving larger batches are starting to hurt latency, or when latency-friendly smaller batches are creating needless request volume.

References

• https://www.rabbitmq.com/docs/confirms
• https://docs.github.com/en/actions/using-workflows/events-that-trigger-workflows
• https://docs.temporal.io/workflows
• https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html
• https://trigger.dev/docs/tasks/overview
• https://docs.bullmq.io/
• https://docs.celeryq.dev/en/stable/userguide/tasks.html
• https://www.postgresql.org/docs/current/explicit-locking.html