One auto-land and one Diffusion bump at a time across hub processes (closes #275) 356183b71fab

- **One auto-land and one Diffusion bump at a time across hub processes (#275):** auto-land's check of a proposal, and Diffusion's bump and landing of a consumer's pin, were serialised only within a process, so two web processes could land one proposal twice. Each now holds a lock named for the proposal or the pin across every process sharing the database (a PostgreSQL advisory lock; on SQLite, one process, as before).

    // And one at a time for this proposal across the hub's processes

    // (#275): two web nodes recording its last checks land it once.

    let Ok(_proposal) = crate::leader::exclusive(&app.db, &format!("autoland:{}", p.id)).await else { return };

        // One bump or landing of this pin at a time across processes (#275).

        let Ok(_pin) = crate::leader::exclusive(&app.db, &format!("diffusion:{}:{path}", consumer.id)).await else { continue };

        // One landing of this pin at a time across processes (#275).

        let Ok(_pin) = crate::leader::exclusive(&app.db, &format!("diffusion:{}:{path}", consumer.id)).await else { return };

/// A lock across every process sharing the hub's database, held while the

/// guard lives (#275): one evaluation of a proposal's auto-land, one bump

/// or landing of a consumer's pin, at a time anywhere. On PostgreSQL a

/// transaction-scoped advisory lock (`pg_advisory_xact_lock`), freed when

/// the guard drops and its transaction ends, or with its connection; on

/// SQLite (one process by definition) a lock in this process.

pub struct Exclusive {

    #[cfg(feature = "postgres")]

    _tx: sqlx::Transaction<'static, sqlx::Postgres>,

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

    _guard: tokio::sync::OwnedMutexGuard<()>,

}

/// Wait for the lock named `name`, then hold it until the guard drops.

#[cfg(feature = "postgres")]

pub async fn exclusive(pool: &crate::db::Pool, name: &str) -> sqlx::Result<Exclusive> {

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

    let schema: String = sqlx::query_scalar("SELECT CAST(current_schema() AS TEXT)").fetch_one(&mut *tx).await?;

    sqlx::query("SELECT pg_advisory_xact_lock($1)").bind(pg::key(&schema, name)).execute(&mut *tx).await?;

    Ok(Exclusive { _tx: tx })

}

/// Wait for the lock named `name`, then hold it until the guard drops.

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

pub async fn exclusive(_pool: &crate::db::Pool, name: &str) -> sqlx::Result<Exclusive> {

    static LOCAL: Mutex<Option<std::collections::HashMap<String, std::sync::Arc<tokio::sync::Mutex<()>>>>> = Mutex::new(None);

    let lock = {

        let mut map = LOCAL.lock().unwrap_or_else(|p| p.into_inner());

        let map = map.get_or_insert_with(Default::default);

        // Locks nobody holds or waits for are forgotten.

        map.retain(|_, l| std::sync::Arc::strong_count(l) > 1);

        map.entry(name.to_string()).or_default().clone()

    };

    Ok(Exclusive { _guard: lock.lock_owned().await })

}

/// oxygenhub#275: a named lock is held by one holder at a time across the

/// database's sessions (on SQLite, the process); other names are free.

async fn an_exclusive_lock_has_one_holder_at_a_time() {

    hermetic();

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

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

    let held = oxygenhub::leader::exclusive(&app.db, "autoland:7").await.unwrap();

    let (tx, mut rx) = tokio::sync::oneshot::channel();

    let db = app.db.clone();

    tokio::spawn(async move {

        let g = oxygenhub::leader::exclusive(&db, "autoland:7").await.unwrap();

        let _ = tx.send(());

        drop(g);

    });

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

    assert!(rx.try_recv().is_err(), "the second holder waits");

    let other = tokio::time::timeout(Duration::from_secs(2), oxygenhub::leader::exclusive(&app.db, "autoland:8")).await;

    assert!(other.is_ok(), "another name is free");

    drop(other);

    drop(held);

    assert!(tokio::time::timeout(Duration::from_secs(5), rx).await.is_ok(), "and takes it once it is let go");

}

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

  Built (#275): `leader::exclusive("autoland:ID")`, a transaction-scoped

  advisory lock per proposal, held for each evaluation, beside the TURN.

  push: idempotent by design, but racing; built (#275): each bump and

  landing of a consumer's pin holds `diffusion:CONSUMER:PATH`.