reliar-store-postgres 0.6.0

PostgreSQL provider for the Reliar transactional outbox: migrations, migrate(), enqueue and the SKIP LOCKED claim.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

//! 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.

// 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 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_complete;
mod outbox_constraint_names;
mod outbox_dead_letters;
mod outbox_dispatcher_end_to_end;
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_pgdog;
mod outbox_poison_sweep_failure;
mod outbox_poisoned_row;
mod outbox_purge;
mod outbox_purge_concurrent_resurrection;
mod outbox_roundtrip;
mod outbox_schema_verification;
mod outbox_statement_timeout;
mod outbox_stats;
mod recorder_spawn_capture;
mod server_version_floor;

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). Kept alive until every trial has finished, then dropped
    // explicitly, below, before this function returns.
    let container = rt.block_on(common::start_shared_container());

    let mut trials = Vec::new();

    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_complete::trials(rt));
    trials.extend(outbox_constraint_names::trials(rt));
    trials.extend(outbox_dead_letters::trials(rt));
    trials.extend(outbox_dispatcher_end_to_end::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_pgdog::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_schema_verification::trials(rt));
    trials.extend(outbox_statement_timeout::trials(rt));
    trials.extend(outbox_stats::trials(rt));
    trials.extend(server_version_floor::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(outbox_enqueue_spans::recorder_trials(rt));
    recorder_trials.extend(outbox_schema_verification::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);

    // Drop the container (and, with it, its volumes) *before* returning — never
    // `Conclusion::exit()`, which calls `process::exit` and would skip this (RELIAR-27).
    // `ContainerAsync`'s own `Drop` needs a Tokio runtime context (it calls
    // `tokio::runtime::Handle::current()` internally to perform its async cleanup), which plain
    // `main()` is not inside once every `rt.block_on(..)` call above has returned — so the drop
    // itself runs inside one more `block_on`, not out here.
    rt.block_on(async move { drop(container) });

    if conclusion.has_failed() || recorder_conclusion.has_failed() {
        ExitCode::FAILURE
    } else {
        ExitCode::SUCCESS
    }
}