A durable work queue, first slice (refs #276) dd327d76062d

-- A durable work queue (oxygenhub#276): work that must not be lost on a

-- restart or a failure, claimed by any worker process (SKIP LOCKED on

-- PostgreSQL), retried with backoff, kept as dead after its last attempt.

CREATE TABLE work_queue (

    id            BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,

    kind          TEXT NOT NULL,

    payload       TEXT NOT NULL DEFAULT '{}',

    -- One live item per kind and key: a second enqueue of the same is dropped.

    dedupe        TEXT NOT NULL DEFAULT '',

    run_after     BIGINT NOT NULL,

    attempts      BIGINT NOT NULL DEFAULT 0,

    last_error    TEXT NOT NULL DEFAULT '',

    claimed_by    TEXT NOT NULL DEFAULT '',

    claimed_until BIGINT NOT NULL DEFAULT 0,

    dead          BOOLEAN NOT NULL DEFAULT FALSE,

    created_ms    BIGINT NOT NULL

);

CREATE INDEX work_queue_due ON work_queue (dead, run_after);

-- A durable work queue (oxygenhub#276): work that must not be lost on a

-- restart or a failure, claimed by any worker process (SKIP LOCKED on

-- PostgreSQL), retried with backoff, kept as dead after its last attempt.

CREATE TABLE work_queue (

    id            INTEGER PRIMARY KEY,

    kind          TEXT NOT NULL,

    payload       TEXT NOT NULL DEFAULT '{}',

    -- One live item per kind and key: a second enqueue of the same is dropped.

    dedupe        TEXT NOT NULL DEFAULT '',

    run_after     INTEGER NOT NULL,

    attempts      INTEGER NOT NULL DEFAULT 0,

    last_error    TEXT NOT NULL DEFAULT '',

    claimed_by    TEXT NOT NULL DEFAULT '',

    claimed_until INTEGER NOT NULL DEFAULT 0,

    dead          INTEGER NOT NULL DEFAULT 0,

    created_ms    INTEGER NOT NULL

);

CREATE INDEX work_queue_due ON work_queue (dead, run_after);

//! The durable work queue (oxygenhub#276, docs/roles.md): items any worker

//! process claims for a lease, retried with backoff, dead after the last

//! attempt. On PostgreSQL a claim skips rows another worker has locked

//! (`FOR UPDATE SKIP LOCKED`); on SQLite the one writer serialises claims.

use super::{now_ms, Pool};

use sqlx::FromRow;

/// Attempts before an item is kept as dead.

pub const MAX_ATTEMPTS: i64 = 8;

#[derive(Clone, Debug, FromRow)]

pub struct Item {

    pub id: i64,

    pub kind: String,

    pub payload: String,

    pub dedupe: String,

    pub run_after: i64,

    pub attempts: i64,

    pub last_error: String,

    pub claimed_by: String,

    pub claimed_until: i64,

    pub dead: bool,

    pub created_ms: i64,

}

/// Queue `kind` with its payload, due after `delay_ms`. With a `dedupe`

/// key, nothing is added while a live item of that kind and key waits.

pub async fn enqueue(pool: &Pool, kind: &str, payload: &serde_json::Value, dedupe: &str, delay_ms: i64) -> sqlx::Result<bool> {

    let now = now_ms();

    let mut tx = super::begin_write(pool).await?;

    if !dedupe.is_empty() {

        let live: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM work_queue WHERE kind = $1 AND dedupe = $2 AND dead = $3")

            .bind(kind)

            .bind(dedupe)

            .bind(false)

            .fetch_one(&mut *tx)

            .await?;

        if live > 0 {

            return Ok(false);

        }

    }

    sqlx::query("INSERT INTO work_queue (kind, payload, dedupe, run_after, created_ms) VALUES ($1, $2, $3, $4, $5)")

        .bind(kind)

        .bind(payload.to_string())

        .bind(dedupe)

        .bind(now + delay_ms.max(0))

        .bind(now)

        .execute(&mut *tx)

        .await?;

    tx.commit().await?;

    Ok(true)

}

/// Claim up to `limit` due items for `worker`, each for `lease_ms`: an item

/// whose worker died is due again when its lease ends.

pub async fn claim(pool: &Pool, worker: &str, limit: i64, lease_ms: i64) -> sqlx::Result<Vec<Item>> {

    let now = now_ms();

    #[cfg(feature = "postgres")]

    let pick = "SELECT id FROM work_queue WHERE dead = $3 AND run_after <= $4 AND claimed_until <= $4 ORDER BY run_after, id LIMIT $5 FOR UPDATE SKIP LOCKED";

    #[cfg(not(feature = "postgres"))]

    let pick = "SELECT id FROM work_queue WHERE dead = $3 AND run_after <= $4 AND claimed_until <= $4 ORDER BY run_after, id LIMIT $5";

    let sql =

        format!("UPDATE work_queue SET claimed_by = $1, claimed_until = $2, attempts = attempts + 1 WHERE id IN ({pick}) RETURNING *");

    let mut tx = pool.begin().await?;

    let items: Vec<Item> =

        sqlx::query_as(&sql).bind(worker).bind(now + lease_ms).bind(false).bind(now).bind(limit).fetch_all(&mut *tx).await?;

    tx.commit().await?;

    Ok(items)

}

/// The item is done: it leaves the queue.

pub async fn done(pool: &Pool, id: i64) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM work_queue WHERE id = $1").bind(id).execute(pool).await?;

    Ok(())

}

/// The item failed: due again after a backoff that doubles each attempt

/// (30 s, 1 min, 2 min, … at most an hour), or dead after the last.

pub async fn failed(pool: &Pool, item: &Item, error: &str) -> sqlx::Result<()> {

    let dead = item.attempts >= MAX_ATTEMPTS;

    let backoff = (30_000i64 << (item.attempts.clamp(1, 8) - 1)).min(3_600_000);

    sqlx::query("UPDATE work_queue SET last_error = $1, dead = $2, run_after = $3, claimed_by = '', claimed_until = 0 WHERE id = $4")

        .bind(error)

        .bind(dead)

        .bind(now_ms() + backoff)

        .bind(item.id)

        .execute(pool)

        .await?;

    Ok(())

}

/// Items waiting (claimed or not) and items dead.

pub async fn counts(pool: &Pool) -> sqlx::Result<(i64, i64)> {

    let waiting: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM work_queue WHERE dead = $1").bind(false).fetch_one(pool).await?;

    let dead: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM work_queue WHERE dead = $1").bind(true).fetch_one(pool).await?;

    Ok((waiting, dead))

}

//! The worker side of the durable work queue (oxygenhub#276,

//! docs/roles.md): every worker process takes due items, a batch at a

//! time, and does them. Work that fails where it starts (a notification's

//! email, the Git mirror after a push) is queued here instead of being

//! lost, and retried with backoff.

use crate::db::{self, queue::Item};

use crate::Shared;

use std::time::Duration;

/// How often an idle worker looks for due items.

pub const POLL: Duration = Duration::from_secs(2);

/// How long a claim lasts before another worker may take the item.

const LEASE_MS: i64 = 5 * 60 * 1000;

/// Items taken at once.

const BATCH: i64 = 10;

/// An email to send again (`mail`).

pub async fn queue_mail(

    app: &Shared,

    to_name: &str,

    to_addr: &str,

    subject: &str,

    body: &str,

    action: Option<(&str, &str)>,

    unsubscribe: Option<&str>,

) {

    let payload = serde_json::json!({

        "to_name": to_name, "to_addr": to_addr, "subject": subject, "body": body,

        "action_label": action.map(|a| a.0), "action_url": action.map(|a| a.1), "unsubscribe": unsubscribe,

    });

    if let Err(e) = db::queue::enqueue(&app.db, "mail", &payload, "", 30_000).await {

        tracing::warn!("queueing an email to {to_addr}: {e}");

    }

}

/// A repository's Git mirror to refresh again (`mirror.refresh`), once.

pub async fn queue_mirror(app: &Shared, repo_id: i64) {

    let payload = serde_json::json!({ "repo_id": repo_id });

    if let Err(e) = db::queue::enqueue(&app.db, "mirror.refresh", &payload, &repo_id.to_string(), 30_000).await {

        tracing::warn!("queueing a Git mirror refresh: {e}");

    }

}

/// This process's name as a claimant: its host and pid.

fn worker_name() -> String {

    let host = std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_default();

    format!("{host}:{}", std::process::id())

}

/// Take and do due items until the process stops.

pub async fn worker(app: Shared) {

    let me = worker_name();

    loop {

        let items = match db::queue::claim(&app.db, &me, BATCH, LEASE_MS).await {

            Ok(items) => items,

            Err(e) => {

                tracing::warn!("work queue: claiming: {e}");

                Vec::new()

            }

        };

        let idle = items.is_empty();

        for item in items {

            run(&app, item).await;

        }

        if idle {

            tokio::time::sleep(POLL).await;

        }

    }

}

/// Do one item and record how it went.

pub async fn run(app: &Shared, item: Item) {

    let outcome = match item.kind.as_str() {

        "mail" => mail(app, &item).await,

        "mirror.refresh" => mirror(app, &item).await,

        other => Err(format!("no worker knows {other:?}")),

    };

    let recorded = match outcome {

        Ok(()) => db::queue::done(&app.db, item.id).await,

        Err(e) => {

            tracing::warn!(kind = %item.kind, attempt = item.attempts, "work queue: {e}");

            // An item nobody can do is dead at once.

            let item = if item.kind != "mail" && item.kind != "mirror.refresh" {

                Item { attempts: db::queue::MAX_ATTEMPTS, ..item }

            } else {

                item

            };

            db::queue::failed(&app.db, &item, &e).await

        }

    };

    if let Err(e) = recorded {

        tracing::warn!("work queue: recording an outcome: {e}");

    }

}

async fn mail(app: &Shared, item: &Item) -> Result<(), String> {

    let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(|e| e.to_string())?;

    let s = |k: &str| p[k].as_str().unwrap_or_default().to_string();

    let (label, url) = (s("action_label"), s("action_url"));

    let action = (!url.is_empty()).then_some(crate::mail::Action { label: &label, url: &url });

    let unsubscribe = s("unsubscribe");

    crate::mail::send_letter(

        app,

        &s("to_name"),

        &s("to_addr"),

        &s("subject"),

        &s("body"),

        action,

        (!unsubscribe.is_empty()).then_some(unsubscribe.as_str()),

    )

    .await

}

async fn mirror(app: &Shared, item: &Item) -> Result<(), String> {

    let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(|e| e.to_string())?;

    let id = p["repo_id"].as_i64().ok_or("no repo_id")?;

    // Gone, or its mirror turned off meanwhile: nothing to do.

    let Some(row) = db::repos::get_by_id(&app.db, id).await.map_err(|e| e.to_string())? else { return Ok(()) };

    crate::mirror::refresh(app, &row).await

}

- **A durable work queue (#276):** work that fails where it starts is queued and retried instead of lost: a notification's email the mail server refused, and a Git mirror export after a push. Any worker process takes due items (on PostgreSQL several at once, each skipping what another holds), retries back off from 30 seconds to an hour, and an item still failing after eight attempts is kept as dead.

pub mod queue;

pub mod queue;

    // The durable work queue (#276): every worker takes its due items.

    app.leader.note("queue.worker");

    tokio::spawn(queue::worker(app.clone()));

pub async fn refresh(app: &Shared, row: &RepoRow) -> Result<(), String> {

        return Ok(());

    let Some(repo) = RepoName::new(&row.owner, &row.name) else { return Ok(()) };

            Ok(())

        Err(e) => {

            tracing::warn!("git mirror of {}: {e}", row.slug());

            Err(format!("git mirror of {}: {e}", row.slug()))

        }

        refresh(&app, &row).await.ok();

        // Sent again from the work queue (#276), when mail is on at all.

        if crate::mail::enabled(app) {

            crate::queue::queue_mail(app, to.name(), &to.email, &subject, &body, Some(("Open it", &link)), Some(&unsubscribe)).await;

        }

                    // Failed, it is done again from the work queue (#276).

                    if crate::mirror::refresh(&app, &row).await.is_err() {

                        crate::queue::queue_mirror(&app, row.id).await;

                    }

        crate::mirror::refresh(app, row).await.ok();

        crate::mirror::refresh(&app, &row).await.ok();

    tasks.extend(["diffusion.backfill", "images.learn_archs", "queue.worker"]);

/// oxygenhub#276: the work queue. A keyed item is queued once; a claim

/// leases it (another claim skips it); a failure backs off and the last

/// attempt leaves it dead; done removes it. An item no worker knows is dead

/// at once.

async fn the_work_queue_leases_retries_and_keeps_the_dead() {

    use oxygenhub::db::queue;

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let app = oxygenhub::build(config(dir.path(), &["web"])).await.unwrap();

    let payload = serde_json::json!({ "repo_id": 7 });

    assert!(queue::enqueue(&app.db, "mirror.refresh", &payload, "7", 0).await.unwrap());

    assert!(!queue::enqueue(&app.db, "mirror.refresh", &payload, "7", 0).await.unwrap(), "once per key");

    let got = queue::claim(&app.db, "a", 10, 60_000).await.unwrap();

    assert_eq!((got.len(), got[0].attempts, got[0].claimed_by.as_str()), (1, 1, "a"));

    assert!(queue::claim(&app.db, "b", 10, 60_000).await.unwrap().is_empty(), "leased to a");

    queue::failed(&app.db, &got[0], "the disk is full").await.unwrap();

    assert!(queue::claim(&app.db, "b", 10, 60_000).await.unwrap().is_empty(), "backing off");

    sqlx::query("UPDATE work_queue SET run_after = 0, attempts = $1").bind(queue::MAX_ATTEMPTS - 1).execute(&app.db).await.unwrap();

    let last = queue::claim(&app.db, "b", 10, 60_000).await.unwrap();

    assert_eq!(last[0].attempts, queue::MAX_ATTEMPTS);

    queue::failed(&app.db, &last[0], "still full").await.unwrap();

    assert_eq!(queue::counts(&app.db).await.unwrap(), (0, 1), "dead after the last attempt");

    // A dead item does not hold its key: the same work queues again.

    assert!(queue::enqueue(&app.db, "mirror.refresh", &payload, "7", 0).await.unwrap());

    let item = queue::claim(&app.db, "a", 10, 60_000).await.unwrap().remove(0);

    // Repository 7 does not exist: nothing to refresh, so it is done.

    oxygenhub::queue::run(&app, item).await;

    assert_eq!(queue::counts(&app.db).await.unwrap(), (0, 1));

    queue::enqueue(&app.db, "no.such.kind", &serde_json::json!({}), "", 0).await.unwrap();

    let item = queue::claim(&app.db, "a", 10, 60_000).await.unwrap().remove(0);

    oxygenhub::queue::run(&app, item).await;

    assert_eq!(queue::counts(&app.db).await.unwrap(), (0, 2), "an unknown kind is dead at once");

}

#[tokio::test(flavor = "multi_thread")]

    assert_eq!(app.leader.tasks().len(), 12, "{:?}", app.leader.tasks());

**Built, first slice (#276):** the `work_queue` table (migration 0091:

kind, payload, a dedupe key, run after, attempts, last error, claimed by,

claimed until, dead) and `db::queue` (enqueue, claim, done, failed,

counts). A claim leases a batch for five minutes (`UPDATE … WHERE id IN

(SELECT … FOR UPDATE SKIP LOCKED) RETURNING *` on Postgres; under the write

lock on SQLite), so any number of worker processes take items, and an item

whose worker died is due again when its lease ends. A failure backs off

(30 s, doubling, at most an hour); the eighth attempt leaves the item dead,

kept for a person to read; a kind no worker knows is dead at once. Every

worker runs `queue.worker` (no leader lock), polling every 2 s. Tenants so

far: a notification's **email** that failed to send (`mail`), and the

**Git mirror** export after a push that failed (`mirror.refresh`, one per

repository); both still try at once and queue only on failure, so the

usual path is as fast as before. Next: webhook deliveries onto it, the

apt rebuild, Diffusion's bumps, and a nudge on the event bus.