Spec: Job Queue — States, Leases & Reaper¶

File: .ai/specs/core-engine/2026-07-02-job-queue-leases.md Status: Draft Owners: Storage lane Related: PR-030/031; composite-key-layout spec (jobs CF, nil-tenant namespace); tombstone-cascade spec (cascade journal rides this queue)

1. Overview¶

Defines job states, the lease protocol that guarantees single-consumer processing without distributed locks, retry/backoff and poison handling, and the reaper that reclaims work from dead workers. Serves ingestion jobs, embed jobs, cascade journals, and system jobs (nil-tenant namespace).

2. Problem Statement¶

Stateless workers polling shared work need exactly one guarantee: no job is processed by two workers simultaneously, and no job is lost when a worker dies mid-flight. Without a defined state machine, "leased" jobs from crashed workers hang forever; without poison handling, one malformed document loops a worker eternally; without atomic lease acquisition, two pollers grab the same job.

3. Proposed Solution¶

State machine pending → leased → completed | failed(retryable) | dead, where every transition is an atomic state-move: one WriteBatch deletes the old state-keyed row and puts the new one (the jobs CF is operational state, not history — the append-only doctrine governs version CFs, not this queue; the audit trail is the job's event log embedded in its value). Leases carry a TTL (default 300s) extended by Heartbeat; the reaper (background task, interval 30s) moves expired leases back to pending with attempt+1 and exponential backoff (base 10s, factor 4, cap 1h), dead-lettering after max_attempts (default 5).

4. Architecture¶

enqueue ─▶ [pending] ─PollJobs(atomic move)─▶ [leased(worker,deadline)]
              ▲                                   │Heartbeat: extend
              │reaper: TTL expired, attempts<max  ├─Submit ok──▶ [completed]
              │                                   ├─Submit err─▶ [failed]→backoff→[pending]
              └───────────────────────────────────┘attempts==max─▶ [dead] (+ alert metric)

5. Data Models¶

Key per composite-key-layout spec: [scope][state 1][priority 1][createdAt 8][jobId 16] — ready-job discovery is a bounded prefix scan on [scope][pending], FIFO within priority. Value: { kind, payload(msgpack), attempts, worker_id?, lease_deadline?, events: [(ts, transition, detail)], idempotency_key }. Job kinds v1: ingest:{format}, embed, cascade, system:{task}. System jobs use SystemScope (nil namespace).

6. API Contracts¶

Engine-internal: enqueue, lease_next(scope_filter, kinds) -> Option<Job> (atomic move), extend(job_ids, worker), complete(job, result), fail(job, reason, retryable). gRPC surface per grpc-contracts spec (PollJobs stream wraps lease_next long-poll; Heartbeat wraps extend; Submit wraps complete/fail). Lease conflicts (stale worker submits after reap) return ABORTED; the submission is discarded (the re-leased attempt owns the job).

7. UI/UX¶

Operator: telha jobs ls --state dead, telha jobs retry <id> (dead→pending, resets attempts), telha jobs show <id> (event log). Dead-letter count exported as an alerting metric.

8. Configuration¶

jobs.lease_ttl (300s), jobs.reaper_interval (30s), jobs.max_attempts (5), jobs.backoff{base,factor,cap}, per-kind overrides allowed.

9. Alternatives Considered¶

Redis/BullMQ for core-internal jobs (rejected: core must not depend on external infrastructure; BullMQ remains app-layer-only). Append-only job history in the queue CF itself (rejected: state-scan efficiency requires state-in-key; history preserved in the value's event log instead). Fencing tokens on Submit (adopted in minimal form: lease generation number checked on submit — covers the stale-worker race without full fencing machinery).

10. Implementation Approach¶

PR-030: this spec → model + key layout → operations with lease-generation check → reaper → contention and recovery tests. PR-031 exposes gRPC. Cascade journal (tombstone-cascade spec) enqueues as kind=cascade with its own resume payload.

11. Migration Path¶

Greenfield. Kind payloads are msgpack with a version byte (same evolution rule as version-record-model spec).

12. Success Metrics¶

Contention test: N concurrent pollers, zero double-leases across 100k jobs. Recovery: kill-9 worker, job re-leased after TTL, completes on retry. Poison: malformed job dead-letters after exactly max_attempts with full event log. Stale-submit: reaped worker's late Submit rejected via lease generation.

13. Open Questions¶

OQ-1: Per-tenant fairness — can one tenant's 1M-job backlog starve others under a global poller? Stance: v1 pollers round-robin known active tenant scopes; measure, and revisit with weighted fair queuing in Phase 4 (EPIC-403 territory).

14. Changelog¶

2026-07-02 — v0.1: initial draft for peer review.
2026-07-05 — v0.2 (PR-060 additions, both additive): (1) JobRecord.max_attempts: Option<u32> — per-job attempt ceiling overriding jobs.max_attempts, set via enqueue_with_max_attempts (clarify jobs budget one attempt per ranked candidate + the steward escalation; old rows decode with the global default). (2) Per-kind backoff-base overrides (jobs.backoff_base_secs_by_kind, internal map, not an env key): an override is both base and at-least-cap, so clarify:temporal retries pace at the escalation window instead of the generic exponential schedule. Consulted by both the fail path and the reaper.