Krolyxon
Commit 4e844bb

Branches

Tags

Refs →

ferreniumsrcindexwatch.rs

266 lines · 9.4 KB · Rust

Raw Blame History Download
1//! Keeps the index fresh: inotify watches (debounced per repository), a
2//! periodic full rescan as a fallback, and rescans requested on SIGHUP.
3//!
4//! Watched paths:
5//! - each git dir, non-recursively (`HEAD`, `packed-refs`, `config`,
6//! `description`) and its `refs/` recursively; objects are never watched;
7//! - every non-repository directory walked during discovery, so repositories
8//! appearing or disappearing trigger a rescan.
9
10use std::collections::{BTreeSet, HashMap};
11use std::path::{Path, PathBuf};
12use std::sync::Arc;
13use std::time::Duration;
14
15use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
16use tokio::sync::mpsc;
17use tokio::time::{Instant, Interval, MissedTickBehavior};
18
19use super::{Refresh, RepoIndex, ScanStats};
20use crate::state::AppState;
21
22#[derive(Debug, Clone, PartialEq, Eq, Hash)]
23enum Task {
24 Rescan,
25 Repo(PathBuf),
26}
27
28/// Run forever. Call after the initial scan; `seen` holds the warnings that
29/// scan already logged.
30pub async fn run(state: Arc<AppState>, mut seen: BTreeSet<String>) {
31 let (tx, mut rx) = mpsc::unbounded_channel::<notify::Result<Event>>();
32 let mut fs = FsWatch::new(&state, tx);
33 fs.sync(&state.index.load());
34
35 let mut pending: HashMap<Task, Instant> = HashMap::new();
36 let mut poll_secs = state.config.load().watch.poll_interval_secs;
37 let mut poll = poll_timer(poll_secs);
38
39 loop {
40 let config = state.config.load_full();
41 if config.watch.poll_interval_secs != poll_secs {
42 poll_secs = config.watch.poll_interval_secs;
43 poll = poll_timer(poll_secs);
44 }
45 let debounce = Duration::from_millis(config.watch.debounce_ms);
46 let next_due = pending.values().min().copied();
47
48 tokio::select! {
49 Some(event) = rx.recv() => {
50 let index = state.index.load();
51 for task in classify(event, &index) {
52 pending.entry(task).or_insert_with(|| Instant::now() + debounce);
53 }
54 }
55 _ = state.rescan.notified() => {
56 pending.insert(Task::Rescan, Instant::now());
57 }
58 _ = async { poll.as_mut().unwrap().tick().await }, if poll.is_some() => {
59 pending.entry(Task::Rescan).or_insert_with(Instant::now);
60 }
61 _ = tokio::time::sleep_until(next_due.unwrap_or_else(Instant::now)), if next_due.is_some() => {
62 let now = Instant::now();
63 let due: Vec<Task> = pending
64 .iter()
65 .filter(|(_, at)| **at <= now)
66 .map(|(task, _)| task.clone())
67 .collect();
68 for task in &due {
69 pending.remove(task);
70 }
71 if due.contains(&Task::Rescan) {
72 // A full scan covers every per-repo refresh.
73 pending.retain(|task, _| *task == Task::Rescan);
74 let st = state.clone();
75 let stats = tokio::task::spawn_blocking(move || {
76 let config = st.config.load_full();
77 st.index.full_scan(&config)
78 })
79 .await;
80 if let Ok(stats) = stats {
81 let index = state.index.load();
82 log_scan(&stats, &index, &mut seen, false);
83 fs.sync(&index);
84 }
85 continue;
86 }
87 for task in due {
88 let Task::Repo(git_dir) = task else { continue };
89 let st = state.clone();
90 let outcome = tokio::task::spawn_blocking(move || {
91 let config = st.config.load_full();
92 st.index.refresh(&config, &git_dir)
93 })
94 .await;
95 match outcome {
96 Ok(Refresh::Updated(name)) => tracing::info!("updated {name}"),
97 Ok(Refresh::Hidden(name)) => tracing::info!("{name} is now hidden"),
98 Ok(Refresh::NeedsRescan) => {
99 pending.insert(Task::Rescan, Instant::now());
100 }
101 Ok(Refresh::Failed(e)) => tracing::warn!("refresh failed: {e}"),
102 Ok(Refresh::Unchanged) | Err(_) => {}
103 }
104 }
105 }
106 }
107 }
108}
109
110fn poll_timer(secs: u64) -> Option<Interval> {
111 (secs > 0).then(|| {
112 let period = Duration::from_secs(secs);
113 let mut interval = tokio::time::interval_at(Instant::now() + period, period);
114 interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
115 interval
116 })
117}
118
119/// Map a filesystem event to the work it implies.
120fn classify(event: notify::Result<Event>, index: &RepoIndex) -> Vec<Task> {
121 let event = match event {
122 Ok(event) => event,
123 Err(e) => {
124 tracing::warn!("watch error: {e}; rescanning");
125 return vec![Task::Rescan];
126 }
127 };
128 if event.need_rescan() {
129 return vec![Task::Rescan];
130 }
131 if matches!(event.kind, EventKind::Access(_)) {
132 return Vec::new();
133 }
134 let mut tasks = Vec::new();
135 for path in &event.paths {
136 if path.extension().is_some_and(|ext| ext == "lock") {
137 continue;
138 }
139 let task = if let Some(git_dir) = path.ancestors().find(|a| index.git_dirs.contains_key(*a))
140 {
141 Task::Repo(git_dir.to_path_buf())
142 } else if index
143 .dirs
144 .iter()
145 .any(|d| d == path || Some(d.as_path()) == path.parent())
146 {
147 Task::Rescan
148 } else {
149 continue;
150 };
151 if !tasks.contains(&task) {
152 tasks.push(task);
153 }
154 }
155 tasks
156}
157
158/// Log a scan: a summary when something changed (always for the initial
159/// scan) and each warning once until it goes away.
160pub fn log_scan(stats: &ScanStats, index: &RepoIndex, seen: &mut BTreeSet<String>, initial: bool) {
161 let ms = stats.elapsed.as_secs_f64() * 1000.0;
162 if initial {
163 tracing::info!(
164 "indexed {} repositories ({} hidden) in {ms:.1} ms",
165 stats.total,
166 stats.hidden
167 );
168 } else if stats.any_change() {
169 tracing::info!(
170 "rescan: {} repositories (+{} −{} ~{}) in {ms:.1} ms",
171 stats.total,
172 stats.added,
173 stats.removed,
174 stats.changed
175 );
176 } else {
177 tracing::debug!("rescan: no changes ({ms:.1} ms)");
178 }
179 let current: BTreeSet<String> = index.warnings.iter().cloned().collect();
180 for warning in current.difference(seen) {
181 tracing::warn!("{warning}");
182 }
183 *seen = current;
184}
185
186/// The inotify watcher and the set of paths it currently watches.
187struct FsWatch {
188 watcher: Option<RecommendedWatcher>,
189 watched: HashMap<PathBuf, RecursiveMode>,
190 warned_limit: bool,
191}
192
193impl FsWatch {
194 fn new(state: &AppState, tx: mpsc::UnboundedSender<notify::Result<Event>>) -> Self {
195 let watcher = if state.config.load().watch.enabled {
196 match notify::recommended_watcher(move |event| {
197 let _ = tx.send(event);
198 }) {
199 Ok(w) => Some(w),
200 Err(e) => {
201 tracing::warn!("cannot start file watcher ({e}); relying on periodic rescans");
202 None
203 }
204 }
205 } else {
206 None
207 };
208 FsWatch {
209 watcher,
210 watched: HashMap::new(),
211 warned_limit: false,
212 }
213 }
214
215 /// Bring the watched set in line with the index.
216 fn sync(&mut self, index: &RepoIndex) {
217 let Some(watcher) = self.watcher.as_mut() else {
218 return;
219 };
220 let mut desired: HashMap<PathBuf, RecursiveMode> = HashMap::new();
221 for dir in &index.dirs {
222 desired.insert(dir.clone(), RecursiveMode::NonRecursive);
223 }
224 for git_dir in index.git_dirs.keys() {
225 desired.insert(git_dir.clone(), RecursiveMode::NonRecursive);
226 desired.insert(git_dir.join("refs"), RecursiveMode::Recursive);
227 }
228
229 let stale: Vec<PathBuf> = self
230 .watched
231 .iter()
232 .filter(|(path, mode)| desired.get(*path) != Some(*mode))
233 .map(|(path, _)| path.clone())
234 .collect();
235 for path in stale {
236 let _ = watcher.unwatch(&path);
237 self.watched.remove(&path);
238 }
239 for (path, mode) in desired {
240 if self.watched.contains_key(&path) {
241 continue;
242 }
243 match watcher.watch(&path, mode) {
244 Ok(()) => {
245 self.watched.insert(path, mode);
246 }
247 Err(e) => watch_failed(&mut self.warned_limit, &path, e),
248 }
249 }
250 }
251}
252
253fn watch_failed(warned_limit: &mut bool, path: &Path, e: notify::Error) {
254 // ENOSPC from inotify_add_watch means the per-user watch limit.
255 let limit = matches!(&e.kind, notify::ErrorKind::MaxFilesWatch)
256 || matches!(&e.kind, notify::ErrorKind::Io(io) if io.raw_os_error() == Some(28));
257 if !limit {
258 tracing::debug!("cannot watch {}: {e}", path.display());
259 } else if !*warned_limit {
260 *warned_limit = true;
261 tracing::warn!(
262 "inotify watch limit reached; raise fs.inotify.max_user_watches \
263 (changes are still picked up by periodic rescans)"
264 );
265 }
266}