alkblobs 0.2.1

//! The joint tx and the sweep lease: `PgJointTx` (the pooled checkout
//! held for the tx's life, the session-level `BEGIN` the clause bodies
//! close), the joint tx's entry half (`begin_joint_tx`/`joint_put`),
//! and the sweep lease (the session-scoped pg advisory lock whose
//! acquired checkout is mutex-held until release); the joint tx's
//! cancellation Drop guard and the accessor live on the type itself
//! (`src/postgres/mod.rs`).

/// The sweep lease's advisory-lock key: FNV-1a over the namespaced
/// string `alkblobs:sweep:{schema}`, folded to positive i64. Deterministic
/// across processes and nodes — the fleet's shared schema derives one
/// key (one sweeper per pool), co-tenant schemas derive distinct keys.
pub(super) fn sweep_lock_key(schema: &str) -> i64 {
    let mut state: u64 = 0xcbf2_9ce4_8422_2325;
    for byte in format!("alkblobs:sweep:{schema}").as_bytes() {
        state ^= u64::from(*byte);
        state = state.wrapping_mul(0x0000_0100_0000_01b3);
    }
    (state & !(1_u64 << 63)) as i64
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::Backend;
    use std::sync::Arc;
    use std::time::{Duration, SystemTime};

    use parking_lot::Mutex;

    use crate::gc::EngineFleetState;
    use crate::gc::EngineWindowState;
    use crate::pg_common::time_to_nanos;
    use crate::pg_common::PgOpts;
    use crate::postgres::pool::DEFAULT_PIN_TTL;
    use crate::postgres::testing::{
        arc_backend, harness_dsn, harness_gc, harness_with, pin_row, pin_rows, raw_connection,
        stored_value, CONTENT,
    };
    use crate::postgres::PostgresBackend;
    use crate::{Error, PinState};

    /// The joint entry+pin transaction publishes together: the entry row
    /// and its pin row commit in one step (pin-before-publish, fleet
    /// form), the joint put rides the same first-writer-wins CAS
    /// semantics as the public `put` (a re-put is dedup), and a re-put's
    /// pin row upsert renews the same instance's row.
    #[tokio::test]
    async fn joint_put_and_pin_commit_publish_together() {
        let Some(backend) = harness_gc("joint").await else {
            return;
        };
        let key = b"\x01joint-key".to_vec();
        let owner = b"\x01instance-id".to_vec();

        let mut tx = backend.begin_joint_tx().await.unwrap();
        EngineFleetState::joint_put(&*backend, &mut tx, &key, CONTENT)
            .await
            .unwrap();
        let expiry = SystemTime::now() + DEFAULT_PIN_TTL;
        backend
            .pin_commit(tx, &[pin_row(&key, &owner, expiry)])
            .await
            .unwrap();
        assert!(backend.has(&key).await.unwrap(), "entry published");
        assert_eq!(
            backend.pin_check(&key).await.unwrap(),
            PinState::Protected,
            "the pin row landed with the entry"
        );
        assert_eq!(stored_value(&backend, &key).await.as_deref(), Some(CONTENT));

        let before = pin_rows(&backend).await;
        assert_eq!(
            before,
            vec![(key.clone(), owner.clone(), time_to_nanos(expiry))],
            "the pin row's identity is (key, owner), expiry the store-resolved one"
        );

        let mut tx = backend.begin_joint_tx().await.unwrap();
        EngineFleetState::joint_put(&*backend, &mut tx, &key, b"overwritten")
            .await
            .unwrap();
        let renewed = SystemTime::now() + DEFAULT_PIN_TTL + Duration::from_secs(7);
        backend
            .pin_commit(tx, &[pin_row(&key, &owner, renewed)])
            .await
            .unwrap();
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(CONTENT),
            "joint re-put does not overwrite (first writer wins)"
        );
        assert_eq!(
            pin_rows(&backend).await,
            vec![(key, owner, time_to_nanos(renewed))],
            "the re-put's pin row upsert renews the same instance's row"
        );
    }

    /// The joint tx atomicity (the stage-hygiene invariant's tx form): a
    /// forced failure at `pin_commit` aborts the entire tx — no pin
    /// rows, no published entry, zero residue.
    #[tokio::test]
    async fn pin_commit_failure_aborts_with_zero_residue() {
        let Some(backend) = harness_gc("joint-fail").await else {
            return;
        };
        let key = b"\x01joint-failure-key".to_vec();
        let owner = b"\x01instance-id".to_vec();

        let failing = PostgresBackend {
            pool: backend.pool.clone(),
            table: backend.table.clone(),
            schema: backend.schema.clone(),
            pins_table: backend.pins_table.clone(),
            window_table: backend.window_table.clone(),
            pin_ttl: backend.pin_ttl,
            lease: Mutex::new(None),
            liveness_sources: Arc::new(Mutex::new(Vec::new())),
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_joint_commit: true,
            #[cfg(test)]
            fail_window_stage: false,
        };
        let mut tx = failing.begin_joint_tx().await.unwrap();
        EngineFleetState::joint_put(&failing, &mut tx, &key, CONTENT)
            .await
            .unwrap();
        let expiry = SystemTime::now() + DEFAULT_PIN_TTL;
        let err = failing
            .pin_commit(tx, &[pin_row(&key, &owner, expiry)])
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::Io(_)),
            "engine joint media failure surfaces as Io"
        );

        assert!(
            !backend.has(&key).await.unwrap(),
            "the aborted tx published nothing"
        );
        assert_eq!(
            backend.pin_check(&key).await.unwrap(),
            PinState::Unprotected,
            "no pin row survived"
        );
        assert_eq!(pin_rows(&backend).await, vec![], "zero residue in gc_pins");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "zero residue in the blobs table itself"
        );
    }

    /// The sweep lease (ADR-013 §2's named exclusive semantics, shipped
    /// as the session-scoped pg advisory lock): atomic test-and-set
    /// (`Ok(true)` = held), one sweep per backend and per schema — a
    /// second acquire and a raw session's try both contend — release is
    /// no-op-safe, and a release actually frees the lock for the next
    /// acquirer (the same-connection discipline did its job).
    #[tokio::test]
    async fn sweep_lock_acquire_contention_release() {
        let Some(backend) = harness_gc("sweep-lock").await else {
            return;
        };

        assert!(
            backend.sweep_lock_acquire().await.unwrap(),
            "first acquire: held"
        );
        assert!(
            !backend.sweep_lock_acquire().await.unwrap(),
            "second acquire on the held backend contended"
        );

        let dsn = harness_dsn().unwrap();
        let (raw, _conn) = raw_connection(&dsn).await;
        let key = sweep_lock_key(&backend.schema);
        let held = raw
            .query_one("SELECT pg_try_advisory_lock($1)", &[&key])
            .await
            .unwrap()
            .get::<_, bool>(0);
        assert!(!held, "a concurrent session's try contends on the same key");

        backend.sweep_lock_release().await.unwrap();
        backend.sweep_lock_release().await.unwrap();
        assert!(
            backend.sweep_lock_acquire().await.unwrap(),
            "acquire after release: held again (release freed the lock)"
        );
        backend.sweep_lock_release().await.unwrap();
    }

    /// Session-loss reaping: a lock held by a foreign session reaps at
    /// that session's death without any release call — the scope whose
    /// holdership ends at engine-session loss (ADR-013 §2).
    #[tokio::test]
    async fn sweep_lock_session_loss_reaps_the_lease() {
        let Some(backend) = harness_gc("sweep-reap").await else {
            return;
        };
        let dsn = harness_dsn().unwrap();
        let key = sweep_lock_key(&backend.schema);
        let (holder, conn) = raw_connection(&dsn).await;
        let held = holder
            .query_one("SELECT pg_try_advisory_lock($1)", &[&key])
            .await
            .unwrap()
            .get::<_, bool>(0);
        assert!(held, "the foreign session holds the lock");

        assert!(
            !backend.sweep_lock_acquire().await.unwrap(),
            "the backend's acquire contends while the foreign session lives"
        );

        drop(holder);
        let _ = conn.await;
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        assert!(
            backend.sweep_lock_acquire().await.unwrap(),
            "session loss reaped the lease — the backend now holds it"
        );
        backend.sweep_lock_release().await.unwrap();
    }

    /// In-instance session-loss reaping (ADR-013 §2): a lease whose
    /// *own* held session dies is reaped at the next acquire — no
    /// release call is owed first. The lease's held checkout is
    /// terminated server-side (`pg_terminate_backend`), so its
    /// advisory lock is gone; the mutex-held closed session must not
    /// block re-acquisition forever.
    #[tokio::test]
    async fn sweep_lock_in_instance_session_loss_reaps_the_lease() {
        let Some(backend) = harness_gc("sweep-reap-own").await else {
            return;
        };

        assert!(backend.sweep_lock_acquire().await.unwrap(), "held");
        let held_client = backend
            .lease
            .lock()
            .take()
            .expect("the acquire stored its checkout");
        let pid = held_client
            .query_one("SELECT pg_backend_pid()", &[])
            .await
            .unwrap()
            .get::<_, i32>(0);
        *backend.lease.lock() = Some(held_client);

        let dsn = harness_dsn().unwrap();
        let (raw, _conn) = raw_connection(&dsn).await;
        raw.query_one("SELECT pg_terminate_backend($1)", &[&pid])
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        assert!(
            backend.sweep_lock_acquire().await.unwrap(),
            "the dead in-instance lease is reaped without a release call"
        );
        assert!(
            !backend.sweep_lock_acquire().await.unwrap(),
            "the re-acquired lease contends again"
        );
        backend.sweep_lock_release().await.unwrap();
    }

    /// The cancellation guard's named test shape (review-001 F-2):
    /// cancel-the-future-mid-joint-tx — a task that aborts holding a
    /// begun joint tx (its entry put inside the open transaction, the
    /// future parked before `pin_commit`) strands the open tx; the
    /// guard's spawned ROLLBACK clears it and the sole pooled session
    /// (`max_size: 1`) comes back clean — the next checkout's
    /// `BEGIN`/`COMMIT` joint tx completes over the very session the
    /// cancelled task held, publishing nothing of the stranded put.
    #[tokio::test]
    async fn cancel_mid_joint_tx_leaves_the_next_checkout_clean() {
        let Some(backend) = harness_with(
            "joint-cancel",
            PgOpts {
                max_size: 1,
                ..PgOpts::default()
            },
        )
        .await
        else {
            return;
        };
        backend.ensure_gc_schema().await.expect("gc schema");
        let stranded_key = b"\x01joint-cancel-key".to_vec();
        let holder = arc_backend(&backend);
        let put_key = stranded_key.clone();

        let (begun_tx, tx_begun) = tokio::sync::oneshot::channel();
        let task = tokio::spawn(async move {
            let mut tx = holder.begin_joint_tx().await.unwrap();
            EngineFleetState::joint_put(&*holder, &mut tx, &put_key, CONTENT)
                .await
                .unwrap();
            let _ = begun_tx.send(());
            std::future::pending::<()>().await;
        });
        tx_begun.await.expect("the task reached the begun joint tx");
        task.abort();
        let _ = task.await;

        let mut clean = backend.begin_joint_tx().await.unwrap();
        let clean_key = b"\x01joint-cancel-clean-key".to_vec();
        let clean_owner = b"\x01joint-cancel-clean-owner".to_vec();
        EngineFleetState::joint_put(&*backend, &mut clean, &clean_key, CONTENT)
            .await
            .unwrap();
        backend
            .pin_commit(
                clean,
                &[pin_row(
                    &clean_key,
                    &clean_owner,
                    SystemTime::now() + DEFAULT_PIN_TTL,
                )],
            )
            .await
            .unwrap();

        assert!(
            !backend.has(&stranded_key).await.unwrap(),
            "the cancelled tx published nothing — its insert was rolled back"
        );
        assert!(
            backend.has(&clean_key).await.unwrap(),
            "the next checkout ran its joint tx (BEGIN/COMMIT) cleanly"
        );
        assert_eq!(
            pin_rows(&backend).await[0].0,
            clean_key,
            "the clean tx's pin row is the only residue in gc_pins"
        );
    }
}