From d5202ebf6027e70ed789dd6f7f310b9ae38c67ac Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 22 Jun 2026 13:30:07 +0200 Subject: [PATCH] heartbeat the matrix-accounts snapshot so as_of tracks daemon liveness --- hive-matrix-mcp/src/accounts.rs | 84 ++++++++++++++++++++++++++++++--- hive-matrix-mcp/src/main.rs | 26 ++++++++++ 2 files changed, 104 insertions(+), 6 deletions(-) diff --git a/hive-matrix-mcp/src/accounts.rs b/hive-matrix-mcp/src/accounts.rs index 5d9e424a..70f0cc22 100644 --- a/hive-matrix-mcp/src/accounts.rs +++ b/hive-matrix-mcp/src/accounts.rs @@ -207,27 +207,67 @@ impl Registry { pub fn write_snapshot(&self, path: &Path) -> anyhow::Result<()> { write_accounts_snapshot(path, &self.list()) } + + /// Heartbeat the snapshot: rewrite it unconditionally so the file + /// mtime advances even when the account set is unchanged. The daemon + /// calls this on a timer (see `main`) so the dashboard's `as_of` + /// (derived from the file mtime) tracks real daemon liveness instead + /// of freezing at the boot-time write — a stalled mtime then means + /// the daemon is down, which the dashboard can render as stale/dimmed. + /// + /// # Errors + /// Propagates filesystem errors from the atomic write. + pub fn heartbeat_snapshot(&self, path: &Path) -> anyhow::Result<()> { + heartbeat_accounts_snapshot(path, &self.list()) + } } /// Atomically write `accounts` as pretty-printed JSON to `path` /// (`.tmp` + rename) so a dashboard reader never sees a partial /// file. Idempotent — skips the rewrite when the on-disk content already -/// matches, keeping the mtime stable. +/// matches, keeping the mtime stable. Used for the boot-time publish. /// /// The daemon rebuilds this file fresh on every boot (and is restarted /// by the `matrix-token*` path-watcher when a new account is /// provisioned), so the file lists exactly the accounts that restored at -/// the last start. A reader wanting true real-time liveness should also -/// confirm the daemon is running — a stale file otherwise reports the -/// last-known-up set. +/// the last start. Real-time liveness is conveyed by the file mtime, +/// which the daemon advances on a timer via +/// [`heartbeat_accounts_snapshot`] — a stalled mtime means the daemon is +/// down, so a reader can treat an old snapshot as stale. /// /// # Errors /// Returns an error if the parent dir can't be created or the /// write/rename fails. pub fn write_accounts_snapshot(path: &Path, accounts: &[AccountStatus]) -> anyhow::Result<()> { + write_accounts_snapshot_inner(path, accounts, false) +} + +/// Like [`write_accounts_snapshot`] but always rewrites (tmp + rename) +/// even when the on-disk content is byte-identical, so the file mtime +/// advances. The daemon calls this on a periodic heartbeat so the +/// dashboard's `as_of` (the file mtime) reflects daemon liveness rather +/// than freezing at the boot-time write. +/// +/// # Errors +/// Returns an error if the parent dir can't be created or the +/// write/rename fails. +pub fn heartbeat_accounts_snapshot(path: &Path, accounts: &[AccountStatus]) -> anyhow::Result<()> { + write_accounts_snapshot_inner(path, accounts, true) +} + +/// Shared body for [`write_accounts_snapshot`] (idempotent) and +/// [`heartbeat_accounts_snapshot`] (`force`). When `force` is false the +/// rewrite is skipped if the on-disk content already matches, keeping the +/// mtime stable; when true the tmp + rename always runs so the mtime +/// advances. +fn write_accounts_snapshot_inner( + path: &Path, + accounts: &[AccountStatus], + force: bool, +) -> anyhow::Result<()> { let body = serde_json::to_string_pretty(accounts).expect("Vec is always serialisable"); - if std::fs::read_to_string(path).ok().as_deref() == Some(body.as_str()) { + if !force && std::fs::read_to_string(path).ok().as_deref() == Some(body.as_str()) { return Ok(()); } if let Some(parent) = path.parent() { @@ -247,7 +287,7 @@ pub fn write_accounts_snapshot(path: &Path, accounts: &[AccountStatus]) -> anyho #[cfg(test)] mod tests { - use super::{AccountStatus, write_accounts_snapshot}; + use super::{AccountStatus, heartbeat_accounts_snapshot, write_accounts_snapshot}; fn sample() -> Vec { vec![ @@ -319,4 +359,36 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } + + #[test] + fn heartbeat_advances_mtime_even_when_unchanged() { + let dir = std::env::temp_dir().join(format!( + "hh-acct-hb-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("matrix-accounts.json"); + + write_accounts_snapshot(&path, &sample()).unwrap(); + let mtime1 = std::fs::metadata(&path).unwrap().modified().unwrap(); + // The heartbeat must rewrite (tmp + rename) even when the content + // is byte-identical, so the mtime advances and the dashboard reads + // a fresh `as_of`. Small sleep so the new mtime is strictly later + // than the first (tmpfs/ext4 have sub-second mtime resolution). + std::thread::sleep(std::time::Duration::from_millis(20)); + heartbeat_accounts_snapshot(&path, &sample()).unwrap(); + let mtime2 = std::fs::metadata(&path).unwrap().modified().unwrap(); + assert!(mtime2 > mtime1, "heartbeat should advance mtime"); + // Content is still the same shape, and no leftover temp file. + let written = std::fs::read_to_string(&path).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&written).unwrap(); + assert_eq!(parsed[0]["name"], "main"); + assert!(!path.with_extension("json.tmp").exists()); + + std::fs::remove_dir_all(&dir).ok(); + } } diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index ee2a1254..5d8d2e0e 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -46,6 +46,12 @@ use accounts::{AccountCfg, Registry}; /// `select_all`) rather than `tokio::spawn`/`JoinSet`. type SyncLoop = std::pin::Pin>>>; +/// How often the daemon rewrites the accounts snapshot to advance its +/// mtime, so the dashboard's `as_of` tracks daemon liveness. 30s keeps +/// the staleness window small while the rewrite cost (one tmp + rename of +/// a tiny file) is negligible. +const ACCOUNTS_HEARTBEAT_SECS: u64 = 30; + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -116,6 +122,26 @@ async fn main() -> Result<()> { // sync loops so the MCP bridge can connect as soon as the first // claude turn fires. let registry = Arc::new(registry); + + // Heartbeat: periodically rewrite the accounts snapshot so its mtime + // advances while the daemon lives. The dashboard derives `as_of` from + // the file mtime, so a stalled mtime now means the daemon is down — + // which lets the dashboard dim accounts whose snapshot is stale rather + // than reporting the boot-time set forever (BE-4 follow-up). + let hb_registry = Arc::clone(®istry); + tokio::spawn(async move { + let mut tick = + tokio::time::interval(std::time::Duration::from_secs(ACCOUNTS_HEARTBEAT_SECS)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let path = paths::accounts_file(); + loop { + tick.tick().await; + if let Err(e) = hb_registry.heartbeat_snapshot(&path) { + tracing::warn!(error = %format!("{e:#}"), "matrix-accounts heartbeat write failed"); + } + } + }); + let socket_listener = mcp_socket.clone(); tokio::spawn(async move { if let Err(e) = socket::serve(&socket_listener, registry).await {