From bd6009ff33fd0bd9f8a5bef06a5f68fcbca7eb5b Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 14 Oct 2026 01:28:01 +0200 Subject: [PATCH] hive-matrix-mcp: test the secondary retry loop and the registry lock The retry loop's body was only reachable through bring_up_account, which needs a live homeserver, so nothing exercised what happens when a down secondary comes up or fails, or that the Registry's new RwLock is never held across the network call. The unit tests covered the pure decision (DownAccounts, retry_delay) only. retry_down now delegates to retry_loop / retry_once, which take the bring-up step and the snapshot path as parameters; retry_down passes bring_up_account and the real path, so behaviour is unchanged. Tests use offline matrix-sdk clients and fake bring-ups to show that: - a secondary that comes up is inserted, leaves the down set, has its sync loop sent on and is published in the snapshot, while the primary and the other accounts stay the same clients; - one that fails (error or no token) stays down, is not published, and waits twice as long next time; one that is not yet due is not tried; - readers are not blocked while a bring-up is pending; - inserts under four busy reader threads neither deadlock nor lose an account; - the loop, under paused time, retries at 2, 6, 14 and 30 minutes. tokio's test-util feature is added as a dev-dependency for the last. --- hive-matrix-mcp/Cargo.toml | 3 + hive-matrix-mcp/src/main.rs | 402 +++++++++++++++++++++++++++++++++--- 2 files changed, 375 insertions(+), 30 deletions(-) diff --git a/hive-matrix-mcp/Cargo.toml b/hive-matrix-mcp/Cargo.toml index 1693d81a..9857502d 100644 --- a/hive-matrix-mcp/Cargo.toml +++ b/hive-matrix-mcp/Cargo.toml @@ -32,3 +32,6 @@ tracing-subscriber.workspace = true [[bin]] name = "hive-matrix-daemon" path = "src/main.rs" + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 10d9bba1..f621a334 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -29,6 +29,7 @@ use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::convert::Infallible; +use std::path::Path; use std::process::ExitCode; use std::sync::Arc; @@ -487,49 +488,92 @@ impl DownAccounts { } } +/// What a bring-up attempt yields; [`bring_up_account`]'s result. +type BringUp = Result>; + /// Every [`LINKED_REFRESH`], try again the secondary [`DownAccounts::due`] -/// names, one per tick. One that comes up joins `registry` and its sync loop -/// goes to `up`; the accounts already up are not touched. Never returns. +/// names, one per tick, with [`bring_up_account`]. Never returns. async fn retry_down( mut down: DownAccounts, registry: Arc, up: tokio::sync::mpsc::UnboundedSender, ) -> Infallible { + retry_loop( + &mut down, + ®istry, + &up, + &paths::accounts_file(), + |cfg: AccountCfg, tag| async move { bring_up_account(&cfg, tag, false).await }, + ) + .await +} + +/// [`retry_down`] with the bring-up step and the snapshot path injected. +async fn retry_loop( + down: &mut DownAccounts, + registry: &Registry, + up: &tokio::sync::mpsc::UnboundedSender, + snapshot: &Path, + mut bring_up: F, +) -> Infallible +where + F: FnMut(AccountCfg, Option) -> Fut, + Fut: std::future::Future, +{ let mut tick = tokio::time::interval(LINKED_REFRESH); tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // The first tick is immediate, and boot has just tried every account. tick.tick().await; loop { let now = tick.tick().await; - let Some((name, (cfg, tag))) = down.due(now) else { - continue; - }; - let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone()); - match bring_up_account(&cfg, tag.clone(), false).await { - Ok(Some((client, sync_loop))) => { - tracing::info!(account = %name, "secondary matrix account came up"); - down.came_up(&name); - registry.insert(name.clone(), client); - if let Err(e) = registry.write_snapshot(&paths::accounts_file()) { - tracing::warn!(error = %format!("{e:#}"), "failed to write matrix-accounts snapshot"); - } - // `drive` holds the receiver for as long as it polls this. - let _ = up.send(named(name, sync_loop)); + retry_once(down, registry, up, snapshot, now, &mut bring_up).await; + } +} + +/// Try the secondary of `down` that is due at `now`, if any. One that comes up +/// leaves `down`, joins `registry` and has its sync loop sent to `up`; the +/// accounts already in `registry` are not touched. One that fails stays in +/// `down`, due again after a longer wait. No lock on `registry` is held while +/// `bring_up` runs. +async fn retry_once( + down: &mut DownAccounts, + registry: &Registry, + up: &tokio::sync::mpsc::UnboundedSender, + snapshot: &Path, + now: Instant, + bring_up: &mut F, +) where + F: FnMut(AccountCfg, Option) -> Fut, + Fut: std::future::Future, +{ + let Some((name, (cfg, tag))) = down.due(now) else { + return; + }; + let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone()); + match bring_up(cfg.clone(), tag.clone()).await { + Ok(Some((client, sync_loop))) => { + tracing::info!(account = %name, "secondary matrix account came up"); + down.came_up(&name); + registry.insert(name.clone(), client); + if let Err(e) = registry.write_snapshot(snapshot) { + tracing::warn!(error = %format!("{e:#}"), "failed to write matrix-accounts snapshot"); } - Ok(None) => { - 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"); - } + // `drive` holds the receiver for as long as it polls this. + let _ = up.send(named(name, sync_loop)); + } + Ok(None) => { + 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) => { - 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"); - } + } + Err(e) => { + 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"); } } } @@ -678,7 +722,17 @@ mod tests { use tokio::time::Instant; - use super::{DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, retry_delay}; + use std::path::Path; + use std::sync::Arc; + + use anyhow::anyhow; + use matrix_sdk::Client; + use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel}; + + use super::{ + AccountCfg, BringUp, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, + Registry, Retry, SyncLoop, retry_delay, retry_loop, retry_once, + }; #[test] fn the_wait_starts_at_one_refresh_and_doubles_up_to_the_cap() { @@ -760,4 +814,292 @@ mod tests { assert_eq!(down.due(t0 + MAX_SECONDARY_RETRY), Some(("ccc", &"first"))); } + + /// A client that never touches the network: built, not restored. + async fn client(host: &str) -> Client { + Client::builder() + .homeserver_url(format!("https://{host}.example")) + .build() + .await + .expect("an offline client") + } + + fn cfg(name: &str) -> AccountCfg { + AccountCfg { + name: name.to_owned(), + token_file: None, + state_dir: std::env::temp_dir().join(format!("hh-main-test-{name}")), + homeserver: Some(format!("https://{name}.example")), + } + } + + fn pending_loop() -> SyncLoop { + Box::pin(std::future::pending()) + } + + /// A registry holding the primary `main` and a running secondary `other`. + async fn registry() -> Registry { + let r = Registry::new("main".to_owned()); + r.insert("main".to_owned(), client("main").await); + r.insert("other".to_owned(), client("other").await); + r + } + + fn down_ccc(at: Instant) -> DownAccounts { + let mut down = DownAccounts::default(); + down.failed("ccc".to_owned(), (cfg("ccc"), None), at, "boot"); + down + } + + fn snapshot_path(tag: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!("hh-main-snap-{tag}-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + dir.join("matrix-accounts.json") + } + + fn names(r: &Registry) -> Vec { + r.list().into_iter().map(|a| a.name).collect() + } + + fn sent(rx: &mut UnboundedReceiver) -> usize { + std::iter::from_fn(|| rx.try_recv().ok()).count() + } + + #[tokio::test] + async fn a_secondary_that_comes_up_joins_the_registry_and_leaves_the_down_set() { + let t0 = Instant::now(); + let mut down = down_ccc(t0); + let reg = registry().await; + let (main_before, other_before) = ( + reg.resolve(Some("main")).unwrap(), + reg.resolve(Some("other")).unwrap(), + ); + let (up, mut rx) = unbounded_channel(); + let snap = snapshot_path("up"); + let mut ccc = Some(client("ccc").await); + let mut bring_up = move |_: AccountCfg, _: Option| { + let c = ccc.take().expect("tried once"); + async move { Ok(Some((c, pending_loop()))) as BringUp } + }; + + let now = t0 + LINKED_REFRESH; + retry_once(&mut down, ®, &up, &snap, now, &mut bring_up).await; + + assert!(reg.resolve(Some("ccc")).is_ok()); + assert!(down.due(now + MAX_SECONDARY_RETRY * 10).is_none()); + assert_eq!(sent(&mut rx), 1); + // The accounts already up are the very same clients. + assert!(Arc::ptr_eq( + &main_before, + ®.resolve(Some("main")).unwrap() + )); + assert!(Arc::ptr_eq( + &other_before, + ®.resolve(Some("other")).unwrap() + )); + let published = std::fs::read_to_string(&snap).unwrap(); + assert!( + published.contains("\"ccc\""), + "snapshot lists it: {published}" + ); + std::fs::remove_dir_all(snap.parent().unwrap()).ok(); + } + + #[tokio::test] + async fn a_secondary_that_fails_stays_down_with_a_longer_wait_and_disturbs_nobody() { + for outcome in [Err(anyhow!("whoami: 502")), Ok(None)] { + let t0 = Instant::now(); + let mut down = down_ccc(t0); + let reg = registry().await; + let main_before = reg.resolve(Some("main")).unwrap(); + let (up, mut rx) = unbounded_channel(); + let snap = snapshot_path("fail"); + let mut outcome = Some(outcome); + let mut bring_up = move |_: AccountCfg, _: Option| { + let r = outcome.take().expect("tried once"); + async move { r as BringUp } + }; + + let now = t0 + LINKED_REFRESH; + retry_once(&mut down, ®, &up, &snap, now, &mut bring_up).await; + + // Still down; its second failure waits twice as long as its first. + assert!(reg.resolve(Some("ccc")).is_err()); + assert_eq!(names(®), ["main", "other"]); + assert_eq!(sent(&mut rx), 0); + assert!(!snap.exists(), "no snapshot for a failed try"); + assert!( + down.due(now + retry_delay(2) - Duration::from_secs(1)) + .is_none() + ); + assert!(down.due(now + retry_delay(2)).is_some()); + assert!(Arc::ptr_eq( + &main_before, + ®.resolve(Some("main")).unwrap() + )); + } + } + + #[tokio::test] + async fn an_account_whose_wait_is_not_over_is_not_tried() { + let t0 = Instant::now(); + let mut down = down_ccc(t0); + let reg = registry().await; + let (up, _rx) = unbounded_channel(); + let mut tried = 0; + let mut bring_up = |_: AccountCfg, _: Option| { + tried += 1; + async { Ok(None) as BringUp } + }; + + let early = t0 + LINKED_REFRESH - Duration::from_secs(1); + retry_once( + &mut down, + ®, + &up, + Path::new("/nonexistent/x.json"), + early, + &mut bring_up, + ) + .await; + + assert_eq!(tried, 0); + } + + #[tokio::test] + async fn readers_are_not_blocked_while_a_bring_up_is_pending() { + let t0 = Instant::now(); + let mut down = down_ccc(t0); + let reg = Arc::new(registry().await); + let (up, _rx) = unbounded_channel(); + let snap = snapshot_path("pending"); + let (gate_tx, gate_rx) = tokio::sync::oneshot::channel::<()>(); + let mut gate_rx = Some(gate_rx); + let mut ccc = Some(client("ccc").await); + let mut bring_up = move |_: AccountCfg, _: Option| { + let gate = gate_rx.take().expect("tried once"); + let c = ccc.take().expect("tried once"); + async move { + gate.await.unwrap(); + Ok(Some((c, pending_loop()))) as BringUp + } + }; + + // A reader on its own thread, so a lock held across the bring-up + // blocks it, not the runtime: it must finish before the gate opens. + let (done_tx, done_rx) = tokio::sync::oneshot::channel(); + let reader_reg = Arc::clone(®); + let reader = std::thread::spawn(move || { + let seen = reader_reg.list().len(); + let resolved = reader_reg.resolve(Some("main")).is_ok(); + let _ = done_tx.send((seen, resolved)); + }); + let retry = retry_once( + &mut down, + ®, + &up, + &snap, + t0 + LINKED_REFRESH, + &mut bring_up, + ); + let open_gate_after_reader = async { + let seen = done_rx.await.unwrap(); + gate_tx.send(()).unwrap(); + seen + }; + let ((), seen) = tokio::time::timeout(Duration::from_secs(10), async { + tokio::join!(retry, open_gate_after_reader) + }) + .await + .expect("a reader was blocked by a pending bring-up"); + + assert_eq!(seen, (2, true)); + assert!(reg.resolve(Some("ccc")).is_ok()); + reader.join().unwrap(); + std::fs::remove_dir_all(snap.parent().unwrap()).ok(); + } + + #[test] + fn inserting_while_readers_run_neither_deadlocks_nor_loses_an_account() { + // The whole scenario runs on its own thread: a deadlock blocks that + // thread, and only the channel timeout below can see it. + let (result_tx, result_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + rt.block_on(async { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + for n in ["a", "b", "c", "d"] { + down.failed(n.to_owned(), (cfg(n), None), t0, "boot"); + } + let reg = Arc::new(registry().await); + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let readers: Vec<_> = (0..4) + .map(|_| { + let (reg, stop) = (Arc::clone(®), Arc::clone(&stop)); + std::thread::spawn(move || { + while !stop.load(std::sync::atomic::Ordering::Relaxed) { + let _ = reg.list(); + assert!(reg.resolve(Some("main")).is_ok()); + } + }) + }) + .collect(); + let (up, _rx) = unbounded_channel(); + let snap = snapshot_path("readers"); + let mut bring_up = |c: AccountCfg, _: Option| async move { + Ok(Some((client(&c.name).await, pending_loop()))) as BringUp + }; + let mut now = t0 + LINKED_REFRESH; + for _ in 0..4 { + retry_once(&mut down, ®, &up, &snap, now, &mut bring_up).await; + now += Duration::from_secs(1); + } + stop.store(true, std::sync::atomic::Ordering::Relaxed); + for r in readers { + r.join().unwrap(); + } + std::fs::remove_dir_all(snap.parent().unwrap()).ok(); + let mut got = names(®); + got.sort(); + result_tx.send(got).unwrap(); + }); + }); + let got = result_rx + .recv_timeout(Duration::from_secs(20)) + .expect("deadlock or panic while inserting under readers"); + assert_eq!(got, ["a", "b", "c", "d", "main", "other"]); + } + + #[tokio::test(start_paused = true)] + async fn the_loop_retries_a_failing_account_on_a_growing_wait() { + let t0 = Instant::now(); + let mut down = down_ccc(t0); + let reg = registry().await; + let (up, _rx) = unbounded_channel(); + let calls = std::cell::RefCell::new(Vec::new()); + let bring_up = |_: AccountCfg, _: Option| { + calls.borrow_mut().push(t0.elapsed().as_secs() / 60); + async { Err(anyhow!("homeserver down")) as BringUp } + }; + + let _ = tokio::time::timeout( + Duration::from_mins(31), + retry_loop( + &mut down, + ®, + &up, + Path::new("/nonexistent/x.json"), + bring_up, + ), + ) + .await; + + // Waits of 2, 4, 8 and 16 minutes after the failures at 0, 2, 6, 14. + assert_eq!(*calls.borrow(), [2, 6, 14, 30]); + assert_eq!(names(®), ["main", "other"]); + } }