reliar-store-postgres 0.7.0

PostgreSQL provider for the Reliar transactional outbox and inbox: 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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218

//! 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_until`, `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_pgdog;
mod inbox_plans;
mod inbox_process;
mod inbox_purge;
mod inbox_schema_verification;
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_complete;
mod outbox_constraint_names;
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_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_schema_verification;
mod outbox_statement_timeout;
mod outbox_stats;
mod outbox_upgrade_from_0004;
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). 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(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_pgdog::trials(rt));
    trials.extend(inbox_plans::trials(rt));
    trials.extend(inbox_process::trials(rt));
    trials.extend(inbox_purge::trials(rt));
    trials.extend(inbox_schema_verification::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_complete::trials(rt));
    trials.extend(outbox_constraint_names::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_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_schema_verification::trials(rt));
    trials.extend(outbox_statement_timeout::trials(rt));
    trials.extend(outbox_stats::trials(rt));
    trials.extend(outbox_upgrade_from_0004::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_dispatcher_observability::recorder_trials(rt));
    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
    }
}