rustlavel_queue/

database.rs

//! The database driver: a `jobs` table, a `failed_jobs` table, and a
//! reservation that is safe when several workers race for the same row.
//!
//! Timestamps are stored as epoch seconds in `bigint` columns rather than as
//! `timestamptz`. rustlavel-db deliberately leaves a timestamp column as text so
//! precision survives, and a queue compares times far more often than it shows
//! them: an integer sorts, compares and indexes without anything having to
//! agree on a text format.

use crate::job::{BoxFuture, FailedJob, QueuedJob, ReservedJob};
use crate::queue::{Queue, record_pushed};
use crate::time::unix_now;
use rustlavel_core::{Error, Json, Result};
use rustlavel_db::schema::{Schema, Table};
use rustlavel_db::dialect::quote_qualified;
use rustlavel_db::{Database, Value};
use std::time::Duration;

/// The conventional table names, matching Laravel's.
pub const JOBS_TABLE: &str = "jobs";
pub const FAILED_JOBS_TABLE: &str = "failed_jobs";

/// Create the queue's two tables.
///
/// Shared by the migration below and by [`DatabaseQueue::migrate`], so a test
/// that wants its own table names gets exactly the schema production has.
pub async fn create_tables(schema: &Schema<'_>, jobs: &str, failed: &str) -> Result<()> {
    schema.create(jobs, define_jobs_table).await?;
    schema.create(failed, define_failed_jobs_table).await?;
    Ok(())
}

/// The `jobs` table. A free function rather than a closure so a test can build
/// the statements without a database, and see the same schema production gets.
pub fn define_jobs_table(t: &mut Table) {
    t.id();
    t.string("queue");
    t.string("name");
    t.json("payload");
    // Counted up on every reservation, never reset: it is the number that
    // decides when a job has had enough.
    t.integer("attempts").default_int(0);
    t.integer("max_tries").default_int(crate::job::DEFAULT_TRIES as i64);
    t.integer("retry_after").default_int(60);
    t.big_integer("available_at");
    // Null means nobody holds it. Set when a worker reserves it, which is also
    // what lets a crashed worker's job be reclaimed.
    t.big_integer("reserved_at").nullable();
    t.big_integer("created_at");
    // The exact shape of the reservation query, so finding the next job is an
    // index scan rather than a table scan taken under a row lock.
    t.index(&["queue", "reserved_at", "available_at"]);
}

/// The dead-letter table.
pub fn define_failed_jobs_table(t: &mut Table) {
    t.id();
    t.string("queue");
    t.string("name");
    t.json("payload");
    t.integer("attempts");
    t.text("error");
    t.big_integer("failed_at");
}

rustlavel_db::migration!(
    CreateQueueTables,
    "2026_08_29_000001_create_queue_tables",
    up: |schema| { crate::database::create_tables(schema, JOBS_TABLE, FAILED_JOBS_TABLE).await },
    down: |schema| {
        schema.drop(FAILED_JOBS_TABLE).await?;
        schema.drop(JOBS_TABLE).await
    },
);

pub const JOB_PROGRESS_TABLE: &str = "job_progress";

/// Where a job's progress is kept, so a worker in one process and a request
/// handler in another read the same row.
pub fn define_job_progress_table(t: &mut Table) {
    t.id();
    // The job id, or `chain:<id>` for a pipeline.
    t.string("key").unique();
    t.integer("percent");
    t.string("stage");
    t.big_integer("updated_at");
    t.boolean("finished").default_bool(false);
    t.text("failed").nullable();
}

// Its own migration, separate from the queue tables: a project that has run
// `CreateQueueTables` already adds this one and touches nothing else.
rustlavel_db::migration!(
    CreateJobProgressTable,
    "2026_09_17_000001_create_job_progress_table",
    up: |schema| { schema.create(JOB_PROGRESS_TABLE, crate::database::define_job_progress_table).await },
    down: |schema| { schema.drop(JOB_PROGRESS_TABLE).await },
);

/// Progress in a table. The one to use once the worker is its own process.
#[derive(Clone)]
pub struct DatabaseProgress {
    db: Database,
    table: String,
}

impl DatabaseProgress {
    pub fn new(db: Database) -> DatabaseProgress {
        DatabaseProgress { db, table: JOB_PROGRESS_TABLE.to_string() }
    }

    pub fn with_table(mut self, table: impl Into<String>) -> DatabaseProgress {
        self.table = table.into();
        self
    }
}

impl crate::progress::ProgressStore for DatabaseProgress {
    fn set<'a>(&'a self, key: &'a str, progress: crate::progress::Progress) -> crate::progress::ProgressFuture<'a, ()> {
        Box::pin(async move {
            let values: Vec<(&str, Value)> = vec![
                ("percent", Value::from(progress.percent as i64)),
                ("stage", Value::from(progress.stage.as_str())),
                ("updated_at", Value::from(progress.updated_at as i64)),
                ("finished", Value::from(progress.finished)),
                ("failed", progress.failed.as_deref().map_or(Value::Null, Value::from)),
            ];

            // An upsert without a dialect-specific `ON CONFLICT`: update, and
            // insert if there was nothing to update. Two writers racing on a
            // brand-new key can both miss the update and one insert then loses
            // to the unique index — in which case the update is tried again.
            // Progress is a hint, so the last writer winning is the right
            // outcome and no write is an error.
            let updated = self.db.table(&self.table).filter("key", key).update(&self.db, &values).await?;
            if updated > 0 {
                return Ok(());
            }
            let mut insert = values.clone();
            insert.push(("key", Value::from(key)));
            if self.db.table(&self.table).insert_without_id(&self.db, &insert).await.is_err() {
                self.db.table(&self.table).filter("key", key).update(&self.db, &values).await?;
            }
            Ok(())
        })
    }

    fn get<'a>(&'a self, key: &'a str) -> crate::progress::ProgressFuture<'a, Option<crate::progress::Progress>> {
        Box::pin(async move {
            let row = self.db.table(&self.table).filter("key", key).first(&self.db).await?;
            Ok(row.map(|row| crate::progress::Progress {
                percent: row.get::<i64>("percent").unwrap_or(0).clamp(0, 100) as u8,
                stage: row.get::<String>("stage").unwrap_or_default(),
                updated_at: row.get::<i64>("updated_at").unwrap_or(0).max(0) as u64,
                finished: row.get::<bool>("finished").unwrap_or(false),
                failed: row.get::<String>("failed").ok().filter(|f| !f.is_empty()),
            }))
        })
    }

    fn purge<'a>(&'a self, before: u64) -> crate::progress::ProgressFuture<'a, usize> {
        Box::pin(async move {
            let removed = self
                .db
                .table(&self.table)
                .filter_op("updated_at", "<", before as i64)
                .delete(&self.db)
                .await?;
            Ok(removed as usize)
        })
    }
}

/// A queue backed by two database tables.
///
/// Survives a restart, and is shared by every process pointed at the same
/// database — which is the whole reason to prefer it over the memory driver.
#[derive(Clone)]
pub struct DatabaseQueue {
    db: Database,
    jobs: String,
    failed: String,
    /// Both names pre-quoted, so no statement builds an identifier at run time.
    jobs_sql: String,
    failed_sql: String,
}

/// How many orphans one reclaim releases.
///
/// Bounded because the release names each row: SQL Server refuses a statement
/// with more than 2,100 parameters, and a crash that orphaned thousands of jobs
/// would otherwise produce exactly that. Reclaim runs whenever the queue looks
/// empty, so a larger backlog is cleared over a few polls rather than in one.
const RECLAIM_BATCH: i64 = 500;

/// The statement that finds orphaned reservations, built for this database.
///
/// The same three dialect pieces as [`reserve_sql`], in the same order —
/// `limit` before the lock clause, which MySQL insists on — so the two locking
/// reads in the queue cannot disagree about how a row is claimed.
fn reclaim_sql(jobs_sql: &str, dialect: &dyn rustlavel_db::Dialect) -> String {
    let (hint, lock) = dialect.skip_locked();
    format!(
        "select id from {jobs_sql}{hint} \
         where queue = {} and reserved_at is not null and reserved_at + retry_after < {} \
         order by id{}{lock}",
        dialect.placeholder(1),
        dialect.placeholder(2),
        dialect.limit_offset(Some(RECLAIM_BATCH), None, true)
    )
}

/// The statement that claims one job, built for whichever database this is
/// talking to.
///
/// A free function so its shape can be asserted without opening a database.
/// Four things here are database-specific and none may be written literally:
/// the **table name's quoting**, the **placeholders** (`$1` on PostgreSQL,
/// `?` elsewhere), the **row-claiming clause** (a suffix on PostgreSQL and
/// MySQL, a table hint on SQL Server, nothing at all on SQLite), and
/// **`limit`**, which SQL Server spells `offset … fetch next … rows only`.
///
/// Every one of those was once written the PostgreSQL way, which made this
/// the only package in the workspace that could not run on the other
/// databases the framework claims to support.
fn reserve_sql(jobs_sql: &str, dialect: &dyn rustlavel_db::Dialect) -> String {
    let (hint, lock) = dialect.skip_locked();
    // `limit` before the lock clause. MySQL requires that order and rejects
    // the other with a syntax error; PostgreSQL accepts both, which is how the
    // wrong order survived — it was only ever run against PostgreSQL.
    format!(
        "select id, name, payload, attempts, max_tries, retry_after \
         from {jobs_sql}{hint} \
         where queue = {} and reserved_at is null and available_at <= {} \
         order by id{}{lock}",
        dialect.placeholder(1),
        dialect.placeholder(2),
        dialect.limit_offset(Some(1), None, true)
    )
}

impl DatabaseQueue {
    /// Use the conventional `jobs` and `failed_jobs` tables.
    pub fn new(db: Database) -> Self {
        DatabaseQueue::with_tables(db, JOBS_TABLE, FAILED_JOBS_TABLE)
            .expect("the built-in table names are valid identifiers")
    }

    /// Use table names of your own. Validated and quoted here, once, for the
    /// database this queue is talking to — `"jobs"` on PostgreSQL and SQLite,
    /// `` `jobs` `` on MySQL, `[jobs]` on SQL Server. The shared
    /// `quote_identifier` helper always writes double quotes, which MySQL
    /// reads as a string literal rather than a table.
    pub fn with_tables(db: Database, jobs: &str, failed: &str) -> Result<Self> {
        Ok(DatabaseQueue {
            jobs_sql: quote_qualified(db.dialect(), jobs)?,
            failed_sql: quote_qualified(db.dialect(), failed)?,
            jobs: jobs.to_string(),
            failed: failed.to_string(),
            db,
        })
    }

    pub fn database(&self) -> &Database {
        &self.db
    }

    /// Create the tables directly, for an application that would rather not
    /// register [`CreateQueueTables`] in its migration registry.
    pub async fn migrate(&self) -> Result<()> {
        let schema = Schema::new(&self.db);
        create_tables(&schema, &self.jobs, &self.failed).await?;
        // The progress table too. `create_if_missing`, not `has_table` then
        // `create`: this runs on every boot, and two workers booting at once
        // would both see no table and one would fail — which is exactly what
        // happened to the test suite the first time this ran.
        schema.create_if_missing(JOB_PROGRESS_TABLE, define_job_progress_table).await
    }

    /// Drop the tables. Intended for tests and `queue:fresh`.
    pub async fn drop_tables(&self) -> Result<()> {
        let schema = Schema::new(&self.db);
        schema.drop(&self.failed).await?;
        schema.drop(&self.jobs).await
    }

    /// Take the next job, holding it against every other worker.
    ///
    /// `select ... for update skip locked` is the primitive this needs, and
    /// nothing weaker will do:
    ///
    /// * A plain `select ... limit 1` followed by an `update` is a read-then-
    ///   write race. Two workers read the same row, both update it, and the job
    ///   runs twice — the one failure mode a queue may not have.
    /// * `for update` alone fixes correctness by making the second worker block
    ///   on the row lock, but it also serialises the pool: with ten workers, nine
    ///   are queued behind the same row instead of taking the nine rows after it.
    /// * `for update skip locked` locks the row this transaction picked and
    ///   makes every other transaction *step over* it rather than wait. Ten
    ///   workers take ten different rows in parallel, and no row is ever handed
    ///   out twice, because the lock is held until this transaction commits the
    ///   `reserved_at` that closes the row to everyone else.
    ///
    /// The select and the update are in one transaction for exactly that
    /// reason: the lock only lasts as long as the transaction does, so the row
    /// must be marked reserved before it is released.
    /// The statement that claims one job, built for whichever database this
    /// is talking to.
    ///
    /// Split out from [`DatabaseQueue::reserve`] so its shape can be asserted
    /// without a server. Three things here are database-specific and none may
    /// be written literally: the **placeholders** (`$1` on PostgreSQL, `?`
    /// everywhere else), the **row-claiming clause** (a suffix on PostgreSQL
    /// and MySQL, a table hint on SQL Server, nothing at all on SQLite), and
    /// **`limit`**, which SQL Server spells `offset … fetch next … rows only`.
    ///
    /// Every one of those was once written the PostgreSQL way, which made this
    /// the only package in the workspace that could not run on the other
    /// databases the framework supports.
    async fn reserve(&self, queue: &str) -> Result<Option<ReservedJob>> {
        let now = unix_now();
        let mut tx = self.db.begin().await?;

        let dialect = self.db.dialect();
        let sql = reserve_sql(&self.jobs_sql, dialect);
        let row = tx.select_one(&sql, &[Value::from(queue), Value::from(now)]).await?;

        let Some(row) = row else {
            // Nothing to do, but the transaction still has to end — an open one
            // would be discarded by the pool rather than reused.
            tx.rollback().await?;
            return Ok(None);
        };

        let id = row.get::<i64>("id")?;
        let attempts = row.get::<i64>("attempts")? as u32 + 1;
        let retry_after = Duration::from_secs(row.get::<i64>("retry_after")?.max(0) as u64);

        tx.execute(
            &format!(
                "update {} set attempts = {}, reserved_at = {} where id = {}",
                self.jobs_sql,
                dialect.placeholder(1),
                dialect.placeholder(2),
                dialect.placeholder(3)
            ),
            &[Value::from(attempts), Value::from(now), Value::from(id)],
        )
        .await?;

        tx.commit().await?;

        // The payload column may hold a chain envelope; `with_stored_payload`
        // unfolds it and leaves an ordinary payload alone.
        let stored: Json = row.get::<Json>("payload")?;
        let job = QueuedJob {
            name: row.get::<String>("name")?,
            payload: Json::Null,
            queue: queue.to_string(),
            max_tries: row.get::<i64>("max_tries")?.max(1) as u32,
            retry_after,
            delay: Duration::ZERO,
            chain: None,
            then: Vec::new(),
        }
        .with_stored_payload(stored);

        Ok(Some(ReservedJob {
            id: id.to_string(),
            job,
            attempts,
        }))
    }

    /// Release jobs whose worker never came back.
    ///
    /// A worker that is killed between reserving and finishing leaves
    /// `reserved_at` set forever. Once the job's own `retry_after` has passed
    /// with no result, the job is assumed orphaned and goes back on the queue.
    ///
    /// Called only when the queue looks empty, so the common path — a worker
    /// picking up one of many waiting jobs — pays nothing for it.
    pub async fn reclaim_expired(&self, queue: &str) -> Result<u64> {
        let now = unix_now();
        let dialect = self.db.dialect();

        // **Two statements, not one, and the first skips locked rows.** This
        // was a single `update … where reserved_at + retry_after < ?`, and on
        // MySQL it deadlocked against the workers it runs beside: an InnoDB
        // update locks every row it *scans*, not just the ones it changes,
        // while each worker holds its own row under `for update skip locked`.
        // Two transactions each holding what the other wanted — MySQL killed
        // one with error 1213, seven runs in fifteen of the queue's race test.
        //
        // Selecting with the same skip-locked clause the reservation uses
        // means reclaim steps over a row somebody holds instead of waiting for
        // it, and every locking read in the queue now does that. A transaction
        // that never waits cannot be part of a cycle. A row skipped here is
        // not an orphan anyway: somebody holds it right now.
        let mut tx = self.db.begin().await?;
        let rows = tx
            .select(&reclaim_sql(&self.jobs_sql, dialect), &[Value::from(queue), Value::from(now)])
            .await?;
        let ids: Vec<i64> =
            rows.iter().map(|row| row.get::<i64>("id")).collect::<Result<_>>()?;
        if ids.is_empty() {
            tx.rollback().await?;
            return Ok(0);
        }

        let placeholders: Vec<String> =
            (0..ids.len()).map(|n| dialect.placeholder(n + 2)).collect();
        let mut params = Vec::with_capacity(ids.len() + 1);
        params.push(Value::from(now));
        params.extend(ids.iter().map(|id| Value::from(*id)));
        let changed = tx
            .execute(
                &format!(
                    "update {} set reserved_at = null, available_at = {} where id in ({})",
                    self.jobs_sql,
                    dialect.placeholder(1),
                    placeholders.join(", ")
                ),
                &params,
            )
            .await?;
        tx.commit().await?;
        Ok(changed)
    }

    fn row_to_failed(row: &rustlavel_db::Row) -> Result<FailedJob> {
        Ok(FailedJob {
            id: row.get::<i64>("id")?.to_string(),
            name: row.get::<String>("name")?,
            queue: row.get::<String>("queue")?,
            payload: row.get::<Json>("payload")?,
            attempts: row.get::<i64>("attempts")?.max(0) as u32,
            error: row.get::<String>("error")?,
            failed_at: row.get::<i64>("failed_at")?,
        })
    }

    fn job_id(job: &ReservedJob) -> Result<i64> {
        job.id.parse::<i64>().map_err(|_| {
            Error::msg(format!(
                "`{}` is not an id from the database queue. Was this job reserved by a \
                 different driver?",
                job.id
            ))
        })
    }
}

impl Queue for DatabaseQueue {
    fn driver(&self) -> &'static str {
        "database"
    }

    fn push(&self, job: QueuedJob) -> BoxFuture<'_, Result<String>> {
        Box::pin(async move {
            let now = unix_now();
            let id = self
                .db
                .table(&self.jobs)
                .insert(
                    &self.db,
                    &[
                        ("queue", Value::from(job.queue.as_str())),
                        ("name", Value::from(job.name.as_str())),
                        ("payload", Value::from(job.stored_payload())),
                        ("attempts", Value::from(0)),
                        ("max_tries", Value::from(job.max_tries)),
                        ("retry_after", Value::from(job.retry_after.as_secs() as i64)),
                        ("available_at", Value::from(now + job.delay.as_secs() as i64)),
                        ("created_at", Value::from(now)),
                    ],
                )
                .await?
                .to_string();

            record_pushed(self.driver(), &job, &id);
            Ok(id)
        })
    }

    fn pop<'a>(&'a self, queue: &'a str) -> BoxFuture<'a, Result<Option<ReservedJob>>> {
        Box::pin(async move {
            if let Some(job) = self.reserve(queue).await? {
                return Ok(Some(job));
            }

            // Nothing was ready. Before reporting an empty queue, check whether
            // a dead worker is sitting on something, and only re-select if it
            // actually released anything.
            if self.reclaim_expired(queue).await? > 0 {
                return self.reserve(queue).await;
            }

            Ok(None)
        })
    }

    fn size<'a>(&'a self, queue: &'a str) -> BoxFuture<'a, Result<u64>> {
        Box::pin(async move {
            let count = self.db.table(&self.jobs).filter("queue", queue).count(&self.db).await?;
            Ok(count.max(0) as u64)
        })
    }

    fn delete<'a>(&'a self, job: &'a ReservedJob) -> BoxFuture<'a, Result<()>> {
        Box::pin(async move {
            self.db
                .table(&self.jobs)
                .filter("id", DatabaseQueue::job_id(job)?)
                .delete(&self.db)
                .await?;
            Ok(())
        })
    }

    fn release<'a>(&'a self, job: &'a ReservedJob, delay: Duration) -> BoxFuture<'a, Result<()>> {
        Box::pin(async move {
            self.db
                .table(&self.jobs)
                .filter("id", DatabaseQueue::job_id(job)?)
                .update(
                    &self.db,
                    &[
                        ("reserved_at", Value::Null),
                        ("available_at", Value::from(unix_now() + delay.as_secs() as i64)),
                    ],
                )
                .await?;
            Ok(())
        })
    }

    fn fail<'a>(&'a self, job: &'a ReservedJob, error: &'a str) -> BoxFuture<'a, Result<()>> {
        Box::pin(async move {
            let id = DatabaseQueue::job_id(job)?;

            // One transaction, so a job can never be in both tables or neither.
            let mut tx = self.db.begin().await?;

            tx.execute(
                &format!(
                    "insert into {} (queue, name, payload, attempts, error, failed_at) \
                     values ({}, {}, {}, {}, {}, {})",
                    self.failed_sql,
                    self.db.dialect().placeholder(1),
                    self.db.dialect().placeholder(2),
                    self.db.dialect().placeholder(3),
                    self.db.dialect().placeholder(4),
                    self.db.dialect().placeholder(5),
                    self.db.dialect().placeholder(6)
                ),
                &[
                    Value::from(job.job.queue.as_str()),
                    Value::from(job.job.name.as_str()),
                    // The envelope, chain included, so a failed step's
                    // remaining steps are visible in the dead-letter table
                    // rather than silently gone.
                    Value::from(job.job.stored_payload()),
                    Value::from(job.attempts),
                    Value::from(error),
                    Value::from(unix_now()),
                ],
            )
            .await?;

            tx.execute(
                &format!(
                    "delete from {} where id = {}",
                    self.jobs_sql,
                    self.db.dialect().placeholder(1)
                ),
                &[Value::from(id)],
            )
            .await?;

            tx.commit().await
        })
    }

    fn failed_jobs(&self) -> BoxFuture<'_, Result<Vec<FailedJob>>> {
        Box::pin(async move {
            let rows = self
                .db
                .select(&format!("select * from {} order by id", self.failed_sql), &[])
                .await?;
            rows.iter().map(DatabaseQueue::row_to_failed).collect()
        })
    }

    fn clear<'a>(&'a self, queue: &'a str) -> BoxFuture<'a, Result<u64>> {
        Box::pin(async move {
            self.db.table(&self.jobs).filter("queue", queue).delete(&self.db).await
        })
    }
}

#[cfg(test)]
mod portability {
    /// **No statement in this file may write a placeholder literally.**
    ///
    /// `$1` is PostgreSQL's spelling. MySQL and SQL Server use `?`, SQLite
    /// uses `?` too, and a query carrying `$1` is a syntax error on three of
    /// the four databases this framework claims to support. Every statement
    /// here went through `dialect.placeholder(n)` — this fails if one stops.
    ///
    /// It was not hypothetical: until this guard existed, the queue was the
    /// only package in the workspace that could not run anywhere but
    /// PostgreSQL, and nothing said so.
    #[test]
    fn no_statement_writes_a_placeholder_literally() {
        let source = include_str!("database.rs");
        let offenders: Vec<(usize, &str)> = source
            .lines()
            .enumerate()
            .map(|(number, line)| (number + 1, line))
            // Everything *before* this module. `skip_while` was the bug: it
            // kept only the guard itself, whose own strings name the very
            // things it forbids, so it failed on its own explanation.
            .take_while(|(_, line)| !line.contains("mod portability"))
            .filter(|(_, line)| !line.trim_start().starts_with("///"))
            .filter(|(_, line)| {
                ["$1", "$2", "$3", "$4", "$5", "$6"].iter().any(|p| line.contains(p))
            })
            .collect();
        assert!(
            offenders.is_empty(),
            "these lines write a PostgreSQL placeholder literally; use \
             `dialect.placeholder(n)`: {offenders:?}"
        );
    }

    /// The statement each database actually receives.
    ///
    /// Asserted as whole strings rather than by searching for fragments: a
    /// clause in the wrong *position* is still present, and SQL Server's hint
    /// belongs after the table while PostgreSQL's belongs at the end.
    #[test]
    fn the_claiming_statement_is_built_for_each_database() {
        use rustlavel_db::dialect::{MySql, Postgres, Sqlite, SqlServer};

        use rustlavel_db::dialect::quote_qualified;

        let sql = |dialect: &dyn rustlavel_db::Dialect| {
            super::reserve_sql(&quote_qualified(dialect, "jobs").unwrap(), dialect)
        };

        // **These strings are only as right as the belief they encode.** They
        // once asserted `… for update skip locked limit 1` for PostgreSQL *and*
        // MySQL, and passed — because that was what the code produced and what
        // its author thought correct. MySQL rejects it: `limit` must come
        // first. PostgreSQL accepts either order, so a suite run only against
        // PostgreSQL could never have said so.
        //
        // What caught it was running the queue's integration tests against a
        // live MySQL. So these assertions pin the shape, and the live suites
        // (`tests/postgres.rs`, pointed at each database in turn) prove it is
        // accepted. Neither is enough alone.
        assert_eq!(
            sql(&Postgres),
            "select id, name, payload, attempts, max_tries, retry_after \
             from \"jobs\" \
             where queue = $1 and reserved_at is null and available_at <= $2 \
             order by id limit 1 for update skip locked"
        );

        assert_eq!(
            sql(&MySql),
            "select id, name, payload, attempts, max_tries, retry_after \
             from `jobs` \
             where queue = ? and reserved_at is null and available_at <= ? \
             order by id limit 1 for update skip locked"
        );

        // Brackets, `@P1` rather than `?`, the hint after the table, and no
        // `limit` anywhere — four differences in one statement, which is the
        // whole reason none of them may be written by hand.
        assert_eq!(
            sql(&SqlServer),
            "select id, name, payload, attempts, max_tries, retry_after \
             from [jobs] with (updlock, readpast, rowlock) \
             where queue = @P1 and reserved_at is null and available_at <= @P2 \
             order by id offset 0 rows fetch next 1 rows only"
        );

        // No locking clause at all: SQLite's `begin immediate` is what stops
        // two workers reading the same row.
        assert_eq!(
            sql(&Sqlite),
            "select id, name, payload, attempts, max_tries, retry_after \
             from \"jobs\" \
             where queue = ? and reserved_at is null and available_at <= ? \
             order by id limit 1"
        );
    }

    /// The reclaim statement locks rows the same way the reservation does —
    /// that is the whole fix: both locking reads skip what somebody holds, so
    /// neither can wait on the other. These pin the shape; the deadlock itself
    /// is only visible against a live MySQL (`many_workers_racing…`, which
    /// failed seven runs in fifteen before this).
    #[test]
    fn the_reclaim_select_claims_rows_like_the_reservation_does() {
        use rustlavel_db::dialect::{MySql, Postgres, SqlServer, Sqlite, quote_qualified};

        let sql = |dialect: &dyn rustlavel_db::Dialect| {
            super::reclaim_sql(&quote_qualified(dialect, "jobs").unwrap(), dialect)
        };

        assert_eq!(
            sql(&Postgres),
            "select id from \"jobs\" \
             where queue = $1 and reserved_at is not null and reserved_at + retry_after < $2 \
             order by id limit 500 for update skip locked"
        );
        assert_eq!(
            sql(&MySql),
            "select id from `jobs` \
             where queue = ? and reserved_at is not null and reserved_at + retry_after < ? \
             order by id limit 500 for update skip locked"
        );
        assert_eq!(
            sql(&SqlServer),
            "select id from [jobs] with (updlock, readpast, rowlock) \
             where queue = @P1 and reserved_at is not null and reserved_at + retry_after < @P2 \
             order by id offset 0 rows fetch next 500 rows only"
        );
        assert_eq!(
            sql(&Sqlite),
            "select id from \"jobs\" \
             where queue = ? and reserved_at is not null and reserved_at + retry_after < ? \
             order by id limit 500"
        );
    }

    /// The row-claiming clause is the dialect's too, for the same reason:
    /// SQL Server spells it as a table hint and SQLite has no such thing.
    #[test]
    fn the_row_lock_is_asked_of_the_dialect_not_written_here() {
        let source = include_str!("database.rs");
        let written_literally = source
            .lines()
            .take_while(|line| !line.contains("mod portability"))
            .any(|line| {
                // Comments of either kind may name the phrase — explaining why
                // it is not written here is the point of half of them. Only a
                // line of code can be a statement.
                let code = line.trim_start();
                !code.starts_with("//") && line.contains("for update skip locked")
            });
        assert!(
            !written_literally,
            "`for update skip locked` is written into a statement here; ask \
             `dialect.skip_locked()` instead — SQL Server needs a table hint"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rustlavel_db::Migration;

    #[test]
    fn table_names_are_validated_once_at_construction() {
        let db = Database::lazy(rustlavel_db::DatabaseConfig::default())
            .expect("the default configuration names a driver that exists");

        assert!(DatabaseQueue::with_tables(db.clone(), "app_jobs", "app_failed").is_ok());

        let error = match DatabaseQueue::with_tables(db, "jobs; drop table users", "failed") {
            Err(error) => error.to_string(),
            Ok(_) => panic!("an injected table name should have been rejected"),
        };
        assert!(error.contains("not a valid SQL identifier"), "{error}");
    }

    #[test]
    fn the_jobs_table_has_the_columns_the_driver_reads() {
        // The tables assert PostgreSQL's spelling; the dialect layer covers the
        // other databases.
        let jobs = rustlavel_db::schema::create_statements(
            &rustlavel_db::dialect::Postgres,
            JOBS_TABLE,
            define_jobs_table,
        )
        .unwrap();

        for column in [
            r#""id" bigserial primary key"#,
            r#""queue" varchar(255) not null"#,
            r#""name" varchar(255) not null"#,
            r#""payload" jsonb not null"#,
            r#""attempts" integer not null default 0"#,
            r#""max_tries" integer not null default 3"#,
            r#""retry_after" integer not null default 60"#,
            r#""available_at" bigint not null"#,
            r#""created_at" bigint not null"#,
        ] {
            assert!(jobs[0].contains(column), "expected {column} in:\n{}", jobs[0]);
        }

        // Nullable on purpose: null is what "nobody holds this job" means, and
        // it is what the reservation query filters on.
        assert!(jobs[0].contains(r#""reserved_at" bigint,"#), "{}", jobs[0]);
        assert!(!jobs[0].contains(r#""reserved_at" bigint not null"#), "{}", jobs[0]);

        assert!(
            jobs[1].contains(r#"on "jobs" ("queue", "reserved_at", "available_at")"#),
            "the reservation query needs a covering index:\n{}",
            jobs[1]
        );
    }

    #[test]
    fn the_failed_jobs_table_keeps_everything_needed_to_diagnose_a_job() {
        let failed =
            rustlavel_db::schema::create_statements(
                &rustlavel_db::dialect::Postgres,
                FAILED_JOBS_TABLE,
                define_failed_jobs_table,
            )
                .unwrap();

        for column in [
            r#""queue" varchar(255) not null"#,
            r#""name" varchar(255) not null"#,
            r#""payload" jsonb not null"#,
            r#""attempts" integer not null"#,
            r#""error" text not null"#,
            r#""failed_at" bigint not null"#,
        ] {
            assert!(failed[0].contains(column), "expected {column} in:\n{}", failed[0]);
        }
    }

    #[test]
    fn the_migration_is_registered_under_a_sortable_name() {
        assert_eq!(CreateQueueTables.name(), "2026_08_29_000001_create_queue_tables");
    }

    #[test]
    fn an_id_from_another_driver_is_rejected_rather_than_silently_ignored() {
        let job = ReservedJob {
            id: "memory-7".to_string(),
            job: QueuedJob::new("x", Json::Null),
            attempts: 1,
        };

        let error = DatabaseQueue::job_id(&job).unwrap_err().to_string();
        assert!(error.contains("not an id from the database queue"), "{error}");
    }
}