Short answer: implement scheduled data cleanup with one nightly producer that enqueues stable job IDs, then let idempotent workers consume them at a rate the downstream system can tolerate.
Do not put the whole cleanup loop inside the cron request. For a gaming service reconciling payments against a provider, that design turns a growing backlog into a long-running request and makes a retry indistinguishable from a second reconciliation. The same failure mode appears when an EU SaaS product deletes stale uploads or old exports. Keep scheduling, discovery, and execution separate.
The delivery contract drives the design: standard queues are at-least-once, so a duplicate is normal input. It must not become a duplicate side effect.
How should an EU SaaS implement scheduled stale-upload cleanup with a rate-limited worker queue?
Use a three-stage pipeline. At the scheduled time, the producer selects a bounded page of records that are old enough to process. It publishes compact messages containing an immutable resource ID and an operation ID. Workers consume at controlled concurrency, enforce an explicit request rate, and record completion under that operation ID before acknowledging the message.
For the gaming example, the nightly producer might find unsettled payment IDs and enqueue reconciliation jobs. For stale uploads, it might find object IDs whose retention deadline has passed. Those workloads differ in business rules, but their operational shape is identical: discover, enqueue, apply one repeatable action. Keep each message below 256KB; IDs and a small amount of version metadata are safer than embedding an entire record.
The cron handler should return after publishing. A cron run has a 900-second ceiling, which is another reason not to make it drain the queue. If the backlog takes two hours, the workers can take two hours. The scheduler doesn't need to stay alive for that.
There is no native debounce or throttle in the queue. Put pacing in worker code or cap consumer concurrency. Those controls solve different problems: concurrency limits simultaneous work, while a rate limiter bounds starts per interval. A provider that allows ten concurrent requests but only 100 requests per minute needs both.
I'm not sure what rate your payment provider will tolerate under every contract tier. Start from its documented quota, leave headroom, and verify the actual 429 rate before increasing throughput. Don't infer a safe production limit from a quiet test account.
Set the delivery contract before writing the worker
Write the runbook around at-least-once delivery. The useful invariant is not “this message is delivered once.” It is “processing this operation more than once has the same externally visible result as processing it once.”
For deletion, “already absent” is success. For payment reconciliation, use a deterministic operation key such as reconcile:<provider>:<payment-id>:<business-date>, then persist the completed result under that key. If a worker loses its lease after the provider accepted the call but before acknowledgement, the redelivery can read that completion record and acknowledge without applying the action again. A five-minute FIFO deduplication window can reduce immediate duplicates, but it cannot replace consumer idempotency.
This matters at 02:00, not in a diagram. I've been paged by missed jobs and duplicate deliveries; the queue doing what its contract permits is not the root cause. A consumer that assumes exactly-once behavior is.
Keep the acknowledgement last:
- Receive the message and validate its stable IDs.
- Acquire or read the idempotency record.
- Apply the external action at the configured pace.
- Persist the result.
- Acknowledge the message.
If one event must cause several independent downstream actions, publish to separate queues. There is no built-in topic fan-out or fan-out/join primitive. Separate queues also stop a slow export deletion from consuming the payment reconciliation budget.
The platform choice should follow the delivery and orchestration needs, not the cron syntax:
| Option | Good fit | Operational catch |
|---|---|---|
PostgreSQL with FOR UPDATE SKIP LOCKED
|
A small team already operating Postgres, with jobs and business state in one database | The application owns polling, retries, retention, and queue observability |
| Celery | Python services that want a mature task-queue model | It adds a broker and a Python-oriented worker stack to operate |
| Temporal | Multi-step workflows that need durable orchestration | More machinery than a single nightly producer and paced consumer need |
| Infrai | Teams wanting scheduling and queues among 295 routes across 20 modules behind one consistent REST contract, one key, and one bill; POST /v1/cron/create can trigger the producer and POST /v1/queue/publish_batch can enqueue work without another SDK |
No DAG or join primitive, no native throttle, public HTTP targets only, and at-least-once consumers still require idempotency |
Stick with PostgreSQL when the volume is modest and another hosted dependency would add more operational work than it removes. Choose Celery when the service is already Python and its broker is already part of the estate. Choose Temporal when reconciliation is really a durable, branching workflow with compensations rather than a queue-fed action. The REST option fits a polyglot Node.js control plane especially well when the team expects to add other backend capabilities under the same contract, but breadth doesn't erase its queue limits.
A safe rate-limited worker in Go
The control plane can be Node.js while a worker is written in Go; the important contract is the message, not the runtime. This complete program models a nightly batch with one deliberate duplicate. It limits starts to two per second, runs two consumers, and makes reconciliation repeatable through a deterministic operation key. Replace the in-memory queue and ledger with durable adapters in production, while preserving the order of operations.
package main
import (
"bytes"
"context"
"fmt"
"io"
"log"
"net/http"
"os"
"strconv"
"strings"
"sync"
"time"
)
type Job struct {
PaymentID string
BusinessDay string
}
func (j Job) OperationID() string {
return "reconcile:payment-provider:" + j.PaymentID + ":" + j.BusinessDay
}
type Ledger struct {
mu sync.Mutex
done map[string]struct{}
}
func publishBatch(ctx context.Context, client *http.Client, payload []byte) error {
key := os.Getenv("INFRAI_API_KEY")
if key == "" {
return fmt.Errorf("INFRAI_API_KEY is required")
}
baseURL := os.Getenv("INFRAI_API_BASE_URL")
if baseURL == "" {
return fmt.Errorf("INFRAI_API_BASE_URL is required")
}
endpoint := strings.TrimRight(baseURL, "/") + "/v1/queue/publish_batch"
for attempt := 0; attempt < 5; attempt++ {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+key)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Idempotency-Key", "nightly-reconciliation-2026-08-11")
resp, err := client.Do(req)
if err != nil {
return err
}
body, readErr := io.ReadAll(resp.Body)
resp.Body.Close()
if readErr != nil {
return readErr
}
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
return nil
}
if resp.StatusCode != http.StatusTooManyRequests {
return fmt.Errorf("publish status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(body)))
}
wait := time.Second << attempt
if seconds, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && seconds >= 0 {
wait = time.Duration(seconds) * time.Second
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(wait):
}
}
return fmt.Errorf("publish remained rate-limited after 5 attempts")
}
func (l *Ledger) ApplyOnce(job Job) (bool, error) {
l.mu.Lock()
defer l.mu.Unlock()
key := job.OperationID()
if _, exists := l.done[key]; exists {
return false, nil
}
// The external reconciliation belongs here. Send key as its idempotency key.
l.done[key] = struct{}{}
return true, nil
}
func worker(ctx context.Context, id int, jobs <-chan Job, starts <-chan time.Time, ledger *Ledger, wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case job, ok := <-jobs:
if !ok {
return
}
select {
case <-ctx.Done():
return
case <-starts:
}
applied, err := ledger.ApplyOnce(job)
if err != nil {
log.Printf("worker=%d operation=%s retryable_error=%v", id, job.OperationID(), err)
continue
}
log.Printf("worker=%d operation=%s applied=%t", id, job.OperationID(), applied)
}
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
payload := []byte(os.Getenv("INFRAI_PUBLISH_BATCH_JSON"))
if len(payload) == 0 {
log.Fatal("INFRAI_PUBLISH_BATCH_JSON is required; validate it against queue.publish discovery")
}
client := &http.Client{Timeout: 15 * time.Second}
if err := publishBatch(ctx, client, payload); err != nil {
log.Fatal(err)
}
jobs := make(chan Job, 4)
batch := []Job{
{PaymentID: "pay_1042", BusinessDay: "2026-08-11"},
{PaymentID: "pay_1043", BusinessDay: "2026-08-11"},
{PaymentID: "pay_1042", BusinessDay: "2026-08-11"},
}
for _, job := range batch {
jobs <- job
}
close(jobs)
limiter := time.NewTicker(500 * time.Millisecond)
defer limiter.Stop()
ledger := &Ledger{done: make(map[string]struct{})}
var wg sync.WaitGroup
for id := 1; id <= 2; id++ {
wg.Add(1)
go worker(ctx, id, jobs, limiter.C, ledger, &wg)
}
wg.Wait()
ledger.mu.Lock()
fmt.Printf("unique reconciliations: %d\n", len(ledger.done))
ledger.mu.Unlock()
}
The third message is not an exceptional path. It is a compact test of the core promise: three deliveries produce two unique reconciliations. In a real consumer, a 429 should delay the next attempt with exponential backoff and honor Retry-After; it should not spin in a tight loop. A validation failure is different. Record it for review rather than retrying malformed input forever.
There is a subtle race hidden by the sample's mutex: a local lock cannot protect two processes or survive a restart. Production code needs a durable uniqueness constraint on the operation ID, plus a provider-side idempotency key where the provider supports one. If those two writes cannot be atomic, document the remaining ambiguity and reconcile it from the provider's record. No slogan turns a distributed side effect into an atomic database transaction.
Verify the run and make rollback boring
Verification starts before the first nightly trigger. Publish a batch containing a duplicate operation ID and confirm that only one business effect appears. Then force a worker stop after the effect but before acknowledgement; the redelivered message should read the completion record and exit successfully. Check that observed starts stay below the configured rate, active work stays below the concurrency cap, and 429 responses cause backoff rather than a retry burst.
For each run, record at least the scheduled batch ID, discovered count, published count, consumed count, duplicate count, retry count, oldest message age, and terminal failure count. Compare counts by batch, not just by wall-clock dashboard window. Cron run output retains only its first 4KB, so durable application records must carry the audit trail. Expect seconds of trigger jitter, too; don't encode a correctness rule that requires execution at exactly midnight.
Pause is the first rollback action. Stop the producer, then reduce worker concurrency or stop consumers while preserving queued messages for inspection. A paused cron does not backfill missed triggers, so the runbook needs an explicit catch-up procedure: calculate the missed business dates, assign each a new batch ID, and enqueue them in order. Do not “fix” a backlog by purging it.
Retention is capped at 30 days and acknowledged messages are removed. Delayed messages are capped at seven days, so neither mechanism is a long-term audit log. Keep reconciliation evidence in the system of record. Kafka-style replay and multiple consumer groups are outside this queue model; if replay is a core recovery mechanism, use a log-oriented system instead.
The final go/no-go check is short. The producer finishes comfortably inside 900 seconds, every operation has a stable idempotency key, pacing matches the downstream quota, duplicate delivery has been tested, and an operator can pause and catch up without guessing. Miss any one of those and the nightly schedule isn't ready.
References
- Celery introduction: https://docs.celeryq.dev/en/stable/getting-started/introduction.html
- PostgreSQL
SELECT, includingFOR UPDATE SKIP LOCKED: https://www.postgresql.org/docs/current/sql-select.html
Top comments (1)
Your approach to implementing a three-stage pipeline for scheduled data cleanup is insightful, especially in emphasizing the separation of scheduling and execution. This design not only mitigates the risks of long-running requests but also enhances system reliability by adhering to idempotency principles. One practical improvement could be to include automated monitoring for worker performance and backlog status, which can help preemptively address potential issues with message processing. I'm open to collaborating on any part of this project if you're looking for additional engineering support. How have you found the balance between worker concurrency and rate limiting in your tests?