reliar-store-postgres 0.9.0

//! The single Postgres-touching test binary for `reliar-store-postgres` (RELIAR-27): 167 leaked
//! containers and 31 GB of volumes from the provider suite, traced to two causes fixed together
//! below — a `static` container handle whose destructor never runs at process exit, and one
//! `[[test]]` target per scenario file starting its own container.
//!
//! **Why this file exists.** Before this, every scenario (`tests/outbox_*.rs`, `tests/migrate.rs`)
//! was its own `[[test]]` binary, each lazily starting its own shared Postgres container in a
//! `static OnceLock<ContainerAsync<..>>`. Two facts made that leak every container, forever:
//! `testcontainers` 0.27 has **no reaper** (no Ryuk) — the *only* removal path is
//! `ContainerAsync::Drop` — and Rust **never runs destructors for `static`s** at process exit, so
//! that `Drop` never ran. ~25 scenario binaries meant ~25 leaked containers (plus volumes) per
//! `cargo test -p reliar-store-postgres` run.
//!
//! **The fix.** `harness = false` + `libtest-mimic` (`Cargo.toml`'s single `[[test]] name =
//! "postgres"` entry) so this `main` owns the shared container as a **local, not a `static`**:
//! it starts the container, runs every scenario (`mod` per former file, `Trial` per former
//! `#[tokio::test]` fn) on one shared Tokio runtime, then **drops the container before
//! exiting**. `main` returning `ExitCode` (rather than calling `Conclusion::exit()`, which calls
//! `process::exit` and would skip every destructor including this one) is what makes that
//! ordering happen — falling out of `main`'s scope runs every local's `Drop` first, and only
//! then does the runtime convert the returned `ExitCode` into the real process exit code.
//!
//! The `watchdog` dev-dependency feature (enabled in `Cargo.toml`) is the belt to this belt: it
//! removes registered containers on SIGINT/SIGTERM/SIGQUIT, which no amount of `Drop` correctness
//! can cover (a killed process runs no destructors either).
//!
//! **Two run phases (RELIAR-66).** `main` calls `libtest_mimic::run` twice: once for every
//! ordinary trial, then again, only after the first call has returned, for the handful of trials
//! that install a thread-local `tracing` recording subscriber (`recorder_trials` below). See
//! `common::install_recording_subscriber`'s doc for why a second, `--test-threads 1` phase — not
//! a mutex — is what makes those trials' transcripts deterministic.
//!
//! **This binary is the Postgres substrate host for the whole workspace, not only its own SQL**
//! (ADR 0043 §3): since no in-memory store fake exists anywhere in this workspace, `reliar-outbox`'s
//! dispatcher (`outbox_dispatcher_*`) and `reliar-inbox`'s `process` flow (`inbox_process`,
//! folded into `inbox_fail`/`inbox_dead`/`inbox_dead_letters`/`inbox_purge`/`inbox_spans`) are
//! proven here too, against `PostgresOutboxStore`/`PostgresInboxStore`, not in either crate's own
//! `tests/`. Every assertion about a stored row, lease, attempt, state or purge count reads real
//! Postgres; a fault is a `FaultyStore` decorator that delegates to the real store, or a publisher
//! stub whose error type is uninhabited (it can only succeed, stall or panic). **This binary's
//! dispatcher trials prove control flow, not delivery** — the oracle here is `outbox`'s own
//! columns (`published_at`, `attempts`, `available_at`, `locked_by`, `dead_at`, `last_error`);
//! that a byte actually reached a broker is asserted only in `tests/system`'s `e2e` suite, against
//! a real NATS.

// Crate-wide (this binary has no other crate root): a `harness = false` target is not compiled
// with `rustc --test`, so it loses whatever implicit "this is test code" exemption normally
// covers restriction lints like these in a standard-harness test binary — every module below
// still needs them for exactly the reasons `unwrap`/`expect`/`panic!` are always fine in a test
// assertion (that's how a test reports failure) and a seeded test id occasionally exceeds an
// `i32`/`i64` cast target harmlessly.
#![allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::cast_possible_truncation
)]

mod common;
mod harness_smoke;
mod inbox_claim;
mod inbox_concurrency;
mod inbox_dead;
mod inbox_dead_letters;
mod inbox_fail;
mod inbox_migration;
mod inbox_not_migrated;
mod inbox_pgdog;
mod inbox_plans;
mod inbox_process;
mod inbox_purge;
mod inbox_send_bounds;
mod inbox_spans;
mod inbox_statement_timeout;
mod migrate;
mod outbox_acquire_skip_locked;
mod outbox_acquire_skip_locked_held_lock;
mod outbox_claim_index_scale;
mod outbox_claim_no_lock_during_publish;
mod outbox_claim_token_fencing;
mod outbox_complete;
mod outbox_dead_letters;
mod outbox_dispatcher_concurrency;
mod outbox_dispatcher_leases;
mod outbox_dispatcher_lifecycle;
mod outbox_dispatcher_observability;
mod outbox_dispatcher_store_faults;
mod outbox_enqueue;
mod outbox_enqueue_bare_and_envelope_agree;
mod outbox_enqueue_serialize_error;
mod outbox_enqueue_spans;
mod outbox_epoch_millis_codec;
mod outbox_error_classification;
mod outbox_fail_retry_dead;
mod outbox_lease_management;
mod outbox_lease_recovery;
mod outbox_non_default_schema;
mod outbox_not_migrated;
mod outbox_pgdog;
mod outbox_plans;
mod outbox_poison_sweep_failure;
mod outbox_poisoned_row;
mod outbox_purge;
mod outbox_purge_concurrent_resurrection;
mod outbox_roundtrip;
mod outbox_statement_timeout;
mod outbox_stats;
mod recorder_spawn_capture;

use std::process::ExitCode;

use libtest_mimic::Arguments;

fn main() -> ExitCode {
    let args = Arguments::from_args();

    // Leaked deliberately: this runtime is a process-lifetime singleton (every trial below
    // shares it), not a value with a meaningful drop — `Box::leak` is the standard way to get a
    // `&'static` out of a value built at runtime, cheaper and clearer here than an `Arc` every
    // scenario module would otherwise need to clone.
    let rt: &'static tokio::runtime::Runtime = Box::leak(Box::new(
        tokio::runtime::Runtime::new().expect("build the shared Tokio runtime"),
    ));

    // The one and only container this whole run starts — a **local**, not a `static` (that
    // `static` was RELIAR-27's bug). Declared here, outside the `catch_unwind` below, and filled
    // in as soon as it starts, so a panic anywhere after that (RELIAR-94: a bug in a `trials()`
    // builder, or in `libtest_mimic::run` itself — every individual trial's own panic is already
    // caught by `libtest_mimic`, so the previous unguarded version never showed the gap in
    // ordinary runs) still leaves the container reachable for the unconditional teardown at the
    // bottom of this function, mirroring `tests/system/tests/e2e/main.rs`'s `main`. Without this,
    // a panic that unwound past `main`'s own frame left no Tokio runtime for
    // `ContainerAsync::Drop`'s internal `Handle::current()` call to find, so `Drop` itself
    // panicked — "panic in a destructor during cleanup" — the process aborted, and the container
    // (with its volume) was left running.
    let mut container = None;
    let mut conclusions = None;

    let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        container = Some(rt.block_on(common::start_shared_container()));

        let mut trials = Vec::new();

        trials.extend(harness_smoke::trials(rt));
        trials.extend(inbox_claim::trials(rt));
        trials.extend(inbox_concurrency::trials(rt));
        trials.extend(inbox_dead::trials(rt));
        trials.extend(inbox_dead_letters::trials(rt));
        trials.extend(inbox_fail::trials(rt));
        trials.extend(inbox_migration::trials(rt));
        trials.extend(inbox_not_migrated::trials(rt));
        trials.extend(inbox_pgdog::trials(rt));
        trials.extend(inbox_plans::trials(rt));
        trials.extend(inbox_process::trials(rt));
        trials.extend(inbox_purge::trials(rt));
        trials.extend(inbox_send_bounds::trials(rt));
        trials.extend(inbox_statement_timeout::trials(rt));
        trials.extend(migrate::trials(rt));
        trials.extend(outbox_acquire_skip_locked::trials(rt));
        trials.extend(outbox_acquire_skip_locked_held_lock::trials(rt));
        trials.extend(outbox_claim_index_scale::trials(rt));
        trials.extend(outbox_claim_no_lock_during_publish::trials(rt));
        trials.extend(outbox_claim_token_fencing::trials(rt));
        trials.extend(outbox_complete::trials(rt));
        trials.extend(outbox_dead_letters::trials(rt));
        trials.extend(outbox_dispatcher_concurrency::trials(rt));
        trials.extend(outbox_dispatcher_leases::trials(rt));
        trials.extend(outbox_dispatcher_lifecycle::trials(rt));
        trials.extend(outbox_dispatcher_observability::trials(rt));
        trials.extend(outbox_dispatcher_store_faults::trials(rt));
        trials.extend(outbox_enqueue::trials(rt));
        trials.extend(outbox_enqueue_bare_and_envelope_agree::trials(rt));
        trials.extend(outbox_enqueue_serialize_error::trials(rt));
        trials.extend(outbox_epoch_millis_codec::trials(rt));
        trials.extend(outbox_error_classification::trials(rt));
        trials.extend(outbox_fail_retry_dead::trials(rt));
        trials.extend(outbox_lease_management::trials(rt));
        trials.extend(outbox_lease_recovery::trials(rt));
        trials.extend(outbox_non_default_schema::trials(rt));
        trials.extend(outbox_not_migrated::trials(rt));
        trials.extend(outbox_pgdog::trials(rt));
        trials.extend(outbox_plans::trials(rt));
        trials.extend(outbox_poisoned_row::trials(rt));
        trials.extend(outbox_purge::trials(rt));
        trials.extend(outbox_purge_concurrent_resurrection::trials(rt));
        trials.extend(outbox_roundtrip::trials(rt));
        trials.extend(outbox_statement_timeout::trials(rt));
        trials.extend(outbox_stats::trials(rt));

        // Every trial that installs a thread-local `tracing` recording subscriber (RELIAR-66) —
        // never run alongside `trials` above; see `common::install_recording_subscriber`'s doc
        // for the hazard this separation closes.
        let mut recorder_trials = Vec::new();
        recorder_trials.extend(inbox_spans::recorder_trials(rt));
        recorder_trials.extend(outbox_claim_token_fencing::recorder_trials(rt));
        recorder_trials.extend(outbox_dispatcher_observability::recorder_trials(rt));
        recorder_trials.extend(outbox_enqueue_spans::recorder_trials(rt));
        recorder_trials.extend(outbox_poison_sweep_failure::recorder_trials(rt));
        recorder_trials.extend(recorder_spawn_capture::recorder_trials(rt));

        let conclusion = libtest_mimic::run(&args, trials);

        // The recorder trials run only now, in a second phase, and only with `test_threads`
        // forced to 1. `libtest_mimic::run` above drives its trials inside `std::thread::scope`,
        // which does not return until every worker thread it spawned has been joined — so by the
        // time control reaches this line, no other *trial's own* thread is still alive to race a
        // `tracing` callsite against a recorder's thread-local default (the shared `rt` and its
        // background tasks are still running, but their callsites were already registered in
        // phase one). `test_threads = 1` additionally keeps the recorder trials from racing
        // *each other* (RELIAR-66).
        let mut recorder_args = args.clone();
        recorder_args.test_threads = Some(1);
        let recorder_conclusion = libtest_mimic::run(&recorder_args, recorder_trials);

        conclusions = Some((conclusion, recorder_conclusion));
    }));

    // Drop the container (and, with it, its volume) *before* returning — never
    // `Conclusion::exit()`, which calls `process::exit` and would skip this (RELIAR-27) — and
    // regardless of whether the closure above panicked (RELIAR-94). `ContainerAsync::Drop` needs
    // a Tokio runtime context, so the drop itself runs inside one more `block_on`, not out here.
    rt.block_on(async move { drop(container) });

    match panic {
        Ok(()) => {
            let (conclusion, recorder_conclusion) =
                conclusions.expect("set on every non-panicking path above");

            if conclusion.has_failed() || recorder_conclusion.has_failed() {
                ExitCode::FAILURE
            } else {
                ExitCode::SUCCESS
            }
        }
        // The container is already torn down above; re-raise so the harness's own panic message
        // and non-zero process exit still surface exactly as they would have without this
        // wrapper.
        Err(payload) => std::panic::resume_unwind(payload),
    }
}