diff --git a/knowledge/architecture/graph-runtime.md b/knowledge/architecture/graph-runtime.md index 4d57f90..4d02265 100644 --- a/knowledge/architecture/graph-runtime.md +++ b/knowledge/architecture/graph-runtime.md @@ -6,7 +6,7 @@ tags: - components - graph-runtime - rust -timestamp: 2026-08-06 +timestamp: 2026-08-17 title: Graph Runtime Architecture type: architecture --- @@ -72,7 +72,7 @@ The Materialization API requests these operations but does not publish paths its ## Refresh behavior -The Repository Refresh Service supports continuous and one-shot refresh. It filters and coalesces filesystem events into bounded batches, resolves canonical repository context, normalizes refresh options, and reuses generation-backed materialization. Transient failures are classified and retried with bounded backoff. Runtime entry also gives the janitor an opportunity to recover abandoned work and retry retirement. +The Repository Refresh Service supports continuous and one-shot refresh. Continuous refresh is a cross-process elected role: one nonblocking `refresh.lock` holder performs a manifest catch-up before creating the watcher, while followers remain read-only and retry election with deterministic jitter. Install schema v3 defaults to `refresh.policy = leader`; `off` starts a watcher-free MCP runtime.\n\nThe leader admits supported source and rebuild-triggering configuration events, never admits CodebaseGraph-owned state or storage roots, and collapses churn into one dirty signal plus a bounded path set. Path-count, byte-count, or watcher-channel overflow becomes one full-rescan marker. After the materialization writer lock is held, refresh intent may discard an unchanged candidate without publishing a generation; explicit builds retain their publication semantics. Transient failures are classified and retried with bounded backoff. ## Source evidence diff --git a/knowledge/architecture/graph-storage-lifecycle.md b/knowledge/architecture/graph-storage-lifecycle.md index a512dbc..4f8e60c 100644 --- a/knowledge/architecture/graph-storage-lifecycle.md +++ b/knowledge/architecture/graph-storage-lifecycle.md @@ -3,11 +3,11 @@ description: Immutable generation storage, recoverable run workspaces, artifact resource: repository-architecture tags: - architecture -- graph-storage - generations +- graph-storage - recovery - runbook -timestamp: 2026-08-06 +timestamp: 2026-08-17 title: Graph Storage Lifecycle and Recovery type: architecture --- @@ -54,7 +54,7 @@ storage_root/ 7. **Paths are confined.** Cleanup accepts only expected descendants of the managed root, rejects symlinks and traversal, and never follows links into user data. 8. **Direct paths remain transactional.** Explicit `--db` and `--manifest` targets use adjacent shadow files plus a checksummed recovery journal so the pair cannot remain half-published after a crash. -These rules replace the stale-file write-intent heuristic and all in-place partition deletion or replacement. The compatibility `atomic_rebuild` request field remains accepted but does not re-enable in-place mutation. +These rules replace the stale-file write-intent heuristic and all in-place partition deletion or replacement. The compatibility `atomic_rebuild` request field remains accepted but does not re-enable in-place mutation.\n\n## Refresh ownership\n\n`refresh.lock` is independent from `writer.lock` and `state.lock`. Its nonblocking exclusive holder is the only process allowed to create a repository watcher. Followers do not materialize and retry election every second with up to 250 ms of deterministic jitter; operating-system lock release enables takeover without a persisted leader record. On acquisition, the new leader reconciles the active manifest before it begins watching.\n\nManaged storage places the lock under `storage_root`. Direct storage derives a destination-scoped lock from the explicit database and manifest pair. Lock files reject symlinks. Refresh candidates still acquire the ordinary writer lock for the complete mutation, and an unchanged refresh may close its write session without publishing after comparing against the latest active manifest. ## Run workspace lifecycle @@ -90,13 +90,13 @@ Health and materialization output expose: - active generation; - reused and rebuilt artifact counts; - pending run count and cleanup status; -- physical and logical database sizes. +- physical and logical database sizes;\n- refresh role, leader process, pending state, coalesced and overflow counts, deduplicated refreshes, and the latest no-op reason. A healthy idle managed store reports format v2, one active generation, zero run directories, and `cleanup_pending = false`. ## Recovery runbook -1. Quiesce the repository watcher and any long-lived readers before reinstalling or investigating retirement. +1. Confirm the process reporting refresh role `leader`, then quiesce the repository watcher and any long-lived readers before reinstalling or investigating retirement. 2. Run health and record the storage format, active generation, pending runs, cleanup status, and physical/logical sizes. 3. If a v2 run or publication was interrupted, enter the runtime through health or another repository operation. The janitor will acquire unlocked run journals and recover them before normal work continues. 4. If `cleanup_pending` remains true, confirm no process holds the run or retired-generation lease, then enter the runtime again. Do not manually delete a locked workspace or generation. diff --git a/knowledge/architecture/invariants.md b/knowledge/architecture/invariants.md index 778ed17..efd42e7 100644 --- a/knowledge/architecture/invariants.md +++ b/knowledge/architecture/invariants.md @@ -5,9 +5,9 @@ tags: - architecture - constraints - decisions -- invariants - graph-storage -timestamp: 2026-08-06 +- invariants +timestamp: 2026-08-17 title: Architecture Invariants type: architecture --- @@ -34,7 +34,7 @@ These constraints are the shortest durable test for whether a change still fits 15. **Cleanup is confined and primary errors survive.** Cleanup rejects symlinks and escaping paths, is idempotent, and never masks the failure that caused abort. 16. **Artifacts optimize parsing, not persistence correctness.** Raw partitions are content-addressed across every invalidation dimension; all partitions are assembled deterministically and global semantic enrichment always reruns. 17. **Legacy state is read-only until explicit reinstall.** Schema-v1 reads remain available; mutations return `legacy_storage_requires_reinstall`. Successful reinstall deletes renamed legacy state immediately after validated v2 activation. -18. **Refresh orchestrates rather than reimplements.** Event filtering, batching, recovery, and retry wrap generation-backed materialization instead of duplicating indexing logic. +18. **Refresh orchestrates rather than reimplements.** Event filtering, batching, recovery, and retry wrap generation-backed materialization instead of duplicating indexing logic.\n19. **Refresh ownership and admission are bounded.** One nonblocking refresh lease holder creates the watcher; followers remain read-only. Event churn collapses to one bounded dirty state, overflow forces a full rescan, CodebaseGraph-owned roots are never admitted, and only refresh intent may close an unchanged writer session without publication. ## Knowledge invariants diff --git a/knowledge/architecture/operation-paths.md b/knowledge/architecture/operation-paths.md index 25c6bd1..7c9d507 100644 --- a/knowledge/architecture/operation-paths.md +++ b/knowledge/architecture/operation-paths.md @@ -8,7 +8,7 @@ tags: - mcp - runtime - storage -timestamp: 2026-08-06 +timestamp: 2026-08-17 title: Public Operations and Runtime Paths type: architecture --- @@ -51,17 +51,17 @@ Every repository-scoped operation resolves one `RepoRuntime`: source root, confi Managed reads resolve `active.json` and lease its generation for the entire operation. Direct reads recover any interrupted paired publication before opening their destinations. Runtime entry also recovers abandoned managed runs and retries pending retirement. -Config schema v2 supplies a managed `storage_root`. Schema-v1 deserialization remains available for reads, but the resolved runtime is not writable until explicit reinstall. +Config schema v3 supplies a managed `storage_root`, refresh policy and backend, and bounded materialization defaults. Schema-v2 remains readable and receives v3 defaults for missing fields. Schema-v1 deserialization remains available for reads, but the resolved runtime is not writable until explicit reinstall. ## Graph read path Health, schema, helper catalogs, architecture catalogs, search, context, and raw query operations dispatch from the core to the Graph Read Service. Search reads native full-text indexes and applies lexical/entity ranking. Context expands selected relationship profiles. Raw statements are parameterized, single-statement, read-only, and result-bounded. -Health reports storage format, writability, active generation, reused and rebuilt artifacts, pending runs, cleanup status, and physical/logical database sizes. +Health reports storage format, writability, active generation, reused and rebuilt artifacts, pending runs, cleanup status, physical/logical database sizes, and refresh ownership/coalescing/no-op state. ## Lifecycle and refresh paths -Repository installation, reinstallation, client registration, and removal are coordinated by the Repository Lifecycle Service. Continuous or one-shot refresh is coordinated by the Repository Refresh Service, which invokes the same Materialization API used by explicit builds. +Repository installation, reinstallation, client registration, and removal are coordinated by the Repository Lifecycle Service. Continuous or one-shot refresh is coordinated by the Repository Refresh Service, which invokes the same Materialization API used by explicit builds. Under the default `leader` policy, one cross-process lock holder owns the watcher and followers remain read-only standbys; `off` starts MCP without refresh. Refresh-only materialization may return `database_written = false` after the writer lock proves the active generation already consumed the change. For schema-v1 state, search, context, query, and health remain available. Build, watch, refresh, and install return `legacy_storage_requires_reinstall`. Reinstall moves the legacy state without copying it, restores it after any pre-activation failure, and deletes it immediately after successful v2 activation and validation; there is no grace-period copy. diff --git a/src/adapters/cli/format/help.rs b/src/adapters/cli/format/help.rs index 7f4ef3b..feb8b5a 100644 --- a/src/adapters/cli/format/help.rs +++ b/src/adapters/cli/format/help.rs @@ -3,7 +3,7 @@ pub(in crate::adapters::cli) fn top_level_help() -> &'static str { } pub(crate) fn mcp_help() -> &'static str { - "codebase-graph mcp\n\nUSAGE:\n codebase-graph mcp install [--client ] [--scope ] [--config-path ] [--client-config-path ] [--dry-run] [--json]\n codebase-graph mcp start [--repo-root ] [--config ] [--db ] [--manifest ]\n codebase-graph mcp http [--repo-root ] [--config ] [--db ] [--manifest ] [--host ] [--port ] [--path ] [--allow-remote] [--auth-token |--auth-token-env ]\n\nOPTIONS:\n --repo-root Repository root override; auto-detected when omitted\n --config Setup config path; defaults to .codebaseGraph/config.json\n --db Ladybug database path override\n --manifest Manifest path override\n --host HTTP bind host; defaults to 127.0.0.1\n --port HTTP bind port; defaults to 8765\n --path HTTP endpoint path; defaults to /mcp\n --allow-remote Permit non-local HTTP bind when an auth token is supplied\n --auth-token Bearer token required for HTTP requests\n --auth-token-env Environment variable containing the bearer token" + "codebase-graph mcp\n\nUSAGE:\n codebase-graph mcp install [--client ] [--scope ] [--config-path ] [--client-config-path ] [--dry-run] [--json]\n codebase-graph mcp start [--repo-root ] [--config ] [--db ] [--manifest ] [--refresh-policy off|leader] [memory options]\n codebase-graph mcp http [--repo-root ] [--config ] [--db ] [--manifest ] [--refresh-policy off|leader] [memory options] [--host ] [--port ] [--path ] [--allow-remote] [--auth-token |--auth-token-env ]\n\nOPTIONS:\n --repo-root Repository root override; auto-detected when omitted\n --config Setup config path; defaults to .codebaseGraph/config.json\n --db Ladybug database path override\n --manifest Manifest path override\n --refresh-policy off or leader; defaults to config or leader\n --worker-memory-mib Worker RSS budget; defaults to config or 768\n --rust-memory-mib Rust working-state budget; defaults to config or 384\n --spill-chunk-mib External-sort chunk budget; defaults to config or 32\n --max-parallelism Materialization worker count; defaults to config or 2\n --host HTTP bind host; defaults to 127.0.0.1\n --port HTTP bind port; defaults to 8765\n --path HTTP endpoint path; defaults to /mcp\n --allow-remote Permit non-local HTTP bind when an auth token is supplied\n --auth-token Bearer token required for HTTP requests\n --auth-token-env Environment variable containing the bearer token" } pub(in crate::adapters::cli) fn mcp_install_help() -> &'static str { diff --git a/src/adapters/cli/tests/dispatch_materialize.rs b/src/adapters/cli/tests/dispatch_materialize.rs index ab516da..beede74 100644 --- a/src/adapters/cli/tests/dispatch_materialize.rs +++ b/src/adapters/cli/tests/dispatch_materialize.rs @@ -485,7 +485,15 @@ fn setup_materializes_graph_and_writes_config() { &fs::read_to_string(root.join(".codebaseGraph").join("config.json")).unwrap(), ) .unwrap(); - assert_eq!(config["schema_version"], 2); + assert_eq!(config["schema_version"], 3); + assert_eq!(config["refresh"]["policy"], "leader"); + assert_eq!(config["refresh"]["backend"], "auto"); + assert_eq!(config["materialization"]["include_fts"], true); + assert_eq!(config["materialization"]["semantic_enrichment"], true); + assert_eq!(config["materialization"]["worker_memory_mib"], 768); + assert_eq!(config["materialization"]["rust_memory_mib"], 384); + assert_eq!(config["materialization"]["spill_chunk_mib"], 32); + assert_eq!(config["materialization"]["max_parallelism"], 2); assert_eq!(config["mcp"]["server_name"], "codebase_graph"); let instructions = fs::read_to_string(root.join("AGENTS.md")).unwrap(); for expected in [ diff --git a/src/adapters/cli/tests/fixtures.rs b/src/adapters/cli/tests/fixtures.rs index 24ee560..c82f0f4 100644 --- a/src/adapters/cli/tests/fixtures.rs +++ b/src/adapters/cli/tests/fixtures.rs @@ -77,6 +77,11 @@ pub(super) fn test_http_options(root: PathBuf, auth_token: Option<&str>) -> McpH db: None, manifest: None, api: None, + refresh_policy: None, + worker_memory_mib: None, + rust_memory_mib: None, + spill_chunk_mib: None, + max_parallelism: None, }, host: "127.0.0.1".to_string(), port: 8765, diff --git a/src/adapters/cli/tests/install.rs b/src/adapters/cli/tests/install.rs index c5d99cc..88e9069 100644 --- a/src/adapters/cli/tests/install.rs +++ b/src/adapters/cli/tests/install.rs @@ -57,8 +57,8 @@ fn install_skips_materialization_when_graph_state_already_exists() { } #[test] -fn install_writes_managed_v2_config_without_static_database_or_manifest_paths() { - let root = unique_temp_dir("codebase-graph-rust-install-managed-v2"); +fn install_writes_schema_v3_managed_config_without_static_database_or_manifest_paths() { + let root = unique_temp_dir("codebase-graph-rust-install-managed-v3"); fs::create_dir_all(&root).unwrap(); fs::write(root.join("service.py"), "def helper():\n return 1\n").unwrap(); @@ -87,7 +87,13 @@ fn install_writes_managed_v2_config_without_static_database_or_manifest_paths() &fs::read_to_string(root.join(".codebaseGraph").join("config.json")).unwrap(), ) .unwrap(); - assert_eq!(config["schema_version"], 2); + assert_eq!(config["schema_version"], 3); + assert_eq!(config["refresh"]["policy"], "leader"); + assert_eq!(config["refresh"]["backend"], "auto"); + assert_eq!(config["materialization"]["worker_memory_mib"], 768); + assert_eq!(config["materialization"]["rust_memory_mib"], 384); + assert_eq!(config["materialization"]["spill_chunk_mib"], 32); + assert_eq!(config["materialization"]["max_parallelism"], 2); assert!(config.get("database_path").is_none()); assert!(config.get("manifest_path").is_none()); let expected_storage_root = diff --git a/src/adapters/cli/tests/mcp.rs b/src/adapters/cli/tests/mcp.rs index 9dd105e..44e195e 100644 --- a/src/adapters/cli/tests/mcp.rs +++ b/src/adapters/cli/tests/mcp.rs @@ -1,5 +1,104 @@ use super::*; +#[test] +fn mcp_options_resolve_config_defaults_and_cli_overrides() { + let root = unique_temp_dir("codebase-graph-rust-mcp-runtime-settings"); + let state = root.join(".codebaseGraph"); + fs::create_dir_all(&state).unwrap(); + fs::write( + state.join("config.json"), + serde_json::to_vec(&json!({ + "schema_version": 3, + "repo_root": root, + "refresh": {"policy": "off", "backend": "auto"}, + "materialization": { + "worker_memory_mib": 900, + "rust_memory_mib": 450, + "spill_chunk_mib": 45, + "max_parallelism": 3 + } + })) + .unwrap(), + ) + .unwrap(); + + let options = McpServeOptions::parse( + &[ + "--repo-root".into(), + root.to_string_lossy().into_owned(), + "--refresh-policy".into(), + "leader".into(), + "--worker-memory-mib".into(), + "1024".into(), + "--rust-memory-mib".into(), + "512".into(), + "--spill-chunk-mib".into(), + "64".into(), + "--max-parallelism".into(), + "4".into(), + ], + format::mcp_help(), + ) + .unwrap(); + let settings = options.runtime_settings().unwrap(); + assert_eq!( + settings.refresh_policy, + crate::api::context::GraphRefreshPolicy::Leader + ); + assert_eq!(settings.worker_memory_mib, 1024); + assert_eq!(settings.rust_memory_mib, 512); + assert_eq!(settings.spill_chunk_mib, 64); + assert_eq!(settings.max_parallelism, 4); + + let http = McpHttpOptions::parse( + &[ + "--repo-root".into(), + root.to_string_lossy().into_owned(), + "--refresh-policy".into(), + "off".into(), + "--worker-memory-mib".into(), + "800".into(), + "--rust-memory-mib".into(), + "400".into(), + ], + format::mcp_help(), + ) + .unwrap(); + assert_eq!( + http.serve.runtime_settings().unwrap().refresh_policy, + crate::api::context::GraphRefreshPolicy::Off + ); + let _ = fs::remove_dir_all(root); +} + +#[test] +fn mcp_options_reject_invalid_resource_limits() { + let error = McpServeOptions::parse( + &["--worker-memory-mib".into(), "0".into()], + format::mcp_help(), + ) + .unwrap_err(); + assert!(error.contains("positive integer")); + + let root = unique_temp_dir("codebase-graph-rust-mcp-invalid-limits"); + fs::create_dir_all(&root).unwrap(); + let options = McpServeOptions::parse( + &[ + "--repo-root".into(), + root.to_string_lossy().into_owned(), + "--worker-memory-mib".into(), + "256".into(), + ], + format::mcp_help(), + ) + .unwrap(); + assert!(options + .runtime_settings() + .unwrap_err() + .contains("rust_memory_mib")); + let _ = fs::remove_dir_all(root); +} + #[test] fn mcp_graph_query_binds_json_parameters() { let root = unique_temp_dir("codebase-graph-rust-mcp-query-params"); @@ -13,6 +112,11 @@ fn mcp_graph_query_binds_json_parameters() { db: None, manifest: None, api: None, + refresh_policy: None, + worker_memory_mib: None, + rust_memory_mib: None, + spill_chunk_mib: None, + max_parallelism: None, }; let result = mcp_call_tool_result( "graph_query", @@ -107,6 +211,11 @@ fn mcp_stdio_serves_tools_and_tool_errors() { db: None, manifest: None, api: None, + refresh_policy: None, + worker_memory_mib: None, + rust_memory_mib: None, + spill_chunk_mib: None, + max_parallelism: None, }; let mut output = Vec::new(); serve_mcp_stdio(&options, std::io::Cursor::new(input), &mut output).unwrap(); @@ -164,6 +273,59 @@ fn mcp_stdio_serves_tools_and_tool_errors() { let _ = fs::remove_dir_all(root); } +#[test] +fn mcp_stdio_refresh_policy_off_is_watcher_free() { + let root = unique_temp_dir("codebase-graph-rust-mcp-refresh-off"); + fs::create_dir_all(&root).unwrap(); + fs::write(root.join("service.py"), "def helper():\n return 1\n").unwrap(); + setup_fixture_repo(&root); + + let input = [ + json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2025-11-25"}, + }), + json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "graph_health", + "arguments": {"include_structured_content": true}, + }, + }), + ] + .iter() + .map(serde_json::to_string) + .collect::, _>>() + .unwrap() + .join("\n") + + "\n"; + let options = McpServeOptions { + repo_root: Some(root.clone()), + config: None, + db: None, + manifest: None, + api: None, + refresh_policy: Some(crate::api::context::GraphRefreshPolicy::Off), + worker_memory_mib: None, + rust_memory_mib: None, + spill_chunk_mib: None, + max_parallelism: None, + }; + let mut output = Vec::new(); + serve_mcp_stdio(&options, std::io::Cursor::new(input), &mut output).unwrap(); + let responses = String::from_utf8(output) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert!(responses[1]["result"]["structuredContent"]["refresh"].is_null()); + let _ = fs::remove_dir_all(root); +} + #[test] fn mcp_http_rejects_remote_bind_without_auth_token() { let error = McpHttpOptions::parse( diff --git a/src/adapters/cli/tests/watch.rs b/src/adapters/cli/tests/watch.rs index 99d37db..32d3475 100644 --- a/src/adapters/cli/tests/watch.rs +++ b/src/adapters/cli/tests/watch.rs @@ -150,7 +150,7 @@ fn watch_filter_honors_ignore_config_and_cli_excludes() { } #[test] -fn watch_filter_keeps_unsupported_files_when_unignored() { +fn watch_filter_rejects_unsupported_files_when_unignored() { let root = unique_temp_dir("codebase-graph-rust-watch-filter-unsupported"); fs::create_dir_all(&root).unwrap(); let filter = watch_filter_for(&root, &[]); @@ -162,13 +162,56 @@ fn watch_filter_keeps_unsupported_files_when_unignored() { &["notes.txt"], ); + assert!(filter.relevant_paths(&event).is_empty()); + let _ = fs::remove_dir_all(root); +} + +#[test] +fn watch_filter_ignores_generated_directories_unless_explicitly_included() { + let root = unique_temp_dir("codebase-graph-rust-watch-filter-generated"); + fs::create_dir_all(&root).unwrap(); + let default_filter = watch_filter_for(&root, &[]); + let included_filter = watch_filter_for(&root, &["--include", ".astro/*"]); + let event = watch_test_event( + &root, + EventKind::Modify(notify::event::ModifyKind::Data( + notify::event::DataChange::Content, + )), + &[".astro/generated/routes.rs"], + ); + + assert!(default_filter.relevant_paths(&event).is_empty()); assert_eq!( - filter.relevant_paths(&event), - BTreeSet::from(["notes.txt".to_string()]) + included_filter.relevant_paths(&event), + BTreeSet::from([".astro/generated/routes.rs".to_string()]) ); let _ = fs::remove_dir_all(root); } +#[test] +fn watch_filter_always_admits_configuration_triggers() { + let root = unique_temp_dir("codebase-graph-rust-watch-filter-config"); + fs::create_dir_all(root.join(".codebaseGraph")).unwrap(); + fs::write(root.join(".codebaseGraphignore"), "*.json\n").unwrap(); + fs::write(root.join(".codebaseGraph/config.json"), "{}\n").unwrap(); + let filter = watch_filter_for(&root, &["--exclude", "*.json"]); + + for path in [".codebaseGraph/config.json", ".codebaseGraphignore"] { + let event = watch_test_event( + &root, + EventKind::Modify(notify::event::ModifyKind::Data( + notify::event::DataChange::Content, + )), + &[path], + ); + assert_eq!( + filter.relevant_paths(&event), + BTreeSet::from([path.to_string()]) + ); + } + let _ = fs::remove_dir_all(root); +} + #[test] fn watch_filter_accepts_relative_notify_paths() { let root = unique_workspace_dir("codebase-graph-rust-watch-relative"); @@ -226,6 +269,7 @@ fn watch_batch_coalesces_burst_events_until_quiet() { &["a.py"], )), &rx, + None, &mut queued, &filter, Duration::from_millis(10), @@ -242,6 +286,86 @@ fn watch_batch_coalesces_burst_events_until_quiet() { let _ = fs::remove_dir_all(root); } +#[test] +fn watch_batch_overflow_collapses_to_one_full_rescan_marker() { + let mut batch = WatchChangeBatch::default(); + batch.extend_paths((0..=4_096).map(|index| format!("src/file-{index}.rs"))); + + assert!(batch.full_rescan); + assert!(batch.paths.is_empty()); + assert_eq!(batch.overflow_count, 1); + + batch.extend_paths(["src/ignored-after-overflow.rs".to_string()]); + assert!(batch.paths.is_empty()); + assert_eq!(batch.overflow_count, 1); +} + +#[test] +fn watch_batch_counts_filtered_paths_without_retaining_them() { + let root = unique_temp_dir("codebase-graph-rust-watch-filter-count"); + fs::create_dir_all(&root).unwrap(); + let filter = watch_filter_for(&root, &[]); + let mut batch = WatchChangeBatch::default(); + + apply_watch_message( + WatchMessage::Event(watch_test_event( + &root, + EventKind::Modify(notify::event::ModifyKind::Any), + &["keep.rs", "notes.txt"], + )), + &filter, + &mut batch, + ) + .unwrap(); + + assert_eq!(batch.paths, BTreeSet::from(["keep.rs".to_string()])); + assert_eq!(batch.filtered_event_count, 1); + let _ = fs::remove_dir_all(root); +} + +#[test] +fn watch_batch_directory_remove_and_rename_force_full_rescan() { + let root = unique_temp_dir("codebase-graph-rust-watch-directory-change"); + fs::create_dir_all(&root).unwrap(); + let filter = watch_filter_for(&root, &[]); + + for event in [ + watch_test_event( + &root, + EventKind::Remove(notify::event::RemoveKind::Folder), + &["src/removed"], + ), + watch_test_event( + &root, + EventKind::Modify(notify::event::ModifyKind::Name( + notify::event::RenameMode::Both, + )), + &["src/old", "src/new"], + ), + ] { + let mut batch = WatchChangeBatch::default(); + apply_watch_message(WatchMessage::Event(event), &filter, &mut batch).unwrap(); + assert!(batch.full_rescan); + assert!(batch.paths.is_empty()); + assert_eq!(batch.event_count, 1); + } + + let mut unsupported_file = WatchChangeBatch::default(); + apply_watch_message( + WatchMessage::Event(watch_test_event( + &root, + EventKind::Remove(notify::event::RemoveKind::File), + &["notes.txt"], + )), + &filter, + &mut unsupported_file, + ) + .unwrap(); + assert!(!unsupported_file.full_rescan); + assert!(unsupported_file.paths.is_empty()); + let _ = fs::remove_dir_all(root); +} + #[test] fn watch_batch_flushes_under_sustained_churn() { let root = unique_temp_dir("codebase-graph-rust-watch-churn"); @@ -272,6 +396,7 @@ fn watch_batch_flushes_under_sustained_churn() { &["initial.py"], )), &rx, + None, &mut queued, &filter, Duration::from_millis(100), @@ -312,6 +437,7 @@ fn watch_batch_coalesces_queued_events_into_follow_up_refresh() { let batch = collect_watch_batch( rx.recv().unwrap(), &rx, + None, &mut queued, &filter, Duration::from_millis(10), @@ -342,6 +468,7 @@ fn watch_batch_propagates_watcher_errors() { let error = collect_watch_batch( WatchMessage::Error("backend failed".to_string()), &rx, + None, &mut queued, &filter, Duration::from_millis(1), @@ -706,6 +833,58 @@ fn watch_once_runs_single_refresh_and_exits() { let _ = fs::remove_dir_all(root); } +#[test] +fn unchanged_refresh_keeps_generation_but_explicit_build_still_publishes() { + let root = unique_temp_dir("codebase-graph-rust-refresh-noop"); + fs::create_dir_all(&root).unwrap(); + fs::write(root.join("service.py"), "def helper():\n return 1\n").unwrap(); + setup_fixture_repo(&root); + let active_path = root.join(".codebaseGraph/storage/active.json"); + let initial_active = fs::read(&active_path).unwrap(); + + let mut watch_output = Vec::new(); + run( + [ + "watch", + "--repo-root", + root.to_str().unwrap(), + "--once", + "--no-git", + "--no-fts", + "--no-semantic-enrichment", + ], + &mut watch_output, + ) + .unwrap(); + assert!(String::from_utf8(watch_output) + .unwrap() + .contains("database_written=false")); + assert_eq!(fs::read(&active_path).unwrap(), initial_active); + + let mut build_output = Vec::new(); + run( + [ + "build", + "--repo-root", + root.to_str().unwrap(), + "--mode", + "changed", + "--no-git", + "--no-fts", + "--no-semantic-enrichment", + "--json", + ], + &mut build_output, + ) + .unwrap(); + assert_eq!( + serde_json::from_slice::(&build_output).unwrap()["database_written"], + true + ); + assert_ne!(fs::read(&active_path).unwrap(), initial_active); + 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"); diff --git a/src/adapters/mcp/http.rs b/src/adapters/mcp/http.rs index 8756588..9021400 100644 --- a/src/adapters/mcp/http.rs +++ b/src/adapters/mcp/http.rs @@ -1,7 +1,7 @@ use super::{ options::McpHttpOptions, protocol::{handle_mcp_message, is_supported_protocol_version, parse_mcp_payload, rpc_error}, - refresh::start_auto_refresh, + refresh::start_configured_api, state::McpHttpState, }; use serde_json::json; @@ -16,7 +16,7 @@ const MAX_HTTP_BODY_BYTES: usize = 1_000_000; pub(crate) fn serve_mcp_http(options: &McpHttpOptions) -> Result<(), String> { let listener = options.bind_listener()?; let mut options = options.clone(); - options.serve.api = Some(start_auto_refresh(&options.serve)); + options.serve.api = Some(start_configured_api(&options.serve)?); serve_mcp_http_listener(&options, listener, None) } diff --git a/src/adapters/mcp/options.rs b/src/adapters/mcp/options.rs index 5fb6bff..515aaa3 100644 --- a/src/adapters/mcp/options.rs +++ b/src/adapters/mcp/options.rs @@ -1,8 +1,40 @@ use super::http::is_local_host; use crate::adapters::required_arg; +use crate::api::context::{ + read_selected_install_config, GraphInstallConfig, GraphRefreshBackend, GraphRefreshPolicy, + DEFAULT_MAX_PARALLELISM, DEFAULT_RUST_MEMORY_MIB, DEFAULT_SPILL_CHUNK_MIB, + DEFAULT_WORKER_MEMORY_MIB, +}; use crate::api::{CodebaseGraphApi, RepoSelector}; use std::{env, net::TcpListener, path::PathBuf}; +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct McpRuntimeSettings { + pub(crate) refresh_policy: GraphRefreshPolicy, + pub(crate) refresh_backend: GraphRefreshBackend, + pub(crate) include_fts: bool, + pub(crate) semantic_enrichment: bool, + pub(crate) worker_memory_mib: u64, + pub(crate) rust_memory_mib: u64, + pub(crate) spill_chunk_mib: u64, + pub(crate) max_parallelism: usize, +} + +impl Default for McpRuntimeSettings { + fn default() -> Self { + Self { + refresh_policy: GraphRefreshPolicy::Leader, + refresh_backend: GraphRefreshBackend::Auto, + include_fts: true, + semantic_enrichment: true, + worker_memory_mib: DEFAULT_WORKER_MEMORY_MIB, + rust_memory_mib: DEFAULT_RUST_MEMORY_MIB, + spill_chunk_mib: DEFAULT_SPILL_CHUNK_MIB, + max_parallelism: DEFAULT_MAX_PARALLELISM, + } + } +} + #[derive(Clone, Debug)] pub(crate) struct McpServeOptions { pub(in crate::adapters) repo_root: Option, @@ -10,44 +42,106 @@ pub(crate) struct McpServeOptions { pub(in crate::adapters) db: Option, pub(in crate::adapters) manifest: Option, pub(in crate::adapters) api: Option, + pub(in crate::adapters) refresh_policy: Option, + pub(in crate::adapters) worker_memory_mib: Option, + pub(in crate::adapters) rust_memory_mib: Option, + pub(in crate::adapters) spill_chunk_mib: Option, + pub(in crate::adapters) max_parallelism: Option, } impl McpServeOptions { pub(crate) fn parse(args: &[String], help: &str) -> Result { - let mut options = Self { + let mut options = Self::empty(); + let mut index = 0; + while index < args.len() { + if let Some(next) = options.parse_common_option(args, index)? { + index = next; + } else { + return Err(format!( + "unknown mcp start option: {}\n\n{help}", + args[index] + )); + } + } + Ok(options) + } + + fn empty() -> Self { + Self { repo_root: None, config: None, db: None, manifest: None, api: None, - }; - let mut index = 0; - while index < args.len() { - match args[index].as_str() { - "--repo-root" => { - options.repo_root = - Some(PathBuf::from(required_arg(args, index, "--repo-root")?)); - index += 2; - } - "--config" => { - options.config = Some(PathBuf::from(required_arg(args, index, "--config")?)); - index += 2; - } - "--db" => { - options.db = Some(PathBuf::from(required_arg(args, index, "--db")?)); - index += 2; - } - "--manifest" => { - options.manifest = - Some(PathBuf::from(required_arg(args, index, "--manifest")?)); - index += 2; - } - other => { - return Err(format!("unknown mcp start option: {other}\n\n{help}")); - } - } + refresh_policy: None, + worker_memory_mib: None, + rust_memory_mib: None, + spill_chunk_mib: None, + max_parallelism: None, } - Ok(options) + } + + fn parse_common_option( + &mut self, + args: &[String], + index: usize, + ) -> Result, String> { + let next = match args[index].as_str() { + "--repo-root" => { + self.repo_root = Some(PathBuf::from(required_arg(args, index, "--repo-root")?)); + index + 2 + } + "--config" => { + self.config = Some(PathBuf::from(required_arg(args, index, "--config")?)); + index + 2 + } + "--db" => { + self.db = Some(PathBuf::from(required_arg(args, index, "--db")?)); + index + 2 + } + "--manifest" => { + self.manifest = Some(PathBuf::from(required_arg(args, index, "--manifest")?)); + index + 2 + } + "--refresh-policy" => { + self.refresh_policy = Some(parse_refresh_policy(required_arg( + args, + index, + "--refresh-policy", + )?)?); + index + 2 + } + "--worker-memory-mib" => { + self.worker_memory_mib = Some(parse_positive_u64( + required_arg(args, index, "--worker-memory-mib")?, + "--worker-memory-mib", + )?); + index + 2 + } + "--rust-memory-mib" => { + self.rust_memory_mib = Some(parse_positive_u64( + required_arg(args, index, "--rust-memory-mib")?, + "--rust-memory-mib", + )?); + index + 2 + } + "--spill-chunk-mib" => { + self.spill_chunk_mib = Some(parse_positive_u64( + required_arg(args, index, "--spill-chunk-mib")?, + "--spill-chunk-mib", + )?); + index + 2 + } + "--max-parallelism" => { + self.max_parallelism = Some(parse_positive_usize( + required_arg(args, index, "--max-parallelism")?, + "--max-parallelism", + )?); + index + 2 + } + _ => return Ok(None), + }; + Ok(Some(next)) } pub(in crate::adapters) fn repo_selector(&self) -> RepoSelector { @@ -58,6 +152,84 @@ impl McpServeOptions { manifest_path: self.manifest.clone(), } } + + pub(crate) fn runtime_settings(&self) -> Result { + let mut settings = read_selected_install_config(&self.repo_selector())? + .map(settings_from_install_config) + .unwrap_or_default(); + if let Some(value) = self.refresh_policy { + settings.refresh_policy = value; + } + if let Some(value) = self.worker_memory_mib { + settings.worker_memory_mib = value; + } + if let Some(value) = self.rust_memory_mib { + settings.rust_memory_mib = value; + } + if let Some(value) = self.spill_chunk_mib { + settings.spill_chunk_mib = value; + } + if let Some(value) = self.max_parallelism { + settings.max_parallelism = value; + } + validate_runtime_settings(settings) + } +} + +fn settings_from_install_config(config: GraphInstallConfig) -> McpRuntimeSettings { + McpRuntimeSettings { + refresh_policy: config.refresh.policy, + refresh_backend: config.refresh.backend, + include_fts: config.materialization.include_fts, + semantic_enrichment: config.materialization.semantic_enrichment, + worker_memory_mib: config.materialization.worker_memory_mib, + rust_memory_mib: config.materialization.rust_memory_mib, + spill_chunk_mib: config.materialization.spill_chunk_mib, + max_parallelism: config.materialization.max_parallelism, + } +} + +fn validate_runtime_settings(settings: McpRuntimeSettings) -> Result { + if settings.worker_memory_mib == 0 + || settings.rust_memory_mib == 0 + || settings.spill_chunk_mib == 0 + || settings.max_parallelism == 0 + { + return Err( + "MCP materialization memory limits and max parallelism must be positive".into(), + ); + } + if settings.rust_memory_mib > settings.worker_memory_mib { + return Err("rust_memory_mib must not exceed worker_memory_mib".into()); + } + if settings.spill_chunk_mib > settings.rust_memory_mib { + return Err("spill_chunk_mib must not exceed rust_memory_mib".into()); + } + Ok(settings) +} + +fn parse_refresh_policy(value: &str) -> Result { + match value { + "off" => Ok(GraphRefreshPolicy::Off), + "leader" => Ok(GraphRefreshPolicy::Leader), + _ => Err("--refresh-policy must be off or leader".to_string()), + } +} + +fn parse_positive_u64(value: &str, option: &str) -> Result { + value + .parse::() + .ok() + .filter(|value| *value > 0) + .ok_or_else(|| format!("{option} must be a positive integer")) +} + +fn parse_positive_usize(value: &str, option: &str) -> Result { + value + .parse::() + .ok() + .filter(|value| *value > 0) + .ok_or_else(|| format!("{option} must be a positive integer")) } #[derive(Clone, Debug)] @@ -73,13 +245,7 @@ pub(crate) struct McpHttpOptions { impl McpHttpOptions { pub(crate) fn parse(args: &[String], help: &str) -> Result { let mut options = Self { - serve: McpServeOptions { - repo_root: None, - config: None, - db: None, - manifest: None, - api: None, - }, + serve: McpServeOptions::empty(), host: "127.0.0.1".to_string(), port: 8765, endpoint_path: "/mcp".to_string(), @@ -88,26 +254,11 @@ impl McpHttpOptions { }; let mut index = 0; while index < args.len() { + if let Some(next) = options.serve.parse_common_option(args, index)? { + index = next; + continue; + } match args[index].as_str() { - "--repo-root" => { - options.serve.repo_root = - Some(PathBuf::from(required_arg(args, index, "--repo-root")?)); - index += 2; - } - "--config" => { - options.serve.config = - Some(PathBuf::from(required_arg(args, index, "--config")?)); - index += 2; - } - "--db" => { - options.serve.db = Some(PathBuf::from(required_arg(args, index, "--db")?)); - index += 2; - } - "--manifest" => { - options.serve.manifest = - Some(PathBuf::from(required_arg(args, index, "--manifest")?)); - index += 2; - } "--host" => { options.host = required_arg(args, index, "--host")?.to_string(); index += 2; diff --git a/src/adapters/mcp/refresh.rs b/src/adapters/mcp/refresh.rs index ab036ab..0bdb783 100644 --- a/src/adapters/mcp/refresh.rs +++ b/src/adapters/mcp/refresh.rs @@ -1,6 +1,23 @@ use super::options::McpServeOptions; -use crate::api::CodebaseGraphApi; +use crate::api::context::GraphRefreshPolicy; +use crate::api::{CodebaseGraphApi, RefreshServiceConfig}; -pub(in crate::adapters) fn start_auto_refresh(options: &McpServeOptions) -> CodebaseGraphApi { - CodebaseGraphApi::with_auto_refresh(options.repo_selector()) +pub(in crate::adapters) fn start_configured_api( + options: &McpServeOptions, +) -> Result { + let settings = options.runtime_settings()?; + match settings.refresh_policy { + GraphRefreshPolicy::Off => Ok(CodebaseGraphApi::new()), + GraphRefreshPolicy::Leader => Ok(CodebaseGraphApi::with_auto_refresh( + options.repo_selector(), + RefreshServiceConfig { + include_fts: settings.include_fts, + semantic_enrichment: settings.semantic_enrichment, + worker_memory_mib: settings.worker_memory_mib, + rust_memory_mib: settings.rust_memory_mib, + spill_chunk_mib: settings.spill_chunk_mib, + max_parallelism: settings.max_parallelism, + }, + )), + } } diff --git a/src/adapters/mcp/stdio.rs b/src/adapters/mcp/stdio.rs index 4fb64ed..a67dd2c 100644 --- a/src/adapters/mcp/stdio.rs +++ b/src/adapters/mcp/stdio.rs @@ -1,7 +1,7 @@ use super::{ options::McpServeOptions, protocol::{handle_mcp_message, parse_mcp_payload, rpc_error, McpSession}, - refresh::start_auto_refresh, + refresh::start_configured_api, }; use serde_json::json; use std::io::{BufRead, Write}; @@ -12,7 +12,7 @@ pub(crate) fn serve_mcp_stdio( output: &mut W, ) -> Result<(), String> { let mut options = options.clone(); - options.api = Some(start_auto_refresh(&options)); + options.api = Some(start_configured_api(&options)?); let mut session = McpSession::default(); while let Some(message) = read_mcp_message(&mut input, output)? { if let Some(response) = handle_mcp_message(message, &mut session, &options) { diff --git a/src/api/context.rs b/src/api/context.rs index f90d547..b074a54 100644 --- a/src/api/context.rs +++ b/src/api/context.rs @@ -97,12 +97,88 @@ impl RepoPaths { } } -#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub(crate) const INSTALL_CONFIG_SCHEMA_VERSION: u64 = 3; +pub(crate) const DEFAULT_WORKER_MEMORY_MIB: u64 = 768; +pub(crate) const DEFAULT_RUST_MEMORY_MIB: u64 = 384; +pub(crate) const DEFAULT_SPILL_CHUNK_MIB: u64 = 32; +pub(crate) const DEFAULT_MAX_PARALLELISM: usize = 2; + +const fn default_true() -> bool { + true +} + +const fn default_worker_memory_mib() -> u64 { + DEFAULT_WORKER_MEMORY_MIB +} + +const fn default_rust_memory_mib() -> u64 { + DEFAULT_RUST_MEMORY_MIB +} + +const fn default_spill_chunk_mib() -> u64 { + DEFAULT_SPILL_CHUNK_MIB +} + +const fn default_max_parallelism() -> usize { + DEFAULT_MAX_PARALLELISM +} + +#[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct GraphInstallMaterializationConfig { #[serde(default)] pub include: Vec, #[serde(default)] pub exclude: Vec, + #[serde(default = "default_true")] + pub include_fts: bool, + #[serde(default = "default_true")] + pub semantic_enrichment: bool, + #[serde(default = "default_worker_memory_mib")] + pub worker_memory_mib: u64, + #[serde(default = "default_rust_memory_mib")] + pub rust_memory_mib: u64, + #[serde(default = "default_spill_chunk_mib")] + pub spill_chunk_mib: u64, + #[serde(default = "default_max_parallelism")] + pub max_parallelism: usize, +} + +impl Default for GraphInstallMaterializationConfig { + fn default() -> Self { + Self { + include: Vec::new(), + exclude: Vec::new(), + include_fts: true, + semantic_enrichment: true, + worker_memory_mib: DEFAULT_WORKER_MEMORY_MIB, + rust_memory_mib: DEFAULT_RUST_MEMORY_MIB, + spill_chunk_mib: DEFAULT_SPILL_CHUNK_MIB, + max_parallelism: DEFAULT_MAX_PARALLELISM, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub(crate) enum GraphRefreshPolicy { + Off, + #[default] + Leader, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub(crate) enum GraphRefreshBackend { + #[default] + Auto, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub(crate) struct GraphInstallRefreshConfig { + #[serde(default)] + pub policy: GraphRefreshPolicy, + #[serde(default)] + pub backend: GraphRefreshBackend, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -136,6 +212,8 @@ pub(crate) struct GraphInstallConfig { #[serde(default)] pub materialization: GraphInstallMaterializationConfig, #[serde(default)] + pub refresh: GraphInstallRefreshConfig, + #[serde(default)] pub mcp: Option, } @@ -192,7 +270,9 @@ pub(crate) fn resolve_runtime(selector: &RepoSelector) -> Result resolve_managed_runtime(repo_root, paths, config_path, config.as_ref()), + Some(2) | Some(INSTALL_CONFIG_SCHEMA_VERSION) => { + resolve_managed_runtime(repo_root, paths, config_path, config.as_ref()) + } Some(1) => Ok(RepoRuntime { repo_root: repo_root.clone(), state_dir: paths.state_dir.clone(), @@ -367,6 +447,20 @@ pub(crate) fn read_install_config(path: &Path) -> Result Result, String> { + let repo_root = resolve_repository_root(selector.repo_root.as_deref())?; + let config_path = selector + .config_path + .clone() + .unwrap_or_else(|| RepoPaths::derive(&repo_root).config_path); + config_path + .exists() + .then(|| read_install_config(&config_path)) + .transpose() +} + fn safe_name(value: &str) -> String { let normalized: String = value .chars() @@ -388,7 +482,11 @@ fn safe_name(value: &str) -> String { #[cfg(test)] mod tests { - use super::{read_install_config, resolve_runtime, RepoPaths}; + use super::{ + read_install_config, resolve_runtime, GraphRefreshBackend, GraphRefreshPolicy, RepoPaths, + DEFAULT_MAX_PARALLELISM, DEFAULT_RUST_MEMORY_MIB, DEFAULT_SPILL_CHUNK_MIB, + DEFAULT_WORKER_MEMORY_MIB, + }; use crate::api::contracts::RepoSelector; use crate::storage::atomic::write_json_atomically; use crate::storage::direct::{DirectPublishJournal, DirectPublishPhase}; @@ -597,6 +695,26 @@ mod tests { assert_eq!(runtime.active_generation.as_deref(), Some("demo")); assert_eq!(runtime.db_path, generation.join("graph.ldb")); assert_eq!(runtime.manifest_path, generation.join("manifest.json")); + + fs::write( + state.join("config.json"), + serde_json::to_vec(&serde_json::json!({ + "schema_version": 3, + "repo_root": root, + "storage_root": storage, + })) + .unwrap(), + ) + .unwrap(); + let schema_v3_runtime = resolve_runtime(&RepoSelector { + repo_root: Some(root.clone()), + config_path: None, + db_path: None, + manifest_path: None, + }) + .expect("schema-v3 managed runtime should resolve"); + assert_eq!(schema_v3_runtime.storage_format(), "managed_v2"); + assert_eq!(schema_v3_runtime.active_generation.as_deref(), Some("demo")); let _ = fs::remove_dir_all(root); } @@ -620,6 +738,26 @@ mod tests { let mcp = config.mcp.expect("mcp config should deserialize"); assert_eq!(mcp.server_name, ""); assert!(mcp.command.is_empty()); + assert_eq!(config.refresh.policy, GraphRefreshPolicy::Leader); + assert_eq!(config.refresh.backend, GraphRefreshBackend::Auto); + assert!(config.materialization.include_fts); + assert!(config.materialization.semantic_enrichment); + assert_eq!( + config.materialization.worker_memory_mib, + DEFAULT_WORKER_MEMORY_MIB + ); + assert_eq!( + config.materialization.rust_memory_mib, + DEFAULT_RUST_MEMORY_MIB + ); + assert_eq!( + config.materialization.spill_chunk_mib, + DEFAULT_SPILL_CHUNK_MIB + ); + assert_eq!( + config.materialization.max_parallelism, + DEFAULT_MAX_PARALLELISM + ); let _ = fs::remove_dir_all(root); } } diff --git a/src/api/facade.rs b/src/api/facade.rs index 15b9a12..d320486 100644 --- a/src/api/facade.rs +++ b/src/api/facade.rs @@ -4,7 +4,7 @@ use crate::api::{ RefreshWatchConfig, RefreshWatchObserver, RepoSelector, }, core::{ApiCore, OperationDescriptor}, - refresh::{run_refresh_watch, start_refresh_service}, + refresh::{run_refresh_watch, start_refresh_service, RefreshServiceConfig}, }; pub trait OperationExecutor { @@ -37,9 +37,9 @@ impl CodebaseGraphApi { self.core.resolve_mcp_operation(tool_name) } - pub(crate) fn with_auto_refresh(selector: RepoSelector) -> Self { + pub(crate) fn with_auto_refresh(selector: RepoSelector, config: RefreshServiceConfig) -> Self { Self { - core: ApiCore::with_refresh_state(Some(start_refresh_service(selector))), + core: ApiCore::with_refresh_state(Some(start_refresh_service(selector, config))), } } diff --git a/src/api/lifecycle.rs b/src/api/lifecycle.rs index f3c2f9f..f916384 100644 --- a/src/api/lifecycle.rs +++ b/src/api/lifecycle.rs @@ -1,6 +1,6 @@ use crate::api::context::{ resolve_runtime, GraphInstallConfig, GraphInstallMaterializationConfig, GraphInstallMcpConfig, - RepoRuntime, + GraphInstallRefreshConfig, RepoRuntime, INSTALL_CONFIG_SCHEMA_VERSION, }; use crate::api::contracts::{ ApiError, McpInstallRequest, RefreshRequest, RepoSelector, RepositoryLifecycleRequest, @@ -858,7 +858,7 @@ pub(crate) fn safe_name(value: &str) -> String { fn setup_config_payload(paths: &GraphStatePaths, repo_root: &Path) -> serde_json::Value { serde_json::to_value(GraphInstallConfig { - schema_version: Some(2), + schema_version: Some(INSTALL_CONFIG_SCHEMA_VERSION), repo_root: Some(repo_root.to_path_buf()), repo_name: Some(paths.repo_name.clone()), state_dir: Some(paths.state_dir.clone()), @@ -868,6 +868,7 @@ fn setup_config_payload(paths: &GraphStatePaths, repo_root: &Path) -> serde_json ontology_version: Some("code_ontology_v1".to_string()), package_version: Some(env!("CARGO_PKG_VERSION").to_string()), materialization: GraphInstallMaterializationConfig::default(), + refresh: GraphInstallRefreshConfig::default(), mcp: Some(GraphInstallMcpConfig { server_name: "codebase_graph".to_string(), command: vec![ diff --git a/src/api/materialization.rs b/src/api/materialization.rs index 17ade5f..ecc49d0 100644 --- a/src/api/materialization.rs +++ b/src/api/materialization.rs @@ -42,6 +42,14 @@ pub(crate) struct MaterializeOptions { pub(crate) parallel: bool, pub(crate) progress: bool, pub(crate) plan_only: bool, + pub(crate) intent: MaterializationIntent, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum MaterializationIntent { + #[default] + ExplicitBuild, + Refresh, } impl Default for MaterializeOptions { @@ -66,6 +74,7 @@ impl Default for MaterializeOptions { parallel: true, progress: false, plan_only: false, + intent: MaterializationIntent::ExplicitBuild, } } } @@ -187,6 +196,18 @@ pub(crate) fn execute_materialization_request( request.previous_manifest = execution.previous_manifest().clone(); request.artifact_root = execution.artifact_root().to_string_lossy().into_owned(); request.manifest_schema_version = MATERIALIZATION_MANIFEST_SCHEMA_VERSION; + if options.intent == MaterializationIntent::Refresh { + let mut response = match crate::plan_materialization(&request) { + Ok(response) => response, + Err(error) => return Err(execution.abort_with_cleanup(error.to_string())), + }; + if refresh_plan_is_current(&response) { + response.storage_format = execution.storage_format().to_string(); + response.active_generation = execution.active_generation(); + execution.finish_without_publish()?; + return Ok((request, response)); + } + } let started = Instant::now(); let final_request = request; let mut response = match crate::execute_materialization_pipeline(&final_request) { @@ -264,6 +285,43 @@ enum StorageExecution { } impl StorageExecution { + fn storage_format(&self) -> &'static str { + match self { + Self::Direct { .. } => "direct", + Self::Managed { .. } => "managed_v2", + } + } + + fn active_generation(&self) -> Option { + match self { + Self::Direct { .. } => None, + Self::Managed { session, .. } => session + .base_generation + .as_ref() + .map(|generation| generation.generation_id.clone()), + } + } + + fn finish_without_publish(self) -> Result<(), String> { + match self { + Self::Direct { + session, workspace, .. + } => { + let cleanup_error = cleanup_direct_candidate(&session).err(); + let workspace_error = workspace.finish().err().map(|error| error.to_string()); + session.finish(); + match (cleanup_error, workspace_error) { + (None, None) => Ok(()), + (Some(primary), secondary) => Err(append_cleanup_error(primary, secondary)), + (None, Some(error)) => Err(error), + } + } + Self::Managed { session, .. } => session + .abort(None) + .map_err(|error| format!("failed to discard no-op materialization: {error}")), + } + } + fn request_db_path(&self) -> PathBuf { match self { Self::Direct { session, .. } => session.db_candidate_path(), @@ -331,6 +389,13 @@ impl StorageExecution { } } +fn refresh_plan_is_current(response: &NativeSyntaxMaterializationResponse) -> bool { + !response.diff.force_rebuild + && response.diff.added.is_empty() + && response.diff.modified.is_empty() + && response.diff.deleted.is_empty() +} + pub(crate) fn build_request( options: &MaterializeOptions, ) -> Result { diff --git a/src/api/mod.rs b/src/api/mod.rs index b56404f..22f3f66 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -31,6 +31,7 @@ pub use lifecycle::{ McpClientRenameOptions, McpExistingEntryPolicy, McpInstallMode, McpServerDescriptor, McpServerRegistration, McpTargetLocality, ResolvedMcpTarget, }; +pub(crate) use refresh::RefreshServiceConfig; #[cfg(test)] pub(crate) use refresh::{ diff --git a/src/api/refresh.rs b/src/api/refresh.rs index a8e8293..d96651f 100644 --- a/src/api/refresh.rs +++ b/src/api/refresh.rs @@ -8,11 +8,16 @@ use crate::{ lifecycle::is_retryable_refresh_failure, materialization::{ default_excluded_parts, execute_candidate_materialization, read_codebase_graph_ignore, - read_materialization_config_rules, MaterializeOptions, + read_materialization_config_rules, MaterializationIntent, MaterializeOptions, }, normalization::normalize_materialize_options, }, + profiles::ProfileSet, protocol::NativeSyntaxMaterializationResponse, + storage::{ + layout::{DirectLayout, ManagedLayout}, + locks::{try_open_locked, LockMode, RefreshLease}, + }, }; use notify::{ event::{AccessKind, AccessMode}, @@ -24,6 +29,7 @@ use std::{ env, fs, path::{Path, PathBuf}, sync::{ + atomic::{AtomicBool, Ordering}, mpsc::{self, Receiver}, Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard, }, @@ -31,14 +37,46 @@ use std::{ time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; +pub(crate) const MAX_PENDING_PATHS: usize = 4_096; +const MAX_PENDING_PATH_BYTES: usize = 1024 * 1024; +const REFRESH_ELECTION_INTERVAL: Duration = Duration::from_secs(1); + +const GENERATED_PARTS: &[&str] = &[".astro", ".kwiki", ".scryer"]; + +#[derive(Clone, Copy, Debug)] +pub(crate) struct RefreshServiceConfig { + pub(crate) include_fts: bool, + pub(crate) semantic_enrichment: bool, + pub(crate) worker_memory_mib: u64, + pub(crate) rust_memory_mib: u64, + pub(crate) spill_chunk_mib: u64, + pub(crate) max_parallelism: usize, +} + +impl Default for RefreshServiceConfig { + fn default() -> Self { + Self { + include_fts: true, + semantic_enrichment: true, + worker_memory_mib: crate::api::context::DEFAULT_WORKER_MEMORY_MIB, + rust_memory_mib: crate::api::context::DEFAULT_RUST_MEMORY_MIB, + spill_chunk_mib: crate::api::context::DEFAULT_SPILL_CHUNK_MIB, + max_parallelism: crate::api::context::DEFAULT_MAX_PARALLELISM, + } + } +} + #[derive(Debug)] pub(crate) struct WatchEventFilter { pub(crate) source_root: PathBuf, pub(crate) current_dir: PathBuf, + config_path: PathBuf, pub(crate) excluded_parts: BTreeSet, pub(crate) include_patterns: Vec, pub(crate) exclude_patterns: Vec, pub(crate) ignore_patterns: Vec, + profiles: ProfileSet, + protected_roots: Vec, } impl WatchEventFilter { @@ -52,6 +90,12 @@ impl WatchEventFilter { request.repo.config_path.clone(), request.include_patterns.clone(), request.exclude_patterns.clone(), + protected_roots( + source_root, + None, + request.repo.db_path.as_deref(), + request.repo.manifest_path.as_deref(), + ), ) } @@ -64,6 +108,12 @@ impl WatchEventFilter { options.config.clone(), options.include_patterns.clone(), options.exclude_patterns.clone(), + protected_roots( + source_root, + options.storage_root.as_deref(), + options.db.as_deref(), + options.manifest.as_deref(), + ), ) } @@ -72,6 +122,7 @@ impl WatchEventFilter { config_path: Option, mut include_patterns: Vec, mut exclude_patterns: Vec, + protected_roots: Vec, ) -> Result { let config_path = config_path.unwrap_or_else(|| config_path_for(source_root)); let config_rules = read_materialization_config_rules(&config_path)?; @@ -80,10 +131,13 @@ impl WatchEventFilter { Ok(Self { source_root: source_root.to_path_buf(), current_dir: env::current_dir().unwrap_or_else(|_| source_root.to_path_buf()), + config_path, excluded_parts: default_excluded_parts().into_iter().collect(), include_patterns, exclude_patterns, ignore_patterns: read_codebase_graph_ignore(source_root)?, + profiles: ProfileSet::new(&[]), + protected_roots, }) } @@ -98,7 +152,69 @@ impl WatchEventFilter { .collect() } + fn directory_rescan_path_count(&self, event: &Event) -> usize { + if !matches!( + event.kind, + EventKind::Remove(_) | EventKind::Modify(notify::event::ModifyKind::Name(_)) + ) { + return 0; + } + event + .paths + .iter() + .filter(|path| self.directory_change_path(path)) + .count() + } + + fn directory_change_path(&self, path: &Path) -> bool { + if self.is_configuration_path(path) || self.is_protected_path(path) { + return false; + } + let Some(relative) = self.relative_event_path(path) else { + return false; + }; + if relative.as_os_str().is_empty() + || relative.components().any(|component| { + self.excluded_parts + .contains(component.as_os_str().to_string_lossy().as_ref()) + }) + { + return false; + } + let relative = relative.to_string_lossy().replace('\\', "/"); + let explicitly_included = watch_matches_any_pattern(&relative, &self.include_patterns); + for generated_part in GENERATED_PARTS { + if let Some(index) = relative.split('/').position(|part| part == *generated_part) { + let generated_prefix = relative + .split('/') + .take(index + 1) + .collect::>() + .join("/"); + let descendant_is_included = self.include_patterns.iter().any(|pattern| { + watch_normalize_pattern(pattern).starts_with(&format!("{generated_prefix}/")) + }); + if !explicitly_included && !descendant_is_included { + return false; + } + } + } + if watch_matches_any_pattern(&relative, &self.ignore_patterns) + || watch_matches_any_pattern(&relative, &self.exclude_patterns) + { + return false; + } + path.is_dir() || path.extension().is_none() + } + pub(crate) fn relevant_path(&self, path: &Path) -> Option { + if self.is_configuration_path(path) { + return self + .relative_event_path(path) + .map(|relative| relative.to_string_lossy().replace('\\', "/")); + } + if self.is_protected_path(path) { + return None; + } let relative = self.relative_event_path(path)?; if relative.as_os_str().is_empty() { return None; @@ -110,13 +226,48 @@ impl WatchEventFilter { return None; } let relative = relative.to_string_lossy().replace('\\', "/"); - if self.ignored_by_patterns(&relative) { + let explicitly_included = watch_matches_any_pattern(&relative, &self.include_patterns); + if relative + .split('/') + .any(|part| GENERATED_PARTS.contains(&part)) + && !explicitly_included + { + return None; + } + if self.ignored_by_patterns(&relative) + || self + .profiles + .language_for_path(Path::new(&relative)) + .is_none() + { None } else { Some(relative) } } + fn is_configuration_path(&self, path: &Path) -> bool { + let absolute = if path.is_absolute() { + path.to_path_buf() + } else { + self.current_dir.join(path) + }; + absolute == self.config_path + || absolute == self.source_root.join(".codebaseGraphignore") + || path == self.config_path + } + + fn is_protected_path(&self, path: &Path) -> bool { + let absolute = if path.is_absolute() { + path.to_path_buf() + } else { + self.current_dir.join(path) + }; + self.protected_roots + .iter() + .any(|root| absolute.starts_with(root)) + } + pub(crate) fn relative_event_path(&self, path: &Path) -> Option { if let Ok(relative) = path.strip_prefix(&self.source_root) { return Some(relative.to_path_buf()); @@ -150,6 +301,26 @@ impl WatchEventFilter { } } +fn protected_roots( + source_root: &Path, + storage_root: Option<&Path>, + db_path: Option<&Path>, + manifest_path: Option<&Path>, +) -> Vec { + let mut roots = vec![source_root.join(".codebaseGraph")]; + roots.extend(storage_root.map(Path::to_path_buf)); + roots.extend(db_path.map(Path::to_path_buf)); + roots.extend(manifest_path.map(Path::to_path_buf)); + if storage_root.is_none() { + if let (Some(db_path), Some(manifest_path)) = (db_path, manifest_path) { + roots.push(DirectLayout::new(db_path, manifest_path).artifact_root_path()); + } + } + roots.sort(); + roots.dedup(); + roots +} + #[cfg(windows)] fn normalize_windows_verbatim_path(path: &Path) -> PathBuf { let normalized = path.to_string_lossy().replace('\\', "/"); @@ -188,6 +359,48 @@ pub(crate) enum WatchMessage { pub(crate) struct WatchChangeBatch { pub(crate) paths: BTreeSet, pub(crate) event_count: usize, + pub(crate) full_rescan: bool, + pub(crate) overflow_count: usize, + pub(crate) filtered_event_count: usize, + path_bytes: usize, +} + +impl WatchChangeBatch { + pub(crate) fn extend_paths(&mut self, paths: impl IntoIterator) { + if self.full_rescan { + return; + } + for path in paths { + if self.paths.contains(&path) { + continue; + } + let Some(next_bytes) = self.path_bytes.checked_add(path.len()) else { + self.mark_overflow(); + return; + }; + if self.paths.len() >= MAX_PENDING_PATHS || next_bytes > MAX_PENDING_PATH_BYTES { + self.mark_overflow(); + return; + } + self.path_bytes = next_bytes; + self.paths.insert(path); + } + } + + fn mark_overflow(&mut self) { + self.mark_full_rescan(); + self.overflow_count = self.overflow_count.saturating_add(1); + } + + fn mark_full_rescan(&mut self) { + self.paths.clear(); + self.path_bytes = 0; + self.full_rescan = true; + } + + fn has_changes(&self) -> bool { + self.full_rescan || !self.paths.is_empty() + } } #[derive(Debug, Default)] @@ -199,20 +412,31 @@ pub(crate) struct WatchProbeOutcome { pub(crate) fn start_native_watcher( source_root: &Path, -) -> Result<(notify::RecommendedWatcher, Receiver), String> { - let (tx, rx) = mpsc::channel(); +) -> Result< + ( + notify::RecommendedWatcher, + Receiver, + Arc, + ), + String, +> { + let (tx, rx) = mpsc::sync_channel(1); + let overflowed = Arc::new(AtomicBool::new(false)); + let callback_overflowed = Arc::clone(&overflowed); let mut watcher = notify::recommended_watcher(move |result: notify::Result| { let message = match result { Ok(event) => WatchMessage::Event(event), Err(error) => WatchMessage::Error(error.to_string()), }; - let _ = tx.send(message); + if tx.try_send(message).is_err() { + callback_overflowed.store(true, Ordering::Release); + } }) .map_err(|error| format!("failed to start filesystem watcher: {error}"))?; watcher .watch(source_root, RecursiveMode::Recursive) .map_err(|error| format!("failed to watch {}: {error}", source_root.display()))?; - Ok((watcher, rx)) + Ok((watcher, rx, overflowed)) } pub(crate) fn probe_native_watcher( @@ -322,10 +546,23 @@ pub(crate) fn apply_watch_message( ) -> Result<(), String> { match message { WatchMessage::Event(event) => { + let candidate_count = event.paths.len(); + let directory_rescan_count = filter.directory_rescan_path_count(&event); let paths = filter.relevant_paths(&event); - if !paths.is_empty() { + batch.filtered_event_count = batch.filtered_event_count.saturating_add( + candidate_count + .saturating_sub(paths.len()) + .saturating_sub(directory_rescan_count), + ); + let has_paths = !paths.is_empty(); + if has_paths { + batch.extend_paths(paths); + } + if directory_rescan_count > 0 { + batch.mark_full_rescan(); + } + if has_paths || directory_rescan_count > 0 { batch.event_count += 1; - batch.paths.extend(paths); } Ok(()) } @@ -336,6 +573,7 @@ pub(crate) fn apply_watch_message( pub(crate) fn collect_watch_batch( first: WatchMessage, rx: &Receiver, + overflowed: Option<&AtomicBool>, queued: &mut VecDeque, filter: &WatchEventFilter, debounce: Duration, @@ -343,7 +581,10 @@ pub(crate) fn collect_watch_batch( ) -> Result, String> { let mut batch = WatchChangeBatch::default(); apply_watch_message(first, filter, &mut batch)?; - if batch.paths.is_empty() { + if overflowed.is_some_and(|flag| flag.swap(false, Ordering::AcqRel)) { + batch.mark_overflow(); + } + if !batch.has_changes() { return Ok(None); } @@ -352,10 +593,16 @@ pub(crate) fn collect_watch_batch( loop { let elapsed = started.elapsed(); if elapsed >= max_wait { + if overflowed.is_some_and(|flag| flag.swap(false, Ordering::AcqRel)) { + batch.mark_overflow(); + } return Ok(Some(batch)); } let quiet_elapsed = last_relevant.elapsed(); if quiet_elapsed >= debounce { + if overflowed.is_some_and(|flag| flag.swap(false, Ordering::AcqRel)) { + batch.mark_overflow(); + } return Ok(Some(batch)); } let timeout = debounce @@ -374,7 +621,12 @@ pub(crate) fn collect_watch_batch( last_relevant = Instant::now(); } } - Err(mpsc::RecvTimeoutError::Timeout) => return Ok(Some(batch)), + Err(mpsc::RecvTimeoutError::Timeout) => { + if overflowed.is_some_and(|flag| flag.swap(false, Ordering::AcqRel)) { + batch.mark_overflow(); + } + return Ok(Some(batch)); + } Err(mpsc::RecvTimeoutError::Disconnected) => { return Err("filesystem watcher stopped".to_string()) } @@ -473,9 +725,14 @@ pub(crate) fn collect_poll_batch( let started = Instant::now(); let mut last_relevant = started; let mut batch = WatchChangeBatch { - paths: changed_paths, + paths: BTreeSet::new(), event_count: 1, + full_rescan: false, + overflow_count: 0, + filtered_event_count: 0, + path_bytes: 0, }; + batch.extend_paths(changed_paths); loop { let elapsed = started.elapsed(); if elapsed >= max_wait { @@ -493,7 +750,7 @@ pub(crate) fn collect_poll_batch( let changed_paths = watch_snapshot_diff(previous_snapshot, ¤t_snapshot); *previous_snapshot = current_snapshot; if !changed_paths.is_empty() { - batch.paths.extend(changed_paths); + batch.extend_paths(changed_paths); batch.event_count += 1; last_relevant = Instant::now(); } @@ -508,8 +765,19 @@ pub(crate) fn run_refresh_watch( ) -> Result<(), String> { let runtime = resolve_refresh_runtime(&request.repo)?; runtime.require_graph_write()?; + let _refresh_lease = if config.once { + None + } else { + try_open_locked(refresh_lock_path(&runtime), LockMode::Exclusive) + .map_err(|error| format!("failed to acquire refresh ownership: {error}"))? + .ok_or_else(|| { + "another process already owns repository refresh monitoring".to_string() + })? + .into() + }; let mut materialize_options = MaterializeOptions::from_request(request, &runtime, false); normalize_materialize_options(&mut materialize_options); + materialize_options.intent = MaterializationIntent::Refresh; let execution = RefreshExecutionPlan::new(request.repo.clone(), materialize_options.clone()); if config.once { @@ -523,18 +791,19 @@ pub(crate) fn run_refresh_watch( refresh_watch_batch(observer, "poll", &execution, batch) }), RefreshBackend::Native => { - let (watcher, rx) = start_native_watcher(&runtime.repo_root)?; + let (watcher, rx, overflowed) = start_native_watcher(&runtime.repo_root)?; run_native_watch( config.loop_config, &filter, watcher, rx, + overflowed, VecDeque::new(), |batch| refresh_watch_batch(observer, "native", &execution, batch), ) } RefreshBackend::Auto => match start_native_watcher(&runtime.repo_root) { - Ok((watcher, rx)) => { + Ok((watcher, rx, overflowed)) => { let probe = probe_native_watcher(&runtime.repo_root, &filter, &rx)?; if probe.delivered { run_native_watch( @@ -542,6 +811,7 @@ pub(crate) fn run_refresh_watch( &filter, watcher, rx, + overflowed, probe.queued, |batch| refresh_watch_batch(observer, "native", &execution, batch), ) @@ -583,6 +853,7 @@ fn refresh_watch_batch( &mut bound_observer, batch.event_count, &batch.paths, + batch.full_rescan, RefreshRetryPolicy::default(), |candidate_paths| execution.execute(candidate_paths), ) @@ -659,6 +930,7 @@ pub(crate) fn run_native_watch( filter: &WatchEventFilter, _watcher: notify::RecommendedWatcher, rx: Receiver, + overflowed: Arc, mut queued: VecDeque, mut refresh: impl FnMut(&WatchChangeBatch) -> Result, ) -> Result<(), String> { @@ -673,6 +945,7 @@ pub(crate) fn run_native_watch( let Some(batch) = collect_watch_batch( first, &rx, + Some(&overflowed), &mut queued, filter, config.debounce, @@ -774,15 +1047,20 @@ pub(crate) fn execute_refresh_with_policy( observer: &mut impl RefreshObserver, event_count: usize, paths: &BTreeSet, + full_rescan: bool, policy: RefreshRetryPolicy, mut refresh: impl FnMut(Vec) -> Result, ) -> Result { let changed_paths = paths.len(); - if changed_paths == 0 { + if changed_paths == 0 && !full_rescan { return Ok(true); } - let candidate_paths = paths.iter().cloned().collect::>(); + let candidate_paths = if full_rescan { + Vec::new() + } else { + paths.iter().cloned().collect::>() + }; let mut delay = policy.initial_delay; loop { observer.before_attempt(event_count, changed_paths)?; @@ -807,6 +1085,9 @@ pub(crate) fn execute_refresh_with_policy( #[derive(Clone, Debug)] pub(crate) struct RefreshStatus { pub(crate) enabled: bool, + pub(crate) role: String, + pub(crate) leader_pid: Option, + pub(crate) worker_pid: Option, pub(crate) backend: String, pub(crate) refreshing: bool, pub(crate) pending: bool, @@ -819,12 +1100,26 @@ pub(crate) struct RefreshStatus { pub(crate) last_rebuilt: usize, pub(crate) last_deleted: usize, pub(crate) last_database_written: bool, + pub(crate) coalesced_event_count: usize, + pub(crate) filtered_event_count: usize, + pub(crate) overflow_count: usize, + pub(crate) deduplicated_refresh_count: usize, + pub(crate) last_noop_reason: Option, + pub(crate) worker_memory_mib: u64, + pub(crate) rust_memory_mib: u64, + pub(crate) spill_chunk_mib: u64, + pub(crate) max_parallelism: usize, + pub(crate) phase_high_water_marks: BTreeMap, + pub(crate) spill_bytes: u64, } impl Default for RefreshStatus { fn default() -> Self { Self { enabled: true, + role: "starting".to_string(), + leader_pid: None, + worker_pid: None, backend: "starting".to_string(), refreshing: false, pending: false, @@ -837,6 +1132,17 @@ impl Default for RefreshStatus { last_rebuilt: 0, last_deleted: 0, last_database_written: false, + coalesced_event_count: 0, + filtered_event_count: 0, + overflow_count: 0, + deduplicated_refresh_count: 0, + last_noop_reason: None, + worker_memory_mib: crate::api::context::DEFAULT_WORKER_MEMORY_MIB, + rust_memory_mib: crate::api::context::DEFAULT_RUST_MEMORY_MIB, + spill_chunk_mib: crate::api::context::DEFAULT_SPILL_CHUNK_MIB, + max_parallelism: crate::api::context::DEFAULT_MAX_PARALLELISM, + phase_high_water_marks: BTreeMap::new(), + spill_bytes: 0, } } } @@ -847,10 +1153,27 @@ pub(crate) struct RefreshState { graph_lock: RwLock<()>, } +struct RefreshCompletion { + event_count: usize, + changed_paths: usize, + rebuilt: usize, + deleted: usize, + database_written: bool, + overflow_count: usize, + filtered_event_count: usize, +} + impl RefreshState { - pub(crate) fn new() -> Self { + pub(crate) fn with_config(config: RefreshServiceConfig) -> Self { + let status = RefreshStatus { + worker_memory_mib: config.worker_memory_mib, + rust_memory_mib: config.rust_memory_mib, + spill_chunk_mib: config.spill_chunk_mib, + max_parallelism: config.max_parallelism, + ..RefreshStatus::default() + }; Self { - status: Mutex::new(RefreshStatus::default()), + status: Mutex::new(status), graph_lock: RwLock::new(()), } } @@ -861,6 +1184,9 @@ impl RefreshState { .map(|status| status.clone()) .unwrap_or_else(|_| RefreshStatus { enabled: false, + role: "failed".to_string(), + leader_pid: None, + worker_pid: None, backend: "failed".to_string(), refreshing: false, pending: false, @@ -873,6 +1199,17 @@ impl RefreshState { last_rebuilt: 0, last_deleted: 0, last_database_written: false, + coalesced_event_count: 0, + filtered_event_count: 0, + overflow_count: 0, + deduplicated_refresh_count: 0, + last_noop_reason: None, + worker_memory_mib: 0, + rust_memory_mib: 0, + spill_chunk_mib: 0, + max_parallelism: 0, + phase_high_water_marks: BTreeMap::new(), + spill_bytes: 0, }) } @@ -880,6 +1217,9 @@ impl RefreshState { let status = self.snapshot(); json!({ "enabled": status.enabled, + "role": status.role, + "leader_pid": status.leader_pid, + "worker_pid": status.worker_pid, "backend": status.backend, "refreshing": status.refreshing, "pending": status.pending, @@ -892,9 +1232,40 @@ impl RefreshState { "last_rebuilt": status.last_rebuilt, "last_deleted": status.last_deleted, "last_database_written": status.last_database_written, + "coalesced_event_count": status.coalesced_event_count, + "filtered_event_count": status.filtered_event_count, + "overflow_count": status.overflow_count, + "deduplicated_refresh_count": status.deduplicated_refresh_count, + "last_noop_reason": status.last_noop_reason, + "memory_limits": { + "worker_memory_mib": status.worker_memory_mib, + "rust_memory_mib": status.rust_memory_mib, + "spill_chunk_mib": status.spill_chunk_mib, + "max_parallelism": status.max_parallelism, + }, + "phase_high_water_marks": status.phase_high_water_marks, + "spill_bytes": status.spill_bytes, }) } + pub(crate) fn mark_leader(&self) { + if let Ok(mut status) = self.status.lock() { + status.role = "leader".to_string(); + status.leader_pid = Some(std::process::id()); + status.enabled = true; + } + } + + pub(crate) fn mark_standby(&self) { + if let Ok(mut status) = self.status.lock() { + status.role = "standby".to_string(); + status.leader_pid = None; + status.backend = "standby".to_string(); + status.refreshing = false; + status.enabled = true; + } + } + pub(crate) fn read_guard(&self) -> Result, String> { self.graph_lock .read() @@ -930,6 +1301,8 @@ impl RefreshState { if let Ok(mut status) = self.status.lock() { status.backend = backend.to_string(); status.enabled = false; + status.role = "disabled".to_string(); + status.leader_pid = None; status.refreshing = false; status.pending = false; status.last_error = Some(error); @@ -972,15 +1345,7 @@ impl RefreshState { } } - pub(crate) fn mark_refreshed( - &self, - backend: &str, - event_count: usize, - changed_paths: usize, - rebuilt: usize, - deleted: usize, - database_written: bool, - ) { + fn mark_refreshed(&self, backend: &str, completion: RefreshCompletion) { if let Ok(mut status) = self.status.lock() { status.backend = backend.to_string(); status.refreshing = false; @@ -989,20 +1354,39 @@ impl RefreshState { 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; - status.last_deleted = deleted; - status.last_database_written = database_written; + status.last_event_count = completion.event_count; + status.last_changed_paths = completion.changed_paths; + status.last_rebuilt = completion.rebuilt; + status.last_deleted = completion.deleted; + status.last_database_written = completion.database_written; + status.coalesced_event_count = status + .coalesced_event_count + .saturating_add(completion.event_count.saturating_sub(1)); + status.overflow_count = status + .overflow_count + .saturating_add(completion.overflow_count); + status.filtered_event_count = status + .filtered_event_count + .saturating_add(completion.filtered_event_count); + if completion.database_written { + status.last_noop_reason = None; + } else { + status.deduplicated_refresh_count = + status.deduplicated_refresh_count.saturating_add(1); + status.last_noop_reason = Some("active_generation_current".to_string()); + } } } } -pub(crate) fn start_refresh_service(selector: RepoSelector) -> Arc { - let state = Arc::new(RefreshState::new()); +pub(crate) fn start_refresh_service( + selector: RepoSelector, + config: RefreshServiceConfig, +) -> Arc { + let state = Arc::new(RefreshState::with_config(config)); let thread_state = Arc::clone(&state); thread::spawn(move || { - if let Err(error) = run_refresh_service(selector, &thread_state) { + if let Err(error) = run_refresh_service(selector, config, &thread_state) { thread_state.set_error("failed", error.clone()); eprintln!( "{}", @@ -1013,12 +1397,55 @@ pub(crate) fn start_refresh_service(selector: RepoSelector) -> Arc state } -fn run_refresh_service(selector: RepoSelector, state: &Arc) -> Result<(), String> { +fn run_refresh_service( + selector: RepoSelector, + config: RefreshServiceConfig, + state: &Arc, +) -> Result<(), String> { let runtime = resolve_refresh_runtime(&selector)?; if let Err(error) = runtime.require_graph_write() { state.disable("disabled", error); return Ok(()); } + let lock_path = refresh_lock_path(&runtime); + loop { + match try_open_locked(&lock_path, LockMode::Exclusive).map_err(|error| { + format!( + "failed to acquire refresh ownership {}: {error}", + lock_path.display() + ) + })? { + Some(lease) => { + state.mark_leader(); + return run_refresh_leader(selector, state, runtime, lease, config); + } + None => { + state.mark_standby(); + thread::sleep(refresh_election_delay()); + } + } + } +} + +fn refresh_lock_path(runtime: &crate::api::context::RepoRuntime) -> PathBuf { + match runtime.storage_root.as_ref() { + Some(storage_root) => ManagedLayout::new(storage_root).refresh_lock_path(), + None => DirectLayout::new(&runtime.db_path, &runtime.manifest_path).refresh_lock_path(), + } +} + +fn refresh_election_delay() -> Duration { + REFRESH_ELECTION_INTERVAL + .saturating_add(Duration::from_millis(u64::from(std::process::id() % 251))) +} + +fn run_refresh_leader( + selector: RepoSelector, + state: &Arc, + runtime: crate::api::context::RepoRuntime, + _lease: RefreshLease, + config: RefreshServiceConfig, +) -> Result<(), String> { let materialize_options = MaterializeOptions { source_root: Some(runtime.repo_root.clone()), config: runtime.config_path.clone(), @@ -1026,10 +1453,11 @@ fn run_refresh_service(selector: RepoSelector, state: &Arc) -> Res manifest: Some(runtime.manifest_path.clone()), storage_root: runtime.storage_root.clone(), mode: "changed".to_string(), - include_fts: true, - semantic_enrichment: true, + include_fts: config.include_fts, + semantic_enrichment: config.semantic_enrichment, semantic_provider_mode: "local_only".to_string(), use_git: false, + intent: MaterializationIntent::Refresh, ..MaterializeOptions::default() }; let execution = RefreshExecutionPlan::new(selector, materialize_options.clone()); @@ -1041,19 +1469,27 @@ fn run_refresh_service(selector: RepoSelector, state: &Arc) -> Res max_iterations: None, }; + let startup_batch = WatchChangeBatch { + full_rescan: true, + ..WatchChangeBatch::default() + }; + if !refresh_batch_with_state(state, "startup", &execution, &startup_batch)? { + return Err("startup repository reconciliation failed".to_string()); + } + match start_native_watcher(&runtime.repo_root) { - Ok((watcher, rx)) => { + Ok((watcher, rx, overflowed)) => { let probe = probe_native_watcher(&runtime.repo_root, &filter, &rx)?; if probe.delivered { state.set_backend("native"); - match run_service_native_loop( - state, + match run_native_watch( loop_config, - &execution, &filter, watcher, rx, + overflowed, probe.queued, + |batch| refresh_batch_with_state(state, "native", &execution, batch), ) { Ok(()) => Ok(()), Err(error) => { @@ -1083,20 +1519,6 @@ fn run_refresh_service(selector: RepoSelector, state: &Arc) -> Res } } -fn run_service_native_loop( - state: &Arc, - config: RefreshLoopConfig, - execution: &RefreshExecutionPlan, - filter: &WatchEventFilter, - watcher: notify::RecommendedWatcher, - rx: Receiver, - queued: VecDeque, -) -> Result<(), String> { - run_native_watch(config, filter, watcher, rx, queued, |batch| { - refresh_batch_with_state(state, "native", execution, batch.event_count, &batch.paths) - }) -} - fn run_service_poll_loop( state: &Arc, config: RefreshLoopConfig, @@ -1105,7 +1527,7 @@ fn run_service_poll_loop( ) -> Result<(), String> { state.set_backend("poll"); run_poll_watch(config, filter, |batch| { - refresh_batch_with_state(state, "poll", execution, batch.event_count, &batch.paths) + refresh_batch_with_state(state, "poll", execution, batch) }) } @@ -1113,14 +1535,19 @@ fn refresh_batch_with_state( state: &Arc, backend: &str, execution: &RefreshExecutionPlan, - event_count: usize, - paths: &BTreeSet, + batch: &WatchChangeBatch, ) -> Result { - let mut observer = StateRefreshObserver::new(state, backend); + let mut observer = StateRefreshObserver::new( + state, + backend, + batch.overflow_count, + batch.filtered_event_count, + ); execute_refresh_with_policy( &mut observer, - event_count, - paths, + batch.event_count, + &batch.paths, + batch.full_rescan, RefreshRetryPolicy::default(), |candidate_paths| execution.execute(candidate_paths), ) @@ -1130,14 +1557,23 @@ struct StateRefreshObserver<'a> { state: &'a Arc, backend: &'a str, guard: Option>, + overflow_count: usize, + filtered_event_count: usize, } impl<'a> StateRefreshObserver<'a> { - fn new(state: &'a Arc, backend: &'a str) -> Self { + fn new( + state: &'a Arc, + backend: &'a str, + overflow_count: usize, + filtered_event_count: usize, + ) -> Self { Self { state, backend, guard: None, + overflow_count, + filtered_event_count, } } } @@ -1159,11 +1595,15 @@ impl RefreshObserver for StateRefreshObserver<'_> { self.guard.take(); self.state.mark_refreshed( self.backend, - event_count, - changed_paths, - response.diff.rebuild_paths().len(), - response.diff.deleted.len(), - response.database_written, + RefreshCompletion { + event_count, + changed_paths, + rebuilt: response.diff.rebuild_paths().len(), + deleted: response.diff.deleted.len(), + database_written: response.database_written, + overflow_count: self.overflow_count, + filtered_event_count: self.filtered_event_count, + }, ); Ok(()) } @@ -1292,6 +1732,47 @@ mod tests { ) } + #[test] + fn refresh_status_reports_configured_limits_and_worker_placeholders() { + let state = RefreshState::with_config(RefreshServiceConfig { + include_fts: false, + semantic_enrichment: false, + worker_memory_mib: 640, + rust_memory_mib: 320, + spill_chunk_mib: 16, + max_parallelism: 1, + }); + + let status = state.as_json(); + assert_eq!(status["worker_pid"], serde_json::Value::Null); + assert_eq!(status["memory_limits"]["worker_memory_mib"], 640); + assert_eq!(status["memory_limits"]["rust_memory_mib"], 320); + assert_eq!(status["memory_limits"]["spill_chunk_mib"], 16); + assert_eq!(status["memory_limits"]["max_parallelism"], 1); + assert_eq!(status["phase_high_water_marks"], serde_json::json!({})); + assert_eq!(status["spill_bytes"], 0); + } + + #[test] + fn watch_filter_never_admits_configured_storage_root() { + let root = unique_temp_dir("codebase-graph-rust-watch-filter-storage"); + let storage_root = root.join("graph-storage"); + fs::create_dir_all(&storage_root).unwrap(); + let options = MaterializeOptions { + source_root: Some(root.clone()), + storage_root: Some(storage_root.clone()), + include_patterns: vec!["graph-storage/*".to_string()], + ..MaterializeOptions::default() + }; + let filter = WatchEventFilter::from_options(&root, &options).unwrap(); + + assert_eq!( + filter.relevant_path(&storage_root.join("generated.rs")), + None + ); + let _ = fs::remove_dir_all(root); + } + struct RecordingObserver { retries: Vec<(bool, String, usize, usize)>, successes: Vec<(usize, usize, usize)>, @@ -1342,6 +1823,7 @@ mod tests { &mut observer, 2, &BTreeSet::from(["src/lib.rs".to_string()]), + false, RefreshRetryPolicy { initial_delay: Duration::from_millis(0), max_delay: Duration::from_millis(0), @@ -1370,6 +1852,7 @@ mod tests { &mut observer, 1, &BTreeSet::from(["src/lib.rs".to_string()]), + false, RefreshRetryPolicy { initial_delay: Duration::from_millis(0), max_delay: Duration::from_millis(0), @@ -1507,12 +1990,15 @@ mod tests { ) .unwrap(); - let state = start_refresh_service(RepoSelector { - repo_root: Some(root.clone()), - config_path: None, - db_path: None, - manifest_path: None, - }); + let state = start_refresh_service( + RepoSelector { + repo_root: Some(root.clone()), + config_path: None, + db_path: None, + manifest_path: None, + }, + RefreshServiceConfig::default(), + ); let mut snapshot = state.snapshot(); for _ in 0..50 { diff --git a/src/profiles.rs b/src/profiles.rs index 7ffbc2a..2d4653c 100644 --- a/src/profiles.rs +++ b/src/profiles.rs @@ -1,6 +1,7 @@ use crate::protocol::{CaptureMapping, LanguageProfile}; use std::collections::BTreeMap; +#[derive(Debug)] pub(crate) struct ProfileSet { by_language: BTreeMap, suffix_to_language: BTreeMap, diff --git a/src/storage/layout.rs b/src/storage/layout.rs index 105ecf7..ee28465 100644 --- a/src/storage/layout.rs +++ b/src/storage/layout.rs @@ -53,6 +53,10 @@ impl ManagedLayout { self.storage_root.join("state.lock") } + pub(crate) fn refresh_lock_path(&self) -> PathBuf { + self.storage_root.join("refresh.lock") + } + pub(crate) fn active_pointer_path(&self) -> PathBuf { self.storage_root.join("active.json") } @@ -214,8 +218,11 @@ impl DirectLayout { } pub(crate) fn writer_lock_path(&self) -> PathBuf { - let parent = self.db_path.parent().unwrap_or_else(|| Path::new(".")); - parent.join(format!(".direct-publish-{}.lock", self.destination_key())) + self.destination_lock_path("publish") + } + + pub(crate) fn refresh_lock_path(&self) -> PathBuf { + self.destination_lock_path("refresh") } pub(crate) fn artifact_root_path(&self) -> PathBuf { @@ -245,6 +252,11 @@ impl DirectLayout { .collect(); hex[..16].to_string() } + + fn destination_lock_path(&self, role: &str) -> PathBuf { + let parent = self.db_path.parent().unwrap_or_else(|| Path::new(".")); + parent.join(format!(".direct-{role}-{}.lock", self.destination_key())) + } } fn sibling_candidate_path(path: &Path) -> PathBuf { diff --git a/src/storage/locks.rs b/src/storage/locks.rs index 43acb28..c4177ee 100644 --- a/src/storage/locks.rs +++ b/src/storage/locks.rs @@ -41,6 +41,7 @@ impl Drop for LockedFile { pub(crate) type WriterLease = LockedFile; pub(crate) type StateLease = LockedFile; pub(crate) type RunLease = LockedFile; +pub(crate) type RefreshLease = LockedFile; pub(crate) fn open_locked( path: impl AsRef, diff --git a/src/storage/test_harness.rs b/src/storage/test_harness.rs index 25725bb..576c591 100644 --- a/src/storage/test_harness.rs +++ b/src/storage/test_harness.rs @@ -24,12 +24,74 @@ use storage::atomic::{ }; use storage::direct::{DirectPublishJournal, DirectPublishPhase, DirectStore}; use storage::layout::{managed_generation_id, DirectLayout, GenerationPaths, ManagedLayout}; -use storage::locks::{try_open_locked, LockMode}; +use storage::locks::{try_open_locked, LockMode, RefreshLease}; use storage::managed::{ActiveGeneration, ManagedStore, ManagedWriteSession}; use storage::run_workspace::{RunJournal, RunPhase, RunWorkspace}; static TEST_SEQUENCE: AtomicU64 = AtomicU64::new(0); +#[test] +fn managed_refresh_lock_path_is_stable_and_separate_from_publication() { + let root = temp_dir("managed_refresh_lock_path"); + let layout = ManagedLayout::new(root.join("storage")); + + assert_eq!( + layout.refresh_lock_path(), + root.join("storage/refresh.lock") + ); + assert_ne!(layout.refresh_lock_path(), layout.writer_lock_path()); +} + +#[test] +fn direct_refresh_lock_paths_are_stable_and_destination_scoped() { + let root = temp_dir("direct_refresh_lock_paths"); + let first = DirectLayout::new(root.join("graph.ldb"), root.join("manifest.json")); + let same = DirectLayout::new(root.join("graph.ldb"), root.join("manifest.json")); + let other = DirectLayout::new(root.join("other.ldb"), root.join("manifest.json")); + + assert_eq!(first.refresh_lock_path(), same.refresh_lock_path()); + assert_eq!(first.refresh_lock_path().parent(), Some(root.as_path())); + assert_ne!(first.refresh_lock_path(), first.writer_lock_path()); + assert_ne!(first.refresh_lock_path(), other.refresh_lock_path()); +} + +#[test] +fn refresh_lease_is_exclusive_nonblocking_and_transfers_after_release() { + let root = temp_dir("refresh_lease_takeover"); + let layout = ManagedLayout::new(root.join("storage")); + + let refresh: RefreshLease = try_open_locked(layout.refresh_lock_path(), LockMode::Exclusive) + .unwrap() + .expect("first refresh owner should acquire the lease"); + assert!( + try_open_locked(layout.refresh_lock_path(), LockMode::Exclusive) + .unwrap() + .is_none() + ); + drop(refresh); + assert!( + try_open_locked(layout.refresh_lock_path(), LockMode::Exclusive) + .unwrap() + .is_some() + ); +} + +#[cfg(unix)] +#[test] +fn refresh_lease_rejects_a_symlinked_lock_file() { + use std::os::unix::fs::symlink; + + let root = temp_dir("refresh_lease_symlink"); + let layout = ManagedLayout::new(root.join("storage")); + fs::create_dir_all(layout.storage_root()).unwrap(); + let target = root.join("outside.lock"); + fs::write(&target, b"").unwrap(); + symlink(&target, layout.refresh_lock_path()).unwrap(); + + let error = try_open_locked(layout.refresh_lock_path(), LockMode::Exclusive).unwrap_err(); + assert!(error.to_string().contains("lock path must be a real file")); +} + #[test] fn atomic_write_failure_preserves_existing_file_and_cleans_temp() { let root = temp_dir("atomic_write_failure_preserves_existing_file_and_cleans_temp");