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
//! ADR 0047 (+ Amendment A) — a store constructor performs **no I/O**: it issues no query, opens
//! no connection and verifies nothing about the database. A `search_path` that does not resolve
//! `outbox`, or a schema stopped mid-migration, is reported by the **first store call** as
//! `PostgresOutboxError::NotMigrated`, never at construction.
use crate::common;
use reliar_core::Classify;
use reliar_outbox::OutboxStore;
use reliar_store_postgres::{PostgresOutboxError, PostgresOutboxStore};
use sqlx::PgPool;
use sqlx::postgres::PgConnectOptions;
async fn pool_without_search_path() -> PgPool {
let base = common::fresh_unmigrated_db().await;
reliar_store_postgres::migrate(&base, reliar_store_postgres::MigrateOptions::default())
.await
.unwrap();
// A pool whose `search_path` explicitly excludes `reliar` — deliberately set to just
// `public` rather than left at the server default, since a local role happening to share
// the schema's name would otherwise resolve it via Postgres's own `"$user", public` default.
let options: PgConnectOptions = base
.connect_options()
.as_ref()
.clone()
.options([("search_path", "public")]);
PgPool::connect_with(options).await.unwrap()
}
/// Against a **fully migrated** database, on a pool whose URL sets no `search_path`:
/// `PostgresOutboxStore::new` returns a store — no `Result`, no `.await`, and (§43.A.36) no
/// statement — and the first `acquire` reports `NotMigrated`: `Permanent`, `source()` the
/// underlying `42P01`, `Display` naming both `migrate()` and `search_path`, and no
/// DSN/host/credential.
async fn the_first_call_reports_not_migrated_without_search_path() {
let pool = pool_without_search_path().await;
// §43.A.36: `PostgresOutboxStore::new`'s signature (`fn new(pool) -> Self`, synchronous,
// infallible) is the actual proof construction issues no statement — there is no `.await`
// for a query to run under. `PgPool`'s own connection counters below only corroborate that no
// *new* connection was opened; they cannot distinguish "no statement" from "a statement on an
// already-idle connection", which is why the signature, not the counters, is the witness.
let size_before = pool.size();
let idle_before = pool.num_idle();
let store = PostgresOutboxStore::new(pool.clone());
assert_eq!(
pool.size(),
size_before,
"constructing a store must acquire no connection"
);
assert_eq!(
pool.num_idle(),
idle_before,
"constructing a store must acquire no connection"
);
let err = store
.acquire(reliar_outbox::AcquireRequest::new(
reliar_outbox::WorkerId::generate(),
))
.await
.unwrap_err();
let PostgresOutboxError::NotMigrated { source } = &err else {
panic!("expected NotMigrated, got {err:?}");
};
assert!(std::error::Error::source(&err).is_some());
assert!(
matches!(source, sqlx::Error::Database(db) if db.code().as_deref() == Some("42P01")),
"expected SQLSTATE 42P01, got {source:?}"
);
assert_eq!(err.kind(), reliar_core::FailureKind::Permanent);
let message = err.to_string();
assert!(message.contains("migrate"), "message: {message:?}");
assert!(message.contains("search_path"), "message: {message:?}");
assert!(!message.contains("postgres://"), "message: {message:?}");
}
/// Reuses the two-stage migration harness (P-1): migrate through `0004` only, build the store
/// (succeeds — construction performs no I/O), then `acquire` fails with PostgreSQL's own
/// missing-column text. Replaces the withdrawn `SchemaOutOfDate` construction-time check (ADR
/// 0044 A.4/A.5 — never reached a released version) with the behaviour that actually ships.
async fn a_schema_stopped_before_the_head_fails_at_the_first_acquire() {
let pool = common::fresh_unmigrated_db().await;
common::apply_migration_prefix(&pool, 4).await;
// `apply_migration_prefix` sets `search_path` only on its own dedicated connection; the
// store's own pool needs it set explicitly too, exactly as `migrate()` would leave a real
// deployment's connection URL.
let options: PgConnectOptions = pool
.connect_options()
.as_ref()
.clone()
.options([("search_path", "reliar,public")]);
let store_pool = PgPool::connect_with(options).await.unwrap();
let store = PostgresOutboxStore::new(store_pool);
let err = store
.acquire(reliar_outbox::AcquireRequest::new(
reliar_outbox::WorkerId::generate(),
))
.await
.unwrap_err();
assert_eq!(err.kind(), reliar_core::FailureKind::Permanent);
let message = err.to_string();
assert!(
message.contains("message_id") || message.contains("claim_token"),
"expected PostgreSQL's own missing-column text, got: {message:?}"
);
}
pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
vec![
libtest_mimic::Trial::test(
"outbox_not_migrated::the_first_call_reports_not_migrated_without_search_path",
move || {
rt.block_on(the_first_call_reports_not_migrated_without_search_path());
Ok(())
},
),
libtest_mimic::Trial::test(
"outbox_not_migrated::a_schema_stopped_before_the_head_fails_at_the_first_acquire",
move || {
rt.block_on(a_schema_stopped_before_the_head_fails_at_the_first_acquire());
Ok(())
},
),
]
}