From f4e8db3c26faea781adacd031f83f7fbae34b8d0 Mon Sep 17 00:00:00 2001 From: rabii-chaarani Date: Mon, 22 Jun 2026 14:13:01 +0930 Subject: [PATCH 1/2] fix: keep graph refresh alive during DB contention Treat Ladybug lock conflicts as expected refresh contention instead of fatal watcher failures. The graph access path now shares transient-error classification, bounded retry, and a repo-local write-intent marker. Incremental writes delete incoming row IDs before COPY so retries and manifest/DB drift do not trip duplicate primary keys. Constraint: Ladybug enforces its own DB lock and can reject concurrent readers or writers during refresh Rejected: Add an external file-lock dependency | stdlib write-intent marker and Ladybug retry handling are sufficient for this recovery path Confidence: high Scope-risk: moderate Directive: Keep MCP refresh loops supervising after recoverable write failures; do not reintroduce fatal propagation from refresh_batch Tested: cargo fmt -- --check; cargo clippy --all-targets -- -D warnings; cargo test; targeted cargo test watch/mcp/graph/db_writer filters; target debug check-health Not-tested: Live restarted MCP daemon under an externally held Ladybug lock --- src/cli/build/command.rs | 14 ++- src/cli/build/request.rs | 2 +- src/cli/graph/health.rs | 16 ++-- src/cli/graph/query.rs | 16 ++-- src/cli/mcp/refresh.rs | 192 ++++++++++++++++++++++++++++++++++---- src/cli/tests/watch.rs | 46 +++++++++ src/cli/watch/mod.rs | 1 + src/cli/watch/native.rs | 20 ++-- src/cli/watch/poll.rs | 20 ++-- src/cli/watch/refresh.rs | 154 ++++++++++++++++++++++++++++++ src/db_writer/access.rs | 165 ++++++++++++++++++++++++++++++++ src/db_writer/deletion.rs | 54 ++++++++++- src/db_writer/mod.rs | 7 +- src/db_writer/tests.rs | 91 +++++++++++++++++- src/db_writer/write.rs | 28 ++++-- 15 files changed, 749 insertions(+), 77 deletions(-) create mode 100644 src/cli/watch/refresh.rs create mode 100644 src/db_writer/access.rs diff --git a/src/cli/build/command.rs b/src/cli/build/command.rs index 8afe32f..394dc45 100644 --- a/src/cli/build/command.rs +++ b/src/cli/build/command.rs @@ -105,15 +105,21 @@ fn write_materialized_database( } else { request.schema_statements.clone() }; + let mut delete_statements = crate::db_writer::partition_delete_statements( + request.previous_manifest.as_ref(), + &response.diff, + ); + delete_statements.extend(crate::db_writer::incoming_row_delete_statements( + request.previous_manifest.as_ref(), + &response.diff, + &response.rebuilt_entries, + )); write_database(LadybugWriteRequest { db_path: request.db_path.clone(), include_fts: request.include_fts, schema_statements, replace_database: response.diff.force_rebuild, - delete_statements: crate::db_writer::partition_delete_statements( - request.previous_manifest.as_ref(), - &response.diff, - ), + delete_statements, copy_statements: response.copy_statements.clone(), }) .map_err(|error| error.to_string()) diff --git a/src/cli/build/request.rs b/src/cli/build/request.rs index 42d0fbc..ef0ae7b 100644 --- a/src/cli/build/request.rs +++ b/src/cli/build/request.rs @@ -210,7 +210,7 @@ pub(in crate::cli) fn default_excluded_parts() -> Vec { .collect() } -#[derive(Debug)] +#[derive(Clone, Debug)] pub(in crate::cli) struct MaterializeOptions { pub(in crate::cli) native_request: Option, pub(in crate::cli) source_root: Option, diff --git a/src/cli/graph/health.rs b/src/cli/graph/health.rs index 96377e1..52e0c0a 100644 --- a/src/cli/graph/health.rs +++ b/src/cli/graph/health.rs @@ -3,7 +3,10 @@ use crate::cli::{ setup::GraphStatePaths, util::{read_json_file, resolve_repo_root}, }; -use lbug::{Connection, Database, SystemConfig, Value}; +use crate::db_writer::{ + connect_ladybug_database, open_ladybug_database, retry_transient_database, READ_RETRY_POLICY, +}; +use lbug::Value; use std::path::{Path, PathBuf}; #[derive(Debug)] @@ -56,14 +59,9 @@ pub(in crate::cli) fn resolve_health_runtime( }) } pub(in crate::cli) fn count_graph_nodes(db_path: &Path) -> Result { - let db = Database::new(db_path, SystemConfig::default().read_only(true)).map_err(|error| { - format!( - "failed to open graph database {}: {error}", - db_path.display() - ) - })?; - let conn = - Connection::new(&db).map_err(|error| format!("failed to connect to graph: {error}"))?; + let db = retry_transient_database(READ_RETRY_POLICY, || open_ladybug_database(db_path, true)) + .map_err(|error| error.to_string())?; + let conn = connect_ladybug_database(&db).map_err(|error| error.to_string())?; let mut result = conn .query("MATCH (n) RETURN count(n) AS total_nodes LIMIT 1") .map_err(|error| format!("failed to query graph health: {error}"))?; diff --git a/src/cli/graph/query.rs b/src/cli/graph/query.rs index 21efc56..dea3fdd 100644 --- a/src/cli/graph/query.rs +++ b/src/cli/graph/query.rs @@ -1,4 +1,7 @@ -use lbug::{Connection, Database, SystemConfig, Value}; +use crate::db_writer::{ + connect_ladybug_database, open_ladybug_database, retry_transient_database, READ_RETRY_POLICY, +}; +use lbug::Value; use serde_json::json; use std::path::Path; @@ -114,14 +117,9 @@ pub(in crate::cli) fn execute_read_only_query( parameters: &serde_json::Map, limit: usize, ) -> Result<(Vec>, bool), String> { - let db = Database::new(db_path, SystemConfig::default().read_only(true)).map_err(|error| { - format!( - "failed to open graph database {}: {error}", - db_path.display() - ) - })?; - let conn = - Connection::new(&db).map_err(|error| format!("failed to connect to graph: {error}"))?; + let db = retry_transient_database(READ_RETRY_POLICY, || open_ladybug_database(db_path, true)) + .map_err(|error| error.to_string())?; + let conn = connect_ladybug_database(&db).map_err(|error| error.to_string())?; let mut result = if parameters.is_empty() { conn.query(statement) .map_err(|error| format!("failed to execute graph query: {error}"))? diff --git a/src/cli/mcp/refresh.rs b/src/cli/mcp/refresh.rs index cc328c4..9df00a5 100644 --- a/src/cli/mcp/refresh.rs +++ b/src/cli/mcp/refresh.rs @@ -7,9 +7,11 @@ use crate::cli::{ watch_file_snapshot, WatchEventFilter, WatchLoopConfig, WatchMessage, }, }; +use crate::db_writer::is_transient_database_error; +use crate::protocol::NativeSyntaxMaterializationResponse; use serde_json::json; use std::{ - collections::VecDeque, + collections::{BTreeSet, VecDeque}, sync::{mpsc::Receiver, Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}, thread, time::{Duration, SystemTime, UNIX_EPOCH}, @@ -17,6 +19,8 @@ use std::{ const DEFAULT_POLL_MS: u64 = 500; const DEFAULT_DEBOUNCE_MS: u64 = 250; +const REFRESH_INITIAL_RETRY_MS: u64 = 100; +const REFRESH_MAX_RETRY_MS: u64 = 1_000; #[derive(Debug)] pub(in crate::cli) struct McpRefreshState { @@ -32,6 +36,8 @@ pub(in crate::cli) struct McpRefreshStatus { pub(in crate::cli) pending: bool, pub(in crate::cli) last_refresh_unix_ms: Option, pub(in crate::cli) last_error: Option, + pub(in crate::cli) last_error_count: usize, + pub(in crate::cli) last_retry_unix_ms: Option, pub(in crate::cli) last_event_count: usize, pub(in crate::cli) last_changed_paths: usize, pub(in crate::cli) last_rebuilt: usize, @@ -48,6 +54,8 @@ impl Default for McpRefreshStatus { pending: false, last_refresh_unix_ms: None, last_error: None, + last_error_count: 0, + last_retry_unix_ms: None, last_event_count: 0, last_changed_paths: 0, last_rebuilt: 0, @@ -76,6 +84,8 @@ impl McpRefreshState { pending: false, last_refresh_unix_ms: None, last_error: Some("refresh status lock poisoned".to_string()), + last_error_count: 1, + last_retry_unix_ms: None, last_event_count: 0, last_changed_paths: 0, last_rebuilt: 0, @@ -93,6 +103,8 @@ impl McpRefreshState { "pending": status.pending, "last_refresh_unix_ms": status.last_refresh_unix_ms, "last_error": status.last_error, + "last_error_count": status.last_error_count, + "last_retry_unix_ms": status.last_retry_unix_ms, "last_event_count": status.last_event_count, "last_changed_paths": status.last_changed_paths, "last_rebuilt": status.last_rebuilt, @@ -128,6 +140,7 @@ impl McpRefreshState { status.refreshing = false; status.pending = false; status.last_error = Some(error); + status.last_error_count = status.last_error_count.saturating_add(1); } } @@ -146,6 +159,26 @@ impl McpRefreshState { } } + fn mark_refresh_error( + &self, + backend: &str, + event_count: usize, + changed_paths: usize, + error: String, + retrying: bool, + ) { + if let Ok(mut status) = self.status.lock() { + status.backend = backend.to_string(); + status.refreshing = false; + status.pending = retrying; + status.last_error = Some(error); + status.last_error_count = status.last_error_count.saturating_add(1); + status.last_retry_unix_ms = retrying.then_some(unix_ms()); + status.last_event_count = event_count; + status.last_changed_paths = changed_paths; + } + } + fn mark_refreshed( &self, backend: &str, @@ -161,6 +194,8 @@ impl McpRefreshState { status.pending = false; status.last_refresh_unix_ms = Some(unix_ms()); status.last_error = None; + status.last_error_count = 0; + status.last_retry_unix_ms = None; status.last_event_count = event_count; status.last_changed_paths = changed_paths; status.last_rebuilt = rebuilt; @@ -212,15 +247,25 @@ fn run_auto_refresh(options: McpServeOptions, state: &Arc) -> R let probe = probe_native_watcher(&runtime.repo_root, &filter, &rx)?; if probe.delivered { state.set_backend("native"); - run_native_refresh_loop( + match run_native_refresh_loop( state, loop_config, - materialize_options, + materialize_options.clone(), filter, watcher, rx, probe.queued, - ) + ) { + Ok(()) => Ok(()), + Err(error) => { + state.set_error("poll", error); + let filter = WatchEventFilter::from_options( + &runtime.repo_root, + &materialize_options, + )?; + run_poll_refresh_loop(state, loop_config, materialize_options, filter) + } + } } else { drop(watcher); state.set_error( @@ -307,26 +352,63 @@ fn refresh_batch( backend: &str, materialize_options: &MaterializeOptions, event_count: usize, - paths: std::collections::BTreeSet, + paths: BTreeSet, +) -> Result<(), String> { + let changed_paths = paths.len(); + if changed_paths == 0 { + return Ok(()); + } + refresh_batch_with(state, backend, event_count, paths, |candidate_paths| { + materialize_candidate_paths(materialize_options, candidate_paths) + .map(|(_, response)| response) + }) +} + +fn refresh_batch_with( + state: &Arc, + backend: &str, + event_count: usize, + paths: BTreeSet, + mut refresh: impl FnMut(Vec) -> Result, ) -> Result<(), String> { let changed_paths = paths.len(); if changed_paths == 0 { return Ok(()); } - state.mark_pending(); - let _guard = state.write_guard()?; - state.mark_refreshing(backend); - let (_, response) = - materialize_candidate_paths(materialize_options, paths.into_iter().collect())?; - state.mark_refreshed( - backend, - event_count, - changed_paths, - response.diff.rebuild_paths().len(), - response.diff.deleted.len(), - response.database_written, - ); - Ok(()) + let candidate_paths = paths.into_iter().collect::>(); + let mut retry_delay = Duration::from_millis(REFRESH_INITIAL_RETRY_MS); + loop { + state.mark_pending(); + let result = { + let _guard = state.write_guard()?; + state.mark_refreshing(backend); + refresh(candidate_paths.clone()) + }; + match result { + Ok(response) => { + state.mark_refreshed( + backend, + event_count, + changed_paths, + response.diff.rebuild_paths().len(), + response.diff.deleted.len(), + response.database_written, + ); + return Ok(()); + } + Err(error) => { + let retrying = is_transient_database_error(&error); + state.mark_refresh_error(backend, event_count, changed_paths, error, retrying); + if !retrying { + return Ok(()); + } + thread::sleep(retry_delay); + retry_delay = retry_delay + .saturating_mul(2) + .min(Duration::from_millis(REFRESH_MAX_RETRY_MS)); + } + } + } } fn unix_ms() -> u128 { @@ -335,3 +417,75 @@ fn unix_ms() -> u128 { .map(|duration| duration.as_millis()) .unwrap_or(0) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::ManifestDiff; + use std::{ + collections::{BTreeMap, BTreeSet}, + sync::atomic::{AtomicUsize, Ordering}, + }; + + #[test] + fn refresh_batch_retries_transient_errors_without_failing_state() { + let state = Arc::new(McpRefreshState::new()); + let attempts = AtomicUsize::new(0); + refresh_batch_with( + &state, + "poll", + 1, + BTreeSet::from(["src/lib.rs".to_string()]), + |_| { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + Err("IO exception: Could not set lock on file".to_string()) + } else { + Ok(NativeSyntaxMaterializationResponse::skipped( + BTreeMap::new(), + ManifestDiff { + added: Vec::new(), + modified: Vec::new(), + unchanged: Vec::new(), + deleted: Vec::new(), + force_rebuild: false, + }, + Vec::new(), + Vec::new(), + BTreeMap::new(), + )) + } + }, + ) + .unwrap(); + + let status = state.snapshot(); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert_eq!(status.backend, "poll"); + assert!(!status.refreshing); + assert!(!status.pending); + assert!(status.last_error.is_none()); + assert_eq!(status.last_error_count, 0); + assert!(status.last_refresh_unix_ms.is_some()); + } + + #[test] + fn refresh_batch_records_non_transient_errors_and_keeps_loop_alive() { + let state = Arc::new(McpRefreshState::new()); + refresh_batch_with( + &state, + "native", + 1, + BTreeSet::from(["src/lib.rs".to_string()]), + |_| Err("parser exploded".to_string()), + ) + .unwrap(); + + let status = state.snapshot(); + assert_eq!(status.backend, "native"); + assert!(!status.refreshing); + assert!(!status.pending); + assert_eq!(status.last_error.as_deref(), Some("parser exploded")); + assert_eq!(status.last_error_count, 1); + assert!(status.last_refresh_unix_ms.is_none()); + } +} diff --git a/src/cli/tests/watch.rs b/src/cli/tests/watch.rs index 939cca0..df14fec 100644 --- a/src/cli/tests/watch.rs +++ b/src/cli/tests/watch.rs @@ -604,3 +604,49 @@ fn watch_once_runs_single_refresh_and_exits() { assert!(root.join(".codebaseGraph").join("manifest.json").exists()); let _ = fs::remove_dir_all(root); } + +#[test] +fn changed_build_recovers_when_manifest_loses_existing_file_entry() { + let root = unique_temp_dir("codebase-graph-rust-manifest-db-drift"); + fs::create_dir_all(&root).unwrap(); + fs::write(root.join("service.py"), "def service():\n return 1\n").unwrap(); + setup_fixture_repo(&root); + + let manifest_path = root.join(".codebaseGraph").join("manifest.json"); + let mut manifest: serde_json::Value = + serde_json::from_str(&fs::read_to_string(&manifest_path).unwrap()).unwrap(); + manifest + .get_mut("files") + .and_then(serde_json::Value::as_object_mut) + .unwrap() + .remove("service.py"); + fs::write( + &manifest_path, + serde_json::to_string_pretty(&manifest).unwrap(), + ) + .unwrap(); + + let mut output = Vec::new(); + run( + [ + "build", + "--repo-root", + root.to_str().unwrap(), + "--mode", + "changed", + "--include", + "service.py", + "--no-git", + "--no-fts", + "--no-semantic-enrichment", + "--json", + ], + &mut output, + ) + .unwrap(); + let text = String::from_utf8(output).unwrap(); + + assert!(text.contains("\"database_written\": true")); + assert!(text.contains("\"service.py\"")); + let _ = fs::remove_dir_all(root); +} diff --git a/src/cli/watch/mod.rs b/src/cli/watch/mod.rs index 57f93b7..beefc41 100644 --- a/src/cli/watch/mod.rs +++ b/src/cli/watch/mod.rs @@ -6,6 +6,7 @@ mod native; mod options; mod output; mod poll; +mod refresh; mod snapshot; mod types; diff --git a/src/cli/watch/native.rs b/src/cli/watch/native.rs index aeacc0a..b5ef774 100644 --- a/src/cli/watch/native.rs +++ b/src/cli/watch/native.rs @@ -2,11 +2,11 @@ use super::{ batch::collect_watch_batch, filter::WatchEventFilter, helpers::watch_max_wait, - output::write_watch_event, + refresh::refresh_watch_batch, types::{WatchMessage, WatchProbeOutcome}, WatchLoopConfig, }; -use crate::cli::build::{materialize_candidate_paths, MaterializeOptions}; +use crate::cli::build::MaterializeOptions; use notify::{Event, RecursiveMode, Watcher}; use std::{ collections::VecDeque, @@ -63,18 +63,16 @@ pub(in crate::cli) fn run_native_watch( Some(batch) => batch, None => continue, }; - let (_, response) = materialize_candidate_paths( - materialize_options, - batch.paths.iter().cloned().collect(), - )?; - write_watch_event( + let refreshed = refresh_watch_batch( stdout, - "refreshed", - Some("native"), + "native", + materialize_options, batch.event_count, - batch.paths.len(), - &response, + &batch.paths, )?; + if !refreshed { + continue; + } refreshes += 1; if loop_config .max_iterations diff --git a/src/cli/watch/poll.rs b/src/cli/watch/poll.rs index 46a3bb6..37be922 100644 --- a/src/cli/watch/poll.rs +++ b/src/cli/watch/poll.rs @@ -1,12 +1,12 @@ use super::{ helpers::watch_max_wait, - output::write_watch_event, + refresh::refresh_watch_batch, snapshot::{watch_file_snapshot, watch_snapshot_diff}, types::WatchChangeBatch, types::WatchFileSnapshot, WatchEventFilter, WatchLoopConfig, }; -use crate::cli::build::{materialize_candidate_paths, MaterializeOptions}; +use crate::cli::build::MaterializeOptions; use std::{ io::Write, time::{Duration, Instant}, @@ -28,18 +28,16 @@ pub(in crate::cli) fn run_poll_watch( Duration::from_millis(loop_config.debounce_ms), watch_max_wait(loop_config.debounce_ms), )?; - let (_, response) = materialize_candidate_paths( - materialize_options, - batch.paths.iter().cloned().collect(), - )?; - write_watch_event( + let refreshed = refresh_watch_batch( stdout, - "refreshed", - Some("poll"), + "poll", + materialize_options, batch.event_count, - batch.paths.len(), - &response, + &batch.paths, )?; + if !refreshed { + continue; + } refreshes += 1; if loop_config .max_iterations diff --git a/src/cli/watch/refresh.rs b/src/cli/watch/refresh.rs new file mode 100644 index 0000000..e4b8aba --- /dev/null +++ b/src/cli/watch/refresh.rs @@ -0,0 +1,154 @@ +use super::output::{write_watch_event, write_watch_status}; +use crate::{ + cli::build::{materialize_candidate_paths, MaterializeOptions}, + db_writer::is_transient_database_error, + protocol::NativeSyntaxMaterializationResponse, +}; +use std::{collections::BTreeSet, io::Write, thread, time::Duration}; + +const INITIAL_RETRY_DELAY: Duration = Duration::from_millis(100); +const MAX_RETRY_DELAY: Duration = Duration::from_millis(1_000); + +pub(in crate::cli) fn refresh_watch_batch( + stdout: &mut W, + backend: &str, + materialize_options: &MaterializeOptions, + event_count: usize, + paths: &BTreeSet, +) -> Result { + refresh_watch_batch_with(stdout, backend, event_count, paths, |candidate_paths| { + materialize_candidate_paths(materialize_options, candidate_paths) + .map(|(_, response)| response) + }) +} + +fn refresh_watch_batch_with( + stdout: &mut W, + backend: &str, + event_count: usize, + paths: &BTreeSet, + mut refresh: impl FnMut(Vec) -> Result, +) -> Result { + let mut delay = INITIAL_RETRY_DELAY; + loop { + match refresh(paths.iter().cloned().collect()) { + Ok(response) => { + write_watch_event( + stdout, + "refreshed", + Some(backend), + event_count, + paths.len(), + &response, + )?; + return Ok(true); + } + Err(error) => { + let transient = is_transient_database_error(&error); + write_watch_status( + stdout, + if transient { "retrying" } else { "error" }, + backend, + Some(&watch_error_reason(&error)), + )?; + if !transient { + return Ok(false); + } + thread::sleep(delay); + delay = delay.saturating_mul(2).min(MAX_RETRY_DELAY); + } + } + } +} + +fn watch_error_reason(error: &str) -> String { + let reason = error.lines().next().unwrap_or("refresh_failed").trim(); + if reason.is_empty() { + "refresh_failed".to_string() + } else { + reason + .split_whitespace() + .collect::>() + .join("_") + .chars() + .take(160) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::ManifestDiff; + use std::{ + collections::BTreeMap, + sync::atomic::{AtomicUsize, Ordering}, + }; + + fn skipped_response() -> NativeSyntaxMaterializationResponse { + NativeSyntaxMaterializationResponse::skipped( + BTreeMap::new(), + ManifestDiff { + added: Vec::new(), + modified: Vec::new(), + unchanged: Vec::new(), + deleted: Vec::new(), + force_rebuild: false, + }, + Vec::new(), + Vec::new(), + BTreeMap::new(), + ) + } + + #[test] + fn watch_error_reason_compacts_multiline_errors() { + assert_eq!( + watch_error_reason("IO exception: Could not set lock\nSee docs"), + "IO_exception:_Could_not_set_lock" + ); + } + + #[test] + fn watch_refresh_retries_transient_errors_before_success() { + let attempts = AtomicUsize::new(0); + let mut output = Vec::new(); + let refreshed = refresh_watch_batch_with( + &mut output, + "poll", + 2, + &BTreeSet::from(["src/lib.rs".to_string()]), + |_| { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + Err("IO exception: Could not set lock on file".to_string()) + } else { + Ok(skipped_response()) + } + }, + ) + .unwrap(); + let text = String::from_utf8(output).unwrap(); + + assert!(refreshed); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert!(text.contains("watch event=retrying backend=poll")); + assert!(text.contains("watch event=refreshed backend=poll")); + } + + #[test] + fn watch_refresh_reports_non_transient_errors_without_success() { + let mut output = Vec::new(); + let refreshed = refresh_watch_batch_with( + &mut output, + "native", + 1, + &BTreeSet::from(["src/lib.rs".to_string()]), + |_| Err("parser exploded".to_string()), + ) + .unwrap(); + let text = String::from_utf8(output).unwrap(); + + assert!(!refreshed); + assert!(text.contains("watch event=error backend=native reason=parser_exploded")); + } +} diff --git a/src/db_writer/access.rs b/src/db_writer/access.rs new file mode 100644 index 0000000..5ad29ad --- /dev/null +++ b/src/db_writer/access.rs @@ -0,0 +1,165 @@ +use crate::error::NativeError; +use lbug::{Connection, Database, SystemConfig}; +use std::{ + fs::{self, OpenOptions}, + io::Write, + path::{Path, PathBuf}, + thread, + time::{Duration, SystemTime}, +}; + +const WRITE_INTENT_FILE: &str = "graph-write-intent.lock"; +const WRITE_INTENT_STALE_AFTER: Duration = Duration::from_secs(60); + +#[derive(Clone, Copy, Debug)] +pub struct RetryPolicy { + max_attempts: usize, + initial_delay: Duration, + max_delay: Duration, +} + +impl RetryPolicy { + pub const fn new(max_attempts: usize, initial_delay: Duration, max_delay: Duration) -> Self { + Self { + max_attempts, + initial_delay, + max_delay, + } + } +} + +pub const READ_RETRY_POLICY: RetryPolicy = + RetryPolicy::new(3, Duration::from_millis(40), Duration::from_millis(160)); +pub const WRITE_RETRY_POLICY: RetryPolicy = + RetryPolicy::new(8, Duration::from_millis(100), Duration::from_millis(1_000)); + +pub struct WriteIntentGuard { + path: PathBuf, +} + +impl Drop for WriteIntentGuard { + fn drop(&mut self) { + let _ = fs::remove_file(&self.path); + } +} + +pub fn is_transient_database_error(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + [ + "could not set lock", + "lock is held", + "database is locked", + "database busy", + "resource busy", + "couldn't replay shadow pages", + "read-only mode", + ] + .iter() + .any(|needle| message.contains(needle)) +} + +pub fn retry_transient_database( + policy: RetryPolicy, + mut operation: impl FnMut() -> Result, +) -> Result { + let max_attempts = policy.max_attempts.max(1); + let mut delay = policy.initial_delay; + for attempt in 1..=max_attempts { + match operation() { + Ok(value) => return Ok(value), + Err(error) => { + if attempt == max_attempts || !is_transient_database_error(&error.to_string()) { + return Err(error); + } + thread::sleep(delay); + delay = delay.saturating_mul(2).min(policy.max_delay); + } + } + } + unreachable!("retry loop always returns") +} + +pub fn open_ladybug_database(db_path: &Path, read_only: bool) -> Result { + if read_only { + wait_for_write_intent(db_path, READ_RETRY_POLICY)?; + } + Database::new(db_path, SystemConfig::default().read_only(read_only)).map_err(|error| { + NativeError::Database(format!( + "failed to open graph database {}: {error}", + db_path.display() + )) + }) +} + +pub fn connect_ladybug_database(database: &Database) -> Result, NativeError> { + Connection::new(database) + .map_err(|error| NativeError::Database(format!("failed to connect to graph: {error}"))) +} + +pub fn acquire_write_intent(db_path: &Path) -> Result { + let path = write_intent_path(db_path); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + retry_transient_database(WRITE_RETRY_POLICY, || { + remove_stale_write_intent(&path)?; + match OpenOptions::new().write(true).create_new(true).open(&path) { + Ok(mut file) => { + writeln!(file, "pid={} unix_ms={}", std::process::id(), unix_ms())?; + Ok(WriteIntentGuard { path: path.clone() }) + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + Err(NativeError::Database(format!( + "graph database busy: write intent exists at {}", + path.display() + ))) + } + Err(error) => Err(NativeError::Io(error)), + } + }) +} + +fn wait_for_write_intent(db_path: &Path, policy: RetryPolicy) -> Result<(), NativeError> { + let path = write_intent_path(db_path); + retry_transient_database(policy, || { + remove_stale_write_intent(&path)?; + if path.exists() { + Err(NativeError::Database(format!( + "graph database busy: refresh/write in progress at {}", + path.display() + ))) + } else { + Ok(()) + } + }) +} + +fn remove_stale_write_intent(path: &Path) -> Result<(), NativeError> { + let Ok(metadata) = fs::metadata(path) else { + return Ok(()); + }; + let Ok(modified) = metadata.modified() else { + return Ok(()); + }; + let Ok(age) = SystemTime::now().duration_since(modified) else { + return Ok(()); + }; + if age >= WRITE_INTENT_STALE_AFTER { + fs::remove_file(path)?; + } + Ok(()) +} + +fn write_intent_path(db_path: &Path) -> PathBuf { + db_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .join(WRITE_INTENT_FILE) +} + +fn unix_ms() -> u128 { + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map(|duration| duration.as_millis()) + .unwrap_or(0) +} diff --git a/src/db_writer/deletion.rs b/src/db_writer/deletion.rs index 070cba6..d2d226f 100644 --- a/src/db_writer/deletion.rs +++ b/src/db_writer/deletion.rs @@ -1,5 +1,5 @@ use super::cypher::{cypher_string_list, quote_identifier}; -use crate::protocol::{ManifestDiff, NativeManifest}; +use crate::protocol::{ManifestDiff, ManifestEntry, NativeManifest}; use std::collections::{BTreeMap, BTreeSet}; const DELETE_BATCH_SIZE: usize = 500; @@ -53,6 +53,58 @@ pub fn partition_delete_statements( edge_deletes } +pub fn incoming_row_delete_statements( + previous_manifest: Option<&NativeManifest>, + diff: &ManifestDiff, + rebuilt_entries: &BTreeMap, +) -> Vec { + if diff.force_rebuild || rebuilt_entries.is_empty() { + return Vec::new(); + } + let (retained_nodes, retained_edges) = retained_manifest_ids(previous_manifest, diff); + let mut edge_deletes = Vec::new(); + let mut node_deletes = Vec::new(); + for entry in rebuilt_entries.values() { + edge_deletes.extend(delete_edge_statements( + &entry.edge_ids, + &entry.edge_types, + &retained_edges, + )); + node_deletes.extend(delete_node_statements( + &entry.node_ids, + &entry.node_types, + &retained_nodes, + )); + } + edge_deletes.extend(node_deletes); + edge_deletes +} + +fn retained_manifest_ids( + previous_manifest: Option<&NativeManifest>, + diff: &ManifestDiff, +) -> (BTreeSet, BTreeSet) { + let Some(manifest) = previous_manifest else { + return (BTreeSet::new(), BTreeSet::new()); + }; + let touched_paths = diff + .deleted + .iter() + .chain(diff.rebuild_paths().iter()) + .cloned() + .collect::>(); + let mut retained_nodes = BTreeSet::new(); + let mut retained_edges = BTreeSet::new(); + for (path, entry) in &manifest.files { + if touched_paths.contains(path) { + continue; + } + retained_nodes.extend(entry.node_ids.iter().cloned()); + retained_edges.extend(entry.edge_ids.iter().cloned()); + } + (retained_nodes, retained_edges) +} + fn delete_edge_statements( edge_ids: &[String], edge_types: &BTreeMap, diff --git a/src/db_writer/mod.rs b/src/db_writer/mod.rs index aade839..ae8ed16 100644 --- a/src/db_writer/mod.rs +++ b/src/db_writer/mod.rs @@ -1,3 +1,4 @@ +mod access; mod cleanup; mod cypher; mod deletion; @@ -6,7 +7,11 @@ mod request; mod schema; mod write; -pub use deletion::partition_delete_statements; +pub use access::{ + acquire_write_intent, connect_ladybug_database, is_transient_database_error, + open_ladybug_database, retry_transient_database, READ_RETRY_POLICY, WRITE_RETRY_POLICY, +}; +pub use deletion::{incoming_row_delete_statements, partition_delete_statements}; pub use extensions::preseed_ladybug_extensions; pub use request::LadybugWriteRequest; pub use write::write_database; diff --git a/src/db_writer/tests.rs b/src/db_writer/tests.rs index eff51bd..cb64440 100644 --- a/src/db_writer/tests.rs +++ b/src/db_writer/tests.rs @@ -1,8 +1,13 @@ -use super::{partition_delete_statements, write_database, LadybugWriteRequest}; +use super::{ + incoming_row_delete_statements, is_transient_database_error, partition_delete_statements, + retry_transient_database, write_database, LadybugWriteRequest, WRITE_RETRY_POLICY, +}; +use crate::error::NativeError; use crate::protocol::{ManifestDiff, ManifestEntry, NativeManifest}; use std::collections::BTreeMap; use std::fs; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; #[test] fn native_writer_loads_json_staging_through_ladybug_copy() { @@ -91,6 +96,90 @@ fn partition_delete_statements_skip_retained_shared_ids() { assert!(!joined.contains("References:shared")); } +#[test] +fn transient_database_error_detection_matches_ladybug_lock_failures() { + assert!(is_transient_database_error( + "IO exception: Could not set lock on file graph.ldb (Lock is held by PID 123)" + )); + assert!(is_transient_database_error( + "Couldn't replay shadow pages under read-only mode" + )); + assert!(!is_transient_database_error( + "Copy exception: Found duplicated primary key value" + )); +} + +#[test] +fn transient_database_retry_replays_operation_until_success() { + let attempts = AtomicUsize::new(0); + retry_transient_database(WRITE_RETRY_POLICY, || { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + Err(NativeError::Database( + "IO exception: Could not set lock on file".to_string(), + )) + } else { + Ok(()) + } + }) + .unwrap(); + + assert_eq!(attempts.load(Ordering::SeqCst), 2); +} + +#[test] +fn incoming_row_delete_statements_skip_retained_shared_ids() { + let mut previous_files = BTreeMap::new(); + previous_files.insert( + "unchanged.py".to_string(), + entry( + &["Symbol:shared"], + &[("Symbol:shared", "Symbol")], + &["References:shared"], + &[("References:shared", "References")], + ), + ); + let manifest = NativeManifest { + schema_version: 1, + ontology: "code_ontology_v1".to_string(), + parser_version: "test".to_string(), + files: previous_files, + }; + let mut rebuilt_entries = BTreeMap::new(); + rebuilt_entries.insert( + "changed.py".to_string(), + entry( + &["Function:changed", "Symbol:shared"], + &[ + ("Function:changed", "Function"), + ("Symbol:shared", "Symbol"), + ], + &["Contains:changed", "References:shared"], + &[ + ("Contains:changed", "Contains"), + ("References:shared", "References"), + ], + ), + ); + + let statements = incoming_row_delete_statements( + Some(&manifest), + &ManifestDiff { + added: Vec::new(), + modified: vec!["changed.py".to_string()], + unchanged: vec!["unchanged.py".to_string()], + deleted: Vec::new(), + force_rebuild: false, + }, + &rebuilt_entries, + ); + let joined = statements.join("\n"); + + assert!(joined.contains("Function:changed")); + assert!(joined.contains("Contains:changed")); + assert!(!joined.contains("Symbol:shared")); + assert!(!joined.contains("References:shared")); +} + fn entry( node_ids: &[&str], node_types: &[(&str, &str)], diff --git a/src/db_writer/write.rs b/src/db_writer/write.rs index 07bedb8..216d50f 100644 --- a/src/db_writer/write.rs +++ b/src/db_writer/write.rs @@ -2,27 +2,35 @@ use super::cleanup::remove_existing_database; use super::extensions::preseed_ladybug_extensions; use super::request::LadybugWriteRequest; use super::schema::schema_statements; +use super::{ + acquire_write_intent, connect_ladybug_database, open_ladybug_database, + retry_transient_database, WRITE_RETRY_POLICY, +}; use crate::error::NativeError; -use lbug::{Connection, Database, SystemConfig}; +use lbug::Connection; +use std::path::Path; pub fn write_database(request: LadybugWriteRequest) -> Result<(), NativeError> { + let _write_intent = acquire_write_intent(Path::new(&request.db_path))?; + retry_transient_database(WRITE_RETRY_POLICY, || write_database_once(&request)) +} + +fn write_database_once(request: &LadybugWriteRequest) -> Result<(), NativeError> { preseed_ladybug_extensions(request.include_fts)?; if request.replace_database { remove_existing_database(&request.db_path)?; } - let database = Database::new(&request.db_path, SystemConfig::default()) - .map_err(|error| NativeError::Database(error.to_string()))?; - let connection = - Connection::new(&database).map_err(|error| NativeError::Database(error.to_string()))?; - for statement in schema_statements(request.include_fts, request.schema_statements) { + let database = open_ladybug_database(Path::new(&request.db_path), false)?; + let connection = connect_ladybug_database(&database)?; + for statement in schema_statements(request.include_fts, request.schema_statements.clone()) { query_ignoring_existing(&connection, &statement)?; } - for statement in request.delete_statements { - query_ignoring_missing(&connection, &statement)?; + for statement in &request.delete_statements { + query_ignoring_missing(&connection, statement)?; } - for statement in request.copy_statements { + for statement in &request.copy_statements { connection - .query(&statement) + .query(statement) .map_err(|error| NativeError::Database(error.to_string()))?; } Ok(()) From e1175a75cd55d1307155563862562b8a7c58167c Mon Sep 17 00:00:00 2001 From: rabii-chaarani Date: Mon, 22 Jun 2026 15:35:51 +0930 Subject: [PATCH 2/2] ci: avoid Windows debug lbug source linking Windows cargo test fell back to building Ladybug from source after the upstream prebuilt Windows archive returned 404, and the debug C++ build exceeds MSVC library/object limits before Rust tests can run. Keep Linux and macOS on the normal debug cargo test path, but run the Windows test leg in release profile, matching the already-passing Windows package build path while still executing the Rust test suite. Constraint: Upstream Ladybug 0.17.1 Windows static archive URL currently returns 404 in CI Rejected: Disable Windows tests | would remove platform coverage instead of preserving the test suite Rejected: Add a new file-lock or build dependency | unnecessary for a CI-only upstream C++ debug-link limit Confidence: medium Scope-risk: narrow Directive: Revisit this once upstream publishes a usable Windows prebuilt lbug archive or the source build no longer exceeds MSVC debug-link limits Tested: cargo run -p xtask -- release-gate Tested: cargo fmt -- --check Tested: cargo test --workspace --locked --release --- .github/workflows/ci.yml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3e0b23e..150f0f8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,14 +43,17 @@ jobs: rustflags: "" prebuilt_lbug: false lbug_archive: "" + test_args: "" - os: macos-latest rustflags: "" prebuilt_lbug: true lbug_archive: liblbug-static-osx-arm64.tar.gz + test_args: "" - os: windows-2022 rustflags: "" prebuilt_lbug: false lbug_archive: "" + test_args: "--release" steps: - name: Check out repository uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 @@ -78,7 +81,7 @@ jobs: - name: Run tests env: RUSTFLAGS: ${{ matrix.rustflags }} - run: cargo test --workspace --locked + run: cargo test --workspace --locked ${{ matrix.test_args }} clippy: name: cargo clippy