diff --git a/app/src/ai/outline/native.rs b/app/src/ai/outline/native.rs index 60223ab5bdb..8fe7023adf8 100644 --- a/app/src/ai/outline/native.rs +++ b/app/src/ai/outline/native.rs @@ -36,6 +36,12 @@ struct OutlineState { status: OutlineStatus, /// Subscriber ID for repository updates (if watching). subscriber_id: Option, + /// Updates that arrived while a recomputation was already in flight (`status` was + /// `Pending`), coalesced via `RepositoryUpdate::merge` into a single accumulator rather + /// than queued individually -- so a burst of churn during one slow recomputation costs one + /// growing update, not one stored copy per update that arrived. Applied as one follow-up + /// recomputation once the in-flight one completes, so nothing is lost. + pending_update: RepositoryUpdate, } pub enum RepoOutlinesEvent { @@ -128,6 +134,7 @@ impl RepoOutlines { repository, status: OutlineStatus::Pending, subscriber_id: None, + pending_update: RepositoryUpdate::default(), }; self.outlines.insert(repo_path.clone(), outline_state); self.outline_queue.push_back(repo_path); @@ -379,36 +386,50 @@ impl RepoOutlines { return; } - match self.outlines.get_mut(repo_path) { - Some(OutlineState { - status: outline_status @ OutlineStatus::Complete(_), - .. - }) => { - let mut outline = OutlineStatus::Pending; - std::mem::swap(outline_status, &mut outline); + let Some(state) = self.outlines.get_mut(repo_path) else { + log::warn!("Failed to update repo outline: repo outline not found"); + return; + }; + + match &mut state.status { + OutlineStatus::Complete(_) => { + let OutlineStatus::Complete(outline) = + std::mem::replace(&mut state.status, OutlineStatus::Pending) + else { + unreachable!("Expected status to be Complete(outline)") + }; let repo_path_clone_inner = repo_path.to_path_buf(); ctx.spawn( async move { - if let OutlineStatus::Complete(mut outline) = outline { - outline.update(update).await; - (outline, repo_path_clone_inner) - } else { - unreachable!("Expected status to be Complete(outline)") - } + let mut outline = outline; + outline.update(update).await; + (outline, repo_path_clone_inner) }, move |me, (outline, repo_path), ctx| { - if let Some(state) = me.outlines.get_mut(&repo_path) { - state.status = OutlineStatus::Complete(outline); - ctx.emit(RepoOutlinesEvent::OutlinesUpdated(repo_path)); + let Some(state) = me.outlines.get_mut(&repo_path) else { + return; + }; + state.status = OutlineStatus::Complete(outline); + ctx.emit(RepoOutlinesEvent::OutlinesUpdated(repo_path.clone())); + + // Apply whatever coalesced in while we were recomputing, so sustained + // filesystem churn under a slow recomputation never loses an update. + if !state.pending_update.is_empty() { + let next_update = std::mem::take(&mut state.pending_update); + me.handle_repository_update(&repo_path, next_update, ctx); } }, ); } - Some(_) => { - log::warn!("Failed to update repo outline: repo outline failed or is pending") + OutlineStatus::Pending => { + // A recomputation is already in flight; merge this update into the pending + // accumulator so it isn't lost, to be applied once that recomputation completes. + state.pending_update.merge(&update); + } + OutlineStatus::Failed => { + log::warn!("Failed to update repo outline: repo outline failed"); } - None => log::warn!("Failed to update repo outline: repo outline not found"), } } } @@ -446,3 +467,7 @@ impl RepositorySubscriber for OutlineRepositorySubscriber { }) } } + +#[cfg(test)] +#[path = "native_tests.rs"] +mod tests; diff --git a/app/src/ai/outline/native_tests.rs b/app/src/ai/outline/native_tests.rs new file mode 100644 index 00000000000..be99bd1ec80 --- /dev/null +++ b/app/src/ai/outline/native_tests.rs @@ -0,0 +1,117 @@ +use std::time::Duration; + +use ai::index::build_outline; +use repo_metadata::{DirectoryWatcher, RepositoryUpdate, TargetFile}; +use tempfile::TempDir; +use warp_util::standardized_path::StandardizedPath; +use warpui::App; +use warpui_core::r#async::Timer; + +use super::{OutlineState, OutlineStatus, RepoOutlines}; + +/// `OutlineRepositorySubscriber` resolves as soon as it enqueues onto its own channel, while the +/// real recomputation (`RepoOutlines::handle_repository_update`) keeps running for much longer. +/// This exercises that two-stage path directly: it drives an actual `Outline` recomputation via +/// `handle_repository_update` and submits a second update while the first is still in flight, to +/// prove the second is coalesced into the accumulator -- not dropped, and not starting a second, +/// overlapping recomputation. +#[test] +fn concurrent_update_while_pending_is_merged_not_dropped_or_double_recomputed() { + App::test((), |mut app| async move { + let temp_dir = TempDir::new().unwrap(); + let repo_path = dunce::canonicalize(temp_dir.path()).unwrap(); + std::fs::write(repo_path.join("existing.rs"), "fn existing() {}\n").unwrap(); + + let baseline_outline = build_outline(&repo_path, None).await.unwrap(); + assert_eq!(baseline_outline.file_count(), 1); + + let watcher_handle = app.add_singleton_model(DirectoryWatcher::new_for_testing); + let repository_handle = watcher_handle + .update(&mut app, |watcher, ctx| { + watcher.add_directory( + StandardizedPath::from_local_canonicalized(&repo_path).unwrap(), + ctx, + ) + }) + .unwrap(); + + let outlines_handle = app.add_singleton_model(RepoOutlines::new_for_test); + outlines_handle.update(&mut app, |outlines, _| { + outlines.outlines.insert( + repo_path.clone(), + OutlineState { + repository: repository_handle, + status: OutlineStatus::Complete(baseline_outline), + subscriber_id: None, + pending_update: RepositoryUpdate::default(), + }, + ); + }); + + std::fs::write(repo_path.join("first.rs"), "fn first() {}\n").unwrap(); + std::fs::write(repo_path.join("second.rs"), "fn second() {}\n").unwrap(); + + let first_update = RepositoryUpdate { + added: [TargetFile::new(repo_path.join("first.rs"), false)].into(), + ..Default::default() + }; + let second_update = RepositoryUpdate { + added: [TargetFile::new(repo_path.join("second.rs"), false)].into(), + ..Default::default() + }; + + outlines_handle.update(&mut app, |outlines, ctx| { + outlines.handle_repository_update(&repo_path, first_update, ctx); + }); + + // `handle_repository_update` flips `status` to `Pending` synchronously, before the + // spawned recomputation is ever polled, so this deterministically observes it in flight. + outlines_handle.read(&app, |outlines, _| { + let state = outlines.outlines.get(&repo_path).unwrap(); + assert!(matches!(state.status, OutlineStatus::Pending)); + }); + + outlines_handle.update(&mut app, |outlines, ctx| { + outlines.handle_repository_update(&repo_path, second_update.clone(), ctx); + }); + + // It must be merged into the accumulator: still `Pending` (no second, overlapping + // recomputation was started), and the accumulator holds exactly the second update. + outlines_handle.read(&app, |outlines, _| { + let state = outlines.outlines.get(&repo_path).unwrap(); + assert!(matches!(state.status, OutlineStatus::Pending)); + assert_eq!(state.pending_update.added, second_update.added); + }); + + // Wait for the in-flight recomputation -- which then applies the merged update as a + // follow-up recomputation -- to fully settle. + let mut waited = Duration::ZERO; + loop { + let is_complete = outlines_handle.read(&app, |outlines, _| { + matches!( + outlines.outlines.get(&repo_path).map(|state| &state.status), + Some(OutlineStatus::Complete(_)) + ) + }); + if is_complete { + break; + } + assert!( + waited < Duration::from_secs(10), + "recomputation never completed" + ); + Timer::after(Duration::from_millis(10)).await; + waited += Duration::from_millis(10); + } + + outlines_handle.read(&app, |outlines, _| { + let state = outlines.outlines.get(&repo_path).unwrap(); + assert!(state.pending_update.is_empty()); + let OutlineStatus::Complete(outline) = &state.status else { + panic!("expected Complete status"); + }; + // The baseline file plus both concurrently-submitted updates are all reflected. + assert_eq!(outline.file_count(), 3); + }); + }); +} diff --git a/crates/repo_metadata/src/repository.rs b/crates/repo_metadata/src/repository.rs index 104d88413a9..f2f6736c20e 100644 --- a/crates/repo_metadata/src/repository.rs +++ b/crates/repo_metadata/src/repository.rs @@ -575,16 +575,16 @@ impl Entity for Repository { } /// Coalescing merge for RepositoryUpdate with normalization rules. -fn merge_repository_updates(acc: &mut RepositoryUpdate, incoming: &RepositoryUpdate) { +pub(crate) fn merge_repository_updates(acc: &mut RepositoryUpdate, incoming: &RepositoryUpdate) { // 1) Moves first for (to, from) in &incoming.moved { if acc.added.remove(from) { acc.added.insert(to.clone()); - return; + continue; } if acc.modified.remove(from) { acc.modified.insert(to.clone()); - return; + continue; } // Collapse chain: if `from` was a prior destination, pull its original source @@ -641,30 +641,128 @@ fn merge_repository_updates(acc: &mut RepositoryUpdate, incoming: &RepositoryUpd acc.remote_ref_updated |= incoming.remote_ref_updated; } +/// Maximum number of buffered filesystem changes (the combined size of `added`, `modified`, +/// `deleted`, and `moved`) a [`BufferingRepositorySubscriber`] accumulates before it forces a +/// flush ahead of the debounce timer. Bounds memory under sustained filesystem churn (e.g. a +/// large `git checkout` or `npm install`) while remaining large enough that ordinary update +/// bursts still coalesce through the debounce window instead of flushing on every batch. +const DEFAULT_MAX_PENDING_ENTRIES: usize = 10_000; + +fn pending_entry_count(update: &RepositoryUpdate) -> usize { + update.added.len() + update.modified.len() + update.deleted.len() + update.moved.len() +} + /// A generic debouncing layer for any RepositorySubscriber. pub struct BufferingRepositorySubscriber { inner: Arc>, state: Arc>, debounce: Duration, + max_pending_entries: usize, } -#[derive(Default)] struct BufferState { pending: RepositoryUpdate, /// Monotonic counter incremented for each incoming update; used to implement true debounce. version: u64, /// Whether the background flusher loop is currently running. flush_handle: Option, + /// A batch that crossed `max_pending_entries` (or was handed off by the debounce flusher) + /// while a delivery to `inner` was already in flight. Further such batches are coalesced + /// into this one via `merge_repository_updates` rather than queued separately, so the + /// backlog stays proportional to the number of distinct un-delivered paths instead of + /// growing with the number of batches that formed while `inner` was busy. Taken and + /// delivered as a single batch once the in-flight delivery completes. + next_delivery: Option, + /// Whether a batch is currently being delivered to `inner`. Only one delivery is ever in + /// flight at a time, so `inner` never sees overlapping or out-of-order calls to + /// `on_files_updated`. + delivering: bool, + /// Cleared on unsubscribe so an in-flight or still-forming delivery becomes a no-op instead + /// of continuing to call `inner` (and retaining memory) after the subscription has ended. + active: bool, +} + +impl Default for BufferState { + fn default() -> Self { + Self { + pending: RepositoryUpdate::default(), + version: 0, + flush_handle: None, + next_delivery: None, + delivering: false, + active: true, + } + } } impl BufferingRepositorySubscriber { pub fn new(inner: S, debounce: Duration) -> Self { + Self::new_with_max_pending_entries(inner, debounce, DEFAULT_MAX_PENDING_ENTRIES) + } + + fn new_with_max_pending_entries( + inner: S, + debounce: Duration, + max_pending_entries: usize, + ) -> Self { Self { inner: Arc::new(Mutex::new(inner)), state: Arc::new(Mutex::new(BufferState::default())), debounce, + max_pending_entries, } } + + /// Like [`Self::new`], but allows overriding the maximum number of buffered filesystem + /// changes before a flush is forced ahead of the debounce timer. Exposed for tests that need + /// to exercise the bound without buffering `DEFAULT_MAX_PENDING_ENTRIES` real entries. + #[cfg(any(test, feature = "test-util"))] + pub fn with_max_pending_entries( + inner: S, + debounce: Duration, + max_pending_entries: usize, + ) -> Self { + Self::new_with_max_pending_entries(inner, debounce, max_pending_entries) + } +} + +/// Splits `update` into a sequence of single-entry updates, in the same phase order +/// `merge_repository_updates` already applies within a single call (moves, then adds, then +/// modifies, then deletes, with the boolean flags folded into one final update). +/// `merge_repository_updates`'s per-phase coalescing only depends on phases being applied in +/// that order, not on how many entries are merged in one call, so merging this sequence one +/// piece at a time is equivalent to merging `update` as a whole. +fn single_entry_updates(update: &RepositoryUpdate) -> impl Iterator + '_ { + let moves = update.moved.iter().map(|(to, from)| RepositoryUpdate { + moved: [(to.clone(), from.clone())].into(), + ..Default::default() + }); + let adds = update.added.iter().map(|p| RepositoryUpdate { + added: [p.clone()].into(), + ..Default::default() + }); + let modifies = update.modified.iter().map(|p| RepositoryUpdate { + modified: [p.clone()].into(), + ..Default::default() + }); + let deletes = update.deleted.iter().map(|p| RepositoryUpdate { + deleted: [p.clone()].into(), + ..Default::default() + }); + let flags = (update.commit_updated || update.index_lock_detected || update.remote_ref_updated) + .then(|| RepositoryUpdate { + commit_updated: update.commit_updated, + index_lock_detected: update.index_lock_detected, + remote_ref_updated: update.remote_ref_updated, + ..Default::default() + }) + .into_iter(); + + moves + .chain(adds) + .chain(modifies) + .chain(deletes) + .chain(flags) } impl RepositorySubscriber for BufferingRepositorySubscriber @@ -681,19 +779,19 @@ where fn on_files_updated( &mut self, - _repository: &Repository, + repository: &Repository, update: &RepositoryUpdate, ctx: &mut ModelContext, ) -> Pin + Send + 'static>> { { let mut st = self.state.lock().unwrap(); - merge_repository_updates(&mut st.pending, update); st.version = st.version.wrapping_add(1); // Start a single background flusher if it's not already running. if st.flush_handle.is_none() { - let inner = Arc::clone(&self.inner); - let state = Arc::clone(&self.state); + let state_for_loop = Arc::clone(&self.state); + let inner_for_completion = Arc::clone(&self.inner); + let state_for_completion = Arc::clone(&self.state); let wait = self.debounce; st.flush_handle = Some(ctx.spawn( @@ -702,7 +800,7 @@ where loop { // Capture current version, then wait. let start_version = { - let st = state.lock().unwrap(); + let st = state_for_loop.lock().unwrap(); st.version }; warpui_core::r#async::Timer::after(wait).await; @@ -712,7 +810,7 @@ where // Yield before flushing to check if the current flush is cancelled. futures_lite::future::yield_now().await; - let mut st = state.lock().unwrap(); + let mut st = state_for_loop.lock().unwrap(); if st.version == start_version { st.flush_handle = None; Some(std::mem::take(&mut st.pending)) @@ -723,24 +821,50 @@ where }; if let Some(merged) = maybe_merged { - break (inner, merged); + break merged; } } }, - |repo_model, (inner, merged), repo_ctx| { + move |repo_model, merged, repo_ctx| { if merged.is_empty() { return; } - if let Ok(mut inner) = inner.lock() { - let fut = inner.on_files_updated(repo_model, &merged, repo_ctx); - // Drive the subscriber's async update to completion. - repo_ctx.spawn(fut, |_, _, _| {}); - } + Self::hand_off(&state_for_completion, merged); + Self::advance_delivery( + inner_for_completion, + state_for_completion, + repo_model, + repo_ctx, + ); }, )); } } + // Merge one entry at a time, so `pending` never grows past `max_pending_entries` even + // when `update` by itself is far larger than that. Whenever a chunk crosses the bound, + // try to deliver it immediately: if nothing else is in flight it goes out right away + // (bounded); if a delivery is already running, it's coalesced into the single pending + // backlog instead of queued, so a slow consumer never accumulates more than one + // (still-bounded-in-content, if not in size) extra batch. + for chunk in single_entry_updates(update) { + let ready = { + let mut st = self.state.lock().unwrap(); + merge_repository_updates(&mut st.pending, &chunk); + (pending_entry_count(&st.pending) >= self.max_pending_entries) + .then(|| std::mem::take(&mut st.pending)) + }; + let Some(ready) = ready else { continue }; + + Self::hand_off(&self.state, ready); + Self::advance_delivery( + Arc::clone(&self.inner), + Arc::clone(&self.state), + repository, + ctx, + ); + } + Box::pin(ready(())) } @@ -751,6 +875,68 @@ where if let Some(handle) = st.flush_handle.take() { handle.abort(); } + // Release any buffered or backlogged work and mark this subscription inactive, so an + // already-in-flight delivery's completion becomes a no-op instead of continuing to call + // `inner` (and keeping this state alive) after the subscription has ended. + st.active = false; + st.next_delivery = None; + st.pending = RepositoryUpdate::default(); + } +} + +impl BufferingRepositorySubscriber +where + S: RepositorySubscriber + Send + Sync + 'static, +{ + /// Coalesces `batch` into the pending delivery backlog, merging with whatever is already + /// there instead of storing it separately. + fn hand_off(state: &Arc>, batch: RepositoryUpdate) { + let mut st = state.lock().unwrap(); + match &mut st.next_delivery { + Some(existing) => merge_repository_updates(existing, &batch), + None => st.next_delivery = Some(batch), + } + } + + /// Delivers the pending backlog to `inner`, unless a delivery is already in flight or the + /// subscription has been unsubscribed. On completion, recurses to pick up whatever + /// coalesced in next, so `inner.on_files_updated` never has two batches in flight at once. + fn advance_delivery( + inner: Arc>, + state: Arc>, + repository: &Repository, + ctx: &mut ModelContext, + ) { + let batch = { + let mut st = state.lock().unwrap(); + if !st.active || st.delivering { + return; + } + let Some(batch) = st.next_delivery.take() else { + return; + }; + st.delivering = true; + batch + }; + + let Ok(mut inner_guard) = inner.lock() else { + state.lock().unwrap().delivering = false; + return; + }; + let fut = inner_guard.on_files_updated(repository, &batch, ctx); + drop(inner_guard); + + let inner_for_next = Arc::clone(&inner); + let state_for_next = Arc::clone(&state); + ctx.spawn(fut, move |repo_model, (), repo_ctx| { + let mut st = state_for_next.lock().unwrap(); + st.delivering = false; + let active = st.active; + drop(st); + if active { + Self::advance_delivery(inner_for_next, state_for_next, repo_model, repo_ctx); + } + }); } } diff --git a/crates/repo_metadata/src/repository_tests.rs b/crates/repo_metadata/src/repository_tests.rs index fdcf7900b06..66db1f37ada 100644 --- a/crates/repo_metadata/src/repository_tests.rs +++ b/crates/repo_metadata/src/repository_tests.rs @@ -1,6 +1,9 @@ +use std::collections::HashSet; use std::future::Future; use std::path::PathBuf; use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use futures::channel::mpsc; @@ -11,8 +14,8 @@ use warpui_core::r#async::Timer; use warpui_core::{App, ModelContext}; use super::{ - Repository, RepositorySubscriber, RepositorySubscription, RepositoryWatchMode, - TrackedRemoteRef, merge_repository_updates, + BufferingRepositorySubscriber, Repository, RepositorySubscriber, RepositorySubscription, + RepositoryWatchMode, TrackedRemoteRef, merge_repository_updates, }; use crate::repositories::stub_git_repository; use crate::watcher::DirectoryWatcher; @@ -45,6 +48,46 @@ impl RepositorySubscriber for RecordingSubscriber { } } +/// A subscriber whose `on_files_updated` takes `delay` to resolve, and asserts that it is never +/// invoked again before the previous call's returned future has resolved. Used to prove that +/// `BufferingRepositorySubscriber` serializes delivery instead of racing overlapping calls. +struct SlowRecordingSubscriber { + update_tx: mpsc::UnboundedSender, + in_flight: Arc, + delay: Duration, +} + +impl RepositorySubscriber for SlowRecordingSubscriber { + fn on_scan( + &mut self, + _repository: &Repository, + _ctx: &mut ModelContext, + ) -> Pin + Send + 'static>> { + Box::pin(async {}) + } + + fn on_files_updated( + &mut self, + _repository: &Repository, + update: &RepositoryUpdate, + _ctx: &mut ModelContext, + ) -> Pin + Send + 'static>> { + assert!( + !self.in_flight.swap(true, Ordering::SeqCst), + "on_files_updated was invoked again before the previous call's future resolved" + ); + let update = update.clone(); + let update_tx = self.update_tx.clone(); + let in_flight = Arc::clone(&self.in_flight); + let delay = self.delay; + Box::pin(async move { + Timer::after(delay).await; + let _ = update_tx.unbounded_send(update); + in_flight.store(false, Ordering::SeqCst); + }) + } +} + fn add_recording_subscriber( repository: &mut Repository, mode: RepositoryWatchMode, @@ -61,6 +104,45 @@ fn add_recording_subscriber( ); subscriber_id } + +/// Awaits `update_rx` until every file in `all_files` has been seen across delivered batches +/// (failing on any duplicate along the way), or `deadline` elapses first -- in which case it +/// panics with the batches seen so far and the files still missing, instead of hanging until +/// nextest's slow-timeout. `on_batch` is called with the 0-based batch index and the batch +/// itself before its files are recorded, so callers can assert per-batch properties without +/// reimplementing the receive loop. Returns the received files and the number of batches. +async fn collect_until_seen( + update_rx: &mut mpsc::UnboundedReceiver, + all_files: &[TargetFile], + deadline: Duration, + mut on_batch: impl FnMut(usize, &RepositoryUpdate), +) -> (HashSet, usize) { + let mut seen = HashSet::new(); + let mut batch_count = 0; + let sleep = futures::FutureExt::fuse(Timer::after(deadline)); + futures::pin_mut!(sleep); + while seen.len() < all_files.len() { + futures::select! { + flushed = update_rx.next().fuse() => { + let flushed = flushed.expect("channel closed before every update was seen"); + on_batch(batch_count, &flushed); + for file in flushed.added { + assert!(seen.insert(file), "duplicate file delivered across batches"); + } + batch_count += 1; + } + _ = sleep => { + let missing: Vec<_> = all_files.iter().filter(|f| !seen.contains(f)).collect(); + panic!( + "timed out after {batch_count} batch(es) waiting for the rest to be \ + delivered; missing: {missing:?}" + ); + } + } + } + (seen, batch_count) +} + #[test] fn tracked_remote_ref_validates_full_ref_names() { assert_eq!( @@ -198,6 +280,104 @@ fn merge_repository_updates_preserves_remote_ref_updates() { ); } +#[test] +fn merge_repository_updates_applies_every_entry_when_a_move_collapses_into_added() { + let source = TargetFile::new(PathBuf::from("/repo/source.txt"), false); + let move_target = TargetFile::new(PathBuf::from("/repo/moved.txt"), false); + let unrelated_move_to = TargetFile::new(PathBuf::from("/repo/d.txt"), false); + let unrelated_move_from = TargetFile::new(PathBuf::from("/repo/c.txt"), false); + let added_file = TargetFile::new(PathBuf::from("/repo/added.txt"), false); + let deleted_file = TargetFile::new(PathBuf::from("/repo/deleted.txt"), false); + + // `acc` already records `source` as added, so the incoming batch's move collapses into it + // via the `acc.added.remove(from)` branch. + let mut acc = RepositoryUpdate { + added: [source.clone()].into(), + ..Default::default() + }; + + let incoming = RepositoryUpdate { + // A pre-existing bug used `return` instead of `continue` in this branch, which would + // abandon the unrelated move plus every later phase below. + moved: [(move_target.clone(), source.clone())].into(), + added: [added_file.clone()].into(), + deleted: [deleted_file.clone()].into(), + remote_ref_updated: true, + ..Default::default() + }; + // A second, independent move exercises that the moves loop itself keeps iterating (not + // just the phases after it); kept separate to avoid HashMap-iteration-order ambiguity with + // the colliding move above. + let mut incoming_with_second_move = incoming.clone(); + incoming_with_second_move + .moved + .insert(unrelated_move_to.clone(), unrelated_move_from.clone()); + + merge_repository_updates(&mut acc, &incoming_with_second_move); + + assert!( + acc.added.contains(&move_target), + "the colliding move should collapse into `added`" + ); + assert!(!acc.added.contains(&source)); + assert_eq!( + acc.moved.get(&unrelated_move_to), + Some(&unrelated_move_from), + "the unrelated move must still be recorded" + ); + assert!( + acc.added.contains(&added_file), + "adds after the moves phase must still be applied" + ); + assert!( + acc.deleted.contains(&deleted_file), + "deletes after the moves phase must still be applied" + ); + assert!(acc.remote_ref_updated, "flags must still be folded in"); +} + +#[test] +fn merge_repository_updates_applies_every_entry_when_a_move_collapses_into_modified() { + let source = TargetFile::new(PathBuf::from("/repo/source.txt"), false); + let move_target = TargetFile::new(PathBuf::from("/repo/moved.txt"), false); + let unrelated_move_to = TargetFile::new(PathBuf::from("/repo/d.txt"), false); + let unrelated_move_from = TargetFile::new(PathBuf::from("/repo/c.txt"), false); + let added_file = TargetFile::new(PathBuf::from("/repo/added.txt"), false); + + // `acc` already records `source` as modified, so the incoming batch's move collapses into + // it via the `acc.modified.remove(from)` branch. + let mut acc = RepositoryUpdate { + modified: [source.clone()].into(), + ..Default::default() + }; + + let mut incoming = RepositoryUpdate { + moved: [(move_target.clone(), source.clone())].into(), + added: [added_file.clone()].into(), + ..Default::default() + }; + incoming + .moved + .insert(unrelated_move_to.clone(), unrelated_move_from.clone()); + + merge_repository_updates(&mut acc, &incoming); + + assert!( + acc.modified.contains(&move_target), + "the colliding move should collapse into `modified`" + ); + assert!(!acc.modified.contains(&source)); + assert_eq!( + acc.moved.get(&unrelated_move_to), + Some(&unrelated_move_from), + "the unrelated move must still be recorded" + ); + assert!( + acc.added.contains(&added_file), + "adds after the moves phase must still be applied" + ); +} + #[test] fn filesystem_only_subscription_does_not_activate_git_tracking() { VirtualFS::test( @@ -433,6 +613,459 @@ fn tracked_remote_ref_change_notifies_subscribers() { }); } +#[test] +fn forced_flush_drains_pending_before_debounce_timer_fires() { + VirtualFS::test( + "forced_flush_drains_pending_before_debounce_timer_fires", + |dirs, mut vfs| { + vfs.mkdir("repo"); + let repo_path = dirs.tests().join("repo"); + + App::test((), |mut app| async move { + let watcher_handle = app.add_singleton_model(DirectoryWatcher::new_for_testing); + let repo_handle = watcher_handle + .update(&mut app, |watcher, ctx| { + watcher.add_directory( + StandardizedPath::from_local_canonicalized(&repo_path).unwrap(), + ctx, + ) + }) + .unwrap(); + + const MAX_PENDING_ENTRIES: usize = 3; + let (update_tx, mut update_rx) = mpsc::unbounded::(); + let start = repo_handle.update(&mut app, |repo, ctx| { + let buffered = BufferingRepositorySubscriber::with_max_pending_entries( + RecordingSubscriber { update_tx }, + // Long enough that the debounce timer cannot plausibly fire during + // the test, so any observed flush must come from the forced path. + Duration::from_secs(3600), + MAX_PENDING_ENTRIES, + ); + repo.start_watching( + RepositoryWatchMode::FilesystemOnly, + Box::new(buffered), + ctx, + ) + }); + std::mem::drop(start.registration_future); + let subscriber_id = start.subscriber_id; + + let expected_files: Vec<_> = (0..MAX_PENDING_ENTRIES) + .map(|i| TargetFile::new(repo_path.join(format!("file{i}.txt")), false)) + .collect(); + for file in &expected_files { + let update = RepositoryUpdate { + added: [file.clone()].into(), + ..Default::default() + }; + repo_handle.update(&mut app, |repo, ctx| { + repo.notify_subscriber(subscriber_id, &update, ctx); + }); + } + + let flushed = update_rx.next().await.expect("forced flush"); + assert_eq!(flushed.added.len(), MAX_PENDING_ENTRIES); + for file in &expected_files { + assert!(flushed.added.contains(file)); + } + }); + }, + ); +} + +#[test] +fn single_incoming_update_exceeding_the_bound_is_delivered_without_loss() { + VirtualFS::test( + "single_incoming_update_exceeding_the_bound_is_delivered_without_loss", + |dirs, mut vfs| { + vfs.mkdir("repo"); + let repo_path = dirs.tests().join("repo"); + + App::test((), |mut app| async move { + let watcher_handle = app.add_singleton_model(DirectoryWatcher::new_for_testing); + let repo_handle = watcher_handle + .update(&mut app, |watcher, ctx| { + watcher.add_directory( + StandardizedPath::from_local_canonicalized(&repo_path).unwrap(), + ctx, + ) + }) + .unwrap(); + + const MAX_PENDING_ENTRIES: usize = 4; + let (update_tx, mut update_rx) = mpsc::unbounded::(); + let start = repo_handle.update(&mut app, |repo, ctx| { + let buffered = BufferingRepositorySubscriber::with_max_pending_entries( + RecordingSubscriber { update_tx }, + // Long enough that the debounce timer cannot plausibly fire during + // the test, so every observed batch must come from the forced, + // within-a-single-update chunking path. + Duration::from_secs(3600), + MAX_PENDING_ENTRIES, + ); + repo.start_watching( + RepositoryWatchMode::FilesystemOnly, + Box::new(buffered), + ctx, + ) + }); + std::mem::drop(start.registration_future); + let subscriber_id = start.subscriber_id; + + // A single incoming update far larger than the configured bound, exactly the + // large-single-event case a coalesced filesystem-watcher batch can produce (a + // huge git checkout or npm install landing as one `RepositoryUpdate`). A clean + // multiple of the bound so nothing is left waiting on the (unreached) debounce. + let total_files = MAX_PENDING_ENTRIES * 3; + let all_files: Vec<_> = (0..total_files) + .map(|i| TargetFile::new(repo_path.join(format!("file{i}.txt")), false)) + .collect(); + let huge_update = RepositoryUpdate { + added: all_files.iter().cloned().collect(), + ..Default::default() + }; + + repo_handle.update(&mut app, |repo, ctx| { + repo.notify_subscriber(subscriber_id, &huge_update, ctx); + }); + + let (seen, batch_count) = collect_until_seen( + &mut update_rx, + &all_files, + Duration::from_secs(5), + |index, flushed| { + if index == 0 { + // Nothing else is in flight yet the first time the bound is + // crossed, so that first batch is delivered immediately, on its + // own, bounded. + assert_eq!(flushed.added.len(), MAX_PENDING_ENTRIES); + } + }, + ) + .await; + + assert_eq!(seen.len(), total_files); + for file in &all_files { + assert!(seen.contains(file)); + } + // Everything after the first bounded batch coalesces into a single backlog + // while that first delivery is in flight (see `BufferState::next_delivery`), so + // this is only guaranteed to be more than one batch, not every batch bounded. + assert!( + batch_count > 1, + "a single update far exceeding the bound should still be split into more than one batch" + ); + }); + }, + ); +} + +#[test] +fn forced_and_debounced_flushes_apply_every_update_exactly_once_with_a_slow_consumer() { + VirtualFS::test( + "forced_and_debounced_flushes_apply_every_update_exactly_once_with_a_slow_consumer", + |dirs, mut vfs| { + vfs.mkdir("repo"); + let repo_path = dirs.tests().join("repo"); + + App::test((), |mut app| async move { + let watcher_handle = app.add_singleton_model(DirectoryWatcher::new_for_testing); + let repo_handle = watcher_handle + .update(&mut app, |watcher, ctx| { + watcher.add_directory( + StandardizedPath::from_local_canonicalized(&repo_path).unwrap(), + ctx, + ) + }) + .unwrap(); + + const MAX_PENDING_ENTRIES: usize = 5; + // Short enough to elapse comfortably within the test's timeout, but long enough + // that the feed loop below (which never awaits between updates) reliably + // finishes well before it fires. + const DEBOUNCE: Duration = Duration::from_millis(30); + // Longer than `DEBOUNCE`, so the debounce timer can fire while a forced batch is + // still being delivered -- exercising the case where a debounce-triggered flush + // has to wait behind an in-flight forced delivery instead of racing it. + const CONSUMER_DELAY: Duration = Duration::from_millis(40); + + let (update_tx, mut update_rx) = mpsc::unbounded::(); + let in_flight = Arc::new(AtomicBool::new(false)); + let start = repo_handle.update(&mut app, |repo, ctx| { + let buffered = BufferingRepositorySubscriber::with_max_pending_entries( + SlowRecordingSubscriber { + update_tx, + in_flight: Arc::clone(&in_flight), + delay: CONSUMER_DELAY, + }, + DEBOUNCE, + MAX_PENDING_ENTRIES, + ); + repo.start_watching( + RepositoryWatchMode::FilesystemOnly, + Box::new(buffered), + ctx, + ) + }); + std::mem::drop(start.registration_future); + let subscriber_id = start.subscriber_id; + + // Two forced flushes' worth, plus a partial batch that only the debounce timer + // flushes once the feed goes quiet. + let total_updates = MAX_PENDING_ENTRIES * 2 + 3; + let all_files: Vec<_> = (0..total_updates) + .map(|i| TargetFile::new(repo_path.join(format!("file{i}.txt")), false)) + .collect(); + for file in &all_files { + let update = RepositoryUpdate { + added: [file.clone()].into(), + ..Default::default() + }; + repo_handle.update(&mut app, |repo, ctx| { + repo.notify_subscriber(subscriber_id, &update, ctx); + }); + } + + // Some number of forced and/or debounced batches account for every update -- + // exactly how many depends on timing (a forced batch that's still in flight + // when more work becomes ready coalesces that work into one backlog rather than + // a separate batch), but nothing may ever be dropped or duplicated. + // `SlowRecordingSubscriber` itself asserts that no two batches are ever + // delivered concurrently. + let (seen, _batch_count) = collect_until_seen( + &mut update_rx, + &all_files, + Duration::from_secs(5), + |_, flushed| assert!(!flushed.added.is_empty()), + ) + .await; + + assert_eq!(seen.len(), total_updates); + for file in &all_files { + assert!(seen.contains(file)); + } + + // Nothing else should ever arrive. + futures::select! { + update = update_rx.next().fuse() => { + panic!("unexpected extra flush: {update:?}"); + } + _ = futures::FutureExt::fuse(Timer::after(Duration::from_millis(300))) => {} + } + }); + }, + ); +} + +#[test] +fn unsubscribe_cancels_pending_delivery_and_releases_the_backlog() { + VirtualFS::test( + "unsubscribe_cancels_pending_delivery_and_releases_the_backlog", + |dirs, mut vfs| { + vfs.mkdir("repo"); + let repo_path = dirs.tests().join("repo"); + + App::test((), |mut app| async move { + let watcher_handle = app.add_singleton_model(DirectoryWatcher::new_for_testing); + let repo_handle = watcher_handle + .update(&mut app, |watcher, ctx| { + watcher.add_directory( + StandardizedPath::from_local_canonicalized(&repo_path).unwrap(), + ctx, + ) + }) + .unwrap(); + + const MAX_PENDING_ENTRIES: usize = 3; + let (update_tx, mut update_rx) = mpsc::unbounded::(); + let in_flight = Arc::new(AtomicBool::new(false)); + let start = repo_handle.update(&mut app, |repo, ctx| { + let buffered = BufferingRepositorySubscriber::with_max_pending_entries( + SlowRecordingSubscriber { + update_tx, + in_flight: Arc::clone(&in_flight), + delay: Duration::from_millis(50), + }, + Duration::from_secs(3600), + MAX_PENDING_ENTRIES, + ); + repo.start_watching( + RepositoryWatchMode::FilesystemOnly, + Box::new(buffered), + ctx, + ) + }); + std::mem::drop(start.registration_future); + let subscriber_id = start.subscriber_id; + + // First threshold crossing: dispatched immediately, taking 50ms to resolve. + // Second threshold crossing: nothing is in flight, so it coalesces into the + // pending backlog instead of being delivered. + let total_updates = MAX_PENDING_ENTRIES * 2; + for i in 0..total_updates { + let update = RepositoryUpdate { + added: [TargetFile::new( + repo_path.join(format!("file{i}.txt")), + false, + )] + .into(), + ..Default::default() + }; + repo_handle.update(&mut app, |repo, ctx| { + repo.notify_subscriber(subscriber_id, &update, ctx); + }); + } + + // Unsubscribe while the first batch is still in flight and the second is only + // backlogged. + repo_handle.update(&mut app, |repo, ctx| { + repo.stop_watching(subscriber_id, ctx); + }); + + // The already-in-flight first batch still completes normally... + let flushed = update_rx + .next() + .await + .expect("the in-flight delivery still completes"); + assert_eq!(flushed.added.len(), MAX_PENDING_ENTRIES); + + // ...but the backlogged second batch must never be delivered: unsubscribing + // released it instead of delivering it late. The channel closing (because + // dropping the subscription released the last reference to the subscriber) is + // an acceptable way for that to manifest, same as the timer simply elapsing. + futures::select! { + result = update_rx.next().fuse() => { + if let Some(update) = result { + panic!("unexpected delivery after unsubscribe: {update:?}"); + } + } + _ = futures::FutureExt::fuse(Timer::after(Duration::from_millis(200))) => {} + } + }); + }, + ); +} + +#[test] +fn coalescing_a_move_that_collapses_into_an_existing_add_still_applies_later_entries() { + VirtualFS::test( + "coalescing_a_move_that_collapses_into_an_existing_add_still_applies_later_entries", + |dirs, mut vfs| { + vfs.mkdir("repo"); + let repo_path = dirs.tests().join("repo"); + + App::test((), |mut app| async move { + let watcher_handle = app.add_singleton_model(DirectoryWatcher::new_for_testing); + let repo_handle = watcher_handle + .update(&mut app, |watcher, ctx| { + watcher.add_directory( + StandardizedPath::from_local_canonicalized(&repo_path).unwrap(), + ctx, + ) + }) + .unwrap(); + + const MAX_PENDING_ENTRIES: usize = 3; + let (update_tx, mut update_rx) = mpsc::unbounded::(); + let in_flight = Arc::new(AtomicBool::new(false)); + let start = repo_handle.update(&mut app, |repo, ctx| { + let buffered = BufferingRepositorySubscriber::with_max_pending_entries( + SlowRecordingSubscriber { + update_tx, + in_flight: Arc::clone(&in_flight), + delay: Duration::from_millis(80), + }, + Duration::from_secs(3600), + MAX_PENDING_ENTRIES, + ); + repo.start_watching( + RepositoryWatchMode::FilesystemOnly, + Box::new(buffered), + ctx, + ) + }); + std::mem::drop(start.registration_future); + let subscriber_id = start.subscriber_id; + + // Batch A: crosses the bound on its own and is dispatched immediately (nothing + // else is in flight yet), taking 80ms to resolve. + let a1 = TargetFile::new(repo_path.join("a1.txt"), false); + let a2 = TargetFile::new(repo_path.join("a2.txt"), false); + let a3 = TargetFile::new(repo_path.join("a3.txt"), false); + let batch_a_update = RepositoryUpdate { + added: [a1.clone(), a2.clone(), a3.clone()].into(), + ..Default::default() + }; + repo_handle.update(&mut app, |repo, ctx| { + repo.notify_subscriber(subscriber_id, &batch_a_update, ctx); + }); + + // Batch B: crosses the bound while A is still in flight, so it becomes the + // initial `next_delivery` backlog via a plain assignment (nothing to merge with + // yet). + let move_target = TargetFile::new(repo_path.join("moved.txt"), false); + let w1 = TargetFile::new(repo_path.join("w1.txt"), false); + let w2 = TargetFile::new(repo_path.join("w2.txt"), false); + let batch_b_update = RepositoryUpdate { + added: [move_target.clone(), w1.clone(), w2.clone()].into(), + ..Default::default() + }; + repo_handle.update(&mut app, |repo, ctx| { + repo.notify_subscriber(subscriber_id, &batch_b_update, ctx); + }); + + // Batch C: also crosses the bound while A is still in flight, so `hand_off` + // merges it into the existing `next_delivery` (batch B) via + // `merge_repository_updates`. Its one move's source is `move_target`, which + // batch B recorded as `added` -- the exact "a move collapses into an existing + // add" condition a pre-existing bug (`return` instead of `continue` in the + // moves loop) would abandon everything after, dropping `extra_added` and + // `extra_deleted` below. A single move (not two) keeps this deterministic, + // independent of `HashMap` iteration order. + let moved_to = TargetFile::new(repo_path.join("moved_to.txt"), false); + let extra_added = TargetFile::new(repo_path.join("extra_added.txt"), false); + let extra_deleted = TargetFile::new(repo_path.join("extra_deleted.txt"), false); + let batch_c_update = RepositoryUpdate { + moved: [(moved_to.clone(), move_target.clone())].into(), + added: [extra_added.clone()].into(), + deleted: [extra_deleted.clone()].into(), + ..Default::default() + }; + repo_handle.update(&mut app, |repo, ctx| { + repo.notify_subscriber(subscriber_id, &batch_c_update, ctx); + }); + + // Batch A completes first. + let flushed_a = update_rx.next().await.expect("batch A"); + assert_eq!(flushed_a.added, [a1, a2, a3].into()); + + // The coalesced backlog (B merged with C) is delivered next, once A's in-flight + // delivery completes. + let flushed_backlog = update_rx.next().await.expect("coalesced backlog"); + assert!( + flushed_backlog.added.contains(&moved_to), + "the move's target should replace the source in `added`" + ); + assert!( + !flushed_backlog.added.contains(&move_target), + "the move's source should no longer be recorded as `added`" + ); + assert!(flushed_backlog.added.contains(&w1)); + assert!(flushed_backlog.added.contains(&w2)); + assert!( + flushed_backlog.added.contains(&extra_added), + "adds after the colliding move must still be applied" + ); + assert!( + flushed_backlog.deleted.contains(&extra_deleted), + "deletes after the colliding move must still be applied" + ); + }); + }, + ); +} + #[test] fn unchanged_tracked_remote_ref_does_not_notify_subscribers() { VirtualFS::test( diff --git a/crates/repo_metadata/src/watcher.rs b/crates/repo_metadata/src/watcher.rs index c65bb8cdff9..1c859172e58 100644 --- a/crates/repo_metadata/src/watcher.rs +++ b/crates/repo_metadata/src/watcher.rs @@ -767,6 +767,17 @@ impl RepositoryUpdate { pub fn contains_added_or_modified(&self, file: &TargetFile) -> bool { self.added.contains(file) || self.modified.contains(file) } + + /// Coalesces `incoming` into `self` as if `incoming` happened chronologically after `self`. + /// This is not a plain union of the two updates: it applies the same normalization a + /// debounce window would (moves before adds before modifies before deletes), so an add + /// followed by a delete of the same path cancels out, a move's target replaces its source + /// wherever the source was already tracked, and so on. Merging out of chronological order, + /// or merging two updates that weren't captured over a contiguous window, can produce a + /// result that doesn't correspond to any real sequence of filesystem events. + pub fn merge(&mut self, incoming: &RepositoryUpdate) { + crate::repository::merge_repository_updates(self, incoming); + } } /// An asynchronous task in a watched repository.