diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index 7e899d19..a6e8d25b 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -599,7 +599,9 @@ two-minute tick it tries again each secondary account that didn't come 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 +after each further one, up to sixteen minutes. It tries one account at a +time and counts a try that hasn't finished after two minutes as a failure, +so one hung account doesn't hold the others back. One that comes up joins the running daemon; the primary and the other accounts aren't restarted. The same token watcher also drives avatar setting: on a restart the daemon re-runs each account's bring-up, which sets the diff --git a/hive-matrix-mcp/src/client.rs b/hive-matrix-mcp/src/client.rs index eff9a70d..5b478fda 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -108,13 +108,12 @@ pub async fn build_and_restore( // or a secondary one, because a single bad secondary token must // NOT take down the whole daemon (and with it every healthy // account). - tracing::warn!( - credential = %token.origin(), - is_primary, - "matrix token rejected (M_UNKNOWN_TOKEN); discarding stale token" - ); token.discard_stale().await; if is_primary { + tracing::warn!( + credential = %token.origin(), + "matrix token rejected (M_UNKNOWN_TOKEN); discarding stale token" + ); // Primary: also drop the matrix-sdk sqlite state keyed to the // now-invalid session, then exit 0; the swarm's backfill // re-mints the token and systemd restarts us once it @@ -137,6 +136,12 @@ pub async fn build_and_restore( // token for the same device. // Return a PermanentBringUpError so the caller can distinguish // "retrying now won't help" from a transient network/DNS failure. + // Every try of a down secondary lands here, so the warning is the + // caller's, once per outage. + tracing::debug!( + credential = %token.origin(), + "matrix token rejected (M_UNKNOWN_TOKEN); discarded stale token" + ); return Err(anyhow::Error::new(PermanentBringUpError( "matrix token rejected (M_UNKNOWN_TOKEN); discarded stale token, leaving account down".into(), ))); diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index e981f706..d88e4bd9 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -90,6 +90,11 @@ 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); +/// How long one try of a down secondary may take before it counts as a +/// failure. Tries run one after another, so this bounds how long one hung +/// account holds back the next one's try. +const RESTORE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(2); + /// Why a secondary is down when it had no token to try. const NO_TOKEN: &str = "no token"; @@ -537,8 +542,9 @@ where /// Try the secondary of `down` that is due at `now`, if any. One that comes up /// leaves `down`, joins `registry` and has its [`account_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 `restore` runs. +/// stays in `down`, due again after a longer wait; so does one whose `restore` +/// outlasts [`RESTORE_TIMEOUT`]. No lock on `registry` is held while `restore` +/// runs. async fn retry_once( down: &mut DownAccounts, registry: &Registry, @@ -554,7 +560,10 @@ async fn retry_once( return; }; let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone()); - match restore(cfg.clone()).await { + let restored = tokio::time::timeout(RESTORE_TIMEOUT, restore(cfg.clone())) + .await + .unwrap_or_else(|_| Err(anyhow::anyhow!("gave up after {RESTORE_TIMEOUT:?}"))); + match restored { Ok(Some(client)) => { tracing::info!(account = %name, "secondary matrix account came up"); let sync_loop = account_sync_loop(&client, tag); @@ -744,13 +753,14 @@ mod tests { use std::path::Path; use std::sync::Arc; - use anyhow::anyhow; + use anyhow::{Context, anyhow}; use matrix_sdk::Client; use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel}; use super::{ - AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, Registry, - Restore, Retry, retry_delay, retry_loop, retry_once, wake, + AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, + RESTORE_TIMEOUT, Registry, Restore, Retry, Token, client, retry_delay, retry_loop, + retry_once, wake, }; #[test] @@ -824,6 +834,16 @@ mod tests { assert!(down.failed("other".to_owned(), (), t0, "no token")); } + #[test] + fn a_failure_after_the_account_came_up_is_news_again() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + down.failed("ccc".to_owned(), (), t0, "rejected"); + down.came_up("ccc"); + + assert!(down.failed("ccc".to_owned(), (), t0, "rejected")); + } + #[test] fn the_first_retry_target_is_kept_across_failures() { let t0 = Instant::now(); @@ -955,12 +975,37 @@ mod tests { } } + /// Set for the child processes [`run_in_child`] starts; the tests it runs + /// do nothing without it, so another run of the ignored tests skips them. + const CHILD: &str = "HIVE_MATRIX_TEST_CHILD"; + + /// Run the ignored test `name` alone in a child process of this test + /// binary, with `env` added to its environment, and panic unless it ran + /// and passed. + async fn run_in_child(name: &str, env: &[(&str, &std::ffi::OsStr)]) { + let child = tokio::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", name, "--ignored"]) + .envs(env.iter().copied()) + .env(CHILD, "1") + .output() + .await + .unwrap(); + let stdout = String::from_utf8_lossy(&child.stdout); + assert!( + child.status.success() && stdout.contains("1 passed"), + "{name} in a child process:\n{stdout}{}", + String::from_utf8_lossy(&child.stderr) + ); + } + + /// Runs [`late_bring_up`] in a child process of this test binary whose + /// `HIVE_AGENT_SOCKET` is a listener here, so a todo op from any code that + /// reads the variable reaches the listener. #[tokio::test] async fn a_secondary_that_comes_up_late_does_not_wipe_the_todos() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; - let snap = snapshot_path("todos"); - let socket = snap.with_file_name("agent.sock"); + let socket = snapshot_path("todos").with_file_name("agent.sock"); let listener = tokio::net::UnixListener::bind(&socket).unwrap(); let ops = Arc::new(std::sync::Mutex::new(Vec::new())); let seen = Arc::clone(&ops); @@ -973,13 +1018,37 @@ mod tests { conn.write_all(b"{\"ok\":true}\n").await.unwrap(); } }); - // Startup's wipe, as `main` runs it: the one the harness should see. + + run_in_child( + "tests::late_bring_up", + &[("HIVE_AGENT_SOCKET", socket.as_os_str())], + ) + .await; + + harness.abort(); + let ops = ops.lock().unwrap().clone(); + // The child's startup wipe, and nothing after it. + let wipes = ops.iter().filter(|op| op.contains("\"all\":true")).count(); + assert_eq!(wipes, 1, "todo ops the harness saw: {ops:?}"); + std::fs::remove_dir_all(socket.parent().unwrap()).ok(); + } + + /// Startup's wipe as `main` runs it, then a secondary that comes up late + /// and runs its sync loop. + #[tokio::test] + #[ignore = "run in a child process by a_secondary_that_comes_up_late_does_not_wipe_the_todos"] + async fn late_bring_up() { + if std::env::var_os(CHILD).is_none() { + return; + } + let socket = wake::agent_socket().expect("the parent sets HIVE_AGENT_SOCKET"); wake::clear_all_todos(&socket).await.unwrap(); let t0 = Instant::now(); let mut down = down_ccc(t0); let reg = registry().await; let (up, mut rx) = unbounded_channel(); + let snap = snapshot_path("late"); let mut ccc = Some(client("ccc").await); let mut restore = move |_: AccountCfg| { let c = ccc.take().expect("tried once"); @@ -997,14 +1066,113 @@ mod tests { // Run the late account's sync loop; an offline client never syncs. let late = rx.try_recv().expect("its sync loop is handed on"); let _ = tokio::time::timeout(Duration::from_secs(2), late).await; - - harness.abort(); - let ops = ops.lock().unwrap().clone(); - let wipes = ops.iter().filter(|op| op.contains("\"all\":true")).count(); - assert_eq!(wipes, 1, "todo ops the harness saw: {ops:?}"); std::fs::remove_dir_all(snap.parent().unwrap()).ok(); } + /// The WARN and ERROR lines logged on this thread while [`Self::capture`]'s + /// guard lives. + #[derive(Clone, Default)] + struct Warnings(Arc>>); + + impl std::io::Write for Warnings { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl Warnings { + fn capture(&self) -> tracing::subscriber::DefaultGuard { + let sink = self.clone(); + tracing::subscriber::set_default( + tracing_subscriber::fmt() + .with_max_level(tracing::Level::WARN) + .with_ansi(false) + .with_writer(move || sink.clone()) + .finish(), + ) + } + + fn lines(&self) -> Vec { + String::from_utf8_lossy(&self.0.lock().unwrap()) + .lines() + .map(str::to_owned) + .collect() + } + } + + /// `restore_account`'s path for an account whose file token `homeserver` + /// rejects, minus the store read. The rejection deletes the file, so it is + /// written afresh each try. + async fn restore_rejected(homeserver: &str, dir: &Path, account: &str) -> Restore { + let token_file = dir.join("token"); + tokio::fs::write(&token_file, "stale").await?; + let token = Token::from_file(&token_file).await?.expect("just written"); + client::build_and_restore(homeserver, &token, &dir.join("state"), false) + .await + .with_context(|| format!("build matrix client for account {account}")) + .map(Some) + } + + /// Runs [`two_rejections_in_a_row`] alone in a child process: tracing caches + /// which callsites are enabled process-wide, and a test on another thread + /// reaching the same `warn!` races the subscriber [`Warnings::capture`] sets. + #[tokio::test] + async fn a_second_rejection_in_a_row_does_not_warn_again() { + run_in_child("tests::two_rejections_in_a_row", &[]).await; + } + + /// Two consecutive tries of an account whose token the homeserver rejects, + /// through the real `build_and_restore`, log one warning between them. + #[tokio::test] + #[ignore = "run in a child process by a_second_rejection_in_a_row_does_not_warn_again"] + async fn two_rejections_in_a_row() { + if std::env::var_os(CHILD).is_none() { + return; + } + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let homeserver = format!("http://{}", listener.local_addr().unwrap()); + let rejects_every_token = axum::Router::new().route( + "/_matrix/client/v3/account/whoami", + axum::routing::get(|| async { + ( + axum::http::StatusCode::UNAUTHORIZED, + r#"{"errcode":"M_UNKNOWN_TOKEN","error":"Unknown access token"}"#, + ) + }), + ); + let server = tokio::spawn(async move { axum::serve(listener, rejects_every_token).await }); + let snap = snapshot_path("rejected"); + let dir = snap.parent().unwrap().to_owned(); + let mut restore = |c: AccountCfg| { + let (homeserver, dir) = (homeserver.clone(), dir.clone()); + async move { restore_rejected(&homeserver, &dir, &c.name).await } + }; + let t0 = Instant::now(); + let mut down = down_ccc(t0); + let reg = registry().await; + let (up, _rx) = unbounded_channel(); + + let warnings = Warnings::default(); + let guard = warnings.capture(); + let first = t0 + LINKED_REFRESH; + retry_once(&mut down, ®, &up, &snap, first, &mut restore).await; + let second = first + retry_delay(2); + retry_once(&mut down, ®, &up, &snap, second, &mut restore).await; + drop(guard); + server.abort(); + + assert!(reg.resolve(Some("ccc")).is_err()); + let lines = warnings.lines(); + assert_eq!(lines.len(), 1, "{lines:#?}"); + assert!(lines[0].contains("will keep retrying"), "{lines:#?}"); + std::fs::remove_dir_all(dir).ok(); + } + #[tokio::test] async fn an_account_whose_wait_is_not_over_is_not_tried() { let t0 = Instant::now(); @@ -1166,4 +1334,34 @@ mod tests { assert_eq!(*calls.borrow(), [2, 6, 14, 30]); assert_eq!(names(®), ["main", "other"]); } + + #[tokio::test(start_paused = true)] + async fn an_account_whose_restore_hangs_does_not_hold_back_the_others() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + // `hung` is due first. + down.failed("hung".to_owned(), (cfg("hung"), None), t0, "boot"); + let fine_at = t0 + Duration::from_secs(1); + down.failed("fine".to_owned(), (cfg("fine"), None), fine_at, "boot"); + let reg = registry().await; + let (up, mut rx) = unbounded_channel(); + let snap = snapshot_path("hung"); + let restore = |c: AccountCfg| async move { + if c.name == "hung" { + std::future::pending::<()>().await; + } + Ok(Some(client(&c.name).await)) as Restore + }; + + let _ = tokio::time::timeout( + LINKED_REFRESH * 2 + RESTORE_TIMEOUT, + retry_loop(&mut down, ®, &up, &snap, restore), + ) + .await; + + assert!(reg.resolve(Some("fine")).is_ok()); + assert!(reg.resolve(Some("hung")).is_err()); + assert_eq!(sent(&mut rx), 1); + std::fs::remove_dir_all(snap.parent().unwrap()).ok(); + } }