distributed_task_queue

(forks: edit the badge URL in README to your repo)

Distributed task orchestrator in Go (CockroachDB + Redis). Design and schema: INSTRUCTIONS.md.

Prerequisites

• Go 1.23+ (see go / toolchain in go.mod; CI uses go-version-file)
• Either Docker (recommended for local dependencies) or your own CockroachDB + Redis

Go build: cannot find package / -mod=vendor

If go build or go run fails with cannot find package, or import lookup disabled by -mod=vendor:

• Incomplete vendor/: When a vendor/ directory exists, Go may use vendor mode (-mod=vendor). Every dependency must appear under vendor/ (for example vendor/github.com/jackc/..., vendor/github.com/redis/...). From the repo root run go mod vendor after go.mod is complete, or delete vendor/ if you prefer module cache builds only.
• GOFLAGS: If you set GOFLAGS=-mod=vendor globally (go env -w, shell profile, or Cursor/VS Code go.buildFlags / go.toolsEnvVars), either keep vendor/ in sync with go mod vendor or remove -mod=vendor when vendor/ is incomplete. Cloud sync (e.g. OneDrive) can sometimes leave vendor/ partially synced—re-run go mod vendor locally if needed.

Reproducibility

Item	How it’s pinned
Go toolchain	go / toolchain in go.mod
Modules	Committed go.sum; run go mod download and go mod verify after clone
Containers	Pinned image tags in docker-compose.yml (cockroachdb/cockroach:v24.3.4, redis:7.4.2-alpine)
Schema	migrations/001_initial.sql via Makefile migrate, scripts/migrate.sh, scripts/migrate.ps1, or scripts/migrate.cmd (Windows cmd)
CI	.github/workflows/ci.yml: go mod verify, go vet, go test ./..., go build ./cmd/...

After a fresh clone (Docker running), from the repo root:

docker compose up -d
scripts\migrate.cmd
set CRDB_DSN=postgresql://root@127.0.0.1:26257/defaultdb?sslmode=disable
go run .\cmd\orchestrator

Use a second cmd window for go run .\cmd\worker with the same CRDB_DSN.

Same checks as CI locally:

make verify

scripts\verify.cmd

Docker Compose (local CockroachDB + Redis)

From the repo root:

docker compose up -d

Stop: docker compose down.

• SQL (from the host): postgresql://root@127.0.0.1:26257/defaultdb?sslmode=disable
• CockroachDB UI (mapped to avoid clashing with the orchestrator): http://127.0.0.1:8089
• Redis: 127.0.0.1:6379

Apply the schema (pick one):

# Make (Git Bash / WSL / Unix shell)
make migrate

# Or bash helper
./scripts/migrate.sh

# Or PowerShell (with Docker Compose Cockroach running)
powershell -ExecutionPolicy Bypass -File .\scripts\migrate.ps1

# PowerShell without Docker: set CRDB_DSN and use the Cockroach CLI on PATH
# $env:CRDB_DSN = "postgresql://root@127.0.0.1:26257/defaultdb?sslmode=disable"
# powershell -ExecutionPolicy Bypass -File .\scripts\migrate.ps1

Copy .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 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.

make cluster-up
make cluster-migrate
make cluster-status

Point both binaries at all three nodes so pgx fails over when one dies:

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:

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:

cockroach sql --url "$CRDB_DSN" -f migrations/001_initial.sql

(Use your cluster URL or postgresql:// DSN with pgx-compatible options.)

Configuration

Loaded by both binaries via internal/config. CRDB_DSN is required.

Variable	Default	Purpose
CRDB_DSN	—	CockroachDB / Postgres DSN for pgx
REDIS_ADDR	127.0.0.1:6379	Redis address
REDIS_KEY_PREFIX	dto:	Prefix for Redis keys (see INSTRUCTIONS)
ORCHESTRATOR_LISTEN	:8080	Orchestrator HTTP bind (GET /healthz, GET /metrics, task/job routes below)
RECONCILE_INTERVAL	30s	Background reconciler: re-enqueue due queued tasks (CRDB → Redis)
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 to your CockroachDB cluster.
2. Start Redis (default 127.0.0.1:6379).
3. In one terminal, run the orchestrator (HTTP API + enqueue).
4. In another, run the worker (registers a demo echo handler).

export CRDB_DSN='postgresql://root@localhost:26257/defaultdb?sslmode=disable'
go run ./cmd/orchestrator
go run ./cmd/worker

Submit a task (orchestrator listens on :8080 by default):

curl -sS -X POST http://127.0.0.1:8080/v1/tasks \
  -H 'Content-Type: application/json' \
  -d '{"kind":"echo","payload":{"msg":"hi"}}'

Multi-task job with a DAG (b runs after a completes successfully):

curl -sS -X POST http://127.0.0.1:8080/v1/jobs \
  -H 'Content-Type: application/json' \
  -d '{"tasks":[{"name":"a","kind":"echo","payload":{"n":1}},{"name":"b","kind":"echo","payload":{"n":2},"depends_on":["a"]}]}'

Task names must be unique per job. Dependency edges are validated (unknown names, self-deps, and cycles return 400). A dependent is promoted to queued only when every dependency is completed. If a task fails permanently, downstream tasks still in pending are cascaded to failed (transitive, BFS).

The worker logs a line like echo: kind=echo attempt=1 payload=.... GET http://127.0.0.1:8080/healthz checks DB + Redis connectivity.

Prometheus — GET http://127.0.0.1:8080/metrics exposes Go process/runtime metrics, orchestrator_http_*, orchestrator_tasks_submitted_total, orchestrator_jobs_submitted_total, and reconciler counters (orchestrator_reconcile_*).

Read APIs — GET /v1/tasks/{id} returns a task row (status, attempts, timestamps, payload). GET /v1/jobs/{id} returns the job plus all tasks in that job. Unknown UUIDs return 404; malformed IDs return 400.

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.

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):

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 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.

Layout

• docker-compose.yml — local CockroachDB + Redis
• docker-compose.multinode.yml — three-node CockroachDB + Redis for failover testing
• 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)
• internal/config — environment configuration
• internal/db — Cockroach pool (pgxpool), jobs/tasks, task_run lifecycle
• internal/redis — Redis client, key layout, ready LIST, lease hashes, scheduled ZSET helpers
• internal/orchestrator — submit path, HTTP handlers (GET/POST v1), reconciler (ReconcileOnce, ReclaimStaleRunningOnce)
• internal/worker — pkg/worker.Runtime implementation
• pkg/worker — task handler API types