Skip to content

Commit 422b574

Browse files
fix(run): stop per-pass advisory churn on large catalogs
Two reconcile-pass costs repeat every pass even when nothing changed: - git-exclude spawned `git rev-parse` per op per pass and failed for workspaces that are plain directories (28 agents on dev3 → ~28 spawns plus captured git stderr every pass). Probe for a `.git` marker up the ancestor chain first — the same probe is_git_tracked already uses — and answer "not a Git worktree" without a process spawn. - print_report re-printed identical warnings every pass (~7k lines in hours on dev3). The supervisor loop now deduplicates warnings that persist across passes while still re-surfacing one that clears and returns; applied to both the catalog and spec loops. Part of #314. Co-authored-by: schickling-assistant <schickling-assistant@users.noreply.github.com> agent-identity: unknown agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.0.3 agent-runtime: OMP 18.0.3 tooling-profile: dotfiles@f33cd9c-dirty
1 parent 11bcf5d commit 422b574

2 files changed

Lines changed: 110 additions & 11 deletions

File tree

src/materialize.rs

Lines changed: 37 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -531,7 +531,8 @@ fn deep_merge(
531531
// longer states are removed, so a matcher group holding a user hook beside an st2
532532
// one keeps the user's; containers left with nothing but empty arrays are dropped.
533533
target.retain_mut(|element| {
534-
if patch.contains(element) || !supersede_managed_hooks
534+
if patch.contains(element)
535+
|| !supersede_managed_hooks
535536
|| !contains_owned_string(element)
536537
{
537538
return true;
@@ -602,7 +603,23 @@ fn contains_owned_string(value: &serde_json::Value) -> bool {
602603
}
603604
}
604605

606+
fn has_git_marker(workspace: &Path) -> bool {
607+
workspace
608+
.ancestors()
609+
.any(|ancestor| ancestor.join(".git").exists())
610+
}
611+
605612
fn git_exclude(workspace: &Path, line: &str) -> Result<bool> {
613+
// A failed `git rev-parse` costs a process spawn per op per pass. On catalogs whose
614+
// workspaces are plain directories this repeats every reconcile (measured live: ~28 spawns
615+
// plus stderr captures per pass on dev3). The marker probe answers "not a repo" for free;
616+
// `.git` may be a directory or a file (linked worktrees), which `exists` covers either way.
617+
if !has_git_marker(workspace) {
618+
anyhow::bail!(
619+
"{} is not a Git worktree (no .git marker)",
620+
workspace.display()
621+
);
622+
}
606623
let output = Command::new("git")
607624
.args(["-C"])
608625
.arg(workspace)
@@ -1169,6 +1186,25 @@ mod tests {
11691186
}
11701187
}
11711188

1189+
#[test]
1190+
fn git_exclude_reports_missing_repo_without_spawning_git() {
1191+
let dir = tempfile::tempdir().unwrap();
1192+
let workspace = dir.path().join("ws");
1193+
std::fs::create_dir_all(&workspace).unwrap();
1194+
1195+
assert!(!has_git_marker(&workspace));
1196+
let error = git_exclude(&workspace, ".st2/").unwrap_err();
1197+
assert!(
1198+
error.to_string().contains("no .git marker"),
1199+
"the no-repo case must be answered by the marker probe: {error:#}"
1200+
);
1201+
1202+
// A `.git` entry anywhere above the workspace (directory or file form, as linked
1203+
// worktrees use) re-enables the real git path.
1204+
std::fs::create_dir_all(dir.path().join(".git")).unwrap();
1205+
assert!(has_git_marker(&workspace));
1206+
}
1207+
11721208
#[test]
11731209
fn deep_merge_preserves_unrelated_keys_and_replaces_arrays() {
11741210
let mut target = serde_json::json!({

src/run.rs

Lines changed: 73 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1856,6 +1856,7 @@ pub fn up_loop_specs(
18561856
let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE);
18571857
let mut presentation_cursor = PresentationPatchCursor::default();
18581858
let mut reported_flapping: HashSet<String> = HashSet::new();
1859+
let mut recurring_warnings = RecurringWarnings::default();
18591860
let park_channel = ParkChannel::for_supervisor(root, this_host);
18601861
loop {
18611862
let mut pre = UpReport::default();
@@ -1874,15 +1875,7 @@ pub fn up_loop_specs(
18741875
reported_flapping.remove(id);
18751876
}
18761877
park_channel.publish(&cap, &mut report);
1877-
for cl in &report.crash_loops {
1878-
if reported_flapping.insert(cl.pty_id.clone()) {
1879-
eprintln!(
1880-
"st2: GAVE UP on '{id}' — crash-looping past its restart{{}} policy (mode=fail); leaving it parked and its last session for inspection. It is reported as parked by `st2 tasks`. Fix the cause, then `st2 unpark {id}` — no supervisor restart needed.",
1881-
id = cl.pty_id
1882-
);
1883-
surface_crash_loop(root, this_host, cl);
1884-
}
1885-
}
1878+
recurring_warnings.filter(&mut report);
18861879
on_report(&report);
18871880
if STOP.load(Ordering::SeqCst) {
18881881
break;
@@ -2048,6 +2041,24 @@ fn best_effort_catalog_watcher(
20482041
}
20492042
}
20502043

2044+
/// Suppresses warnings that persist across passes while still re-surfacing one that clears and
2045+
/// returns. An unchanged advisory failure (a non-Git workspace failing its git-exclude, say) must
2046+
/// be diagnosed once, not once per reconcile pass.
2047+
#[derive(Default)]
2048+
struct RecurringWarnings {
2049+
emitted: HashSet<String>,
2050+
}
2051+
2052+
impl RecurringWarnings {
2053+
fn filter(&mut self, report: &mut UpReport) {
2054+
let current: HashSet<_> = report.warnings.iter().cloned().collect();
2055+
self.emitted.retain(|warning| current.contains(warning));
2056+
report
2057+
.warnings
2058+
.retain(|warning| self.emitted.insert(warning.clone()));
2059+
}
2060+
}
2061+
20512062
/// The supervisor loop: reconcile on a timer AND on folder changes until interrupted. The fs-watch is
20522063
/// best-effort; the `interval` timer is the always-on fallback. `on_report` is called once per pass
20532064
pub fn up_loop(
@@ -2093,6 +2104,7 @@ fn up_loop_until(
20932104
// agent's supervisor over the native bus, so a crash-loop isn't only visible to whoever is
20942105
// watching the log.
20952106
let mut reported_flapping: HashSet<String> = HashSet::new();
2107+
let mut recurring_warnings = RecurringWarnings::default();
20962108
let park_channel = ParkChannel::for_supervisor(root, this_host);
20972109

20982110
loop {
@@ -2114,9 +2126,10 @@ fn up_loop_until(
21142126
}
21152127
// A recovered task that crash-loops again is a new crash-loop, so it must be able to surface
21162128
// again. Leaving the id in the dedup set would make every park after the first one silent.
2117-
for id in &report.unparked {
2129+
for id in report.unparked.iter() {
21182130
reported_flapping.remove(id);
21192131
}
2132+
recurring_warnings.filter(&mut report);
21202133
park_channel.publish(&cap, &mut report);
21212134
for cl in &report.crash_loops {
21222135
if reported_flapping.insert(cl.pty_id.clone()) {
@@ -2707,6 +2720,56 @@ mod tests {
27072720
);
27082721
}
27092722

2723+
#[cfg(target_os = "linux")]
2724+
#[test]
2725+
fn persistent_advisory_warnings_surface_once_not_per_pass() {
2726+
let catalog = tempfile::tempdir().unwrap();
2727+
let agent = catalog.path().join("agents/test-host/live");
2728+
std::fs::create_dir_all(&agent).unwrap();
2729+
std::fs::create_dir_all(catalog.path().join("workspace")).unwrap();
2730+
std::fs::write(
2731+
agent.join("agent.kdl"),
2732+
r#"agent "live" {
2733+
host "test-host"
2734+
command "true"
2735+
workspace "$CATALOG/workspace"
2736+
render { git-exclude "scratch.txt" }
2737+
}"#,
2738+
)
2739+
.unwrap();
2740+
let stop = AtomicBool::new(false);
2741+
let mut passes = 0usize;
2742+
let mut warnings_seen = 0usize;
2743+
2744+
std::thread::scope(|scope| {
2745+
scope.spawn(|| {
2746+
std::thread::sleep(Duration::from_millis(300));
2747+
stop.store(true, Ordering::SeqCst);
2748+
});
2749+
up_loop_until(
2750+
catalog.path(),
2751+
"test-host",
2752+
&SpawnCountingRunner::default(),
2753+
Duration::from_millis(50),
2754+
&stop,
2755+
|_, _| None,
2756+
|report| {
2757+
passes += 1;
2758+
warnings_seen += report.warnings.len();
2759+
},
2760+
)
2761+
.unwrap();
2762+
});
2763+
2764+
assert!(
2765+
passes >= 3,
2766+
"the loop must have run several passes for this to say anything: {passes}"
2767+
);
2768+
assert_eq!(
2769+
warnings_seen, 1,
2770+
"an unchanged advisory failure must be diagnosed once across {passes} passes"
2771+
);
2772+
}
27102773
/// A pass can execute a plan the task was never in: `up_once` drops an owner whose
27112774
/// materialization failed, `gate_harness_launches_on_hooks` strips gated launches, and
27122775
/// `defer_flickers` removes debounced ones — each after the pass is already committed to

0 commit comments

Comments
 (0)