Scheduled Cleanup of Stale Uploads in EU SaaS: Node.js Pacing

Use a small nightly scheduler to create bounded cleanup work, then let rate-limited workers delete stale uploads; do not make the scheduler hold the whole backlog open. That separation is the practical answer for an EU SaaS where a fixed reservation or upload hold window can expire thousands of objects at once and the important trade-off is latency versus cost.

Short answer: select candidates in short, repeatable pages, enqueue one tenant-scoped command per upload, and acknowledge each command only after an idempotent delete and state update. The schedule tells you when to look. It should not be responsible for how fast an external storage system can absorb the work.

This is an architecture decision record, not a recipe for one queue product. The invariants are more valuable than the brand: a stale object must not become live again while cleanup is in flight, a duplicate delivery must be harmless, and a tenant must never be inferred from an object name alone. I've been paged for missed jobs and duplicate deliveries; both incidents started with a deceptively small cron handler.

Decision record: put each failure boundary somewhere explicit

The application owns eligibility. A scheduled trigger starts a pass and records a run identifier. A database query selects only a bounded page of uploads whose hold window has expired and whose version still matches the candidate. A publisher hands off commands containing the tenant identifier, upload identifier, and expected version. The worker rechecks eligibility, applies the downstream rate limit, performs the delete, writes the terminal state, and only then acknowledges the command.

That order matters. If an upload is renewed after selection, the worker's second check prevents a stale snapshot from deleting the renewed object. If the process stops after deletion but before acknowledgement, the queue may deliver the command again. The second delivery should see an absent object or a recorded terminal result and finish without changing a newer upload. At-least-once delivery is a design input, not a rare anomaly to hide in logs.

The database remains authoritative.

Consider the awkward middle of the operation, because that is where clean diagrams become misleading. The worker has selected tenant acme, upload u-1842, and version 7. It checks the row, waits for a rate slot, and sends the delete. During that wait, the customer may retry an upload or extend its hold; after the delete returns, the conditional update must match both u-1842 and version 7, so it cannot mark version 8 as deleted. Now reverse the timing: the delete succeeds, the worker loses its connection before acknowledgement, and a second worker receives the same command. The second worker must not treat absence as a dangerous surprise, nor may it blindly write a deletion state without checking the version. Those two checks turn a timing race into an ordinary terminal outcome. The command's operation identifier helps trace the duplicate, but it does not replace the conditional database write. This is the sort of failure a test should force with a fake clock and a storage adapter that drops the connection after completing the delete; a green happy-path test says very little about cleanup safety.

The query and the delete also need different clocks. Use a durable cutoff such as hold_expires_at < pass_started_at - safety_margin, rather than selecting rows whose timestamp equals the scheduler's exact firing time. A paused schedule can miss a trigger, and a retry can start a second pass. Repeating a range-based cleanup is easier to audit than trying to reconstruct one lost instant.

The cost control is equally concrete: cap the candidate page, cap worker concurrency, and pace delete starts against the downstream quota. Concurrency controls how many requests are in flight; a rate limiter controls how frequently requests begin. Treating them as the same setting is how a cleanup pass becomes a burst.

Slow is intentional.

How should Node.js schedule stale upload cleanup with a rate-limited worker queue?

Node.js is a good fit for the orchestration layer, but the algorithm does not depend on its event loop. The scheduler should do little more than create a run and publish a bounded page. The worker should use an explicit concurrency limiter and a shared token bucket or equivalent start gate. If each worker has an independent limiter, four workers can quietly multiply the intended rate.

Here is the critical path in Python so the state transitions stay visible; the same interfaces can be implemented with Node.js timers, an HTTP client, and the queue client already present in the service. The endpoint names are intentionally application-owned. The database transaction claims work, but it never waits for a network quota slot while holding a row lock.

import asyncio
import time


class RateGate:
    def __init__(self, starts_per_second: float):
        self.interval = 1.0 / starts_per_second
        self.next_start = 0.0
        self.lock = asyncio.Lock()

    async def wait_for_slot(self) -> None:
        async with self.lock:
            now = time.monotonic()
            delay = max(0.0, self.next_start - now)
            self.next_start = max(now, self.next_start) + self.interval
        await asyncio.sleep(delay)


async def handle(command, db, storage, gate):
    upload = await db.get_upload(command["tenant_id"], command["upload_id"])
    if upload is None or upload["version"] != command["expected_version"]:
        return "skipped"
    if upload["hold_expires_at"] >= command["cutoff"]:
        return "renewed"

    await gate.wait_for_slot()
    await storage.delete(upload["object_key"], idempotency_key=command["operation_id"])
    await db.mark_deleted_if_version_matches(
        command["tenant_id"], command["upload_id"], command["expected_version"]
    )
    return "deleted"


async def consume(commands, db, storage, starts_per_second, concurrency):
    gate = RateGate(starts_per_second)
    semaphore = asyncio.Semaphore(concurrency)

    async def run(command):
        async with semaphore:
            result = await handle(command, db, storage, gate)
            await acknowledge(command)
            return result

    return await asyncio.gather(*(run(command) for command in commands))

The storage adapter must define what a terminal absence means. Some systems return a successful result for an already absent object; others expose a distinct not-found response. Either behavior can be mapped to the same application outcome, provided that a later version cannot be marked deleted by an earlier command. A transient response should remain retryable with backoff and a bounded attempt policy. Do not acknowledge an exception merely to make the queue look quiet.

The example's acknowledge call is deliberately after the state transition. In production, make that transition durable and record the reason the item qualified, the attempt number, and the final outcome. Scheduler output is a poor audit database. Your mileage may vary on the exact retry delay because the storage provider's quota and retry contract are external inputs; measure them and document the chosen rate.

Which cleanup design fits the retention workload?

There is no universal winner here. A database-owned worker is often the least complicated choice when the candidate rows and their lock state already belong to PostgreSQL. FOR UPDATE SKIP LOCKED can let concurrent claimers avoid waiting on rows another worker has claimed, but it does not make object deletion transactional with the database. That split still needs an idempotency key and a reconciliation pass.

Celery is a reasonable alternative when a team already operates its worker model and wants task retries, routing, and worker lifecycle to remain in that estate. Its introduction documentation describes a distributed task queue rather than a guarantee that an arbitrary storage delete is safe to repeat, so the application still owns the delete contract. A queue can carry work; it cannot invent domain idempotency.

The catch is operational latency. A shared limiter intentionally makes the oldest item wait, so the service needs a backlog-age metric and an alert based on the hold policy, not just a count of failed requests. Track selected, published, started, skipped-after-recheck, deleted, retrying, and permanently rejected outcomes. Include tenant and run identifiers in logs, but avoid putting object names or upload contents into a general log stream if they can contain personal data.

The rejected design is a single nightly loop that reads every stale row and deletes inline. It is attractive because it has fewer moving parts, and it is valid for a demonstrably tiny bounded set with a hard maximum and generous execution headroom. It is not suitable when an import, incident, or migration can create a backlog large enough to turn scheduler runtime into customer-visible latency. In that case, the queue is not decoration; it is the boundary that lets the trigger finish while work continues at a controlled pace.

Operational checks before the first nightly pass

Test the state machine with four cases: a renewed upload after selection, a duplicate command, a process stop after the external delete, and a tenant mismatch. Then test the boring cases that become expensive at scale: a full queue, a quota response, a dead-lettered command, and a schedule that was paused for several days.

Run one dry pass that only records candidate counts and the oldest eligible timestamp. Compare that count with the retention policy before enabling deletion. During the first real pass, keep the page size and start rate conservative, and make the worker's terminal states visible to the operator who owns storage cost.

The decision rule is simple: use a scheduler for timing, a durable handoff for variable volume, and a worker for pacing and final authorization. Keep direct deletion for a proven small workload. If the system cannot state what happens after delete-before-ack, it is not ready for unattended cleanup.

References

• https://docs.celeryq.dev/en/stable/getting-started/introduction.html
• https://www.postgresql.org/docs/current/sql-select.html