1use std::{
2 collections::HashSet,
3 future::Future,
4 path::{Path, PathBuf},
5 pin::Pin,
6 sync::{
7 Arc,
8 atomic::{AtomicU64, Ordering},
9 },
10 time::Duration,
11};
12
13use notify::{Config, EventKind, RecommendedWatcher, RecursiveMode, Watcher as NotifyWatcher};
14use serde::{Deserialize, Serialize};
15use tokio::sync::{RwLock, mpsc, oneshot, watch};
16use tracing;
17
18use super::{
19 ContentHashCache, WatchedRepository,
20 config::WatcherConfig,
21 event_filter::WatcherEventFilter,
22 hash_cache::content_hash64,
23 task_seed::{
24 ChangedPathSnapshot, build_incremental_task_seed, changed_content_fingerprint,
25 unreadable_path_fingerprint,
26 },
27};
28
29const DEBOUNCE_CHANNEL_CAPACITY: usize = 4096;
30const WATCHER_COMMAND_CHANNEL_CAPACITY: usize = 128;
31const WATCHER_COMMAND_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
32
33type TaskQueueFuture = Pin<Box<dyn Future<Output = Result<(), String>> + Send>>;
34type TaskQueueSink =
35 Arc<dyn Fn(crate::storage::CodeIndexTaskSeed) -> TaskQueueFuture + Send + Sync>;
36
37fn boxed_task_sink<F, Fut>(task_sink: F) -> TaskQueueSink
38where
39 F: Fn(crate::storage::CodeIndexTaskSeed) -> Fut + Send + Sync + 'static,
40 Fut: Future<Output = Result<(), String>> + Send + 'static,
41{
42 Arc::new(move |task| Box::pin(task_sink(task)))
43}
44
45#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(rename_all = "snake_case")]
47pub enum WatcherState {
48 Disabled,
49 Active,
50 Degraded,
51 Failed,
52}
53
54impl WatcherState {
55 pub const fn as_str(self) -> &'static str {
56 match self {
57 Self::Disabled => "disabled",
58 Self::Active => "active",
59 Self::Degraded => "degraded",
60 Self::Failed => "failed",
61 }
62 }
63
64 pub fn parse(value: &str) -> Option<Self> {
65 match value {
66 "disabled" => Some(Self::Disabled),
67 "active" => Some(Self::Active),
68 "degraded" => Some(Self::Degraded),
69 "failed" => Some(Self::Failed),
70 _ => None,
71 }
72 }
73}
74
75#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
76pub struct WatcherDiagnostics {
77 pub state: WatcherState,
78 pub watched_repository_count: usize,
79 pub total_events_received: u64,
80 pub total_events_filtered: u64,
81 pub total_index_tasks_queued: u64,
82 pub total_events_dropped: u64,
83 #[serde(skip_serializing_if = "Option::is_none")]
84 pub last_error: Option<String>,
85 #[serde(skip_serializing_if = "Option::is_none")]
86 pub degraded_reason: Option<String>,
87}
88
89impl Default for WatcherDiagnostics {
90 fn default() -> Self {
91 Self {
92 state: WatcherState::Disabled,
93 watched_repository_count: 0,
94 total_events_received: 0,
95 total_events_filtered: 0,
96 total_index_tasks_queued: 0,
97 total_events_dropped: 0,
98 last_error: None,
99 degraded_reason: None,
100 }
101 }
102}
103
104#[derive(Debug, Clone)]
105pub struct WatcherHandle {
106 diagnostics: watch::Receiver<WatcherDiagnostics>,
107 shutdown: watch::Sender<bool>,
108 state: Arc<RwLock<WatcherInternalState>>,
109 command_tx: Option<mpsc::Sender<WatcherCommand>>,
110}
111
112impl WatcherHandle {
113 pub fn diagnostics(&self) -> WatcherDiagnostics {
114 self.diagnostics.borrow().clone()
115 }
116
117 pub async fn updated_diagnostics(&self) -> WatcherDiagnostics {
118 let mut rx = self.diagnostics.clone();
119 let _ = rx.changed().await;
120 rx.borrow().clone()
121 }
122
123 pub fn request_shutdown(&self) {
124 let _ = self.shutdown.send(true);
125 }
126
127 pub async fn add_repository(&self, repo: WatchedRepository) -> bool {
128 let Some(command_tx) = &self.command_tx else {
129 return false;
130 };
131 let (response_tx, response_rx) = oneshot::channel();
132 let command = WatcherCommand::Add {
133 repository: repo,
134 response: response_tx,
135 };
136 if command_tx.send(command).await.is_err() {
137 return false;
138 }
139 match tokio::time::timeout(WATCHER_COMMAND_RESPONSE_TIMEOUT, response_rx).await {
140 Ok(Ok(updated)) => updated,
141 _ => false,
142 }
143 }
144
145 pub async fn remove_repository(&self, alias: &str) -> bool {
146 let Some(command_tx) = &self.command_tx else {
147 return false;
148 };
149 let (response_tx, response_rx) = oneshot::channel();
150 let command = WatcherCommand::Remove {
151 alias_or_id: alias.to_owned(),
152 response: response_tx,
153 };
154 if command_tx.send(command).await.is_err() {
155 return false;
156 }
157 match tokio::time::timeout(WATCHER_COMMAND_RESPONSE_TIMEOUT, response_rx).await {
158 Ok(Ok(updated)) => updated,
159 _ => false,
160 }
161 }
162
163 pub async fn repository_count(&self) -> usize {
164 self.state.read().await.repositories.len()
165 }
166}
167
168#[derive(Debug)]
169struct WatcherInternalState {
170 repositories: Vec<WatchedRepository>,
171 hash_cache: ContentHashCache,
172 events_received: u64,
173 events_filtered: u64,
174 index_tasks_queued: u64,
175}
176
177enum WatcherCommand {
178 Add {
179 repository: WatchedRepository,
180 response: oneshot::Sender<bool>,
181 },
182 Remove {
183 alias_or_id: String,
184 response: oneshot::Sender<bool>,
185 },
186}
187
188struct WatcherLoopContext {
189 state: Arc<RwLock<WatcherInternalState>>,
190 diag_tx: watch::Sender<WatcherDiagnostics>,
191 dropped_events: Arc<AtomicU64>,
192 debounce: Duration,
193 max_watch_dirs: usize,
194 task_sink: TaskQueueSink,
195}
196
197enum WatchRegistrationPlan {
198 Add {
199 watch_root: bool,
200 },
201 Replace {
202 index: usize,
203 previous_root: PathBuf,
204 watch_new_root: bool,
205 unwatch_previous_root: bool,
206 },
207}
208
209pub struct FileWatcher {
210 config: WatcherConfig,
211}
212
213impl FileWatcher {
214 pub fn new(config: WatcherConfig) -> Self {
215 Self { config }
216 }
217
218 pub fn start(self, repositories: Vec<WatchedRepository>) -> Result<WatcherHandle, String> {
219 self.start_with_sink(repositories, |_| async { Ok(()) })
220 }
221
222 pub fn start_with_sink<F, Fut>(
223 self,
224 repositories: Vec<WatchedRepository>,
225 task_sink: F,
226 ) -> Result<WatcherHandle, String>
227 where
228 F: Fn(crate::storage::CodeIndexTaskSeed) -> Fut + Send + Sync + 'static,
229 Fut: Future<Output = Result<(), String>> + Send + 'static,
230 {
231 if !self.config.enabled {
232 let (_diag_tx, diag_rx) = watch::channel(WatcherDiagnostics {
233 state: WatcherState::Disabled,
234 ..WatcherDiagnostics::default()
235 });
236 let (shutdown_tx, _) = watch::channel(false);
237 return Ok(WatcherHandle {
238 diagnostics: diag_rx,
239 shutdown: shutdown_tx,
240 state: Arc::new(RwLock::new(WatcherInternalState {
241 repositories: Vec::new(),
242 hash_cache: ContentHashCache::new(self.config.hash_cache_capacity),
243 events_received: 0,
244 events_filtered: 0,
245 index_tasks_queued: 0,
246 })),
247 command_tx: None,
248 });
249 }
250
251 let (diag_tx, diag_rx) = watch::channel(WatcherDiagnostics {
252 state: WatcherState::Active,
253 ..WatcherDiagnostics::default()
254 });
255 let (shutdown_tx, shutdown_rx) = watch::channel(false);
256 let (command_tx, command_rx) = mpsc::channel(WATCHER_COMMAND_CHANNEL_CAPACITY);
257
258 let state = Arc::new(RwLock::new(WatcherInternalState {
259 repositories: Vec::new(),
260 hash_cache: ContentHashCache::new(self.config.hash_cache_capacity),
261 events_received: 0,
262 events_filtered: 0,
263 index_tasks_queued: 0,
264 }));
265
266 let handle = WatcherHandle {
267 diagnostics: diag_rx,
268 shutdown: shutdown_tx,
269 state: state.clone(),
270 command_tx: Some(command_tx),
271 };
272
273 let diag_sender = diag_tx;
274 let debounce = self.config.debounce;
275 let max_watch_dirs = self.config.max_watch_dirs;
276 let dropped_events = Arc::new(AtomicU64::new(0));
277 let task_sink = boxed_task_sink(task_sink);
278
279 tokio::spawn(async move {
280 run_watcher_loop(
281 WatcherLoopContext {
282 state,
283 diag_tx: diag_sender,
284 dropped_events,
285 debounce,
286 max_watch_dirs,
287 task_sink,
288 },
289 shutdown_rx,
290 repositories,
291 command_rx,
292 )
293 .await;
294 });
295
296 Ok(handle)
297 }
298}
299
300async fn run_watcher_loop(
301 context: WatcherLoopContext,
302 mut shutdown_rx: watch::Receiver<bool>,
303 initial_repositories: Vec<WatchedRepository>,
304 mut command_rx: mpsc::Receiver<WatcherCommand>,
305) {
306 let (event_tx, mut event_rx) = mpsc::channel::<PathBuf>(DEBOUNCE_CHANNEL_CAPACITY);
307 let state = &context.state;
308 let diag_tx = &context.diag_tx;
309 let dropped_events = &context.dropped_events;
310
311 let mut watcher = match create_notify_watcher(event_tx.clone(), Arc::clone(dropped_events)) {
312 Ok(watcher) => watcher,
313 Err(error) => {
314 update_diagnostics_failed(diag_tx, state, dropped_events, &error).await;
315 return;
316 }
317 };
318 for repo in initial_repositories {
319 watch_repository(
320 &mut watcher,
321 state,
322 diag_tx,
323 dropped_events,
324 repo,
325 context.max_watch_dirs,
326 )
327 .await;
328 }
329
330 let mut pending_paths: HashSet<PathBuf> = HashSet::new();
331 let mut debounce_deadline: Option<tokio::time::Instant> = None;
332
333 loop {
334 if let Some(deadline) = debounce_deadline {
335 tokio::select! {
336 maybe_path = event_rx.recv() => {
337 if !handle_path_event(maybe_path, state, diag_tx, dropped_events, &mut pending_paths, context.debounce, &mut debounce_deadline).await {
338 flush_pending(state, diag_tx, dropped_events, &mut pending_paths, &context.task_sink).await;
339 update_diagnostics_failed(diag_tx, state, dropped_events, "event channel closed").await;
340 unwatch_all_repositories(&mut watcher, state).await;
341 return;
342 }
343 }
344 maybe_command = command_rx.recv() => {
345 match maybe_command {
346 Some(command) => {
347 handle_watcher_command(command, &mut watcher, state, diag_tx, dropped_events, context.max_watch_dirs).await;
348 }
349 None => {
350 flush_pending(state, diag_tx, dropped_events, &mut pending_paths, &context.task_sink).await;
351 unwatch_all_repositories(&mut watcher, state).await;
352 return;
353 }
354 }
355 }
356 _ = tokio::time::sleep_until(deadline) => {
357 let changed_paths: Vec<PathBuf> = pending_paths.drain().collect();
358 if !changed_paths.is_empty() {
359 process_debounced_paths(state, diag_tx, dropped_events, &changed_paths, &context.task_sink).await;
360 }
361 debounce_deadline = None;
362 }
363 _ = shutdown_rx.changed() => {
364 flush_pending(state, diag_tx, dropped_events, &mut pending_paths, &context.task_sink).await;
365 unwatch_all_repositories(&mut watcher, state).await;
366 return;
367 }
368 }
369 } else {
370 tokio::select! {
371 maybe_path = event_rx.recv() => {
372 if !handle_path_event(maybe_path, state, diag_tx, dropped_events, &mut pending_paths, context.debounce, &mut debounce_deadline).await {
373 update_diagnostics_failed(diag_tx, state, dropped_events, "event channel closed").await;
374 unwatch_all_repositories(&mut watcher, state).await;
375 return;
376 }
377 }
378 maybe_command = command_rx.recv() => {
379 match maybe_command {
380 Some(command) => {
381 handle_watcher_command(command, &mut watcher, state, diag_tx, dropped_events, context.max_watch_dirs).await;
382 }
383 None => {
384 unwatch_all_repositories(&mut watcher, state).await;
385 return;
386 }
387 }
388 }
389 _ = shutdown_rx.changed() => {
390 unwatch_all_repositories(&mut watcher, state).await;
391 return;
392 }
393 }
394 }
395 }
396}
397
398async fn handle_path_event(
399 maybe_path: Option<PathBuf>,
400 state: &Arc<RwLock<WatcherInternalState>>,
401 diag_tx: &watch::Sender<WatcherDiagnostics>,
402 dropped_events: &Arc<AtomicU64>,
403 pending_paths: &mut HashSet<PathBuf>,
404 debounce: Duration,
405 debounce_deadline: &mut Option<tokio::time::Instant>,
406) -> bool {
407 let Some(path) = maybe_path else {
408 return false;
409 };
410 {
411 let mut state_guard = state.write().await;
412 state_guard.events_received += 1;
413 }
414
415 let should_process = {
416 let state_guard = state.read().await;
417 should_process_path(&state_guard, &path)
418 };
419
420 if should_process {
421 pending_paths.insert(path);
422 *debounce_deadline = Some(tokio::time::Instant::now() + debounce);
423 } else {
424 let mut state_guard = state.write().await;
425 state_guard.events_filtered += 1;
426 drop(state_guard);
427 emit_diagnostics(state, diag_tx, dropped_events).await;
428 }
429 true
430}
431
432async fn handle_watcher_command(
433 command: WatcherCommand,
434 watcher: &mut RecommendedWatcher,
435 state: &Arc<RwLock<WatcherInternalState>>,
436 diag_tx: &watch::Sender<WatcherDiagnostics>,
437 dropped_events: &Arc<AtomicU64>,
438 max_watch_dirs: usize,
439) {
440 match command {
441 WatcherCommand::Add {
442 repository,
443 response,
444 } => {
445 let watched = watch_repository(
446 watcher,
447 state,
448 diag_tx,
449 dropped_events,
450 repository,
451 max_watch_dirs,
452 )
453 .await;
454 let _ = response.send(watched);
455 }
456 WatcherCommand::Remove {
457 alias_or_id,
458 response,
459 } => {
460 let removed =
461 unwatch_repository(watcher, state, diag_tx, dropped_events, &alias_or_id).await;
462 let _ = response.send(removed);
463 }
464 }
465}
466
467async fn watch_repository(
468 watcher: &mut RecommendedWatcher,
469 state: &Arc<RwLock<WatcherInternalState>>,
470 diag_tx: &watch::Sender<WatcherDiagnostics>,
471 dropped_events: &Arc<AtomicU64>,
472 repo: WatchedRepository,
473 max_watch_dirs: usize,
474) -> bool {
475 let plan = {
476 let state_guard = state.read().await;
477 if let Some(index) = state_guard.repositories.iter().position(|watched| {
478 watched.alias == repo.alias || watched.repository_id == repo.repository_id
479 }) {
480 let watched = &state_guard.repositories[index];
481 if watched == &repo {
482 return false;
483 }
484 let root_changed = watched.root != repo.root;
485 let new_root_already_watched = root_changed
486 && state_guard
487 .repositories
488 .iter()
489 .enumerate()
490 .any(|(repo_index, watched)| repo_index != index && watched.root == repo.root);
491 let previous_root_still_watched = root_changed
492 && state_guard
493 .repositories
494 .iter()
495 .enumerate()
496 .any(|(repo_index, existing)| {
497 repo_index != index && existing.root == watched.root
498 });
499 if root_changed && !new_root_already_watched {
500 let root_count_after = watched_root_count(&state_guard.repositories) + 1
501 - usize::from(!previous_root_still_watched);
502 if root_count_after > max_watch_dirs {
503 drop(state_guard);
504 update_diagnostics_degraded(
505 diag_tx,
506 state,
507 dropped_events,
508 &format!(
509 "exceeded max watch directories limit ({max_watch_dirs}); repository '{}' not watched",
510 repo.alias
511 ),
512 )
513 .await;
514 return false;
515 }
516 }
517 WatchRegistrationPlan::Replace {
518 index,
519 previous_root: watched.root.clone(),
520 watch_new_root: root_changed && !new_root_already_watched,
521 unwatch_previous_root: root_changed && !previous_root_still_watched,
522 }
523 } else {
524 let root_already_watched = state_guard
525 .repositories
526 .iter()
527 .any(|watched| watched.root == repo.root);
528 if !root_already_watched
529 && watched_root_count(&state_guard.repositories) >= max_watch_dirs
530 {
531 drop(state_guard);
532 update_diagnostics_degraded(
533 diag_tx,
534 state,
535 dropped_events,
536 &format!(
537 "exceeded max watch directories limit ({max_watch_dirs}); repository '{}' not watched",
538 repo.alias
539 ),
540 )
541 .await;
542 return false;
543 }
544 WatchRegistrationPlan::Add {
545 watch_root: !root_already_watched,
546 }
547 }
548 };
549
550 match plan {
551 WatchRegistrationPlan::Add { watch_root } => {
552 if watch_root {
553 if let Err(error) = watcher.watch(&repo.root, RecursiveMode::Recursive) {
554 tracing::warn!(
555 repository = %repo.alias,
556 path = %repo.root.display(),
557 error = %error,
558 "failed to watch repository directory"
559 );
560 update_diagnostics_degraded(
561 diag_tx,
562 state,
563 dropped_events,
564 &format!("watch failed for {}: {error}", repo.alias),
565 )
566 .await;
567 return false;
568 }
569 }
570
571 let mut state_guard = state.write().await;
572 state_guard.repositories.push(repo);
573 drop(state_guard);
574 emit_diagnostics(state, diag_tx, dropped_events).await;
575 true
576 }
577 WatchRegistrationPlan::Replace {
578 index,
579 previous_root,
580 watch_new_root,
581 unwatch_previous_root,
582 } => {
583 if watch_new_root {
584 if let Err(error) = watcher.watch(&repo.root, RecursiveMode::Recursive) {
585 tracing::warn!(
586 repository = %repo.alias,
587 path = %repo.root.display(),
588 error = %error,
589 "failed to watch replacement repository directory"
590 );
591 update_diagnostics_degraded(
592 diag_tx,
593 state,
594 dropped_events,
595 &format!("watch refresh failed for {}: {error}", repo.alias),
596 )
597 .await;
598 return false;
599 }
600 }
601
602 let mut state_guard = state.write().await;
603 state_guard.repositories[index] = repo.clone();
604 drop(state_guard);
605
606 if unwatch_previous_root {
607 if let Err(error) = watcher.unwatch(&previous_root) {
608 tracing::warn!(
609 repository = %repo.alias,
610 path = %previous_root.display(),
611 error = %error,
612 "failed to unwatch replaced repository directory"
613 );
614 update_diagnostics_degraded(
615 diag_tx,
616 state,
617 dropped_events,
618 &format!("watch refresh cleanup failed for {}: {error}", repo.alias),
619 )
620 .await;
621 return true;
622 }
623 }
624
625 emit_diagnostics(state, diag_tx, dropped_events).await;
626 true
627 }
628 }
629}
630
631fn watched_root_count(repositories: &[WatchedRepository]) -> usize {
632 let mut roots = HashSet::new();
633 for repo in repositories {
634 roots.insert(repo.root.clone());
635 }
636 roots.len()
637}
638
639async fn unwatch_repository(
640 watcher: &mut RecommendedWatcher,
641 state: &Arc<RwLock<WatcherInternalState>>,
642 diag_tx: &watch::Sender<WatcherDiagnostics>,
643 dropped_events: &Arc<AtomicU64>,
644 alias_or_id: &str,
645) -> bool {
646 let (repo, unwatch_root) = {
647 let mut state_guard = state.write().await;
648 let Some(index) = state_guard
649 .repositories
650 .iter()
651 .position(|repo| repo.alias == alias_or_id || repo.repository_id == alias_or_id)
652 else {
653 return false;
654 };
655 let repo = state_guard.repositories.remove(index);
656 let unwatch_root = !state_guard
657 .repositories
658 .iter()
659 .any(|remaining| remaining.root == repo.root);
660 (repo, unwatch_root)
661 };
662
663 if !unwatch_root {
664 emit_diagnostics(state, diag_tx, dropped_events).await;
665 return true;
666 }
667
668 if let Err(error) = watcher.unwatch(&repo.root) {
669 tracing::warn!(
670 repository = %repo.alias,
671 path = %repo.root.display(),
672 error = %error,
673 "failed to remove repository watcher"
674 );
675 update_diagnostics_degraded(
676 diag_tx,
677 state,
678 dropped_events,
679 &format!("unwatch failed for {}: {error}", repo.alias),
680 )
681 .await;
682 } else {
683 emit_diagnostics(state, diag_tx, dropped_events).await;
684 }
685 true
686}
687
688async fn unwatch_all_repositories(
689 watcher: &mut RecommendedWatcher,
690 state: &Arc<RwLock<WatcherInternalState>>,
691) {
692 let repositories = state.read().await.repositories.clone();
693 let mut unwatched_roots = HashSet::new();
694 for repo in repositories {
695 if unwatched_roots.insert(repo.root.clone()) {
696 let _ = watcher.unwatch(&repo.root);
697 }
698 }
699}
700
701async fn flush_pending(
702 state: &Arc<RwLock<WatcherInternalState>>,
703 diag_tx: &watch::Sender<WatcherDiagnostics>,
704 dropped_events: &Arc<AtomicU64>,
705 pending: &mut HashSet<PathBuf>,
706 task_sink: &TaskQueueSink,
707) {
708 if pending.is_empty() {
709 return;
710 }
711 let changed_paths: Vec<PathBuf> = pending.drain().collect();
712 process_debounced_paths(state, diag_tx, dropped_events, &changed_paths, task_sink).await;
713}
714
715async fn process_debounced_paths(
716 state: &Arc<RwLock<WatcherInternalState>>,
717 diag_tx: &watch::Sender<WatcherDiagnostics>,
718 dropped_events: &Arc<AtomicU64>,
719 paths: &[PathBuf],
720 task_sink: &TaskQueueSink,
721) {
722 let mut changed_snapshots = Vec::new();
723 for path in paths {
724 let read_result = tokio::task::spawn_blocking({
725 let path = path.clone();
726 move || std::fs::read(&path).map(|content| content_hash64(&content))
727 })
728 .await;
729 let mut state_guard = state.write().await;
730 match read_result {
731 Ok(Ok(content_hash)) => {
732 let observation = state_guard.hash_cache.observe_hash(path, content_hash);
733 if observation.changed {
734 changed_snapshots.push(ChangedPathSnapshot {
735 path: path.clone(),
736 content_hash: observation.hash,
737 });
738 } else {
739 state_guard.events_filtered += 1;
740 }
741 }
742 Ok(Err(_)) | Err(_) => {
743 let content_hash = unreadable_path_fingerprint(path);
744 let observation = state_guard.hash_cache.observe_hash(path, content_hash);
745 if observation.changed {
746 changed_snapshots.push(ChangedPathSnapshot {
747 path: path.clone(),
748 content_hash,
749 });
750 } else {
751 state_guard.events_filtered += 1;
752 }
753 }
754 }
755 }
756
757 if !changed_snapshots.is_empty() {
758 let repositories = state.read().await.repositories.clone();
759 let now_ms = std::time::SystemTime::now()
760 .duration_since(std::time::UNIX_EPOCH)
761 .unwrap_or_default()
762 .as_millis() as u64;
763 let mut queued_tasks = 0u64;
764 let mut queue_failed = false;
765 let mut queued_paths = HashSet::new();
766
767 for repo in &repositories {
768 let repo_changes = changed_snapshots
769 .iter()
770 .filter(|change| repository_should_process_path(repo, &change.path))
771 .collect::<Vec<_>>();
772 let repo_paths: Vec<PathBuf> = repo_changes
773 .iter()
774 .map(|change| change.path.clone())
775 .collect();
776 let content_fingerprint = changed_content_fingerprint(repo, &repo_changes);
777 if let Some(seed) = build_incremental_task_seed(
778 repo,
779 &repo_paths,
780 "HEAD",
781 "",
782 "",
783 content_fingerprint,
784 now_ms,
785 ) {
786 match task_sink(seed).await {
787 Ok(()) => {
788 queued_tasks += 1;
789 for change in repo_changes {
790 queued_paths.insert(change.path.clone());
791 }
792 }
793 Err(error) => {
794 queue_failed = true;
795 update_diagnostics_degraded(
796 diag_tx,
797 state,
798 dropped_events,
799 &format!("code index task queue failed for {}: {error}", repo.alias),
800 )
801 .await;
802 }
803 }
804 }
805 }
806
807 if queued_tasks > 0 {
808 let mut state_guard = state.write().await;
809 state_guard.index_tasks_queued += queued_tasks;
810 if !queue_failed {
811 for snapshot in changed_snapshots
812 .iter()
813 .filter(|snapshot| queued_paths.contains(&snapshot.path))
814 {
815 state_guard
816 .hash_cache
817 .record_hash(snapshot.path.clone(), snapshot.content_hash);
818 }
819 }
820 }
821 }
822
823 emit_diagnostics(state, diag_tx, dropped_events).await;
824}
825
826fn should_process_path(state: &WatcherInternalState, path: &Path) -> bool {
827 for repo in &state.repositories {
828 if repository_should_process_path(repo, path) {
829 return true;
830 }
831 }
832 false
833}
834
835fn repository_should_process_path(repo: &WatchedRepository, path: &Path) -> bool {
836 WatcherEventFilter::new(
837 repo.root.clone(),
838 repo.path_filters.clone(),
839 repo.language_filters.clone(),
840 )
841 .should_process_path(path)
842}
843
844fn create_notify_watcher(
845 event_tx: mpsc::Sender<PathBuf>,
846 dropped_events: Arc<AtomicU64>,
847) -> Result<RecommendedWatcher, String> {
848 let tx = event_tx;
849 let watcher = RecommendedWatcher::new(
850 move |result: Result<notify::Event, notify::Error>| {
851 if let Ok(event) = result {
852 match event.kind {
853 EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => {
854 for path in &event.paths {
855 if let Err(e) = tx.try_send(path.clone()) {
856 dropped_events.fetch_add(1, Ordering::Relaxed);
857 tracing::debug!(
858 path = %path.display(),
859 error = %e,
860 "watcher event dropped: debounce channel full or closed"
861 );
862 }
863 }
864 }
865 _ => {}
866 }
867 }
868 },
869 Config::default(),
870 )
871 .map_err(|error| format!("failed to create file watcher: {error}"))?;
872
873 Ok(watcher)
874}
875
876async fn emit_diagnostics(
877 state: &Arc<RwLock<WatcherInternalState>>,
878 diag_tx: &watch::Sender<WatcherDiagnostics>,
879 dropped_events: &Arc<AtomicU64>,
880) {
881 let state_guard = state.read().await;
882 let current = diag_tx.borrow().clone();
883 let updated = WatcherDiagnostics {
884 watched_repository_count: state_guard.repositories.len(),
885 total_events_received: state_guard.events_received,
886 total_events_filtered: state_guard.events_filtered,
887 total_index_tasks_queued: state_guard.index_tasks_queued,
888 total_events_dropped: dropped_events.load(Ordering::Relaxed),
889 ..current
890 };
891 let _ = diag_tx.send(updated);
892}
893
894async fn update_diagnostics_failed(
895 diag_tx: &watch::Sender<WatcherDiagnostics>,
896 state: &Arc<RwLock<WatcherInternalState>>,
897 dropped_events: &Arc<AtomicU64>,
898 error: &str,
899) {
900 let mut current = diag_tx.borrow().clone();
901 current.state = WatcherState::Failed;
902 current.last_error = Some(error.to_owned());
903 let state_guard = state.read().await;
904 current.watched_repository_count = state_guard.repositories.len();
905 current.total_events_received = state_guard.events_received;
906 current.total_events_filtered = state_guard.events_filtered;
907 current.total_index_tasks_queued = state_guard.index_tasks_queued;
908 current.total_events_dropped = dropped_events.load(Ordering::Relaxed);
909 let _ = diag_tx.send(current);
910}
911
912async fn update_diagnostics_degraded(
913 diag_tx: &watch::Sender<WatcherDiagnostics>,
914 state: &Arc<RwLock<WatcherInternalState>>,
915 dropped_events: &Arc<AtomicU64>,
916 reason: &str,
917) {
918 let mut current = diag_tx.borrow().clone();
919 current.state = WatcherState::Degraded;
920 current.degraded_reason = Some(reason.to_owned());
921 let state_guard = state.read().await;
922 current.watched_repository_count = state_guard.repositories.len();
923 current.total_events_received = state_guard.events_received;
924 current.total_events_filtered = state_guard.events_filtered;
925 current.total_index_tasks_queued = state_guard.index_tasks_queued;
926 current.total_events_dropped = dropped_events.load(Ordering::Relaxed);
927 let _ = diag_tx.send(current);
928}
929
930#[cfg(test)]
931#[path = "engine_tests.rs"]
932mod tests;