use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use notify::{
event::{EventKind, ModifyKind},
RecursiveMode, Watcher,
};
const EWMA_DECAY_PER_SEC: f64 = 0.87;
const PRUNE_IDLE: Duration = Duration::from_secs(30);
const MAX_TRACKED: usize = 4096;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ActivityKind {
Modified,
Created,
Removed,
Metadata,
Renamed,
Other,
}
impl ActivityKind {
pub fn label(&self) -> &'static str {
match self {
ActivityKind::Modified => "modify",
ActivityKind::Created => "create",
ActivityKind::Removed => "remove",
ActivityKind::Metadata => "meta",
ActivityKind::Renamed => "rename",
ActivityKind::Other => "other",
}
}
fn from_event(kind: &EventKind) -> Self {
match kind {
EventKind::Create(_) => ActivityKind::Created,
EventKind::Remove(_) => ActivityKind::Removed,
EventKind::Modify(ModifyKind::Name(_)) => ActivityKind::Renamed,
EventKind::Modify(ModifyKind::Metadata(_)) => ActivityKind::Metadata,
EventKind::Modify(_) => ActivityKind::Modified,
_ => ActivityKind::Other,
}
}
}
#[derive(Debug, Clone)]
pub struct FileActivity {
pub path: PathBuf,
pub events_per_sec: f64,
pub total_events: u64,
pub last_kind: ActivityKind,
pub last_seen: Instant,
}
#[derive(Default)]
pub struct HotFileState {
pub activity: HashMap<PathBuf, FileActivity>,
pub total_events: u64,
pub watch_roots: Vec<PathBuf>,
pub error: Option<String>,
}
impl HotFileState {
fn record(&mut self, path: PathBuf, kind: ActivityKind) {
self.total_events += 1;
let now = Instant::now();
let entry = self
.activity
.entry(path.clone())
.or_insert_with(|| FileActivity {
path,
events_per_sec: 0.0,
total_events: 0,
last_kind: kind,
last_seen: now,
});
entry.total_events += 1;
entry.last_kind = kind;
entry.events_per_sec += 1.0;
entry.last_seen = now;
}
}
pub struct HotFileWatcher {
pub state: Arc<Mutex<HotFileState>>,
_watcher: Option<notify::RecommendedWatcher>,
}
impl HotFileWatcher {
pub fn start(roots: &[&Path]) -> Self {
let state = Arc::new(Mutex::new(HotFileState::default()));
let mut s = state.lock().unwrap();
s.watch_roots = roots.iter().map(|p| p.to_path_buf()).collect();
drop(s);
let state_w = state.clone();
let watcher_result =
notify::recommended_watcher(move |result: notify::Result<notify::Event>| {
let Ok(event) = result else { return };
let kind = ActivityKind::from_event(&event.kind);
let Ok(mut s) = state_w.lock() else { return };
for p in event.paths {
s.record(p, kind);
}
});
let mut watcher = match watcher_result {
Ok(w) => w,
Err(e) => {
state.lock().unwrap().error = Some(format!("watcher init failed: {}", e));
return Self {
state,
_watcher: None,
};
}
};
for r in roots {
if let Err(e) = watcher.watch(r, RecursiveMode::Recursive) {
state.lock().unwrap().error =
Some(format!("failed to watch {}: {}", r.display(), e));
}
}
Self {
state,
_watcher: Some(watcher),
}
}
pub fn decay(&self, elapsed: Duration) {
let mut s = self.state.lock().unwrap();
let now = Instant::now();
let factor = EWMA_DECAY_PER_SEC.powf(elapsed.as_secs_f64());
s.activity.retain(|_, a| {
if now.duration_since(a.last_seen) > PRUNE_IDLE {
return false;
}
a.events_per_sec *= factor;
if a.events_per_sec < 0.01 {
a.events_per_sec = 0.0;
}
true
});
if s.activity.len() > MAX_TRACKED {
let mut by_age: Vec<(PathBuf, Instant)> = s
.activity
.iter()
.map(|(k, v)| (k.clone(), v.last_seen))
.collect();
by_age.sort_by_key(|(_, t)| *t);
let drop_n = s.activity.len() - MAX_TRACKED;
for (k, _) in by_age.into_iter().take(drop_n) {
s.activity.remove(&k);
}
}
}
pub fn top(&self, n: usize) -> Vec<FileActivity> {
let s = self.state.lock().unwrap();
let mut v: Vec<FileActivity> = s.activity.values().cloned().collect();
v.sort_by(|a, b| {
b.events_per_sec
.partial_cmp(&a.events_per_sec)
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| b.total_events.cmp(&a.total_events))
});
v.truncate(n);
v
}
pub fn snapshot_meta(&self) -> (u64, Vec<PathBuf>, Option<String>) {
let s = self.state.lock().unwrap();
(s.total_events, s.watch_roots.clone(), s.error.clone())
}
}
pub fn default_roots() -> Vec<PathBuf> {
let mut roots = Vec::new();
if let Some(home) = std::env::var_os("HOME") {
roots.push(PathBuf::from(home));
}
#[cfg(target_os = "macos")]
{
roots.push(PathBuf::from("/private/var/log"));
roots.push(PathBuf::from("/private/tmp"));
}
#[cfg(target_os = "linux")]
{
roots.push(PathBuf::from("/var/log"));
roots.push(PathBuf::from("/tmp"));
}
roots
}