diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index 9c6c3d68..7e899d19 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -596,7 +596,8 @@ exited. A running daemon re-lists its linked accounts in the store every two minutes and, when the set changed, exits with status 75, which the unit's `RestartForceExitStatus` restarts. On the same two-minute tick it tries again each secondary account that didn't come -up — no token yet, a token the homeserver rejected, or a homeserver that +up — a declared account with no token yet, a token the homeserver +rejected, or a homeserver that was down — waiting two minutes after its first failure and twice as long after each further one, up to sixteen minutes. One that comes up joins the running daemon; the primary and the other accounts aren't diff --git a/hive-matrix-mcp/src/accounts.rs b/hive-matrix-mcp/src/accounts.rs index 0bb08534..a02d6c86 100644 --- a/hive-matrix-mcp/src/accounts.rs +++ b/hive-matrix-mcp/src/accounts.rs @@ -420,7 +420,7 @@ impl Registry { } /// Atomically write `accounts` as pretty-printed JSON to `path` -/// (`.tmp` + rename) so a dashboard reader never sees a partial +/// (a unique `.….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. Used for the boot-time publish and /// when a secondary comes up later. @@ -462,6 +462,9 @@ fn write_accounts_snapshot_inner( accounts: &[AccountStatus], force: bool, ) -> anyhow::Result<()> { + // Names a tmp file unique per write: the heartbeat and a late-joining + // account write concurrently. + static WRITES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); let body = serde_json::to_string_pretty(accounts).expect("Vec is always serialisable"); if !force && std::fs::read_to_string(path).ok().as_deref() == Some(body.as_str()) { @@ -470,7 +473,8 @@ fn write_accounts_snapshot_inner( if let Some(parent) = path.parent() { std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; } - let tmp = path.with_extension("json.tmp"); + let n = WRITES.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let tmp = path.with_extension(format!("json.{}.{n}.tmp", std::process::id())); std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?; std::fs::rename(&tmp, path).with_context(|| { format!( @@ -671,6 +675,28 @@ mod tests { ] } + #[test] + fn concurrent_writers_do_not_clobber_each_other_s_tmp_file() { + let dir = std::env::temp_dir().join(format!("hh-acct-race-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("matrix-accounts.json"); + let writers: Vec<_> = (0..4) + .map(|_| { + let path = path.clone(); + std::thread::spawn(move || { + for _ in 0..200 { + heartbeat_accounts_snapshot(&path, &sample()).unwrap(); + } + }) + }) + .collect(); + for w in writers { + w.join().unwrap(); + } + assert_eq!(std::fs::read_dir(&dir).unwrap().count(), 1); + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn snapshot_writes_json_and_cleans_up_tmp() { let dir = std::env::temp_dir().join(format!( @@ -694,7 +720,11 @@ mod tests { assert_eq!(parsed[1]["homeserver"], "https://matrix.org"); assert!(parsed[1]["user_id"].is_null()); // No leftover temp file after the atomic publish. - assert!(!path.with_extension("json.tmp").exists()); + assert_eq!( + std::fs::read_dir(&dir).unwrap().count(), + 1, + "only the snapshot is left" + ); std::fs::remove_dir_all(&dir).ok(); } @@ -750,7 +780,11 @@ mod tests { 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()); + assert_eq!( + std::fs::read_dir(&dir).unwrap().count(), + 1, + "only the snapshot is left" + ); std::fs::remove_dir_all(&dir).ok(); } diff --git a/hive-matrix-mcp/src/client.rs b/hive-matrix-mcp/src/client.rs index 364fd7f5..eff9a70d 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -131,9 +131,10 @@ pub async fn build_and_restore( } std::process::exit(0); } - // Secondary: token discarded (so it's cleanly skipped next boot - // rather than re-erroring); leave the sdk state in place in case - // the operator re-provisions a fresh token for the same device. + // Secondary: a file token is gone, a store token is untouched; the + // daemon tries the account again later and reads the token afresh. + // The sdk state stays in place in case the operator re-provisions a + // token for the same device. // Return a PermanentBringUpError so the caller can distinguish // "retrying now won't help" from a transient network/DNS failure. return Err(anyhow::Error::new(PermanentBringUpError( diff --git a/hive-matrix-mcp/src/credential.rs b/hive-matrix-mcp/src/credential.rs index 97324ca0..cc7b2ce2 100644 --- a/hive-matrix-mcp/src/credential.rs +++ b/hive-matrix-mcp/src/credential.rs @@ -172,7 +172,7 @@ impl Token { Origin::File(p) => { let _ = tokio::fs::remove_file(p).await; } - Origin::Store(path) => tracing::warn!( + Origin::Store(path) => tracing::debug!( credential = %path, "the stored matrix credential was rejected; leaving it alone (this agent may only read it)" ), @@ -543,8 +543,8 @@ mod tests { #[tokio::test] async fn an_unprovisioned_account_has_no_token_rather_than_an_error() { - // The daemon skips such an account (a secondary) or waits on the path - // watcher (the primary); neither is a failure. + // The daemon leaves such a secondary down and retries it, or waits on + // the path watcher (the primary); neither is a failure. let missing = std::env::temp_dir().join("hh-cred-definitely-not-here/matrix-token"); assert!(Token::from_file(&missing).await.unwrap().is_none()); } diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 77e49c27..10d9bba1 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -89,6 +89,13 @@ const LINKED_REFRESH: std::time::Duration = std::time::Duration::from_mins(2); /// takes to come up. const MAX_SECONDARY_RETRY: std::time::Duration = std::time::Duration::from_mins(16); +/// Why a secondary is down when it had no token to try. +const NO_TOKEN: &str = "no token"; + +/// Why a secondary is down when start-up gave up on it; the reason was logged +/// there. +const FAILED_AT_START: &str = "failed at start"; + /// The exit status the daemon leaves with when the store's linked accounts /// changed. `nix/agent-modules/matrix.nix` names the same number in the unit's /// `RestartForceExitStatus` and `SuccessExitStatus`, so systemd restarts it @@ -146,7 +153,7 @@ async fn main() -> Result { } Ok(None) => { tracing::warn!(account = %cfg.name, "secondary matrix account has no token; will retry"); - down.failed(cfg.name.clone(), (cfg, tag), Instant::now()); + down.failed(cfg.name.clone(), (cfg, tag), Instant::now(), NO_TOKEN); } // The primary failing to restore is fatal (propagate so // systemd retries on a transient blip — matches legacy @@ -158,7 +165,14 @@ async fn main() -> Result { sync_loops.push(named(cfg.name.clone(), sync_loop)); registry.insert(cfg.name, client); } - None => down.failed(cfg.name.clone(), (cfg, tag), Instant::now()), + None => { + down.failed( + cfg.name.clone(), + (cfg, tag), + Instant::now(), + FAILED_AT_START, + ); + } }, } } @@ -426,6 +440,8 @@ struct Down { retry: T, failures: u32, not_before: Instant, + /// Why the last try failed. + reason: String, } impl Default for DownAccounts { @@ -437,16 +453,22 @@ impl Default for DownAccounts { } impl DownAccounts { - /// Record that `name` failed to come up at `at`. `retry` is kept from the - /// first failure on. - fn failed(&mut self, name: String, retry: T, at: Instant) { + /// Record that `name` failed to come up at `at`, for `reason`. `retry` is + /// kept from the first failure on. True when the failure is news — the + /// account's first, or for a different reason than the last — and so + /// worth a warning. + fn failed(&mut self, name: String, retry: T, at: Instant, reason: &str) -> bool { let down = self.by_name.entry(name).or_insert(Down { retry, failures: 0, not_before: at, + reason: String::new(), }); + let news = down.failures == 0 || down.reason != reason; + reason.clone_into(&mut down.reason); down.failures += 1; down.not_before = at + retry_delay(down.failures); + news } /// Record that `name` came up; it is never due again. @@ -495,16 +517,19 @@ async fn retry_down( let _ = up.send(named(name, sync_loop)); } Ok(None) => { - tracing::warn!(account = %name, "secondary matrix account still has no token; will retry"); - down.failed(name, (cfg, tag), now); + if down.failed(name.clone(), (cfg, tag), now, NO_TOKEN) { + tracing::warn!(account = %name, "secondary matrix account has no token; will keep retrying"); + } else { + tracing::debug!(account = %name, "secondary matrix account still has no token"); + } } Err(e) => { - tracing::warn!( - account = %name, - error = %format!("{e:#}"), - "secondary matrix account still failing; will retry" - ); - down.failed(name, (cfg, tag), now); + let reason = format!("{e:#}"); + if down.failed(name.clone(), (cfg, tag), now, &reason) { + tracing::warn!(account = %name, error = %reason, "secondary matrix account failing; will keep retrying"); + } else { + tracing::debug!(account = %name, error = %reason, "secondary matrix account still failing"); + } } } } @@ -668,7 +693,7 @@ mod tests { fn a_failed_account_is_due_once_its_wait_is_over_and_not_before() { let t0 = Instant::now(); let mut down = DownAccounts::default(); - down.failed("ccc".to_owned(), 7, t0); + down.failed("ccc".to_owned(), 7, t0, "r"); assert_eq!(down.due(t0), None); assert_eq!(down.due(t0 + LINKED_REFRESH - Duration::from_secs(1)), None); @@ -679,9 +704,9 @@ mod tests { fn each_failure_in_a_row_pushes_the_next_try_further_out() { let t0 = Instant::now(); let mut down = DownAccounts::default(); - down.failed("ccc".to_owned(), (), t0); + down.failed("ccc".to_owned(), (), t0, "r"); let t1 = t0 + LINKED_REFRESH; - down.failed("ccc".to_owned(), (), t1); + down.failed("ccc".to_owned(), (), t1, "r"); assert_eq!(down.due(t1 + LINKED_REFRESH), None); assert_eq!(down.due(t1 + LINKED_REFRESH * 2), Some(("ccc", &()))); @@ -691,7 +716,7 @@ mod tests { fn an_account_that_came_up_is_never_due_again() { let t0 = Instant::now(); let mut down = DownAccounts::default(); - down.failed("ccc".to_owned(), (), t0); + down.failed("ccc".to_owned(), (), t0, "r"); down.came_up("ccc"); assert_eq!(down.due(t0 + MAX_SECONDARY_RETRY * 10), None); @@ -701,21 +726,37 @@ mod tests { fn the_longest_overdue_account_goes_first_and_the_others_keep_their_schedule() { let t0 = Instant::now(); let mut down = DownAccounts::default(); - down.failed("late".to_owned(), "late cfg", t0 + Duration::from_secs(30)); - down.failed("early".to_owned(), "early cfg", t0); + down.failed( + "late".to_owned(), + "late cfg", + t0 + Duration::from_secs(30), + "r", + ); + down.failed("early".to_owned(), "early cfg", t0, "r"); let now = t0 + LINKED_REFRESH * 2; assert_eq!(down.due(now), Some(("early", &"early cfg"))); - down.failed("early".to_owned(), "early cfg", now); + down.failed("early".to_owned(), "early cfg", now, "r"); assert_eq!(down.due(now), Some(("late", &"late cfg"))); } + #[test] + fn a_failure_is_news_the_first_time_and_when_its_reason_changes() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + assert!(down.failed("ccc".to_owned(), (), t0, "rejected")); + assert!(!down.failed("ccc".to_owned(), (), t0, "rejected")); + assert!(down.failed("ccc".to_owned(), (), t0, "no token")); + assert!(!down.failed("ccc".to_owned(), (), t0, "no token")); + assert!(down.failed("other".to_owned(), (), t0, "no token")); + } + #[test] fn the_first_retry_target_is_kept_across_failures() { let t0 = Instant::now(); let mut down = DownAccounts::default(); - down.failed("ccc".to_owned(), "first", t0); - down.failed("ccc".to_owned(), "second", t0); + down.failed("ccc".to_owned(), "first", t0, "r"); + down.failed("ccc".to_owned(), "second", t0, "r"); assert_eq!(down.due(t0 + MAX_SECONDARY_RETRY), Some(("ccc", &"first"))); }