Fix the CI test flakes of K22 and the three of 2026-10-29

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;

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 {

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<String> {
@@ -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

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<Vec<Finding>>;
 
@@ -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<T>(c: &mut Client, f: impl FnOnce(&mut Client) -> Result<T>) -> Result<T> {
     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))?;

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::<Result<Vec<_>>>()
+        {
+            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