From b2828d83dce54f4dc240a68ab1bb5f8460248074 Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 15 Oct 2026 07:29:39 +0200 Subject: [PATCH] matrix mcp: a restore that hangs at boot times out instead of blocking every account after it RESTORE_TIMEOUT bounded only retry_down's restore. The boot loop in main (primary and secondaries) and bring_up_secondary_with_retry's backoff retries called bring_up_account with no bound, so a restore that never answered held up every account after it, and the retry loop that would recover it never started. A hung primary also never reached the fatal Err path, so systemd never restarted the daemon. The boot loop moves out of main into bring_up_all, which takes the restore as a closure (the same seam retry_loop already has) so a test can stand in a restore that never resolves. Every restore, at boot, on a boot retry and in retry_once, now goes through within_restore_timeout, so RESTORE_TIMEOUT is applied in one place. A timeout is an Err naming the account: fatal for the primary, retried for a secondary, as any other restore error. bring_up_account had no caller left and is removed. Closes #5266 --- hive-matrix-mcp/src/main.rs | 246 ++++++++++++++++++++++++------------ 1 file changed, 165 insertions(+), 81 deletions(-) diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index d88e4bd9..bcb641c5 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -90,9 +90,9 @@ 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. +/// How long one restore of an account, at boot or on a retry, may take before +/// it counts as a failure. Restores run one after another, so this bounds how +/// long one hung account holds back the next one's. const RESTORE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(2); /// Why a secondary is down when it had no token to try. @@ -131,58 +131,13 @@ async fn main() -> Result { let booted = account_names(&declared, &linked); let mut cfgs = declared.clone(); cfgs.extend(linked); - let multi = cfgs.len() > 1; let primary = cfgs[0].name.clone(); let registry = Registry::new(primary); - let mut sync_loops: Vec = Vec::new(); - let mut down = DownAccounts::default(); - - for (idx, cfg) in cfgs.into_iter().enumerate() { - let is_primary = idx == 0; - // Account-tag the todos only in multi-account mode so single- - // account todo summaries stay byte-identical to the legacy format. - let tag = multi.then(|| cfg.name.clone()); - match bring_up_account(&cfg, tag.clone(), is_primary).await { - Ok(Some((client, sync_loop))) => { - sync_loops.push(named(cfg.name.clone(), sync_loop)); - registry.insert(cfg.name, client); - } - // No token yet: for the primary that means the daemon isn't - // useful — exit 0, and systemd restarts us once a token exists - // (the path-watcher for a file, the re-check timer for the store). - Ok(None) if is_primary => { - tracing::warn!( - account = %cfg.name, - "primary matrix account has no token yet; exiting cleanly \ - (systemd restarts us once one exists)" - ); - return Ok(ExitCode::SUCCESS); - } - Ok(None) => { - tracing::warn!(account = %cfg.name, "secondary matrix account has no token; will retry"); - 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 - // behaviour); a secondary failing is retried with backoff - // here, then left to `retry_down`. - Err(e) if is_primary => return Err(e.context("bring up primary matrix account")), - Err(e) => match bring_up_secondary_with_retry(&cfg, tag.clone(), e).await { - Some((client, sync_loop)) => { - sync_loops.push(named(cfg.name.clone(), sync_loop)); - registry.insert(cfg.name, client); - } - None => { - down.failed( - cfg.name.clone(), - (cfg, tag), - Instant::now(), - FAILED_AT_START, - ); - } - }, - } - } + let restore = + |cfg: AccountCfg, is_primary| async move { restore_account(&cfg, is_primary).await }; + let Some(Boot { sync_loops, down }) = bring_up_all(cfgs, ®istry, restore).await? else { + return Ok(ExitCode::SUCCESS); + }; if registry.is_empty() { tracing::warn!("no matrix accounts restored; exiting cleanly"); @@ -360,16 +315,114 @@ async fn wait_for_linked_change( } } +/// What [`bring_up_all`] brought up: the sync loops of the accounts that came +/// up, and the secondaries left to [`retry_down`]. +struct Boot { + sync_loops: Vec, + down: DownAccounts, +} + +/// Try each account of `cfgs` once with `restore`, the first as the primary, +/// and insert each that comes up into `registry`. `Ok(None)` when the primary +/// has no token. Each `restore` is bounded by [`within_restore_timeout`]. +/// +/// # Errors +/// The primary failing to restore, a timeout included. +async fn bring_up_all( + cfgs: Vec, + registry: &Registry, + mut restore: F, +) -> Result> +where + F: FnMut(AccountCfg, bool) -> Fut, + Fut: std::future::Future, +{ + let multi = cfgs.len() > 1; + let mut sync_loops: Vec = Vec::new(); + let mut down = DownAccounts::default(); + + for (idx, cfg) in cfgs.into_iter().enumerate() { + let is_primary = idx == 0; + // Account-tag the todos only in multi-account mode so single- + // account todo summaries stay byte-identical to the legacy format. + let tag = multi.then(|| cfg.name.clone()); + match within_restore_timeout(&cfg.name, restore(cfg.clone(), is_primary)).await { + Ok(Some(client)) => { + sync_loops.push(named(cfg.name.clone(), account_sync_loop(&client, tag))); + registry.insert(cfg.name, client); + } + // No token yet: for the primary that means the daemon isn't + // useful — exit 0, and systemd restarts us once a token exists + // (the path-watcher for a file, the re-check timer for the store). + Ok(None) if is_primary => { + tracing::warn!( + account = %cfg.name, + "primary matrix account has no token yet; exiting cleanly \ + (systemd restarts us once one exists)" + ); + return Ok(None); + } + Ok(None) => { + tracing::warn!(account = %cfg.name, "secondary matrix account has no token; will retry"); + 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 + // behaviour); a secondary failing is retried with backoff + // here, then left to `retry_down`. + Err(e) if is_primary => return Err(e.context("bring up primary matrix account")), + Err(e) => match bring_up_secondary_with_retry(&cfg, e, &mut restore).await { + Some(client) => { + sync_loops.push(named(cfg.name.clone(), account_sync_loop(&client, tag))); + registry.insert(cfg.name, client); + } + None => { + down.failed( + cfg.name.clone(), + (cfg, tag), + Instant::now(), + FAILED_AT_START, + ); + } + }, + } + } + Ok(Some(Boot { sync_loops, down })) +} + +/// `restore` of `account`, failing once it has run for [`RESTORE_TIMEOUT`]. +/// `restore` is boxed before the returned future exists, so a caller's future +/// does not grow by the size of [`restore_account`]'s. +fn within_restore_timeout<'a>( + account: &'a str, + restore: impl std::future::Future + 'a, +) -> impl std::future::Future + 'a { + let restore = Box::pin(restore); + async move { + tokio::time::timeout(RESTORE_TIMEOUT, restore) + .await + .unwrap_or_else(|_| { + Err(anyhow::anyhow!( + "restoring matrix account {account} gave up after {RESTORE_TIMEOUT:?}" + )) + }) + } +} + /// Try to bring up a secondary account, retrying with exponential backoff /// on transient failures (anything that is NOT a `PermanentBringUpError`). -/// Returns `Some((client, sync_loop))` on success, or `None` to signal +/// Returns `Some(client)` on success, or `None` to signal /// that the account stays down and is left to [`retry_down`] (permanent /// failure, no token, or all retries exhausted). -async fn bring_up_secondary_with_retry( +async fn bring_up_secondary_with_retry( cfg: &AccountCfg, - tag: Option, first_error: anyhow::Error, -) -> Option<(Client, SyncLoop)> { + restore: &mut F, +) -> Option +where + F: FnMut(AccountCfg, bool) -> Fut, + Fut: std::future::Future, +{ // Permanent failure: the homeserver rejected the token. Trying the same // token again at once won't help — leave it to `retry_down`. if first_error @@ -396,10 +449,10 @@ async fn bring_up_secondary_with_retry( "retrying secondary account bring-up after backoff" ); tokio::time::sleep(std::time::Duration::from_secs(delay)).await; - match bring_up_account(cfg, tag.clone(), false).await { - Ok(Some((client, sync_loop))) => { + match within_restore_timeout(&cfg.name, restore(cfg.clone(), false)).await { + Ok(Some(client)) => { tracing::info!(account = %cfg.name, "secondary matrix account recovered"); - return Some((client, sync_loop)); + return Some(client); } Ok(None) => { tracing::warn!(account = %cfg.name, "secondary matrix account has no token; leaving it down"); @@ -560,9 +613,7 @@ async fn retry_once( return; }; let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone()); - let restored = tokio::time::timeout(RESTORE_TIMEOUT, restore(cfg.clone())) - .await - .unwrap_or_else(|_| Err(anyhow::anyhow!("gave up after {RESTORE_TIMEOUT:?}"))); + let restored = within_restore_timeout(&name, restore(cfg.clone())).await; match restored { Ok(Some(client)) => { tracing::info!(account = %name, "secondary matrix account came up"); @@ -658,22 +709,6 @@ async fn ensure_recovery(client: &Client, account: &str) { } } -/// Restore one account's client (when its token exists) and build its sync -/// loop. Returns `Ok(Some((client, sync_loop)))` when the account came up, -/// `Ok(None)` when it has no token or homeserver (caller decides -/// primary-vs-secondary handling). The returned sync loop is driven by the -/// caller. -async fn bring_up_account( - cfg: &AccountCfg, - tag: Option, - is_primary: bool, -) -> Result> { - Ok(restore_account(cfg, is_primary).await?.map(|client| { - let sync_loop = account_sync_loop(&client, tag); - (client, sync_loop) - })) -} - /// Restore one account's client, when it has a homeserver and a token: /// everything in bringing an account up that talks to the network. async fn restore_account(cfg: &AccountCfg, is_primary: bool) -> Result> { @@ -759,8 +794,8 @@ mod tests { use super::{ AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, - RESTORE_TIMEOUT, Registry, Restore, Retry, Token, client, retry_delay, retry_loop, - retry_once, wake, + RESTORE_TIMEOUT, Registry, Restore, Retry, Token, bring_up_all, client, retry_delay, + retry_loop, retry_once, wake, }; #[test] @@ -1364,4 +1399,53 @@ mod tests { assert_eq!(sent(&mut rx), 1); std::fs::remove_dir_all(snap.parent().unwrap()).ok(); } + + /// A restore that never resolves for `hung`, and an offline client for any + /// other account. + async fn hang_for_hung(c: AccountCfg, _: bool) -> Restore { + if c.name == "hung" { + std::future::pending::<()>().await; + } + Ok(Some(client(&c.name).await)) + } + + #[tokio::test(start_paused = true)] + async fn a_primary_whose_restore_hangs_at_boot_fails_the_boot() { + let reg = Registry::new("hung".to_owned()); + let boot = tokio::time::timeout( + Duration::from_hours(1), + bring_up_all(vec![cfg("hung"), cfg("fine")], ®, hang_for_hung), + ) + .await + .expect("boot never finished"); + + let err = boot.err().expect("a hung primary fails the boot"); + assert!(format!("{err:#}").contains("hung"), "{err:#}"); + assert!(reg.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_secondary_whose_restore_hangs_at_boot_does_not_hold_back_the_others() { + let reg = Registry::new("main".to_owned()); + let boot = tokio::time::timeout( + Duration::from_hours(1), + bring_up_all( + vec![cfg("main"), cfg("hung"), cfg("fine")], + ®, + hang_for_hung, + ), + ) + .await + .expect("boot never finished") + .expect("the primary came up") + .expect("the primary had a token"); + + assert_eq!(names(®), ["main", "fine"]); + assert_eq!(boot.sync_loops.len(), 2); + assert!( + boot.down + .due(Instant::now() + MAX_SECONDARY_RETRY) + .is_some_and(|(name, _)| name == "hung") + ); + } }