alkblobs 0.1.0

//! The postgres engine's constructor family: the deadpool pool setup
//! (the shipped driver posture rides [`crate::pg_common::build_pool`]) with
//! the idempotent bootstrap DDL, the same-instance probe's
//! `system_identifier` read, and the shipped pin-TTL constants — the fleet
//! constructors' `pin_ttl` source. Prepared statements ride
//! `prepare_cached` per pooled session (the recorded discipline; the
//! module doc).

use std::sync::Arc;
use std::time::Duration;

use deadpool_postgres::Pool;
use parking_lot::Mutex;

use super::schema::bootstrap_ddl;
use super::PostgresBackend;
use crate::pg_common::{build_pool, err, pg_err, qualified_table, PgOpts, DEFAULT_SCHEMA};
use crate::Error;

impl PostgresBackend {
    /// Open (or create) the engine-owned schema and pool table and
    /// establish the shipped posture: the deadpool pool
    /// (`RecyclingMethod::Fast`, the alkstore precedent) with the
    /// per-session `synchronous_commit` knob merged into every pooled
    /// connection's connect options, then the idempotent bootstrap DDL
    /// ([`bootstrap_ddl`]) on a pool connection. Construction fails —
    /// never silently downgrades posture — when the server is
    /// unreachable, the credentials/config are unparseable, the pool
    /// cannot be built (`max_size = 0` fails at entry), or the DDL
    /// errors.
    pub(crate) async fn new(config: &str, opts: PgOpts) -> Result<Self, Error> {
        Self::new_with_pin_ttl(config, opts, DEFAULT_PIN_TTL).await
    }

    /// Open the engine with a constructor-carried pin TTL — the fleet
    /// constructors' instrument (wave 6's declaration landed): the
    /// store's fleet `pin_ttl` (ADR-008 §2, default [`DEFAULT_PIN_TTL`])
    /// configures the backend so [`EngineFleetState::pin_renew`]'s
    /// successor-time body reads the TTL it was constructed with. The
    /// non-fleet constructors ride [`Self::new`] (the shipped default;
    /// nothing renews under `fleet: false`, so the value is dormant
    /// there).
    pub(crate) async fn new_with_pin_ttl(
        config: &str,
        opts: PgOpts,
        pin_ttl: Duration,
    ) -> Result<Self, Error> {
        let pool = build_pool(config, &opts)?;
        let schema = if opts.schema.is_empty() {
            DEFAULT_SCHEMA.to_owned()
        } else {
            opts.schema
        };
        {
            let client = pool.get().await.map_err(err)?;
            client
                .batch_execute(&bootstrap_ddl(&schema))
                .await
                .map_err(err)?;
        }
        let pins_table = qualified_table(&schema, "gc_pins");
        let window_table = qualified_table(&schema, "gc_window");
        Ok(Self {
            pool,
            table: qualified_table(&schema, "blobs"),
            schema,
            pins_table,
            window_table,
            pin_ttl,
            lease: Mutex::new(None),
            liveness_sources: Arc::new(Mutex::new(Vec::new())),
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_joint_commit: false,
            #[cfg(test)]
            fail_window_stage: false,
        })
    }

    /// The pg instance's `system_identifier` (`pg_control_system()`,
    /// PG14+) — read across one pooled connection. The fleet
    /// constructor's same-instance probe's instrument (ADR-012 §4):
    /// the kv engine's and the pg-lo engine's identifiers compared —
    /// server-side instance identity, immune to DSN aliases. A probe
    /// failure surfaces as an error (the probe errors closed — never
    /// a silent pass).
    #[allow(dead_code)]
    pub(crate) async fn system_identifier(&self) -> Result<u64, Error> {
        Self::system_identifier_of(&self.pool).await
    }

    /// Test-only injection arm: force [`EngineFleetState::pin_commit`]'s
    /// failure form (the kv joint tx's terminal step) so the store-core
    /// driver's A1 propagation path is exercised through the public
    /// `put`. Mirrors the other engines' cfg(test) knobs.
    #[cfg(test)]
    pub(crate) fn inject_fail_joint_commit(&mut self) {
        self.fail_joint_commit = true;
    }

    /// Test-only disarm arm: clears the forced joint-commit failure so
    /// a follow-up put runs clean (the failure tests' pool-health
    /// assertions' driver).
    #[cfg(test)]
    pub(crate) fn clear_fail_joint_commit(&mut self) {
        self.fail_joint_commit = false;
    }

    /// The probe's body over a pool — shared with the pg-lo engine's
    /// twin.
    async fn system_identifier_of(pool: &Pool) -> Result<u64, Error> {
        let client = pool.get().await.map_err(err)?;
        let row = client
            .query_one("SELECT system_identifier FROM pg_control_system()", &[])
            .await
            .map_err(pg_err)?;
        let id: i64 = row.try_get(0).map_err(pg_err)?;
        Ok(id as u64)
    }
}

/// The pin-TTL constant's seconds ([`DEFAULT_PIN_TTL`]) — ADR-008's
/// default; the store re-exports it as the fleet constructors' shipped
/// `pin_ttl` value ([`crate::store::DEFAULT_PIN_TTL`]), and the fleet
/// constructors tune the backend's `pin_ttl` through
/// [`PostgresBackend::new_with_pin_ttl`].
pub(crate) const DEFAULT_PIN_TTL_SECS: u64 = 60;

/// The shipped pin-TTL ([`PostgresBackend::pin_renew`]'s successor-time
/// source): every renewed row's expiry moves to `now` + this.
pub(super) const DEFAULT_PIN_TTL: Duration = Duration::from_secs(DEFAULT_PIN_TTL_SECS);

#[cfg(test)]
mod tests {
    use super::*;

    use crate::postgres::testing::{
        fresh_schema, harness, harness_dsn, harness_with, shown_application_name,
        shown_synchronous_commit, skip_note,
    };

    /// The shipped `synchronous_commit` posture: the parity knob
    /// defaults to `on` (the ship config) on the pool's connections.
    #[tokio::test]
    async fn opens_with_synchronous_commit_on_by_default() {
        let Some(backend) = harness("sync-default").await else {
            return;
        };
        assert_eq!(
            shown_synchronous_commit(&backend).await,
            "on",
            "the shipped tier is synchronous_commit = on"
        );
    }

    /// The parity knob is configurable per engine instance: `off` wires
    /// onto the pooled connections (the documented
    /// sqlite-NORMAL-alike latency trade, B5 ~2-3x).
    #[tokio::test]
    async fn synchronous_commit_knob_wires_per_session() {
        let Some(backend) = harness_with(
            "sync-off",
            PgOpts {
                synchronous_commit: false,
                ..Default::default()
            },
        )
        .await
        else {
            return;
        };
        assert_eq!(
            shown_synchronous_commit(&backend).await,
            "off",
            "the instance's knob wires onto pooled connections"
        );
    }

    /// The shipped `application_name` tag posture: the field's default
    /// names the crate, and the pool's sessions carry it.
    #[tokio::test]
    async fn opens_with_the_default_crate_tag() {
        let Some(backend) = harness("app-name-default").await else {
            return;
        };
        assert_eq!(
            shown_application_name(&backend).await,
            "alkblobs",
            "the default tag is the crate-identifying name"
        );
    }

    /// The per-instance tag is configurable and wires onto the pool's
    /// sessions (the observability probe's scope boundary in tests =
    /// a unique tag per test instance).
    #[tokio::test]
    async fn application_name_tag_wires_per_session() {
        let Some(backend) = harness_with(
            "app-name-custom",
            PgOpts {
                application_name: "alkblobs-test-tag".to_owned(),
                ..Default::default()
            },
        )
        .await
        else {
            return;
        };
        assert_eq!(
            shown_application_name(&backend).await,
            "alkblobs-test-tag",
            "the instance's tag wires onto pooled connections"
        );
    }

    /// A `max_size` of zero fails construction at entry, before any
    /// round trip — deadpool cannot grant checkouts from an empty
    /// pool (the alkstore precedent).
    #[tokio::test]
    async fn constructor_rejects_zero_pool_size() {
        let Some(dsn) = harness_dsn() else {
            skip_note();
            return;
        };
        let opened = PostgresBackend::new(
            &dsn,
            PgOpts {
                schema: fresh_schema("zero-pool"),
                max_size: 0,
                ..Default::default()
            },
        )
        .await;
        match opened {
            Ok(_) => panic!("max_size 0 must fail at entry"),
            Err(Error::Io(message)) => assert!(
                message.contains("max_size must be positive"),
                "typed entry validation: {message}"
            ),
            Err(other) => panic!("max_size 0 must fail typed, got: {other}"),
        }
    }
}