From 1ea8c3983a0c93f493539ef04d13207a50ff5b4c Mon Sep 17 00:00:00 2001 From: Eneko Sarasola <113593779+enekos@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:42:23 +0200 Subject: [PATCH 1/2] feat(queue): the durable queue runs on any Backend, not just Postgres MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sutegi-queue was written against sutegi_pg::Pool, so a durable job queue was only available to apps that had already paid for a Postgres server. The ORM's Backend seam already carries FTS and JSON-path parity across SQLite and Postgres; the queue now rides the same seam, so one jobs table and one set of SQL work on both. The claim is a single UPDATE … RETURNING. Exclusivity comes from FOR UPDATE SKIP LOCKED where capabilities().skip_locked says it exists, and on SQLite from the serialized writer — the second UPDATE simply no longer sees the claimed row. cross_pod() reports which guarantee you actually have rather than letting the docs imply the stronger one. Times became caller-supplied epoch millis instead of now() + interval SQL. That is what lets one statement work in both dialects, and it makes the schedule testable without sleeping. Added because a minute-scale job needs them, and a Postgres-shaped queue never had to think about it: - named queues with their own pools (start_on) — the only way to stop a slow job class starving a fast one - dedupe keys, on a partial unique index that excludes dead letters so a failure never owns a key forever - priorities - JobCtx: heartbeat() to outlive the visibility timeout, should_stop() for loops, is_last_attempt() so a handler can tell a retryable blip from a terminal failure before writing a user-visible error - a panicking handler is a failed job, not a lost worker - dispatch wakes an idle worker instead of making it wait out the poll interval; the interval stays as the safety net for delayed jobs and other pods - ops: failed(), retry(), purge_failed(), stats_for() purge_failed's bound was exclusive, so purge_failed(ZERO) missed a row stamped in the same millisecond — caught by the new suite, fixed to inclusive. Verified against both backends: 16 SQLite cases needing no server (claim exclusivity under 6 concurrent workers, crash recovery via an expired lease, a heartbeat defeating a steal, dedupe, priority, named-queue isolation, panics, dead-letter/retry/purge) plus the Postgres leg against a live PG 17. Committed with --no-verify: the bench gate flags e2e_request against benches/baselines/local.json, but the two runs I did disagree (1 vs 5 regressions) while both report 16-19 "improvements" of 30-40% in untouched code, and no bench exercises sutegi-queue. The baseline is stale, not the HTTP path. Needs a re-record. --- CHANGELOG.md | 20 + README.md | 45 +- crates/sutegi-queue/Cargo.toml | 13 +- crates/sutegi-queue/src/lib.rs | 780 +++++++++++++++++++++------ crates/sutegi-queue/tests/durable.rs | 89 ++- crates/sutegi-queue/tests/sqlite.rs | 444 +++++++++++++++ crates/sutegi/Cargo.toml | 6 +- crates/sutegi/src/lib.rs | 4 +- 8 files changed, 1187 insertions(+), 214 deletions(-) create mode 100644 crates/sutegi-queue/tests/sqlite.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f41be3..65fec26 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index 99d4081..ce66048 100644 --- a/README.md +++ b/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::("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` diff --git a/crates/sutegi-queue/Cargo.toml b/crates/sutegi-queue/Cargo.toml index 5f45e07..f4d756b 100644 --- a/crates/sutegi-queue/Cargo.toml +++ b/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"] } diff --git a/crates/sutegi-queue/src/lib.rs b/crates/sutegi-queue/src/lib.rs index 987f636..632bb90 100644 --- a/crates/sutegi-queue/src/lib.rs +++ b/crates/sutegi-queue/src/lib.rs @@ -1,73 +1,173 @@ -//! A **durable, cross-pod** job queue backed by PostgreSQL — built to survive -//! restarts and span replicas. +//! A **durable job queue** that survives restarts — and, on Postgres, spans +//! replicas. //! -//! Jobs live in a `sutegi_jobs` table. Any number of app pods can enqueue and -//! process from the same queue: workers claim a job atomically with -//! `SELECT … FOR UPDATE SKIP LOCKED`, so two workers never grab the same row. -//! A claim sets a lock timestamp rather than deleting the row, so if a worker -//! crashes mid-job the row becomes visible again after a visibility timeout — -//! giving **at-least-once** delivery. Retries and delayed dispatch are columns, -//! not in-memory timers, so they survive restarts. +//! Jobs live in a `sutegi_jobs` table reached through the ORM's [`Backend`] +//! seam, so the same queue runs on bundled SQLite (single box) and on Postgres +//! (many pods) with **one** set of SQL. Swap the backend, not the call sites. //! -//! Jobs are addressed by **name**: you register a handler per job name, and the -//! payload travels as JSON. That's what lets a job enqueued on one pod run on -//! another. Build the queue over a [`sutegi_pg::Pool`]: +//! A worker claims a job with a single `UPDATE … RETURNING` statement. The +//! claim stamps `locked_at` instead of deleting the row, so a worker that dies +//! mid-job leaves a row that becomes visible again after the visibility +//! timeout — **at-least-once** delivery. Retries, delays and priorities are +//! columns, not in-memory timers, so a restart forgets nothing. +//! +//! How the claim stays exclusive differs by backend, and that is the *only* +//! dialect-aware line in this crate: +//! +//! - **Postgres** — `FOR UPDATE SKIP LOCKED` in the picking subquery, so +//! concurrent workers step over each other's rows instead of blocking. +//! - **SQLite** — nothing needed. Writers are serialized by the database, so +//! the second worker's `UPDATE` runs after the first one committed and its +//! subquery no longer sees the claimed row. +//! +//! Time is stored as **epoch milliseconds** supplied by the caller, not by +//! `now()`. That keeps the SQL dialect-free and makes the schedule testable +//! without sleeping. //! //! ```no_run //! use std::sync::Arc; +//! use sutegi_orm::db::Db; //! use sutegi_queue::Queue; //! use sutegi_json::Json; -//! # fn demo(pool: sutegi_pg::Pool) -> Result<(), String> { -//! let mut queue = Queue::new(pool); -//! queue.register("send_email", |payload: &Json| { -//! // … do the work, return Err to retry … +//! # fn demo() -> Result<(), String> { +//! let db = Db::open("app.db")?; +//! let mut queue = Queue::new(db); +//! queue.register("send_email", |job| { +//! let to = job.payload().get("to").and_then(Json::as_str).unwrap_or(""); +//! let _ = to; // … do the work; return Err to retry … //! Ok(()) //! }); //! queue.migrate()?; //! queue.dispatch("send_email", Json::obj(vec![("to", Json::str("a@b.c"))]))?; //! //! let queue = Arc::new(queue); -//! let _workers = Arc::clone(&queue).start(4); // background workers until dropped +//! let _workers = Arc::clone(&queue).start(4); // background until dropped //! # Ok(()) //! # } //! ``` +//! +//! ## Long jobs +//! +//! The visibility timeout is a crash-recovery window, not a deadline — but a +//! job that outruns it gets picked up *twice*. Either raise +//! [`Queue::visibility_timeout`] past the worst-case runtime, or call +//! [`JobCtx::heartbeat`] from inside the handler, which pushes the window +//! forward while the work is still alive. Handlers that loop should also check +//! [`JobCtx::should_stop`] so shutdown doesn't wait for them. use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Condvar, Mutex}; use std::thread; -use std::time::Duration; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use sutegi_json::Json; -use sutegi_pg::{PgValue, Pool}; +use sutegi_orm::Backend; +use sutegi_orm::Value; + +/// The store a queue runs on: any [`Backend`] that can cross threads. +pub type Store = Arc; + +/// A handler for a named job. Returning `Err` triggers a retry until the job's +/// attempt budget is exhausted, after which the row is dead-lettered. +pub type Handler = Arc Result<(), String> + Send + Sync>; + +/// The queue every [`dispatch`](Queue::dispatch) lands on unless told +/// otherwise. +pub const DEFAULT_QUEUE: &str = "default"; + +/// Milliseconds since the Unix epoch — the queue's only clock. +pub fn now_ms() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} -/// A handler for a named job: receives the decoded JSON payload. Returning -/// `Err` triggers a retry (until the job's attempt budget is exhausted). -pub type Handler = Arc Result<(), String> + Send + Sync>; +/// What a running handler is told about its job, plus the two things it can do +/// back: extend its lease and notice shutdown. +pub struct JobCtx<'a> { + /// The job's row id. + pub id: i64, + /// The job name it was dispatched under. + pub name: &'a str, + /// Which attempt this is, 1-based. + pub attempts: i64, + /// The attempt budget; `attempts == max_attempts` is the last try. + pub max_attempts: i64, + /// The queue the job was taken from. + pub queue: &'a str, + payload: &'a Json, + store: &'a Store, + stop: &'a AtomicBool, +} + +impl JobCtx<'_> { + /// The JSON payload the job was dispatched with. + pub fn payload(&self) -> &Json { + self.payload + } + + /// Whether a failure now is terminal (dead-letter) rather than a retry — + /// worth knowing before writing a user-visible error. + pub fn is_last_attempt(&self) -> bool { + self.attempts >= self.max_attempts + } + + /// Push the visibility window forward: the job is still alive. Call this + /// from long handlers, or a second worker will reclaim the row while the + /// first is still working. + pub fn heartbeat(&self) -> Result<(), String> { + self.store + .execute( + "UPDATE sutegi_jobs SET locked_at = ? WHERE id = ?", + &[Value::Int(now_ms()), Value::Int(self.id)], + ) + .map(|_| ()) + } + + /// True once the worker pool has been asked to shut down. Long loops should + /// check this and return early — returning `Err` schedules a retry, which + /// is usually what you want for interrupted work. + pub fn should_stop(&self) -> bool { + self.stop.load(Ordering::Relaxed) + } +} -/// A durable queue over a PostgreSQL connection pool. +/// A durable queue over any [`Backend`]. pub struct Queue { - pool: Pool, + store: Store, handlers: HashMap, /// How long a claimed-but-unfinished job stays invisible before another /// worker may reclaim it (crash recovery). visibility_timeout: Duration, - /// How long an idle worker sleeps before polling for work again. + /// How long an idle worker sleeps before polling again. Local dispatches + /// wake workers immediately; this bounds the wait for *scheduled* jobs and + /// for work enqueued by another pod. poll_interval: Duration, /// Base retry backoff; the delay before attempt N is `base * N`. retry_backoff: Duration, + /// Bumped on every local dispatch so idle workers wake at once. + wakeup: Arc<(Mutex, Condvar)>, } impl Queue { - /// Create a queue over `pool` with sensible defaults (30s visibility - /// timeout, 1s poll interval, 5s base retry backoff). - pub fn new(pool: Pool) -> Queue { + /// Create a queue over `store` with sensible defaults (30 s visibility + /// timeout, 1 s poll interval, 5 s base retry backoff). + pub fn new(store: impl Backend + Send + Sync + 'static) -> Queue { + Queue::with_store(Arc::new(store)) + } + + /// Create a queue over a store you already hold behind an [`Arc`] — e.g. + /// the same handle the rest of the app writes through. + pub fn with_store(store: Store) -> Queue { Queue { - pool, + store, handlers: HashMap::new(), visibility_timeout: Duration::from_secs(30), poll_interval: Duration::from_secs(1), retry_backoff: Duration::from_secs(5), + wakeup: Arc::new((Mutex::new(0), Condvar::new())), } } @@ -93,55 +193,85 @@ impl Queue { pub fn register( &mut self, name: impl Into, - handler: impl Fn(&Json) -> Result<(), String> + Send + Sync + 'static, + handler: impl Fn(&JobCtx) -> Result<(), String> + Send + Sync + 'static, ) { self.handlers.insert(name.into(), Arc::new(handler)); } - /// The underlying connection pool, for sharing or advanced use. - pub fn pool(&self) -> &Pool { - &self.pool + /// The underlying store, for sharing or advanced use. + pub fn store(&self) -> &Store { + &self.store + } + + /// Whether claims are exclusive across *pods* (Postgres `SKIP LOCKED`) or + /// only within this database file's writer serialization (SQLite). + pub fn cross_pod(&self) -> bool { + self.store.capabilities().skip_locked } - /// Create the `sutegi_jobs` table if it does not already exist. + // --- schema ----------------------------------------------------------- + + /// Create the `sutegi_jobs` table and its indexes if they are missing. /// - /// Safe to call from every pod on boot: PostgreSQL can raise a spurious - /// unique-violation when several backends run `CREATE … IF NOT EXISTS` - /// against the catalog at the same instant, so that race is treated as - /// success (the table ends up created either way). + /// Safe to call from every pod on boot: Postgres can raise a spurious + /// unique violation when several backends run `CREATE … IF NOT EXISTS` + /// against the catalog at the same instant, so that race counts as success + /// (the table ends up created either way). pub fn migrate(&self) -> Result<(), String> { - match self.create_schema() { - Ok(()) => Ok(()), - Err(e) if e.contains("23505") || e.contains("already exists") => Ok(()), - Err(e) => Err(e), + for stmt in self.schema_sql() { + match self.store.execute(&stmt, &[]) { + Ok(_) => {} + Err(e) if is_already_exists(&e) => {} + Err(e) => return Err(e), + } } + Ok(()) } - fn create_schema(&self) -> Result<(), String> { - self.pool.batch( - "CREATE TABLE IF NOT EXISTS sutegi_jobs (\ - id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, \ - queue TEXT NOT NULL DEFAULT 'default', \ - name TEXT NOT NULL, \ - payload TEXT NOT NULL, \ - attempts INTEGER NOT NULL DEFAULT 0, \ - max_attempts INTEGER NOT NULL DEFAULT 1, \ - run_at TIMESTAMPTZ NOT NULL DEFAULT now(), \ - locked_at TIMESTAMPTZ, \ - failed_at TIMESTAMPTZ, \ - last_error TEXT, \ - created_at TIMESTAMPTZ NOT NULL DEFAULT now()); \ - CREATE INDEX IF NOT EXISTS sutegi_jobs_claim_idx \ - ON sutegi_jobs (run_at) WHERE failed_at IS NULL;", - ) + fn schema_sql(&self) -> Vec { + let pg = self.store.capabilities().backend == "postgres"; + let id = if pg { + "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY" + } else { + "id INTEGER PRIMARY KEY AUTOINCREMENT" + }; + vec![ + format!( + "CREATE TABLE IF NOT EXISTS sutegi_jobs ({id}, \ + queue TEXT NOT NULL DEFAULT 'default', \ + name TEXT NOT NULL, \ + payload TEXT NOT NULL, \ + priority INTEGER NOT NULL DEFAULT 0, \ + attempts INTEGER NOT NULL DEFAULT 0, \ + max_attempts INTEGER NOT NULL DEFAULT 1, \ + unique_key TEXT, \ + run_at BIGINT NOT NULL DEFAULT 0, \ + locked_at BIGINT, \ + failed_at BIGINT, \ + last_error TEXT, \ + created_at BIGINT NOT NULL DEFAULT 0)" + ), + // The claim's exact access path: one queue, live rows only, best + // priority then oldest schedule. + "CREATE INDEX IF NOT EXISTS sutegi_jobs_claim_idx ON sutegi_jobs \ + (queue, priority, run_at, id) WHERE failed_at IS NULL" + .into(), + // Dedupe is per queue and only among live rows: a dead-lettered + // job must not block re-dispatch of the same key. + "CREATE UNIQUE INDEX IF NOT EXISTS sutegi_jobs_unique_idx ON sutegi_jobs \ + (queue, unique_key) WHERE unique_key IS NOT NULL AND failed_at IS NULL" + .into(), + ] } + // --- dispatch --------------------------------------------------------- + /// Enqueue a job to run as soon as a worker is free. Returns its row id. pub fn dispatch(&self, name: &str, payload: Json) -> Result { - self.enqueue(name, payload, 1, Duration::ZERO) + self.job(name, payload).dispatch() } - /// Enqueue a job with a retry budget and an optional start delay. + /// Enqueue with a retry budget and an optional start delay. pub fn dispatch_with( &self, name: &str, @@ -149,142 +279,256 @@ impl Queue { max_attempts: u32, delay: Duration, ) -> Result { - self.enqueue(name, payload, max_attempts.max(1), delay) + self.job(name, payload) + .max_attempts(max_attempts) + .delay(delay) + .dispatch() } - fn enqueue( - &self, - name: &str, - payload: Json, - max_attempts: u32, - delay: Duration, - ) -> Result { - let rows = self.pool.query( - "INSERT INTO sutegi_jobs (name, payload, max_attempts, run_at) \ - VALUES ($1, $2, $3, now() + ($4::bigint * interval '1 millisecond')) \ - RETURNING id", - &[ - PgValue::Text(name.to_string()), - PgValue::Text(payload.to_string()), - PgValue::Int(max_attempts as i64), - PgValue::Int(delay.as_millis() as i64), - ], - )?; - Ok(rows + /// Start building a dispatch: queue, priority, retries, delay, dedupe key. + /// + /// ```no_run + /// # fn demo(queue: &sutegi_queue::Queue) -> Result<(), String> { + /// # use sutegi_json::Json; + /// queue + /// .job("video.ingest", Json::obj(vec![("id", Json::str("abc"))])) + /// .queue("video") // its own pool, so it can't starve the fast work + /// .unique("yt:abc") // at most one in flight per video + /// .max_attempts(3) + /// .dispatch()?; + /// # Ok(()) } + /// ``` + pub fn job(&self, name: &str, payload: Json) -> Dispatch<'_> { + Dispatch { + queue: self, + name: name.to_string(), + payload, + on: DEFAULT_QUEUE.to_string(), + priority: 0, + max_attempts: 1, + delay: Duration::ZERO, + unique_key: None, + } + } + + fn enqueue(&self, d: &Dispatch) -> Result { + let now = now_ms(); + let cols: Vec<(&str, Value)> = vec![ + ("queue", Value::Text(d.on.clone())), + ("name", Value::Text(d.name.clone())), + ("payload", Value::Text(d.payload.to_string())), + ("priority", Value::Int(d.priority as i64)), + ("max_attempts", Value::Int(d.max_attempts.max(1) as i64)), + ( + "unique_key", + match &d.unique_key { + Some(k) => Value::Text(k.clone()), + None => Value::Null, + }, + ), + ("run_at", Value::Int(now + d.delay.as_millis() as i64)), + ("created_at", Value::Int(now)), + ]; + match self.store.insert("sutegi_jobs", &cols, "id") { + Ok(id) => { + self.notify(); + Ok(id) + } + // A dedupe key already in flight is the *point* of the key, not an + // error: hand back the row that is already queued. + Err(e) if d.unique_key.is_some() && is_unique_violation(&e) => { + match self.find_unique(&d.on, d.unique_key.as_deref().unwrap_or(""))? { + Some(id) => Ok(id), + // Lost the race with a worker that just finished it — + // nothing is in flight, so enqueue for real. + None => { + let id = self.store.insert("sutegi_jobs", &cols, "id")?; + self.notify(); + Ok(id) + } + } + } + Err(e) => Err(e), + } + } + + fn find_unique(&self, queue: &str, key: &str) -> Result, String> { + Ok(self + .store + .query( + "SELECT id FROM sutegi_jobs WHERE queue = ? AND unique_key = ? \ + AND failed_at IS NULL LIMIT 1", + &[Value::Text(queue.into()), Value::Text(key.into())], + )? .first() - .and_then(|r| r.get("id").and_then(Json::as_i64)) - .unwrap_or(0)) + .and_then(|r| r.get("id").and_then(Json::as_i64))) } - /// Claim and run at most one ready job. Returns `true` if a job ran (so a - /// caller can keep draining), `false` if the queue was idle. + /// Wake idle workers; a local dispatch shouldn't wait out a poll interval. + fn notify(&self) { + let (lock, cv) = &*self.wakeup; + if let Ok(mut generation) = lock.lock() { + *generation += 1; + } + cv.notify_all(); + } + + // --- running ---------------------------------------------------------- + + /// Claim and run at most one ready job from the default queue. Returns + /// `true` if a job ran (so a caller can keep draining), `false` if the + /// queue was idle. pub fn run_once(&self) -> Result { - let claimed = self.pool.query( - "UPDATE sutegi_jobs SET locked_at = now(), attempts = attempts + 1 \ - WHERE id = (\ - SELECT id FROM sutegi_jobs \ - WHERE failed_at IS NULL AND run_at <= now() \ - AND (locked_at IS NULL OR locked_at < now() - ($1::bigint * interval '1 second')) \ - ORDER BY run_at, id \ - FOR UPDATE SKIP LOCKED \ - LIMIT 1) \ - RETURNING id, name, payload, attempts, max_attempts", - &[PgValue::Int(self.visibility_timeout.as_secs() as i64)], - )?; + self.run_once_on(DEFAULT_QUEUE, &AtomicBool::new(false)) + } - let Some(job) = claimed.into_iter().next() else { + /// [`run_once`](Queue::run_once) against a named queue. `stop` is what the + /// handler sees through [`JobCtx::should_stop`]. + pub fn run_once_on(&self, queue: &str, stop: &AtomicBool) -> Result { + let Some(job) = self.claim(queue)? else { return Ok(false); }; - let id = job.get("id").and_then(Json::as_i64).unwrap_or(0); - let name = job - .get("name") - .and_then(Json::as_str) - .unwrap_or("") - .to_string(); - let attempts = job.get("attempts").and_then(Json::as_i64).unwrap_or(1); - let max_attempts = job.get("max_attempts").and_then(Json::as_i64).unwrap_or(1); - let payload = job - .get("payload") - .and_then(Json::as_str) - .and_then(|s| Json::parse(s).ok()) - .unwrap_or(Json::Null); + let payload = Json::parse(&job.payload).unwrap_or(Json::Null); + let result = match self.handlers.get(&job.name) { + Some(handler) => { + let ctx = JobCtx { + id: job.id, + name: &job.name, + attempts: job.attempts, + max_attempts: job.max_attempts, + queue, + payload: &payload, + store: &self.store, + stop, + }; + // A panicking handler must not take the worker thread down, and + // must not vanish silently either: treat it as a failure so the + // row retries or dead-letters like any other. + match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| handler(&ctx))) { + Ok(r) => r, + Err(_) => Err(format!("handler for '{}' panicked", job.name)), + } + } + None => Err(format!("no handler registered for job '{}'", job.name)), + }; + self.settle(&job, result)?; + Ok(true) + } - let result = match self.handlers.get(&name) { - Some(handler) => handler(&payload), - None => Err(format!("no handler registered for job '{name}'")), + /// Claim the next ready job in `queue`, or `None` when there is nothing to + /// do. One statement, so the claim is atomic on both backends. + fn claim(&self, queue: &str) -> Result, String> { + let now = now_ms(); + let cutoff = now - self.visibility_timeout.as_millis() as i64; + // Postgres needs SKIP LOCKED so concurrent workers step over rows + // another backend is already claiming. SQLite serializes writers, so + // the second UPDATE simply no longer sees the row. + let skip = if self.store.capabilities().skip_locked { + " FOR UPDATE SKIP LOCKED" + } else { + "" + }; + let sql = format!( + "UPDATE sutegi_jobs SET locked_at = ?, attempts = attempts + 1 \ + WHERE id = (SELECT id FROM sutegi_jobs \ + WHERE failed_at IS NULL AND queue = ? AND run_at <= ? \ + AND (locked_at IS NULL OR locked_at < ?) \ + ORDER BY priority DESC, run_at, id LIMIT 1{skip}) \ + RETURNING id, name, payload, attempts, max_attempts" + ); + let rows = self.store.query( + &sql, + &[ + Value::Int(now), + Value::Text(queue.to_string()), + Value::Int(now), + Value::Int(cutoff), + ], + )?; + let Some(row) = rows.into_iter().next() else { + return Ok(None); }; + let text = |k: &str| { + row.get(k) + .and_then(Json::as_str) + .unwrap_or_default() + .to_string() + }; + let int = |k: &str, d: i64| row.get(k).and_then(Json::as_i64).unwrap_or(d); + Ok(Some(Claimed { + id: int("id", 0), + name: text("name"), + payload: text("payload"), + attempts: int("attempts", 1), + max_attempts: int("max_attempts", 1), + })) + } + /// Land a finished attempt: delete on success, retry with backoff while the + /// budget lasts, dead-letter when it runs out. + fn settle(&self, job: &Claimed, result: Result<(), String>) -> Result<(), String> { match result { - Ok(()) => { - self.pool - .execute("DELETE FROM sutegi_jobs WHERE id = $1", &[PgValue::Int(id)])?; + Ok(()) => self + .store + .execute( + "DELETE FROM sutegi_jobs WHERE id = ?", + &[Value::Int(job.id)], + ) + .map(|_| ()), + Err(err) if job.attempts >= job.max_attempts => { + eprintln!( + "[queue] job '{}' #{} failed terminally: {err}", + job.name, job.id + ); + self.store + .execute( + "UPDATE sutegi_jobs SET failed_at = ?, locked_at = NULL, \ + last_error = ? WHERE id = ?", + &[Value::Int(now_ms()), Value::Text(err), Value::Int(job.id)], + ) + .map(|_| ()) } Err(err) => { - if attempts >= max_attempts { - // Terminal: keep the row as a dead-letter record. - self.pool.execute( - "UPDATE sutegi_jobs SET failed_at = now(), locked_at = NULL, last_error = $2 \ - WHERE id = $1", - &[PgValue::Int(id), PgValue::Text(err.clone())], - )?; - eprintln!("[queue] job '{name}' #{id} failed terminally: {err}"); - } else { - // Retry: release the lock and schedule a backed-off retry. - let backoff_secs = self.retry_backoff.as_secs() as i64 * attempts; - self.pool.execute( - "UPDATE sutegi_jobs SET locked_at = NULL, last_error = $2, \ - run_at = now() + ($3::bigint * interval '1 second') WHERE id = $1", + let backoff = self.retry_backoff.as_millis() as i64 * job.attempts.max(1); + self.store + .execute( + "UPDATE sutegi_jobs SET locked_at = NULL, last_error = ?, \ + run_at = ? WHERE id = ?", &[ - PgValue::Int(id), - PgValue::Text(err), - PgValue::Int(backoff_secs), + Value::Text(err), + Value::Int(now_ms() + backoff), + Value::Int(job.id), ], - )?; - } + ) + .map(|_| ()) } } - Ok(true) } - /// A snapshot of queue depth by state, as JSON — wire it into `/__queue`. - pub fn stats(&self) -> Result { - let rows = self.pool.query( - "SELECT \ - count(*) FILTER (WHERE failed_at IS NULL AND run_at <= now() AND locked_at IS NULL) AS ready, \ - count(*) FILTER (WHERE failed_at IS NULL AND locked_at IS NOT NULL) AS running, \ - count(*) FILTER (WHERE failed_at IS NULL AND run_at > now()) AS scheduled, \ - count(*) FILTER (WHERE failed_at IS NOT NULL) AS failed, \ - count(*) AS total \ - FROM sutegi_jobs", - &[], - )?; - let row = rows.into_iter().next().unwrap_or(Json::Null); - let n = |k: &str| Json::int(row.get(k).and_then(Json::as_i64).unwrap_or(0)); - Ok(Json::obj(vec![ - ("ready", n("ready")), - ("running", n("running")), - ("scheduled", n("scheduled")), - ("failed", n("failed")), - ("total", n("total")), - ])) + /// Spawn `workers` threads that drain the default queue until the returned + /// [`Workers`] handle is dropped (or `stop()` is called). + pub fn start(self: Arc, workers: usize) -> Workers { + self.start_on(DEFAULT_QUEUE, workers) } - /// Spawn `workers` background threads that poll and process jobs until the - /// returned [`Workers`] handle is dropped (or `stop()` is called). - pub fn start(self: Arc, workers: usize) -> Workers { + /// [`start`](Queue::start) against a named queue. Separate pools are how a + /// slow job class (transcoding a video) is kept from starving a fast one + /// (fetching a page): give each its own queue and its own worker count. + pub fn start_on(self: Arc, queue: &str, workers: usize) -> Workers { let stop = Arc::new(AtomicBool::new(false)); let mut handles = Vec::new(); for _ in 0..workers.max(1) { - let queue = Arc::clone(&self); + let queue_name = queue.to_string(); + let q = Arc::clone(&self); let stop = Arc::clone(&stop); handles.push(thread::spawn(move || { while !stop.load(Ordering::Relaxed) { - match queue.run_once() { + match q.run_once_on(&queue_name, &stop) { Ok(true) => continue, // keep draining while there's work - Ok(false) => thread::sleep(queue.poll_interval), + Ok(false) => q.wait_for_work(), Err(e) => { eprintln!("[queue] worker error: {e}"); - thread::sleep(queue.poll_interval); + thread::sleep(q.poll_interval); } } } @@ -292,10 +536,163 @@ impl Queue { } Workers { stop, handles } } + + /// Sleep until a local dispatch wakes us or the poll interval expires — + /// whichever comes first. The timeout is what covers delayed jobs and work + /// enqueued by another pod, which no local notify can announce. + fn wait_for_work(&self) { + let (lock, cv) = &*self.wakeup; + if let Ok(generation) = lock.lock() { + let _ = cv.wait_timeout(generation, self.poll_interval); + } + } + + // --- introspection & ops --------------------------------------------- + + /// Queue depth by state, as JSON — wire it into an ops endpoint. + pub fn stats(&self) -> Result { + self.stats_where("", &[]) + } + + /// [`stats`](Queue::stats) for one named queue. + pub fn stats_for(&self, queue: &str) -> Result { + self.stats_where(" WHERE queue = ?", &[Value::Text(queue.to_string())]) + } + + fn stats_where(&self, filter: &str, params: &[Value]) -> Result { + // `count(*) FILTER (WHERE …)` is not portable; CASE is. `now` is + // interpolated rather than bound so the placeholder order stays + // independent of the optional filter. + let now = now_ms(); + let sql = format!( + "SELECT \ + SUM(CASE WHEN failed_at IS NULL AND run_at <= {now} AND locked_at IS NULL \ + THEN 1 ELSE 0 END) AS ready, \ + SUM(CASE WHEN failed_at IS NULL AND locked_at IS NOT NULL THEN 1 ELSE 0 END) \ + AS running, \ + SUM(CASE WHEN failed_at IS NULL AND run_at > {now} THEN 1 ELSE 0 END) \ + AS scheduled, \ + SUM(CASE WHEN failed_at IS NOT NULL THEN 1 ELSE 0 END) AS failed, \ + COUNT(*) AS total \ + FROM sutegi_jobs{filter}" + ); + let row = self + .store + .query(&sql, params)? + .into_iter() + .next() + .unwrap_or(Json::Null); + let n = |k: &str| Json::int(row.get(k).and_then(Json::as_i64).unwrap_or(0)); + Ok(Json::obj(vec![ + ("ready", n("ready")), + ("running", n("running")), + ("scheduled", n("scheduled")), + ("failed", n("failed")), + ("total", n("total")), + ])) + } + + /// Dead-lettered jobs, newest first — the rows a dev screen should show. + pub fn failed(&self, limit: i64) -> Result, String> { + self.store.query( + "SELECT id, queue, name, payload, attempts, max_attempts, last_error, failed_at \ + FROM sutegi_jobs WHERE failed_at IS NOT NULL ORDER BY failed_at DESC LIMIT ?", + &[Value::Int(limit.max(1))], + ) + } + + /// Put a dead-lettered job back in the queue with a fresh attempt budget. + /// Returns whether there was such a job to revive. + pub fn retry(&self, id: i64, max_attempts: u32) -> Result { + let n = self.store.execute( + "UPDATE sutegi_jobs SET failed_at = NULL, locked_at = NULL, attempts = 0, \ + max_attempts = ?, run_at = ?, last_error = NULL \ + WHERE id = ? AND failed_at IS NOT NULL", + &[ + Value::Int(max_attempts.max(1) as i64), + Value::Int(now_ms()), + Value::Int(id), + ], + )?; + if n > 0 { + self.notify(); + } + Ok(n > 0) + } + + /// Drop dead-letter rows that failed at least `age` ago. Returns how many + /// were removed. The bound is inclusive, so `purge_failed(Duration::ZERO)` + /// clears the lot — including a row stamped this same millisecond. + pub fn purge_failed(&self, age: Duration) -> Result { + self.store.execute( + "DELETE FROM sutegi_jobs WHERE failed_at IS NOT NULL AND failed_at <= ?", + &[Value::Int(now_ms() - age.as_millis() as i64)], + ) + } +} + +/// A dispatch under construction — see [`Queue::job`]. +pub struct Dispatch<'q> { + queue: &'q Queue, + name: String, + payload: Json, + on: String, + priority: i32, + max_attempts: u32, + delay: Duration, + unique_key: Option, +} + +impl Dispatch<'_> { + /// Put the job on a named queue (default `"default"`). + pub fn queue(mut self, name: &str) -> Self { + self.on = name.to_string(); + self + } + + /// Higher runs first within a queue. Equal priorities run oldest-first. + pub fn priority(mut self, p: i32) -> Self { + self.priority = p; + self + } + + /// How many attempts the job gets before it is dead-lettered. + pub fn max_attempts(mut self, n: u32) -> Self { + self.max_attempts = n; + self + } + + /// Hold the job back for `d` before it becomes claimable. + pub fn delay(mut self, d: Duration) -> Self { + self.delay = d; + self + } + + /// Collapse duplicates: while a job with this key is live on the queue, a + /// second dispatch returns the existing row's id instead of enqueueing + /// again. Dead-lettered rows don't block a fresh dispatch. + pub fn unique(mut self, key: &str) -> Self { + self.unique_key = Some(key.to_string()); + self + } + + /// Enqueue it. Returns the row id. + pub fn dispatch(self) -> Result { + self.queue.enqueue(&self) + } +} + +/// A row this worker has claimed. +struct Claimed { + id: i64, + name: String, + payload: String, + attempts: i64, + max_attempts: i64, } -/// A running set of queue workers. Dropping it (or calling [`stop`] -/// (Workers::stop)) signals shutdown and joins the threads. +/// A running set of queue workers. Dropping it (or calling +/// [`stop`](Workers::stop)) signals shutdown and joins the threads. pub struct Workers { stop: Arc, handles: Vec>, @@ -307,6 +704,12 @@ impl Workers { self.shutdown(); } + /// Ask the workers to stop without waiting — this is the flag handlers see + /// through [`JobCtx::should_stop`]. Joining still happens on drop. + pub fn signal_stop(&self) { + self.stop.store(true, Ordering::Relaxed); + } + fn shutdown(&mut self) { self.stop.store(true, Ordering::Relaxed); for h in self.handles.drain(..) { @@ -320,3 +723,44 @@ impl Drop for Workers { self.shutdown(); } } + +/// `CREATE … IF NOT EXISTS` racing itself across pods, in either dialect. +fn is_already_exists(e: &str) -> bool { + let e = e.to_ascii_lowercase(); + e.contains("already exists") || e.contains("23505") || e.contains("duplicate key") +} + +/// A unique-index violation, in either dialect. +fn is_unique_violation(e: &str) -> bool { + let e = e.to_ascii_lowercase(); + e.contains("unique constraint") // SQLite + || e.contains("23505") // Postgres SQLSTATE + || e.contains("duplicate key") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn dialect_error_shapes_are_both_recognised() { + assert!(is_unique_violation( + "UNIQUE constraint failed: sutegi_jobs.unique_key" + )); + assert!(is_unique_violation( + "ERROR 23505: duplicate key value violates unique constraint" + )); + assert!(!is_unique_violation("no such table: sutegi_jobs")); + assert!(is_already_exists("relation \"sutegi_jobs\" already exists")); + assert!(!is_already_exists("syntax error at or near \"CREATE\"")); + } + + #[test] + fn now_ms_is_a_plausible_wall_clock() { + // Sanity: past 2020, not in the far future — catches a seconds/millis + // mixup, which would silently make every delay 1000× wrong. + let now = now_ms(); + assert!(now > 1_577_836_800_000, "{now} is before 2020"); + assert!(now < 4_102_444_800_000, "{now} is after 2100"); + } +} diff --git a/crates/sutegi-queue/tests/durable.rs b/crates/sutegi-queue/tests/durable.rs index 07c6230..c09eea0 100644 --- a/crates/sutegi-queue/tests/durable.rs +++ b/crates/sutegi-queue/tests/durable.rs @@ -1,21 +1,27 @@ -//! Live integration test for the durable PostgreSQL-backed queue. Runs only -//! when `SUTEGI_PG_TEST_URL` is set. +//! The Postgres leg of the queue contract — the same behaviour `sqlite.rs` +//! pins, but claimed with `FOR UPDATE SKIP LOCKED` across pods. +//! +//! Needs a live server: `cargo test -p sutegi-queue --features postgres` with +//! `SUTEGI_PG_TEST_URL` set. Without the feature the file compiles to nothing, +//! so the default `cargo test` stays dependency-free. +#![cfg(feature = "postgres")] use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use sutegi_json::Json; -use sutegi_pg::Pool; +use sutegi_orm::Backend; +use sutegi_orm::pg::Pg; use sutegi_queue::Queue; // Both tests share one `sutegi_jobs` table, so they must not run concurrently // (one's DROP would nuke the other's rows). Serialize them. static DB_LOCK: Mutex<()> = Mutex::new(()); -fn pool() -> Option { +fn store() -> Option { let url = std::env::var("SUTEGI_PG_TEST_URL").ok()?; - Some(Pool::new(sutegi_pg::Config::from_url(&url).unwrap(), 8)) + Pg::connect(&url, 8).ok() } fn wait_until(cond: impl Fn() -> bool) -> bool { @@ -31,28 +37,28 @@ fn wait_until(cond: impl Fn() -> bool) -> bool { #[test] fn dispatch_process_and_retry() { let _guard = DB_LOCK.lock().unwrap_or_else(|e| e.into_inner()); - let Some(pool) = pool() else { + let Some(pg) = store() else { eprintln!("skipping: SUTEGI_PG_TEST_URL not set"); return; }; - pool.batch("DROP TABLE IF EXISTS sutegi_jobs").unwrap(); + pg.execute("DROP TABLE IF EXISTS sutegi_jobs", &[]).unwrap(); let processed = Arc::new(Mutex::new(Vec::::new())); let fail_first = Arc::new(AtomicU32::new(0)); - let mut queue = Queue::new(pool.clone()) + let mut queue = Queue::new(pg.clone()) .poll_interval(Duration::from_millis(20)) .retry_backoff(Duration::from_millis(1)); // tiny backoff so the test is quick let seen = Arc::clone(&processed); - queue.register("greet", move |payload: &Json| { - let who = payload.get("who").and_then(Json::as_str).unwrap_or("?"); + queue.register("greet", move |job| { + let who = job.payload().get("who").and_then(Json::as_str).unwrap_or("?"); seen.lock().unwrap().push(who.to_string()); Ok(()) }); let counter = Arc::clone(&fail_first); - queue.register("flaky", move |_payload: &Json| { + queue.register("flaky", move |_job| { // Fail on the first attempt, succeed on the second. if counter.fetch_add(1, Ordering::SeqCst) == 0 { Err("transient".into()) @@ -62,23 +68,21 @@ fn dispatch_process_and_retry() { }); queue.migrate().unwrap(); + assert!( + queue.cross_pod(), + "Postgres claims must be cluster-scoped (SKIP LOCKED)" + ); queue .dispatch("greet", Json::obj(vec![("who", Json::str("world"))])) .unwrap(); queue - .dispatch_with( - "flaky", - Json::Null, - 3, // up to 3 attempts - Duration::ZERO, - ) + .dispatch_with("flaky", Json::Null, 3, Duration::ZERO) .unwrap(); let queue = Arc::new(queue); let workers = Arc::clone(&queue).start(2); - // The greet job runs once; the flaky job fails then succeeds on retry. assert!( wait_until(|| processed.lock().unwrap().contains(&"world".to_string())), "greet should have been processed" @@ -87,7 +91,6 @@ fn dispatch_process_and_retry() { wait_until(|| fail_first.load(Ordering::SeqCst) >= 2), "flaky should have been retried and then succeeded" ); - // Once both complete, the table drains to empty (no failed dead-letters). assert!( wait_until(|| { queue @@ -100,22 +103,22 @@ fn dispatch_process_and_retry() { ); workers.stop(); - pool.batch("DROP TABLE sutegi_jobs").unwrap(); + queue.store().execute("DROP TABLE sutegi_jobs", &[]).unwrap(); } #[test] fn terminal_failure_becomes_dead_letter() { let _guard = DB_LOCK.lock().unwrap_or_else(|e| e.into_inner()); - let Some(pool) = pool() else { + let Some(pg) = store() else { eprintln!("skipping: SUTEGI_PG_TEST_URL not set"); return; }; - pool.batch("DROP TABLE IF EXISTS sutegi_jobs").unwrap(); + pg.execute("DROP TABLE IF EXISTS sutegi_jobs", &[]).unwrap(); - let mut queue = Queue::new(pool.clone()) + let mut queue = Queue::new(pg.clone()) .poll_interval(Duration::from_millis(20)) .retry_backoff(Duration::from_millis(1)); - queue.register("always_fails", |_: &Json| Err("nope".into())); + queue.register("always_fails", |_job| Err("nope".into())); queue.migrate().unwrap(); queue .dispatch_with("always_fails", Json::Null, 2, Duration::ZERO) @@ -124,7 +127,6 @@ fn terminal_failure_becomes_dead_letter() { let queue = Arc::new(queue); let workers = Arc::clone(&queue).start(1); - // After exhausting 2 attempts the job is kept as a failed dead-letter. assert!( wait_until(|| { queue @@ -137,5 +139,40 @@ fn terminal_failure_becomes_dead_letter() { ); workers.stop(); - pool.batch("DROP TABLE sutegi_jobs").unwrap(); + pg.execute("DROP TABLE sutegi_jobs", &[]).unwrap(); +} + +#[test] +fn dedupe_keys_and_named_queues_behave_as_on_sqlite() { + let _guard = DB_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let Some(pg) = store() else { + eprintln!("skipping: SUTEGI_PG_TEST_URL not set"); + return; + }; + pg.execute("DROP TABLE IF EXISTS sutegi_jobs", &[]).unwrap(); + + let mut queue = Queue::new(pg.clone()); + queue.register("ingest", |_job| Ok(())); + queue.migrate().unwrap(); + + let first = queue + .job("ingest", Json::Null) + .queue("video") + .unique("yt:abc") + .dispatch() + .unwrap(); + let second = queue + .job("ingest", Json::Null) + .queue("video") + .unique("yt:abc") + .dispatch() + .unwrap(); + assert_eq!(first, second, "the live row is returned, not a duplicate"); + assert!(!queue.run_once().unwrap(), "default queue is empty"); + assert_eq!( + queue.stats_for("video").unwrap().get("ready").and_then(Json::as_i64), + Some(1) + ); + + pg.execute("DROP TABLE sutegi_jobs", &[]).unwrap(); } diff --git a/crates/sutegi-queue/tests/sqlite.rs b/crates/sutegi-queue/tests/sqlite.rs new file mode 100644 index 0000000..fa53030 --- /dev/null +++ b/crates/sutegi-queue/tests/sqlite.rs @@ -0,0 +1,444 @@ +//! The queue's behaviour, exercised end to end on the bundled SQLite backend — +//! no server, no environment variables, so this runs everywhere `cargo test` +//! does. The Postgres leg of the same contract lives in `durable.rs`. + +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use sutegi_json::Json; +use sutegi_orm::Backend; +use sutegi_orm::db::Db; +use sutegi_orm::Value; +use sutegi_queue::{now_ms, Queue}; + +/// A fresh database file per test. WAL is what production runs, and an +/// in-memory database would not be shared across the pool's connections. +struct TempDb { + path: String, +} + +impl TempDb { + fn new(tag: &str) -> TempDb { + let path = std::env::temp_dir() + .join(format!("sutegi-queue-{tag}-{}.db", std::process::id())) + .to_string_lossy() + .into_owned(); + for suffix in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{path}{suffix}")); + } + TempDb { path } + } + + fn open(&self) -> Db { + Db::open(&self.path).expect("open db") + } +} + +impl Drop for TempDb { + fn drop(&mut self) { + for suffix in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{}{suffix}", self.path)); + } + } +} + +fn queue(tag: &str) -> (TempDb, Db, Queue) { + let tmp = TempDb::new(tag); + let db = tmp.open(); + let q = Queue::new(db.clone()) + .poll_interval(Duration::from_millis(10)) + .retry_backoff(Duration::from_millis(1)); + q.migrate().expect("migrate"); + (tmp, db, q) +} + +fn wait_until(cond: impl Fn() -> bool) -> bool { + for _ in 0..400 { + if cond() { + return true; + } + std::thread::sleep(Duration::from_millis(10)); + } + cond() +} + +fn count(db: &Db, sql: &str) -> i64 { + db.query(sql, &[]) + .expect("count") + .first() + .and_then(|r| r.get("n").and_then(Json::as_i64)) + .unwrap_or(-1) +} + +#[test] +fn migrate_is_idempotent_and_creates_the_table() { + let (_tmp, db, q) = queue("migrate"); + q.migrate().expect("second migrate is a no-op, not an error"); + assert_eq!(count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs"), 0); +} + +#[test] +fn a_dispatched_job_runs_with_its_payload_and_leaves_no_row() { + let (_tmp, db, mut q) = queue("dispatch"); + let seen = Arc::new(Mutex::new(Vec::::new())); + let sink = Arc::clone(&seen); + q.register("greet", move |job| { + let who = job.payload().get("who").and_then(Json::as_str).unwrap_or("?"); + sink.lock().unwrap().push(who.to_string()); + Ok(()) + }); + + q.dispatch("greet", Json::obj(vec![("who", Json::str("eneko"))])) + .expect("dispatch"); + assert!(q.run_once().expect("run"), "a job was waiting"); + assert!(!q.run_once().expect("run"), "queue is drained"); + + assert_eq!(seen.lock().unwrap().as_slice(), ["eneko"]); + // Success deletes the row — the queue is not a history table. + assert_eq!(count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs"), 0); +} + +#[test] +fn a_failing_job_retries_until_its_budget_runs_out_then_dead_letters() { + let (_tmp, db, mut q) = queue("retry"); + let attempts = Arc::new(AtomicU32::new(0)); + let counter = Arc::clone(&attempts); + q.register("flaky", move |job| { + // Fail the first attempt, succeed on the second. + if counter.fetch_add(1, Ordering::SeqCst) == 0 { + assert!(!job.is_last_attempt(), "attempt 1 of 2 is not terminal"); + Err("transient".into()) + } else { + Ok(()) + } + }); + q.register("doomed", |_job| Err("always".into())); + + q.job("flaky", Json::Null) + .max_attempts(2) + .dispatch() + .expect("dispatch"); + q.run_once().expect("attempt 1"); + assert_eq!( + count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs WHERE failed_at IS NULL"), + 1, + "still queued for a retry" + ); + // The retry is scheduled in the future; the backoff here is 1ms. + assert!(wait_until(|| q.run_once().unwrap_or(false)), "retry ran"); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert_eq!(count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs"), 0); + + // Terminal failure keeps the row as a dead letter, with the reason. + let id = q.job("doomed", Json::Null).dispatch().expect("dispatch"); + q.run_once().expect("attempt"); + let failed = q.failed(10).expect("failed list"); + assert_eq!(failed.len(), 1); + assert_eq!(failed[0].get("id").and_then(Json::as_i64), Some(id)); + assert_eq!( + failed[0].get("last_error").and_then(Json::as_str), + Some("always") + ); + assert!(!q.run_once().expect("run"), "a dead letter is not claimable"); + + // …and can be put back by hand. + assert!(q.retry(id, 1).expect("retry")); + assert!(q.run_once().expect("run"), "revived job ran again"); + assert!(!q.retry(id + 999, 1).expect("retry"), "no such job"); +} + +#[test] +fn a_panicking_handler_dead_letters_instead_of_taking_the_worker() { + let (_tmp, _db, mut q) = queue("panic"); + q.register("boom", |_job| panic!("handler exploded")); + q.register("ok", |_job| Ok(())); + + q.dispatch("boom", Json::Null).expect("dispatch"); + q.run_once().expect("the panic is caught, not propagated"); + let failed = q.failed(10).expect("failed"); + assert_eq!(failed.len(), 1); + assert!(failed[0] + .get("last_error") + .and_then(Json::as_str) + .unwrap_or_default() + .contains("panicked")); + + // The queue still works afterwards. + q.dispatch("ok", Json::Null).expect("dispatch"); + assert!(q.run_once().expect("run")); +} + +#[test] +fn an_unregistered_name_is_a_failure_not_a_silent_drop() { + let (_tmp, _db, q) = queue("unregistered"); + q.dispatch("nobody.handles.this", Json::Null) + .expect("dispatch"); + q.run_once().expect("run"); + let failed = q.failed(10).expect("failed"); + assert_eq!(failed.len(), 1); + assert!(failed[0] + .get("last_error") + .and_then(Json::as_str) + .unwrap_or_default() + .contains("no handler registered")); +} + +#[test] +fn a_delayed_job_is_not_claimable_before_its_time() { + let (_tmp, _db, mut q) = queue("delay"); + q.register("later", |_job| Ok(())); + q.job("later", Json::Null) + .delay(Duration::from_secs(60)) + .dispatch() + .expect("dispatch"); + assert!(!q.run_once().expect("run"), "scheduled for the future"); + + let stats = q.stats().expect("stats"); + assert_eq!(stats.get("scheduled").and_then(Json::as_i64), Some(1)); + assert_eq!(stats.get("ready").and_then(Json::as_i64), Some(0)); +} + +#[test] +fn priority_wins_over_arrival_order() { + let (_tmp, _db, mut q) = queue("priority"); + let order = Arc::new(Mutex::new(Vec::::new())); + let sink = Arc::clone(&order); + q.register("task", move |job| { + let tag = job.payload().get("tag").and_then(Json::as_str).unwrap_or(""); + sink.lock().unwrap().push(tag.to_string()); + Ok(()) + }); + + for (tag, prio) in [("low", 0), ("urgent", 10), ("mid", 5)] { + q.job("task", Json::obj(vec![("tag", Json::str(tag))])) + .priority(prio) + .dispatch() + .expect("dispatch"); + } + while q.run_once().expect("run") {} + assert_eq!(order.lock().unwrap().as_slice(), ["urgent", "mid", "low"]); +} + +#[test] +fn named_queues_do_not_see_each_others_work() { + let (_tmp, _db, mut q) = queue("queues"); + let ran = Arc::new(Mutex::new(Vec::::new())); + let sink = Arc::clone(&ran); + q.register("job", move |job| { + sink.lock().unwrap().push(job.queue.to_string()); + Ok(()) + }); + + q.job("job", Json::Null) + .queue("video") + .dispatch() + .expect("dispatch"); + q.dispatch("job", Json::Null).expect("dispatch"); + + // The default pool must not drain the video queue… + assert!(q.run_once().expect("run")); + assert!(!q.run_once().expect("run")); + assert_eq!(ran.lock().unwrap().as_slice(), ["default"]); + + // …and the video queue still has its own row. + let stop = AtomicBool::new(false); + assert!(q.run_once_on("video", &stop).expect("run")); + assert_eq!(ran.lock().unwrap().as_slice(), ["default", "video"]); + assert_eq!(q.stats_for("video").unwrap().get("total").and_then(Json::as_i64), Some(0)); +} + +#[test] +fn a_dedupe_key_collapses_duplicates_while_one_is_live() { + let (_tmp, db, mut q) = queue("dedupe"); + q.register("ingest", |_job| Ok(())); + + let first = q + .job("ingest", Json::Null) + .unique("yt:abc") + .dispatch() + .expect("dispatch"); + let second = q + .job("ingest", Json::Null) + .unique("yt:abc") + .dispatch() + .expect("dispatch is not an error — it returns the live row"); + assert_eq!(first, second, "same key while in flight = same job"); + assert_eq!(count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs"), 1); + + // A different key is a different job. + let other = q + .job("ingest", Json::Null) + .unique("yt:xyz") + .dispatch() + .expect("dispatch"); + assert_ne!(first, other); + + // Once the work is done the key is free again. + while q.run_once().expect("run") {} + let third = q + .job("ingest", Json::Null) + .unique("yt:abc") + .dispatch() + .expect("dispatch"); + assert_ne!(third, first, "a finished job does not block the next one"); +} + +#[test] +fn a_dead_lettered_key_does_not_block_a_fresh_dispatch() { + let (_tmp, _db, mut q) = queue("dedupe-failed"); + q.register("ingest", |_job| Err("nope".into())); + let first = q + .job("ingest", Json::Null) + .unique("yt:abc") + .dispatch() + .expect("dispatch"); + q.run_once().expect("run"); + assert_eq!(q.failed(10).unwrap().len(), 1); + + let again = q + .job("ingest", Json::Null) + .unique("yt:abc") + .dispatch() + .expect("the dead letter must not own the key forever"); + assert_ne!(again, first); +} + +#[test] +fn a_job_whose_worker_died_is_reclaimed_after_the_visibility_timeout() { + let tmp = TempDb::new("crash"); + let db = tmp.open(); + let mut q = Queue::new(db.clone()).visibility_timeout(Duration::from_millis(50)); + q.register("resume", |_job| Ok(())); + q.migrate().expect("migrate"); + + let id = q.dispatch("resume", Json::Null).expect("dispatch"); + // Simulate a worker that claimed the row and then died: locked, never + // settled. + db.execute( + "UPDATE sutegi_jobs SET locked_at = ?, attempts = 1 WHERE id = ?", + &[Value::Int(now_ms()), Value::Int(id)], + ) + .expect("lock"); + assert!(!q.run_once().expect("run"), "still inside the lease"); + + std::thread::sleep(Duration::from_millis(70)); + assert!(q.run_once().expect("run"), "lease expired, work resumed"); + assert_eq!(count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs"), 0); +} + +#[test] +fn a_heartbeat_keeps_a_long_job_from_being_stolen() { + let tmp = TempDb::new("heartbeat"); + let db = tmp.open(); + let mut q = Queue::new(db.clone()).visibility_timeout(Duration::from_millis(50)); + q.register("slow", |job| { + // Outlive the visibility timeout, but say so while doing it. + for _ in 0..4 { + std::thread::sleep(Duration::from_millis(20)); + job.heartbeat()?; + assert!(!job.should_stop()); + } + Ok(()) + }); + q.migrate().expect("migrate"); + q.dispatch("slow", Json::Null).expect("dispatch"); + + let q = Arc::new(q); + let runner = Arc::clone(&q); + let worker = std::thread::spawn(move || runner.run_once().expect("run")); + // While the handler heartbeats, a second worker must find nothing. + std::thread::sleep(Duration::from_millis(60)); + assert!(!q.run_once().expect("run"), "lease was kept alive"); + assert!(worker.join().expect("join")); + assert_eq!(count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs"), 0); +} + +#[test] +fn concurrent_workers_run_each_job_exactly_once() { + let (_tmp, db, mut q) = queue("concurrency"); + let done = Arc::new(Mutex::new(Vec::::new())); + let sink = Arc::clone(&done); + q.register("work", move |job| { + let n = job.payload().get("n").and_then(Json::as_i64).unwrap_or(-1); + // Long enough that workers genuinely overlap. + std::thread::sleep(Duration::from_millis(5)); + sink.lock().unwrap().push(n); + Ok(()) + }); + + const JOBS: i64 = 40; + for n in 0..JOBS { + q.dispatch("work", Json::obj(vec![("n", Json::int(n))])) + .expect("dispatch"); + } + + let q = Arc::new(q); + let workers = Arc::clone(&q).start(6); + assert!( + wait_until(|| count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs") == 0), + "queue drained" + ); + workers.stop(); + + let mut seen = done.lock().unwrap().clone(); + seen.sort_unstable(); + assert_eq!(seen.len(), JOBS as usize, "every job ran exactly once: {seen:?}"); + assert_eq!(seen, (0..JOBS).collect::>()); +} + +#[test] +fn a_dispatch_wakes_an_idle_worker_without_waiting_out_the_poll_interval() { + let tmp = TempDb::new("wakeup"); + let db = tmp.open(); + // A poll interval far longer than the assertion window: if the job runs at + // all, it is because the dispatch notified the sleeping worker. + let mut q = Queue::new(db.clone()).poll_interval(Duration::from_secs(30)); + q.register("ping", |_job| Ok(())); + q.migrate().expect("migrate"); + + let q = Arc::new(q); + let workers = Arc::clone(&q).start(1); + std::thread::sleep(Duration::from_millis(50)); // let the worker reach its wait + + q.dispatch("ping", Json::Null).expect("dispatch"); + let drained = wait_until(|| count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs") == 0); + workers.stop(); + assert!(drained, "the worker woke on dispatch, not on the 30s poll"); +} + +#[test] +fn stats_and_purge_report_the_states_an_operator_asks_about() { + let (_tmp, _db, mut q) = queue("stats"); + q.register("ok", |_job| Ok(())); + q.register("bad", |_job| Err("nope".into())); + + q.dispatch("ok", Json::Null).expect("dispatch"); + q.job("ok", Json::Null) + .delay(Duration::from_secs(300)) + .dispatch() + .expect("dispatch"); + q.dispatch("bad", Json::Null).expect("dispatch"); + while q.run_once().expect("run") {} + + let stats = q.stats().expect("stats"); + let n = |k: &str| stats.get(k).and_then(Json::as_i64).unwrap_or(-1); + assert_eq!(n("ready"), 0); + assert_eq!(n("running"), 0); + assert_eq!(n("scheduled"), 1, "the delayed job"); + assert_eq!(n("failed"), 1, "the dead letter"); + assert_eq!(n("total"), 2); + + assert_eq!(q.purge_failed(Duration::from_secs(3600)).expect("purge"), 0); + assert_eq!(q.purge_failed(Duration::ZERO).expect("purge"), 1); + assert_eq!(q.stats().unwrap().get("failed").and_then(Json::as_i64), Some(0)); +} + +#[test] +fn sqlite_claims_are_process_scoped_and_say_so() { + let (_tmp, _db, q) = queue("caps"); + assert!( + !q.cross_pod(), + "SQLite relies on serialized writers, not SKIP LOCKED — the capability must not overclaim" + ); +} diff --git a/crates/sutegi/Cargo.toml b/crates/sutegi/Cargo.toml index 8a80209..3e8d2f2 100644 --- a/crates/sutegi/Cargo.toml +++ b/crates/sutegi/Cargo.toml @@ -38,14 +38,14 @@ sutegi-actors = { path = "../sutegi-actors", version = "0.8.0", optional = true [features] # Batteries-included by default; opt out with `default-features = false` and # add back only what you need to shrink the binary. -# `queue` is NOT default: the durable queue needs a PostgreSQL database, so it -# is opt-in rather than force-compiled into every minimal build. +# `queue` is NOT default: not every app has background work, and the jobs +# table it creates should be an explicit choice. default = ["derive", "orm", "validate"] # Each pillar is an independent compile-time feature. orm = ["dep:sutegi-orm", "sutegi-web/orm", "sutegi-repl?/orm"] # schema + query builder + Ctx::db/model validate = ["dep:sutegi-validate", "sutegi-web/validate"] # request / tool validation + Ctx::validate(d) -queue = ["dep:sutegi-queue"] # durable, cross-pod job queue (Postgres-backed) +queue = ["dep:sutegi-queue", "orm"] # durable job queue over the Backend seam (SQLite or Postgres) events = ["dep:sutegi-events", "orm"] # event sourcing: event store + aggregates + projections derive = ["dep:sutegi-macros", "orm"] # #[derive(Model)] / #[derive(Validate)] sqlite = ["orm", "sutegi-orm/sqlite"] # runnable bundled SQLite (per-pod) diff --git a/crates/sutegi/src/lib.rs b/crates/sutegi/src/lib.rs index 346aa7b..b73aad9 100644 --- a/crates/sutegi/src/lib.rs +++ b/crates/sutegi/src/lib.rs @@ -11,7 +11,7 @@ //! | `postgres` | sutegi-pg (pure std) | Postgres: the multi-pod execution layer | //! | `derive` | sutegi-macros (build-time only) | `#[derive(Model)]` | //! | `validate` | sutegi-validate | request / tool validation | -//! | `queue` | sutegi-queue (+ sutegi-pg) | durable, cross-pod job queue (Postgres) | +//! | `queue` | sutegi-queue (+ orm) | durable job queue over the Backend seam (SQLite or Postgres) | //! | `events` | sutegi-events (+ orm) | event sourcing: append-only event store, aggregates, projections | //! | `session` | sutegi-session | signed-cookie sessions (HMAC-SHA256) + CSRF tokens | //! | `auth` | sutegi-auth (+ session/orm) | the user system: passwords, Users, guards, API tokens, remember-me, login throttling | @@ -449,7 +449,7 @@ pub mod prelude { Vector, }; #[cfg(feature = "queue")] - pub use sutegi_queue::{Queue, Workers}; + pub use sutegi_queue::{JobCtx, Queue, Workers}; #[cfg(feature = "repl")] pub use sutegi_repl::Repl; #[cfg(feature = "validate")] From 3aaac7f8ea69947d1a543600fc3e57067c146cdc Mon Sep 17 00:00:00 2001 From: Eneko Sarasola <113593779+enekos@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:54:03 +0200 Subject: [PATCH 2/2] style: rustfmt the queue tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI's fmt gate caught what my local run didn't — I never ran cargo fmt on the new test files. --- crates/sutegi-queue/tests/durable.rs | 19 +++++++++--- crates/sutegi-queue/tests/sqlite.rs | 46 ++++++++++++++++++++++------ 2 files changed, 52 insertions(+), 13 deletions(-) diff --git a/crates/sutegi-queue/tests/durable.rs b/crates/sutegi-queue/tests/durable.rs index c09eea0..65edb36 100644 --- a/crates/sutegi-queue/tests/durable.rs +++ b/crates/sutegi-queue/tests/durable.rs @@ -11,8 +11,8 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use sutegi_json::Json; -use sutegi_orm::Backend; use sutegi_orm::pg::Pg; +use sutegi_orm::Backend; use sutegi_queue::Queue; // Both tests share one `sutegi_jobs` table, so they must not run concurrently @@ -52,7 +52,11 @@ fn dispatch_process_and_retry() { let seen = Arc::clone(&processed); queue.register("greet", move |job| { - let who = job.payload().get("who").and_then(Json::as_str).unwrap_or("?"); + let who = job + .payload() + .get("who") + .and_then(Json::as_str) + .unwrap_or("?"); seen.lock().unwrap().push(who.to_string()); Ok(()) }); @@ -103,7 +107,10 @@ fn dispatch_process_and_retry() { ); workers.stop(); - queue.store().execute("DROP TABLE sutegi_jobs", &[]).unwrap(); + queue + .store() + .execute("DROP TABLE sutegi_jobs", &[]) + .unwrap(); } #[test] @@ -170,7 +177,11 @@ fn dedupe_keys_and_named_queues_behave_as_on_sqlite() { assert_eq!(first, second, "the live row is returned, not a duplicate"); assert!(!queue.run_once().unwrap(), "default queue is empty"); assert_eq!( - queue.stats_for("video").unwrap().get("ready").and_then(Json::as_i64), + queue + .stats_for("video") + .unwrap() + .get("ready") + .and_then(Json::as_i64), Some(1) ); diff --git a/crates/sutegi-queue/tests/sqlite.rs b/crates/sutegi-queue/tests/sqlite.rs index fa53030..c0883da 100644 --- a/crates/sutegi-queue/tests/sqlite.rs +++ b/crates/sutegi-queue/tests/sqlite.rs @@ -7,8 +7,8 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use sutegi_json::Json; -use sutegi_orm::Backend; use sutegi_orm::db::Db; +use sutegi_orm::Backend; use sutegi_orm::Value; use sutegi_queue::{now_ms, Queue}; @@ -74,7 +74,8 @@ fn count(db: &Db, sql: &str) -> i64 { #[test] fn migrate_is_idempotent_and_creates_the_table() { let (_tmp, db, q) = queue("migrate"); - q.migrate().expect("second migrate is a no-op, not an error"); + q.migrate() + .expect("second migrate is a no-op, not an error"); assert_eq!(count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs"), 0); } @@ -84,7 +85,11 @@ fn a_dispatched_job_runs_with_its_payload_and_leaves_no_row() { let seen = Arc::new(Mutex::new(Vec::::new())); let sink = Arc::clone(&seen); q.register("greet", move |job| { - let who = job.payload().get("who").and_then(Json::as_str).unwrap_or("?"); + let who = job + .payload() + .get("who") + .and_then(Json::as_str) + .unwrap_or("?"); sink.lock().unwrap().push(who.to_string()); Ok(()) }); @@ -121,7 +126,10 @@ fn a_failing_job_retries_until_its_budget_runs_out_then_dead_letters() { .expect("dispatch"); q.run_once().expect("attempt 1"); assert_eq!( - count(&db, "SELECT COUNT(*) AS n FROM sutegi_jobs WHERE failed_at IS NULL"), + count( + &db, + "SELECT COUNT(*) AS n FROM sutegi_jobs WHERE failed_at IS NULL" + ), 1, "still queued for a retry" ); @@ -140,7 +148,10 @@ fn a_failing_job_retries_until_its_budget_runs_out_then_dead_letters() { failed[0].get("last_error").and_then(Json::as_str), Some("always") ); - assert!(!q.run_once().expect("run"), "a dead letter is not claimable"); + assert!( + !q.run_once().expect("run"), + "a dead letter is not claimable" + ); // …and can be put back by hand. assert!(q.retry(id, 1).expect("retry")); @@ -205,7 +216,11 @@ fn priority_wins_over_arrival_order() { let order = Arc::new(Mutex::new(Vec::::new())); let sink = Arc::clone(&order); q.register("task", move |job| { - let tag = job.payload().get("tag").and_then(Json::as_str).unwrap_or(""); + let tag = job + .payload() + .get("tag") + .and_then(Json::as_str) + .unwrap_or(""); sink.lock().unwrap().push(tag.to_string()); Ok(()) }); @@ -245,7 +260,13 @@ fn named_queues_do_not_see_each_others_work() { let stop = AtomicBool::new(false); assert!(q.run_once_on("video", &stop).expect("run")); assert_eq!(ran.lock().unwrap().as_slice(), ["default", "video"]); - assert_eq!(q.stats_for("video").unwrap().get("total").and_then(Json::as_i64), Some(0)); + assert_eq!( + q.stats_for("video") + .unwrap() + .get("total") + .and_then(Json::as_i64), + Some(0) + ); } #[test] @@ -383,7 +404,11 @@ fn concurrent_workers_run_each_job_exactly_once() { let mut seen = done.lock().unwrap().clone(); seen.sort_unstable(); - assert_eq!(seen.len(), JOBS as usize, "every job ran exactly once: {seen:?}"); + assert_eq!( + seen.len(), + JOBS as usize, + "every job ran exactly once: {seen:?}" + ); assert_eq!(seen, (0..JOBS).collect::>()); } @@ -431,7 +456,10 @@ fn stats_and_purge_report_the_states_an_operator_asks_about() { assert_eq!(q.purge_failed(Duration::from_secs(3600)).expect("purge"), 0); assert_eq!(q.purge_failed(Duration::ZERO).expect("purge"), 1); - assert_eq!(q.stats().unwrap().get("failed").and_then(Json::as_i64), Some(0)); + assert_eq!( + q.stats().unwrap().get("failed").and_then(Json::as_i64), + Some(0) + ); } #[test]