Guide 03

Job lifecycle and schema

Audience: Backend engineers implementing user-visible async work. Rule: The broker is transport. Product state lives in your database.

TL;DR

Insert job row → publish command → return 202 + job_id
Worker: claim → run → succeed / fail / retry
Client polls GET /jobs/{id} (or webhook)

---

1. State machine

text

pending ──► running ──► succeeded
 │
 ├──► failed (terminal after max attempts)
 ├──► cancelled (optional)
 └──► pending/retrying (scheduled retry; keep attempts++)

Keep statuses few. Map framework retries onto attempts + timestamps rather than inventing ten enums.

Terminal vs non-terminal

Status	Terminal?	Client meaning
pending	No	Queued or waiting for retry
running	No	Worker claimed it
succeeded	Yes	Done; result ready
failed	Yes	Give up; show error
cancelled	Yes	User or system aborted

2. Schema

sql

-- PSEUDOCODE DDL
CREATE TABLE jobs (
 id UUID PRIMARY KEY,
 type TEXT NOT NULL, -- send_email | export_csv | ...
 status TEXT NOT NULL, -- pending|running|succeeded|failed|cancelled
 idempotency_key TEXT, -- UNIQUE when present
 payload_ref TEXT, -- small JSON or object-storage key
 result_ref TEXT, -- optional output location
 entity_id TEXT, -- user_id / order_id for listing
 attempts INT NOT NULL DEFAULT 0,
 max_attempts INT NOT NULL DEFAULT 5,
 last_error TEXT,
 created_at TIMESTAMPTZ NOT NULL,
 updated_at TIMESTAMPTZ NOT NULL,
 started_at TIMESTAMPTZ,
 finished_at TIMESTAMPTZ
);

CREATE UNIQUE INDEX jobs_idempotency_uidx
 ON jobs (idempotency_key) WHERE idempotency_key IS NOT NULL;
CREATE INDEX jobs_status_created_idx ON jobs (status, created_at);
CREATE INDEX jobs_entity_idx ON jobs (entity_id, created_at DESC);

3. Enqueue contract

http

POST /jobs/exports
Idempotency-Key: client-generated-or-body
→ 202 Accepted
{ "job_id": "...", "status": "pending" }

python

# PSEUDOCODE
async def enqueue_export(cmd, db, bus):
 existing = await db.find_by_idempotency(cmd.idempotency_key)
 if existing:
 return existing # no double publish side effects

 async with db.transaction():
 job = await db.insert_job(type="export_csv", status="pending", ...)
 await bus.publish("jobs.heavy", {"job_id": str(job.id)})
 # or write outbox row instead of direct publish
 return job

Failure modes on enqueue

Failure	Response	Recovery
Validation	400	Client fixes input
Duplicate key	202 with existing job	None
Broker down, no outbox	503	Client retry with same key
Broker down, with outbox	202	Relay publishes later

4. Worker contract

python

# PSEUDOCODE
async def handle_export(message):
 job_id = message["job_id"]
 job = await db.get_job(job_id)
 if job.status in ("succeeded", "cancelled"):
 return # idempotent ack

 claimed = await db.try_mark_running(job_id, expected=("pending",))
 if not claimed:
 return

 try:
 result = await do_export(job)
 await db.mark_succeeded(job_id, result_ref=result)
 except TransientError as e:
 await db.bump_attempt(job_id, error=str(e))
 raise # framework retries
 except PermanentError as e:
 await db.mark_failed(job_id, error=str(e))
 # do not retry

5. Idempotency keys

Client sends key for “create export” / “send receipt”
Server stores unique key on job
Replays return the same job_id

Worker-side: natural keys (invoice_id + action) or inbox table : see 04 Outbox/inbox.

6. Payload design

Do	Don't
Pass IDs and small refs	Pass multi-MB blobs on the bus
Version message schema	Dump ORM instances
Keep secrets out of messages	Log full PII payloads

Large inputs/outputs → object storage; job row holds keys.

7. Observability

Log and metric:

enqueue rate, success rate, attempts, runtime
time-to-start, time-to-complete SLOs
backlog (queue depth)

Status for users always from DB, not Flower.

8. API sketch