From cc50825a173715152155aad83f2baab361ead138 Mon Sep 17 00:00:00 2001 From: "warp-agent-staging[bot]" <240773466+warp-agent-staging[bot]@users.noreply.github.com> Date: Thu, 20 Aug 2026 03:33:49 +0000 Subject: [PATCH 1/4] Bound BufferingRepositorySubscriber's debounce buffer (APP-4829) Under sustained filesystem churn (a large git checkout, npm install, etc.), BufferingRepositorySubscriber::on_files_updated coalesced every incoming RepositoryUpdate into BufferState.pending with no maximum size or age. As long as updates kept arriving, the debounce flusher never observed a quiet period, so the pending HashSets/HashMap grew without bound, driving multi-gigabyte heap spikes (Sentry issue 7259255054). Force an early flush of the pending buffer once its combined entry count (added + modified + deleted + moved) reaches 10,000, independent of the debounce timer. The forced flush drains the same coalesced buffer merge_repository_updates already maintains and hands it to the inner subscriber exactly like the debounced flush does, so updates are only ever split into more, smaller batches -- never dropped or reordered. The debounce flusher, once spawned, keeps draining whatever accumulates between forced flushes, so pending is never left non-empty without a flusher scheduled against it. Adds a test-only with_max_pending_entries constructor so tests can exercise the bound without buffering 10,000 real entries, plus regression tests proving a flush is forced before the debounce timer fires and that no updates are lost across repeated forced flushes. --- crates/repo_metadata/src/repository.rs | 55 ++++++- crates/repo_metadata/src/repository_tests.rs | 146 ++++++++++++++++++- 2 files changed, 197 insertions(+), 4 deletions(-) diff --git a/crates/repo_metadata/src/repository.rs b/crates/repo_metadata/src/repository.rs index 104d88413a9..db4175517dd 100644 --- a/crates/repo_metadata/src/repository.rs +++ b/crates/repo_metadata/src/repository.rs @@ -641,11 +641,24 @@ 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; + +/// Total number of individual filesystem changes buffered in `update`. +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)] @@ -659,12 +672,33 @@ struct BufferState { 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) + } } impl RepositorySubscriber for BufferingRepositorySubscriber @@ -681,11 +715,11 @@ where fn on_files_updated( &mut self, - _repository: &Repository, + repository: &Repository, update: &RepositoryUpdate, ctx: &mut ModelContext, ) -> Pin + Send + 'static>> { - { + let forced_batch = { let mut st = self.state.lock().unwrap(); merge_repository_updates(&mut st.pending, update); st.version = st.version.wrapping_add(1); @@ -739,6 +773,23 @@ where }, )); } + + // Force a flush ahead of the debounce timer once the buffer grows past the bound, so + // memory stays bounded under sustained filesystem churn. This never drops or reorders + // anything: the drained batch already went through `merge_repository_updates`'s + // coalescing, and the flusher spawned above (or already running from an earlier call) + // keeps draining whatever accumulates afterward. + (pending_entry_count(&st.pending) >= self.max_pending_entries) + .then(|| std::mem::take(&mut st.pending)) + }; + + if let Some(batch) = forced_batch + && !batch.is_empty() + && let Ok(mut inner) = self.inner.lock() + { + let fut = inner.on_files_updated(repository, &batch, ctx); + // Drive the subscriber's async update to completion. + ctx.spawn(fut, |_, _, _| {}); } Box::pin(ready(())) diff --git a/crates/repo_metadata/src/repository_tests.rs b/crates/repo_metadata/src/repository_tests.rs index fdcf7900b06..f5ba787d0f3 100644 --- a/crates/repo_metadata/src/repository_tests.rs +++ b/crates/repo_metadata/src/repository_tests.rs @@ -11,8 +11,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; @@ -433,6 +433,148 @@ 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 forced_flushes_do_not_drop_updates_while_debounce_never_settles() { + VirtualFS::test( + "forced_flushes_do_not_drop_updates_while_debounce_never_settles", + |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; + 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 never fires during the test, so + // updates keep arriving into a buffer that never observes a quiet period. + 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; + + // Enough updates to force two flushes, with a partial batch left over that only + // the (unreached, 3600s) debounce timer would flush. + 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); + }); + } + + let mut flushed_files = std::collections::HashSet::new(); + for _ in 0..2 { + let flushed = update_rx.next().await.expect("forced flush"); + assert_eq!(flushed.added.len(), MAX_PENDING_ENTRIES); + for file in flushed.added { + assert!(flushed_files.insert(file), "duplicate flushed file"); + } + } + + assert_eq!(flushed_files.len(), MAX_PENDING_ENTRIES * 2); + for file in all_files.iter().take(MAX_PENDING_ENTRIES * 2) { + assert!(flushed_files.contains(file)); + } + + // The remaining updates stay buffered; without waiting out the debounce, no + // third flush should be observed. + futures::select! { + update = update_rx.next().fuse() => { + panic!("unexpected extra flush: {update:?}"); + } + _ = futures::FutureExt::fuse(Timer::after(Duration::from_millis(100))) => {} + } + }); + }, + ); +} + #[test] fn unchanged_tracked_remote_ref_does_not_notify_subscribers() { VirtualFS::test( From 5d590aad2c78163d37b742e67c013b5db6ba9fed Mon Sep 17 00:00:00 2001 From: "warp-agent-staging[bot]" <240773466+warp-agent-staging[bot]@users.noreply.github.com> Date: Thu, 20 Aug 2026 04:14:46 +0000 Subject: [PATCH 2/4] Bound the merge, not just the flush; serialize forced-flush delivery Review found two correctness gaps in the previous revision: 1. The threshold was checked only after merging the whole incoming RepositoryUpdate, so a single large filesystem-watcher batch (e.g. from a huge git checkout) could still balloon `pending` past the configured bound before anything drained it. 2. Forced-flush batches were dispatched fire-and-forget, so a fast producer could submit a new batch before the previous one's async delivery finished. The production consumer (RepoOutlines) silently drops updates that arrive while a previous recomputation is still in flight, so overlapping/rapid forced flushes could lose updates. Fix 1: merge each incoming update into `pending` one entry at a time, in the same phase order `merge_repository_updates` already applies (moves, adds, modifies, deletes, then flags), checking the bound after every entry. This keeps `pending` bounded even when a single incoming update by itself exceeds the limit. Fix 2: route every flush (forced or debounced) through a delivery queue drained strictly one batch at a time, so `inner.on_files_updated` never has two batches in flight and always applies them in order. Fix 2 also reaches into RepoOutlines (the only production consumer): it now retains and applies updates that arrive while a recomputation is Pending instead of dropping them, closing the remaining loss scenario a slow recomputation could otherwise still hit downstream of the channel. Also: rewrote the losslessness test to actually let the debounce elapse and account for every update (it previously left a partial batch unverified), added a regression test proving a single oversized update is split into bounded batches, added a slow-consumer regression test proving serialized, in-order, lossless delivery across forced and debounced flushes, and removed a redundant doc comment. --- app/src/ai/outline/native.rs | 57 ++++-- crates/repo_metadata/src/repository.rs | 185 +++++++++++++++---- crates/repo_metadata/src/repository_tests.rs | 178 ++++++++++++++++-- 3 files changed, 351 insertions(+), 69 deletions(-) diff --git a/app/src/ai/outline/native.rs b/app/src/ai/outline/native.rs index 60223ab5bdb..e21849070de 100644 --- a/app/src/ai/outline/native.rs +++ b/app/src/ai/outline/native.rs @@ -36,6 +36,10 @@ 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`). Drained, in arrival order, once that recomputation completes, so sustained + /// filesystem churn never loses an update just because it arrived mid-recomputation. + queued_updates: VecDeque, } pub enum RepoOutlinesEvent { @@ -128,6 +132,7 @@ impl RepoOutlines { repository, status: OutlineStatus::Pending, subscriber_id: None, + queued_updates: VecDeque::new(), }; self.outlines.insert(repo_path.clone(), outline_state); self.outline_queue.push_back(repo_path); @@ -379,36 +384,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 queued up while we were recomputing, so sustained + // filesystem churn under a slow recomputation never loses an update. + let next_update = state.queued_updates.pop_front(); + if let Some(next_update) = next_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; queue this update so it isn't lost, to + // be applied once that recomputation completes. + state.queued_updates.push_back(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"), } } } diff --git a/crates/repo_metadata/src/repository.rs b/crates/repo_metadata/src/repository.rs index db4175517dd..319324eb27f 100644 --- a/crates/repo_metadata/src/repository.rs +++ b/crates/repo_metadata/src/repository.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::future::Future; #[cfg(feature = "local_fs")] use std::path::{Component, Path, PathBuf}; @@ -648,7 +648,6 @@ fn merge_repository_updates(acc: &mut RepositoryUpdate, incoming: &RepositoryUpd /// bursts still coalesce through the debounce window instead of flushing on every batch. const DEFAULT_MAX_PENDING_ENTRIES: usize = 10_000; -/// Total number of individual filesystem changes buffered in `update`. fn pending_entry_count(update: &RepositoryUpdate) -> usize { update.added.len() + update.modified.len() + update.deleted.len() + update.moved.len() } @@ -668,6 +667,12 @@ struct BufferState { version: u64, /// Whether the background flusher loop is currently running. flush_handle: Option, + /// Batches that crossed `max_pending_entries` (or were handed off by the debounce flusher) + /// and are waiting to be delivered to `inner`. Drained strictly one at a time so `inner` + /// never sees overlapping or out-of-order calls to `on_files_updated`. + delivery_queue: VecDeque, + /// Whether a batch from `delivery_queue` is currently being delivered to `inner`. + delivering: bool, } impl BufferingRepositorySubscriber { @@ -701,6 +706,81 @@ impl BufferingRepositorySubscriber { } } +/// Merges `update` into `st.pending` one entry at a time, 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 in last), pushing the accumulated +/// batch onto `st.delivery_queue` every time the combined entry count reaches +/// `max_pending_entries`. `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 this +/// keeps `pending` bounded even when a single incoming `update` by itself exceeds the limit, +/// without changing the coalesced result. +fn merge_update_bounded( + st: &mut BufferState, + update: &RepositoryUpdate, + max_pending_entries: usize, +) { + fn flush_if_over_bound(st: &mut BufferState, max_pending_entries: usize) { + if pending_entry_count(&st.pending) >= max_pending_entries { + st.delivery_queue.push_back(std::mem::take(&mut st.pending)); + } + } + + for (to, from) in &update.moved { + merge_repository_updates( + &mut st.pending, + &RepositoryUpdate { + moved: [(to.clone(), from.clone())].into(), + ..Default::default() + }, + ); + flush_if_over_bound(st, max_pending_entries); + } + for p in &update.added { + merge_repository_updates( + &mut st.pending, + &RepositoryUpdate { + added: [p.clone()].into(), + ..Default::default() + }, + ); + flush_if_over_bound(st, max_pending_entries); + } + for p in &update.modified { + merge_repository_updates( + &mut st.pending, + &RepositoryUpdate { + modified: [p.clone()].into(), + ..Default::default() + }, + ); + flush_if_over_bound(st, max_pending_entries); + } + for p in &update.deleted { + merge_repository_updates( + &mut st.pending, + &RepositoryUpdate { + deleted: [p.clone()].into(), + ..Default::default() + }, + ); + flush_if_over_bound(st, max_pending_entries); + } + + // The flags never contribute to `pending_entry_count`, so folding them in can't itself + // cross the bound; merge them once, after every path-bearing entry above. + if update.commit_updated || update.index_lock_detected || update.remote_ref_updated { + merge_repository_updates( + &mut st.pending, + &RepositoryUpdate { + commit_updated: update.commit_updated, + index_lock_detected: update.index_lock_detected, + remote_ref_updated: update.remote_ref_updated, + ..Default::default() + }, + ); + } +} + impl RepositorySubscriber for BufferingRepositorySubscriber where S: RepositorySubscriber + Send + Sync + 'static, @@ -719,15 +799,16 @@ where update: &RepositoryUpdate, ctx: &mut ModelContext, ) -> Pin + Send + 'static>> { - let forced_batch = { + { let mut st = self.state.lock().unwrap(); - merge_repository_updates(&mut st.pending, update); + merge_update_bounded(&mut st, update, self.max_pending_entries); 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( @@ -736,7 +817,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; @@ -746,7 +827,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)) @@ -757,41 +838,39 @@ 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, |_, _, _| {}); - } + state_for_completion + .lock() + .unwrap() + .delivery_queue + .push_back(merged); + Self::advance_delivery( + inner_for_completion, + state_for_completion, + repo_model, + repo_ctx, + ); }, )); } - - // Force a flush ahead of the debounce timer once the buffer grows past the bound, so - // memory stays bounded under sustained filesystem churn. This never drops or reorders - // anything: the drained batch already went through `merge_repository_updates`'s - // coalescing, and the flusher spawned above (or already running from an earlier call) - // keeps draining whatever accumulates afterward. - (pending_entry_count(&st.pending) >= self.max_pending_entries) - .then(|| std::mem::take(&mut st.pending)) - }; - - if let Some(batch) = forced_batch - && !batch.is_empty() - && let Ok(mut inner) = self.inner.lock() - { - let fut = inner.on_files_updated(repository, &batch, ctx); - // Drive the subscriber's async update to completion. - ctx.spawn(fut, |_, _, _| {}); } + // Deliver whatever just crossed the bound (or is otherwise queued), one batch at a + // time. A no-op if a delivery is already in flight or nothing is queued. + Self::advance_delivery( + Arc::clone(&self.inner), + Arc::clone(&self.state), + repository, + ctx, + ); + Box::pin(ready(())) } @@ -805,6 +884,48 @@ where } } +impl BufferingRepositorySubscriber +where + S: RepositorySubscriber + Send + Sync + 'static, +{ + /// Delivers the next queued batch to `inner`, unless a delivery is already in flight. On + /// completion, recurses to pick up whatever queued up next, so `inner.on_files_updated` + /// never has two batches in flight at once and batches are always applied in the order they + /// were queued. + fn advance_delivery( + inner: Arc>, + state: Arc>, + repository: &Repository, + ctx: &mut ModelContext, + ) { + let batch = { + let mut st = state.lock().unwrap(); + if st.delivering { + return; + } + let Some(batch) = st.delivery_queue.pop_front() 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| { + state_for_next.lock().unwrap().delivering = false; + Self::advance_delivery(inner_for_next, state_for_next, repo_model, repo_ctx); + }); + } +} + #[cfg(test)] #[path = "repository_tests.rs"] mod tests; diff --git a/crates/repo_metadata/src/repository_tests.rs b/crates/repo_metadata/src/repository_tests.rs index f5ba787d0f3..8ef2bab4baa 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; @@ -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, @@ -495,9 +538,9 @@ fn forced_flush_drains_pending_before_debounce_timer_fires() { } #[test] -fn forced_flushes_do_not_drop_updates_while_debounce_never_settles() { +fn single_incoming_update_exceeding_the_bound_is_split_into_bounded_batches() { VirtualFS::test( - "forced_flushes_do_not_drop_updates_while_debounce_never_settles", + "single_incoming_update_exceeding_the_bound_is_split_into_bounded_batches", |dirs, mut vfs| { vfs.mkdir("repo"); let repo_path = dirs.tests().join("repo"); @@ -513,13 +556,14 @@ fn forced_flushes_do_not_drop_updates_while_debounce_never_settles() { }) .unwrap(); - const MAX_PENDING_ENTRIES: usize = 5; + 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 never fires during the test, so - // updates keep arriving into a buffer that never observes a quiet period. + // 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, ); @@ -532,8 +576,103 @@ fn forced_flushes_do_not_drop_updates_while_debounce_never_settles() { std::mem::drop(start.registration_future); let subscriber_id = start.subscriber_id; - // Enough updates to force two flushes, with a partial batch left over that only - // the (unreached, 3600s) debounce timer would flush. + // 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 mut seen = HashSet::new(); + let mut batch_count = 0; + while seen.len() < total_files { + let flushed = update_rx.next().await.expect("bounded batch"); + assert!( + flushed.added.len() <= MAX_PENDING_ENTRIES, + "batch of {} entries exceeded the configured bound of {MAX_PENDING_ENTRIES}", + flushed.added.len() + ); + for file in flushed.added { + assert!(seen.insert(file), "duplicate file delivered across batches"); + } + batch_count += 1; + } + + assert_eq!(seen.len(), total_files); + for file in &all_files { + assert!(seen.contains(file)); + } + assert!( + batch_count > 1, + "a single update far exceeding the bound should be split into multiple batches" + ); + }); + }, + ); +} + +#[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)) @@ -548,27 +687,30 @@ fn forced_flushes_do_not_drop_updates_while_debounce_never_settles() { }); } - let mut flushed_files = std::collections::HashSet::new(); - for _ in 0..2 { - let flushed = update_rx.next().await.expect("forced flush"); - assert_eq!(flushed.added.len(), MAX_PENDING_ENTRIES); + // Exactly three batches account for every update: two forced (5 each) and one + // debounced remainder (3). `SlowRecordingSubscriber` itself asserts that no two + // of them are ever delivered concurrently. + let mut seen = HashSet::new(); + for _ in 0..3 { + let flushed = update_rx.next().await.expect("flush"); + assert!(!flushed.added.is_empty()); + assert!(flushed.added.len() <= MAX_PENDING_ENTRIES); for file in flushed.added { - assert!(flushed_files.insert(file), "duplicate flushed file"); + assert!(seen.insert(file), "duplicate file delivered across batches"); } } - assert_eq!(flushed_files.len(), MAX_PENDING_ENTRIES * 2); - for file in all_files.iter().take(MAX_PENDING_ENTRIES * 2) { - assert!(flushed_files.contains(file)); + assert_eq!(seen.len(), total_updates); + for file in &all_files { + assert!(seen.contains(file)); } - // The remaining updates stay buffered; without waiting out the debounce, no - // third flush should be observed. + // 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(100))) => {} + _ = futures::FutureExt::fuse(Timer::after(Duration::from_millis(300))) => {} } }); }, From 045016e99c5d9420ccd41115b8801958758dbf91 Mon Sep 17 00:00:00 2001 From: "warp-agent-staging[bot]" <240773466+warp-agent-staging[bot]@users.noreply.github.com> Date: Thu, 20 Aug 2026 06:22:51 +0000 Subject: [PATCH 3/4] Coalesce delivery backlog instead of queueing; fix unsubscribe teardown Second review round found that the previous revision fixed the single-input peak and correctly serialized delivery, but relocated the unbounded accumulation into two new queues (BufferState's delivery_queue and OutlineState's queued_updates), each storing one full RepositoryUpdate per batch that formed while the consumer was busy -- worse than the original bug, since the same path could now be stored once per batch. Replace both queues with a single coalescing accumulator: - BufferState::next_delivery: Option, merged via merge_repository_updates instead of pushed. on_files_updated now interleaves per-entry merging with immediate delivery attempts: the first batch to cross the bound while nothing is in flight is dispatched immediately (bounded); anything that forms while a delivery is already running is merged into next_delivery. - OutlineState::pending_update: RepositoryUpdate, merged via a new public RepositoryUpdate::merge() (repo_metadata crate) instead of queued, and applied as one follow-up recomputation. This bounds the backlog to O(distinct un-applied paths) instead of O(events x paths-per-event) -- the minimum retainable without dropping anything, though not an absolute cap under a sustained, never-catching-up consumer. The PR description is updated to state that honestly rather than overclaiming a hard cap. Also fixes a real leak: on_unsubscribe only aborted the debounce flusher, leaving any pending/in-flight delivery live and able to call the (already unsubscribed) inner subscriber. It now marks the buffer inactive and clears pending/next_delivery; advance_delivery's completion callback becomes a no-op once inactive. Adds: - unsubscribe_cancels_pending_delivery_and_releases_the_backlog: proves an in-flight delivery still completes but a backlogged one is released, not delivered late. - concurrent_update_while_pending_is_merged_not_dropped_or_double_recomputed (app/src/ai/outline/native_tests.rs): drives a real RepoOutlines recomputation via ai::index::build_outline against a temp dir, submits a second update while the first is Pending, and asserts it merges into the accumulator instead of being dropped or triggering a second overlapping recomputation. Relaxes the two existing repo_metadata tests that asserted every batch stays within the bound: with coalescing, only the very first batch formed while idle is guaranteed bounded; everything else is guaranteed lossless and duplicate-free but may exceed the bound in size when the consumer is slow. Both now loop until every fed update is accounted for instead of asserting a fixed batch count/size. --- app/src/ai/outline/native.rs | 26 ++- app/src/ai/outline/native_tests.rs | 117 ++++++++++ crates/repo_metadata/src/repository.rs | 216 ++++++++++--------- crates/repo_metadata/src/repository_tests.rs | 122 +++++++++-- crates/repo_metadata/src/watcher.rs | 7 + 5 files changed, 363 insertions(+), 125 deletions(-) create mode 100644 app/src/ai/outline/native_tests.rs diff --git a/app/src/ai/outline/native.rs b/app/src/ai/outline/native.rs index e21849070de..8fe7023adf8 100644 --- a/app/src/ai/outline/native.rs +++ b/app/src/ai/outline/native.rs @@ -37,9 +37,11 @@ struct OutlineState { /// Subscriber ID for repository updates (if watching). subscriber_id: Option, /// Updates that arrived while a recomputation was already in flight (`status` was - /// `Pending`). Drained, in arrival order, once that recomputation completes, so sustained - /// filesystem churn never loses an update just because it arrived mid-recomputation. - queued_updates: VecDeque, + /// `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 { @@ -132,7 +134,7 @@ impl RepoOutlines { repository, status: OutlineStatus::Pending, subscriber_id: None, - queued_updates: VecDeque::new(), + pending_update: RepositoryUpdate::default(), }; self.outlines.insert(repo_path.clone(), outline_state); self.outline_queue.push_back(repo_path); @@ -411,19 +413,19 @@ impl RepoOutlines { state.status = OutlineStatus::Complete(outline); ctx.emit(RepoOutlinesEvent::OutlinesUpdated(repo_path.clone())); - // Apply whatever queued up while we were recomputing, so sustained + // Apply whatever coalesced in while we were recomputing, so sustained // filesystem churn under a slow recomputation never loses an update. - let next_update = state.queued_updates.pop_front(); - if let Some(next_update) = next_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); } }, ); } OutlineStatus::Pending => { - // A recomputation is already in flight; queue this update so it isn't lost, to - // be applied once that recomputation completes. - state.queued_updates.push_back(update); + // 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"); @@ -465,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 319324eb27f..3e10cc2c332 100644 --- a/crates/repo_metadata/src/repository.rs +++ b/crates/repo_metadata/src/repository.rs @@ -1,4 +1,4 @@ -use std::collections::{HashMap, VecDeque}; +use std::collections::HashMap; use std::future::Future; #[cfg(feature = "local_fs")] use std::path::{Component, Path, PathBuf}; @@ -575,7 +575,7 @@ 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) { @@ -660,19 +660,39 @@ pub struct BufferingRepositorySubscriber { 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, - /// Batches that crossed `max_pending_entries` (or were handed off by the debounce flusher) - /// and are waiting to be delivered to `inner`. Drained strictly one at a time so `inner` - /// never sees overlapping or out-of-order calls to `on_files_updated`. - delivery_queue: VecDeque, - /// Whether a batch from `delivery_queue` is currently being delivered to `inner`. + /// 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 { @@ -706,79 +726,43 @@ impl BufferingRepositorySubscriber { } } -/// Merges `update` into `st.pending` one entry at a time, in the same phase order +/// 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 in last), pushing the accumulated -/// batch onto `st.delivery_queue` every time the combined entry count reaches -/// `max_pending_entries`. `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 this -/// keeps `pending` bounded even when a single incoming `update` by itself exceeds the limit, -/// without changing the coalesced result. -fn merge_update_bounded( - st: &mut BufferState, - update: &RepositoryUpdate, - max_pending_entries: usize, -) { - fn flush_if_over_bound(st: &mut BufferState, max_pending_entries: usize) { - if pending_entry_count(&st.pending) >= max_pending_entries { - st.delivery_queue.push_back(std::mem::take(&mut st.pending)); - } - } +/// 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(); - for (to, from) in &update.moved { - merge_repository_updates( - &mut st.pending, - &RepositoryUpdate { - moved: [(to.clone(), from.clone())].into(), - ..Default::default() - }, - ); - flush_if_over_bound(st, max_pending_entries); - } - for p in &update.added { - merge_repository_updates( - &mut st.pending, - &RepositoryUpdate { - added: [p.clone()].into(), - ..Default::default() - }, - ); - flush_if_over_bound(st, max_pending_entries); - } - for p in &update.modified { - merge_repository_updates( - &mut st.pending, - &RepositoryUpdate { - modified: [p.clone()].into(), - ..Default::default() - }, - ); - flush_if_over_bound(st, max_pending_entries); - } - for p in &update.deleted { - merge_repository_updates( - &mut st.pending, - &RepositoryUpdate { - deleted: [p.clone()].into(), - ..Default::default() - }, - ); - flush_if_over_bound(st, max_pending_entries); - } - - // The flags never contribute to `pending_entry_count`, so folding them in can't itself - // cross the bound; merge them once, after every path-bearing entry above. - if update.commit_updated || update.index_lock_detected || update.remote_ref_updated { - merge_repository_updates( - &mut st.pending, - &RepositoryUpdate { - commit_updated: update.commit_updated, - index_lock_detected: update.index_lock_detected, - remote_ref_updated: update.remote_ref_updated, - ..Default::default() - }, - ); - } + moves + .chain(adds) + .chain(modifies) + .chain(deletes) + .chain(flags) } impl RepositorySubscriber for BufferingRepositorySubscriber @@ -801,7 +785,6 @@ where ) -> Pin + Send + 'static>> { { let mut st = self.state.lock().unwrap(); - merge_update_bounded(&mut st, update, self.max_pending_entries); st.version = st.version.wrapping_add(1); // Start a single background flusher if it's not already running. @@ -846,11 +829,7 @@ where if merged.is_empty() { return; } - state_for_completion - .lock() - .unwrap() - .delivery_queue - .push_back(merged); + Self::hand_off(&state_for_completion, merged); Self::advance_delivery( inner_for_completion, state_for_completion, @@ -862,14 +841,29 @@ where } } - // Deliver whatever just crossed the bound (or is otherwise queued), one batch at a - // time. A no-op if a delivery is already in flight or nothing is queued. - Self::advance_delivery( - Arc::clone(&self.inner), - Arc::clone(&self.state), - repository, - 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(())) } @@ -881,6 +875,12 @@ 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(); } } @@ -888,10 +888,19 @@ impl BufferingRepositorySubscriber where S: RepositorySubscriber + Send + Sync + 'static, { - /// Delivers the next queued batch to `inner`, unless a delivery is already in flight. On - /// completion, recurses to pick up whatever queued up next, so `inner.on_files_updated` - /// never has two batches in flight at once and batches are always applied in the order they - /// were queued. + /// 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>, @@ -900,10 +909,10 @@ where ) { let batch = { let mut st = state.lock().unwrap(); - if st.delivering { + if !st.active || st.delivering { return; } - let Some(batch) = st.delivery_queue.pop_front() else { + let Some(batch) = st.next_delivery.take() else { return; }; st.delivering = true; @@ -920,8 +929,13 @@ where let inner_for_next = Arc::clone(&inner); let state_for_next = Arc::clone(&state); ctx.spawn(fut, move |repo_model, (), repo_ctx| { - state_for_next.lock().unwrap().delivering = false; - Self::advance_delivery(inner_for_next, state_for_next, 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 8ef2bab4baa..e573ab1be49 100644 --- a/crates/repo_metadata/src/repository_tests.rs +++ b/crates/repo_metadata/src/repository_tests.rs @@ -538,9 +538,9 @@ fn forced_flush_drains_pending_before_debounce_timer_fires() { } #[test] -fn single_incoming_update_exceeding_the_bound_is_split_into_bounded_batches() { +fn single_incoming_update_exceeding_the_bound_is_delivered_without_loss() { VirtualFS::test( - "single_incoming_update_exceeding_the_bound_is_split_into_bounded_batches", + "single_incoming_update_exceeding_the_bound_is_delivered_without_loss", |dirs, mut vfs| { vfs.mkdir("repo"); let repo_path = dirs.tests().join("repo"); @@ -596,12 +596,12 @@ fn single_incoming_update_exceeding_the_bound_is_split_into_bounded_batches() { let mut seen = HashSet::new(); let mut batch_count = 0; while seen.len() < total_files { - let flushed = update_rx.next().await.expect("bounded batch"); - assert!( - flushed.added.len() <= MAX_PENDING_ENTRIES, - "batch of {} entries exceeded the configured bound of {MAX_PENDING_ENTRIES}", - flushed.added.len() - ); + let flushed = update_rx.next().await.expect("batch"); + if batch_count == 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); + } for file in flushed.added { assert!(seen.insert(file), "duplicate file delivered across batches"); } @@ -612,9 +612,12 @@ fn single_incoming_update_exceeding_the_bound_is_split_into_bounded_batches() { 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 be split into multiple batches" + "a single update far exceeding the bound should still be split into more than one batch" ); }); }, @@ -687,14 +690,16 @@ fn forced_and_debounced_flushes_apply_every_update_exactly_once_with_a_slow_cons }); } - // Exactly three batches account for every update: two forced (5 each) and one - // debounced remainder (3). `SlowRecordingSubscriber` itself asserts that no two - // of them are ever delivered concurrently. + // 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 mut seen = HashSet::new(); - for _ in 0..3 { + while seen.len() < total_updates { let flushed = update_rx.next().await.expect("flush"); assert!(!flushed.added.is_empty()); - assert!(flushed.added.len() <= MAX_PENDING_ENTRIES); for file in flushed.added { assert!(seen.insert(file), "duplicate file delivered across batches"); } @@ -717,6 +722,95 @@ fn forced_and_debounced_flushes_apply_every_update_exactly_once_with_a_slow_cons ); } +#[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 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..14e79a5c424 100644 --- a/crates/repo_metadata/src/watcher.rs +++ b/crates/repo_metadata/src/watcher.rs @@ -767,6 +767,13 @@ impl RepositoryUpdate { pub fn contains_added_or_modified(&self, file: &TargetFile) -> bool { self.added.contains(file) || self.modified.contains(file) } + + /// Coalesces `incoming` into `self`, using the same normalization rules `Repository`'s + /// internal debounce buffer applies (moves, then adds, then modifies, then deletes; + /// cancelling adds/deletes of the same path within the merged window, and so on). + pub fn merge(&mut self, incoming: &RepositoryUpdate) { + crate::repository::merge_repository_updates(self, incoming); + } } /// An asynchronous task in a watched repository. From 5e9f897d50061ac172bc24665c8fdddf3dc426b3 Mon Sep 17 00:00:00 2001 From: "warp-agent-staging[bot]" <240773466+warp-agent-staging[bot]@users.noreply.github.com> Date: Thu, 20 Aug 2026 06:59:31 +0000 Subject: [PATCH 4/4] Fix return-vs-continue bug in merge_repository_updates moves loop Third review round: the coalescing accumulators, unsubscribe lifecycle, public RepositoryUpdate::merge wrapper, and the outline test all held up. One item reversed a prior scope call: the moves loop's `return` instead of `continue` (when a move's source collapses into an already-recorded add or modify) was previously judged a narrow, pre-existing issue unrelated to this change. hand_off's coalescing now reachably calls merge_repository_updates with real, multi-entry batches on exactly the path sustained churn exercises (merging a bounded batch into the backlog while a delivery is in flight), so the bug is now live on the path this PR's losslessness claim depends on. Fixed by changing both `return`s to `continue`s. Added: - Two direct unit tests (merge_repository_updates_applies_every_entry _when_a_move_collapses_into_added / _into_modified) pinning both branches. Verified both fail without the fix and pass with it. - coalescing_a_move_that_collapses_into_an_existing_add_still_applies _later_entries: an end-to-end regression through the actual reachable path (accumulate a backlog while a delivery is in flight, then coalesce a batch whose move collides with the backlog). Verified it fails without the fix and passes with it. - collect_until_seen test helper: the two loop-until-every-update- seen tests now use a deterministic receive timeout instead of an unbounded await, so a real regression fails fast with the missing paths and batch count instead of hanging until nextest's slow-timeout. - Expanded RepositoryUpdate::merge's doc comment: it's not a generic union, it applies debounce-style normalization, and merging out of chronological order can produce a result that doesn't correspond to any real sequence of filesystem events. Re-ran cargo nextest run -p warp (full app unit test suite, excluding integration tests) explicitly to confirm no OOM: 6479 tests run in 80s, 6476 passed. The 3 failures (server::server_api::ai::tests:: ambient_agent_headers_for_task_overrides_existing_cloud_agent_header, terminal::input::tests::test_histignorespace_support_in_zsh, terminal::input::decorations::tests:: test_decorations_with_multibyte_chars) are pre-existing and unrelated: confirmed they fail identically with this branch's repo_metadata/outline changes reverted, in files this PR never touches. --- crates/repo_metadata/src/repository.rs | 4 +- crates/repo_metadata/src/repository_tests.rs | 299 +++++++++++++++++-- crates/repo_metadata/src/watcher.rs | 10 +- 3 files changed, 286 insertions(+), 27 deletions(-) diff --git a/crates/repo_metadata/src/repository.rs b/crates/repo_metadata/src/repository.rs index 3e10cc2c332..f2f6736c20e 100644 --- a/crates/repo_metadata/src/repository.rs +++ b/crates/repo_metadata/src/repository.rs @@ -580,11 +580,11 @@ pub(crate) fn merge_repository_updates(acc: &mut RepositoryUpdate, incoming: &Re 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 diff --git a/crates/repo_metadata/src/repository_tests.rs b/crates/repo_metadata/src/repository_tests.rs index e573ab1be49..66db1f37ada 100644 --- a/crates/repo_metadata/src/repository_tests.rs +++ b/crates/repo_metadata/src/repository_tests.rs @@ -104,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!( @@ -241,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( @@ -593,20 +730,20 @@ fn single_incoming_update_exceeding_the_bound_is_delivered_without_loss() { repo.notify_subscriber(subscriber_id, &huge_update, ctx); }); - let mut seen = HashSet::new(); - let mut batch_count = 0; - while seen.len() < total_files { - let flushed = update_rx.next().await.expect("batch"); - if batch_count == 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); - } - for file in flushed.added { - assert!(seen.insert(file), "duplicate file delivered across batches"); - } - batch_count += 1; - } + 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 { @@ -696,14 +833,13 @@ fn forced_and_debounced_flushes_apply_every_update_exactly_once_with_a_slow_cons // a separate batch), but nothing may ever be dropped or duplicated. // `SlowRecordingSubscriber` itself asserts that no two batches are ever // delivered concurrently. - let mut seen = HashSet::new(); - while seen.len() < total_updates { - let flushed = update_rx.next().await.expect("flush"); - assert!(!flushed.added.is_empty()); - for file in flushed.added { - assert!(seen.insert(file), "duplicate file delivered across batches"); - } - } + 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 { @@ -811,6 +947,125 @@ fn unsubscribe_cancels_pending_delivery_and_releases_the_backlog() { ); } +#[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 14e79a5c424..1c859172e58 100644 --- a/crates/repo_metadata/src/watcher.rs +++ b/crates/repo_metadata/src/watcher.rs @@ -768,9 +768,13 @@ impl RepositoryUpdate { self.added.contains(file) || self.modified.contains(file) } - /// Coalesces `incoming` into `self`, using the same normalization rules `Repository`'s - /// internal debounce buffer applies (moves, then adds, then modifies, then deletes; - /// cancelling adds/deletes of the same path within the merged window, and so on). + /// 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); }