heartbeat the matrix-accounts snapshot so as_of tracks daemon liveness
This commit is contained in:
parent
f38bc13293
commit
d5202ebf60
2 changed files with 104 additions and 6 deletions
|
|
@ -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`
|
||||
/// (`<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<AccountStatus> 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<AccountStatus> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -46,6 +46,12 @@ use accounts::{AccountCfg, Registry};
|
|||
/// `select_all`) rather than `tokio::spawn`/`JoinSet`.
|
||||
type SyncLoop = std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>>>>;
|
||||
|
||||
/// 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 {
|
||||
|
|
|
|||
Loading…
Reference in a new issue