Skip to repository content1090 lines · 44.0 KB · rust
tenant.openagents/omega
No repository description is available.
OpenAgents Git authority 2026-07-28T03:57:50.339Z Public web read
NIP-34 coordinate
30617:7649603503856e5148d571eac2766b288a8ff1e9e35d380337a1d2b0015b4f92:omegaMaintainersHidden in public view
References2 branches · 1 tag
Read-only clone
git clone https://openagents.com/git/tenant.openagents/omega.gitBrowse files
project_search.rs
1use std::{
2 cell::LazyCell,
3 collections::BTreeSet,
4 io::{BufRead, BufReader},
5 ops::Range,
6 path::{Path, PathBuf},
7 pin::pin,
8 sync::Arc,
9 time::Duration,
10};
11
12use anyhow::Context;
13use async_channel::{Receiver, Sender, bounded, unbounded};
14use collections::HashSet;
15use fs::Fs;
16use futures::FutureExt as _;
17use futures::{SinkExt, StreamExt, select_biased, stream::FuturesOrdered};
18use gpui::{App, AppContext, AsyncApp, BackgroundExecutor, Entity, Priority, Task};
19use language::{Buffer, BufferSnapshot, Point};
20use parking_lot::Mutex;
21use postage::oneshot;
22use rpc::{AnyProtoClient, proto};
23
24use util::{ResultExt, maybe, paths::compare_rel_paths, rel_path::RelPath};
25use worktree::{Entry, ProjectEntryId, Snapshot, Worktree, WorktreeSettings};
26
27use crate::{
28 Project, ProjectItem, ProjectPath, RemotelyCreatedModels,
29 buffer_store::BufferStore,
30 search::{LineHint, SearchQuery, SearchResult},
31 worktree_store::WorktreeStore,
32};
33
34pub struct Search {
35 buffer_store: Entity<BufferStore>,
36 worktree_store: Entity<WorktreeStore>,
37 limit: usize,
38 kind: SearchKind,
39}
40
41/// Represents search setup, before it is actually kicked off with Search::into_results
42enum SearchKind {
43 /// Search for candidates by inspecting file contents on file system, avoiding loading the buffer unless we know that a given file contains a match.
44 Local {
45 fs: Arc<dyn Fs>,
46 worktrees: Vec<Entity<Worktree>>,
47 },
48 /// Query remote host for candidates. As of writing, the host runs a local search in "buffers with matches only" mode.
49 Remote {
50 client: AnyProtoClient,
51 remote_id: u64,
52 models: Arc<Mutex<RemotelyCreatedModels>>,
53 },
54 /// Run search against a known set of candidates. Even when working with a remote host, this won't round-trip to host.
55 OpenBuffersOnly,
56}
57
58/// Represents results of project search and allows one to either obtain match positions OR
59/// just the handles to buffers that may match the search. Grabbing the handles is cheaper than obtaining full match positions, because in that case we'll look for
60/// at most one match in each file.
61#[must_use]
62pub struct SearchResultsHandle {
63 results: Receiver<SearchResult>,
64 matching_buffers: Receiver<(Entity<Buffer>, LineHint)>,
65 trigger_search: Box<dyn FnOnce(&mut App) -> Task<()> + Send + Sync>,
66}
67
68pub struct SearchResults<T> {
69 pub task_handle: Task<()>,
70 pub rx: Receiver<T>,
71}
72impl SearchResultsHandle {
73 pub fn results(self, cx: &mut App) -> SearchResults<SearchResult> {
74 SearchResults {
75 task_handle: (self.trigger_search)(cx),
76 rx: self.results,
77 }
78 }
79 pub fn matching_buffers(self, cx: &mut App) -> SearchResults<(Entity<Buffer>, LineHint)> {
80 SearchResults {
81 task_handle: (self.trigger_search)(cx),
82 rx: self.matching_buffers,
83 }
84 }
85}
86
87#[derive(Clone)]
88enum FindSearchCandidates {
89 Local {
90 fs: Arc<dyn Fs>,
91 /// Start off with all paths in project and filter them based on:
92 /// - Include filters
93 /// - Exclude filters
94 /// - Only open buffers
95 /// - Scan ignored files
96 /// Put another way: filter out files that can't match (without looking at file contents)
97 input_paths_rx: Receiver<InputPath>,
98 /// After that, if the buffer is not yet loaded, we'll figure out if it contains at least one match
99 /// based on disk contents of a buffer. This step is not performed for buffers we already have in memory.
100 confirm_contents_will_match_tx: Sender<MatchingEntry>,
101 confirm_contents_will_match_rx: Receiver<MatchingEntry>,
102 },
103 Remote,
104 OpenBuffersOnly,
105}
106
107impl Search {
108 pub fn local(
109 fs: Arc<dyn Fs>,
110 buffer_store: Entity<BufferStore>,
111 worktree_store: Entity<WorktreeStore>,
112 limit: usize,
113 cx: &mut App,
114 ) -> Self {
115 let worktrees = worktree_store.read(cx).visible_worktrees(cx).collect();
116 Self {
117 kind: SearchKind::Local { fs, worktrees },
118 buffer_store,
119 worktree_store,
120 limit,
121 }
122 }
123
124 pub(crate) fn remote(
125 buffer_store: Entity<BufferStore>,
126 worktree_store: Entity<WorktreeStore>,
127 limit: usize,
128 client_state: (AnyProtoClient, u64, Arc<Mutex<RemotelyCreatedModels>>),
129 ) -> Self {
130 Self {
131 kind: SearchKind::Remote {
132 client: client_state.0,
133 remote_id: client_state.1,
134 models: client_state.2,
135 },
136 buffer_store,
137 worktree_store,
138 limit,
139 }
140 }
141 pub(crate) fn open_buffers_only(
142 buffer_store: Entity<BufferStore>,
143 worktree_store: Entity<WorktreeStore>,
144 limit: usize,
145 ) -> Self {
146 Self {
147 kind: SearchKind::OpenBuffersOnly,
148 buffer_store,
149 worktree_store,
150 limit,
151 }
152 }
153
154 pub(crate) const MAX_SEARCH_RESULT_FILES: usize = 5_000;
155 pub const MAX_SEARCH_RESULT_RANGES: usize = 10_000;
156 /// Prepares a project search run. The resulting [`SearchResultsHandle`] has to be used to specify whether you're interested in matching buffers
157 /// or full search results.
158 pub fn into_handle(mut self, query: SearchQuery, cx: &mut App) -> SearchResultsHandle {
159 let mut open_buffers = HashSet::default();
160 let mut unnamed_buffers = Vec::new();
161 const MAX_CONCURRENT_BUFFER_OPENS: usize = 64;
162 let buffers = self.buffer_store.read(cx);
163 for handle in buffers.buffers() {
164 let buffer = handle.read(cx);
165 if !buffers.is_searchable(&buffer.remote_id()) {
166 continue;
167 } else if buffer
168 .file()
169 .is_some_and(|file| file.disk_state().is_deleted())
170 {
171 continue;
172 } else if let Some(entry_id) = buffer.entry_id(cx) {
173 open_buffers.insert(entry_id);
174 } else {
175 self.limit = self.limit.saturating_sub(1);
176 unnamed_buffers.push(handle)
177 };
178 }
179 let open_buffers = Arc::new(open_buffers);
180 let executor = cx.background_executor().clone();
181 let (tx, rx) = unbounded();
182 let (grab_buffer_snapshot_tx, grab_buffer_snapshot_rx) =
183 unbounded::<(Entity<Buffer>, LineHint)>();
184 let matching_buffers = grab_buffer_snapshot_rx.clone();
185 let trigger_search = Box::new(move |cx: &mut App| {
186 cx.spawn(async move |cx| {
187 for buffer in unnamed_buffers {
188 _ = grab_buffer_snapshot_tx
189 .send((buffer, LineHint::default()))
190 .await;
191 }
192
193 let (find_all_matches_tx, find_all_matches_rx) =
194 bounded(MAX_CONCURRENT_BUFFER_OPENS);
195 let query = Arc::new(query);
196 let (candidate_searcher, tasks) = match self.kind {
197 SearchKind::OpenBuffersOnly => {
198 let open_buffers = cx.update(|cx| self.all_loaded_buffers(&query, cx));
199 let fill_requests = cx
200 .background_spawn(async move {
201 for buffer in open_buffers {
202 if let Err(_) = grab_buffer_snapshot_tx
203 .send((buffer, LineHint::default()))
204 .await
205 {
206 return;
207 }
208 }
209 })
210 .boxed_local();
211 (FindSearchCandidates::OpenBuffersOnly, vec![fill_requests])
212 }
213 SearchKind::Local {
214 fs,
215 ref mut worktrees,
216 } => {
217 let (get_buffer_for_full_scan_tx, get_buffer_for_full_scan_rx) =
218 unbounded();
219 let (confirm_contents_will_match_tx, confirm_contents_will_match_rx) =
220 bounded(64);
221 let (sorted_search_results_tx, sorted_search_results_rx) = unbounded();
222
223 let (input_paths_tx, input_paths_rx) = unbounded();
224 let tasks = vec![
225 cx.spawn(Self::provide_search_paths(
226 std::mem::take(worktrees),
227 query.clone(),
228 input_paths_tx,
229 sorted_search_results_tx,
230 tx.clone(),
231 ))
232 .boxed_local(),
233 Self::open_buffers(
234 self.buffer_store,
235 get_buffer_for_full_scan_rx,
236 grab_buffer_snapshot_tx,
237 cx.clone(),
238 )
239 .boxed_local(),
240 cx.background_spawn(Self::maintain_sorted_search_results(
241 sorted_search_results_rx,
242 get_buffer_for_full_scan_tx,
243 self.limit,
244 ))
245 .boxed_local(),
246 ];
247 (
248 FindSearchCandidates::Local {
249 fs,
250 confirm_contents_will_match_tx,
251 confirm_contents_will_match_rx,
252 input_paths_rx,
253 },
254 tasks,
255 )
256 }
257 SearchKind::Remote {
258 client,
259 remote_id,
260 models,
261 } => {
262 let (handle, rx) = self
263 .buffer_store
264 .update(cx, |this, _| this.register_project_search_result_handle());
265
266 let cancel_ongoing_search = util::defer({
267 let client = client.clone();
268 move || {
269 _ = client.send(proto::FindSearchCandidatesCancelled {
270 project_id: remote_id,
271 handle,
272 });
273 }
274 });
275 let request = client.request(proto::FindSearchCandidates {
276 project_id: remote_id,
277 query: Some(query.to_proto()),
278 limit: self.limit as _,
279 handle,
280 });
281
282 let buffer_store = self.buffer_store;
283 let guard = cx.update(|cx| {
284 Project::retain_remotely_created_models_impl(
285 &models,
286 &buffer_store,
287 &self.worktree_store,
288 cx,
289 )
290 });
291
292 let issue_remote_buffers_request = cx
293 .spawn(async move |cx| {
294 let _ = maybe!(async move {
295 request.await?;
296
297 let (buffer_tx, buffer_rx) = bounded(24);
298
299 let wait_for_remote_buffers = cx.spawn(async move |cx| {
300 while let Ok(buffer_id) = rx.recv().await {
301 let buffer =
302 buffer_store.update(cx, |buffer_store, cx| {
303 buffer_store
304 .wait_for_remote_buffer(buffer_id, cx)
305 });
306 buffer_tx.send(buffer).await?;
307 }
308 anyhow::Ok(())
309 });
310
311 let forward_buffers = cx.background_spawn(async move {
312 while let Ok(buffer) = buffer_rx.recv().await {
313 let _ = grab_buffer_snapshot_tx
314 .send((buffer.await?, LineHint::default()))
315 .await;
316 }
317 anyhow::Ok(())
318 });
319 let (left, right) = futures::future::join(
320 wait_for_remote_buffers,
321 forward_buffers,
322 )
323 .await;
324 left?;
325 right?;
326
327 drop(guard);
328 cancel_ongoing_search.abort();
329 anyhow::Ok(())
330 })
331 .await
332 .log_err();
333 })
334 .boxed_local();
335 (
336 FindSearchCandidates::Remote,
337 vec![issue_remote_buffers_request],
338 )
339 }
340 };
341
342 let should_find_all_matches = !tx.is_closed();
343
344 let _executor = executor.clone();
345 let worker_pool = executor.spawn(async move {
346 let num_cpus = _executor.num_cpus();
347
348 assert!(num_cpus > 0);
349 _executor
350 .scoped(|scope| {
351 let worker_count = (num_cpus - 1).max(1);
352 for _ in 0..worker_count {
353 let worker = Worker {
354 query: query.clone(),
355 open_buffers: open_buffers.clone(),
356 candidates: candidate_searcher.clone(),
357 find_all_matches_rx: find_all_matches_rx.clone(),
358 };
359 scope.spawn(worker.run());
360 }
361
362 drop(find_all_matches_rx);
363 drop(candidate_searcher);
364 })
365 .await;
366 });
367
368 let (sorted_matches_tx, sorted_matches_rx) = unbounded();
369 // The caller of `into_handle` decides whether they're interested in all matches (files that matched + all matching ranges) or
370 // just the files. *They are using the same stream as the guts of the project search do*.
371 // This means that we cannot grab values off of that stream unless it's strictly needed for making a progress in project search.
372 //
373 // Grabbing buffer snapshots is only necessary when we're looking for all matches. If the caller decided that they're not interested
374 // in all matches, running that task unconditionally would hinder caller's ability to observe all matching file paths.
375 let buffer_snapshots = if should_find_all_matches {
376 Some(
377 Self::grab_buffer_snapshots(
378 grab_buffer_snapshot_rx,
379 find_all_matches_tx,
380 sorted_matches_tx,
381 cx.clone(),
382 )
383 .boxed_local(),
384 )
385 } else {
386 drop(find_all_matches_tx);
387 None
388 };
389 let ensure_matches_are_reported_in_order = if should_find_all_matches {
390 Some(
391 Self::ensure_matched_ranges_are_reported_in_order(sorted_matches_rx, tx)
392 .boxed_local(),
393 )
394 } else {
395 drop(tx);
396
397 None
398 };
399
400 futures::future::join_all(
401 [worker_pool.boxed_local()]
402 .into_iter()
403 .chain(buffer_snapshots)
404 .chain(ensure_matches_are_reported_in_order)
405 .chain(tasks),
406 )
407 .await;
408 })
409 });
410
411 SearchResultsHandle {
412 results: rx,
413 matching_buffers,
414 trigger_search,
415 }
416 }
417
418 fn provide_search_paths(
419 worktrees: Vec<Entity<Worktree>>,
420 query: Arc<SearchQuery>,
421 tx: Sender<InputPath>,
422 results: Sender<oneshot::Receiver<(ProjectPath, LineHint)>>,
423 results_tx: Sender<SearchResult>,
424 ) -> impl AsyncFnOnce(&mut AsyncApp) {
425 async move |cx| {
426 _ = maybe!(async move {
427 let gitignored_tracker = PathInclusionMatcher::new(query.clone());
428 let include_ignored = query.include_ignored();
429 for worktree in worktrees {
430 let scan_complete = worktree.read_with(cx, |worktree, _| {
431 worktree.as_local().map(|local| local.scan_complete())
432 });
433 if let Some(scan_complete) = scan_complete {
434 let mut scan_complete = pin!(scan_complete);
435 if scan_complete.as_mut().now_or_never().is_none() {
436 _ = results_tx.send(SearchResult::WaitingForScan).await;
437 scan_complete.await;
438 _ = results_tx.send(SearchResult::Searching).await;
439 }
440 }
441
442 let (mut snapshot, worktree_settings) = worktree
443 .read_with(cx, |this, _| {
444 Some((this.snapshot(), this.as_local()?.settings()))
445 })
446 .context("The worktree is not local")?;
447 if query.include_ignored() {
448 // Pre-fetch all of the ignored directories as they're going to be searched.
449 let mut entries_to_refresh = vec![];
450
451 for entry in snapshot.entries(query.include_ignored(), 0) {
452 if gitignored_tracker.should_scan_gitignored_dir(
453 entry,
454 &snapshot,
455 &worktree_settings,
456 ) {
457 entries_to_refresh.push(entry.path.clone());
458 }
459 }
460 let barrier = worktree.update(cx, |this, _| {
461 let local = this.as_local_mut()?;
462 let barrier = entries_to_refresh
463 .into_iter()
464 .map(|path| local.add_path_prefix_to_scan(path).into_future())
465 .collect::<Vec<_>>();
466 Some(barrier)
467 });
468 if let Some(barriers) = barrier {
469 futures::future::join_all(barriers).await;
470 }
471 snapshot = worktree.read_with(cx, |this, _| this.snapshot());
472 }
473 let tx = tx.clone();
474 let results = results.clone();
475
476 cx.background_executor()
477 .spawn(async move {
478 for entry in snapshot.files(include_ignored, 0) {
479 let (should_scan_tx, should_scan_rx) = oneshot::channel();
480
481 let Ok(_) = tx
482 .send(InputPath {
483 entry: entry.clone(),
484 snapshot: snapshot.clone(),
485 should_scan_tx,
486 })
487 .await
488 else {
489 return;
490 };
491 if results.send(should_scan_rx).await.is_err() {
492 return;
493 };
494 }
495 })
496 .await;
497 }
498 anyhow::Ok(())
499 })
500 .await;
501 }
502 }
503
504 async fn maintain_sorted_search_results(
505 rx: Receiver<oneshot::Receiver<(ProjectPath, LineHint)>>,
506 paths_for_full_scan: Sender<(ProjectPath, LineHint)>,
507 limit: usize,
508 ) {
509 let mut rx = pin!(rx);
510 let mut matched = 0;
511 while let Some(mut next_path_result) = rx.next().await {
512 let Some((successful_path, line_hint)) = next_path_result.next().await else {
513 // This file did not produce a match, hence skip it.
514 continue;
515 };
516 if paths_for_full_scan
517 .send((successful_path, line_hint))
518 .await
519 .is_err()
520 {
521 return;
522 };
523 matched += 1;
524 if matched >= limit {
525 break;
526 }
527 }
528 }
529
530 /// Background workers cannot open buffers by themselves, hence main thread will do it on their behalf.
531 async fn open_buffers(
532 buffer_store: Entity<BufferStore>,
533 rx: Receiver<(ProjectPath, LineHint)>,
534 find_all_matches_tx: Sender<(Entity<Buffer>, LineHint)>,
535 mut cx: AsyncApp,
536 ) {
537 let mut rx = pin!(rx.ready_chunks(64));
538 _ = maybe!(async move {
539 while let Some(requested_paths) = rx.next().await {
540 let line_hints: Vec<LineHint> =
541 requested_paths.iter().map(|(_, line)| *line).collect();
542 let mut buffers = buffer_store.update(&mut cx, |this, cx| {
543 requested_paths
544 .into_iter()
545 .map(|(path, _)| this.open_buffer(path, cx))
546 .collect::<FuturesOrdered<_>>()
547 });
548 let mut line_hints = line_hints.into_iter();
549 while let Some(buffer) = buffers.next().await {
550 let line_hint = line_hints.next().unwrap_or(LineHint::default());
551 if let Some(buffer) = buffer.log_err() {
552 find_all_matches_tx.send((buffer, line_hint)).await?;
553 }
554 }
555 }
556 Result::<_, anyhow::Error>::Ok(())
557 })
558 .await;
559 }
560
561 async fn grab_buffer_snapshots(
562 rx: Receiver<(Entity<Buffer>, LineHint)>,
563 find_all_matches_tx: Sender<FindAllMatchesRequest>,
564 results: Sender<oneshot::Receiver<(Entity<Buffer>, Vec<Range<language::Anchor>>)>>,
565 mut cx: AsyncApp,
566 ) {
567 _ = maybe!(async move {
568 while let Ok((buffer, line_hint)) = rx.recv().await {
569 let snapshot = buffer.read_with(&mut cx, |this, _| this.snapshot());
570 let (tx, rx) = oneshot::channel();
571 find_all_matches_tx
572 .send(FindAllMatchesRequest {
573 buffer,
574 snapshot,
575 line_hint,
576 report_matches: tx,
577 })
578 .await?;
579 results.send(rx).await?;
580 }
581 debug_assert!(rx.is_empty());
582 Result::<_, anyhow::Error>::Ok(())
583 })
584 .await;
585 }
586
587 async fn ensure_matched_ranges_are_reported_in_order(
588 rx: Receiver<oneshot::Receiver<(Entity<Buffer>, Vec<Range<language::Anchor>>)>>,
589 tx: Sender<SearchResult>,
590 ) {
591 use postage::stream::Stream;
592 _ = maybe!(async move {
593 let mut matched_buffers = 0;
594 let mut matches = 0;
595 while let Ok(mut next_buffer_matches) = rx.recv().await {
596 let Some((buffer, ranges)) = next_buffer_matches.recv().await else {
597 continue;
598 };
599
600 if matched_buffers > Search::MAX_SEARCH_RESULT_FILES
601 || matches > Search::MAX_SEARCH_RESULT_RANGES
602 {
603 _ = tx.send(SearchResult::LimitReached).await;
604 break;
605 }
606 matched_buffers += 1;
607 matches += ranges.len();
608
609 _ = tx.send(SearchResult::Buffer { buffer, ranges }).await?;
610 }
611 anyhow::Ok(())
612 })
613 .await;
614 }
615
616 fn all_loaded_buffers(&self, search_query: &SearchQuery, cx: &App) -> Vec<Entity<Buffer>> {
617 let worktree_store = self.worktree_store.read(cx);
618 let mut buffers = search_query
619 .buffers()
620 .into_iter()
621 .flatten()
622 .filter(|buffer| {
623 let b = buffer.read(cx);
624 if let Some(file) = b.file() {
625 if file.disk_state().is_deleted() {
626 return false;
627 }
628 if !search_query.match_path(file.path()) {
629 return false;
630 }
631 if !search_query.include_ignored()
632 && let Some(entry) = b
633 .entry_id(cx)
634 .and_then(|entry_id| worktree_store.entry_for_id(entry_id, cx))
635 && entry.is_ignored
636 {
637 return false;
638 }
639 }
640 true
641 })
642 .cloned()
643 .collect::<Vec<_>>();
644 buffers.sort_by(|a, b| {
645 let a = a.read(cx);
646 let b = b.read(cx);
647 match (a.file(), b.file()) {
648 (None, None) => a.remote_id().cmp(&b.remote_id()),
649 (None, Some(_)) => std::cmp::Ordering::Less,
650 (Some(_), None) => std::cmp::Ordering::Greater,
651 (Some(a), Some(b)) => compare_rel_paths((a.path(), true), (b.path(), true)),
652 }
653 });
654
655 buffers
656 }
657}
658
659struct Worker {
660 query: Arc<SearchQuery>,
661 open_buffers: Arc<HashSet<ProjectEntryId>>,
662 candidates: FindSearchCandidates,
663 /// Ok, we're back in background: run full scan & find all matches in a given buffer snapshot.
664 /// Then, when you're done, share them via the channel you were given.
665 find_all_matches_rx: Receiver<FindAllMatchesRequest>,
666}
667
668impl Worker {
669 async fn run(self) {
670 let (
671 input_paths_rx,
672 confirm_contents_will_match_rx,
673 mut confirm_contents_will_match_tx,
674 fs,
675 ) = match self.candidates {
676 FindSearchCandidates::Local {
677 fs,
678 input_paths_rx,
679 confirm_contents_will_match_rx,
680 confirm_contents_will_match_tx,
681 } => (
682 input_paths_rx,
683 confirm_contents_will_match_rx,
684 confirm_contents_will_match_tx,
685 Some(fs),
686 ),
687 FindSearchCandidates::Remote | FindSearchCandidates::OpenBuffersOnly => {
688 (unbounded().1, unbounded().1, unbounded().0, None)
689 }
690 };
691 // WorkerA: grabs a request for "find all matches in file/a" <- takes 5 minutes
692 // right after: WorkerB: grabs a request for "find all matches in file/b" <- takes 5 seconds
693 let mut find_all_matches = pin!(self.find_all_matches_rx.fuse());
694 let mut find_first_match = pin!(confirm_contents_will_match_rx.fuse());
695 let mut scan_path = pin!(input_paths_rx.fuse());
696
697 loop {
698 let handler = RequestHandler {
699 query: &self.query,
700 open_entries: &self.open_buffers,
701 fs: fs.as_deref(),
702 confirm_contents_will_match_tx: &confirm_contents_will_match_tx,
703 };
704 // Whenever we notice that some step of a pipeline is closed, we don't want to close subsequent
705 // steps straight away. Another worker might be about to produce a value that will
706 // be pushed there, thus we'll replace current worker's pipe with a dummy one.
707 // That way, we'll only ever close a next-stage channel when ALL workers do so.
708 select_biased! {
709 find_all_matches = find_all_matches.next() => {
710 let Some(matches) = find_all_matches else {
711 continue;
712 };
713 handler.handle_find_all_matches(matches).await;
714 },
715 find_first_match = find_first_match.next() => {
716 if let Some(buffer_with_at_least_one_match) = find_first_match {
717 handler.handle_find_first_match(buffer_with_at_least_one_match).await;
718 }
719 },
720 scan_path = scan_path.next() => {
721 if let Some(path_to_scan) = scan_path {
722 handler.handle_scan_path(path_to_scan).await;
723 } else {
724 // If we're the last worker to notice that this is not producing values, close the upstream.
725 confirm_contents_will_match_tx = bounded(1).0;
726 }
727
728 }
729 complete => {
730 break
731 },
732
733 }
734 }
735 }
736}
737
738struct RequestHandler<'worker> {
739 query: &'worker SearchQuery,
740 fs: Option<&'worker dyn Fs>,
741 open_entries: &'worker HashSet<ProjectEntryId>,
742 confirm_contents_will_match_tx: &'worker Sender<MatchingEntry>,
743}
744
745impl RequestHandler<'_> {
746 async fn handle_find_all_matches(&self, request: FindAllMatchesRequest) {
747 let FindAllMatchesRequest {
748 buffer,
749 snapshot,
750 line_hint,
751 mut report_matches,
752 } = request;
753 let range_offset = if line_hint > 0 {
754 snapshot.point_to_offset(Point::new(line_hint, 0))
755 } else {
756 0
757 };
758 let subrange = (range_offset > 0).then(|| range_offset..snapshot.len());
759 let ranges = self
760 .query
761 .search(&snapshot, subrange)
762 .await
763 .iter()
764 .map(|range| {
765 snapshot.anchor_before(range.start + range_offset)
766 ..snapshot.anchor_after(range.end + range_offset)
767 })
768 .collect::<Vec<_>>();
769
770 _ = report_matches.send((buffer, ranges)).await;
771 }
772
773 async fn handle_find_first_match(&self, mut entry: MatchingEntry) {
774 async move {
775 let abs_path = entry.worktree_root.join(entry.path.path.as_std_path());
776 let Some(file) = self
777 .fs
778 .context("Trying to query filesystem in remote project search")?
779 .open_sync(&abs_path)
780 .await
781 .log_err()
782 else {
783 return anyhow::Ok(());
784 };
785
786 let mut file = BufReader::new(file);
787 let file_start = file.fill_buf()?;
788
789 if let Err(Some(starting_position)) =
790 std::str::from_utf8(file_start).map_err(|e| e.error_len())
791 {
792 // Before attempting to match the file content, throw away files that have invalid UTF-8 sequences early on;
793 // That way we can still match files in a streaming fashion without having look at "obviously binary" files.
794 log::debug!(
795 "Invalid UTF-8 sequence in file {abs_path:?} \
796 at byte position {starting_position}"
797 );
798 return Ok(());
799 }
800
801 if let Some(line_hint) = self.query.detect(file).await.ok().flatten() {
802 // Yes, we should scan the whole file.
803 entry.should_scan_tx.send((entry.path, line_hint)).await?;
804 }
805 Ok(())
806 }
807 .await
808 .ok();
809 }
810
811 async fn handle_scan_path(&self, req: InputPath) {
812 _ = maybe!(async move {
813 let InputPath {
814 entry,
815 snapshot,
816 mut should_scan_tx,
817 } = req;
818
819 if entry.is_fifo || !entry.is_file() {
820 return Ok(());
821 }
822
823 if self.query.filters_path() {
824 let matched_path = if self.query.match_full_paths() {
825 let mut full_path = snapshot.root_name().to_owned();
826 full_path.push(&entry.path);
827 self.query.match_path(&full_path)
828 } else {
829 self.query.match_path(&entry.path)
830 };
831 if !matched_path {
832 return Ok(());
833 }
834 }
835
836 if self.open_entries.contains(&entry.id) {
837 // The buffer is already in memory and that's the version we want to scan;
838 // hence skip the dilly-dally and look for all matches straight away.
839 should_scan_tx
840 .send((
841 ProjectPath {
842 worktree_id: snapshot.id(),
843 path: entry.path.clone(),
844 },
845 LineHint::default(),
846 ))
847 .await?;
848 } else {
849 self.confirm_contents_will_match_tx
850 .send(MatchingEntry {
851 should_scan_tx: should_scan_tx,
852 worktree_root: snapshot.abs_path().clone(),
853 path: ProjectPath {
854 worktree_id: snapshot.id(),
855 path: entry.path.clone(),
856 },
857 })
858 .await?;
859 }
860
861 anyhow::Ok(())
862 })
863 .await;
864 }
865}
866
867struct InputPath {
868 entry: Entry,
869 snapshot: Snapshot,
870 should_scan_tx: oneshot::Sender<(ProjectPath, LineHint)>,
871}
872
873struct MatchingEntry {
874 worktree_root: Arc<Path>,
875 path: ProjectPath,
876 should_scan_tx: oneshot::Sender<(ProjectPath, LineHint)>,
877}
878
879struct FindAllMatchesRequest {
880 buffer: Entity<Buffer>,
881 snapshot: BufferSnapshot,
882 line_hint: LineHint,
883 report_matches: oneshot::Sender<(Entity<Buffer>, Vec<Range<language::Anchor>>)>,
884}
885
886/// This struct encapsulates the logic to decide whether a given gitignored directory should be
887/// scanned based on include/exclude patterns of a search query (as include/exclude parameters may match paths inside it).
888/// It is kind-of doing an inverse of glob. Given a glob pattern like `src/**/` and a parent path like `src`, we need to decide whether the parent
889/// may contain glob hits.
890pub struct PathInclusionMatcher {
891 included: BTreeSet<PathBuf>,
892 query: Arc<SearchQuery>,
893}
894
895impl PathInclusionMatcher {
896 pub fn new(query: Arc<SearchQuery>) -> Self {
897 let mut included = BTreeSet::new();
898 // To do an inverse glob match, we split each glob into it's prefix and the glob part.
899 // For example, `src/**/*.rs` becomes `src/` and `**/*.rs`. The glob part gets dropped.
900 // Then, when checking whether a given directory should be scanned, we check whether it is a non-empty substring of any glob prefix.
901 if query.filters_path() {
902 included.extend(
903 query
904 .files_to_include()
905 .sources()
906 .flat_map(|glob| Some(wax::Glob::new(glob).ok()?.partition().0)),
907 );
908 }
909 Self { included, query }
910 }
911
912 pub fn should_scan_gitignored_dir(
913 &self,
914 entry: &Entry,
915 snapshot: &Snapshot,
916 worktree_settings: &WorktreeSettings,
917 ) -> bool {
918 if !entry.is_ignored || !entry.kind.is_unloaded() {
919 return false;
920 }
921 if !self.query.include_ignored() {
922 return false;
923 }
924 if worktree_settings.is_path_excluded(&entry.path) {
925 return false;
926 }
927 if !self.query.filters_path() {
928 return true;
929 }
930
931 let as_abs_path = LazyCell::new(move || snapshot.absolutize(&entry.path));
932 let entry_path = &entry.path;
933 // 3. Check Exclusions (Pruning)
934 // If the current path is a child of an excluded path, we stop.
935 let is_excluded = self.path_is_definitely_excluded(&entry_path, snapshot);
936
937 if is_excluded {
938 return false;
939 }
940
941 // 4. Check Inclusions (Traversal)
942 if self.included.is_empty() {
943 return true;
944 }
945
946 // We scan if the current path is a descendant of an include prefix
947 // OR if the current path is an ancestor of an include prefix (we need to go deeper to find it).
948 let is_included = self.included.iter().any(|prefix| {
949 let (prefix_matches_entry, entry_matches_prefix) = if prefix.is_absolute() {
950 (
951 prefix.starts_with(&**as_abs_path),
952 as_abs_path.starts_with(prefix),
953 )
954 } else {
955 RelPath::new(prefix, snapshot.path_style()).map_or((false, false), |prefix| {
956 (
957 prefix.starts_with(entry_path),
958 entry_path.starts_with(&prefix),
959 )
960 })
961 };
962
963 // Logic:
964 // 1. entry_matches_prefix: We are inside the target zone (e.g. glob: src/, current: src/lib/). Keep scanning.
965 // 2. prefix_matches_entry: We are above the target zone (e.g. glob: src/foo/, current: src/). Keep scanning to reach foo.
966 prefix_matches_entry || entry_matches_prefix
967 });
968
969 is_included
970 }
971 fn path_is_definitely_excluded(&self, path: &RelPath, snapshot: &Snapshot) -> bool {
972 if !self.query.files_to_exclude().sources().next().is_none() {
973 let mut path = if self.query.match_full_paths() {
974 let mut full_path = snapshot.root_name().to_owned();
975 full_path.push(path);
976 full_path
977 } else {
978 path.to_owned()
979 };
980 loop {
981 if self.query.files_to_exclude().is_match(&path) {
982 return true;
983 } else if !path.pop() {
984 return false;
985 }
986 }
987 } else {
988 false
989 }
990 }
991}
992
993type IsTerminating = bool;
994/// Adaptive batcher that starts eager (small batches) and grows batch size
995/// when items arrive quickly, reducing RPC overhead while preserving low latency
996/// for slow streams.
997pub struct AdaptiveBatcher<T> {
998 items: Sender<T>,
999 flush_batch: Sender<IsTerminating>,
1000 _batch_task: Task<()>,
1001}
1002
1003impl<T: 'static + Send> AdaptiveBatcher<T> {
1004 pub fn new(cx: &BackgroundExecutor) -> (Self, Receiver<Vec<T>>) {
1005 let (items, rx) = unbounded();
1006 let (batch_tx, batch_rx) = unbounded();
1007 let (flush_batch_tx, flush_batch_rx) = unbounded();
1008 let flush_batch = flush_batch_tx.clone();
1009 let executor = cx.clone();
1010 let _batch_task = cx.spawn_with_priority(gpui::Priority::High, async move {
1011 let mut current_batch = vec![];
1012 let mut items_produced_so_far = 0_u64;
1013
1014 let mut _schedule_flush_after_delay: Option<Task<()>> = None;
1015 let _time_elapsed_since_start_of_search = std::time::Instant::now();
1016 let mut flush = pin!(flush_batch_rx);
1017 let mut terminating = false;
1018 loop {
1019 select_biased! {
1020 item = rx.recv().fuse() => {
1021 match item {
1022 Ok(new_item) => {
1023 let is_fresh_batch = current_batch.is_empty();
1024 items_produced_so_far += 1;
1025 current_batch.push(new_item);
1026 if is_fresh_batch {
1027 // Chosen arbitrarily based on some experimentation with plots.
1028 let desired_duration_ms = (20 * (items_produced_so_far + 2).ilog2() as u64).min(300);
1029 let desired_duration = Duration::from_millis(desired_duration_ms);
1030 let _executor = executor.clone();
1031 let _flush = flush_batch_tx.clone();
1032 let new_timer = executor.spawn_with_priority(Priority::High, async move {
1033 _executor.timer(desired_duration).await;
1034 _ = _flush.send(false).await;
1035 });
1036 _schedule_flush_after_delay = Some(new_timer);
1037 }
1038 }
1039 Err(_) => {
1040 // Items channel closed - send any remaining batch before exiting
1041 if !current_batch.is_empty() {
1042 _ = batch_tx.send(std::mem::take(&mut current_batch)).await;
1043 }
1044 break;
1045 }
1046 }
1047 }
1048 should_break_afterwards = flush.next() => {
1049 if !current_batch.is_empty() {
1050 _ = batch_tx.send(std::mem::take(&mut current_batch)).await;
1051 _schedule_flush_after_delay = None;
1052 }
1053 if should_break_afterwards.unwrap_or_default() {
1054 terminating = true;
1055 }
1056 }
1057 complete => {
1058 break;
1059 }
1060 }
1061 if terminating {
1062 // Drain any remaining items before exiting
1063 while let Ok(new_item) = rx.try_recv() {
1064 current_batch.push(new_item);
1065 }
1066 if !current_batch.is_empty() {
1067 _ = batch_tx.send(std::mem::take(&mut current_batch)).await;
1068 }
1069 break;
1070 }
1071 }
1072 });
1073 let this = Self {
1074 items,
1075 _batch_task,
1076 flush_batch,
1077 };
1078 (this, batch_rx)
1079 }
1080
1081 pub async fn push(&self, item: T) {
1082 _ = self.items.send(item).await;
1083 }
1084
1085 pub async fn flush(self) {
1086 _ = self.flush_batch.send(true).await;
1087 self._batch_task.await;
1088 }
1089}
1090