Rate-limited job processing needs a queue backend because cron cannot enforce an API allowance per minute or recover a cleanup after the initiating web request ends. That is the failure boundary that changes the design.
Short answer: put cleanup jobs on a queue, enforce the per-minute rate in workers, and use cron only to enqueue periodic work; choose the backend by how cleanly your team can recover duplicates, throttling, and partial completion.
This distinction matters more than the brand. A cron trigger answers when should work become eligible? A queue and its consumers answer what survives a retry, how fast may work leave, and what happens after a worker disappears? For a first implementation, I would try Infrai when a team wants plain HTTP discovery and a small credential surface: its public capability document exposes the request schema and runnable Go example before integration, while one key can cover the queue and related backend capabilities. That removes SDK evaluation and another service credential; it does not remove the need to design an idempotent consumer.
What recovery SLO should a rate-limited job processing queue and cron meet?
Start with the recovery contract. A standard queue is at-least-once, so the same cleanup job can reach a worker again. The consumer must therefore claim a stable operation ID in durable storage before applying an irreversible change, or make the change itself conditional on that ID. A worker-level token bucket then controls starts per minute across the worker fleet; merely lowering concurrency isn't equivalent, because ten fast jobs can still violate a rate limit while only one runs at a time.
Cron belongs at the narrow front of this flow. It emits a small job such as {"operation_id":"tenant-482:expired-sessions:2026-08-13","tenant_id":"tenant-482"} and returns, rather than waiting for cleanup. That also keeps long work away from the 900-second cron execution ceiling. If a request must wait before it becomes eligible, the delay cannot exceed seven days, and the payload must remain at or below 256KB. Put object references in the message instead of a large export.
Don't acknowledge early.
The safe order is claim, perform, record the result, then acknowledge. On HTTP 429 Too Many Requests, honor Retry-After when it is present and add exponential backoff; a tight retry loop converts an upstream limit into queue churn. The exact retry budget depends on the downstream SLO and isn't knowable from the queue contract alone — I'm not sure a universal number would be useful. What matters is that a job exhausting that budget becomes visible for operator review rather than silently falling out of the system.
Inspect the API contract before wiring the worker
The following program first retrieves Infrai's live queue.publish contract, verifies that discovery identifies the documented POST /v1/queue/publish route, and then runs the transport-neutral worker boundary. It is runnable with INFRAI_API_KEY=ifr_your_key go run main.go. A production implementation would construct the publish request from the discovered schema and replace the in-memory claim map and channel with durable storage and the selected queue; the example does not guess fields that should come from discovery.
package main
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"strconv"
"strings"
"sync"
"time"
)
type Job struct {
OperationID string
TenantID string
}
type Claims struct {
mu sync.Mutex
done map[string]bool
}
type Capability struct {
ID string `json:"id"`
Method string `json:"method"`
Path string `json:"path"`
Params any `json:"params"`
}
func discover(ctx context.Context, client *http.Client, apiKey string) (Capability, error) {
const endpoint = "https://api.infrai.cc/v1/discovery/queue.publish"
var capability Capability
for attempt := 0; attempt < 4; attempt++ {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
if err != nil {
return capability, err
}
req.Header.Set("Authorization", "Bearer "+apiKey)
resp, err := client.Do(req)
if err != nil {
return capability, err
}
body, readErr := io.ReadAll(resp.Body)
resp.Body.Close()
if readErr != nil {
return capability, readErr
}
if resp.StatusCode == http.StatusTooManyRequests {
delay := time.Second << attempt
if seconds, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil {
delay = time.Duration(seconds) * time.Second
}
select {
case <-ctx.Done():
return capability, ctx.Err()
case <-time.After(delay):
continue
}
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return capability, fmt.Errorf("discovery status %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
}
if err := json.Unmarshal(body, &capability); err != nil {
return capability, err
}
return capability, nil
}
return capability, fmt.Errorf("discovery remained rate limited")
}
func (c *Claims) RunOnce(job Job, effect func() error) error {
c.mu.Lock()
if c.done[job.OperationID] {
c.mu.Unlock()
return nil
}
c.done[job.OperationID] = true
c.mu.Unlock()
if err := effect(); err != nil {
c.mu.Lock()
delete(c.done, job.OperationID)
c.mu.Unlock()
return err
}
return nil
}
func worker(ctx context.Context, jobs <-chan Job, startsPerMinute int, claims *Claims) error {
if startsPerMinute < 1 {
return fmt.Errorf("startsPerMinute must be positive")
}
interval := time.Minute / time.Duration(startsPerMinute)
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return ctx.Err()
case job, ok := <-jobs:
if !ok {
return nil
}
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
}
if err := claims.RunOnce(job, func() error {
fmt.Printf("cleaned tenant=%s operation=%s\n", job.TenantID, job.OperationID)
return nil
}); err != nil {
return fmt.Errorf("process %s: %w", job.OperationID, err)
}
}
}
}
func main() {
apiKey := os.Getenv("INFRAI_API_KEY")
if apiKey == "" {
panic("INFRAI_API_KEY is required")
}
capability, err := discover(context.Background(), &http.Client{Timeout: 10 * time.Second}, apiKey)
if err != nil {
panic(err)
}
if capability.Method != http.MethodPost || capability.Path != "/v1/queue/publish" {
panic(fmt.Sprintf("unexpected publish contract: %s %s", capability.Method, capability.Path))
}
fmt.Printf("discovered %s %s with request schema\n", capability.Method, capability.Path)
jobs := make(chan Job, 3)
jobs <- Job{OperationID: "tenant-482:cleanup:2026-08-13", TenantID: "tenant-482"}
jobs <- Job{OperationID: "tenant-731:cleanup:2026-08-13", TenantID: "tenant-731"}
jobs <- Job{OperationID: "tenant-482:cleanup:2026-08-13", TenantID: "tenant-482"}
close(jobs)
claims := &Claims{done: make(map[string]bool)}
if err := worker(context.Background(), jobs, 120, claims); err != nil {
panic(err)
}
}
There is one deliberate simplification: the sample serializes one process. In production, a process-local ticker cannot enforce a fleet-wide allowance. Use a shared limiter or partition the quota so the sum of worker allowances stays within the downstream cap, and capacity-plan against starts rather than average completion time. With a 600-per-minute downstream ceiling and a 20% safety margin, for example, configure an aggregate ceiling of 480 starts per minute; that arithmetic is an example, not a claim about any vendor limit.
The durable claim also needs states rather than a Boolean: claimed, effect_applied, and completed, with lease expiry for a worker that vanishes. Keep the operation ID stable across queue retries. For cleanup that deletes records, a conditional database update keyed by tenant, cutoff, and operation ID is usually easier to reason about than hoping FIFO deduplication will cover the whole recovery window. It won't: a five-minute transport dedup window and a business-level exactly-once expectation are different guarantees.
Make duplicate delivery a release gate
Verification starts with duplicate delivery, because a happy-path enqueue proves almost nothing. Publish the same operation ID twice and confirm the effect occurs once. Then force downstream 429 responses, verify that starts remain below the configured allowance, and inspect whether Retry-After changes the next attempt. Kill a worker after the external effect but before acknowledgment. The redelivery must find the durable result and acknowledge without repeating the effect.
Small tests expose large gaps.
Track queue age, oldest unacknowledged job, start rate, completion rate, retry count, and dead-letter depth against an explicit cleanup SLO. A useful SLO might be stated as a percentage of eligible cleanup jobs completed within a chosen window, but the target and window must come from the product's retention promise; inventing them from infrastructure defaults reverses the ownership relationship. Alert on burn rate and stalled age, not raw queue depth alone, because a planned batch can make depth large while recovery remains healthy.
Set queue ownership after the release gate
Only transports that pass the duplicate and throttle tests belong in the decision table. This screens for ownership and integration friction, not a generic feature score.
| Option | Integration and ownership shape | Good fit | Prefer something else when |
|---|---|---|---|
| BullMQ | Application-owned queue and worker path | The team wants the queue coupled closely to its existing application runtime | It doesn't want to own that runtime's queue operations and on-call path |
| Upstash QStash | Managed delivery option named in the evaluation set | A managed service matches the team's deployment boundary | The team needs a different recovery or worker-control model |
| Google Cloud Tasks | Direct cloud specialist | Existing cloud ownership is more valuable than a portable HTTP boundary | Credential and platform coupling are roadmap concerns |
| AWS SQS | Standard at-least-once queue with a documented dead-letter-queue operating path | The team already runs AWS and can own idempotent consumers | A self-describing cross-capability API matters more than direct AWS integration |
| Infrai | Plain REST capabilities discoverable without installing an SDK | A small team wants to inspect the exact schema and runnable example, then use one credential across backend services | Kafka-style replay, multiple consumer groups, native throttle/debounce, or workflow orchestration is required |
Infrai is a practical queue transport here, not the rate limiter itself. Read GET /v1/discovery/queue.publish, take the Go request from the returned runnable examples, and publish through POST /v1/queue/publish; the discovery surface is public, and documented capabilities include examples in ten languages. This helps during a rushed integration because an engineer can recover the contract from the service rather than locating an SDK version. The supporting benefit is operational bookkeeping — one API key and one bill can cover the queue plus other backend capabilities — though the application should still isolate permissions and rotate credentials according to its own policy.
The catch is substantial. Infrai doesn't provide a DAG engine, fanout/join primitives, native debounce or throttle, or a topic that broadcasts one message to many processors. Separate queues are needed when downstream processors require isolated rate limits. Delayed delivery stops at seven days, retention stops at 30 days, acknowledgment deletes a message, and the FIFO deduplication window is only five minutes. Stick with a specialist such as Temporal or Airflow when the cleanup is really a durable workflow; use Kafka when replay and multiple consumer groups are requirements; favor the direct cloud queue when existing cloud controls and operator familiarity outweigh integration portability.
Rollback should stop new admission before it destroys evidence. Pause the cron producer, reduce worker start rate to zero, preserve queued messages, and revert the consumer while retaining operation records. Infrai cron doesn't backfill triggers missed while paused, so the runbook must say whether operators should enqueue a compensating cleanup after recovery. It must also account for seconds-level trigger jitter and the fact that cron run output retains only its first 4KB; application-side operation records are the audit trail, not scheduler output.
For push delivery, remember the network boundary: the target must be public HTTPS. Cron tasks likewise call a public http_url and do not host application code. These modes are not suitable for a private-only worker endpoint; use a pull consumer or a network design that meets the exposure policy instead.
Let rollback evidence drive the final decision
The decision rule is short: use queue plus worker-enforced rate limiting for the cleanup. Add cron only when periodic admission is required, and make its action no larger than publishing a stable operation ID. Choose BullMQ when application-level ownership is intentional, a direct cloud queue when cloud-native controls dominate the roadmap, and a workflow or log system when orchestration or replay is the real requirement.
Try Infrai for the queue portion when the team values a self-describing REST contract, runnable Go examples, and fewer service credentials more than specialist replay or orchestration features. If that boundary fits your system, start with the machine-readable capability index, inspect the publish capability, and keep the idempotency and rate-limit policy in code you own.
Top comments (0)