From 1618ad064513f3ef2168253b28bd8caa6a00e794 Mon Sep 17 00:00:00 2001 From: Matthew Klahn Date: Mon, 20 Jul 2026 13:00:32 -0400 Subject: [PATCH] perf: make repository monitoring non-interfering Use lock-free passive Git reads and debounced filesystem events to eliminate index contention and refresh storms. Supervise child processes and join monitoring workers so normal shutdown leaves no orphaned repository work. --- changelog.md | 4 ++ memory.md | 4 ++ src/adapters/command.rs | 108 +++++++++++++++++++++++++++++++--- src/adapters/git.rs | 28 ++++++--- src/main.rs | 11 ++-- src/refresh/mod.rs | 12 ++-- src/refresh/upstream.rs | 6 +- src/refresh/watcher.rs | 126 ++++++++++++++++++++++++++++++++++++++-- 8 files changed, 263 insertions(+), 36 deletions(-) diff --git a/changelog.md b/changelog.md index 2d13cbf..f6e1b08 100644 --- a/changelog.md +++ b/changelog.md @@ -2,6 +2,10 @@ ## 2026-07-20 +- Made monitoring non-interfering and event-driven: passive Git commands disable optional locks, relevant filesystem events debounce after a quiet period, noisy `.git` paths are ignored, and periodic reconciliation moved from 30 seconds to five minutes. +- Separated passive and mutating Git execution so checkout and atomic deletion retain normal locks with a safer 30-second deadline, while timed-out commands receive a graceful TERM window before forced termination. +- Hardened shutdown by stopping and joining watcher/refresh/upstream workers and terminating every registered subprocess group instead of detaching active repository work. +- Verified formatting, strict offline Clippy, 176 all-target/all-feature tests, release build, and a live FactMachine-monorepo smoke test with 0.0% settled CPU, no index lock, and no process left after quit. - Published the hardened source repository at `https://github.com/fieldsofland/stackmap`, protected `main` with required ARM64/Intel/stable/policy checks, protected prerelease tags, enabled security reporting and dependency updates, and configured the maintainer-approved prerelease environment. - Installed `stackmap 0.1.0-alpha.1` from verified public `main` at `/Users/matt/.cargo/bin/stackmap`; version/help output and disposable-repository startup/quit smoke tests pass. - Prepared Stackmap `0.1.0-alpha.1` for public open-source development with an MIT license, contribution and conduct policies, private vulnerability reporting guidance, issue forms, pull-request guidance, and feature/support/release documentation. diff --git a/memory.md b/memory.md index 09ec3b2..5155e73 100644 --- a/memory.md +++ b/memory.md @@ -38,6 +38,10 @@ - Release benchmarks measured about 0.108 ms per 500-branch projection, 1.10 ms per 5,000-branch projection, and 0.94 ms per projection of one 5,000-branch deep stack. A 5,000-level/10,000-branch deep comb emits iteratively without call-stack recursion; broad attach-parent lookup is indexed. - Public development lives at `https://github.com/fieldsofland/stackmap` on protected `main`. Native macOS ARM64/Intel CI, current-stable compatibility, and dependency policy pass at commit `0f01e4346448c700c0474061734005f08e767405`. - `stackmap 0.1.0-alpha.1` is installed at `/Users/matt/.cargo/bin/stackmap`, resolves on `PATH`, and passes startup/quit smoke testing in a disposable Git repository. +- Passive Git monitoring runs with optional locks disabled, so inventory/status/diff commands cannot refresh or lock the index. Explicit checkout and deletion retain normal Git locking and use a separate 30-second mutation deadline. +- Repository notifications are path-filtered and quiet-period debounced; transient locks, object-store writes, logs, and temporary files do not trigger structural refreshes. Five-minute reconciliation is a dropped-event safety net rather than the primary monitor. +- Normal quit, terminal-close signals, and recoverable error exits stop the watcher, terminate registered subprocess groups, and join refresh/upstream workers. Timed-out children receive TERM before KILL. +- A release-build smoke test against the FactMachine monorepo settled at 0.0% CPU, created no `index.lock`, and left no Stackmap or Git child after `q`. ## Next steps diff --git a/src/adapters/command.rs b/src/adapters/command.rs index 9b098a5..da732eb 100644 --- a/src/adapters/command.rs +++ b/src/adapters/command.rs @@ -3,6 +3,7 @@ use std::io::{Read, Write}; use std::path::Path; use std::process::{Command, ExitStatus, Stdio}; use std::sync::mpsc; +use std::sync::{Mutex, OnceLock}; use std::thread; use std::time::{Duration, Instant}; @@ -44,7 +45,28 @@ where I: IntoIterator, S: AsRef, { - run_bounded_inner(program, args, cwd, timeout, output_limit, None) + run_bounded_inner(program, args, cwd, timeout, output_limit, None, false) +} + +pub fn run_bounded_read_only_git( + args: I, + cwd: &Path, + timeout: Duration, + output_limit: usize, +) -> Result +where + I: IntoIterator, + S: AsRef, +{ + run_bounded_inner( + OsStr::new("git"), + args, + cwd, + timeout, + output_limit, + None, + true, + ) } pub fn run_bounded_with_stdin( @@ -62,7 +84,15 @@ where if stdin.len() > 64 * 1024 { return Err(CommandError::InputTooLarge(stdin.len())); } - run_bounded_inner(program, args, cwd, timeout, output_limit, Some(stdin)) + run_bounded_inner( + program, + args, + cwd, + timeout, + output_limit, + Some(stdin), + false, + ) } fn run_bounded_inner( @@ -72,6 +102,7 @@ fn run_bounded_inner( timeout: Duration, output_limit: usize, stdin: Option<&[u8]>, + disable_optional_locks: bool, ) -> Result where I: IntoIterator, @@ -89,12 +120,16 @@ where }) .stdout(Stdio::piped()) .stderr(Stdio::piped()); + if disable_optional_locks { + command.env("GIT_OPTIONAL_LOCKS", "0"); + } #[cfg(unix)] { use std::os::unix::process::CommandExt; command.process_group(0); } let mut child = command.spawn().map_err(CommandError::Spawn)?; + let active_child = ActiveChild::register(child.id()); let stdin_result = if let Some(input) = stdin { let mut stream = child @@ -144,6 +179,7 @@ where None => thread::sleep(Duration::from_millis(10)), } }; + drop(active_child); let mut stdout = (Vec::new(), false); let mut stderr = (Vec::new(), false); @@ -205,17 +241,71 @@ fn terminate(child: &mut std::process::Child) { // Each subprocess owns its process group, so a timeout also stops hooks // or helpers that inherited its pipes instead of leaving reader threads // waiting for descendants after the direct child exits. - let group = format!("-{}", child.id()); - let _ = Command::new("/bin/kill") - .args(["-KILL", &group]) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .status(); + signal_process_group(child.id(), "-TERM"); + let grace_started = Instant::now(); + while grace_started.elapsed() < Duration::from_millis(250) { + if child.try_wait().ok().flatten().is_some() { + return; + } + thread::sleep(Duration::from_millis(10)); + } + signal_process_group(child.id(), "-KILL"); } let _ = child.kill(); } +fn active_children() -> &'static Mutex> { + static ACTIVE: OnceLock>> = OnceLock::new(); + ACTIVE.get_or_init(|| Mutex::new(std::collections::HashSet::new())) +} + +struct ActiveChild(u32); + +impl ActiveChild { + fn register(pid: u32) -> Self { + if let Ok(mut active) = active_children().lock() { + active.insert(pid); + } + Self(pid) + } +} + +impl Drop for ActiveChild { + fn drop(&mut self) { + if let Ok(mut active) = active_children().lock() { + active.remove(&self.0); + } + } +} + +pub fn terminate_active_commands() { + let pids = active_children() + .lock() + .map(|active| active.iter().copied().collect::>()) + .unwrap_or_default(); + for &pid in &pids { + signal_process_group(pid, "-TERM"); + } + thread::sleep(Duration::from_millis(100)); + for pid in pids { + signal_process_group(pid, "-KILL"); + } +} + +#[cfg(unix)] +fn signal_process_group(pid: u32, signal: &str) { + let group = format!("-{pid}"); + let _ = Command::new("/bin/kill") + .args([signal, &group]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); +} + +#[cfg(not(unix))] +fn signal_process_group(_pid: u32, _signal: &str) {} + fn read_limited( mut stream: Box, limit: usize, diff --git a/src/adapters/git.rs b/src/adapters/git.rs index 5c76f97..047ca67 100644 --- a/src/adapters/git.rs +++ b/src/adapters/git.rs @@ -8,13 +8,14 @@ use anyhow::{Context, Result, anyhow, bail}; #[cfg(test)] use super::command::CommandError; -use super::command::{CommandOutput, run_bounded}; +use super::command::{CommandOutput, run_bounded, run_bounded_read_only_git}; #[cfg(test)] use super::graphite::raw_branch_metadata_presence; use super::graphite::{raw_branch_metadata_has_child, read_topology}; use crate::model::{BranchId, ConfiguredUpstream, DiffStat, GraphiteProvenance, RepositoryState}; const GIT_TIMEOUT: Duration = Duration::from_secs(3); +const GIT_MUTATION_TIMEOUT: Duration = Duration::from_secs(30); const UPSTREAM_GIT_TIMEOUT: Duration = Duration::from_millis(250); const OUTPUT_LIMIT: usize = 16 * 1024 * 1024; const UPSTREAM_OUTPUT_LIMIT: usize = 256 * 1024; @@ -166,8 +167,7 @@ impl GitAdapter { pub fn containing_remote_ref(&self, oid: &str) -> Result>> { let contains = format!("--contains={oid}"); - let output = run_bounded( - OsStr::new("git"), + let output = run_bounded_read_only_git( [ OsStr::new("for-each-ref"), OsStr::new("--sort=refname"), @@ -270,7 +270,7 @@ impl GitAdapter { { bail!("branch {branch} is checked out at {}", path.display()); } - let output = self.git_os( + let output = self.git_mutation( &[ OsStr::new("switch"), OsStr::new("--"), @@ -394,7 +394,7 @@ impl GitAdapter { ); } let reference = format!("refs/heads/{}", request.branch); - let output = self.git_os( + let output = self.git_mutation( &[ OsStr::new("update-ref"), OsStr::new("-d"), @@ -560,8 +560,7 @@ impl GitAdapter { } fn git_upstream(&self, args: &[&str]) -> Result { - run_bounded( - OsStr::new("git"), + run_bounded_read_only_git( args.iter().map(OsStr::new), &self.start_dir, UPSTREAM_GIT_TIMEOUT, @@ -571,9 +570,20 @@ impl GitAdapter { } fn git_os(&self, args: &[&OsStr], limit: usize) -> Result { - run_bounded(OsStr::new("git"), args, &self.start_dir, GIT_TIMEOUT, limit) + run_bounded_read_only_git(args, &self.start_dir, GIT_TIMEOUT, limit) .map_err(|error| anyhow!(error)) } + + fn git_mutation(&self, args: &[&OsStr], limit: usize) -> Result { + run_bounded( + OsStr::new("git"), + args, + &self.start_dir, + GIT_MUTATION_TIMEOUT, + limit, + ) + .map_err(|error| anyhow!(error)) + } } fn ensure_success(output: &CommandOutput, action: &str) -> Result<()> { @@ -589,7 +599,7 @@ fn ensure_success(output: &CommandOutput, action: &str) -> Result<()> { fn discover_paths(start_dir: &Path) -> Result<(PathBuf, PathBuf, PathBuf)> { let run = |args: &[&str], action| -> Result { - let output = run_bounded(OsStr::new("git"), args, start_dir, GIT_TIMEOUT, 64 * 1024) + let output = run_bounded_read_only_git(args, start_dir, GIT_TIMEOUT, 64 * 1024) .map_err(|error| anyhow!(error))?; ensure_success(&output, action)?; Ok(output) diff --git a/src/main.rs b/src/main.rs index 121638e..68afecb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -26,7 +26,7 @@ use stackmap::runtime::{ RefreshHandle, github, platform, render, }; -const RECONCILE_INTERVAL: Duration = Duration::from_secs(30); +const RECONCILE_INTERVAL: Duration = Duration::from_secs(300); const GITHUB_TTL: Duration = Duration::from_secs(30); const CONFIG_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); @@ -205,7 +205,13 @@ fn run() -> Result<()> { let mut redraw = true; loop { + if shutdown.load(Ordering::Acquire) { + break; + } while let Some(event) = refresh.try_event() { + if shutdown.load(Ordering::Acquire) { + break; + } redraw = true; match event { RefreshEvent::Structural { @@ -304,9 +310,6 @@ fn run() -> Result<()> { refresh.request(); last_reconcile = Instant::now(); } - if shutdown.load(Ordering::Acquire) { - break; - } if event::poll(Duration::from_millis(100))? { match event::read()? { Event::Key(key) => { diff --git a/src/refresh/mod.rs b/src/refresh/mod.rs index d6c6ce1..10901a6 100644 --- a/src/refresh/mod.rs +++ b/src/refresh/mod.rs @@ -41,7 +41,7 @@ pub struct RefreshHandle { diff_state: Arc<(Mutex, Condvar)>, upstream: UpstreamCoordinator, shutdown: Arc, - _watcher: Option, + watcher: Option, } #[derive(Clone)] @@ -200,7 +200,7 @@ impl RefreshHandle { diff_state, upstream, shutdown, - _watcher: watcher, + watcher, }; handle.request(); Ok(handle) @@ -235,6 +235,7 @@ fn push_event(events: &Mutex>, event: RefreshEvent) { impl Drop for RefreshHandle { fn drop(&mut self) { + self.watcher.take(); self.shutdown.store(true, Ordering::Release); let (state, wake) = &*self.diff_state; if let Ok(mut state) = state.lock() { @@ -242,9 +243,10 @@ impl Drop for RefreshHandle { } wake.notify_all(); let _ = self.requester.requests.try_send(Request::Shutdown); - // A running bounded Git command may still be finishing. Dropping its - // JoinHandle detaches it so terminal restoration is never delayed. - self.workers.clear(); + crate::adapters::command::terminate_active_commands(); + for worker in self.workers.drain(..) { + let _ = worker.join(); + } } } diff --git a/src/refresh/upstream.rs b/src/refresh/upstream.rs index 3b539f4..0d030d9 100644 --- a/src/refresh/upstream.rs +++ b/src/refresh/upstream.rs @@ -200,9 +200,9 @@ impl Drop for UpstreamCoordinator { } self.revision.fetch_add(1, Ordering::AcqRel); wake.notify_all(); - // A bounded Git command may still be finishing. Detach instead of - // delaying terminal restoration. - self.worker.take(); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } } } diff --git a/src/refresh/watcher.rs b/src/refresh/watcher.rs index 4961026..760c68e 100644 --- a/src/refresh/watcher.rs +++ b/src/refresh/watcher.rs @@ -1,19 +1,133 @@ use notify::{RecommendedWatcher, RecursiveMode, Watcher}; use std::path::Path; +use std::sync::mpsc; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +const QUIET_PERIOD: Duration = Duration::from_millis(200); + +pub(super) struct RepositoryWatcher { + watcher: Option, + stop: mpsc::SyncSender<()>, + worker: Option>, +} pub(super) fn watch_repository( git_dir: &Path, common_dir: &Path, requester: super::RefreshRequester, -) -> notify::Result { +) -> notify::Result { + let git_dir = git_dir.to_path_buf(); + let common_dir = common_dir.to_path_buf(); + let (events_send, events_receive) = mpsc::sync_channel(1); + let (stop_send, stop_receive) = mpsc::sync_channel(1); + let callback_git_dir = git_dir.clone(); + let callback_common_dir = common_dir.clone(); let mut watcher = notify::recommended_watcher(move |event: notify::Result| { - if event.is_ok() { - requester.request(); + if event.as_ref().is_ok_and(|event| { + event + .paths + .iter() + .any(|path| relevant(path, &callback_git_dir, &callback_common_dir)) + }) { + let _ = events_send.try_send(()); } })?; - watcher.watch(git_dir, RecursiveMode::Recursive)?; + watcher.watch(&git_dir, RecursiveMode::Recursive)?; if common_dir != git_dir { - watcher.watch(common_dir, RecursiveMode::Recursive)?; + watcher.watch(&common_dir, RecursiveMode::Recursive)?; + } + let worker = thread::Builder::new() + .name("stackmap-watcher".into()) + .spawn(move || { + loop { + if stop_receive.try_recv().is_ok() { + break; + } + if events_receive + .recv_timeout(Duration::from_millis(100)) + .is_err() + { + continue; + } + while events_receive.recv_timeout(QUIET_PERIOD).is_ok() {} + if stop_receive.try_recv().is_ok() { + break; + } + requester.request(); + } + }) + .map_err(notify::Error::io)?; + Ok(RepositoryWatcher { + watcher: Some(watcher), + stop: stop_send, + worker: Some(worker), + }) +} + +fn relevant(path: &Path, git_dir: &Path, common_dir: &Path) -> bool { + let relative = path + .strip_prefix(git_dir) + .or_else(|_| path.strip_prefix(common_dir)) + .unwrap_or(path); + let components: Vec<_> = relative.iter().collect(); + let first = components.first().and_then(|part| part.to_str()); + let file = relative.file_name().and_then(|part| part.to_str()); + + if file.is_some_and(|name| name.ends_with(".lock") || name.ends_with(".tmp")) { + return false; + } + matches!( + first, + Some("refs" | "worktrees" | "rebase-apply" | "rebase-merge") + ) || matches!( + file, + Some( + "HEAD" + | "ORIG_HEAD" + | "MERGE_HEAD" + | "CHERRY_PICK_HEAD" + | "REVERT_HEAD" + | "BISECT_LOG" + | "packed-refs" + | "index" + | "config" + | ".graphite_repo_config" + | ".graphite_metadata.db" + | ".graphite_metadata.db-wal" + | ".graphite_metadata.db-shm" + ) + ) || is_stackmap_config(relative) +} + +fn is_stackmap_config(path: &Path) -> bool { + path == Path::new("stackmap/config.toml") +} + +impl Drop for RepositoryWatcher { + fn drop(&mut self) { + self.watcher.take(); + let _ = self.stop.try_send(()); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ignores_transient_and_object_store_activity() { + let git = Path::new("/repo/.git"); + assert!(!relevant(Path::new("/repo/.git/index.lock"), git, git)); + assert!(!relevant(Path::new("/repo/.git/objects/ab/cdef"), git, git)); + assert!(relevant(Path::new("/repo/.git/index"), git, git)); + assert!(relevant( + Path::new("/repo/.git/refs/heads/feature"), + git, + git + )); } - Ok(watcher) }