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"]); + } }