alkblobs 0.1.0

//! The orphan-recovery sweep (ADR-009 §5): the legacy/bypass class —
//! committed LOs without companion rows — candidate-enumerated from
//! `pg_largeobject` (the sole sanctioned read; candidate list only),
//! validated against `lo_entries.loid` (the truth), the unclaimed
//! unlinked in one tx. Declaration-gated per ADR-016:
//! [`PgOpts::orphan_reap_permitted`] (default false) is the operator's
//! assertion that this database hosts no other large-object consumer —
//! [`PgLoBackend::reap_orphans`] refuses with a named error while it is
//! absent, because `pg_largeobject` is database-level while the sweep's
//! claim probe is companion-table-scoped.

use deadpool_postgres::Object;
use tokio_postgres::error::SqlState;

use super::PgLoBackend;
use crate::pg_common::{err, pg_err};
use crate::Error;

#[allow(dead_code)]
impl PgLoBackend {
    /// The crate-internal orphan-recovery sweep (ADR-009 §5): reap the
    /// legacy/bypass class — committed LOs without companion rows. One
    /// tx: candidates enumerate `pg_largeobject`'s distinct oids
    /// (candidate list only), each re-validated at delete time against
    /// `lo_entries.loid`, the unclaimed unlinked. Returns the count
    /// reaped.
    ///
    /// **Declaration-gated (ADR-016).** An engine constructed with
    /// `orphan_reap_permitted = false` (the default) refuses with a
    /// named error: `pg_largeobject` is database-level, so the sweep
    /// can reach another large-object consumer's committed LOs — the
    /// flag is the operator's assertion that no such consumer exists.
    ///
    /// **The missing-LO pass (ADR-016's riding fix).** The candidate
    /// snapshot can be stale at delete time — a loid may vanish between
    /// the snapshot and its unlink (a concurrent same-engine delete, or
    /// a foreign consumer's lifecycle): the READ COMMITTED race a plain
    /// unlink would turn into a spurious whole-sweep failure, because a
    /// failed statement aborts the tx and every subsequent statement
    /// dies with it. Each unlink therefore rides a savepoint: the
    /// SQLSTATE 42704 (`UNDEFINED_OBJECT` — "large object NNNN does not
    /// exist") pass rolls back to the savepoint and continues, and is
    /// not counted as reaped — the sweep tolerates it and reports
    /// honestly; candidates are re-validated at delete time, never
    /// trusted from the snapshot. Any other unlink failure is a media
    /// failure and aborts the sweep.
    pub(crate) async fn reap_orphans(&self) -> Result<u64, Error> {
        if !self.orphan_reap_permitted {
            return Err(Error::Io(
                "postgres: orphan reap refused (ADR-016): pg_largeobject is \
                 database-level, so this sweep can unlink another large-object \
                 consumer's committed LOs co-tenning this database — construct \
                 the engine with orphan_reap_permitted = true, the operator's \
                 assertion that this database hosts no other large-object \
                 consumer (deployments needing two consumers use two databases)"
                    .to_owned(),
            ));
        }
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        let reaped = match client
            .query("SELECT DISTINCT loid FROM pg_largeobject", &[])
            .await
            .map_err(pg_err)
        {
            Ok(rows) => {
                let candidates: Vec<u32> = rows.iter().map(|row| row.get::<_, u32>(0)).collect();
                match self.reap_orphans_tx(&client, &candidates).await {
                    Ok(n) => match client.batch_execute("COMMIT").await {
                        Ok(()) => Ok(n),
                        Err(e) => {
                            let _ = client.batch_execute("ROLLBACK").await;
                            Err(pg_err(e))
                        }
                    },
                    Err(e) => {
                        let _ = client.batch_execute("ROLLBACK").await;
                        Err(e)
                    }
                }
            }
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Err(e)
            }
        };
        drop(client);
        reaped
    }

    async fn reap_orphans_tx(&self, client: &Object, candidates: &[u32]) -> Result<u64, Error> {
        let claim_sql = format!("SELECT 1 FROM {} WHERE loid = $1", self.table);
        let claim_stmt = client.prepare_cached(&claim_sql).await.map_err(pg_err)?;
        let unlink_stmt = client
            .prepare_cached("SELECT lo_unlink($1)")
            .await
            .map_err(pg_err)?;
        let mut reaped = 0u64;
        for &loid in candidates {
            let claimed = client
                .query_opt(&claim_stmt, &[&loid])
                .await
                .map_err(pg_err)?
                .is_some();
            if claimed {
                continue;
            }
            client
                .batch_execute("SAVEPOINT orphan_reap_unlink")
                .await
                .map_err(pg_err)?;
            match client.query_one(&unlink_stmt, &[&loid]).await {
                Ok(_) => {
                    client
                        .batch_execute("RELEASE SAVEPOINT orphan_reap_unlink")
                        .await
                        .map_err(pg_err)?;
                    reaped += 1;
                }
                Err(e) if e.code() == Some(&SqlState::UNDEFINED_OBJECT) => {
                    client
                        .batch_execute("ROLLBACK TO SAVEPOINT orphan_reap_unlink")
                        .await
                        .map_err(pg_err)?;
                }
                Err(e) => {
                    let _ = client
                        .batch_execute("ROLLBACK TO SAVEPOINT orphan_reap_unlink")
                        .await;
                    return Err(pg_err(e));
                }
            }
        }
        Ok(reaped)
    }
}

#[cfg(test)]
mod tests {
    use crate::backend::Backend;
    use crate::pg_common::PgOpts;
    use crate::pg_lo::testing::{harness, harness_with, stored_value, Harness, CONTENT};
    use crate::pg_lo::INV_WRITE;

    /// A harness engine carrying the ADR-016 declaration — the sweep
    /// tests' fixture.
    async fn reap_harness(tag: &str) -> Option<Harness> {
        harness_with(
            tag,
            PgOpts {
                orphan_reap_permitted: true,
                ..PgOpts::default()
            },
        )
        .await
    }

    /// A committed plant (companion row + LO — the sweep's protected
    /// class) plus a planted orphan (`lo_create` + `lowrite` + COMMIT,
    /// no companion row — the legacy/bypass class crash-free stores
    /// never produce). Returns the committed key and the orphan's loid.
    async fn plant_committed_and_orphan(backend: &super::PgLoBackend) -> (Vec<u8>, u32) {
        let committed_key = b"\x01committed-key".to_vec();
        backend.put(&committed_key, CONTENT).await.unwrap();

        let client = backend.pool.get().await.unwrap();
        client.batch_execute("BEGIN").await.unwrap();
        let loid: u32 = client
            .query_one("SELECT lo_create(0) AS loid", &[])
            .await
            .unwrap()
            .get(0);
        let fd: i32 = client
            .query_one("SELECT lo_open($1, $2)", &[&loid, &INV_WRITE])
            .await
            .unwrap()
            .get(0);
        client
            .query_one("SELECT lowrite($1, $2)", &[&fd, &b"orphan bytes".to_vec()])
            .await
            .unwrap();
        client.batch_execute("COMMIT").await.unwrap();
        (committed_key, loid)
    }

    /// The orphan-recovery sweep (ADR-009 §5, the C6/pgdiag10 shape)
    /// with the ADR-016 declaration carried: a planted orphan is reaped
    /// by `reap_orphans`; committed content survives the sweep
    /// byte-exact; nothing is reaped when there are no orphans.
    #[tokio::test]
    async fn orphan_recovery_sweep_reaps_the_bypass_class() {
        let Some(backend) = reap_harness("orphan").await else {
            return;
        };
        let (committed_key, orphan) = plant_committed_and_orphan(&backend).await;

        assert_eq!(
            backend.reap_orphans().await.unwrap(),
            1,
            "exactly the planted orphan is reaped"
        );
        let client = backend.pool.get().await.unwrap();
        let orphan_pages: i64 = client
            .query_one(
                "SELECT count(*) FROM pg_largeobject WHERE loid = $1",
                &[&orphan],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(orphan_pages, 0, "the orphan's pages are unlinked");
        drop(client);

        assert_eq!(
            stored_value(&backend, &committed_key).await.as_deref(),
            Some(CONTENT),
            "committed content survives the sweep byte-exact (validated against lo_entries)"
        );
        assert_eq!(
            backend.reap_orphans().await.unwrap(),
            0,
            "no orphans remain: committed LOs are never reaped"
        );
    }

    /// The declaration gate (ADR-016): an engine constructed without
    /// `orphan_reap_permitted` (the default) refuses the sweep with the
    /// named error — and nothing runs (the catalog is untouched), so an
    /// undeclared sweep can never silently reach a co-tenant consumer's
    /// committed LOs. The hostile co-tenancy shape this bounds is the
    /// ADR's §3: schema isolation of rows is unaffected; reaping is
    /// what the declaration gates.
    #[tokio::test]
    async fn the_sweep_refuses_without_the_orphan_reap_declaration() {
        let Some(backend) = harness("reap-refused").await else {
            return;
        };
        let (_committed_key, _orphan) = plant_committed_and_orphan(&backend).await;

        let err = backend.reap_orphans().await.unwrap_err();
        assert!(
            err.to_string().contains("orphan reap refused"),
            "the refusal names itself: {err}"
        );
        assert!(
            err.to_string().contains("orphan_reap_permitted"),
            "the refusal names the declaration: {err}"
        );

        let client = backend.pool.get().await.unwrap();
        let loids: i64 = client
            .query_one("SELECT count(DISTINCT loid) FROM pg_largeobject", &[])
            .await
            .unwrap()
            .get(0);
        assert_eq!(
            loids, 2,
            "nothing ran: the refused sweep never opened its tx (committed + orphan both stand)"
        );
    }

    /// The claim probe's supporting index (ADR-016's riding fix):
    /// `lo_entries(loid)` exists after construction, and an existing
    /// database gains it on next construction (the idempotent DDL — the
    /// index dropped by hand comes back at the next open).
    #[tokio::test]
    async fn the_loid_claim_index_exists_and_an_existing_database_gains_it_on_next_construction() {
        let Some(backend) = reap_harness("reap-index").await else {
            return;
        };
        assert!(
            loid_index_def(&backend).await.contains("(loid)"),
            "the index is over (loid)"
        );

        {
            let client = backend.pool.get().await.unwrap();
            client
                .batch_execute(&format!(
                    "DROP INDEX {}.lo_entries_loid_idx",
                    crate::pg_common::quote_identifier(&backend.schema)
                ))
                .await
                .unwrap();
        }

        let second = super::PgLoBackend::new(&backend.dsn, PgOpts::default())
            .await
            .expect("re-open over the existing database");
        assert!(
            loid_index_def(&second).await.contains("(loid)"),
            "an existing database gains the index on next construction"
        );
    }

    /// The missing-LO pass (ADR-016's riding fix), driven
    /// deterministically the way the harness allows: a stale candidate
    /// list — a loid that was in the sweep's snapshot but vanished
    /// before its unlink. A concurrent same-engine delete and a foreign
    /// consumer's lifecycle present the identical shape at delete time
    /// (the row is unclaimed, the LO is gone): the sweep continues past
    /// it and reports honestly — the vanished loid is not counted, the
    /// committed class survives, the tx is alive to the end.
    #[tokio::test]
    async fn a_loid_vanished_between_snapshot_and_unlink_is_a_pass_not_an_abort() {
        let Some(backend) = reap_harness("reap-vanished").await else {
            return;
        };
        let (committed_key, orphan) = plant_committed_and_orphan(&backend).await;

        let client = backend.pool.get().await.unwrap();
        let candidates: Vec<u32> = {
            let rows = client
                .query("SELECT DISTINCT loid FROM pg_largeobject", &[])
                .await
                .unwrap();
            rows.iter().map(|row| row.get::<_, u32>(0)).collect()
        };
        assert_eq!(candidates.len(), 2, "the snapshot sees both loids");

        // The concurrent delete racing the sweep: the orphan's loid
        // vanishes (a committed unlink) after the snapshot was taken —
        // the stale candidate the sweep must pass over.
        client
            .query_one("SELECT lo_unlink($1)", &[&orphan])
            .await
            .unwrap();

        let reaped = {
            client.batch_execute("BEGIN").await.unwrap();
            let reaped = backend
                .reap_orphans_tx(&client, &candidates)
                .await
                .expect("the stale candidate passes the sweep, never aborts it");
            client.batch_execute("COMMIT").await.unwrap();
            reaped
        };
        assert_eq!(
            reaped, 0,
            "the report is honest: the vanished loid is a pass, the committed loid is claimed — nothing reaped"
        );
        let orphan_pages: i64 = client
            .query_one(
                "SELECT count(*) FROM pg_largeobject WHERE loid = $1",
                &[&orphan],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(orphan_pages, 0, "the vanished orphan stays gone");
        assert_eq!(
            stored_value(&backend, &committed_key).await.as_deref(),
            Some(CONTENT),
            "the sweep tx reached its end with the committed class intact"
        );
    }

    /// The (loid) index's definition, asserted present — the index
    /// probes' helper.
    async fn loid_index_def(backend: &super::PgLoBackend) -> String {
        let client = backend.pool.get().await.unwrap();
        let rows = client
            .query(
                "SELECT indexdef::text FROM pg_indexes \
                 WHERE schemaname = $1 AND tablename = 'lo_entries' \
                 AND indexname = 'lo_entries_loid_idx'",
                &[&backend.schema],
            )
            .await
            .unwrap();
        assert_eq!(rows.len(), 1, "exactly the (loid) index exists: {rows:?}");
        rows[0].get::<_, String>(0)
    }
}