Bound BufferingRepositorySubscriber's debounce buffer (APP-4829)

app/src/ai/outline/native.rs

@@ -36,6 +36,12 @@ struct OutlineState {
     status: OutlineStatus,
     /// Subscriber ID for repository updates (if watching).
     subscriber_id: Option<SubscriberId>,
+    /// 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;

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);
+        });
+    });
+}