Run the hub as web and worker processes, singletons under a leader lock e57765dee61a

//! Leader locks (docs/roles.md, #272): a background task that must run

//! once across every process sharing the hub's database runs only in the

//! process holding a lock named for it.

//!

//! On PostgreSQL the lock is a session advisory lock

//! (`pg_try_advisory_lock`) on one connection of this process's own, kept

//! apart from the pool for as long as the process runs. A process that does

//! not hold a task's lock asks again every [`RETRY`]. One that holds it

//! checks its connection every [`CHECK`]; if the connection has gone (the

//! database restarted, the network dropped, the session was ended), the

//! server has freed the lock, so the task is stopped (its future dropped)

//! and the process contends again on a new connection. A process that dies

//! closes its connection, freeing its locks, and another takes over within

//! seconds. Each task reads its work from the database, so a successor

//! picking up where it stopped is safe.

//!

//! On SQLite (one machine, one process by definition) every lock is held.

use crate::Shared;

use std::collections::BTreeSet;

use std::future::Future;

use std::sync::Mutex;

use std::time::Duration;

/// How often a process that does not hold a lock asks for it again.

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

/// How often a holder makes sure its connection, and so its lock, is still

/// there; and how long it waits for the database's answer.

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

/// This process's background tasks and the locks it holds.

#[derive(Default)]

pub struct Leader {

    /// Every background task this process started, led or not.

    tasks: Mutex<BTreeSet<&'static str>>,

    /// The singleton tasks running here now: the locks this process holds.

    leading: Mutex<BTreeSet<&'static str>>,

    /// The loops contending for locks, stopped by [`Leader::resign`].

    contenders: Mutex<Vec<tokio::task::AbortHandle>>,

    #[cfg(feature = "postgres")]

    session: pg::Session,

}

impl Leader {

    /// The background tasks this process started, by name.

    pub fn tasks(&self) -> Vec<&'static str> {

        self.tasks.lock().unwrap_or_else(|p| p.into_inner()).iter().copied().collect()

    }

    /// The leader locks this process holds: the singleton tasks running here.

    pub fn leading(&self) -> Vec<&'static str> {

        self.leading.lock().unwrap_or_else(|p| p.into_inner()).iter().copied().collect()

    }

    /// Record a background task that needs no lock (it is safe anywhere).

    pub fn note(&self, name: &'static str) {

        self.tasks.lock().unwrap_or_else(|p| p.into_inner()).insert(name);

    }

    /// Stop every singleton task here and give up their locks, as a process

    /// that stops does: another process takes them over.

    pub async fn resign(&self) {

        for c in self.contenders.lock().unwrap_or_else(|p| p.into_inner()).drain(..) {

            c.abort();

        }

        #[cfg(feature = "postgres")]

        self.session.close().await;

    }

    /// The process id of the database session holding this process's

    /// locks, if it has one (for tests that end it).

    #[cfg(feature = "postgres")]

    pub async fn session_pid(&self) -> Option<i32> {

        self.session.pid().await

    }

}

// Taking, checking and giving back a lock: on PostgreSQL through the

// session; on SQLite every lock is this process's.

#[cfg(feature = "postgres")]

impl Leader {

    /// Take `name`'s lock now if it is free: the session's generation it

    /// was taken in, which [`Leader::held`] checks against.

    async fn acquire(&self, app: &Shared, name: &'static str) -> Option<u64> {

        self.session.acquire(&app.db, name).await

    }

    /// Whether a lock taken in generation `term` is still held.

    async fn held(&self, term: u64) -> bool {

        self.session.held(term).await

    }

    /// Give `name`'s lock back, its task done.

    async fn release(&self, name: &'static str, term: u64) {

        self.session.release(name, term).await

    }

}

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

impl Leader {

    async fn acquire(&self, _app: &Shared, _name: &'static str) -> Option<u64> {

        Some(0)

    }

    async fn held(&self, _term: u64) -> bool {

        true

    }

    async fn release(&self, _name: &'static str, _term: u64) {}

}

/// Marks a task as led here while it runs; dropping it (the task ended,

/// was stopped, or the loop was aborted) unmarks it.

struct Leading<'a>(&'a Leader, &'static str);

impl<'a> Leading<'a> {

    fn new(leader: &'a Leader, name: &'static str) -> Self {

        leader.leading.lock().unwrap_or_else(|p| p.into_inner()).insert(name);

        Leading(leader, name)

    }

}

impl Drop for Leading<'_> {

    fn drop(&mut self) {

        self.0.leading.lock().unwrap_or_else(|p| p.into_inner()).remove(self.1);

    }

}

/// Run the singleton task `name` in the background, only while this

/// process holds its lock. `task` makes the task's future; it is made

/// again each time the lock is taken after being lost. A task that ends by

/// itself gives its lock back and is not run again.

pub fn run<F, Fut>(app: &Shared, name: &'static str, task: F)

where

    F: Fn(Shared) -> Fut + Send + 'static,

    Fut: Future<Output = ()> + Send + 'static,

{

    app.leader.note(name);

    let handle = tokio::spawn(contend(app.clone(), name, task));

    app.leader.contenders.lock().unwrap_or_else(|p| p.into_inner()).push(handle.abort_handle());

}

async fn contend<F, Fut>(app: Shared, name: &'static str, task: F)

where

    F: Fn(Shared) -> Fut,

    Fut: Future<Output = ()>,

{

    let leader = &app.leader;

    let mut waiting = false;

    loop {

        let Some(term) = leader.acquire(&app, name).await else {

            if !waiting {

                tracing::info!("leader: {name} runs elsewhere; standing by");

                waiting = true;

            }

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

            continue;

        };

        if waiting {

            tracing::info!("leader: took over {name}");

            waiting = false;

        }

        let _leading = Leading::new(leader, name);

        let work = task(app.clone());

        tokio::pin!(work);

        let finished = loop {

            tokio::select! {

                _ = &mut work => break true,

                _ = tokio::time::sleep(CHECK) => {

                    if !leader.held(term).await {

                        break false;

                    }

                }

            }

        };

        if finished {

            leader.release(name, term).await;

            return;

        }

        // The future is dropped here: the task stops where it was.

        tracing::warn!("leader: lost the lock for {name}; stopped it, contending again");

    }

}

#[cfg(feature = "postgres")]

mod pg {

    use super::CHECK;

    use crate::db::Pool;

    use sqlx::postgres::PgConnection;

    use sqlx::Connection;

    use std::time::Instant;

    /// The connection holding this process's locks.

    struct Conn {

        conn: PgConnection,

        /// The schema the hub's tables are in: two hubs in two schemas of

        /// one database are two hubs, and do not share locks.

        schema: String,

        pid: i32,

        /// When the database last answered on it.

        checked: Instant,

    }

    #[derive(Default)]

    pub(super) struct Session {

        conn: tokio::sync::Mutex<Option<Conn>>,

        /// Counts connections lost: a lock taken in an earlier generation

        /// went with its connection.

        term: std::sync::atomic::AtomicU64,

        closed: std::sync::atomic::AtomicBool,

    }

    /// The lock's key: a stable 64-bit hash of the schema and the name.

    pub(super) fn key(schema: &str, name: &str) -> i64 {

        let h = blake3::hash(format!("oxygenhub leader\0{schema}\0{name}").as_bytes());

        i64::from_le_bytes(h.as_bytes()[..8].try_into().expect("eight bytes"))

    }

    impl Session {

        fn term(&self) -> u64 {

            self.term.load(std::sync::atomic::Ordering::SeqCst)

        }

        /// The connection is gone, and every lock with it.

        fn lose(&self, slot: &mut Option<Conn>) {

            if slot.take().is_some() {

                self.term.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

            }

        }

        async fn connect(db: &Pool) -> anyhow::Result<Conn> {

            let options = db.connect_options();

            let mut conn = PgConnection::connect_with(&options).await?;

            let (schema, pid): (String, i32) =

                sqlx::query_as("SELECT CAST(current_schema() AS TEXT), pg_backend_pid()").fetch_one(&mut conn).await?;

            Ok(Conn { conn, schema, pid, checked: Instant::now() })

        }

        pub(super) async fn acquire(&self, db: &Pool, name: &'static str) -> Option<u64> {

            let mut slot = self.conn.lock().await;

            if self.closed.load(std::sync::atomic::Ordering::SeqCst) {

                return None;

            }

            if slot.is_none() {

                match tokio::time::timeout(CHECK * 5, Self::connect(db)).await {

                    Ok(Ok(c)) => *slot = Some(c),

                    Ok(Err(e)) => {

                        tracing::warn!("leader: connecting for {name}'s lock: {e}");

                        return None;

                    }

                    Err(_) => {

                        tracing::warn!("leader: connecting for {name}'s lock: timed out");

                        return None;

                    }

                }

            }

            let term = self.term();

            let c = slot.as_mut()?;

            let ask = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)").bind(key(&c.schema, name));

            match tokio::time::timeout(CHECK, ask.fetch_one(&mut c.conn)).await {

                Ok(Ok(got)) => {

                    c.checked = Instant::now();

                    got.then_some(term)

                }

                Ok(Err(e)) => {

                    tracing::warn!("leader: asking for {name}'s lock: {e}");

                    self.lose(&mut slot);

                    None

                }

                Err(_) => {

                    tracing::warn!("leader: asking for {name}'s lock: timed out");

                    self.lose(&mut slot);

                    None

                }

            }

        }

        pub(super) async fn held(&self, term: u64) -> bool {

            let mut slot = self.conn.lock().await;

            if self.term() != term {

                return false;

            }

            let Some(c) = slot.as_mut() else { return false };

            // One answer serves every task's check made soon after it.

            if c.checked.elapsed() < CHECK / 2 {

                return true;

            }

            let ping = sqlx::query_scalar::<_, i32>("SELECT 1").fetch_one(&mut c.conn);

            match tokio::time::timeout(CHECK, ping).await {

                Ok(Ok(_)) => {

                    c.checked = Instant::now();

                    true

                }

                _ => {

                    tracing::warn!("leader: the connection holding the locks is gone");

                    self.lose(&mut slot);

                    false

                }

            }

        }

        pub(super) async fn release(&self, name: &'static str, term: u64) {

            let mut slot = self.conn.lock().await;

            if self.term() != term {

                return;

            }

            let Some(c) = slot.as_mut() else { return };

            let unlock = sqlx::query_scalar::<_, bool>("SELECT pg_advisory_unlock($1)").bind(key(&c.schema, name));

            if !matches!(tokio::time::timeout(CHECK, unlock.fetch_one(&mut c.conn)).await, Ok(Ok(_))) {

                self.lose(&mut slot);

            }

        }

        pub(super) async fn close(&self) {

            self.closed.store(true, std::sync::atomic::Ordering::SeqCst);

            let mut slot = self.conn.lock().await;

            if let Some(c) = slot.take() {

                self.term.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

                let _ = c.conn.close().await;

            }

        }

        pub(super) async fn pid(&self) -> Option<i32> {

            self.conn.lock().await.as_ref().map(|c| c.pid)

        }

    }

    #[cfg(test)]

    mod tests {

        #[test]

        fn keys_are_stable_and_per_schema() {

            assert_eq!(super::key("hub_a", "flow.scheduler"), super::key("hub_a", "flow.scheduler"));

            assert_ne!(super::key("hub_a", "flow.scheduler"), super::key("hub_b", "flow.scheduler"));

            assert_ne!(super::key("hub_a", "flow.scheduler"), super::key("hub_a", "flow.reaper"));

        }

    }

}

//! Roles (docs/roles.md, #272): a web-only hub starts none of the worker's

//! background tasks, a worker-only hub serves no pages, and on PostgreSQL

//! two hubs on one database run each singleton task once between them.

use axum::body::Body;

use axum::http::{header, Request, StatusCode};

use http_body_util::BodyExt;

use oxygenhub::config::Config;

use oxygenhub::Shared;

use std::path::Path;

use std::time::{Duration, Instant};

use tower::ServiceExt;

/// The worker's singleton tasks that run for as long as the process does.

const WORKER_SINGLETONS: [&str; 7] =

    ["backups.scheduler", "diffusion.reconciler", "flow.placement", "flow.reaper", "flow.scheduler", "hooks.worker", "maintenance.nightly"];

fn hermetic() {

    static ONCE: std::sync::Once = std::sync::Once::new();

    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();

    ONCE.call_once(|| {

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

        std::env::set_var("OXY_HOME", d.path());

        DIR.set(d).ok();

    });

}

fn config(dir: &Path, roles: &[&str]) -> Config {

    let mut config = Config::load(dir).unwrap();

    config.server.roles = roles.iter().map(|r| r.to_string()).collect();

    config.registration.setup_code = "test-setup".into();

    config

}

async fn get(app: &Shared, path: &str, token: Option<&str>) -> (StatusCode, String) {

    let mut req = Request::get(path);

    if let Some(t) = token {

        req = req.header(header::AUTHORIZATION, format!("Bearer {t}"));

    }

    let res = oxygenhub::router(app.clone()).oneshot(req.body(Body::empty()).unwrap()).await.unwrap();

    let status = res.status();

    let body = res.into_body().collect().await.unwrap().to_bytes();

    (status, String::from_utf8_lossy(&body).into_owned())

}

async fn health(app: &Shared) -> serde_json::Value {

    let (status, body) = get(app, "/health", None).await;

    assert_eq!(status, StatusCode::OK, "{body}");

    serde_json::from_str(&body).unwrap()

}

/// Wait up to `limit` for `ok`, checking every 100 ms.

async fn until(limit: Duration, mut ok: impl FnMut() -> bool) -> bool {

    let start = Instant::now();

    while start.elapsed() < limit {

        if ok() {

            return true;

        }

        tokio::time::sleep(Duration::from_millis(100)).await;

    }

    ok()

}

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

async fn a_web_only_hub_starts_none_of_the_workers_tasks() {

    hermetic();

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

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

    // Only the transparency log's checkpointer, which goes with the log's

    // appends in the web role.

    assert_eq!(app.leader.tasks(), ["tlog.checkpointer"]);

    assert!(until(Duration::from_secs(5), || app.leader.leading() == ["tlog.checkpointer"]).await, "{:?}", app.leader.leading());

    let h = health(&app).await;

    assert_eq!(h["roles"], serde_json::json!(["web"]));

    assert_eq!(h["leading"], serde_json::json!(["tlog.checkpointer"]));

    // It serves pages.

    let (status, body) = get(&app, "/explore", None).await;

    assert_eq!(status, StatusCode::OK, "{body}");

}

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

async fn a_worker_only_hub_serves_its_health_and_metrics_and_no_pages() {

    hermetic();

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

    let mut cfg = config(dir.path(), &["worker"]);

    cfg.server.metrics_token = "m".into();

    let app = oxygenhub::build(cfg).await.unwrap();

    let mut tasks = WORKER_SINGLETONS.to_vec();

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

    tasks.sort();

    assert_eq!(app.leader.tasks(), tasks, "every worker task, and not the log's checkpointer");

    assert!(

        until(Duration::from_secs(5), || WORKER_SINGLETONS.iter().all(|n| app.leader.leading().contains(n))).await,

        "{:?}",

        app.leader.leading()

    );

    let h = health(&app).await;

    assert_eq!(h["roles"], serde_json::json!(["worker"]));

    assert!(h["leading"].as_array().unwrap().iter().any(|n| n == "flow.scheduler"), "{h}");

    for path in ["/", "/explore", "/login", "/api/v1/version", "/andy/demo", "/log/checkpoint"] {

        assert_eq!(get(&app, path, None).await.0, StatusCode::NOT_FOUND, "{path} is the web role's");

    }

    let (status, body) = get(&app, "/metrics", Some("m")).await;

    assert_eq!(status, StatusCode::OK, "{body}");

    // It reads the transparency log but never writes it.

    assert!(app.tlog.append(&[]).is_err());

}

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

async fn a_hub_takes_every_role_by_default() {

    hermetic();

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

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

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

    assert_eq!(health(&app).await["roles"], serde_json::json!(["web", "worker"]));

    assert_eq!(get(&app, "/explore", None).await.0, StatusCode::OK);

    let mut bad = config(dir.path(), &["web", "cron"]);

    bad.server.roles.push(String::new());

    let e = oxygenhub::build(bad).await.err().expect("an unknown role is refused").to_string();

    assert!(e.contains("unknown role \"cron\""), "{e}");

}

/// Two hubs on one PostgreSQL database (one schema): exactly one leads

/// each singleton; when its connection drops it stops them, and when it

/// stops the other takes every one over within seconds.

#[cfg(feature = "postgres")]

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

async fn two_hubs_on_one_database_run_each_singleton_once_and_fail_over() {

    hermetic();

    let (da, dbb) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

    let schema = format!("hub_roles_{}", da.path().file_name().unwrap().to_string_lossy());

    let pg = |dir: &Path| {

        let mut c = config(dir, &[]);

        c.database.schema = schema.clone();

        c

    };

    let a = oxygenhub::build(pg(da.path())).await.unwrap();

    let b = oxygenhub::build(pg(dbb.path())).await.unwrap();

    let mut singletons = WORKER_SINGLETONS.to_vec();

    singletons.push("tlog.checkpointer");

    let once = |a: &Shared, b: &Shared| {

        let (la, lb) = (a.leader.leading(), b.leader.leading());

        singletons.iter().all(|n| la.contains(n) != lb.contains(n))

    };

    assert!(until(Duration::from_secs(10), || once(&a, &b)).await, "a {:?}, b {:?}", a.leader.leading(), b.leader.leading());

    // A's connection is ended, as when the network drops: its locks free,

    // it stops what it led, and between them each runs once again.

    let pid = a.leader.session_pid().await.expect("a has a session");

    let ended: bool = sqlx::query_scalar("SELECT pg_terminate_backend($1)").bind(pid).fetch_one(&b.db).await.unwrap();

    assert!(ended);

    let start = Instant::now();

    loop {

        let reconnected = a.leader.session_pid().await.is_some_and(|p| p != pid);

        if reconnected && once(&a, &b) {

            break;

        }

        assert!(start.elapsed() < Duration::from_secs(15), "a {:?}, b {:?}", a.leader.leading(), b.leader.leading());

        tokio::time::sleep(Duration::from_millis(100)).await;

    }

    // The hub leading the scheduler stops; the other takes everything over.

    let (gone, left) = if a.leader.leading().contains(&"flow.scheduler") { (&a, &b) } else { (&b, &a) };

    gone.leader.resign().await;

    let start = Instant::now();

    assert!(

        until(Duration::from_secs(10), || singletons.iter().all(|n| left.leader.leading().contains(n))).await,

        "left {:?}",

        left.leader.leading()

    );

    eprintln!("took over in {:.1} s", start.elapsed().as_secs_f64());

    assert!(gone.leader.leading().is_empty(), "{:?}", gone.leader.leading());

    assert_eq!(health(left).await["leading"].as_array().unwrap().len(), left.leader.leading().len());

}

- **Run the hub as web and worker processes (#272):** `oxygenhub serve --role web,worker` (or `[server] roles`, `OXYHUB_SERVER_ROLES`; both by default, so one process works as before). A web process serves pages, the API, the wire protocol and the runner API; a worker runs the background tasks and serves only `/health` and `/metrics`. On PostgreSQL each singleton task (Flow's scheduler, reaper and placement pass, webhooks, nightly upkeep, backups, Diffusion's index, the log's checkpointer) runs in one process at a time under an advisory lock, and another takes it over within seconds when that process stops or loses its connection. `/health` reports the process's `roles` and the locks it is `leading`. Until the transparency log has a cluster-wide writer (#274), run one web process per data directory. See docs/roles.md.

    /// What this process does (docs/roles.md, #272): `web` (pages, the

    /// API, the wire protocol, the runner API) and `worker` (the background

    /// tasks). Empty, both. `serve --role` overrides it.

    pub roles: Vec<String>,

/// One part of the hub's work a process can take (docs/roles.md, #272).

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub enum Role {

    /// Pages, the JSON API, the wire protocol, live pages and the runner

    /// API, and the transparency log they append to.

    Web,

    /// The background tasks: webhooks, Flow's scheduler, reaper and

    /// placement pass, nightly upkeep, backups and Diffusion's index.

    Worker,

}

impl Role {

    pub const ALL: [Role; 2] = [Role::Web, Role::Worker];

    pub fn name(self) -> &'static str {

        match self {

            Role::Web => "web",

            Role::Worker => "worker",

        }

    }

}

impl std::str::FromStr for Role {

    type Err = String;

    fn from_str(s: &str) -> Result<Role, String> {

        match s.trim().to_ascii_lowercase().as_str() {

            "web" => Ok(Role::Web),

            "worker" => Ok(Role::Worker),

            _ => Err(format!("unknown role {:?}: the roles are web and worker", s.trim())),

        }

    }

}

/// The roles one process takes: every role unless some are named.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub struct Roles {

    pub web: bool,

    pub worker: bool,

}

impl Default for Roles {

    fn default() -> Self {

        Roles { web: true, worker: true }

    }

}

impl Roles {

    /// The roles named, each `web` or `worker`; none named is every role.

    pub fn from_names<S: AsRef<str>>(names: &[S]) -> Result<Roles, String> {

        let names: Vec<&str> = names.iter().map(|s| s.as_ref().trim()).filter(|s| !s.is_empty()).collect();

        if names.is_empty() {

            return Ok(Roles::default());

        }

        let mut roles = Roles { web: false, worker: false };

        for n in names {

            match n.parse::<Role>()? {

                Role::Web => roles.web = true,

                Role::Worker => roles.worker = true,

            }

        }

        Ok(roles)

    }

    pub fn has(self, role: Role) -> bool {

        match role {

            Role::Web => self.web,

            Role::Worker => self.worker,

        }

    }

    /// The roles' names, in a fixed order: what `/health` reports.

    pub fn names(self) -> Vec<&'static str> {

        Role::ALL.into_iter().filter(|r| self.has(*r)).map(Role::name).collect()

    }

}

            roles: Vec::new(),

        cfg.roles()?;

        if let Some(v) = get("OXYHUB_SERVER_ROLES") {

            self.server.roles = v.split(',').map(str::trim).filter(|s| !s.is_empty()).map(String::from).collect();

        }

    }

    /// This process's roles (`[server] roles`); an unknown name is an error.

    pub fn roles(&self) -> anyhow::Result<Roles> {

        Roles::from_names(&self.server.roles).map_err(|e| anyhow::anyhow!("[server] roles: {e}"))

    }

    #[test]

    fn roles_default_to_all_and_unknown_names_are_errors() {

        let cfg = Config::default();

        assert_eq!(cfg.roles().unwrap(), Roles { web: true, worker: true });

        assert_eq!(cfg.roles().unwrap().names(), ["web", "worker"]);

        assert_eq!(Roles::from_names(&["worker"]).unwrap(), Roles { web: false, worker: true });

        assert_eq!(Roles::from_names(&[" Web ", "web"]).unwrap().names(), ["web"]);

        assert_eq!(Roles::from_names(&["worker", "web"]).unwrap().names(), ["web", "worker"]);

        assert_eq!(Roles::from_names::<&str>(&[]).unwrap(), Roles::default());

        let e = Roles::from_names(&["web", "edge"]).unwrap_err();

        assert_eq!(e, "unknown role \"edge\": the roles are web and worker");

        let mut cfg = Config::parse("[server]\nroles = [\"worker\"]\n").unwrap();

        assert_eq!(cfg.roles().unwrap().names(), ["worker"]);

        cfg.apply_env(|k| (k == "OXYHUB_SERVER_ROLES").then(|| "web, worker".into()));

        assert_eq!(cfg.roles().unwrap(), Roles::default());

        cfg.apply_env(|k| (k == "OXYHUB_SERVER_ROLES").then(|| "store".into()));

        assert!(cfg.roles().unwrap_err().to_string().contains("unknown role \"store\""));

    }

    #[test]

    fn an_unknown_role_in_hub_toml_stops_loading() {

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

        std::fs::write(dir.path().join("hub.toml"), "[server]\nroles = [\"wrker\"]\n").unwrap();

        let e = Config::load(dir.path()).unwrap_err().to_string();

        assert!(e.contains("unknown role \"wrker\""), "{e}");

//! One process, or several sharing a database: an axum router serving

//! pages, the JSON API and the wire protocol (the web role); the background

//! tasks (the worker role, singletons under a leader lock: docs/roles.md);

//! SQLite or PostgreSQL for hub state; repositories as bare Oxygen stores.

pub mod leader;

    /// What this process does: web, worker or both (docs/roles.md, #272).

    pub roles: config::Roles,

    /// The background tasks started here and the leader locks held (#272).

    pub leader: leader::Leader,

    let roles = config.roles()?;

    // The web role appends to the log; a worker alone only reads it (#272).

    let (tlog_dir, origin) = (config.data.join("tlog"), tlog::origin_of(config.base_url()));

    let tlog = Arc::new(if roles.web { tlog::Log::open(&tlog_dir, &origin)? } else { tlog::Log::open_reader(&tlog_dir, &origin)? });

        roles,

        leader: leader::Leader::default(),

    start_background(&app);

    setup::ensure(&app).await?;

    Ok(app)

}

/// Start this process's background tasks (docs/roles.md, #272). The

/// worker role's are singletons across every process sharing the database,

/// each run under its leader lock; the transparency log's checkpointer goes

/// with the web role, which appends to the log, and is a singleton too.

fn start_background(app: &Shared) {

    if app.roles.web {

        // The checkpointer signs the log this process appends to (#238).

        leader::run(app, "tlog.checkpointer", tlog::checkpointer);

    }

    if !app.roles.worker {

        return;

    }

    leader::run(app, "hooks.worker", hooks::worker);

    leader::run(app, "flow.reaper", flow::reaper);

    leader::run(app, "flow.scheduler", flow::scheduler);

    leader::run(app, "flow.placement", flow::placement);

    leader::run(app, "maintenance.nightly", maintenance::nightly);

    leader::run(app, "backups.scheduler", backups::scheduler);

    // The reverse pin index for repositories pushed before it existed (#148):

    // once, by whichever worker leads it first (idempotent if run again).

    leader::run(app, "diffusion.backfill", |app| async move { diffusion::backfill(&app).await });

    leader::run(app, "diffusion.reconciler", diffusion::reconciler);

    // What each image is for, read or learned once (oxygenhub#173): safe

    // in every worker, so it takes no lock.

    app.leader.note("images.learn_archs");

}

/// The router this process serves: everything for the web role; for a

/// worker alone, `/health` and `/metrics`, so it can be watched.

pub fn router(app: Shared) -> axum::Router {

    if !app.roles.web {

        return worker_router(app);

    }

    full_router(app)

}

/// What a worker-only process answers: its health and its metrics.

fn worker_router(app: Shared) -> axum::Router {

    use axum::routing::get;

    let hsts = headers::hsts(app.config.base_url(), app.config.server.hsts_max_age);

    axum::Router::new()

        .route("/health", get(web::home::health))

        .route("/metrics", get(metrics::metrics))

        .with_state(app)

        .layer(axum::middleware::from_fn_with_state(hsts, headers::security_headers))

fn full_router(app: Shared) -> axum::Router {

    if http.is_none() && https.is_none() && app.roles.web {

    tracing::info!("roles: {}", app.roles.names().join(", "));

        // A worker with no address to serve on runs until it is stopped.

        r = tasks.join_next(), if !tasks.is_empty() => { r.transpose()?.transpose()?; Ok(()) }

    Serve {

        /// What this process does: web (pages, API, wire protocol, runner

        /// API), worker (the background tasks), or both, comma-separated.

        /// Overrides [server] roles; every role when neither is given

        /// (docs/roles.md)

        #[arg(long = "role", value_name = "ROLE", value_delimiter = ',')]

        roles: Vec<oxygenhub::config::Role>,

    },

        Cmd::Serve { roles } => {

            let mut config = Config::load(&cli.data)?;

            if !roles.is_empty() {

                config.server.roles = roles.iter().map(|r| r.name().to_string()).collect();

            }

    /// Whether this process writes the log: the web role's (#272). A

    /// process without it only reads what the log held when it opened.

    writer: bool,

        Self::open_as(dir, origin, true)

    }

    /// Open the log to read only, as a process without the web role does

    /// (#272): it neither appends nor signs, and leaves the files as they

    /// are, so it never cuts short an append the writer is making. Until

    /// the log has a cluster-wide writer (#274), one process per data

    /// directory has the web role.

    pub fn open_reader(dir: &Path, origin: &str) -> anyhow::Result<Log> {

        Self::open_as(dir, origin, false)

    }

    fn open_as(dir: &Path, origin: &str, writer: bool) -> anyhow::Result<Log> {

        }

        if !writer {

            // A reader takes the leaves whose hashes are all there and

            // changes nothing.

            let files = Files { entries, offsets, hashes, size: whole, entries_len };

            return Ok(Log {

                dir: dir.to_path_buf(),

                origin: origin.to_string(),

                files: Mutex::new(files),

                checkpoint: Mutex::new(Self::read_checkpoint(dir, whole)),

                witnesses: Default::default(),

                witness_for: Default::default(),

                writer,

            });

        offsets.set_len(size * 8)?;

        entries.set_len(entries_len)?;

            checkpoint: Mutex::new(Self::read_checkpoint(dir, size)),

            writer,

    /// The signed checkpoint kept in `dir`, if it is of at most `size` entries.

    fn read_checkpoint(dir: &Path, size: u64) -> Option<(u64, String)> {

        std::fs::read_to_string(dir.join("checkpoint")).ok().and_then(|note| {

            let n = tlog::Note::parse(&note).ok()?;

            let c = Checkpoint::parse(&n.text).ok()?;

            (c.size <= size).then_some((c.size, note))

        })

    }

        if !self.writer {

            anyhow::bail!("this process does not write the log: it has no web role");

        }

        if !self.writer {

            return Ok(false);

        }

    #[test]

    fn a_reader_leaves_the_files_alone_and_cannot_append() {

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

        let log = Log::open(dir.path(), "hub.example/log").unwrap();

        log.append(&[entry(0), entry(1)]).unwrap();

        // A torn append: an offset whose entry never arrived.

        std::fs::OpenOptions::new().append(true).open(dir.path().join("offsets")).unwrap().write_all(&9999u64.to_le_bytes()).unwrap();

        let before = std::fs::metadata(dir.path().join("offsets")).unwrap().len();

        let reader = Log::open_reader(dir.path(), "hub.example/log").unwrap();

        assert_eq!(reader.size(), 2);

        assert_eq!(std::fs::metadata(dir.path().join("offsets")).unwrap().len(), before, "the writer's files are untouched");

        assert!(reader.append(&[entry(2)]).is_err());

        assert_eq!(reader.tree_hash(2).unwrap(), log.tree_hash(2).unwrap());

    }

        // What this process does, and the singleton tasks it leads (#272).

        "roles": app.roles.names(),

        "leading": app.leader.leading(),

Status: design, 2026-10-19 (overnight loop 17). H2 (`--role` and the

leader lock, #272) is built; the event bus (H3) is next.

| `tlog::checkpointer` | signs checkpoints, gathers witness cosignatures | web, beside the log's appends, until #274 | **yes** | the log's single writer (below) |

`leader::run(app, name, task)` (`crates/hub/src/leader.rs`): a process

runs a singleton task only while it holds a lock named for it. On

Postgres, `pg_try_advisory_lock(key)`, the key a stable 64-bit hash

(BLAKE3) of the hub's schema and the task's name, so two hubs in two

schemas of one database never share a lock. Every lock of one process is

held on one connection of its own, outside the pool, for the process's

life. A process that does not hold a lock asks again every 2 seconds. One

that holds it checks the connection every 2 seconds (one `SELECT 1`

answers for every task); if it has gone, the server has freed its locks,

so each task it led is stopped (its future dropped) and the process

contends again on a new connection. A process that dies closes its

connection and frees its locks; another takes over within seconds (2 s in

the test). A holder whose connection drops can overlap its successor for

at most those 2 seconds, which the tasks tolerate. On SQLite (one machine

by definition) the lock is always held. Each task is safe to run again

after a failover: they read their work from the database. A task that

ends by itself (`diffusion.backfill`) gives its lock back; the next

worker to take it runs it again, harmlessly.

The locks: `hooks.worker`, `flow.reaper`, `flow.scheduler`,

`flow.placement`, `maintenance.nightly`, `backups.scheduler`,

`diffusion.backfill`, `diffusion.reconciler` (worker role) and

`tlog.checkpointer` (web role, below). `images.learn_archs` runs in every

worker, without a lock.

`/health` says what the process does and which locks it holds:

`"roles": ["web", "worker"], "leading": ["backups.scheduler", …]`.

## Running the roles apart (H2)

`oxygenhub serve --role web,worker` (or `[server] roles` in `hub.toml`,

`OXYHUB_SERVER_ROLES`); every role when none is given. An unknown role is

an error.

- **web** serves the whole router. It also owns the transparency log: it

  appends to it (`after_push`, kept receipts and approvals) and runs its

  checkpointer, under the `tlog.checkpointer` lock. The checkpointer moved

  from the worker's list above because the log's state is in the process

  that appends: a checkpointer elsewhere would sign a log it never sees

  grow, and the web process would have no checkpoint to serve.

- **worker** starts the background tasks, and serves only `/health` and

  `/metrics` (when `metrics_token` is set) on its bind address, so it can

  be watched; with no bind address it serves nothing. It opens the

  transparency log to read only: it never cuts short an append the web

  process is making.

- **Until the log has a cluster-wide writer (#274), only one process per

  data directory may have the web role.** Any number may be workers.

  Webhook nudges from a push reach only the process that took it, so a

  separate worker delivers on its once-a-minute pass until the event bus

  (H3) carries the nudge.

   locks; always held on SQLite). **Built (#272):** two hubs on one

   Postgres run each singleton on exactly one, and the other takes over

   within seconds when it stops or loses its connection. One web process

   per data directory until step 3.

3. The transparency log's appends under a cluster-wide lock (#274), and

   its checkpointer back with the workers.

roles = ["web", "worker"]       # what this process does (docs/roles.md); empty or absent: both.

                                # `serve --role` overrides it. web: pages, API, wire protocol,

                                # runner API, the transparency log; worker: background tasks,

                                # singletons under a leader lock, serving only /health and /metrics.

                                # One web process per data directory until #274.

oxygenhub serve --data DIR [--role web,worker] run, taking these roles (all by default)