pg open path: validate max_size > 0 at entry (typed Database before any round trip — 'max_size: 0' previously hung open forever at the bootstrap checkout, deadpool 0.13/0.14 neither validates nor defaults timeouts) and append the engine's '-c synchronous_commit' SET to the DSN's parse-carried options (was: setter replaced them, a silent override); pins: the zero-guard test (bounded by inner timeout, server-less-capable — fires pre-connect) and the DSN-options coexistence test (consumer SHOW statement_timeout + engine SHOW synchronous_commit, both settings) (task pg-fix-open-path, review 002 Finding 3 + options note)
This commit is contained in:
1 parent
9a5d1f4705
commit
86719a39cc
4 files changed
+184
-12
No files matched your search
@@ -50,7 +50,10 @@ pub struct PgOpts {
|
||||
/// queries, claims, and all non-transactional work ride these.
|
||||
/// Default [`DEFAULT_MAX_SIZE`] = 8 (the deployment matrix's
|
||||
/// budget line; the listener adds +1 per LISTEN-ing process
|
||||
/// *outside* the pool — sizing rule `max_size + 1`).
|
||||
/// *outside* the pool — sizing rule `max_size + 1`). A value of
|
||||
/// `0` fails [`open`](crate::open) with [`Error::Database`]
|
||||
/// immediately (validated at entry — no round trip; deadpool
|
||||
/// cannot grant checkouts from an empty pool).
|
||||
pub max_size: usize,
|
||||
/// The per-session `synchronous_commit` durability knob, wired as
|
||||
/// a connect-options `-c synchronous_commit=…` on every pooled
|
||||
|
||||
@@ -143,17 +143,38 @@ pub async fn open(config: &str, opts: PgOpts) -> alkstore::Result<Box<dyn Store>
|
||||
}
|
||||
|
||||
pub(crate) async fn open_store(config: &str, opts: PgOpts) -> alkstore::Result<PgStore> {
|
||||
// Validate at entry (the engine-wide posture): a `max_size: 0`
|
||||
// pool can never grant a checkout, and deadpool (0.13/0.14) does
|
||||
// not validate the size or apply default timeouts — without this
|
||||
// guard `open` would hang forever at the bootstrap checkout
|
||||
// instead of failing typed. This fires before any round trip.
|
||||
if opts.max_size == 0 {
|
||||
return Err(database_error(format!(
|
||||
"max_size must be positive, got {}",
|
||||
opts.max_size
|
||||
)));
|
||||
}
|
||||
|
||||
let cfg: tokio_postgres::Config = config.parse().map_err(pg_error)?;
|
||||
|
||||
// Per-session durability knob: the POC's connect-options SET
|
||||
// mechanics (`-c synchronous_commit=…` on every pooled
|
||||
// connection; POC-verified — pool.rs's `set_sync_commit_session`
|
||||
// probe shape).
|
||||
// probe shape). The engine's option is *appended* to whatever
|
||||
// options the consumer's DSN carried (tokio-postgres's setter
|
||||
// replaces — reading the parse-carried value via `get_options()`
|
||||
// first keeps the consumer's `options=` alive alongside; postgres
|
||||
// parses repeated `-c` flags, so both ride the connection).
|
||||
let mut pool_cfg = cfg.clone();
|
||||
pool_cfg.options(format!(
|
||||
let engine_opt = format!(
|
||||
"-c synchronous_commit={}",
|
||||
if opts.synchronous_commit { "on" } else { "off" }
|
||||
));
|
||||
);
|
||||
let merged = match pool_cfg.get_options() {
|
||||
Some(existing) => format!("{existing} {engine_opt}"),
|
||||
None => engine_opt,
|
||||
};
|
||||
pool_cfg.options(merged);
|
||||
|
||||
let mgr_cfg = ManagerConfig {
|
||||
recycling_method: RecyclingMethod::Fast,
|
||||
|
||||
@@ -701,6 +701,95 @@ async fn store_trait_methods_are_wired() {
|
||||
drop_schema(&admin, &schema).await;
|
||||
}
|
||||
|
||||
/// Acceptance (review 002 Finding 3): `max_size: 0` fails `open` with
|
||||
/// the typed `Database` error *before any round trip* — promptly, not
|
||||
/// the deadpool hang (a `max_size: 0` pool can never grant a checkout
|
||||
/// and deadpool 0.13/0.14 neither validates it nor defaults timeouts).
|
||||
/// The test's inner timeout bounds the hang: a regression re-hangs and
|
||||
/// fails here by timeout. This needs no server — the guard fires at
|
||||
/// entry, before any connection attempt (its "before any round trip"
|
||||
/// property pinned by running against an unreachable endpoint).
|
||||
#[tokio::test]
|
||||
async fn open_fails_database_promptly_on_zero_max_size() {
|
||||
let dsn = harness_dsn().unwrap_or_else(unreachable_dsn);
|
||||
let schema = instance_namer("zero")();
|
||||
let opts = PgOpts {
|
||||
schema,
|
||||
max_size: 0,
|
||||
..PgOpts::default()
|
||||
};
|
||||
let result = tokio::time::timeout(Duration::from_secs(2), open_store(&dsn, opts)).await;
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"open with max_size: 0 must fail promptly — \
|
||||
an inner timeout tripped, the deadpool hang regressed"
|
||||
);
|
||||
let err = result.unwrap().unwrap_err();
|
||||
match err {
|
||||
alkstore::Error::Database(source) => {
|
||||
let detail = source.to_string();
|
||||
assert!(
|
||||
detail.contains("max_size") && detail.contains("positive"),
|
||||
"the source chain says what was rejected, got: {detail}"
|
||||
);
|
||||
}
|
||||
other => panic!("max_size: 0 must fail typed Database, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Acceptance (review 002 options note): the consumer DSN's `options=`
|
||||
/// survive `open` — the engine's `synchronous_commit` `-c` SET is
|
||||
/// *appended*, not substituted (tokio-postgres's setter replaces; the
|
||||
/// parse-carried value is read back and preserved). Both knob values
|
||||
/// ride every pooled connection: the consumer's
|
||||
/// (`statement_timeout` via `SHOW`) and the engine's
|
||||
/// (`synchronous_commit` via `SHOW` on both settings).
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn consumer_dsn_options_survive_alongside_the_engine_set() {
|
||||
let Some(dsn) = harness_dsn() else {
|
||||
eprintln!("skip: no harness server");
|
||||
return;
|
||||
};
|
||||
let consumer_dsn = format!("{dsn} options='-c statement_timeout=30000'");
|
||||
|
||||
for (sync_commit, expected) in [(true, "on"), (false, "off")] {
|
||||
let schema = instance_namer("dsnops")();
|
||||
let store = open_store(
|
||||
&consumer_dsn,
|
||||
PgOpts {
|
||||
schema: schema.to_string(),
|
||||
synchronous_commit: sync_commit,
|
||||
..PgOpts::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap_or_else(|e| panic!("open with a DSN carrying options must succeed: {e}"));
|
||||
let conn = store.pool().get().await.unwrap();
|
||||
|
||||
let row = conn.query_one("SHOW statement_timeout", &[]).await.unwrap();
|
||||
let timeout: String = row.get(0);
|
||||
assert_eq!(
|
||||
timeout, "30s",
|
||||
"the consumer's DSN options must survive open, got: {timeout}"
|
||||
);
|
||||
|
||||
let row = conn
|
||||
.query_one("SHOW synchronous_commit", &[])
|
||||
.await
|
||||
.unwrap();
|
||||
let value: String = row.get(0);
|
||||
assert_eq!(
|
||||
value, expected,
|
||||
"the engine's SET must still ride the appended options, got: {value}"
|
||||
);
|
||||
|
||||
drop(conn);
|
||||
store.close();
|
||||
let admin = harness_client().await.unwrap();
|
||||
drop_schema(&admin, &schema).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// The error-mapping helpers exist and are the documented reuse point:
|
||||
/// a mapped error is `Error::Database` with the source chain preserved
|
||||
/// (the mapping trio the mechanism tasks reuse; the io-Error string
|
||||
|
||||
Reference in new issue
Block a user