fs: Improve filesystem rescan diagnostics (#64253)
## Summary Replace the broken rescan-warning rate limiter, which repeatedly emitted “not logging more” without advancing its counter. - Keep the last 10 queued path occurrences per watcher backend, preserving full paths. - Build and write rescan reports in the existing background dispatch task, keeping blocking log I/O off the native watcher reader. - Emit one error line per dispatched rescan after coalescing duplicate pathless rescans and before registration fan-out, containing the backend-provided reason, affected paths, and preceding paths. - Remove the redundant rescan warnings and rate-limit bookkeeping without changing rescan scheduling. - Add one focused test for the path history and report contents. ## Verification - `corgi test -p fs --lib` — 23 tests passed. - `corgi test -p fs --features test-support --test integration` — 24 tests passed. - `corgi fmt -p fs -- --check` - `git diff --check` Release Notes: - Improved filesystem watcher rescan diagnostics and removed repetitive “not logging more” warnings.
This commit is contained in:
1 parent
eb655653f2
commit
8f843d073d
2 files changed
+75
-35
No files matched your search
+18
-35
@@ -695,10 +695,6 @@ fn push_notify_event(
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
if event.need_rescan() {
|
||||
if !watcher_logging_rate_limited() {
|
||||
log::warn!("filesystem watcher lost sync for {watched_root:?}; scheduling rescan");
|
||||
}
|
||||
|
||||
// A rescan event names the subtrees that lost sync (FSEvents
|
||||
// `MustScanSubDirs`); only a pathless one (inotify queue overflow) means
|
||||
// the whole watched root is suspect.
|
||||
@@ -717,28 +713,6 @@ fn push_notify_event(
|
||||
enqueue_path_events(tx, pending_path_events, path_events);
|
||||
}
|
||||
|
||||
fn watcher_logging_rate_limited() -> bool {
|
||||
static LAST_WARN: Mutex<Option<(Instant, usize)>> = Mutex::new(None);
|
||||
let Some((ref mut started, ref mut emitted)) = *LAST_WARN.lock() else {
|
||||
*LAST_WARN.lock() = Some((Instant::now(), 0));
|
||||
return false;
|
||||
};
|
||||
|
||||
if started.elapsed().as_secs() < 1 {
|
||||
if *emitted < 20 {
|
||||
log::warn!("filesystem watcher lost sync for many files, not logging more");
|
||||
return true;
|
||||
} else {
|
||||
*emitted += 1;
|
||||
}
|
||||
} else {
|
||||
*emitted = 0;
|
||||
*started = Instant::now()
|
||||
}
|
||||
|
||||
true
|
||||
}
|
||||
|
||||
fn coalesce_pending_rescans(pending_paths: &mut Vec<PathEvent>, path_events: &mut Vec<PathEvent>) {
|
||||
if !path_events
|
||||
.iter()
|
||||
@@ -939,8 +913,9 @@ impl OsWatcher {
|
||||
let dispatch_task = executor.spawn({
|
||||
let state = state.clone();
|
||||
async move {
|
||||
let mut rescan_history = diagnostics::RescanHistory::default();
|
||||
while let Ok(first) = event_rx.recv().await {
|
||||
dispatch_batch(kind, &state, first, &event_rx);
|
||||
dispatch_batch(kind, &state, &mut rescan_history, first, &event_rx);
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -1058,7 +1033,13 @@ impl OsWatcher {
|
||||
first: notify::Result<notify::Event>,
|
||||
event_rx: &async_channel::Receiver<notify::Result<notify::Event>>,
|
||||
) {
|
||||
dispatch_batch(self.kind, &self.state, first, event_rx);
|
||||
dispatch_batch(
|
||||
self.kind,
|
||||
&self.state,
|
||||
&mut diagnostics::RescanHistory::default(),
|
||||
first,
|
||||
event_rx,
|
||||
);
|
||||
}
|
||||
|
||||
fn start_native_watch_limit_cooldown(&self, path: &Path) {
|
||||
@@ -1219,16 +1200,11 @@ fn dispatch(
|
||||
let callbacks = {
|
||||
let state = state.lock();
|
||||
if event.need_rescan() {
|
||||
let callbacks = state
|
||||
state
|
||||
.watchers
|
||||
.values()
|
||||
.map(|registration| registration.callback.clone())
|
||||
.collect::<Vec<_>>();
|
||||
log::warn!(
|
||||
"{kind:?} filesystem watcher lost sync; scheduling rescans for {} registrations",
|
||||
callbacks.len()
|
||||
);
|
||||
callbacks
|
||||
.collect::<Vec<_>>()
|
||||
} else {
|
||||
let mut ids = Vec::new();
|
||||
for path in &event.paths {
|
||||
@@ -1252,6 +1228,7 @@ fn dispatch(
|
||||
fn dispatch_batch(
|
||||
kind: OsWatcherKind,
|
||||
state: &Mutex<WatcherState>,
|
||||
rescan_history: &mut diagnostics::RescanHistory,
|
||||
first: notify::Result<notify::Event>,
|
||||
event_rx: &async_channel::Receiver<notify::Result<notify::Event>>,
|
||||
) {
|
||||
@@ -1272,6 +1249,12 @@ fn dispatch_batch(
|
||||
rescan_dispatched = true;
|
||||
}
|
||||
|
||||
// Log writes can block, so keep them off the native watcher reader.
|
||||
if let Ok(event) = &event
|
||||
&& let Some(report) = rescan_history.record(event)
|
||||
{
|
||||
log::error!("{kind:?} filesystem watcher requested rescan: {report}");
|
||||
}
|
||||
dispatch(kind, state, event);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ use std::{
|
||||
};
|
||||
|
||||
const EVENT_CAPACITY: usize = 10_000;
|
||||
const RESCAN_PATH_HISTORY_CAPACITY: usize = 10;
|
||||
|
||||
/// An opt-in recording. Watchers hold only weak references to its buffer, so
|
||||
/// dropping this value stops collection without changing any watches.
|
||||
@@ -122,6 +123,37 @@ impl DiagnosticRecorder {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub(super) struct RescanHistory {
|
||||
paths: VecDeque<PathBuf>,
|
||||
}
|
||||
|
||||
impl RescanHistory {
|
||||
pub(super) fn record(&mut self, event: ¬ify::Event) -> Option<String> {
|
||||
let report = event.need_rescan().then(|| {
|
||||
// Capture before inserting the rescan so the report includes all ten
|
||||
// preceding paths, even when this is the start of an overflow burst.
|
||||
format!(
|
||||
"reason={:?}, paths={:?}, recent_paths={:?}",
|
||||
event.info().unwrap_or("unspecified"),
|
||||
event.paths,
|
||||
self.paths,
|
||||
)
|
||||
});
|
||||
let first_path = event
|
||||
.paths
|
||||
.len()
|
||||
.saturating_sub(RESCAN_PATH_HISTORY_CAPACITY);
|
||||
for path in event.paths.iter().skip(first_path) {
|
||||
if self.paths.len() == RESCAN_PATH_HISTORY_CAPACITY {
|
||||
self.paths.pop_front();
|
||||
}
|
||||
self.paths.push_back(path.clone());
|
||||
}
|
||||
report
|
||||
}
|
||||
}
|
||||
|
||||
impl WatchRecording {
|
||||
pub(crate) fn new(watchers: impl IntoIterator<Item = Arc<OsWatcher>>) -> Self {
|
||||
let state = Arc::new(Mutex::new(RecordingState {
|
||||
@@ -203,6 +235,31 @@ mod tests {
|
||||
use notify::{Event, EventKind, event::Flag};
|
||||
use std::path::Path;
|
||||
|
||||
#[test]
|
||||
fn rescan_report_contains_ten_preceding_paths_and_reason() {
|
||||
let mut history = RescanHistory::default();
|
||||
for range in [0..12, 12..13] {
|
||||
let mut event = Event::new(EventKind::Create(notify::event::CreateKind::File));
|
||||
event.paths = range
|
||||
.map(|index| PathBuf::from(format!("file-{index}")))
|
||||
.collect();
|
||||
assert!(history.record(&event).is_none());
|
||||
}
|
||||
let rescan = Event::new(EventKind::Other)
|
||||
.set_flag(Flag::Rescan)
|
||||
.set_info("rescan: kernel dropped")
|
||||
.add_path("root".into());
|
||||
let report = history
|
||||
.record(&rescan)
|
||||
.expect("the first rescan should produce a report");
|
||||
|
||||
assert_eq!(
|
||||
report,
|
||||
r#"reason="rescan: kernel dropped", paths=["root"], recent_paths=["file-3", "file-4", "file-5", "file-6", "file-7", "file-8", "file-9", "file-10", "file-11", "file-12"]"#
|
||||
);
|
||||
assert_eq!(history.paths.len(), RESCAN_PATH_HISTORY_CAPACITY);
|
||||
}
|
||||
|
||||
struct Backend;
|
||||
|
||||
impl WatchBackend for Backend {
|
||||
|
||||
Reference in new issue
Block a user