reliar-store-postgres 0.9.0

//! Inbox contract §5.2 I-P13 — `claim` then `complete` through `PgDog`, the transaction-mode
//! pooler `outbox_pgdog.rs` already proves the URL-`options`/`ALTER ROLE` `search_path` paths
//! against (ADR 0021); this scenario does not repeat that proof. What is new here: `claim`'s
//! in-flight guard is a **transaction** advisory lock (`pg_try_advisory_xact_lock`), released by
//! the transaction's own commit/rollback rather than by the session — exactly the property a
//! transaction-mode pooler could break if it ever handed one client transaction's statements to
//! more than one backend connection. `claim` and `complete` spanning one transaction, end to end,
//! through the pooler is the proof that it does not.

use std::io::Write as _;

use reliar_core::MessageId;
use reliar_inbox::{InboxClaim, InboxHandler, InboxScope, InboxStore};
use reliar_store_postgres::{PostgresInboxSettings, PostgresInboxStore};

use crate::common::inbox::{InsertBusinessRow, business_row_count, create_business_table};

use sqlx::PgPool;
use testcontainers::core::{IntoContainerPort, Mount, WaitFor};
use testcontainers::runners::AsyncRunner;
use testcontainers::{GenericImage, ImageExt};
use testcontainers_modules::postgres::Postgres;

// Same pin as `outbox_pgdog.rs`, enforced by the same ci.yaml step — bump both together.
const PGDOG_IMAGE: &str = "ghcr.io/pgdogdev/pgdog";
const PGDOG_TAG: &str = "v0.1.46";

/// Writes `pgdog.toml` + `users.toml` into a fresh directory, exactly as `outbox_pgdog.rs`'s own
/// helper does — duplicated rather than shared, since the two scenario files run in different
/// test binaries with no common non-`common` module today.
fn write_pgdog_config(pg_host: &str) -> std::path::PathBuf {
    let dir = std::env::temp_dir().join(format!(
        "reliar-inbox-pgdog-{}",
        uuid::Uuid::now_v7().simple()
    ));

    std::fs::create_dir_all(&dir).expect("create pgdog config dir");

    let pgdog_toml = format!(
        r#"[general]
host = "0.0.0.0"
port = 6432
pooler_mode = "transaction"

[[databases]]
name = "postgres"
host = "{pg_host}"
port = 5432
database_name = "postgres"
user = "postgres"
"#
    );
    let users_toml = r#"[[users]]
name = "postgres"
database = "postgres"
password = "postgres"
"#;

    let mut f = std::fs::File::create(dir.join("pgdog.toml")).unwrap();
    f.write_all(pgdog_toml.as_bytes()).unwrap();
    let mut f = std::fs::File::create(dir.join("users.toml")).unwrap();
    f.write_all(users_toml.as_bytes()).unwrap();

    dir
}

async fn claim_and_complete_through_pgdog_in_one_transaction() {
    let network = format!("reliar-inbox-pgdog-{}", uuid::Uuid::now_v7().simple());
    let pg_name = format!("reliar-pg-{}", uuid::Uuid::now_v7().simple());

    let pg = Postgres::default()
        .with_tag("18-alpine")
        .with_container_name(&pg_name)
        .with_network(&network)
        .with_label("reliar.test", "true")
        .start()
        .await
        .expect("start postgres");
    let pg_direct_port = pg.get_host_port_ipv4(5432).await.expect("postgres port");
    let direct_url = format!("postgres://postgres:postgres@127.0.0.1:{pg_direct_port}/postgres");

    // DDL and the role-level `search_path` default run direct — see `outbox_pgdog.rs`'s own
    // comment on why `migrate()` must never go through a transaction-mode pooler.
    let direct_pool = PgPool::connect(&direct_url).await.expect("connect direct");

    reliar_store_postgres::migrate(
        &direct_pool,
        reliar_store_postgres::MigrateOptions::default(),
    )
    .await
    .expect("migrate direct");

    sqlx::query("ALTER ROLE postgres SET search_path = reliar, public")
        .execute(&direct_pool)
        .await
        .expect("alter role");

    let config_dir = write_pgdog_config(&pg_name);
    let pgdog = GenericImage::new(PGDOG_IMAGE, PGDOG_TAG)
        .with_exposed_port(6432.tcp())
        .with_wait_for(WaitFor::message_on_stderr("PgDog listening on"))
        .with_network(&network)
        .with_mount(Mount::bind_mount(
            config_dir.join("pgdog.toml").to_string_lossy().into_owned(),
            "/pgdog/pgdog.toml",
        ))
        .with_mount(Mount::bind_mount(
            config_dir.join("users.toml").to_string_lossy().into_owned(),
            "/pgdog/users.toml",
        ))
        .with_container_name(format!(
            "reliar-inbox-pgdog-{}",
            uuid::Uuid::now_v7().simple()
        ))
        .with_label("reliar.test", "true")
        .start()
        .await
        .expect("start pgdog");
    let pgdog_port = pgdog.get_host_port_ipv4(6432).await.expect("pgdog port");

    // No URL `options` — the role default (set above, direct) is what resolves `search_path`
    // through the pooler.
    let pool = PgPool::connect(&format!(
        "postgres://postgres:postgres@127.0.0.1:{pgdog_port}/postgres"
    ))
    .await
    .expect("connect through pgdog");

    create_business_table(&pool).await;
    let store = PostgresInboxStore::with_settings(pool.clone(), PostgresInboxSettings::default())
        .expect("construction succeeds through pgdog once the role default is in place");
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();

    // `claim`, the business write and `complete` all inside one transaction through the pooler —
    // if the pooler ever handed these to different backend connections, the transaction advisory
    // lock `claim` takes (and PostgreSQL itself would refuse to release mid-transaction on a
    // connection it was never taken on) would surface as a protocol or lock error here, not a
    // silent pass.
    let mut tx = pool.begin().await.unwrap();
    let claim = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

    let output = InsertBusinessRow { value: 1 }
        .handle(&mut tx)
        .await
        .unwrap();
    assert_eq!(output, 1);

    store.complete(&mut tx, &scope, id).await.unwrap();
    tx.commit().await.unwrap();

    assert_eq!(business_row_count(&pool).await, 1);
    let record = store.find(&scope, id).await.unwrap().unwrap();
    assert!(record.completed_at.is_some());

    // The `InProgress` path through the pooler: a second transaction racing the first while it
    // still holds the lock. Proven here with a fresh id claimed and held open, rather than a real
    // concurrent race, since the pooler scenario's own value is the *routing*, not the race —
    // that race is already covered by `inbox_concurrency.rs` against a direct connection.
    let id2 = MessageId::new();
    let mut holder = pool.begin().await.unwrap();
    let held = store
        .claim(&mut holder, &scope, crate::common::inbox::message(id2))
        .await
        .unwrap();
    assert_eq!(held, InboxClaim::Claimed { attempt: 1 });

    let mut second_tx = pool.begin().await.unwrap();
    let second_claim = store
        .claim(&mut second_tx, &scope, crate::common::inbox::message(id2))
        .await
        .unwrap();
    assert_eq!(second_claim, InboxClaim::InProgress);
    second_tx.rollback().await.unwrap();
    holder.rollback().await.unwrap();

    drop(pgdog);
    let _ = std::fs::remove_dir_all(&config_dir);
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![libtest_mimic::Trial::test(
        "inbox_pgdog::claim_and_complete_through_pgdog_in_one_transaction",
        move || {
            rt.block_on(claim_and_complete_through_pgdog_in_one_transaction());
            Ok(())
        },
    )]
}