feat(queue): the durable queue runs on any Backend, not just Postgres

CHANGELOG.md

@@ -5,6 +5,26 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+
+- **`sutegi-queue` runs on any `Backend`, not just Postgres.** The durable queue moved off `sutegi_pg::Pool` onto the ORM's `Backend` seam, so one jobs table and one set of SQL now work on bundled SQLite (a single box) and on Postgres (many pods) — the same parity `tsvector`↔FTS5 and `jsonb`↔JSON1 already had. Claims stay exclusive on both: `FOR UPDATE SKIP LOCKED` wherever `capabilities().skip_locked` says it exists, and on SQLite the serialized writer already provides it, since the second `UPDATE … RETURNING` no longer sees the claimed row. `queue.cross_pod()` reports which guarantee you actually have instead of letting the docs imply the stronger one. The `queue` feature no longer drags in a Postgres driver.
+- Queue: **times are epoch milliseconds supplied by the caller**, not `now()` + `interval` SQL. That is what lets one statement work in both dialects, and it makes schedules testable without sleeping.
+- Queue: `Queue::new` takes any `Backend + Send + Sync + 'static` (`Queue::with_store` for a store already behind an `Arc`), and handlers receive a `&JobCtx` rather than a `&Json` — **breaking** against 0.5.1: the payload is now `job.payload()`.
+- Queue: `sutegi_jobs` gains `priority` and `unique_key`, and its timestamp columns are integers. Existing tables are not migrated — drop and recreate (pre-1.0, and the queue is not a history table).
+
+### Added
+
+- Queue: **named queues with their own worker pools** — `queue.job(name, payload).queue("video")` plus `Arc::clone(&q).start_on("video", 1)`. This is how a slow job class is kept from starving a fast one, which a single pool cannot express.
+- Queue: **dedupe keys** — `.unique("yt:abc")` hands back the live row's id instead of enqueueing a second copy, backed by a partial unique index that deliberately excludes dead letters, so a failure never owns a key forever.
+- Queue: **priorities** — `.priority(10)` runs ahead of older work within a queue.
+- Queue: **`JobCtx`** — `payload()`, `attempts`/`max_attempts`, `is_last_attempt()` (so a handler can tell a retryable blip from a terminal failure *before* writing a user-visible error), `heartbeat()` (pushes the visibility window forward, which is what lets a job outlive the timeout), and `should_stop()` for handlers that loop.
+- Queue: **a panicking handler is a failed job, not a lost worker** — the panic is caught, the row retries or dead-letters like any other, and the pool keeps going.
+- Queue: **dispatch wakes an idle worker** over a condvar instead of making it wait out the poll interval; the interval stays as the safety net for delayed jobs and for work enqueued by another pod.
+- Queue ops: `failed(limit)` lists dead letters, `retry(id, max_attempts)` revives one, `purge_failed(age)` clears them (inclusive bound, so `Duration::ZERO` really does clear the lot), and `stats_for(queue)` reports a single pool. `stats()` no longer uses Postgres-only `FILTER`.
+- Queue tests: a full SQLite suite (`tests/sqlite.rs`, 16 cases — claim exclusivity under six concurrent workers, crash recovery through an expired lease, a heartbeat defeating a steal, dedupe, priority, named-queue isolation, panics, dead-letter/retry/purge) that needs no server, plus the Postgres leg (`tests/durable.rs`, `--features postgres`) pinning the same contract. Both verified against real backends.
+
 ## [0.8.0] - 2026-07-27
 
 The production user-system release: everything Laravel's auth scaffolding does, still zero third-party dependencies.

README.md

@@ -95,7 +95,7 @@ fn main() -> std::io::Result<()> {
 | `sutegi-storage` | File/object storage behind one `Storage` trait: local fs, database blobs (over `Backend`), and a pure-`std` S3 SigV4 presigner. |
 | `sutegi-macros` | `#[derive(Model)]` (schema, hydration, `save`, `from_input`) and `#[derive(Validate)]` (field-attr rulesets). Compile-time only (syn/quote never reach your binary). |
 | `sutegi-validate` | Fluent `Validator`-style rule sets **and** a JSON Schema subset validator, with structured errors. |
-| `sutegi-queue` | Durable, cross-pod job queue backed by Postgres (`FOR UPDATE SKIP LOCKED` claim, visibility-timeout retries, dead-letter). |
+| `sutegi-queue` | Durable job queue over the `Backend` seam — same SQL on SQLite or Postgres (`UPDATE … RETURNING` claim, `SKIP LOCKED` where the backend has it, visibility-timeout retries, priorities, named queues, dedupe keys, dead-letter). |
 | `sutegi-hexagon`  | Opinionated hexagonal/clean-architecture primitives: `AppError`, `UseCase` ports, `respond` adapter glue. |
 | `sutegi`      | Facade crate + `prelude`. |
 | `sutegi-cli`  | The `sutegi` command: scaffold apps/models/routes, `introspect` a live app. |
@@ -113,7 +113,7 @@ what you use:
 | `validate` | ✓ | request / tool validation + `Ctx::validate`/`validated` |
 | `sqlite`   |   | SQLite backend — the **single-node** runnable store (bundled) |
 | `postgres` |   | Postgres backend — the **multi-pod** runnable store (pure std) |
-| `queue`    |   | durable, cross-pod job queue (Postgres-backed) |
+| `queue`    |   | durable job queue (SQLite or Postgres) |
 | `graceful` |   | SIGTERM/SIGINT draining (libc) |
 | `hexagon`  |   | hexagonal/clean-architecture primitives |
 | `session`  |   | signed-cookie sessions (HMAC-SHA256) |
@@ -384,28 +384,47 @@ state, returning a ready `404`/`500` `Error` you can `?`:
 .get("/api/todos/:id", "show", |c| c.model::<Todo, Db>("id").map(|t| t.to_json()))
 ```
 
-### Background jobs (durable, cross-pod)
+### Background jobs (durable)
 
-The queue is Postgres-backed (`queue` feature), so jobs survive a crash and are
-claimed exactly once across pods (`FOR UPDATE SKIP LOCKED` + a visibility
-timeout, with retries and a dead-letter column):
+The queue (`queue` feature) runs over the `Backend` seam, so one jobs table and
+one set of SQL work on bundled SQLite and on Postgres. Jobs survive a crash: the
+claim stamps a lease instead of deleting the row, so a dead worker's job becomes
+visible again after the visibility timeout (at-least-once). Retries, delays,
+priorities and dedupe keys are columns, so a restart forgets nothing.
+
+Claims are exclusive on both backends — `FOR UPDATE SKIP LOCKED` on Postgres,
+and on SQLite the serialized writer already gives it, since the second `UPDATE`
+no longer sees the claimed row. Postgres additionally makes that exclusivity
+*cross-pod*; `queue.cross_pod()` tells you which one you have.
 
 ```rust
 use std::sync::Arc;
 use sutegi::queue::Queue;
 
-let mut queue = Queue::new(pg.pool().clone());   // over a Pg connection pool
-queue.register("notify", |args| { /* send … */ Ok(()) });  // named handler
-queue.migrate()?;                                 // create sutegi_jobs
+let mut queue = Queue::new(db.clone());            // any Backend: Db or Pg
+queue.register("notify", |job| { /* send job.payload() … */ Ok(()) });
+queue.migrate()?;                                  // create sutegi_jobs
 
-// Enqueue from anywhere (any pod):
+// Enqueue from anywhere (any pod, on Postgres):
 queue.dispatch("notify", Json::obj(vec![("to", Json::str("a@b.com"))]))?;
 
-// Run workers (crash-safe, at-least-once, cross-pod):
-let workers = Arc::new(queue).start(4);           // 4 worker threads
-// … later: workers.stop();
+// Or shape the dispatch: its own pool, one in flight per key, 3 tries.
+queue
+    .job("video.ingest", Json::obj(vec![("id", Json::str("abc"))]))
+    .queue("video")
+    .unique("yt:abc")
+    .max_attempts(3)
+    .dispatch()?;
+
+let queue = Arc::new(queue);
+let fast = Arc::clone(&queue).start(4);             // 4 workers on "default"
+let slow = Arc::clone(&queue).start_on("video", 1); // 1 on the slow queue
+// … later: fast.stop(); slow.stop();
 ```
 
+A handler that can outrun the visibility timeout should say it is still alive
+(`job.heartbeat()`) and notice shutdown (`job.should_stop()`).
+
 ## Auth: the user system
 
 `--features auth,sqlite` (or `auth,postgres`) gives you the Laravel `auth`

crates/sutegi-queue/Cargo.toml

@@ -6,8 +6,17 @@ rust-version.workspace = true
 license.workspace = true
 repository.workspace = true
 authors.workspace = true
-description = "Durable, cross-pod job queue for sutegi, backed by PostgreSQL (via the pure-std sutegi-pg driver): atomic FOR UPDATE SKIP LOCKED claim, visibility-timeout crash recovery, retries, delayed dispatch, dead-letter, introspectable stats."
+description = "Durable job queue for sutegi, over the ORM Backend seam: one set of SQL on bundled SQLite or PostgreSQL — atomic UPDATE…RETURNING claim, visibility-timeout crash recovery, retries with backoff, priorities, named queues, dedupe keys, dead-letter, introspectable stats."
 
 [dependencies]
-sutegi-pg = { path = "../sutegi-pg", version = "0.8.0" }
+sutegi-orm = { path = "../sutegi-orm", version = "0.8.0" }
 sutegi-json = { path = "../sutegi-json", version = "0.8.0" }
+
+# The queue is backend-agnostic — these only decide which backend its own tests
+# can reach. An app enables its backend on sutegi-orm (or the sutegi facade).
+[features]
+sqlite = ["sutegi-orm/sqlite"]
+postgres = ["sutegi-orm/postgres"]
+
+[dev-dependencies]
+sutegi-orm = { path = "../sutegi-orm", version = "0.8.0", features = ["sqlite"] }