pravit-amp · pravit-amp · Aug 11, 2026 · Aug 11, 2026 · Aug 11, 2026 · Aug 11, 2026
diff --git a/.env.example b/.env.example
@@ -13,3 +13,7 @@ LEASE_DURATION=30s
 STALE_RUNNING_AFTER=1m
 RETRY_BACKOFF=5s
 WORKER_CONCURRENCY=1
+# How often the worker upserts its workers row (minimum 1s)
+WORKER_HEARTBEAT_INTERVAL=30s
+# Stable worker identity; defaults to a random UUID per process
+# WORKER_ID=worker-1
diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md
@@ -59,7 +59,7 @@ flowchart LR
   W2 --> CRDB
 ```
 
-**Invariant:** If Redis is flushed or inconsistent, the system remains correct by reconciling from CRDB: **(implemented)** the orchestrator periodically selects **`queued`** tasks with `scheduled_at <= now()` and LPUSHes task IDs after a Redis **pending** marker (`SET NX`) to avoid duplicate enqueue storms; it also **reclaims** **`running`** tasks that are older than a configurable threshold **and** have **no Redis lease key** (worker heartbeats stopped), closing open `task_runs` and returning the task to `queued`. Redis is an optimization, not the system of record.
+**Invariant:** If Redis is flushed or inconsistent, the system remains correct by reconciling from CRDB: **(implemented)** the orchestrator periodically selects **`queued`** tasks with `scheduled_at <= now()` and LPUSHes task IDs after a Redis **pending** marker (`SET NX`) to avoid duplicate enqueue storms; it also **reclaims** **`running`** tasks that are older than a configurable threshold **and** have **no Redis lease key** (worker heartbeats stopped), closing open `task_runs` as **`dead`** and returning the task to `queued` without consuming a retry. Redis is an optimization, not the system of record.
 
 ---
 
@@ -104,14 +104,17 @@ distributed_task_queue/
 - Timestamps: **`TIMESTAMPTZ`** everywhere.
 - Status fields: use **`STRING`** with enumerated values documented below (or `ENUM`-like check constraints if preferred).
 - **Transaction boundaries:** creating a job with all tasks and dependency rows MUST occur in **one transaction**. Transitioning a task and inserting a `task_runs` row for a new attempt SHOULD be one transaction where practical.
+- **`attempt` is a retry-budget counter, not a run key.** `tasks.attempt` counts how much of `max_attempts` has been consumed. Reclaiming a stale `running` task rolls it back by one — deliberately, so losing a worker does not cost the task a retry — and the next claim re-issues the same number. `(task_id, attempt_number)` in `task_runs` is therefore **not unique** and MUST NOT be treated as one: a reclaimed-and-retried task holds a `dead` row and a later row sharing that number. The run key is `task_runs.id`; history is ordered by `started_at` (the index is built for exactly this). Distinguish an abandoned run from its successor by `status = 'dead'`, not by the `error` text and not by attempt number.
 
 ### Enumerated values
 
 **`jobs.status`:** `pending`, `running`, `completed`, `failed`, `cancelled`
 
 **`tasks.status`:** `pending`, `queued`, `running`, `completed`, `failed`, `cancelled`
 
-**`task_runs.status`:** `running`, `succeeded`, `failed` (implementation; a `dead` / stale-run distinction may be added later)
+**`task_runs.status`:** `running`, `succeeded`, `failed`, `dead`
+
+`failed` means the handler ran and reported an error. `dead` means the run was abandoned — the worker stopped heartbeating and the orchestrator reclaimed the task — so no result was ever reported. Reclaim also writes the reason into `error`, but the status is the field to branch on.
 
 ### DDL
 
@@ -158,6 +161,7 @@ CREATE TABLE task_dependencies (
 
 CREATE INDEX idx_task_dependencies_depends ON task_dependencies (depends_on_task_id);
 
+-- attempt_number is intentionally not unique per task_id (see the invariant under Conventions).
 CREATE TABLE task_runs (
     id UUID NOT NULL PRIMARY KEY DEFAULT gen_random_uuid(),
     task_id UUID NOT NULL REFERENCES tasks (id) ON DELETE CASCADE,
@@ -211,11 +215,21 @@ Per claimed task, store lease metadata so other workers do not claim it until ex
   Fields: `worker_id`, `deadline_ms` (or `deadline` as Unix ms string), optional `run_id` (UUID of `task_runs` row).
 - **TTL:** set **EXPIRE** on the hash to slightly exceed lease duration so orphaned keys disappear; correctness still comes from CRDB reconciliation if a worker dies without releasing.
 
-**Claim flow (implemented):** worker `BRPOP` → CRDB `ClaimQueuedTask` (queued→running) → `InsertTaskRun` → `SetLease` on the hash → handler runs → `DeleteLease` on success path; heartbeat extends key TTL until completion.
+**Claim flow (implemented):** worker `BRPOP` → CRDB `ClaimQueuedTask` (queued→running) → release the pending marker → `InsertTaskRun` → `SetLease` on the hash → handler runs → `DeleteLease` on success path; heartbeat extends key TTL until completion. If the claim does **not** stick (another worker won it, or the DB call failed), the marker is released anyway — `BRPOP` already removed the id from the ready list, so this worker owns the delivery and must hand it back.
 
 ### Pending enqueue deduplication
 
-**Implemented:** `{prefix}queue:pending:{task_id}` — short TTL via `SET NX` so the reconciler and other producers do not LPUSH the same task repeatedly while it is already queued or in flight. Released when a worker successfully claims the task.
+**Implemented:** `{prefix}queue:pending:{task_id}` — set via `SET NX` with a TTL so the reconciler and other producers do not LPUSH the same task repeatedly while it is already queued or in flight.
+
+**The marker is a lock on the delivery, and whoever ends a delivery MUST release it.** While it is set, `EnqueueDueTaskID` will not push the task, so a marker that outlives its delivery leaves the task `queued`, due, and unreachable — a deadlock that only the TTL breaks. Release is required on **all three** exit paths, not just the successful one:
+
+| Path | Released by |
+|------|-------------|
+| Worker claims the task | `Runtime.processTask` after `ClaimQueuedTask` succeeds |
+| Worker pops the id but the claim fails or is lost | `Runtime.processTask` on the error path |
+| Orchestrator reclaims a stale `running` task, abandoning the in-flight delivery | `ReclaimStaleRunningOnce` before it re-enqueues |
+
+The TTL is a **backstop** for a release that never happened (the process died between `LPUSH` and claim), not the primary mechanism. It does not need to exceed the worst-case `LPUSH`→claim latency: expiring while the task is still on the ready list only lets a producer push a duplicate id, and duplicates are harmless because `ClaimQueuedTask` is a single conditional `UPDATE` — the second worker to pop it gets `ErrTaskNotClaimable`. Prefer a **short** TTL: expiring early costs a wasted `BRPOP`, expiring late stalls a task for the whole TTL.
 
 ### Delayed / scheduled tasks
 
@@ -239,6 +253,8 @@ Per claimed task, store lease metadata so other workers do not claim it until ex
 
 **CockroachDB always wins** for `tasks.status` and attempts. Redis entries that disagree with CRDB are harmless at claim time (`ClaimQueuedTask` gates execution). Recovery: reconciler re-enqueues due **`queued`** rows; separate path **reclaims** stale **`running`** rows when the Redis lease key is absent (see §2 invariant).
 
+One asymmetry is worth stating, because it is the way this design can actually stall: Redis state that says **"there is more work than there is"** self-corrects at claim time, but Redis state that says **"a delivery is already in flight"** — a pending marker — *suppresses* recovery instead of triggering it. The reconciler treats the marker as authoritative and skips the task, so a leaked marker is not harmless the way a duplicate ready-list entry is. That is why the release paths above are mandatory rather than best-effort, and why the marker carries a TTL. Applies only to the pending marker; the lease hash fails the safe way, since a missing lease is what *causes* reclaim.
+
 ---
 
 ## 6. Worker interface (Go)
@@ -289,7 +305,7 @@ Implementations in `internal/worker` construct `Runtime` with Redis + DB clients
 
 | Topic | Behavior |
 |--------|----------|
-| **Delivery** | **At-least-once.** The same logical attempt may be redelivered after crash, lease expiry, or network partition. |
+| **Delivery** | **At-least-once.** The same logical attempt may be redelivered after crash, lease expiry, or network partition. Redelivery reuses the attempt number, so `task_runs` holds one `dead` row and one live row sharing it — see the `attempt` invariant in §4. |
 | **Success** | Handler returns `nil` → runtime **acks**: clear lease in Redis, update CRDB task to `completed`, close `task_runs` as `succeeded`, orchestrator may enqueue dependents. |
 | **Failure** | Handler returns error → runtime records error, increments attempt if under `max_attempts`, applies **backoff** to `scheduled_at`, sets status to `pending` or `queued` per policy, may re-enqueue to Redis after delay. |
 | **Lease / heartbeat** | While `Handler` runs, periodically extend Redis lease (and optionally refresh `task_runs.started_at` semantics); if extension fails, cancel handler `ctx` so shutdown is cooperative. |
@@ -316,6 +332,7 @@ The control plane is **HTTP only** (no gRPC in this repo). Bind address: `ORCHES
 | `GET` | `/v1/tasks/{id}` | Task row JSON (status, attempts, timestamps, payload); **404** if unknown; **400** if `id` is not a UUID |
 | `POST` | `/v1/jobs` | JSON body `{"tasks":[...]}` — DAG job with `name`, `kind`, `payload`, optional `depends_on` (names); **201** + `job_id` and `tasks` name→id map; **400** on validation (cycles, unknown deps, etc.) |
 | `GET` | `/v1/jobs/{id}` | Job metadata + all tasks in the job; **404** / **400** as above |
+| `GET` | `/v1/workers` | Workers that heartbeat recently, from the `workers` table. Optional `?active_since=` duration (default `2m`); **400** if not a positive duration |
 
 ---
 
@@ -333,6 +350,7 @@ Loaded from the environment by both `cmd/orchestrator` and `cmd/worker` (`intern
 | `STALE_RUNNING_AFTER` | `2 × LEASE_DURATION` | Minimum time a task may stay `running` before reclaim is considered (still requires Redis lease to be absent) |
 | `WORKER_ID` | random UUID | Stable worker identity for `task_runs.worker_id` |
 | `WORKER_CONCURRENCY` | `1` | Parallel BRPOP loops in the worker |
+| `WORKER_HEARTBEAT_INTERVAL` | `30s` | How often the worker upserts its `workers` row; must be ≥ `1s`. Registry only — the Redis **lease** heartbeat is `pkg/worker.Options.HeartbeatEvery` (5s default, code-configured, no env var) |
 | `LEASE_DURATION` | `30s` | Logical lease window; Redis key TTL adds a buffer |
 | `RETRY_BACKOFF` | `5s` | Delay before a failed task is re-queued when attempts remain |
 

diff --git a/README.md b/README.md
@@ -80,6 +80,35 @@ powershell -ExecutionPolicy Bypass -File .\scripts\migrate.ps1
 
 Copy [`.env.example`](.env.example) to `.env` and load it in your shell if you use a tool that reads `.env` automatically; otherwise set `CRDB_DSN` / `REDIS_ADDR` as in the table below.
 
+## Three-node cluster (failover testing)
+
+[`docker-compose.multinode.yml`](docker-compose.multinode.yml) runs three CockroachDB nodes plus Redis, so you can kill a node and watch the system recover. The single-node compose above is enough for normal development.
+
+```bash
+make cluster-up
+make cluster-migrate
+make cluster-status
+```
+
+Point both binaries at **all three** nodes so `pgx` fails over when one dies:
+
+```bash
+export CRDB_DSN='postgresql://root@127.0.0.1:26257,127.0.0.1:26258,127.0.0.1:26259/defaultdb?sslmode=disable'
+```
+
+- **SQL:** `26257` / `26258` / `26259` (nodes 1–3) — **Admin UI:** `8089` / `8090` / `8091` — **Redis:** `6379`
+- Tear down with `make cluster-down` (this also removes volumes).
+
+To exercise the reclaim and re-delivery paths, submit load while killing a node. The cluster keeps quorum with two of three nodes, so writes continue while in-flight workers fail mid-commit:
+
+```bash
+docker kill dtq-multinode-crdb1-1
+```
+
+Restart it with `docker start dtq-multinode-crdb1-1`. Tasks whose worker failed to commit are picked up by `ReclaimStaleRunningOnce` once `STALE_RUNNING_AFTER` passes with no lease. Lower `RECONCILE_INTERVAL` and `LEASE_DURATION` (e.g. `5s` and `10s`) to see it happen in seconds rather than minutes.
+
+> The integration tests share one database and `ReconcileOnce` scans it globally, so a running orchestrator or worker will compete with the test suite for tasks and cause spurious failures. Stop both before running `go test`.
+
 ## Database migrations (any cluster)
 
 Apply the initial schema:
@@ -104,9 +133,15 @@ Loaded by both binaries via [`internal/config`](internal/config/config.go). `CRD
 | `STALE_RUNNING_AFTER` | `2 × LEASE_DURATION` | Min time a task stays `running` before the reconciler may reclaim it if the Redis lease key is missing |
 | `WORKER_ID` | random UUID | Stable worker identity if set |
 | `WORKER_CONCURRENCY` | `1` | Parallel BRPOP worker loops |
+| `WORKER_HEARTBEAT_INTERVAL` | `30s` | How often the worker upserts its `workers` row (minimum `1s`). **Not** the Redis lease heartbeat — see below |
 | `LEASE_DURATION` | `30s` | Logical lease window; Redis key TTL adds a buffer |
 | `RETRY_BACKOFF` | `5s` | Delay before a failed task is re-queued (when attempts remain) |
 
+There are **two** unrelated heartbeats, and only one is configurable by environment:
+
+- **Worker registry heartbeat** — `WORKER_HEARTBEAT_INTERVAL`, one per process, upserts the `workers` row for dashboards. Nothing in the claim path depends on it.
+- **Lease heartbeat** — one per *running task*, extends the Redis lease TTL so the orchestrator does not reclaim work that is still running. It is `pkg/worker.Options.HeartbeatEvery` (**5s** default), set in code when constructing the runtime; `cmd/worker` passes the zero value, so **no environment variable changes it**. This is the one that matters for reclaim: if it stops, the lease expires and `ReclaimStaleRunningOnce` takes the task back.
+
 ## Run (end-to-end)
 
 1. Apply [migrations](migrations/001_initial.sql) to your CockroachDB cluster.
@@ -149,19 +184,28 @@ curl -sS "http://127.0.0.1:8080/v1/tasks/<task_id>"
 curl -sS "http://127.0.0.1:8080/v1/jobs/<job_id>"
 ```
 
+`GET /v1/workers` lists workers that have heartbeated recently (see `WORKER_HEARTBEAT_INTERVAL`). The window defaults to 2 minutes; `?active_since=` takes any positive duration.
+
+```bash
+curl -sS "http://127.0.0.1:8080/v1/workers?active_since=5m"
+```
+
 Integration tests that hit the DB and Redis run when `CRDB_DSN` is set (e.g. after `docker compose up`):
 
 ```bash
 set CRDB_DSN=postgresql://root@127.0.0.1:26257/defaultdb?sslmode=disable
 go test ./internal/orchestrator/ -count=1 -v
 ```
 
-Jobs move **`pending` → `running` → `completed`** (or **`failed`**) as tasks finish; the orchestrator **reconciler** periodically re-enqueues **`queued`** rows that are due (`scheduled_at <= now()`), using Redis **pending** markers to avoid spamming duplicate LPUSHes. If a worker dies after claiming a task, the Redis lease **TTL** expires (heartbeats stop); the reconciler also **reclaims** long-running `running` rows with **no lease**—closing the open `task_run`, re-queuing the task (same retry budget), and LPUSHing again.
+Jobs move **`pending` → `running` → `completed`** (or **`failed`**) as tasks finish; the orchestrator **reconciler** periodically re-enqueues **`queued`** rows that are due (`scheduled_at <= now()`), using Redis **pending** markers to avoid spamming duplicate LPUSHes. If a worker dies after claiming a task, the Redis lease **TTL** expires (heartbeats stop); the reconciler also **reclaims** long-running `running` rows with **no lease**—closing the open `task_run` as **`dead`**, re-queuing the task (same retry budget), releasing the pending marker left by the abandoned delivery, and LPUSHing again.
+
+A reclaimed task keeps its attempt number, so `task_runs` holds a `dead` row and its retry sharing one `attempt_number`. That is intended: `(task_id, attempt_number)` is **not** unique — see the `attempt` invariant in [INSTRUCTIONS.md §4](INSTRUCTIONS.md).
 
 ## Layout
 
 - [`docker-compose.yml`](docker-compose.yml) — local CockroachDB + Redis
-- [`Makefile`](Makefile) — `make compose-up`, `make migrate`, …
+- [`docker-compose.multinode.yml`](docker-compose.multinode.yml) — three-node CockroachDB + Redis for failover testing
+- [`Makefile`](Makefile) — `make compose-up`, `make migrate`, `make cluster-up`, …
 - `scripts/` — `migrate.sh` / `migrate.ps1` / `migrate.cmd`, `verify.cmd` (same checks as CI / `make verify`)
 - `cmd/orchestrator` — HTTP API, DB ping, task submission
 - `cmd/worker` — worker process (`echo` demo handler)

diff --git a/internal/db/jobs.go b/internal/db/jobs.go
@@ -127,40 +127,46 @@ INSERT INTO task_dependencies (task_id, depends_on_task_id) VALUES ($1, $2)`, ti
 
 // RefreshJobStatus sets jobs.status from tasks: any failed → failed; all completed → completed;
 // all cancelled → cancelled; otherwise running.
+//
+// The counts and the write are one statement on purpose. Splitting them into a SELECT and a
+// later UPDATE loses updates: two workers finishing tasks of the same job can both read, and
+// the one that read the earlier (less complete) snapshot can commit its status last, leaving a
+// finished job stuck at 'running' with nothing left to trigger another refresh. As a single
+// statement this is an implicit transaction, so the read of tasks and the write to jobs share
+// one timestamp and CockroachDB retries it internally when a concurrent refresh conflicts.
 func RefreshJobStatus(ctx context.Context, pool *pgxpool.Pool, jobID uuid.UUID) error {
+	// n_total = 0 leaves the row untouched rather than deriving a status from no tasks;
+	// it is returned so the caller can tell "job has no tasks" from "job does not exist".
 	const q = `
-SELECT
-	count(*) FILTER (WHERE status = 'failed') AS n_failed,
-	count(*) FILTER (WHERE status = 'completed') AS n_done,
-	count(*) FILTER (WHERE status = 'cancelled') AS n_cancelled,
-	count(*) AS n_total
-FROM tasks WHERE job_id = $1`
-	var nFailed, nDone, nCancel, nTotal int
-	if err := pool.QueryRow(ctx, q, jobID).Scan(&nFailed, &nDone, &nCancel, &nTotal); err != nil {
+WITH agg AS (
+	SELECT
+		count(*) FILTER (WHERE status = 'failed') AS n_failed,
+		count(*) FILTER (WHERE status = 'completed') AS n_done,
+		count(*) FILTER (WHERE status = 'cancelled') AS n_cancelled,
+		count(*) AS n_total
+	FROM tasks WHERE job_id = $1
+)
+UPDATE jobs SET
+	status = CASE
+		WHEN agg.n_total = 0 THEN jobs.status
+		WHEN agg.n_failed > 0 THEN 'failed'
+		WHEN agg.n_cancelled = agg.n_total THEN 'cancelled'
+		WHEN agg.n_done + agg.n_cancelled = agg.n_total THEN 'completed'
+		ELSE 'running'
+	END,
+	updated_at = CASE WHEN agg.n_total = 0 THEN jobs.updated_at ELSE now() END
+FROM agg
+WHERE jobs.id = $1
+RETURNING agg.n_total`
+	var nTotal int
+	if err := pool.QueryRow(ctx, q, jobID).Scan(&nTotal); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return fmt.Errorf("db: refresh job: no job row for %s", jobID)
+		}
 		return err
 	}
 	if nTotal == 0 {
 		return fmt.Errorf("db: refresh job: no tasks for job %s", jobID)
 	}
-	var status string
-	switch {
-	case nFailed > 0:
-		status = "failed"
-	case nDone+nCancel == nTotal:
-		if nCancel == nTotal {
-			status = "cancelled"
-		} else {
-			status = "completed"
-		}
-	default:
-		status = "running"
-	}
-	tag, err := pool.Exec(ctx, `UPDATE jobs SET status = $2, updated_at = now() WHERE id = $1`, jobID, status)
-	if err != nil {
-		return err
-	}
-	if tag.RowsAffected() == 0 {
-		return fmt.Errorf("db: refresh job: no job row for %s", jobID)
-	}
 	return nil
 }