vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

drop the hive-side matrix code: sweep, hivectl matrix and the apiUrl option

Browse source
This commit is contained in:
damocles 2026-10-28 22:16:16 +01:00 • committed by mara
commit 6dbb18b6d4
7 changed files with 49 additions and 1007 deletions
Download patch file Download diff file Expand all files Collapse all files

57
hive-c0re/src/job_queue/exec.rs
View file

@ -106,8 +106,6 @@ pub(super) trait Host: Sync {
fn purge_dirs(&self, name: &str) -> Vec<PathBuf>;
/// Publish the hive's forge credential for nix.
async fn forge_ensure_all(&self);
/// One matrix sender-token pass; whether the token is in place.
async fn matrix_ensure_all(&self) -> bool;
/// One `/knowledge` pull.
async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()>;
/// One swarm wanted-state pull.
@ -245,10 +243,6 @@ impl Host for RealHost {
crate::forge::ensure_all().await;
}
async fn matrix_ensure_all(&self) -> bool {
crate::matrix::ensure_all().await
}
async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()> {
crate::workers::knowledge::pull(coord).await
}
@ -350,7 +344,6 @@ async fn dispatch(
// until their whole subtree settles.
NodeKind::AgentWindow { .. } => Ok(()),
NodeKind::ForgeSweep => run_forge_sweep(host).await,
NodeKind::MatrixSweep => run_matrix_sweep(host, MATRIX_SWEEP_DEADLINE).await,
NodeKind::KnowledgePull => run_knowledge_pull(host, coord, KNOWLEDGE_PULL_DEADLINE).await,
NodeKind::WantedPull => run_wanted_pull(host, coord).await,
};
@ -367,56 +360,6 @@ async fn run_forge_sweep(host: &impl Host) -> Result<()> {
Ok(())
}
/// Matrix sender-token sweep as a DAG node — see [`NodeKind::MatrixSweep`].
/// Every pass reports here, whoever submitted it: as the node's own error on
/// the dashboard, and into the debounced warning.
async fn run_matrix_sweep(host: &impl Host, deadline: std::time::Duration) -> Result<()> {
let ok = tokio::time::timeout(deadline, host.matrix_ensure_all())
.await
.unwrap_or_else(|_| {
tracing::warn!(?deadline, "matrix: sweep timed out");
false
});
let mut health = matrix_sweep_health()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if ok {
health.record_ok();
Ok(())
} else {
health.record_err(matrix_sweep_warning);
anyhow::bail!("matrix ensure_all: one or more agents failed sync (see logs)")
}
}
/// A pass still running holds [`Resource::MatrixSweep`] against every later
/// one. Each HTTP call has its own 10 s timeout, but the call count grows with
/// the agent count and `lifecycle::list()` has none. A pass dropped here leaves
/// at worst a created but unpersisted room, which the next pass adopts by name.
///
/// [`Resource::MatrixSweep`]: super::resource::Resource::MatrixSweep
const MATRIX_SWEEP_DEADLINE: std::time::Duration = std::time::Duration::from_mins(10);
/// Warning for a matrix sweep that keeps failing. Two misses in a row, so one
/// bad pass (homeserver mid-restart, a transient HTTP blip) doesn't flap the
/// dashboard; cleared by the next clean pass.
fn matrix_sweep_health() -> &'static Mutex<SweepHealth> {
static HEALTH: OnceLock<Mutex<SweepHealth>> = OnceLock::new();
HEALTH.get_or_init(|| Mutex::new(SweepHealth::new("matrix_ensure_all", "warn", 2)))
}
fn matrix_sweep_warning(ctx: sweep_health::SweepFailure) -> String {
let age = ctx.since_last_ok.map_or_else(
|| "no success this session".to_owned(),
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
);
format!(
"matrix sender-token sweep failing ({} consecutive, {age}) \
— `hivectl matrix invite` may not work",
ctx.consecutive
)
}
/// `/knowledge` pull as a DAG node — see [`NodeKind::KnowledgePull`]. Every
/// pass reports here, whoever submitted it: as the node's own error on the
/// dashboard, and into the debounced warning.

74
hive-c0re/src/job_queue/exec/tests.rs
View file

@ -11,12 +11,11 @@ use hive_jobq::NodeId;
use hive_jobq::scheduler::Scheduler;
use super::{
Host, dispatch, drift, matrix_sweep_warning, run_create, run_destroy_bookkeeping,
run_destroy_container, run_drain, run_emit_rebuilt, run_forge_sweep, run_knowledge_pull,
run_matrix_sweep, run_meta_sync, run_pause_drain, run_pause_signal, run_prebuild,
run_provision, run_purge_state, run_rebuild_bookkeeping, run_reconcile, run_set_wanted,
run_signal, run_start, run_stop, run_stop_for_update, run_swap, run_wanted_pull,
run_write_dropin, substitute,
Host, dispatch, drift, run_create, run_destroy_bookkeeping, run_destroy_container, run_drain,
run_emit_rebuilt, run_forge_sweep, run_knowledge_pull, run_meta_sync, run_pause_drain,
run_pause_signal, run_prebuild, run_provision, run_purge_state, run_rebuild_bookkeeping,
run_reconcile, run_set_wanted, run_signal, run_start, run_stop, run_stop_for_update, run_swap,
run_wanted_pull, run_write_dropin, substitute,
};
use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
use crate::job_queue::model::NodeKind;
@ -43,8 +42,7 @@ struct FakeHost {
/// Store paths a prebuild had to build because the store lacked them.
builds: Mutex<Vec<String>>,
/// Names of the calls that fail (`kill`, `realise`, `prebuild`, `swap_update`,
/// `set_paused`, ...). `recorded_build` fails by finding nothing and
/// `matrix_ensure_all` by reporting false.
/// `set_paused`, ...). `recorded_build` fails by finding nothing.
failing: Vec<&'static str>,
/// The sweeps and pulls never finish, so only a deadline ends them.
hang: bool,
@ -277,14 +275,6 @@ impl Host for FakeHost {
self.record("forge_ensure_all".to_owned());
}
async fn matrix_ensure_all(&self) -> bool {
self.record("matrix_ensure_all".to_owned());
if self.hang {
std::future::pending::<()>().await;
}
!self.fails("matrix_ensure_all")
}
async fn knowledge_pull(&self, _coord: &Arc<Coordinator>) -> anyhow::Result<()> {
if self.hang {
std::future::pending::<()>().await;
@ -1088,31 +1078,6 @@ async fn forge_sweep_runs_once_and_never_fails() {
assert_eq!(host.calls(), ["forge_ensure_all"]);
}
#[tokio::test]
async fn matrix_sweep_fails_the_node_when_an_agent_failed_sync() {
run_matrix_sweep(&FakeHost::up(), Duration::from_secs(5))
.await
.expect("ok");
let err = run_matrix_sweep(
&FakeHost::up().failing("matrix_ensure_all"),
Duration::from_secs(5),
)
.await
.expect_err("failed sync");
assert!(format!("{err:#}").contains("failed sync"), "{err:#}");
}
#[tokio::test]
async fn matrix_sweep_past_its_deadline_fails_instead_of_hanging() {
let host = FakeHost {
hang: true,
..FakeHost::up()
};
run_matrix_sweep(&host, Duration::from_millis(50))
.await
.expect_err("deadline");
}
#[tokio::test]
async fn knowledge_pull_passes_the_result_through_and_times_out() {
let (_dir, coord) = coordinator();
@ -1274,7 +1239,6 @@ async fn dispatch_routes_the_agentless_and_hostless_kinds() {
NodeKind::ForgeSweep,
NodeKind::WantedPull,
NodeKind::KnowledgePull,
NodeKind::MatrixSweep,
alice(|agent| NodeKind::AgentWindow { agent }),
alice(|agent| NodeKind::Signal { agent }),
alice(|agent| NodeKind::Drain { agent }),
@ -1292,12 +1256,7 @@ async fn dispatch_routes_the_agentless_and_hostless_kinds() {
);
}
let calls = host.calls();
for want in [
"forge_ensure_all",
"wanted_pull",
"knowledge_pull",
"matrix_ensure_all",
] {
for want in ["forge_ensure_all", "wanted_pull", "knowledge_pull"] {
assert!(calls.iter().any(|c| c == want), "{want} in {calls:?}");
}
}
@ -1511,22 +1470,3 @@ async fn purge_keeps_going_past_a_dir_it_cannot_remove() {
assert!(stuck.exists(), "remove_dir_all refuses a plain file");
assert!(!applied.exists());
}
#[test]
fn matrix_sweep_warning_says_how_long_it_has_been_failing() {
use crate::stats::sweep_health::SweepFailure;
let never = matrix_sweep_warning(SweepFailure {
consecutive: 3,
since_last_ok: None,
});
assert!(
never.contains("3 consecutive, no success this session"),
"{never}"
);
let after_ok = matrix_sweep_warning(SweepFailure {
consecutive: 2,
since_last_ok: Some(Duration::from_secs(90)),
});
assert!(after_ok.contains("2 consecutive, last ok "), "{after_ok}");
assert!(!after_ok.contains("no success"), "{after_ok}");
}

8
hive-c0re/src/job_queue/model.rs
View file

@ -99,9 +99,6 @@ pub enum NodeKind {
/// One-shot boot-time publish of the hive's forge read token for nix
/// (`forge::ensure_all`). Agentless.
ForgeSweep,
/// Matrix sender-token sweep (`matrix::ensure_all`): at boot and every 30
/// minutes. Agentless.
MatrixSweep,
/// `/knowledge` pull (`knowledge::pull`): at boot, on the swarm
/// knowledge-changed event, and hourly. Agentless.
KnowledgePull,
@ -168,10 +165,7 @@ impl NodeKind {
| NodeKind::AgentWindow { agent }
| NodeKind::EmitRebuilt { agent, .. }
| NodeKind::SetWanted { agent, .. } => agent,
NodeKind::ForgeSweep
| NodeKind::MatrixSweep
| NodeKind::KnowledgePull
| NodeKind::WantedPull => "",
NodeKind::ForgeSweep | NodeKind::KnowledgePull | NodeKind::WantedPull => "",
}
}

4
hive-c0re/src/job_queue/resource.rs
View file

@ -31,9 +31,6 @@ pub enum Resource {
/// overlapping `reset` / `clean` / `pull` passes there fail on
/// `.git/index.lock` and the remote-tracking ref lock.
KnowledgeTree,
/// The hive's sender-token file, held by every `MatrixSweep`. Two
/// overlapping passes would race on writing it.
MatrixSweep,
}
/// This resource's name on the generic graph wire.
@ -49,7 +46,6 @@ impl hive_jobq_wire::WireResource for Resource {
Resource::Agent(agent) => format!("agent:{agent}"),
Resource::KnowledgeTree => "knowledge-tree".to_owned(),
Resource::MatrixSweep => "matrix-sweep".to_owned(),
}
}
}

25
hive-c0re/src/job_queue/scheduler.rs
View file

@ -245,31 +245,6 @@ mod tests {
.count()
}
#[tokio::test]
async fn two_passes_of_one_sweep_never_run_together() {
let q = JobQueue::new(1);
for _ in 0..2 {
q.insert_job(|b| vec![templates::matrix_sweep(b).guid()])
.expect("insert");
}
q.insert_job(|b| vec![templates::knowledge_pull(b).guid()])
.expect("insert");
let first = claim(&q).expect("a matrix pass is runnable");
let other = claim(&q).expect("a different sweep runs alongside it");
assert!(
claim(&q).is_none(),
"the second matrix pass waits for the first"
);
assert_eq!(running(&q), ["knowledge_pull", "matrix_sweep"]);
first.await;
other.await;
let second = claim(&q).expect("the second matrix pass runs once the first is done");
assert_eq!(running(&q), ["matrix_sweep"]);
second.await;
}
/// A tick landing while a pass is already live queues a second pass; the
/// shared `Resource::KnowledgeTree` dep holds it back until the first
/// finishes, so the two never run together.

8
hive-c0re/src/job_queue/templates.rs
View file

@ -370,14 +370,6 @@ pub fn destroy(builder: &JobBuilder, agent: &str, purge: bool) {
// `Reconcile` root per drifted agent) — no anchor node and no per-agent child
// DAGs.
/// One matrix sender-token sweep. Holding [`Resource::MatrixSweep`] is what keeps
/// two passes from overlapping, whichever caller submitted each.
pub fn matrix_sweep(builder: &JobBuilder) -> Handle<'_> {
builder
.node(NodeKind::MatrixSweep)
.needs(Resource::MatrixSweep)
}
/// One `/knowledge` pull, holding [`Resource::KnowledgeTree`] for its duration.
pub fn knowledge_pull(builder: &JobBuilder) -> Handle<'_> {
builder

8
hive-c0re/src/job_queue/tests.rs
View file

@ -1507,19 +1507,13 @@ fn destroy_container_must_not_suppress_crash_watch() {
#[test]
fn each_sweep_holds_its_own_resource() {
// Same kind, same capacity-1 resource: two passes serialise. Different
// kinds, disjoint resources and no edges: they run side by side. That the
// Same kind, same capacity-1 resource: two passes serialise. That the
// scheduler honours this is `hive_jobq`'s
// `unrelated_nodes_needing_the_same_resource_are_serialized`.
let q = JobQueue::new(1);
insert(&q, |builder| {
let _ = templates::matrix_sweep(builder);
let _ = templates::knowledge_pull(builder);
});
assert_eq!(
declared_resources_of_kind(&q, "matrix_sweep"),
vec![vec![Resource::MatrixSweep]]
);
assert_eq!(
declared_resources_of_kind(&q, "knowledge_pull"),
vec![vec![Resource::KnowledgeTree]]