Skip to content

CONTRACT.md

Latest commit

 

History

History
299 lines (263 loc) · 16 KB

CONTRACT.md

File metadata and controls

299 lines (263 loc) · 16 KB
Raw

DTQ Contract v2 (normative for all subsystems)

Supersedes v1. Every package, app, and test must follow this exactly. Read this file first, then build.

1. Domain model

class TaskStatus(str, Enum):
    PENDING = "pending"              # accepted, not yet in a stream (delayed)
    QUEUED = "queued"               # entry exists in a dtq:stream:<queue>:p<N> stream
    RUNNING = "running"            # claimed by a worker, handler executing
    SUCCEEDED = "succeeded"        # terminal: handler returned a result
    RETRYING = "retrying"          # terminal attempt recorded, sits in retry schedule
    FAILED = "failed"              # terminal: attempts exhausted, DLQ disabled
    DEAD_LETTERED = "dead_lettered"# terminal: attempts exhausted, moved to DLQ stream
    CANCELLED = "cancelled"        # terminal: cancelled before completion
    EXPIRED = "expired"            # terminal: passed metadata.ttl_seconds before running

Valid transitions (enforce in code, test them): PENDING -> QUEUED | CANCELLED | EXPIRED QUEUED -> RUNNING | CANCELLED | EXPIRED RUNNING -> SUCCEEDED | RETRYING | FAILED | DEAD_LETTERED | CANCELLED RETRYING -> QUEUED | CANCELLED Any non-terminal -> CANCELLED. Terminal states have no outgoing transitions.

Task fields:

  • task_id: str, uuid4 hex, generated by the API at ingest. Never sequential.
  • idempotency_key: str | None
  • task_type: str, must be registered in the handler registry
  • payload: dict, JSON object. Max DTQ_MAX_PAYLOAD_BYTES (default 262144). Never executable code.
  • created_at: datetime, UTC, timezone-aware
  • available_at: datetime, UTC. Eligibility time (delay_seconds at ingest, or next retry time). Delayed tasks wait in the retry schedule, not the stream.
  • attempt: int. The attempt number being executed, starting at 1. Incremented per execution start. Crash redelivery counts as an execution.
  • max_attempts: int, 1..DTQ_MAX_ATTEMPTS (default cap 10), default 5.
  • priority: int, 0..9, default 5. STRICT priority via separate streams (section 2). Documented as strict, not faked.
  • timeout_ms: int, default DTQ_TASK_TIMEOUT_MS (30000), max 3600000.
  • status: TaskStatus
  • queue: str, default "default". Validated: ^[a-z0-9][a-z0-9_-]{0,63}$
  • metadata: dict, JSON object, max 4KB serialized. Recognized: "ttl_seconds".

Execution metadata recorded on the task hash per attempt: started_at, finished_at, duration_ms, worker_id, error_class, error_message (sanitized, max 500 chars), retryable (bool), retry_delay_ms.

2. Redis schema (exact key names; document all in docs)

  • dtq:stream:<queue>:p<0-9> : task streams, one per queue per priority band. Consumer group workers on each. Strict priority: workers drain p9 -> p0.
  • dtq:stream:dead-letter : DLQ stream. Bounded via MAXLEN ~10000 on XADD.
  • dtq:retry:schedule : sorted set, member = task_id, score = due epoch ms. Holds retries AND delayed (not-yet-available) tasks.
  • dtq:retry:claimed : sorted set, member = task_id, score = claim epoch ms. Scheduler claims move here; reaper returns stale claims (>60s) to schedule.
  • dtq:task:<task_id> : hash with all task fields + execution metadata. EXPIRE DTQ_TASK_RETENTION_S (default 604800 = 7 days), refreshed on updates.
  • dtq:idempotency:<queue>:<key> : JSON {task_id, state, updated_at}, state in {processing, completed, failed}. SET NX, EX DTQ_IDEM_TTL_S (86400).
  • dtq:worker:<worker_id> : hash {worker_id, hostname, started_at, last_heartbeat, active_tasks, concurrency, tasks_processed, tasks_failed, tasks_retried, status}. EXPIRE refreshed by heartbeat (3x interval).
  • dtq:workers : set of worker ids (best effort; liveness from heartbeat age).
  • dtq:queue:<queue>:paused : "1" when paused.
  • dtq:cancelled : set of task_ids cancelled while queued.
  • dtq:tasks:recent : zset task_id by created epoch ms, trimmed to 10000.
  • dtq:stream:audit : stream of admin actions, MAXLEN ~10000.
  • dtq:lock:scheduler : SET NX PX 10000, scheduler leader lock.

Stream entry fields (XADD to task streams): task_id, queue, task_type, payload (JSON), attempt, max_attempts, priority, timeout_ms, idempotency_key ("" if none). DLQ entry fields: same plus failed_at_ms, attempts_made, last_error, last_error_class, retryable, worker_id, original_queue.

TTLs: every key above has a TTL or a bounded trim strategy. No unbounded state.

3. ACK order and failure windows (document in delivery-guarantees.md)

Default order: read -> validate -> execute -> record completion/idempotency state -> ACK. Never ACK right after receiving.

  • Crash before execute: entry stays pending, redelivered via XAUTOCLAIM.
  • Crash during execute: same, redelivered; attempt increments.
  • Crash after execute but before hash update: redelivered, handler re-runs (side effects may repeat; handlers must be idempotent).
  • Crash after hash update + event publish but before XACK: redelivered; worker sees terminal status in hash and ACKs without re-executing.
  • Crash after XACK: done, no redelivery. ACK happens only after: task hash updated to post-execution status, lifecycle event published, follow-up write done (schedule ZADD or DLQ XADD).

4. Retry policy

  • Error classification: retryable = TimeoutError/asyncio.TimeoutError, redis connection errors, handler-raised dtq_tasks.RetryableError. Permanent = dtq_tasks.PermanentError, Pydantic validation errors, unknown task_type, payload decode errors, error_message over limit. Never retry everything; infinite loops are forbidden.
  • On failure with attempt < max_attempts and error retryable: capped = min(DTQ_RETRY_MAX_DELAY, DTQ_RETRY_BASE_DELAY * 2 ** (attempt - 1)) actual = capped * (1 + random.uniform(-DTQ_RETRY_JITTER, DTQ_RETRY_JITTER)) actual = max(0, actual). Defaults: base 1.0s, max 60.0s, jitter 0.2. status=RETRYING, ZADD dtq:retry:schedule (due = now + actual).
  • Permanent error, or attempt >= max_attempts: DLQ enabled -> XADD dtq:stream:dead-letter, status=DEAD_LETTERED; else status=FAILED.
  • Unknown task_type: permanent, one attempt, then DLQ/FAILED with last_error="unknown_task_type:".
  • Scheduler: Lua script atomically moves due members schedule -> claimed (ZRANGEBYSCORE + ZADD + ZREM in one script). Leader lock dtq:lock:scheduler (SET NX PX 10000, refreshed). Scheduler XADDs each claimed task to its priority stream (status QUEUED), then ZREMs from claimed. Claim reaper: claims older than 60s move back. Duplicate enqueue is absorbed: worker skips an entry when the task hash shows a newer-or-equal attempt already QUEUED/RUNNING, or a terminal status (then it just ACKs). Document: duplicates become absorbed attempts, never new logical tasks.

5. Idempotency

  • Header Idempotency-Key wins over body idempotency_key.
  • Scope: (queue, key). SET NX dtq:idempotency::.
  • Key absent -> create record {task_id, state: processing}, ingest task.
  • Key present, state=processing -> HTTP 409, code IDEMPOTENCY_IN_PROGRESS.
  • Key present, state=completed|failed -> HTTP 200, return stored task, "duplicate": true. Never create a second logical task.
  • Worker sets state=completed|failed when the task reaches terminal status. Crash window: side effect done but state not yet recorded -> redelivery may re-execute. Guarantee: duplicate submissions map to one logical task; the system cannot make arbitrary side effects exactly-once. Document this.

6. Worker behavior

  • worker_id = "worker-" + 8 hex chars (uuid4). Never hostname alone.
  • Loops (each with jitter, all cancellable):
    1. reaper: XAUTOCLAIM group workers on each priority stream p9->p0, min_idle_ms = DTQ_VISIBILITY_TIMEOUT_S*1000 (default 30s). Reclaimed entries processed like fresh ones (attempt from entry).
    2. heartbeat + touch: HSET dtq:worker:, EXPIRE 3x interval; XCLAIM in-flight entries to self with IDLE 0 to prevent false reclamation of long-running tasks. Document: heartbeat_interval < visibility_timeout, and task timeouts should stay below visibility timeout.
    3. scheduler (only with leader lock): claim due retries/delayed (Lua), XADD to priority streams, ZREM claimed; requeue stale claims.
    4. poll: if queue paused, sleep. Else capacity = concurrency - in_flight; for p in 9..0: XREADGROUP COUNT min(capacity, 32) BLOCK 500ms on dtq:stream::p

      ">" ; stop when capacity exhausted.

  • Before executing a claimed entry:
    • terminal status in hash -> XACK, skip.
    • task_id in dtq:cancelled or hash status CANCELLED -> XACK, event, skip.
    • TTL expired (metadata.ttl_seconds) -> EXPIRED, XACK, event, skip.
    • duplicate check (section 4) -> XACK, event task.duplicate_absorbed, skip.
  • Execute: semaphore-bounded; asyncio.wait_for(asyncio.timeout) with timeout_ms; sync handlers via asyncio.to_thread (documented GIL-bound). Cancel: if hash.cancel_requested, cancel the asyncio task (best effort; thread handlers run to completion, result discarded, status CANCELLED).
  • Heartbeat fields per section 2. Status: starting -> ready (idle) / busy (in_flight>0) -> draining (SIGTERM) -> stopped. STALE is observer-computed (last_heartbeat older than 3x interval), never self-reported.
  • Graceful shutdown: stop polling/scheduling, wait DRAIN_TIMEOUT_S (30s) for in-flight, then exit. Never ACK unexecuted work. Unacked entries are reclaimed by peers.

7. Backpressure

  • API admission: DTQ_MAX_QUEUE_DEPTH (10000). Ingest sums XLEN across the queue's priority streams; depth >= max -> 429 QUEUE_FULL with Retry-After. Approximate under concurrent ingest; document.
  • Payload: DTQ_MAX_PAYLOAD_BYTES (262144) -> 413 PAYLOAD_TOO_LARGE.
  • Workers: prefetch bounded by concurrency; never pull unbounded entries.
  • Retry pressure: dtq_retry_scheduled gauge; scheduler batch capped at 100.
  • DLQ bounded (MAXLEN 10000). WS per-client buffer 1000; slow-client policy: drop non-critical events, keep terminal ones, disconnect past the cap. Document all in backpressure.md.

8. Events (JSON on dtq:events; also fanned to WebSocket)

{event_id (uuid), ts (ISO UTC), type, task_id, queue, task_type, attempt, worker_id, metadata?} Types: TASK_QUEUED, TASK_STARTED, TASK_SUCCEEDED, TASK_FAILED, TASK_RETRY_SCHEDULED, TASK_RETRIED, TASK_DEAD_LETTERED, TASK_CANCELLED, WORKER_STARTED, WORKER_STOPPED, WORKER_STALE, CONFIGURATION_CHANGED.

9. Metrics (prometheus_client; labels ONLY task_type, queue, status)

  • dtq_tasks_submitted_total{queue,task_type}
  • dtq_tasks_started_total{queue,task_type}
  • dtq_tasks_succeeded_total{queue,task_type}
  • dtq_tasks_failed_total{queue,task_type}
  • dtq_tasks_retried_total{queue,task_type}
  • dtq_tasks_dead_lettered_total{queue,task_type}
  • dtq_tasks_completed_total{queue,task_type,status} (status: succeeded|failed|dead_lettered|cancelled|expired)
  • dtq_task_attempts_total{queue,task_type,result} (result: success|failure|timeout|error)
  • dtq_queue_depth{queue} gauge (sum of priority streams)
  • dtq_queue_pending{queue} gauge (XPENDING total)
  • dtq_retry_scheduled gauge, dtq_dlq_depth gauge
  • dtq_workers_active gauge
  • dtq_worker_inflight{worker_id} gauge (worker_id label is bounded: one per process; acceptable)
  • dtq_worker_tasks_total{worker_id,task_type,result}
  • dtq_task_duration_seconds{queue,task_type} histogram (end-to-end)
  • dtq_task_execution_seconds{queue,task_type} histogram (handler only)
  • dtq_task_wait_seconds{queue,task_type} histogram (created -> started)
  • dtq_retry_delay_seconds{queue,task_type} histogram
  • dtq_redis_operations_total{operation,result} counter
  • dtq_ingest_rejected_total{queue,reason} counter Never task_id, user id, request id, exception text, or payload as labels.

10. Config (DTQ_ prefix, pydantic-settings, validate on startup, fail fast)

DTQ_REDIS_URL=redis://localhost:6379/0, DTQ_API_HOST=0.0.0.0, DTQ_API_PORT=8000, DTQ_WS_PATH=/ws/events, DTQ_WORKER_CONCURRENCY=4, DTQ_WORKER_ID= (auto), DTQ_VISIBILITY_TIMEOUT_S=30, DTQ_HEARTBEAT_INTERVAL_S=5, DTQ_MAX_QUEUE_DEPTH=10000, DTQ_MAX_PAYLOAD_BYTES=262144, DTQ_TASK_TIMEOUT_MS=30000, DTQ_MAX_ATTEMPTS=10, DTQ_RETRY_BASE_DELAY=1.0, DTQ_RETRY_MAX_DELAY=60.0, DTQ_RETRY_JITTER=0.2, DTQ_TASK_RETENTION_S=604800, DTQ_RESULT_RETENTION_S=3600, DTQ_RESULT_MAX_BYTES=8192, DTQ_IDEM_TTL_S=86400, DTQ_DRAIN_TIMEOUT_S=30, DTQ_DLQ_ENABLED=true, DTQ_DLQ_MAXLEN=10000, DTQ_API_KEY= (empty = local dev, auth disabled; document), DTQ_LOG_LEVEL=INFO, DTQ_LOG_JSON=false.

11. API surface (FastAPI, prefix /api/v1; /health /ready /metrics /ws at root)

  • POST /api/v1/tasks -> 202 {task_id, status} new; 200 + duplicate:true replay; 409 IDEMPOTENCY_IN_PROGRESS; 429 QUEUE_FULL; 413 PAYLOAD_TOO_LARGE; 400 INVALID_TASK/UNKNOWN_TASK_TYPE. Body: {task_type, payload, queue="default", idempotency_key?, max_attempts=5, priority=5, timeout_ms=30000, delay_seconds=0, metadata={}}. Idempotency-Key header wins.
  • GET /api/v1/tasks/{task_id} -> full task incl. execution metadata.
  • POST /api/v1/tasks/{task_id}/cancel -> 202 (queued) / 409 (running, cancel requested) / 404 / 410 (already terminal). Auth required (section 13).
  • GET /api/v1/tasks?queue=&status=&limit=50 (recent first).
  • GET /api/v1/queues -> per queue {depth, pending, paused, retry_scheduled}.
  • GET /api/v1/queues/{queue} -> above + dlq_depth, workers_active. Label each value exact or approximate in the docs.
  • GET /api/v1/workers, GET /api/v1/workers/{id} (liveness from heartbeat age).
  • GET /api/v1/dlq?limit=50, GET /api/v1/dlq/{task_id}, POST /api/v1/dlq/{task_id}/requeue (attempt reset to 0, original history kept in metadata.previous_attempts; status QUEUED), POST /api/v1/dlq/{task_id}/discard (marks DISCARDED in hash? No: removes from DLQ stream via XDEL and records audit; task hash status stays DEAD_LETTERED with metadata.discarded=true. Never silently destroy.), POST /api/v1/dlq/purge (XTRIM to 0 + audit).
  • POST /api/v1/queue/{queue}/pause | /resume (auth).
  • GET /health (process alive), GET /ready (Redis ping + group exists; 503 with dependency detail when not), GET /metrics, WS /ws/events.

12. CLI (dtq, argparse, dtq_cli.cli:main; talks to Redis directly)

dtq submit --type T --payload '{}' [--queue Q] [--idempotency-key K] [--max-attempts N] [--priority P] [--timeout-ms MS] [--delay-seconds S] dtq get dtq list [--queue Q] [--status S] [--limit N] dtq cancel dtq dlq list [--limit N] dtq retry-dlq dtq discard-dlq dtq purge-dlq [--yes] dtq purge --queue Q [--yes] dtq queue pause | dtq queue resume dtq queue stats dtq worker run [--concurrency N] [--queue Q]

13. Security

  • Workers execute ONLY @task("name") registered handlers. No eval/exec/ shell=True/subprocess/dynamic import of payload data. No Python source in the API or Redis. Unknown type -> permanent failure path, never execution.
  • Auth: DTQ_API_KEY. When set, destructive ops (cancel, dlq requeue/discard/ purge, queue pause/resume/purge) require X-API-Key. Reads and submit stay open for local use. When unset, everything is open (documented local-dev).
  • Audit: every destructive op appends to dtq:stream:audit {actor, action, task_id, ts, request_id} and logs it.
  • Validation: Pydantic everywhere; queue name regex; payload size; metadata 4KB; log injection: sanitize newlines in logged fields.
  • Logging: JSON when DTQ_LOG_JSON=true with fields timestamp, level, service, worker_id, task_id, task_type, queue, attempt, status, duration_ms, event_type. Never log payloads by default or secrets.
  • Docker: non-root users, no exposed Redis port by default, minimal images.

14. Demo task types (examples/, synthetic data only)

echo_task, sleep_task(seconds, max 30), fibonacci_task(n, CPU-bound, marked), flaky_task(fail_times then succeed), always_fail_task, counter_task (atomic Redis INCR demo), slow_task. No shell, no network, no real data.

15. Handler contract (dtq_tasks)

@task("name") registers fn(payload: dict, ctx: TaskContext) -> JSON-serializable. TaskContext: task_id, idempotency_key, attempt, worker_id, queue, deadline. Async functions awaited; sync run in threadpool. Exceptions: dtq_tasks.RetryableError -> retryable; dtq_tasks.PermanentError -> permanent. Helper: dtq_tasks.idempotent_operation(redis, key, ttl, fn) -> runs fn once per key via SET NX (for handler-level dedup). Document with an example.