feat: add database-backed job coordinator

docs/jobcoordinator.md

@@ -0,0 +1,143 @@
+# Job Coordinator (database-backed)
+
+The job coordinator turns **any executable** into a distributed worker. It is a
+database-backed sibling of the file-based grid wrapper: instead of
+`.LOCKED/.PROCESSED/.FAILED` marker files on a shared filesystem, the queue and
+its coordination state live in a database (SQLite or PostgreSQL).
+
+A **job** is a whole batch/run; a **task** is one unit of work within it
+(typically a file path). Two commands make up the workflow:
+
+| Command | Purpose |
+| --- | --- |
+| `claimed propagate_jobs` | Populate a job with tasks from a glob/directory (status `pending`). |
+| `claimed work_jobs`      | Claim the job's tasks one-by-one and run a worker script per task. |
+
+Because the database is the single source of truth and claiming is atomic, many
+workers — on one machine or across many nodes — cooperate safely and never
+process the same task twice.
+
+## Concepts
+
+Each task is one row in the `claimed_jobs` table, identified by its **job** (the
+batch namespace) plus its **task_name** (the file path emitted by
+`propagate_jobs`). The job is a mandatory namespace: it lets many independent
+batches share one database, and workers only ever claim tasks from the job they
+were pointed at. Uniqueness is per `(job, task_name)`, so the same file path may
+appear in more than one job. A task moves through:
+
+```
+pending ── claim ──▶ processing ── worker exit 0 ──▶ succeeded
+                               └── worker exit ≠0 ──▶ failed
+```
+
+The table also records `worker_id`, `attempts`, timestamps, and (on failure) an
+`error` message.
+
+## `propagate_jobs`
+
+```bash
+claimed propagate_jobs --db <url> --job <name> <pattern>
+```
+
+- `--db <url>` — a SQLite file path (e.g. `jobs.db`, `/tmp/jobs.db`) or a full
+  database URL. Bare paths and `*.db` / `*.sqlite` become SQLite; `sqlite:///…`
+  and `postgresql://…` (also the legacy `postgres://…`) are used as given.
+- `--job <name>` — **mandatory** namespace for this batch. Use a distinct job
+  name per independent run so several runs can share one database.
+- `<pattern>` — a glob (with `**` recursion) or a directory. **Quote it** so the
+  shell does not expand the glob before CLAIMED sees it.
+
+The command is **idempotent**: re-running only inserts `(job, task_name)` pairs
+that do not already exist (`INSERT … ON CONFLICT DO NOTHING`), so you can grow
+the queue incrementally.
+
+```bash
+claimed propagate_jobs --db /tmp/jobs.db --job run-2026 'data/**/*.tif'
+# Inserted 42 tasks into job 'run-2026' (0 already existed). Total pending: 42
+```
+
+## `work_jobs`
+
+```bash
+claimed work_jobs --db <url> --job <name> --worker <script> [options]
+```
+
+The worker loop claims the oldest pending task **in `--job`**, marks it
+`processing`, runs the worker, then marks it `succeeded` (exit 0) or `failed`
+(any other exit code or a launch error). It exits when the job's queue is
+empty. The job must match the one used in `propagate_jobs`.
+
+The task name is passed to the worker **both** ways:
+
+- as the first positional argument (`$1`)
+- as the `CLAIMED_TASK` environment variable
+
+`CLAIMED_JOB` and `CLAIMED_WORKER_ID` are also exported.
+
+Options:
+
+| Option | Default | Meaning |
+| --- | --- | --- |
+| `--worker-id NAME`   | `<hostname>-<pid>` | Identifier recorded on claimed tasks. |
+| `--max-tasks N`      | drain queue        | Stop after N tasks this run. |
+| `--poll-interval S`  | `0`                | When empty, wait S seconds and retry instead of exiting. |
+
+### Running workers in parallel
+
+Just launch `work_jobs` more than once against the same `--db` and `--job`:
+
+```bash
+for i in 1 2 3 4; do
+  claimed work_jobs --db postgresql://user:pass@host/claimed --job run-2026 --worker ./worker.sh &
+done
+wait
+```
+
+Two runs that use **different** `--job` values against the same database run
+fully independently — neither sees or claims the other's tasks.
+
+## Atomic claiming
+
+Claiming is always scoped to the worker's `--job`.
+
+- **PostgreSQL** uses `SELECT … FOR UPDATE SKIP LOCKED` so concurrent workers
+  skip rows another worker is already claiming — no contention, no duplicates.
+- **SQLite** has no `SKIP LOCKED`; writers are serialized with `BEGIN IMMEDIATE`
+  plus a guarded `UPDATE … WHERE status='pending'` and a short retry loop, with
+  WAL mode and a busy timeout enabled for cross-process concurrency.
+
+## PostgreSQL support
+
+PostgreSQL needs the `psycopg2` driver, available via the project extra:
+
+```bash
+pip install 'claimed[postgresql]'
+```
+
+Remote hosts should use SSL — pass it in the URL, e.g.
+`postgresql://user:pass@host:5432/claimed?sslmode=require`.
+
+## Writing a worker
+
+Any executable works. Read the task name, do the work, and use the exit code to
+report success or failure:
+
+```bash
+#!/usr/bin/env bash
+set -euo pipefail
+TASK="${1:-$CLAIMED_TASK}"
+python process.py --input "$TASK" --output "out/$(basename "$TASK")"
+```
+
+A complete, runnable example lives in
+[`examples/jobcoordinator_example/`](https://github.com/claimed-framework/claimed/tree/main/examples/jobcoordinator_example).
+
+## Relationship to the grid wrapper
+
+The [grid wrapper](c3/create-gridwrapper.md) coordinates work through marker
+files on a shared filesystem (or object store) and wraps a CLAIMED *component*.
+The job coordinator coordinates through a database and wraps an *arbitrary
+script*. Use the grid wrapper when you already have a containerised component and
+a shared filesystem; use the job coordinator when you want a database as the
+source of truth, robust multi-node claiming, or a plain shell/Python worker.

examples/jobcoordinator_example/README.md

@@ -0,0 +1,78 @@
+# Job Coordinator example
+
+A database-backed coordinator that turns any shell script into a distributed
+worker. One command fills a **job** (a batch of tasks) from a set of files;
+another pulls the job's **tasks** one-by-one and runs your worker for each. The
+database (SQLite or PostgreSQL) holds the queue and provides atomic claiming, so
+many worker sessions — on the same machine or across many nodes — can cooperate
+without a shared filesystem and without ever processing the same task twice.
+
+See [`../../docs/jobcoordinator.md`](../../docs/jobcoordinator.md) for the full
+reference.
+
+## 1. Create some input files
+
+```bash
+mkdir -p /tmp/jc/data
+touch /tmp/jc/data/a.tif /tmp/jc/data/b.tif /tmp/jc/data/c.tif
+```
+
+## 2. Propagate tasks (status: pending)
+
+Every batch is namespaced by a mandatory `--job`, so many independent jobs can
+share one database without colliding.
+
+Local SQLite (no server needed — a bare path becomes a SQLite database):
+
+```bash
+claimed propagate_jobs --db /tmp/jc/jobs.db --job scenes-2026 '/tmp/jc/data/**/*.tif'
+# Inserted 3 tasks into job 'scenes-2026' (0 already existed). Total pending: 3
+```
+
+PostgreSQL (real multi-node coordination):
+
+```bash
+claimed propagate_jobs \
+  --db postgresql://user:pass@localhost:5432/claimed \
+  --job scenes-2026 \
+  '/data/scenes/**/*.tif'
+```
+
+Quote the glob so your shell does not expand it before CLAIMED sees it.
+`propagate_jobs` is idempotent: re-running skips `(job, task_name)` pairs that
+already exist.
+
+## 3. Work the queue
+
+```bash
+claimed work_jobs --db /tmp/jc/jobs.db --job scenes-2026 --worker ./worker.sh
+```
+
+Each task is claimed atomically and marked `processing`, then `succeeded` /
+`failed` based on the worker's exit code. The task name (the file path) is passed
+to the worker as `$1` **and** as `$CLAIMED_TASK` (with `$CLAIMED_JOB` and
+`$CLAIMED_WORKER_ID` also exported). Workers only claim tasks from their own
+`--job`.
+
+Run several workers at once for parallelism — just launch the command multiple
+times (or on multiple nodes) against the same `--db` and `--job`:
+
+```bash
+claimed work_jobs --db /tmp/jc/jobs.db --job scenes-2026 --worker ./worker.sh &
+claimed work_jobs --db /tmp/jc/jobs.db --job scenes-2026 --worker ./worker.sh &
+wait
+```
+
+### Useful options
+
+- `--worker-id NAME`   label recorded on claimed tasks (default `<host>-<pid>`)
+- `--max-tasks N`      stop after N tasks (default: drain the queue)
+- `--poll-interval S`  when the queue is empty, wait S seconds and retry instead
+                       of exiting — handy when producers keep adding tasks
+
+## Writing your own worker
+
+`worker.sh` here is a template. Read the task name from `$1` or
+`$CLAIMED_TASK`, do your work, and exit `0` on success or non-zero on
+failure. Anything executable works — a Python script, a `bsub`/`sbatch`
+submission wrapper, a container invocation, etc.

examples/jobcoordinator_example/worker.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+#
+# Example CLAIMED job-coordinator worker.
+#
+# The coordinator (`claimed work_jobs`) invokes this script once per task and
+# passes the task name (the file path recorded by `claimed propagate_jobs`) in
+# two ways, so use whichever is convenient:
+#
+#   $1                  -> the task name (first positional argument)
+#   $CLAIMED_TASK       -> the task name (environment variable)
+#   $CLAIMED_JOB        -> the job (batch) this task belongs to
+#   $CLAIMED_WORKER_ID  -> id of the worker session running this task
+#
+# Contract:
+#   exit 0        -> the coordinator marks the task "succeeded"
+#   exit non-zero -> the coordinator marks the task "failed" (error recorded)
+#
+# Replace the body with your real work: load the file at "$TASK", process it,
+# and write results wherever you need.
+set -euo pipefail
+
+TASK="${1:-${CLAIMED_TASK:-}}"
+
+if [[ -z "$TASK" ]]; then
+  echo "worker.sh: no task name provided" >&2
+  exit 2
+fi
+
+echo "worker ${CLAIMED_WORKER_ID:-?} handling task: $TASK"
+
+# --- do the real work here -------------------------------------------------
+# e.g. gdalinfo "$TASK"; python process.py --input "$TASK" --output out/
+sleep 1
+# ---------------------------------------------------------------------------
+
+# Demo of the failure path: any task whose name contains FAIL exits non-zero.
+if [[ "$TASK" == *FAIL* ]]; then
+  echo "worker.sh: simulated failure for $TASK" >&2
+  exit 1
+fi
+
+echo "worker done: $TASK"

mkdocs.yml

@@ -81,6 +81,7 @@ nav:
       - create_operator: c3/create-operator.md
       - create_gridwrapper: c3/create-gridwrapper.md
       - operator_utils: c3/operator-utils.md
+  - Job Coordinator: jobcoordinator.md
   - MLX Backend:
       - Overview: mlx/index.md
       - cos_backend: mlx/cos-backend.md

pyproject.toml

@@ -96,6 +96,8 @@ test = [
 ]
 utility    = ["geobench"]
 nvidia     = ["pynvml"]
+# Used by iterate2's PostgreSQL coordinator plugin and by the claimed job
+# coordinator (`claimed propagate_jobs` / `claimed work_jobs`) for Postgres URLs.
 postgresql = ["psycopg2-binary>=2.9"]
 amd        = ["pyrsmi"]
 

src/claimed/claimed.py

@@ -116,6 +116,11 @@ def main():
         _run_module(sys.argv[2:])
         return
 
+    if len(sys.argv) > 1 and sys.argv[1] in ('propagate_jobs', 'work_jobs'):
+        from claimed.jobcoordinator.cli import main as jobcoordinator_main
+        jobcoordinator_main(sys.argv[1], sys.argv[2:])
+        return
+
     dir_path = os.path.dirname(os.path.realpath(__file__))
     return subprocess.call(
         f'{dir_path}/scripts/claimed ' + ' '.join(sys.argv[1:]), shell=True

src/claimed/jobcoordinator/__init__.py

@@ -0,0 +1,13 @@
+"""Database-backed job coordinator for CLAIMED.
+
+A **job** is a whole batch/run; a **task** is one unit of work within it
+(typically a file path).  Populate a job with tasks from a glob/directory of
+files (``propagate_jobs``) and let many worker sessions pull the job's tasks
+one-by-one (``work_jobs``), invoking any shell script as the worker.  The
+database is the single source of truth and provides atomic claiming via row
+locks, so workers can run across many nodes without a shared filesystem.
+
+See :mod:`claimed.jobcoordinator.db` for the storage/claim logic and
+:mod:`claimed.jobcoordinator.cli` for the command-line entry points wired into
+``claimed propagate_jobs`` / ``claimed work_jobs``.
+"""