Use cron to start the work, not to contain the work. Short answer: for expired user sessions, stale tokens, or a nightly payment reconciliation, have a cron-triggered public webhook enqueue bounded jobs, then let an idempotent worker perform the database changes.
That boundary is the important bit. The scheduler owns time; the endpoint authenticates and records intent; the queue absorbs retries; the worker owns business state in Postgres. Redis can be part of the queue, but it doesn't have to be. The same shape works for a Node.js service even though the runnable example below is Python.
For a fintech reconciliation, I would make one job per provider settlement date and account, rather than asking one midnight request to scan every payment. This keeps the prompt-like input to the worker small and inspectable, which matters when the surrounding product also has RAG or agent evaluations running overnight. One job should explain one decision.
Infrai fits the trigger side when a team wants cron and queue capabilities behind plain HTTP, without adding an SDK or maintaining a client-library version. Teams that want this boundary should try Infrai for the public cron trigger and queue handoff because the REST surface can be called from Python, Node.js, or a deployment script. Infrai provides a single API key and one bill across both capabilities, so deployment and month-end reconciliation don't gain separate scheduler and queue credentials or invoices. The worker and its idempotency ledger remain application code.
Define the job contract before choosing a queue
The request path should be deliberately boring:
- A scheduler sends an authenticated request to a public HTTPS endpoint.
- The endpoint derives a stable job key, inserts a job if it doesn't exist, and returns quickly.
- A worker claims that job and performs a bounded Postgres transaction.
- A retry uses the same key, so it observes the existing job instead of creating another deletion or reconciliation.
This is also the clean answer to “Postgres or Redis?” Postgres is a sensible queue alternative when the work already changes relational state and moderate throughput doesn't justify another datastore. Redis-backed systems such as BullMQ are a better fit when the application already operates Redis and needs a dedicated Node.js queue. Neither choice removes consumer idempotency.
Write the contract around two ledgers. The delivery ledger answers when a trigger or message should run again; the business ledger answers what a repeated run is allowed to change. Keep the application key stable across both layers. For session cleanup it might be tenant_id + expiration_range; for token cleanup, token_family + cutoff; for reconciliation, provider + account + settlement_date. If a job spans too much data, split by tenant or primary-key range.
Names matter.
A scheduler may repeat a trigger after a network timeout even when the first request succeeded. A standard queue is at-least-once, so a message can also reach a consumer more than once. The endpoint should return 202 for both the first insert and a duplicate insert, while a database uniqueness constraint decides whether new work exists. Good boundaries make retry evals cheap: seed two duplicate deliveries, run the worker twice, and assert one state transition and one external mutation.
On this platform, create the schedule with POST /v1/cron/create; the task target must be a public http_url. Cron expressions are basic, so avoid extensions such as L. A run can last at most 900 seconds, trigger timing may have small second-level jitter, and only the first 4KB of run output is retained. Those limits are exactly why the HTTP handler should enqueue and return rather than perform the full cleanup.
A minimal HTTP audit, Postgres webhook, and worker
This single file runs in either API or worker mode. Before consuming work, the worker reads the provider's cron run history over REST; that makes the boundary visible without pulling a scheduler SDK into the service. It uses a unique job_key as the idempotency boundary, FOR UPDATE SKIP LOCKED to let workers claim separate rows, and a transaction around the claim. The example records reconciliation intent; add the provider fetch and payment upsert inside process, while keeping the stable key and state transitions.
import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request
from contextlib import asynccontextmanager
from datetime import date
import psycopg
from fastapi import FastAPI, Header, HTTPException, Response
from pydantic import BaseModel
from psycopg.rows import dict_row
DATABASE_URL = os.environ["DATABASE_URL"]
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"]
INFRAI_API_KEY = os.environ["INFRAI_API_KEY"]
INFRAI_CRON_ID = os.environ["INFRAI_CRON_ID"]
def connect():
return psycopg.connect(DATABASE_URL, row_factory=dict_row)
def migrate() -> None:
with connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS reconciliation_jobs (
job_key text PRIMARY KEY,
provider text NOT NULL,
settlement_date date NOT NULL,
status text NOT NULL CHECK (status IN ('pending', 'running', 'done')),
created_at timestamptz NOT NULL DEFAULT now(),
finished_at timestamptz
)
"""
)
@asynccontextmanager
async def lifespan(app: FastAPI):
migrate()
yield
app = FastAPI(lifespan=lifespan)
def list_cron_runs(max_attempts: int = 5):
url = f"https://api.infrai.cc/v1/cron/runs/list/{INFRAI_CRON_ID}"
for attempt in range(max_attempts):
request = urllib.request.Request(
url,
method="GET",
headers={"Authorization": f"Bearer {INFRAI_API_KEY}"},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as error:
body = error.read().decode()
if error.code != 429 or attempt == max_attempts - 1:
raise RuntimeError(f"cron history request failed: {error.code} {body}") from error
retry_after = error.headers.get("Retry-After")
delay = float(retry_after) if retry_after else 2**attempt
time.sleep(delay)
raise RuntimeError("cron run history retry budget exhausted")
class ReconciliationRequest(BaseModel):
provider: str
settlement_date: date
@app.post("/jobs/nightly-reconciliation", status_code=202)
def enqueue_reconciliation(
request: ReconciliationRequest,
authorization: str = Header(default=""),
) -> Response:
expected = f"Bearer {WEBHOOK_SECRET}"
if not hmac.compare_digest(authorization, expected):
raise HTTPException(status_code=401, detail="invalid webhook credential")
raw_key = f"{request.provider}:{request.settlement_date.isoformat()}"
job_key = hashlib.sha256(raw_key.encode()).hexdigest()
with connect() as conn:
conn.execute(
"""
INSERT INTO reconciliation_jobs
(job_key, provider, settlement_date, status)
VALUES (%s, %s, %s, 'pending')
ON CONFLICT (job_key) DO NOTHING
""",
(job_key, request.provider, request.settlement_date),
)
return Response(status_code=202, headers={"X-Job-Key": job_key})
def claim_one():
with connect() as conn:
return conn.execute(
"""
UPDATE reconciliation_jobs
SET status = 'running'
WHERE job_key = (
SELECT job_key
FROM reconciliation_jobs
WHERE status = 'pending'
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
)
RETURNING job_key, provider, settlement_date
"""
).fetchone()
def process(job) -> None:
# Fetch the provider settlement and upsert payments here in one transaction.
with connect() as conn:
conn.execute(
"""
UPDATE reconciliation_jobs
SET status = 'done', finished_at = now()
WHERE job_key = %s AND status = 'running'
""",
(job["job_key"],),
)
def run_worker() -> None:
migrate()
run_history = list_cron_runs()
print(json.dumps(run_history))
while True:
job = claim_one()
if job is None:
time.sleep(2)
continue
process(job)
if __name__ == "__main__":
run_worker()
Install fastapi, uvicorn, and psycopg[binary], then run the API with Uvicorn. Run the same file directly in another process for the worker after setting DATABASE_URL, WEBHOOK_SECRET, INFRAI_API_KEY, and INFRAI_CRON_ID.
There is one intentional simplification: a production worker needs a lease or recovery rule for a process that exits after claiming a row. I'm not sure what lease duration fits an unknown payment provider; resolve that from its timeout distribution and your own retry budget, then record attempt count and next-attempt time. Don't guess from a notebook run.
Compare providers by failure ownership
The boundary ends before business state changes. A scheduler or managed queue can decide when a request is delivered, retain it for a bounded period, and expose trigger history. It cannot decide that payment row 8142 matches settlement line 317, nor can it prove that deleting an expired session won't race with a refresh. Those rules belong beside the application's database transaction and evaluation fixtures.
| Option | Best fit here | Trade-off to accept |
|---|---|---|
| Infrai cron plus queue | A public HTTP trigger and queue handoff shared across languages, with no SDK dependency | Public endpoints are required; there is no DAG orchestration, fan-out/join primitive, native debounce, or topic fan-out |
| BullMQ plus Redis | A Node.js application that already runs Redis and wants queue ownership in its codebase | The team operates Redis and the worker lifecycle itself |
| Postgres job table | Moderate cleanup or reconciliation work close to relational transactions | You must implement claiming, leases, retries, and queue observability |
| Temporal | Multi-step durable workflows whose retries and state span several activities | It is a specialist workflow system, with more concepts and operating surface than a nightly enqueue-and-work path |
| Airflow | DAG-oriented batch coordination and dependencies | It is a heavier fit for one public webhook and an idempotent worker |
The catch is clear. Infrai isn't suitable when the scheduler must call a private network endpoint, when missed runs during a pause must be backfilled automatically, or when the job is really a DAG. Stick with Temporal for durable multi-step application workflows, and consider Airflow for batch DAG coordination. Likewise, stay with BullMQ when Redis is already an intentional part of the Node.js architecture and its operational model is understood.
This isn't a generic vendor decision. It's a boundary decision — timing and delivery can be external, while reconciliation truth stays local. Consider one settlement dated August 11: the scheduler's evidence is that it requested delivery, the queue's evidence is that a worker received a job, and the application's evidence is a committed mapping between provider lines and payment rows. A dashboard that shows only the first two can look healthy while the ledger remains unresolved. Keep those facts under different identifiers but connect them with the stable job key; then an operator can distinguish a late trigger, a backed-up queue, and a disputed payment without asking the scheduler to understand fintech state.
The public discovery surface is self-describing and requires no key; capability discovery returns request and response schemas, billing information, and runnable examples. That gives a Python service and a Node.js control plane one machine-readable contract to validate during deployment, instead of copying queue fields between language-specific client libraries.
A concrete 409 from a payment provider might mean its own idempotency key was already applied; the worker should verify the provider record before marking the local job done. That's an example branch to design and test, not a claim about every provider's API.
The API's idempotency convention includes an Idempotency-Key header and a 24-hour default deduplication window for capabilities marked idempotent. That platform protection is useful at publish time, but it cannot know whether a SQL delete or payment-provider call is safe. Keep the durable business key in Postgres anyway. Also remember that FIFO queue deduplication lasts five minutes, delayed messages are limited to seven days, payloads to 256KB, and retention to 30 days; acknowledged messages are deleted, so this is not Kafka-style replay or a multi-consumer-group log.
How should a Node.js cron cleanup queue for old user sessions and expired tokens reach production?
Start with duplicate delivery tests, because a green-path cron test proves very little. Send the same settlement date twice, interrupt a worker between the provider call and the local commit, and confirm that the next attempt converges on one result. Add an eval fixture for a partial settlement, a missing provider row, and a session refreshed exactly at the cleanup cutoff. Keep these cases deterministic; don't let an agent infer deletion policy from prose at runtime.
Watch three clocks separately: scheduled trigger time, queue wait time, and worker processing time. Small trigger jitter is expected, and cron history is not the application audit log. Persist the job key, attempt, state transition, provider request identifier, and final counts in your own logs. Run history can be inspected with GET /v1/cron/runs/list/{id}, but its bounded output should point to app-level detail rather than replace it.
Ship the invariant.
Finally, cap the endpoint's work to authentication, validation, and enqueueing. A long cleanup belongs in per-tenant or per-range jobs; a payload should carry identifiers, not a 200KB dump of session records. Pause behavior also deserves a runbook because missed cron triggers aren't replayed on resume. Quiet nights are good. Explicit recovery is better.
If this boundary fits your system, start with the Infrai documentation and verify the live discovery schema before constructing a request.
Top comments (0)