From 4f3e1f9f4b9c28c52be6bd0d6c5230287ac83050 Mon Sep 17 00:00:00 2001 From: Jacob Magar Date: Mon, 1 Jun 2026 10:48:07 -0400 Subject: [PATCH 1/2] fix(db): AI rollup refresh staging+swap (syslog-mcp-rvcz) Move the ~4s full GROUP-BY recompute OFF the single WAL writer slot to fix ingest writer starvation. Build the full aggregation into a connection-local TEMP staging table under a read snapshot (zero write lock held), then swap under a sub-millisecond BEGIN IMMEDIATE (DELETE + INSERT-from-staging + meta stamp). Stays a FULL recompute, so it remains correct under retention DELETEs (no MIN/first_seen staleness, no ghost sessions, no event_count drift) where a watermark-incremental refresh would corrupt. Build and swap pin the SAME rusqlite Connection (TEMP tables are connection-local); guarded by an empty-staging-with-live-rows bail (R1) plus debug_assert row-count parity to catch a future pool.get() split refactor. Add rollup_stays_correct_under_concurrent_retention: purges the oldest rows of a surviving session and fully purges another, then asserts first_seen advances to the surviving MIN, the purged session is evicted (no ghost), event_count == COUNT(*), and the rollup equals a from-scratch live recompute. refs syslog-mcp-rvcz --- src/db/queries.rs | 136 ++++++++++++++++++++++++-------- src/db/queries_tests.rs | 171 ++++++++++++++++++++++++++++++++++++---- 2 files changed, 260 insertions(+), 47 deletions(-) diff --git a/src/db/queries.rs b/src/db/queries.rs index 9357d71f..e265074a 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use rusqlite::{params, OptionalExtension}; +use rusqlite::{OptionalExtension, params}; use crate::config::StorageConfig; @@ -592,50 +592,118 @@ pub fn refresh_ai_session_rollup_if_stale(pool: &DbPool) -> Result Result { + // ONE connection for BOTH phases — the TEMP staging table is + // connection-local (see the INVARIANT in the doc comment above). let mut conn = pool.get()?; - // IMMEDIATE (not DEFERRED): take the write lock up front. This function now - // reads (the watermark) before it writes (the DELETE); a DEFERRED tx would - // take a WAL read snapshot on that first read, then have to UPGRADE to a - // writer at the DELETE — and if a concurrent ingest connection committed in - // between, the upgrade fails with SQLITE_BUSY_SNAPSHOT, which busy_timeout - // does NOT retry. Holding the write lock from the start also makes the - // watermark and the GROUP BY observe one consistent state. + + // --- Phase 1: BUILD under a read snapshot (no write lock held) --------- + // A DEFERRED transaction takes a WAL read snapshot on its first read and + // never upgrades to a writer here (we only CREATE TEMP + SELECT), so it + // does not contend for the single WAL writer slot. The watermark and the + // GROUP BY both read from this one consistent snapshot, so the stored + // fingerprint exactly describes the data we aggregate. + let (src_count, src_max_id, staged) = { + let build = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Deferred)?; + let (src_count, src_max_id) = ai_rows_watermark(&build)?; + // TEMP table: connection-local, spills to the temp store (/tmp), never + // to /data. Rebuilt every refresh, so drop any stale prior copy. + build.execute("DROP TABLE IF EXISTS _ai_rollup_staging", [])?; + build.execute( + "CREATE TEMP TABLE _ai_rollup_staging AS + SELECT ai_project, ai_tool, ai_session_id, hostname, + MIN(ai_transcript_path) AS ai_transcript_path, + MIN(timestamp) AS first_seen, + MAX(timestamp) AS last_seen, + COUNT(*) AS event_count + FROM logs + WHERE ai_project IS NOT NULL AND ai_project != '' + AND ai_tool IS NOT NULL AND ai_tool != '' + AND ai_session_id IS NOT NULL AND ai_session_id != '' + GROUP BY ai_project, ai_tool, ai_session_id, hostname", + [], + )?; + let staged: i64 = + build.query_row("SELECT COUNT(*) FROM _ai_rollup_staging", [], |r| r.get(0))?; + // Commit the read snapshot (releases the read lock). The TEMP table + // survives the commit — it is tied to the connection, not the txn. + build.commit()?; + (src_count, src_max_id, staged) + }; + + // R1 guardrail (bead syslog-mcp-rvcz security addendum): the same-connection + // requirement is NOT compile-time enforceable. If a refactor ever ran the + // build on a different pooled connection, the TEMP table would be invisible + // here and the swap would wipe the rollup. The staging table is rebuilt + // every call, so a missing/empty staging table when source rows exist is a + // bug, not a valid empty rollup. Bail BEFORE the destructive swap. + debug_assert!(staged >= 0, "staging row count must be non-negative"); + if staged == 0 && src_count > 0 { + return Err(anyhow::anyhow!( + "ai_session_rollup staging table is empty despite {src_count} live AI \ + rows present — the build and swap MUST share one Connection (TEMP \ + tables are connection-local); refusing to wipe the rollup" + )); + } + + // --- Phase 2: SWAP under a sub-millisecond IMMEDIATE write lock --------- + // IMMEDIATE (not DEFERRED): take the write lock up front. We read nothing + // before the DELETE here, but IMMEDIATE keeps the swap a single short + // writer that never risks an SQLITE_BUSY_SNAPSHOT upgrade failure (which + // busy_timeout does NOT retry). The GROUP BY is already done, so this lock + // is held only for the DELETE + INSERT-from-staging + meta UPDATE. let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; - // Capture the source fingerprint FIRST, within the same transaction as the - // aggregation. Under the held write lock the watermark and the GROUP BY - // observe identical committed state, so the stored fingerprint exactly - // describes the data we aggregate. Any rows that land after this commit - // advance the live fingerprint, correctly marking the rollup stale. - let (src_count, src_max_id) = ai_rows_watermark(&tx)?; tx.execute("DELETE FROM ai_session_rollup", [])?; tx.execute( "INSERT INTO ai_session_rollup (ai_project, ai_tool, ai_session_id, hostname, ai_transcript_path, first_seen, last_seen, event_count) SELECT ai_project, ai_tool, ai_session_id, hostname, - MIN(ai_transcript_path) AS ai_transcript_path, - MIN(timestamp) AS first_seen, - MAX(timestamp) AS last_seen, - COUNT(*) AS event_count - FROM logs - WHERE ai_project IS NOT NULL AND ai_project != '' - AND ai_tool IS NOT NULL AND ai_tool != '' - AND ai_session_id IS NOT NULL AND ai_session_id != '' - GROUP BY ai_project, ai_tool, ai_session_id, hostname", + ai_transcript_path, first_seen, last_seen, event_count + FROM _ai_rollup_staging", [], )?; let row_count: i64 = tx.query_row("SELECT COUNT(*) FROM ai_session_rollup", [], |r| r.get(0))?; + // The swap MUST be faithful: the rollup now holds exactly what we staged. + debug_assert_eq!( + row_count, staged, + "swap row count must equal staged row count" + ); tx.execute( "UPDATE ai_session_rollup_meta SET refreshed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), @@ -646,6 +714,10 @@ pub fn refresh_ai_session_rollup(pool: &DbPool) -> Result { params![row_count, src_count, src_max_id], )?; tx.commit()?; + // Drop the TEMP table so a long-lived pooled connection doesn't carry it + // back into the pool. Best-effort: a failure here doesn't affect the + // already-committed swap. + let _ = conn.execute("DROP TABLE IF EXISTS _ai_rollup_staging", []); Ok(row_count as usize) } @@ -1880,7 +1952,7 @@ pub fn severity_to_num(s: &str) -> Option { return SEVERITY_LEVELS .iter() .position(|&l| l == other) - .map(|i| i as u8) + .map(|i| i as u8); } }; SEVERITY_LEVELS diff --git a/src/db/queries_tests.rs b/src/db/queries_tests.rs index 789b801f..652c8cb0 100644 --- a/src/db/queries_tests.rs +++ b/src/db/queries_tests.rs @@ -1,6 +1,6 @@ use super::*; use crate::config::StorageConfig; -use crate::db::{init_pool, insert_logs_batch, AiRelatedWindow, DbPool, LogBatchEntry}; +use crate::db::{AiRelatedWindow, DbPool, LogBatchEntry, init_pool, insert_logs_batch}; fn test_storage_config(db_path: std::path::PathBuf) -> StorageConfig { StorageConfig::for_test(db_path) @@ -434,9 +434,10 @@ fn tail_logs_filters_multiple_severities() { assert_eq!(rows.len(), 2); assert!(rows.iter().all(|row| row.hostname == "host-a")); - assert!(rows - .iter() - .all(|row| ["err", "warning"].contains(&row.severity.as_str()))); + assert!( + rows.iter() + .all(|row| ["err", "warning"].contains(&row.severity.as_str())) + ); } #[test] @@ -587,9 +588,10 @@ fn search_logs_filters_by_source_ip_prefix_without_fts() { .unwrap(); assert_eq!(rows.len(), 2); - assert!(rows - .iter() - .all(|row| row.source_ip.starts_with("docker://dookie/cortex/"))); + assert!( + rows.iter() + .all(|row| row.source_ip.starts_with("docker://dookie/cortex/")) + ); } #[test] @@ -1119,10 +1121,12 @@ fn investigate_ai_incidents_exact_id_can_fetch_beyond_top_ten() { }, ) .unwrap(); - assert!(!top_ten - .evidence - .iter() - .any(|bundle| bundle.incident.incident_id == target_id)); + assert!( + !top_ten + .evidence + .iter() + .any(|bundle| bundle.incident.incident_id == target_id) + ); let exact = investigate_ai_incidents( &pool, @@ -1448,10 +1452,12 @@ fn rollup_status_reports_refresh_time() { seed_ai_sessions(&pool); // Never refreshed => no staleness timestamp. - assert!(ai_session_rollup_status(&pool) - .unwrap() - .refreshed_at - .is_none()); + assert!( + ai_session_rollup_status(&pool) + .unwrap() + .refreshed_at + .is_none() + ); refresh_ai_session_rollup(&pool).unwrap(); let status = ai_session_rollup_status(&pool).unwrap(); @@ -1517,6 +1523,141 @@ fn rollup_is_exact_after_deletes_recompute_min_max() { } } +/// The staging+swap refresh (bead syslog-mcp-rvcz) MUST stay a correct FULL +/// recompute under retention DELETEs — the exact case a watermark-incremental +/// refresh would silently corrupt. This is the test that distinguishes the +/// (correct) staging+swap from the (corrupt) incremental trap: +/// * purge the OLDEST rows of a SURVIVING session -> first_seen must advance +/// to the surviving MIN (an append-keyed incremental would keep the stale +/// deleted minimum); +/// * fully purge ANOTHER session's rows entirely -> its rollup row must be +/// EVICTED (an append-keyed incremental would leave a ghost session); +/// * event_count must equal the live COUNT(*) -> no upward drift. +/// The post-refresh rollup must be byte-for-byte equal to a from-scratch live +/// aggregation over the surviving rows. +#[test] +fn rollup_stays_correct_under_concurrent_retention() { + let (pool, _dir) = test_pool(); + seed_ai_sessions(&pool); + // Initial full materialization (all 12 seeded sessions present). + refresh_ai_session_rollup(&pool).unwrap(); + let sessions_before = list_ai_sessions(&pool, &default_session_params()) + .unwrap() + .len(); + assert!( + sessions_before >= 2, + "need >=2 sessions to exercise eviction" + ); + + // Pick a SURVIVING session and capture its current first_seen + the + // timestamp of its single oldest row (which we will purge). + let (surviving, old_first_seen, oldest_ts): (String, String, String) = { + let conn = pool.get().unwrap(); + conn.query_row( + "SELECT ai_session_id, MIN(timestamp) FROM logs + WHERE ai_session_id IS NOT NULL GROUP BY ai_session_id + ORDER BY COUNT(*) DESC LIMIT 1", + [], + |r| { + let sid: String = r.get(0)?; + let min_ts: String = r.get(1)?; + Ok((sid.clone(), min_ts.clone(), min_ts)) + }, + ) + .unwrap() + }; + + // Pick a DIFFERENT session to fully purge. + let purged: String = { + let conn = pool.get().unwrap(); + conn.query_row( + "SELECT ai_session_id FROM logs + WHERE ai_session_id IS NOT NULL AND ai_session_id != ?1 + LIMIT 1", + params![surviving], + |r| r.get(0), + ) + .unwrap() + }; + assert_ne!(surviving, purged); + + // Retention purge (mimics maintenance.rs deleting oldest/budget rows with + // NO severity exemption): drop the oldest row of the surviving session AND + // every row of the purged session. + { + let conn = pool.get().unwrap(); + let dropped_old = conn + .execute( + "DELETE FROM logs + WHERE ai_session_id = ?1 AND timestamp = ?2", + params![surviving, oldest_ts], + ) + .unwrap(); + assert!( + dropped_old >= 1, + "must purge the surviving session's oldest row" + ); + let dropped_all = conn + .execute("DELETE FROM logs WHERE ai_session_id = ?1", params![purged]) + .unwrap(); + assert!(dropped_all >= 1, "must fully purge the other session"); + } + + // Refresh AFTER the purge. A correct full recompute (staging+swap) restores + // exactness; the incremental trap would not. + refresh_ai_session_rollup(&pool).unwrap(); + + let rolled = list_ai_sessions(&pool, &default_session_params()).unwrap(); + let live = list_ai_sessions_live(&pool, &default_session_params()).unwrap(); + + // (1) Ghost eviction: the fully-purged session must NOT remain in the rollup. + assert!( + rolled.iter().all(|s| s.ai_session_id != purged), + "fully-purged session must be evicted from the rollup (no ghost row)" + ); + assert_eq!( + rolled.len(), + sessions_before - 1, + "exactly one session should have been evicted" + ); + + // (2) first_seen advanced: the surviving session's MIN must move past the + // now-deleted oldest row. + let surv = rolled + .iter() + .find(|s| s.ai_session_id == surviving) + .expect("surviving session must remain in the rollup"); + assert_ne!( + surv.first_seen, old_first_seen, + "first_seen must advance after the oldest row was purged (incremental \ + would keep the stale minimum)" + ); + let live_surv = live + .iter() + .find(|s| s.ai_session_id == surviving) + .expect("surviving session must be in live aggregation"); + assert_eq!( + surv.first_seen, live_surv.first_seen, + "first_seen must equal the surviving MIN(timestamp)" + ); + + // (3) Byte-for-byte equal to a from-scratch live recompute over survivors. + assert_eq!( + rolled.len(), + live.len(), + "row count must match live recompute" + ); + for (r, l) in rolled.iter().zip(live.iter()) { + assert_eq!(r.ai_session_id, l.ai_session_id); + assert_eq!(r.first_seen, l.first_seen, "first_seen drift vs live"); + assert_eq!(r.last_seen, l.last_seen, "last_seen drift vs live"); + assert_eq!( + r.event_count, l.event_count, + "event_count must equal actual COUNT(*) (no drift)" + ); + } +} + #[test] fn rollup_read_uses_last_seen_index_no_temp_btree() { let (pool, _dir) = test_pool(); From 30228ecd4f320f730a8fceee550d6fbe367fff50 Mon Sep 17 00:00:00 2001 From: Jacob Magar Date: Mon, 1 Jun 2026 13:44:52 -0400 Subject: [PATCH 2/2] fix(db): R1 guard uses full rollup predicate, not broad watermark count MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The R1 empty-staging guard compared staged TEMP-table rows against src_count from ai_rows_watermark, whose predicate (ai_project NOT NULL/!='') is broader than the rollup GROUP BY (which also requires ai_tool and ai_session_id). Rows with ai_project but no recognized tool/session (e.g. OTLP project.path logs) made staged==0 while src_count>0, so refresh ERRORED forever for that data shape. Now count rollup-eligible rows under the SAME read snapshot using the identical full predicate as the staging INSERT, and only bail when staged==0 AND that count>0 — preserving R1's connection-split regression detection while allowing a legitimately-empty rollup (still stamping the meta/fingerprint). bead: syslog-mcp-rvcz --- src/db/queries.rs | 39 ++++++++++++++++++------ src/db/queries_tests.rs | 66 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 9 deletions(-) diff --git a/src/db/queries.rs b/src/db/queries.rs index e265074a..43296ed7 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -637,7 +637,7 @@ pub fn refresh_ai_session_rollup(pool: &DbPool) -> Result { // does not contend for the single WAL writer slot. The watermark and the // GROUP BY both read from this one consistent snapshot, so the stored // fingerprint exactly describes the data we aggregate. - let (src_count, src_max_id, staged) = { + let (src_count, src_max_id, staged, rollup_eligible) = { let build = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Deferred)?; let (src_count, src_max_id) = ai_rows_watermark(&build)?; // TEMP table: connection-local, spills to the temp store (/tmp), never @@ -659,24 +659,45 @@ pub fn refresh_ai_session_rollup(pool: &DbPool) -> Result { )?; let staged: i64 = build.query_row("SELECT COUNT(*) FROM _ai_rollup_staging", [], |r| r.get(0))?; + // Rollup-eligible row count under the SAME read snapshot, using the + // EXACT predicate as the staging INSERT above. Must be computed inside + // this transaction (not after commit / on a fresh connection): under + // one snapshot, any row matching this predicate yields >=1 GROUP BY + // group, so `staged == 0` IMPLIES `rollup_eligible == 0`. The R1 guard + // below relies on that mutual consistency; counting under a different + // snapshot would let a concurrent INSERT revive a false positive. + let rollup_eligible: i64 = build.query_row( + "SELECT COUNT(*) FROM logs + WHERE ai_project IS NOT NULL AND ai_project != '' + AND ai_tool IS NOT NULL AND ai_tool != '' + AND ai_session_id IS NOT NULL AND ai_session_id != ''", + [], + |r| r.get(0), + )?; // Commit the read snapshot (releases the read lock). The TEMP table // survives the commit — it is tied to the connection, not the txn. build.commit()?; - (src_count, src_max_id, staged) + (src_count, src_max_id, staged, rollup_eligible) }; // R1 guardrail (bead syslog-mcp-rvcz security addendum): the same-connection // requirement is NOT compile-time enforceable. If a refactor ever ran the // build on a different pooled connection, the TEMP table would be invisible - // here and the swap would wipe the rollup. The staging table is rebuilt - // every call, so a missing/empty staging table when source rows exist is a - // bug, not a valid empty rollup. Bail BEFORE the destructive swap. + // here and the swap would wipe the rollup. We must distinguish that + // regression from a LEGITIMATELY empty rollup: rows can have `ai_project` + // set but no recognized `ai_tool`/`ai_session_id` (e.g. OTLP logs carrying + // only project.path), which the watermark counts (`src_count > 0`) but the + // rollup GROUP BY correctly excludes (`staged == 0`). Comparing against + // `src_count` would error forever on that data shape. Instead, only bail + // when staging is empty AND rows matching the FULL rollup predicate exist — + // i.e. the build genuinely produced groups but the TEMP table is invisible. debug_assert!(staged >= 0, "staging row count must be non-negative"); - if staged == 0 && src_count > 0 { + if staged == 0 && rollup_eligible > 0 { return Err(anyhow::anyhow!( - "ai_session_rollup staging table is empty despite {src_count} live AI \ - rows present — the build and swap MUST share one Connection (TEMP \ - tables are connection-local); refusing to wipe the rollup" + "ai_session_rollup staging table is empty despite {rollup_eligible} \ + rollup-eligible AI rows present — the build and swap MUST share one \ + Connection (TEMP tables are connection-local); refusing to wipe the \ + rollup" )); } diff --git a/src/db/queries_tests.rs b/src/db/queries_tests.rs index 652c8cb0..d60a606e 100644 --- a/src/db/queries_tests.rs +++ b/src/db/queries_tests.rs @@ -1658,6 +1658,72 @@ fn rollup_stays_correct_under_concurrent_retention() { } } +/// Regression for the R1 guard (bead syslog-mcp-rvcz): rows that carry +/// `ai_project` but have NO recognized `ai_tool`/`ai_session_id` (e.g. OTLP +/// logs with only project.path) are counted by the broad `ai_rows_watermark` +/// predicate but correctly EXCLUDED by the rollup's full GROUP BY predicate. +/// The old guard compared `staged` against the broad watermark `src_count`, so +/// for this data shape `staged == 0` while `src_count > 0` and the refresh +/// ERRORED forever. A legitimately-empty rollup must SUCCEED (returning 0 and +/// still stamping the meta/fingerprint), not raise the R1 error. +#[test] +fn rollup_empty_when_only_broad_project_rows_present_succeeds() { + let (pool, _dir) = test_pool(); + + // Insert rows with ai_project set but ai_tool / ai_session_id EMPTY. These + // match the broad watermark predicate (ai_project NOT NULL/!='') but fail + // the full rollup predicate (which also requires ai_tool and ai_session_id + // NOT NULL/!=''), so the staging GROUP BY yields zero groups. + let batch = vec![ + // empty ai_tool + make_ai_entry( + "2026-05-01T00:00:00Z", + "host0", + "", + "/proj/a", + "sess-1", + "otlp event, no tool", + ), + // empty ai_session_id + make_ai_entry( + "2026-05-01T00:01:00Z", + "host0", + "claude", + "/proj/a", + "", + "otlp event, no session", + ), + // both empty + make_ai_entry( + "2026-05-01T00:02:00Z", + "host1", + "", + "/proj/b", + "", + "otlp event, project only", + ), + ]; + insert_logs_batch(&pool, &batch).unwrap(); + + // The watermark sees these rows (broad predicate) so src_count > 0; the + // rollup predicate excludes them all so staging is legitimately empty. + // Pre-fix this raised the R1 error; post-fix it must SUCCEED with 0 rows. + let total = refresh_ai_session_rollup(&pool).unwrap(); + assert_eq!( + total, 0, + "rollup must be legitimately empty (no rollup-eligible rows)" + ); + + // The meta/fingerprint MUST still be stamped on an empty rollup so + // refresh_ai_session_rollup_if_stale can skip subsequent no-op refreshes. + let status = ai_session_rollup_status(&pool).unwrap(); + assert_eq!(status.row_count, 0, "rollup row_count must be 0"); + assert!( + status.refreshed_at.is_some(), + "meta/fingerprint must be stamped even for an empty rollup" + ); +} + #[test] fn rollup_read_uses_last_seen_index_no_temp_btree() { let (pool, _dir) = test_pool();