From d34dfa63f32c0b03ff1c2f3917bea8a0f2cd33d5 Mon Sep 17 00:00:00 2001 From: Mikael Rinne <40919111+rorychatt@users.noreply.github.com> Date: Thu, 29 Oct 2026 14:46:32 +0100 Subject: [PATCH] Fix the CI test flakes of K22 and the three of 2026-10-29 Product (obop-storage, schema.rs and pg_hist.rs): a process that waited for the schema advisory lock looked at a catalogue cache older than the commit of the process before it, found the history part missing and ran its DDL again, which could deadlock (40P01) with obop_history_ensure_partition. schema::take_lock now reads pg_class once the lock is held. Tests: random 16 byte canary in backup_cli; bounded wait for the pool gauges and a 10 s bound for /readyz in pool.rs; bounded warm-up in the read-lane test; connections held until released and no wall-clock bounds in the lock pool test; kill_sessions waits for the server and a longer acquire window in the connection kill tests; each test port handed out once per binary and a server that exited is noticed in multi_process.rs and pool.rs. K22 in docs/ISSUES.md says which are fixed and what is left. Co-Authored-By: Claude Sonnet 5.5 --- crates/kernel/obop-storage/src/pg_hist.rs | 2 +- .../kernel/obop-storage/src/pg_hist_tests.rs | 66 ++++++++++ .../kernel/obop-storage/src/pg_pool_tests.rs | 117 ++++++++++++------ crates/kernel/obop-storage/src/schema.rs | 23 +++- crates/kernel/obop-storage/src/seams_tests.rs | 20 ++- crates/obop/tests/backup_cli.rs | 37 ++++-- crates/obop/tests/multi_process.rs | 31 ++++- crates/obop/tests/pool.rs | 75 ++++++++--- docs/ISSUES.md | 2 +- 9 files changed, 299 insertions(+), 74 deletions(-) diff --git a/crates/kernel/obop-storage/src/pg_hist.rs b/crates/kernel/obop-storage/src/pg_hist.rs index 09aec367..64a3dc47 100644 --- a/crates/kernel/obop-storage/src/pg_hist.rs +++ b/crates/kernel/obop-storage/src/pg_hist.rs @@ -1329,7 +1329,7 @@ impl Import { if self.needs_install { // The backup carries history and the database has none yet: make the part, as the // start of a process would, under the schema lock and in this transaction. - c.batch_execute(schema::LOCK_SQL)?; + schema::take_lock(c)?; schema::ensure_parts(c, &[PART], Ddl::IfMissing)?; self.exists = true; self.needs_install = false; diff --git a/crates/kernel/obop-storage/src/pg_hist_tests.rs b/crates/kernel/obop-storage/src/pg_hist_tests.rs index cfd9f475..a27d1e82 100644 --- a/crates/kernel/obop-storage/src/pg_hist_tests.rs +++ b/crates/kernel/obop-storage/src/pg_hist_tests.rs @@ -398,6 +398,72 @@ fn processes_that_need_the_part_together_make_it_once() { assert_eq!(n, 4); } +/// A process that waits for the schema lock while another makes the part must see the part when +/// the lock is its: the question "is it there?" is answered from the catalogue as of then, not as +/// of the moment the transaction began or the names were last asked for. Before this was held, the +/// waiter read the cache of its session (a name it had asked for earlier and found missing stayed +/// missing), took the part for absent and ran all its SQL again over the objects of the other: +/// `CREATE INDEX IF NOT EXISTS` takes a lock on the table and on every partition before it finds +/// the index there, which collided with a third process making a partition (SQLSTATE 40P01). +#[test] +fn a_process_that_waited_for_the_schema_lock_sees_what_the_one_before_it_made() { + let Some(db) = TestDb::new() else { + return; + }; + let mut waiter = db.connect(); + // As every process does at its start: ask once, with nothing there. The session now holds a + // negative answer for each of these names. + crate::schema::with_lock(&mut waiter, |c| { + assert!(!(crate::pg_hist::PART.check)(c)?); + Ok(()) + }) + .unwrap(); + // The maker holds the lock, as a process does while it makes the part. + let mut maker = db.connect(); + maker.batch_execute("BEGIN").unwrap(); + maker.batch_execute(crate::schema::LOCK_SQL).unwrap(); + let mut watcher = db.connect(); + let seen = std::thread::scope(|s| { + let waiting = s.spawn(|| { + crate::schema::with_lock(&mut waiter, |c| { + Ok(( + (crate::pg_hist::PART.check)(c)?, + crate::schema::present(c, &crate::pg_hist::RELATIONS)?, + )) + }) + }); + // Wait until the waiter is queued for the lock, which is the situation to be tested; the + // maker lets go only then. + let deadline = std::time::Instant::now() + crate::testing::time_budget(30); + loop { + let queued: i64 = watcher + .query_one( + "SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND NOT granted AND database = (SELECT oid FROM pg_database WHERE datname = current_database())", + &[], + ) + .unwrap() + .get(0); + if queued == 1 { + break; + } + assert!( + std::time::Instant::now() < deadline, + "the waiter never queued for the schema lock" + ); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + crate::schema::install_all(&mut maker).unwrap(); + maker.batch_execute("COMMIT").unwrap(); + waiting.join().unwrap() + }) + .unwrap(); + assert_eq!( + seen, + (true, true), + "the waiter has the lock after the other committed the part: it sees the part (check, relations)" + ); +} + /// A model that keeps history, with `names` as plain fields. fn history_of(names: &[&str]) -> Track { Track { diff --git a/crates/kernel/obop-storage/src/pg_pool_tests.rs b/crates/kernel/obop-storage/src/pg_pool_tests.rs index 88861db1..6b7eb8ba 100644 --- a/crates/kernel/obop-storage/src/pg_pool_tests.rs +++ b/crates/kernel/obop-storage/src/pg_pool_tests.rs @@ -81,6 +81,44 @@ fn sleep_query(c: &mut postgres::Client, secs: u32) -> crate::pgpool::PR<()> { Ok(()) } +/// The acquire window of the tests that kill connections. The pool replaces the dead connection +/// with a new one, and opening one (TCP, authentication) can take longer than the default 2 s on a +/// machine that is busy with other work; how long it takes is not what these tests are about, and +/// a window this long cannot make a lost replacement pass. +const SLOW_CONNECT: Duration = Duration::from_secs(30); + +/// Ends every session of the application `app` on the server and returns when the server has none +/// left, so that the next use of a pooled connection finds it dead whatever the speed of the +/// machine. (`pg_terminate_backend` only asks: a backend that is not scheduled for a while on a +/// loaded machine is still there, and a fixed pause after the call is a guess about that while.) +fn kill_sessions(url: &str, app: &str) { + let mut admin = PgConnector::new(url).unwrap().connect().unwrap(); + admin + .execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE application_name = $1 AND pid <> pg_backend_pid()", + &[&app], + ) + .unwrap(); + let deadline = Instant::now() + crate::testing::time_budget(30); + loop { + let left: i64 = admin + .query_one( + "SELECT count(*) FROM pg_stat_activity WHERE application_name = $1 AND pid <> pg_backend_pid()", + &[&app], + ) + .unwrap() + .get(0); + if left == 0 { + return; + } + assert!( + Instant::now() < deadline, + "{left} sessions of {app} are still on the server after they were ended" + ); + std::thread::sleep(Duration::from_millis(5)); + } +} + // ---- configuration ------------------------------------------------------------------------- fn lookup(pairs: &'static [(&'static str, &'static str)]) -> impl Fn(&str) -> Option { @@ -589,11 +627,11 @@ fn the_lock_pool_does_not_take_interactive_capacity_and_the_other_way_round() { } assert_eq!(pg.lock_pool.report().in_use, 8, "eight lock connections"); // Interactive traffic is not affected: both main connections are free. - let t = Instant::now(); + // (A get that had to wait for a connection would be refused after the acquire timeout, so + // the gets succeeding is the proof; how fast they are depends on the machine.) for _ in 0..50 { assert_eq!(pg.get(&model, "a").unwrap().unwrap()["n"], 1); } - assert!(t.elapsed() < Duration::from_secs(2), "{:?}", t.elapsed()); assert_eq!(pg.main.stats().timeouts, 0); // ... and a ninth lock holder is refused (busy), not queued without bound. let t = Instant::now(); @@ -602,35 +640,53 @@ fn the_lock_pool_does_not_take_interactive_capacity_and_the_other_way_round() { panic!("a ninth lock connection does not exist"); }; assert_eq!(Fault::of_message(&err), Some(Fault::Busy), "{err}"); - assert!(t.elapsed() < Duration::from_secs(4), "{:?}", t.elapsed()); + // The refusal comes after the acquire timeout (1.5 s); the bound only catches a queue that + // never ends. + assert!( + t.elapsed() < crate::testing::time_budget(30), + "{:?}", + t.elapsed() + ); release.wait(); for h in holders { h.join().unwrap(); } - // The other way round: with every interactive connection busy a lock is still granted. - let sleepers: Vec<_> = (0..2) + // The other way round: with every interactive connection busy a lock is still granted. The + // two connections are held until the test lets go (also when it unwinds: the gates close), + // not for a guessed time: on a loaded machine the steps below can take longer than any guess, + // and the interactive pool would no longer be full when it is asked. + let (busy_tx, busy_rx) = std::sync::mpsc::channel(); + let mut gates = Vec::new(); + let holders: Vec<_> = (0..2) .map(|_| { - let pg = pg.clone(); - std::thread::spawn(move || pg.main.run(Class::Point, |c| sleep_query(c, 2))) + let (pg, busy_tx) = (pg.clone(), busy_tx.clone()); + let (gate, opens) = std::sync::mpsc::channel::<()>(); + gates.push(gate); + std::thread::spawn(move || { + pg.main.run(Class::Point, |_c| { + busy_tx.send(()).unwrap(); + let _ = opens.recv_timeout(Duration::from_secs(60)); + Ok(()) + }) + }) }) .collect(); - while pg.main.stats().in_use < 2 { - std::thread::sleep(Duration::from_millis(5)); + for _ in 0..2 { + busy_rx + .recv_timeout(crate::testing::time_budget(30)) + .expect("the interactive connections were never both taken"); } - let t = Instant::now(); + // (A lock that had to wait for the interactive pool would be refused after the acquire + // timeout, so being granted is the proof.) let g = pg.lock(&unique("free")).unwrap(); - assert!( - t.elapsed() < Duration::from_millis(1000), - "{:?}", - t.elapsed() - ); assert!( pg.get(&model, "a").is_err(), "the interactive pool is the one that is full" ); drop(g); - for s in sleepers { - s.join().unwrap().unwrap(); + drop(gates); + for h in holders { + h.join().unwrap().unwrap(); } } @@ -1322,22 +1378,13 @@ fn a_slow_consumer_is_not_cut_off_by_the_statement_limit_but_a_stalled_one_is_en fn a_connection_killed_by_the_server_is_noticed_and_replaced() { let (url, _db) = pg_db!(); let sep = if url.contains('?') { '&' } else { '?' }; - let kill = |app: &str| { - let mut admin = PgConnector::new(&url).unwrap().connect().unwrap(); - admin - .execute( - "SELECT pg_terminate_backend(pid) FROM pg_stat_activity \ - WHERE application_name = $1 AND pid <> pg_backend_pid()", - &[&app], - ) - .unwrap(); - std::thread::sleep(Duration::from_millis(200)); - }; + let kill = |app: &str| kill_sessions(&url, app); let model = unique("killed"); // 1. With a health check on every checkout the dead connection is never handed out. let app = unique("kill-health"); let cfg = PgPoolConfig { health_check_idle: Duration::ZERO, + acquire_timeout: SLOW_CONNECT, ..PgPoolConfig::default() }; let pg = connect(&format!("{url}{sep}application_name={app}"), cfg); @@ -1350,6 +1397,7 @@ fn a_connection_killed_by_the_server_is_noticed_and_replaced() { let app = unique("kill-read"); let cfg = PgPoolConfig { health_check_idle: Duration::from_secs(3600), + acquire_timeout: SLOW_CONNECT, ..PgPoolConfig::default() }; let pg = connect(&format!("{url}{sep}application_name={app}"), cfg); @@ -1369,6 +1417,7 @@ fn a_connection_killed_by_the_server_is_noticed_and_replaced() { &format!("{url}{sep}application_name={app}"), PgPoolConfig { health_check_idle: Duration::from_secs(3600), + acquire_timeout: SLOW_CONNECT, ..PgPoolConfig::default() }, ); @@ -1389,19 +1438,11 @@ fn only_a_statement_on_a_connection_that_broke_is_in_flight() { // caller was sent (the limit of the session, BEGIN), where nothing is in doubt. let (url, _db) = pg_db!(); let sep = if url.contains('?') { '&' } else { '?' }; - let kill = |app: &str| { - let mut admin = PgConnector::new(&url).unwrap().connect().unwrap(); - admin - .execute( - "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE application_name = $1 AND pid <> pg_backend_pid()", - &[&app], - ) - .unwrap(); - std::thread::sleep(Duration::from_millis(200)); - }; + let kill = |app: &str| kill_sessions(&url, app); let model = unique("inflight"); let cfg = || PgPoolConfig { health_check_idle: Duration::from_secs(3600), + acquire_timeout: SLOW_CONNECT, ..PgPoolConfig::default() }; // A write on a pooled connection that was killed: the statement itself meets the dead diff --git a/crates/kernel/obop-storage/src/schema.rs b/crates/kernel/obop-storage/src/schema.rs index 32332706..d348f660 100644 --- a/crates/kernel/obop-storage/src/schema.rs +++ b/crates/kernel/obop-storage/src/schema.rs @@ -47,6 +47,24 @@ use sha2::{Digest, Sha256}; pub(crate) const LOCK_SQL: &str = "SELECT pg_advisory_xact_lock(hashtextextended('obop-schema', 0))"; +/// Makes the session read the catalogue as it is now. A session keeps what it has looked up (does +/// this name exist?), and what it is told about changes other sessions committed in the meantime +/// is taken in when it opens a relation or begins a transaction, not when it gets a lock that is +/// no relation (this one). A process that waited for [`LOCK_SQL`] while another made the schema +/// therefore saw the part as it was before the wait, took it for missing and ran all its SQL +/// again over the objects of the other (see `pg_hist_tests`, `a_process_that_waited_for_...`); +/// reading a catalogue table first, as this does, takes the message in. A catalogue table cannot +/// be missing or refused, and no privilege beyond connecting is needed. +const FRESH_SQL: &str = "SELECT 1 FROM pg_catalog.pg_class LIMIT 0"; + +/// Take the schema lock for the rest of the transaction and see what the one who held it before +/// committed (see [`FRESH_SQL`]). Every question "is it there?" that decides on DDL is asked after +/// this, in the transaction that holds the lock. +pub(crate) fn take_lock(c: &mut Client) -> std::result::Result<(), postgres::Error> { + c.batch_execute(LOCK_SQL)?; + c.batch_execute(FRESH_SQL) +} + /// What a part says about its data: findings for [`inspect`]. pub type Audit = fn(&mut Client) -> Result>; @@ -285,10 +303,7 @@ pub fn install_all(c: &mut Client) -> Result<()> { /// Run `f` in a transaction that holds the schema lock, committing when it returns `Ok`. pub(crate) fn with_lock(c: &mut Client, f: impl FnOnce(&mut Client) -> Result) -> Result { c.batch_execute("BEGIN").map_err(|e| describe(&e))?; - let done = c - .batch_execute(LOCK_SQL) - .map_err(|e| describe(&e)) - .and_then(|()| f(c)); + let done = take_lock(c).map_err(|e| describe(&e)).and_then(|()| f(c)); match done { Ok(v) => { c.batch_execute("COMMIT").map_err(|e| describe(&e))?; diff --git a/crates/kernel/obop-storage/src/seams_tests.rs b/crates/kernel/obop-storage/src/seams_tests.rs index 1ee99069..1077f9fb 100644 --- a/crates/kernel/obop-storage/src/seams_tests.rs +++ b/crates/kernel/obop-storage/src/seams_tests.rs @@ -10,7 +10,7 @@ use crate::pgpool::{Class, PgPoolConfig}; use crate::refusal::{self, RefusalCode}; use crate::testing::TestDb; use std::sync::mpsc; -use std::time::Duration; +use std::time::{Duration, Instant}; fn fields(n: i64) -> Fields { let mut m = Fields::new(); @@ -263,8 +263,22 @@ fn read_tx_takes_a_normal_connection_and_not_a_place_in_the_long_read_lane() { ) .unwrap(); // Open the connections before anything is measured: opening one (TCP, authentication) can - // take longer than the window on a loaded machine and is not what is tested. - let warm: Vec<_> = (0..3).map(|_| pg.main.lease(false).unwrap()).collect(); + // take longer than the window on a loaded machine and is not what is tested. A lease that + // gives up leaves its connection to finish opening and land idle in the pool, so asking again + // gets it; the wait is for the three connections to exist, bounded, and says what it saw. + let deadline = Instant::now() + crate::testing::time_budget(60); + let warm = loop { + match (0..3) + .map(|_| pg.main.lease(false)) + .collect::>>() + { + Ok(leases) => break leases, + Err(e) => assert!( + Instant::now() < deadline, + "three connections did not open within the time budget: {e}" + ), + } + }; drop(warm); std::thread::scope(|s| { // The ends of the channels are locals of this closure, so that they are dropped when an diff --git a/crates/obop/tests/backup_cli.rs b/crates/obop/tests/backup_cli.rs index 1ac4bc4c..b2f9fb13 100644 --- a/crates/obop/tests/backup_cli.rs +++ b/crates/obop/tests/backup_cli.rs @@ -58,11 +58,31 @@ fn text(o: &Output) -> String { ) } -/// A database with one account. +/// A text that is in no file by chance: `canary-` and 16 random bytes in hex, new for every call. +/// It is also the name of the account of [`seeded_as`], so that it is in the records of the +/// backup, and in the file of a plain one. A short canary is no proof: a three-letter word such as +/// `ada` is made of hex digits, and the header of a sealed file is hex (the backup id, the key id, +/// the nonce, the wrapped key), so it was found in a sealed file 3 times in 120 backups. Sixteen +/// random bytes are found by chance never (2^-128 per position), and the `canary-` in front is +/// no hex digit at all. +fn canary() -> String { + let hex: String = obop_keys::random::<16>() + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + format!("canary-{hex}") +} + +/// A database with one account, `ada`. fn seeded(dir: &Dir, name: &str) -> String { + seeded_as(dir, name, "ada") +} + +/// A database with one account, `user`. +fn seeded_as(dir: &Dir, name: &str, user: &str) -> String { let url = dir.url(name); let mut child = obop(&[]) - .args(["admin", "create", "ada", "--all-groups", "--db", &url]) + .args(["admin", "create", user, "--all-groups", "--db", &url]) .stdin(Stdio::piped()) .stdout(Stdio::null()) .stderr(Stdio::null()) @@ -91,7 +111,9 @@ fn only_file(dir: &Path, ext: &str) -> PathBuf { #[test] fn a_key_in_the_environment_makes_the_backup_sealed_and_restore_needs_it() { let dir = Dir::new("sealed"); - let url = seeded(&dir, "src.db"); + // The name of the account is the canary the scan below looks for. + let canary = canary(); + let url = seeded_as(&dir, "src.db", &canary); let out = dir.p("backups"); let o = run( &[("OBOP_BACKUP_KEY", KEY_A)], @@ -116,7 +138,7 @@ fn a_key_in_the_environment_makes_the_backup_sealed_and_restore_needs_it() { "verify", file.to_str().unwrap(), "--canary", - "ada", + &canary, ], ); assert!(v.status.success(), "{}", text(&v)); @@ -168,7 +190,7 @@ fn a_key_in_the_environment_makes_the_backup_sealed_and_restore_needs_it() { ); assert!(r.status.success(), "{}", text(&r)); let u = run(&[], &["admin", "users", "--db", &dst]); - assert!(text(&u).contains("ada"), "{}", text(&u)); + assert!(text(&u).contains(&canary), "{}", text(&u)); } #[test] @@ -276,7 +298,8 @@ fn the_server_key_seals_when_there_is_no_backup_key_and_no_key_at_all_means_a_pl #[test] fn a_canary_scan_of_a_plain_backup_fails_and_a_sealed_one_cannot_be_given_a_gz_name() { let dir = Dir::new("canary"); - let url = seeded(&dir, "src.db"); + let canary = canary(); + let url = seeded_as(&dir, "src.db", &canary); let out = dir.p("plain"); let o = run( &[], @@ -293,7 +316,7 @@ fn a_canary_scan_of_a_plain_backup_fails_and_a_sealed_one_cannot_be_given_a_gz_n "verify", plain.to_str().unwrap(), "--canary", - "ada", + &canary, ], ); assert!(!v.status.success(), "{}", text(&v)); diff --git a/crates/obop/tests/multi_process.rs b/crates/obop/tests/multi_process.rs index b8c7a2ad..be3b511f 100644 --- a/crates/obop/tests/multi_process.rs +++ b/crates/obop/tests/multi_process.rs @@ -71,12 +71,25 @@ impl Drop for Server { } } +/// A port nobody listens on, and one this test binary has not handed out before. The tests of a +/// binary run side by side and each starts servers: a port that is free when it is chosen is not +/// bound until the server has opened its database, and in that time another test could be given +/// the same port. Two servers then ask for one port, the later one exits, and the test talks to +/// the other test's server (a token of one database refused as unknown by the "other process"). fn free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .unwrap() - .local_addr() - .unwrap() - .port() + static GIVEN: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + loop { + let port = TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let mut given = GIVEN.lock().unwrap_or_else(|e| e.into_inner()); + if !given.contains(&port) { + given.push(port); + return port; + } + } } fn obop() -> Command { @@ -108,9 +121,15 @@ fn start_cmd(kind: &str, db: &str, env: &[(&str, &str)]) -> Server { .stdout(Stdio::null()) .stderr(std::fs::File::create(&log).unwrap()); let child = cmd.spawn().expect("spawn obop serve"); - let s = Server { child, port, log }; + let mut s = Server { child, port, log }; let deadline = Instant::now() + Duration::from_secs(60); loop { + if let Some(status) = s.child.try_wait().expect("wait for obop serve") { + panic!( + "obop serve exited ({status}) before it came up: {}", + std::fs::read_to_string(&s.log).unwrap_or_default() + ); + } if TcpStream::connect(("127.0.0.1", port)).is_ok() { return s; } diff --git a/crates/obop/tests/pool.rs b/crates/obop/tests/pool.rs index 45659a41..0e322165 100644 --- a/crates/obop/tests/pool.rs +++ b/crates/obop/tests/pool.rs @@ -79,12 +79,25 @@ impl Drop for Server { } } +/// A port nobody listens on, and one this test binary has not handed out before. The tests of a +/// binary run side by side and each starts servers: a port that is free when it is chosen is not +/// bound until the server has opened its database, and in that time another test could be given +/// the same port. Two servers then ask for one port, the later one exits, and the test talks to +/// the other test's server (a token of one database refused as unknown by the "other process"). fn free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .unwrap() - .local_addr() - .unwrap() - .port() + static GIVEN: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + loop { + let port = TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let mut given = GIVEN.lock().unwrap_or_else(|e| e.into_inner()); + if !given.contains(&port) { + given.push(port); + return port; + } + } } fn obop() -> Command { @@ -108,9 +121,15 @@ fn start(db: &str, env: &[(&str, &str)]) -> Server { .stderr(std::fs::File::create(&log).unwrap()) .spawn() .expect("spawn obop serve"); - let s = Server { child, port, log }; + let mut s = Server { child, port, log }; let deadline = Instant::now() + Duration::from_secs(60); loop { + if let Some(status) = s.child.try_wait().expect("wait for obop serve") { + panic!( + "obop serve exited ({status}) before it came up: {}", + std::fs::read_to_string(&s.log).unwrap_or_default() + ); + } if TcpStream::connect(("127.0.0.1", port)).is_ok() { return s; } @@ -291,6 +310,31 @@ fn series(text: &str, name: &str) -> f64 { .unwrap_or_else(|| panic!("no series {name} in:\n{text}")) } +/// The metrics at the first moment the gauge `name` reads `want`, asked again every 50 ms for at +/// most 30 s; the last reading is in the failure. +/// +/// A gauge of the pool is the reading of one instant, and the server is never quite idle: the +/// connectors worker of `obop serve` takes a connection for a few tens of milliseconds every five +/// seconds of uptime (its `tick`, at uptime 5, 10, 15, ... s) and after a create (it reads the +/// connections to deliver the event). A test that reads "nothing in use" once, at the wrong +/// instant, finds that worker (the CI run that failed read `in_use` 1 at uptime 10.5 s). What is +/// asserted is that the pool gets back to nothing in use and nobody waiting: a connection that is +/// never given back, or a caller that never leaves the queue, still fails, at the deadline. +fn metrics_when(port: u16, name: &str, want: f64) -> String { + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let m = metrics(port); + if series(&m, name) == want { + return m; + } + assert!( + Instant::now() < deadline, + "{name} did not get back to {want} within 30 s:\n{m}" + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + #[test] fn an_exhausted_pool_answers_503_with_retry_after_never_a_401_and_the_server_recovers() { let Some(base) = pg_base() else { return }; @@ -416,11 +460,9 @@ fn an_exhausted_pool_answers_503_with_retry_after_never_a_401_and_the_server_rec >= 190.0 ); assert!(series(&m, "obop_db_statement_timeouts_total") >= 1.0); - assert_eq!( - series(&m, "obop_db_pool_waiting{pool=\"main\"}"), - 0.0, - "{m}" - ); + // Nobody is left in the queue: the 200 requests are answered, and the worker of the server, + // which may be queued behind them, leaves it within the acquire timeout. + metrics_when(port, "obop_db_pool_waiting{pool=\"main\"}", 0.0); // `/readyz` stays ready while every connection is busy: the database is there, it is only // full of work. (Two requests hold both connections until their time limit.) @@ -460,7 +502,12 @@ fn an_exhausted_pool_answers_503_with_retry_after_never_a_401_and_the_server_rec "{}", ready.body ); - assert!(ready.took < Duration::from_secs(3), "{:?}", ready.took); + // The probe is answered at its own limit at the latest: `READY_TIMEOUT` is 3 s (obop-http, + // ops.rs), and a probe that has not finished by then, with the pool full, answers ready at that + // limit, which is a few milliseconds after the 3 s (3.0016 s in 2 of 208 runs, where a bound + // of exactly 3 s counted them as failures). What is told apart here is a probe that answers + // from one that hangs (the reads of `http` stop after 60 s). + assert!(ready.took < Duration::from_secs(10), "{:?}", ready.took); // Four times as many requests: the wait queue is bounded, so memory does not follow the load. let (more, _, rss_peak_more) = stampede(port, &token, pid, 800); @@ -516,8 +563,8 @@ fn an_exhausted_pool_answers_503_with_retry_after_never_a_401_and_the_server_rec Some(&json!({"name": "Acme Ltd"})), ); assert_eq!(made.status, 200, "{}", made.body); - let m = metrics(port); - assert_eq!(series(&m, "obop_db_pool_in_use{pool=\"main\"}"), 0.0, "{m}"); + // Every connection comes back: none is lost to the requests that were stuck behind the lock. + metrics_when(port, "obop_db_pool_in_use{pool=\"main\"}", 0.0); let log = std::fs::read_to_string(&server.log).unwrap_or_default(); assert!( !log.contains("panicked"), diff --git a/docs/ISSUES.md b/docs/ISSUES.md index 9ba3b966..a73949e2 100644 --- a/docs/ISSUES.md +++ b/docs/ISSUES.md @@ -203,7 +203,7 @@ The rows K38 and K39 are not storage foundations: they were found while writing | K19 | Found by the check of fail-closed II and not fixed there: (medium, one verifier of two) a pending OAuth flow of a connection that was revoked can still finish and store its tokens, because revocation does not end pending flows and `store_tokens` does not check the state (`obop-connectors` `runtime.rs:1124`); (low) a passkey assertion's three reads (`take_challenge`, the passkey row, `handle_of`, `obop-auth` `passkey.rs:232-237,305-311,160`) still turn a refused read into a counted "denied" answer, so repeated transient refusals can lock a person out for 60 s; a half-finished sign-in made by an older server process is refused by a newer one once the account has an epoch (`mfa.rs:455`); an agent token or OAuth access token issued during an offboarding is not compared with the epoch or the disabled flag after the insert (`lib.rs:493`); the origin entry of an imported record is read by nothing and removed by neither erasure, revocation nor retention (`ingress.rs:298`); `consent.withdraw` can answer "partial, repeat it" for work that is complete (`handlers.rs:690`); `obop site seed` still rewrites settings that are not a JSON object (`obop-site` `seed.rs:66`); `obop agent-token` reports a refused read as "no user X", exit 1 (`main.rs:866`); a self-migrating start still opens the uploaded-file store when no backup directory is set (`ops.rs:93`); the scheduled-backup site is tested through a helper and not through the scheduler (`backup_gate_tests.rs:70`) | open | | K20 | The hosting offer (OR-22) is only a page: before the first sale it needs a processing agreement and terms for hosted instances, a payment provider and invoicing with VAT, a written agreement with Sliplane about resale, the sub-processor and Data Act questions for counsel, the abuse and sanctions approach, and the billing and provisioning of a dedicated server per customer in the control plane (today the Sliplane provisioner makes a service per instance on a shared server and a database on a shared cell). The prices are one table, `web/apps/site/data/hosting.ts` | open (owner: counsel, payment provider) | | K21 | Review of the storage seams (wave 2 package A2, `obop-storage`), what is recorded and not fixed there: (a) interim behaviour change: `Tracked` refuses every write to a model named `record_change` (`tracked::PSEUDO_MODELS`) while the install gate does not yet refuse that name (`RESERVED_MODELS` of `obop-registry` has no caller). Sequence: a package or a module-builder model named `record_change` installs, and its create, update and delete answer "record_change is a reserved model name ...", also for records it already holds; reads work. The gate rule arrives with the D75 package, which turns the failure into a refusal at install; (b) `OBOP_PRODUCTION` is not read and `set_production` has no caller. Sequence if it were read now: `OBOP_PRODUCTION=1` in the environment of `obop restore` into a fresh PostgreSQL database, of the restore drill, of `obop migrate rehearse --scratch`, or of the control plane's first start makes `open` refuse the empty database with "missing: base", and no command installs the schema as the owner (`schema::owner_install` is a Rust function). The D77 package adds `obop migrate run --owner-db`, calls `set_production` from `serve --production` and reads the variable (the rule is `production_from`); (c) a `ctx::Scope` that is forgotten (`std::mem::forget`) stays in force on its thread; scopes dropped out of order are handled; (d) `refusal::parse` accepts only a text that starts with the code word, has one separator and ends with the JSON, so input echoed into another error cannot make a refusal; an error text that a caller composed entirely from input could still look like one, so surfaces pass the errors of the store through unchanged; (e) `schema::inspect` returns `Result, String>` (a check that cannot be answered is an error, not a missing part), where the plan says `Vec`; (f) the byte store's table is `schema::BLOB`, created by `install_all` and by `DbBytes` under `Ddl::IfMissing`, required under `Never`; `inspect` leaves it out because a deployment with `OBOP_BLOB_DIR` has no such table; (g) the `Flaky` of `obop-ops` forwards the new `Backend` methods now and the forwarding lint reads its file by path. Done in the same round: the check is fallible, the retry on a deadlock is recognised by the start of `describe`'s text, scopes are removed by token, the lane test of `read_tx` fails instead of hanging | open (D75, D77 packages) | -| K22 | Test flakes found by the last reviews, not fixed: the warm-up of the read-lane test `read_tx_takes_a_normal_connection_and_not_a_place_in_the_long_read_lane` (`obop-storage/src/seams_tests.rs:263`) opens two connections inside the same 500 ms acquire window it claims not to measure, so a slow connect on a loaded runner panics in the warm-up (warm up on a pool with a long timeout); `backup_cli::a_key_in_the_environment_makes_the_backup_sealed_and_restore_needs_it` (`crates/obop/tests/backup_cli.rs`) looks for the 3-byte canary `ada` in the sealed file and finds it by chance in the plaintext hex of the file's header (the backup id, the key id, the nonce prefix, the wrapped key): 3 of 120 runs of a script that makes the same backup, at byte offsets 47, 252 and 347 (use a longer canary); two pool timing tests (`the_lock_pool_does_not_take_interactive_capacity_and_the_other_way_round`, `a_connection_killed_by_the_server_is_noticed_and_replaced`) failed under heavy machine load and pass alone; and `multi_process::a_session_started_on_one_process_works_on_the_other_and_data_is_shared` (`crates/obop/tests/multi_process.rs:272`) failed once on the main CI run of v0.2.0-r29 (`/api/meta` on the second process answered 401 right after a login on the first) and passed on the next main run (r30) and on the branch runs; the cause is not found, so it is not known whether it is a load flake or a short delay before a new session is visible on a second process (if it recurs, read the session lookup and its caching in `obop-auth` before calling it a flake); and `pg_hist_tests::processes_that_need_the_part_together_make_it_once` failed once with SQLSTATE 40P01 (deadlock detected) on the CI run of the docs-only head of PR 29 and passed on the rerun; the cause is not found | open | +| K22 | Test flakes of the CI runs, found by the last reviews and by the three failures of 2026-10-29. Fixed (branch `fix/ci-flakes`, each measured before and after by repeated runs, the numbers are in the PR): (a) `pg_hist_tests::processes_that_need_the_part_together_make_it_once` failed with SQLSTATE 40P01. This one was a product bug, not a test one: a process that waited for the schema advisory lock (`schema::with_lock`) asked the catalogue about the history part from a cached negative lookup that was older than the commit of the process before it (advisory locks do not refresh the cache), found the part missing, and ran its DDL again, whose `CREATE INDEX IF NOT EXISTS` took ShareLock on the history tables in the opposite order to `obop_history_ensure_partition` (the lock of a batch that writes history). `schema::take_lock` now reads `pg_class` once the lock is held, so the check sees what the process before it made (`pg_hist::Import::flush` takes it too); 24 of 200 runs failed on PostgreSQL 17 and 10 of 200 on 18 before, none of 200 on either after, and the new test `a_process_that_waited_for_the_schema_lock_sees_what_the_one_before_it_made` fails on the old code every time; (b) `backup_cli::a_key_in_the_environment_makes_the_backup_sealed_and_restore_needs_it` found its 3-byte canary `ada` by chance in the hex of the file header (4 of 200 runs): the canary is now 16 random bytes per run (the control test with a plain backup still fails the scan every time); (c) `pool::an_exhausted_pool_answers_503_with_retry_after_never_a_401_and_the_server_recovers` read the gauge `obop_db_pool_in_use` once and found 1 on CI: the server is never idle, its connectors worker takes a connection at uptime 5, 10, 15 s and after every create (shown with an idle server); the test now waits, bounded at 30 s, for the gauges `in_use` and `waiting` to read 0, and a leaked connection still fails at the deadline; the same run showed that its `/readyz` bound of exactly 3 s is the product's own `READY_TIMEOUT` (2 of 208 runs took 3.0016 s), now 10 s; 2 of 208 failed before, none of 208 after (the `in_use` reading itself did not recur locally); (d) the warm-up of `seams_tests::read_tx_takes_a_normal_connection_and_not_a_place_in_the_long_read_lane` opened three connections inside the 500 ms window it claims not to measure: it now retries until they exist (bounded); 50 of 204 runs failed before and none of 204 after, on a machine loaded with 48 busy processes; (e) `pg_pool_tests::the_lock_pool_does_not_take_interactive_capacity_and_the_other_way_round` held its two interactive connections with `pg_sleep(2)` and asserted wall-clock bounds of 1 and 2 s, so a slow machine lost the "pool is full" moment: the holders now hold until the test lets go, and the bounds are gone (the refusals and grants themselves are the proof); 51 of 204 failed before, 1 of 204 and 0 of 204 in two runs after, under 28 busy processes; (f) `a_connection_killed_by_the_server_is_noticed_and_replaced` (and `only_a_statement_on_a_connection_that_broke_is_in_flight`, same helper) guessed 200 ms for the server to end the sessions, and replaced the dead connection inside the default 2 s acquire window: `kill_sessions` now waits until `pg_stat_activity` shows none and the configs have a 30 s window (1 of 204 failed before, none of 204 after, same load). Left open: `multi_process::a_session_started_on_one_process_works_on_the_other_and_data_is_shared` (401 on the second process right after a login on the first, once on CI) is not reproduced (15 runs of the whole binary, none locally) and its cause is not proven; the one explanation that fits (the tests of the binary run side by side and chose their ports with a bind-and-close, so two servers could be given one port, the later exits and the test talks to the other test's server, which does not know the token) is made impossible in the tests (`free_port` hands each port out once per binary, the start-up loop notices a server that exited), but if the 401 recurs read the session lookup in `obop-auth` before calling it a flake; and in the product, found while fixing (a) and not changed: a part that is genuinely partial (an operator dropped one object) is repaired by DDL that can still deadlock with `obop_history_ensure_partition` of another process, `ensure_schema` has no retry for 40P01 (a batch has one), and the `require()` check of a part runs outside the schema lock and can be briefly stale (it heals on the next call) | open | | K23 | Found while closing the review of the first slice of D75 (`obop-storage`, branch w2/d75-core), not fixed there: (a) on main, `Pg::export` calls `r.get::<_, J>(3)` (`pg.rs`, the export loop) and panics with "error deserializing column 3" for a record that holds a number near `f64::MAX` (`1.7976931348623157e308` stored in a jsonb column is read back as a 309-digit integer that `serde_json` refuses); sequence: `insert` a record with that number, then `obop backup` on PostgreSQL; the error should be returned as a `Result`, and the golden script of `golden_tests.rs` avoids the value for that reason; (b) K21 (a) is closed by this slice: the gate refuses a model named `record_change` and the other reserved model names for an install or an upgrade, and the records of such a model from an older package still load, export and restore (the history rows of a backup carry the key `$kind`, `sysfields::ENTRY_KEY`; a row of that model without it is an ordinary record); (c) PostgreSQL `import` still stores a row of the model `record_change` as an ordinary record whether or not it carries `$kind`, because PostgreSQL keeps no history yet: the PostgreSQL history package (D75 WP3) must route rows with the key to its history table and refuse one it cannot read, as memory and SQLite do; (d) `Backend::total()` counts the history entries of memory and SQLite as rows, which the "is the target empty" checks of restore and the drill rely on | done (D75 WP3, the PostgreSQL history): (a) `Pg::export` reads every record through `try_get` and returns the error (test `a_number_a_json_reader_cannot_hold_makes_the_export_an_error_and_not_a_panic`); (c) PostgreSQL `import` routes a row with `$kind` to its history table and refuses one it cannot read, and refuses a backup that carries entries when the history part is not installed; (d) `Backend::total()` counts the entries on PostgreSQL as on the other two stores, so the "is the target empty" checks hold for all three | | K24 | Low items of the check of the D75 core fixes, not fixed: the error table row for `bad action name` in docs/MODULES.md (about line 365) still lists seven generated names and not `history` with its `show_history` hint; the test `a_database_from_before_the_history_opens_without_a_schema_change_and_gets_the_table_when_needed` refuses its batch before the history is written, so its "no table behind" check cannot fail for the lazy creation (use a batch that fails after the history write, and read the schema of the fresh file); an imported history entry number of `i64::MAX` makes the next history write overflow in Memory (`memory.rs:442`, `last_seq + 1`: checked add); `UNSTAMPED` can disagree with `sysfields::row_version` for an integer above `i64::MAX`, so the SQLite backfill skips and under-counts such a record (`sqlite.rs:78`) | open | | K25 | Low items of the review of the default-language package (r21, PR 15), not fixed: (a) a shell file (index.html) larger than 4 MiB (`MAX_SHELL_BYTES` in `obop-http/src/shell.rs`) is passed through without the language tag and nothing warns at start-up or in the docs (a real shell is a few KB; `Shell::check_shell` should warn about the size and the docs row of OBOP_DEFAULT_LANGUAGE should say it); (b) the unknown-length overflow branch of `Shell::serve` (`Unread::TooLarge`, which re-asks and serves the document unchanged) is pinned by no test and its comment says 'No length was declared', a case ServeDir 0.6.11 never produces (the branch is only reachable if the file grows between the stat and the read); (c) the serverMessage test 'falls back to the generic table when the field cannot be known' no longer pins the new split (title kept when only the option words disagree) and its comment is wrong; (d) a `refresh()` already running in `web/apps/backoffice/src/App.tsx` when the person signs out re-applies the old session's language and currency and sets ready afterwards; (e) sign-out resets the language and currency but not the instance theme, settings or modules state that `refresh()` applied (the sign-in screen after an in-app sign-out differs from one after a reload); also ten older rows of the environment reference table (backup, update and migration variables) lack a Default cell | open |