hive-agent: no-pressure hint when state dir top level hits 30+ entries
New state_entry_watch.rs, mirroring disk_watch.rs's shape exactly: periodic in-process probe, Todos::upsert with a stable count-bucketed summary (anti-nag - drifting inside one bucket stays silent, crossing a bucket speaks up again), clears once back under threshold. Top-level entry count only, deliberately - a large subdirectory (git clone, build tree) counts as one entry regardless of what's inside it, which is disk_watch's problem to catch on its own axis (bytes), not this one's. Wired into main.rs's spawn_todo_socket alongside disk_watch::run. closes #3464
This commit is contained in:
parent
41f0d7e036
commit
301a576feb
2 changed files with 192 additions and 0 deletions
|
|
@ -24,6 +24,7 @@ mod questions;
|
||||||
mod reminder_timer;
|
mod reminder_timer;
|
||||||
mod reminders;
|
mod reminders;
|
||||||
mod serve_common;
|
mod serve_common;
|
||||||
|
mod state_entry_watch;
|
||||||
mod stats;
|
mod stats;
|
||||||
mod stream_enrich;
|
mod stream_enrich;
|
||||||
mod todo_server;
|
mod todo_server;
|
||||||
|
|
@ -522,6 +523,9 @@ fn spawn_todo_socket(
|
||||||
// shares this store + wake directly instead of dialling the
|
// shares this store + wake directly instead of dialling the
|
||||||
// socket the out-of-process producers use.
|
// socket the out-of-process producers use.
|
||||||
tokio::spawn(disk_watch::run(store.clone(), todo_wake.clone()));
|
tokio::spawn(disk_watch::run(store.clone(), todo_wake.clone()));
|
||||||
|
// Same shape, different signal: nudge on a crowded state-dir
|
||||||
|
// top level instead of disk pressure.
|
||||||
|
tokio::spawn(state_entry_watch::run(store.clone(), todo_wake.clone()));
|
||||||
Some(store)
|
Some(store)
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
|
|
||||||
188
hive-agent/src/state_entry_watch.rs
Normal file
188
hive-agent/src/state_entry_watch.rs
Normal file
|
|
@ -0,0 +1,188 @@
|
||||||
|
//! State-dir entry-count watch: raises a loose-ends-v2 todo when the
|
||||||
|
//! top level of this agent's own state directory accumulates 30+ entries,
|
||||||
|
//! as a low-pressure nudge to tidy up before it gets unwieldy — not a
|
||||||
|
//! hard limit, nothing here blocks or deletes anything.
|
||||||
|
//!
|
||||||
|
//! Same shape as [`crate::disk_watch`] (periodic in-process probe,
|
||||||
|
//! `Todos::upsert` with a *stable, bucketed* summary), reused rather than
|
||||||
|
//! reinvented: a producer whose only signal is "the count crossed a
|
||||||
|
//! threshold" wants the exact same anti-nag property disk pressure did.
|
||||||
|
//!
|
||||||
|
//! Anti-nag: the todo is keyed (`state`/[`TODO_KEY`]), and the summary is
|
||||||
|
//! bucketed by count ([`BUCKET`]) rather than carrying the raw number, so
|
||||||
|
//! `upsert` reports `changed == false` while the count drifts inside one
|
||||||
|
//! bucket (31 → 35 stays quiet; crossing 40 speaks up again). Dropping
|
||||||
|
//! back under [`WARN_COUNT`] clears the todo, same as `disk_watch`.
|
||||||
|
//!
|
||||||
|
//! Top-level only, deliberately: a single large subdirectory (a git
|
||||||
|
//! clone, a build output tree) counts as one entry here regardless of
|
||||||
|
//! what's inside it — that's a different problem `disk_watch` already
|
||||||
|
//! covers on its own axis (bytes, not directory-entry count).
|
||||||
|
|
||||||
|
use std::path::Path;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use tokio::sync::Notify;
|
||||||
|
|
||||||
|
use crate::todos::Todos;
|
||||||
|
|
||||||
|
/// Same cadence as `disk_watch` — no reason for this to poll more often
|
||||||
|
/// than disk pressure does, and the `read_dir` here is cheaper still.
|
||||||
|
const CHECK_INTERVAL: Duration = Duration::from_mins(15);
|
||||||
|
/// Entry count past which the hint appears. Matches the ask exactly
|
||||||
|
/// ("30+ entries").
|
||||||
|
const WARN_COUNT: u64 = 30;
|
||||||
|
/// Counts are reported in buckets this wide — the anti-nag knob, same
|
||||||
|
/// role as `disk_watch::BUCKET_PCT`.
|
||||||
|
const BUCKET: u64 = 10;
|
||||||
|
/// Producing subsystem marker for the todo row.
|
||||||
|
const SUBSYSTEM: &str = "state";
|
||||||
|
/// Fixed dedup key — there is only ever one state-entry-count todo per
|
||||||
|
/// agent.
|
||||||
|
const TODO_KEY: &str = "entries";
|
||||||
|
|
||||||
|
/// Background loop: periodically reconcile the state-entry-count todo.
|
||||||
|
/// Detached task — runs for the harness's lifetime; errors are logged,
|
||||||
|
/// never fatal.
|
||||||
|
pub async fn run(todos: Arc<Todos>, wake: Arc<Notify>) {
|
||||||
|
loop {
|
||||||
|
tokio::time::sleep(CHECK_INTERVAL).await;
|
||||||
|
let summary = match tokio::task::spawn_blocking(probe).await {
|
||||||
|
Ok(s) => s,
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!(error = %e, "state entry watch probe panicked");
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
reconcile(&todos, &wake, summary.as_deref());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Apply one probe result to the todo store: raise/refresh while over
|
||||||
|
/// threshold, clear once back under. Split from [`run`] so the store
|
||||||
|
/// interaction is testable without a timer.
|
||||||
|
fn reconcile(todos: &Todos, wake: &Notify, summary: Option<&str>) {
|
||||||
|
match summary {
|
||||||
|
Some(summary) => match todos.upsert(SUBSYSTEM, Some(TODO_KEY), summary, None, false) {
|
||||||
|
Ok((_, true)) => wake.notify_one(),
|
||||||
|
Ok((_, false)) => {}
|
||||||
|
Err(e) => tracing::warn!(error = ?e, "state entry todo upsert failed"),
|
||||||
|
},
|
||||||
|
None => {
|
||||||
|
if let Err(e) = todos.clear(SUBSYSTEM, Some(TODO_KEY)) {
|
||||||
|
tracing::warn!(error = ?e, "state entry todo clear failed");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One probe: count the top-level entries in the agent's state dir and,
|
||||||
|
/// when at/over threshold, render the hint. `None` means "nothing to
|
||||||
|
/// report" — under threshold, or the directory couldn't be read.
|
||||||
|
fn probe() -> Option<String> {
|
||||||
|
let count = top_level_count(&crate::paths::state_dir())?;
|
||||||
|
summary_for(count)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Pure rendering half of [`probe`]: the todo text for a given entry
|
||||||
|
/// count. Separated so the threshold and the *stability* of the summary
|
||||||
|
/// are unit-tested without a real filesystem.
|
||||||
|
fn summary_for(count: u64) -> Option<String> {
|
||||||
|
if count < WARN_COUNT {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
let bucket = count / BUCKET * BUCKET;
|
||||||
|
Some(format!(
|
||||||
|
"your state dir has {bucket}+ entries at the top level — worth a look if it's \
|
||||||
|
gotten unwieldy, no pressure. Splitting a long-lived file into dated pieces or \
|
||||||
|
moving one-off scratch out of the top level are both fine ways to trim it; \
|
||||||
|
nothing here is a hard limit and nothing gets deleted for you."
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Count entries directly inside `path` — non-recursive, so a large
|
||||||
|
/// subdirectory (a git clone, a build tree) still counts as exactly one
|
||||||
|
/// entry. `None` if the directory can't be read (matches `disk_watch`'s
|
||||||
|
/// treatment of a failed syscall: silence, not an error the agent has to
|
||||||
|
/// act on).
|
||||||
|
fn top_level_count(path: &Path) -> Option<u64> {
|
||||||
|
let entries = std::fs::read_dir(path).ok()?;
|
||||||
|
Some(entries.flatten().count() as u64)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn under_threshold_is_silent() {
|
||||||
|
assert!(summary_for(WARN_COUNT - 1).is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn at_threshold_speaks_up() {
|
||||||
|
let s = summary_for(WARN_COUNT).expect("todo");
|
||||||
|
assert!(s.contains("30+ entries"));
|
||||||
|
assert!(s.contains("no pressure"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The anti-nag property: a count drifting inside one bucket must
|
||||||
|
/// render byte-identically, so `Todos::upsert` reports
|
||||||
|
/// `changed == false` and nothing wakes the agent.
|
||||||
|
#[test]
|
||||||
|
fn drift_inside_a_bucket_is_identical() {
|
||||||
|
let a = summary_for(31).expect("todo");
|
||||||
|
let b = summary_for(39).expect("todo");
|
||||||
|
assert_eq!(a, b);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn crossing_a_bucket_changes_the_summary() {
|
||||||
|
let a = summary_for(39).expect("todo");
|
||||||
|
let b = summary_for(40).expect("todo");
|
||||||
|
assert_ne!(a, b);
|
||||||
|
assert!(b.contains("40+ entries"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The store round-trip: first raise wakes, an identical re-probe does
|
||||||
|
/// not, and dropping back under threshold clears the row.
|
||||||
|
#[test]
|
||||||
|
fn reconcile_wakes_once_then_clears() {
|
||||||
|
let dir = tempfile::tempdir().expect("tempdir");
|
||||||
|
let todos = Todos::open(&dir.path().join("state.sqlite")).expect("open");
|
||||||
|
let wake = Notify::new();
|
||||||
|
let summary = summary_for(WARN_COUNT).expect("todo");
|
||||||
|
|
||||||
|
reconcile(&todos, &wake, Some(&summary));
|
||||||
|
assert_eq!(todos.list(Some(SUBSYSTEM)).expect("list").len(), 1);
|
||||||
|
|
||||||
|
reconcile(&todos, &wake, Some(&summary));
|
||||||
|
assert_eq!(todos.list(Some(SUBSYSTEM)).expect("list").len(), 1);
|
||||||
|
|
||||||
|
reconcile(&todos, &wake, None);
|
||||||
|
assert!(todos.list(Some(SUBSYSTEM)).expect("list").is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn top_level_count_is_not_recursive() {
|
||||||
|
let dir = tempfile::tempdir().expect("tempdir");
|
||||||
|
std::fs::create_dir(dir.path().join("big_subdir")).expect("mkdir");
|
||||||
|
for i in 0..50 {
|
||||||
|
std::fs::write(dir.path().join("big_subdir").join(format!("f{i}")), b"x")
|
||||||
|
.expect("write");
|
||||||
|
}
|
||||||
|
std::fs::write(dir.path().join("top.txt"), b"x").expect("write");
|
||||||
|
// One subdirectory (however many files it hides) + one file = 2,
|
||||||
|
// not 51 — the whole point of "top level only".
|
||||||
|
assert_eq!(top_level_count(dir.path()), Some(2));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn missing_dir_is_silent_not_an_error() {
|
||||||
|
assert_eq!(
|
||||||
|
top_level_count(Path::new("/definitely/does/not/exist")),
|
||||||
|
None
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue