hive-matrix-mcp: bound each secondary retry, warn once per outage

Three follow-ups to argus's review of #5237:

- A down secondary's retry runs `restore` under RESTORE_TIMEOUT (2 min).
  Tries run one per tick on one future, so a restore that never resolved
  held back every other down account's retries for good; a try that
  outlasts the timeout now counts as a failure like any other.
- `client::build_and_restore` warns about a rejected token only for the
  primary, which exits right after. For a secondary it logs at debug: the
  retry loop already warns once per outage (`DownAccounts::failed` is news
  on the first failure and on a changed reason, and an account that came
  up starts afresh), so the extra warn repeated on every try.
- `a_secondary_that_comes_up_late_does_not_wipe_the_todos` runs the late
  bring-up in a child process of the test binary with HIVE_AGENT_SOCKET
  pointed at the test's listener. In-process the variable was unset, so a
  wipe placed back in the bring-up path was a no-op and the test passed.

New tests: a_second_rejection_in_a_row_does_not_warn_again (captured logs, alone
in a child process since tracing caches callsite interest process-wide;
a stub whoami answering M_UNKNOWN_TOKEN, the real build_and_restore),
a_failure_after_the_account_came_up_is_news_again, and
an_account_whose_restore_hangs_does_not_hold_back_the_others (paused time).

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<F, Fut>(
     down: &mut DownAccounts<Retry>,
     registry: &Registry,
@@ -554,7 +560,10 @@ async fn retry_once<F, Fut>(
         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<std::sync::Mutex<Vec<u8>>>);
+
+    impl std::io::Write for Warnings {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            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> {
+            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, &reg, &up, &snap, first, &mut restore).await;
+        let second = first + retry_delay(2);
+        retry_once(&mut down, &reg, &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(&reg), ["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, &reg, &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();
+    }
 }