fix(#1643): filter destroyed-agent targets from dashboard schedule view
This commit is contained in:
parent
8b991b2cc5
commit
e08122a206
4 changed files with 117 additions and 8 deletions
|
|
@ -696,6 +696,10 @@ pub async fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) -> Resul
|
||||||
// purge — recompute either way).
|
// purge — recompute either way).
|
||||||
coord.rescan_containers_and_emit().await;
|
coord.rescan_containers_and_emit().await;
|
||||||
crate::dashboard::emit_tombstones_snapshot(coord).await;
|
crate::dashboard::emit_tombstones_snapshot(coord).await;
|
||||||
|
// Re-emit the schedules snapshot: the rescan above refreshed the live
|
||||||
|
// roster, so any schedule that still targets the just-destroyed agent
|
||||||
|
// now drops that ghost column live (no page reload needed).
|
||||||
|
coord.emit_schedules_snapshot();
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,7 @@
|
||||||
//! socket, and the per-agent sockets: the broker, configured `agent_flake`,
|
//! socket, and the per-agent sockets: the broker, configured `agent_flake`,
|
||||||
//! and the map of registered agent sockets.
|
//! and the map of registered agent sockets.
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::{HashMap, HashSet};
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::atomic::{AtomicU64, Ordering};
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
|
|
@ -512,7 +512,7 @@ impl Coordinator {
|
||||||
/// after each tick that fires or rearms a row) so the dashboard's
|
/// after each tick that fires or rearms a row) so the dashboard's
|
||||||
/// scheduled-prompts tab updates live without polling.
|
/// scheduled-prompts tab updates live without polling.
|
||||||
pub fn emit_schedules_snapshot(self: &Arc<Self>) {
|
pub fn emit_schedules_snapshot(self: &Arc<Self>) {
|
||||||
let schedules = match self.scheduled_prompts.list() {
|
let mut schedules: Vec<hive_sh4re::WireSchedule> = match self.scheduled_prompts.list() {
|
||||||
Ok(rows) => rows
|
Ok(rows) => rows
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(crate::manager_server::schedule_to_wire_public)
|
.map(crate::manager_server::schedule_to_wire_public)
|
||||||
|
|
@ -522,12 +522,34 @@ impl Coordinator {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
// Strip ghost targets (destroyed agents) so the dashboard doesn't
|
||||||
|
// render dead columns. Best-effort: if the roster cache is
|
||||||
|
// momentarily contended we emit unfiltered rather than block this
|
||||||
|
// sync path — the next snapshot / page reload corrects it.
|
||||||
|
if let Some(live) = self.live_container_names_blocking() {
|
||||||
|
crate::manager_server::filter_ghost_schedule_targets(&mut schedules, &live);
|
||||||
|
}
|
||||||
self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
|
self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
|
||||||
seq: self.next_seq(),
|
seq: self.next_seq(),
|
||||||
schedules,
|
schedules,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Best-effort synchronous snapshot of live container names (the
|
||||||
|
/// logical agent names from the last `nixos-container list` scan).
|
||||||
|
/// Returns `None` when the roster cache lock is momentarily
|
||||||
|
/// contended, so a transient miss is treated as "roster unknown,
|
||||||
|
/// don't filter" rather than hiding live schedule targets. The async
|
||||||
|
/// `containers_snapshot` is the reliable path for request handlers;
|
||||||
|
/// this exists for the sync `emit_schedules_snapshot` SSE emit.
|
||||||
|
#[must_use]
|
||||||
|
pub fn live_container_names_blocking(&self) -> Option<HashSet<String>> {
|
||||||
|
self.last_containers
|
||||||
|
.try_lock()
|
||||||
|
.ok()
|
||||||
|
.map(|m| m.keys().cloned().collect())
|
||||||
|
}
|
||||||
|
|
||||||
/// Emit a `RemindersChanged` snapshot event. Called from every
|
/// Emit a `RemindersChanged` snapshot event. Called from every
|
||||||
/// reminder mutation site (agent `remind` calls, operator cancel /
|
/// reminder mutation site (agent `remind` calls, operator cancel /
|
||||||
/// retry, and the scheduler after each delivery batch) so the
|
/// retry, and the scheduler after each delivery batch) so the
|
||||||
|
|
|
||||||
|
|
@ -18,12 +18,24 @@ use super::{AppState, error_response};
|
||||||
/// so the frontend can render without an extra translation layer.
|
/// so the frontend can render without an extra translation layer.
|
||||||
pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
|
pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
|
||||||
match state.coord.scheduled_prompts.list() {
|
match state.coord.scheduled_prompts.list() {
|
||||||
Ok(rows) => axum::Json(
|
Ok(rows) => {
|
||||||
rows.into_iter()
|
let mut wire: Vec<hive_sh4re::WireSchedule> = rows
|
||||||
|
.into_iter()
|
||||||
.map(crate::manager_server::schedule_to_wire_public)
|
.map(crate::manager_server::schedule_to_wire_public)
|
||||||
.collect::<Vec<_>>(),
|
.collect();
|
||||||
)
|
// Drop ghost targets (agents that no longer exist) so the
|
||||||
.into_response(),
|
// table never shows dead columns. Uses the reliable async
|
||||||
|
// roster snapshot here on the request path.
|
||||||
|
let live: std::collections::HashSet<String> = state
|
||||||
|
.coord
|
||||||
|
.containers_snapshot()
|
||||||
|
.await
|
||||||
|
.into_iter()
|
||||||
|
.map(|c| c.name)
|
||||||
|
.collect();
|
||||||
|
crate::manager_server::filter_ghost_schedule_targets(&mut wire, &live);
|
||||||
|
axum::Json(wire).into_response()
|
||||||
|
}
|
||||||
Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
|
Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -831,6 +831,27 @@ pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh
|
||||||
schedule_to_wire(s)
|
schedule_to_wire(s)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Drop schedule targets that point at agents which no longer exist, so
|
||||||
|
/// the dashboard's schedule table doesn't render ghost columns for
|
||||||
|
/// destroyed agents. `live` is the set of logical agent names from the
|
||||||
|
/// last `nixos-container list` scan (stopped agents included, destroyed
|
||||||
|
/// ones absent); the `operator` pseudo-target is always retained since
|
||||||
|
/// it isn't a container. Applied only to the dashboard wire paths
|
||||||
|
/// (`api_schedules` + the `SchedulesChanged` SSE emit) — the
|
||||||
|
/// manager-facing `list_schedules` stays unfiltered so agents can still
|
||||||
|
/// see and cancel stale targets. This is a view filter: the underlying
|
||||||
|
/// schedule rows keep every target, so a re-spawned agent's targets
|
||||||
|
/// reappear on their own.
|
||||||
|
pub(crate) fn filter_ghost_schedule_targets(
|
||||||
|
schedules: &mut [hive_sh4re::WireSchedule],
|
||||||
|
live: &std::collections::HashSet<String>,
|
||||||
|
) {
|
||||||
|
for s in schedules.iter_mut() {
|
||||||
|
s.targets
|
||||||
|
.retain(|t| t.target == hive_sh4re::OPERATOR_RECIPIENT || live.contains(&t.target));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
|
fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
|
||||||
hive_sh4re::WireSchedule {
|
hive_sh4re::WireSchedule {
|
||||||
id: s.id,
|
id: s.id,
|
||||||
|
|
@ -906,7 +927,57 @@ pub fn spawn_question_watchdog(coord: &Arc<Coordinator>, id: i64, ttl_secs: u64)
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::validate_commit_ref;
|
use super::{filter_ghost_schedule_targets, validate_commit_ref};
|
||||||
|
|
||||||
|
fn target(name: &str) -> hive_sh4re::WireScheduleTarget {
|
||||||
|
hive_sh4re::WireScheduleTarget {
|
||||||
|
target: name.to_owned(),
|
||||||
|
cancelled_at_unix: None,
|
||||||
|
last_fired_at_unix: None,
|
||||||
|
last_result: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn schedule(targets: &[&str]) -> hive_sh4re::WireSchedule {
|
||||||
|
hive_sh4re::WireSchedule {
|
||||||
|
id: 1,
|
||||||
|
owner: "operator".to_owned(),
|
||||||
|
body: "ping".to_owned(),
|
||||||
|
interval_seconds: None,
|
||||||
|
next_fire_at_unix: 0,
|
||||||
|
created_at_unix: 0,
|
||||||
|
source: hive_sh4re::WireScheduleSource::Operator,
|
||||||
|
cancelled_at_unix: None,
|
||||||
|
description: None,
|
||||||
|
targets: targets.iter().map(|t| target(t)).collect(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ghost_filter_drops_dead_agents_keeps_live_and_operator() {
|
||||||
|
let live: std::collections::HashSet<String> = ["iris".to_owned(), "damocles".to_owned()]
|
||||||
|
.into_iter()
|
||||||
|
.collect();
|
||||||
|
let mut schedules = vec![schedule(&["iris", "ghost", "operator", "damocles"])];
|
||||||
|
filter_ghost_schedule_targets(&mut schedules, &live);
|
||||||
|
let kept: Vec<&str> = schedules[0]
|
||||||
|
.targets
|
||||||
|
.iter()
|
||||||
|
.map(|t| t.target.as_str())
|
||||||
|
.collect();
|
||||||
|
// `ghost` (destroyed) dropped; live agents + operator pseudo-target kept.
|
||||||
|
assert_eq!(kept, vec!["iris", "operator", "damocles"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ghost_filter_can_empty_targets_when_all_dead() {
|
||||||
|
let live: std::collections::HashSet<String> = std::collections::HashSet::new();
|
||||||
|
let mut schedules = vec![schedule(&["gone1", "gone2"])];
|
||||||
|
filter_ghost_schedule_targets(&mut schedules, &live);
|
||||||
|
// operator is never in the live set but is always retained; here
|
||||||
|
// there's no operator target, so everything drops.
|
||||||
|
assert!(schedules[0].targets.is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn accepts_short_and_full_sha() {
|
fn accepts_short_and_full_sha() {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue