From c063639a01a29b313e82e11d91626f095ed159a9 Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 12:54:32 -0400 Subject: [PATCH 1/7] fix(db): serialize tx-age sweep test against concurrent tx registration Commit-early sweep: implementer leg ended without committing (build-lane contention). Local gates deferred to CI. Co-Authored-By: Claude Fable 5 --- crates/khive-db/src/checkpoint.rs | 66 ++++++++++++++++++++++++++++++- 1 file changed, 65 insertions(+), 1 deletion(-) diff --git a/crates/khive-db/src/checkpoint.rs b/crates/khive-db/src/checkpoint.rs index 5a11675b9..12c017e3d 100644 --- a/crates/khive-db/src/checkpoint.rs +++ b/crates/khive-db/src/checkpoint.rs @@ -2789,8 +2789,22 @@ mod tests { // warm page cache, so sleep past the cap instead of assuming // the elapsed work already crossed it. std::thread::sleep(Duration::from_millis(5)); + // `tx_registry` is a process-wide singleton shared by every test in + // this binary (cargo runs `#[test]`s in parallel threads of the same + // process): `#[serial(tx_registry)]` only excludes other tests that + // carry the same key, not every production write path elsewhere in + // the crate (e.g. `graph_upsert_edges`) that also calls `register()` + // as ordinary telemetry. If one of those happens to still be open and + // was registered before this test's own handle, raw `oldest()` would + // return THAT entry instead of the fixture's reader — see #926. Look + // up this test's own entry by its known label instead of trusting + // global `oldest()`, so the assertion is immune to that noise. + let our_entry = khive_storage::tx_registry::snapshot() + .into_iter() + .find(|(_, label)| label.as_deref() == Some("tx_age_sweep_reader_pin_test")) + .expect("this test's own tx_registry entry must still be open"); let mut tx_age_state = TxAgeSweepState::default(); - let emissions = tx_age_state.observe(khive_storage::tx_registry::oldest(), &config); + let emissions = tx_age_state.observe(Some((tx_id(1), our_entry.0, our_entry.1)), &config); assert!( emissions.iter().any(|e| e.rung == TxAgeRung::Stale && e.label.as_deref() == Some("tx_age_sweep_reader_pin_test")), @@ -2802,6 +2816,56 @@ mod tests { drop(_tx_handle); } + /// Regression for #926: reproduces the exact race that made + /// `tx_age_sweep_names_long_lived_reader_pinning_wal_past_high_water` + /// flaky, directly rather than hoping cargo's test scheduler happens to + /// interleave two unrelated tests. `tx_registry` is a process-wide + /// singleton; a decoy entry registered before this test's own entry is + /// genuinely older, so raw `oldest()` returns the decoy — exactly what an + /// unrelated, concurrently-running write path (e.g. `graph_upsert_edges`) + /// could do in the real suite. The fix (looking up this test's own entry + /// by label via `snapshot()` instead of trusting global `oldest()`) must + /// still correctly name and escalate THIS entry despite that older decoy. + #[test] + #[serial(tx_registry, checkpoint_skip_metrics)] + fn tx_age_sweep_own_entry_survives_concurrent_older_registration() { + let _decoy = khive_storage::tx_registry::register(Some("decoy_unrelated_span".to_string())); + std::thread::sleep(Duration::from_millis(2)); + let _own = khive_storage::tx_registry::register(Some("this_test_own_span".to_string())); + std::thread::sleep(Duration::from_millis(5)); + + // Confirm the race condition is actually reproduced: the decoy must + // currently be the process-wide oldest entry, same as the bug report. + let global_oldest = khive_storage::tx_registry::oldest().expect("registry not empty"); + assert_eq!( + global_oldest.2.as_deref(), + Some("decoy_unrelated_span"), + "test setup must reproduce the race: an older, unrelated entry must be \ + the current global oldest, got: {global_oldest:?}" + ); + + let our_entry = khive_storage::tx_registry::snapshot() + .into_iter() + .find(|(_, label)| label.as_deref() == Some("this_test_own_span")) + .expect("this test's own tx_registry entry must still be open"); + + let config = CheckpointConfig { + tx_warn_secs: Duration::from_millis(1), + tx_max_age_secs: Duration::from_millis(1), + ..CheckpointConfig::default() + }; + let mut state = TxAgeSweepState::default(); + let emissions = state.observe(Some((tx_id(2), our_entry.0, our_entry.1)), &config); + assert!( + emissions + .iter() + .any(|e| e.rung == TxAgeRung::Stale + && e.label.as_deref() == Some("this_test_own_span")), + "expected a Stale emission naming this test's own span despite an older, \ + unrelated concurrent registration, got: {emissions:?}" + ); + } + /// `KHIVE_WAL_WARN_SUSTAINED_CYCLES` overrides the default and rejects 0. #[test] #[serial] From ddfdbb1f3a48c2aae4305b82e67f8411c46191ab Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 17:49:45 -0400 Subject: [PATCH 2/7] test(mcp): keep schedule routing regression hermetic --- crates/khive-mcp/src/serve.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/crates/khive-mcp/src/serve.rs b/crates/khive-mcp/src/serve.rs index 6116fac2f..a9fd7a163 100644 --- a/crates/khive-mcp/src/serve.rs +++ b/crates/khive-mcp/src/serve.rs @@ -5048,11 +5048,21 @@ backend = "kg-backend" ); use clap::Parser; - let args = Args::parse_from(["mcp", "--config", config_path.to_str().expect("utf8 path")]); + let args = Args::parse_from([ + "mcp", + "--config", + config_path.to_str().expect("utf8 path"), + "--no-embed", + ]); let (server, schedule_rt) = build_server(&args).expect("build_server must succeed"); // No `[packs.schedule]` entry above, so it defaults to "main". let rt = schedule_rt.expect("schedule pack is loaded by default"); + assert!( + rt.config().embedding_model.is_none() + && rt.config().additional_embedding_models.is_empty(), + "the backend-routing fixture must remain independent of external embedding models" + ); let marker = "adr106-multi-backend-dispatch-marker"; let action_dsl = format!("create(kind=\"observation\", content=\"{marker}\")"); From 7871897f1dde987658144b9a30e6ed37b94dfe46 Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 17:02:23 -0400 Subject: [PATCH 3/7] test(khive-db): serialize writer-task tests on tx_registry to fix flaky tx-age sweep run_writer_task registers a writer_task_tx handle in the process-wide tx_registry singleton; the checkpoint tx_age_sweep_* tests read that singleton under #[serial(tx_registry)]. The writer_task.rs tests that spawn a writer task were missing from the group, leaking a longer-lived writer_task_tx into the sweep's oldest() read and intermittently failing the assertion on main CI. Join them to the serial group. No prod change. --- crates/khive-db/src/writer_task.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/crates/khive-db/src/writer_task.rs b/crates/khive-db/src/writer_task.rs index c13590065..17738a9a5 100644 --- a/crates/khive-db/src/writer_task.rs +++ b/crates/khive-db/src/writer_task.rs @@ -420,6 +420,7 @@ async fn run_writer_task( mod tests { use super::*; use crate::pool::PoolConfig; + use serial_test::serial; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -432,7 +433,14 @@ mod tests { ConnectionPool::new(cfg).expect("pool open") } + // `#[serial(tx_registry)]`: `run_writer_task` registers a `writer_task_tx` + // handle in the process-wide `tx_registry` singleton for the life of each + // `BEGIN IMMEDIATE`. Tests that observe the registry (the checkpoint + // `tx_age_sweep_*` group) read `tx_registry::oldest()`; an un-serialized + // spawning test here would leak a longer-lived `writer_task_tx` into that + // read and make the sweep name the wrong transaction. Share the key. #[tokio::test] + #[serial(tx_registry)] async fn begin_immediate_failure_replies_error_without_running_op() { // Real lock contention, not a simulation: hold the database-level // write lock from the pool's own writer connection (the unmigrated @@ -504,7 +512,11 @@ mod tests { ); } + // `#[serial(tx_registry)]`: shares the key with the checkpoint + // `tx_age_sweep_*` tests — see the note on + // `begin_immediate_failure_replies_error_without_running_op`. #[tokio::test] + #[serial(tx_registry)] async fn writer_task_executes_op_and_commits() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("writer_task_commit.db"); @@ -627,7 +639,13 @@ mod tests { first.abort(); } + // `#[serial(tx_registry)]`: this test deliberately keeps a request (and + // thus its `writer_task_tx` registry handle) alive past a timeout, so it is + // the worst polluter of the checkpoint `tx_age_sweep_*` reads if left + // un-serialized. Shares the key — see the note on + // `begin_immediate_failure_replies_error_without_running_op`. #[tokio::test] + #[serial(tx_registry)] async fn send_with_timeout_returns_op_result_when_op_outlives_the_timeout() { // `send_with_timeout`'s timeout must bound ONLY the enqueue step — // never the reply-wait. An accepted request (channel not full) must From 1e789bb75d19bdef651fe5f9b8ecd2b7cc8ee5d3 Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 17:56:25 -0400 Subject: [PATCH 4/7] test(kkernel): make code-ingest vector test hermetic --- crates/kkernel/src/code_ingest.rs | 81 +++++++++++++++++++++++++++++-- 1 file changed, 78 insertions(+), 3 deletions(-) diff --git a/crates/kkernel/src/code_ingest.rs b/crates/kkernel/src/code_ingest.rs index 4d0bf5d5e..2d49613eb 100644 --- a/crates/kkernel/src/code_ingest.rs +++ b/crates/kkernel/src/code_ingest.rs @@ -104,6 +104,16 @@ pub async fn run_code_ingest(args: CodeIngestArgs) -> Result<()> { /// after validation has already succeeded and a real (non-dry-run) write is /// about to occur. async fn code_ingest_batch(args: CodeIngestArgs) -> Result { + code_ingest_batch_with_runtime_setup(args, |_| Ok(())).await +} + +async fn code_ingest_batch_with_runtime_setup( + args: CodeIngestArgs, + runtime_setup: F, +) -> Result +where + F: FnOnce(&KhiveRuntime) -> Result<()>, +{ let bytes = std::fs::read(&args.findings) .with_context(|| format!("failed to read {}", args.findings.display()))?; @@ -161,6 +171,7 @@ async fn code_ingest_batch(args: CodeIngestArgs) -> Result { } let runtime = KhiveRuntime::new(cfg).map_err(|e| anyhow::anyhow!("{e}"))?; + runtime_setup(&runtime)?; let resolved_ns = runtime.config().default_namespace.clone(); let token = runtime .authorize(resolved_ns) @@ -479,10 +490,63 @@ fn wal_sidecar_path(db_path: &Path) -> PathBuf { #[cfg(test)] mod tests { + use std::sync::Arc; + + use async_trait::async_trait; + use khive_runtime::{EmbedderProvider, RuntimeError}; + use lattice_embed::{EmbedError, EmbeddingModel, EmbeddingService}; use serial_test::serial; use super::*; + struct FixedEmbeddingService { + dimensions: usize, + } + + #[async_trait] + impl EmbeddingService for FixedEmbeddingService { + async fn embed( + &self, + texts: &[String], + _model: EmbeddingModel, + ) -> std::result::Result>, EmbedError> { + Ok(texts + .iter() + .map(|_| vec![1.0_f32; self.dimensions]) + .collect()) + } + + fn supports_model(&self, _model: EmbeddingModel) -> bool { + true + } + + fn name(&self) -> &'static str { + "code-ingest-test" + } + } + + struct FixedEmbeddingProvider { + name: String, + dimensions: usize, + } + + #[async_trait] + impl EmbedderProvider for FixedEmbeddingProvider { + fn name(&self) -> &str { + &self.name + } + + fn dimensions(&self) -> usize { + self.dimensions + } + + async fn build(&self) -> std::result::Result, RuntimeError> { + Ok(Arc::new(FixedEmbeddingService { + dimensions: self.dimensions, + })) + } + } + fn base_args(findings: PathBuf, db: PathBuf) -> CodeIngestArgs { CodeIngestArgs { findings, @@ -847,9 +911,20 @@ mod tests { let findings = write_valid_findings(tmp.path()); let db = tmp.path().join("scratch.db"); - code_ingest_batch(base_args(findings, db.clone())) - .await - .expect("ingest must succeed"); + code_ingest_batch_with_runtime_setup(base_args(findings, db.clone()), |runtime| { + let model_names = runtime.registered_embedding_model_names(); + assert!( + !model_names.is_empty(), + "test requires at least one configured embedding model" + ); + for name in model_names { + let dimensions = runtime.resolve_embedding_model(Some(&name))?.dimensions(); + runtime.register_embedder(FixedEmbeddingProvider { name, dimensions }); + } + Ok(()) + }) + .await + .expect("ingest must succeed"); let cfg = resolve_runtime_config(RuntimeConfigInputs { db: Some(db.to_str().expect("utf8 path")), From 81f0d8a14a95cbeabd0ab5a01eac9bd4fc211456 Mon Sep 17 00:00:00 2001 From: Leo Date: Tue, 14 Jul 2026 08:25:35 -0400 Subject: [PATCH 5/7] chore: rescope to checkpoint.rs fix only (strays belong to their own lanes) --- crates/khive-mcp/src/serve.rs | 12 +---- crates/kkernel/src/code_ingest.rs | 81 ++----------------------------- 2 files changed, 4 insertions(+), 89 deletions(-) diff --git a/crates/khive-mcp/src/serve.rs b/crates/khive-mcp/src/serve.rs index a9fd7a163..6116fac2f 100644 --- a/crates/khive-mcp/src/serve.rs +++ b/crates/khive-mcp/src/serve.rs @@ -5048,21 +5048,11 @@ backend = "kg-backend" ); use clap::Parser; - let args = Args::parse_from([ - "mcp", - "--config", - config_path.to_str().expect("utf8 path"), - "--no-embed", - ]); + let args = Args::parse_from(["mcp", "--config", config_path.to_str().expect("utf8 path")]); let (server, schedule_rt) = build_server(&args).expect("build_server must succeed"); // No `[packs.schedule]` entry above, so it defaults to "main". let rt = schedule_rt.expect("schedule pack is loaded by default"); - assert!( - rt.config().embedding_model.is_none() - && rt.config().additional_embedding_models.is_empty(), - "the backend-routing fixture must remain independent of external embedding models" - ); let marker = "adr106-multi-backend-dispatch-marker"; let action_dsl = format!("create(kind=\"observation\", content=\"{marker}\")"); diff --git a/crates/kkernel/src/code_ingest.rs b/crates/kkernel/src/code_ingest.rs index 2d49613eb..4d0bf5d5e 100644 --- a/crates/kkernel/src/code_ingest.rs +++ b/crates/kkernel/src/code_ingest.rs @@ -104,16 +104,6 @@ pub async fn run_code_ingest(args: CodeIngestArgs) -> Result<()> { /// after validation has already succeeded and a real (non-dry-run) write is /// about to occur. async fn code_ingest_batch(args: CodeIngestArgs) -> Result { - code_ingest_batch_with_runtime_setup(args, |_| Ok(())).await -} - -async fn code_ingest_batch_with_runtime_setup( - args: CodeIngestArgs, - runtime_setup: F, -) -> Result -where - F: FnOnce(&KhiveRuntime) -> Result<()>, -{ let bytes = std::fs::read(&args.findings) .with_context(|| format!("failed to read {}", args.findings.display()))?; @@ -171,7 +161,6 @@ where } let runtime = KhiveRuntime::new(cfg).map_err(|e| anyhow::anyhow!("{e}"))?; - runtime_setup(&runtime)?; let resolved_ns = runtime.config().default_namespace.clone(); let token = runtime .authorize(resolved_ns) @@ -490,63 +479,10 @@ fn wal_sidecar_path(db_path: &Path) -> PathBuf { #[cfg(test)] mod tests { - use std::sync::Arc; - - use async_trait::async_trait; - use khive_runtime::{EmbedderProvider, RuntimeError}; - use lattice_embed::{EmbedError, EmbeddingModel, EmbeddingService}; use serial_test::serial; use super::*; - struct FixedEmbeddingService { - dimensions: usize, - } - - #[async_trait] - impl EmbeddingService for FixedEmbeddingService { - async fn embed( - &self, - texts: &[String], - _model: EmbeddingModel, - ) -> std::result::Result>, EmbedError> { - Ok(texts - .iter() - .map(|_| vec![1.0_f32; self.dimensions]) - .collect()) - } - - fn supports_model(&self, _model: EmbeddingModel) -> bool { - true - } - - fn name(&self) -> &'static str { - "code-ingest-test" - } - } - - struct FixedEmbeddingProvider { - name: String, - dimensions: usize, - } - - #[async_trait] - impl EmbedderProvider for FixedEmbeddingProvider { - fn name(&self) -> &str { - &self.name - } - - fn dimensions(&self) -> usize { - self.dimensions - } - - async fn build(&self) -> std::result::Result, RuntimeError> { - Ok(Arc::new(FixedEmbeddingService { - dimensions: self.dimensions, - })) - } - } - fn base_args(findings: PathBuf, db: PathBuf) -> CodeIngestArgs { CodeIngestArgs { findings, @@ -911,20 +847,9 @@ mod tests { let findings = write_valid_findings(tmp.path()); let db = tmp.path().join("scratch.db"); - code_ingest_batch_with_runtime_setup(base_args(findings, db.clone()), |runtime| { - let model_names = runtime.registered_embedding_model_names(); - assert!( - !model_names.is_empty(), - "test requires at least one configured embedding model" - ); - for name in model_names { - let dimensions = runtime.resolve_embedding_model(Some(&name))?.dimensions(); - runtime.register_embedder(FixedEmbeddingProvider { name, dimensions }); - } - Ok(()) - }) - .await - .expect("ingest must succeed"); + code_ingest_batch(base_args(findings, db.clone())) + .await + .expect("ingest must succeed"); let cfg = resolve_runtime_config(RuntimeConfigInputs { db: Some(db.to_str().expect("utf8 path")), From 734bc398120049bd9aa087919d6908ba7feba68e Mon Sep 17 00:00:00 2001 From: Leo Date: Tue, 14 Jul 2026 08:58:29 -0400 Subject: [PATCH 6/7] test(db): remove tx sweep global-oldest assumption (#926) --- crates/khive-db/src/checkpoint.rs | 22 +++++++++++++--------- implementation_notes.md | 24 ++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 9 deletions(-) create mode 100644 implementation_notes.md diff --git a/crates/khive-db/src/checkpoint.rs b/crates/khive-db/src/checkpoint.rs index 12c017e3d..2850d5451 100644 --- a/crates/khive-db/src/checkpoint.rs +++ b/crates/khive-db/src/checkpoint.rs @@ -2821,11 +2821,12 @@ mod tests { /// flaky, directly rather than hoping cargo's test scheduler happens to /// interleave two unrelated tests. `tx_registry` is a process-wide /// singleton; a decoy entry registered before this test's own entry is - /// genuinely older, so raw `oldest()` returns the decoy — exactly what an - /// unrelated, concurrently-running write path (e.g. `graph_upsert_edges`) - /// could do in the real suite. The fix (looking up this test's own entry - /// by label via `snapshot()` instead of trusting global `oldest()`) must - /// still correctly name and escalate THIS entry despite that older decoy. + /// genuinely older, so raw `oldest()` cannot return the test fixture — + /// exactly what an unrelated, concurrently-running write path (e.g. + /// `graph_upsert_edges`) could do in the real suite. The fix (looking up + /// this test's own entry by label via `snapshot()` instead of trusting + /// global `oldest()`) must still correctly name and escalate THIS entry + /// despite that older decoy. #[test] #[serial(tx_registry, checkpoint_skip_metrics)] fn tx_age_sweep_own_entry_survives_concurrent_older_registration() { @@ -2834,12 +2835,15 @@ mod tests { let _own = khive_storage::tx_registry::register(Some("this_test_own_span".to_string())); std::thread::sleep(Duration::from_millis(5)); - // Confirm the race condition is actually reproduced: the decoy must - // currently be the process-wide oldest entry, same as the bug report. + // Confirm the race condition is actually reproduced: an entry older + // than this test's own span must currently lead the process-wide + // registry. Another concurrently running test may have registered an + // entry before the decoy, so do not assume the decoy is globally + // oldest; the required invariant is only that our span is not. let global_oldest = khive_storage::tx_registry::oldest().expect("registry not empty"); - assert_eq!( + assert_ne!( global_oldest.2.as_deref(), - Some("decoy_unrelated_span"), + Some("this_test_own_span"), "test setup must reproduce the race: an older, unrelated entry must be \ the current global oldest, got: {global_oldest:?}" ); diff --git a/implementation_notes.md b/implementation_notes.md new file mode 100644 index 000000000..9a2220a33 --- /dev/null +++ b/implementation_notes.md @@ -0,0 +1,24 @@ +# PR #975 implementation notes + +## Change + +- Relaxed the `tx_age_sweep_own_entry_survives_concurrent_older_registration` + setup assertion so it verifies the test's own span is not the process-wide + oldest entry without assuming the local decoy is globally oldest. +- Updated the regression-test documentation to describe the interleaving-safe + invariant. The decoy handle remains live for the whole test. + +## Verification + +- `cargo fmt --manifest-path crates/Cargo.toml --all -- --check` — passed. +- `cargo check --manifest-path crates/Cargo.toml --workspace` — passed. +- `cargo clippy --manifest-path crates/Cargo.toml --workspace --all-targets -- -D warnings` — passed. +- `cargo test --manifest-path crates/Cargo.toml -p khive-db tx_age_sweep_ -- --nocapture` — 10 passed. +- `cargo test --manifest-path crates/Cargo.toml -p khive-db` — 345 passed across unit and contract tests. +- `cargo test --manifest-path crates/Cargo.toml --workspace` — compiled successfully and began running; the environment terminated it with exit 137 after the first crate's 58 tests passed. No test assertion failed before termination. + +## Domain utility + +`medium` — the concurrency-test briefing reinforced that the regression oracle +must express an interleaving-independent invariant; the repository source and +review finding determined the exact code change. From 781cd2579dd46809e2abe712b5ad280f6467df9e Mon Sep 17 00:00:00 2001 From: Leo Date: Tue, 14 Jul 2026 10:38:37 -0400 Subject: [PATCH 7/7] chore: remove workflow scratch notes from tree Merge main and drop implementation_notes.md, a stray process-notes file that does not belong in the source tree. --- implementation_notes.md | 24 ------------------------ 1 file changed, 24 deletions(-) delete mode 100644 implementation_notes.md diff --git a/implementation_notes.md b/implementation_notes.md deleted file mode 100644 index 9a2220a33..000000000 --- a/implementation_notes.md +++ /dev/null @@ -1,24 +0,0 @@ -# PR #975 implementation notes - -## Change - -- Relaxed the `tx_age_sweep_own_entry_survives_concurrent_older_registration` - setup assertion so it verifies the test's own span is not the process-wide - oldest entry without assuming the local decoy is globally oldest. -- Updated the regression-test documentation to describe the interleaving-safe - invariant. The decoy handle remains live for the whole test. - -## Verification - -- `cargo fmt --manifest-path crates/Cargo.toml --all -- --check` — passed. -- `cargo check --manifest-path crates/Cargo.toml --workspace` — passed. -- `cargo clippy --manifest-path crates/Cargo.toml --workspace --all-targets -- -D warnings` — passed. -- `cargo test --manifest-path crates/Cargo.toml -p khive-db tx_age_sweep_ -- --nocapture` — 10 passed. -- `cargo test --manifest-path crates/Cargo.toml -p khive-db` — 345 passed across unit and contract tests. -- `cargo test --manifest-path crates/Cargo.toml --workspace` — compiled successfully and began running; the environment terminated it with exit 137 after the first crate's 58 tests passed. No test assertion failed before termination. - -## Domain utility - -`medium` — the concurrency-test briefing reinforced that the regression oracle -must express an interleaving-independent invariant; the repository source and -review finding determined the exact code change.