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
This commit is contained in:
parent
6f2455f84a
commit
b2828d83dc
1 changed files with 165 additions and 81 deletions
|
|
@ -90,9 +90,9 @@ const LINKED_REFRESH: std::time::Duration = std::time::Duration::from_mins(2);
|
||||||
/// takes to come up.
|
/// takes to come up.
|
||||||
const MAX_SECONDARY_RETRY: std::time::Duration = std::time::Duration::from_mins(16);
|
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
|
/// How long one restore of an account, at boot or on a retry, may take before
|
||||||
/// failure. Tries run one after another, so this bounds how long one hung
|
/// it counts as a failure. Restores run one after another, so this bounds how
|
||||||
/// account holds back the next one's try.
|
/// long one hung account holds back the next one's.
|
||||||
const RESTORE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(2);
|
const RESTORE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(2);
|
||||||
|
|
||||||
/// Why a secondary is down when it had no token to try.
|
/// Why a secondary is down when it had no token to try.
|
||||||
|
|
@ -131,58 +131,13 @@ async fn main() -> Result<ExitCode> {
|
||||||
let booted = account_names(&declared, &linked);
|
let booted = account_names(&declared, &linked);
|
||||||
let mut cfgs = declared.clone();
|
let mut cfgs = declared.clone();
|
||||||
cfgs.extend(linked);
|
cfgs.extend(linked);
|
||||||
let multi = cfgs.len() > 1;
|
|
||||||
let primary = cfgs[0].name.clone();
|
let primary = cfgs[0].name.clone();
|
||||||
let registry = Registry::new(primary);
|
let registry = Registry::new(primary);
|
||||||
let mut sync_loops: Vec<NamedSyncLoop> = Vec::new();
|
let restore =
|
||||||
let mut down = DownAccounts::default();
|
|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 {
|
||||||
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);
|
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,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if registry.is_empty() {
|
if registry.is_empty() {
|
||||||
tracing::warn!("no matrix accounts restored; exiting cleanly");
|
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<NamedSyncLoop>,
|
||||||
|
down: DownAccounts<Retry>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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<F, Fut>(
|
||||||
|
cfgs: Vec<AccountCfg>,
|
||||||
|
registry: &Registry,
|
||||||
|
mut restore: F,
|
||||||
|
) -> Result<Option<Boot>>
|
||||||
|
where
|
||||||
|
F: FnMut(AccountCfg, bool) -> Fut,
|
||||||
|
Fut: std::future::Future<Output = Restore>,
|
||||||
|
{
|
||||||
|
let multi = cfgs.len() > 1;
|
||||||
|
let mut sync_loops: Vec<NamedSyncLoop> = 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<Output = Restore> + 'a,
|
||||||
|
) -> impl std::future::Future<Output = Restore> + '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
|
/// Try to bring up a secondary account, retrying with exponential backoff
|
||||||
/// on transient failures (anything that is NOT a `PermanentBringUpError`).
|
/// 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
|
/// that the account stays down and is left to [`retry_down`] (permanent
|
||||||
/// failure, no token, or all retries exhausted).
|
/// failure, no token, or all retries exhausted).
|
||||||
async fn bring_up_secondary_with_retry(
|
async fn bring_up_secondary_with_retry<F, Fut>(
|
||||||
cfg: &AccountCfg,
|
cfg: &AccountCfg,
|
||||||
tag: Option<String>,
|
|
||||||
first_error: anyhow::Error,
|
first_error: anyhow::Error,
|
||||||
) -> Option<(Client, SyncLoop)> {
|
restore: &mut F,
|
||||||
|
) -> Option<Client>
|
||||||
|
where
|
||||||
|
F: FnMut(AccountCfg, bool) -> Fut,
|
||||||
|
Fut: std::future::Future<Output = Restore>,
|
||||||
|
{
|
||||||
// Permanent failure: the homeserver rejected the token. Trying the same
|
// Permanent failure: the homeserver rejected the token. Trying the same
|
||||||
// token again at once won't help — leave it to `retry_down`.
|
// token again at once won't help — leave it to `retry_down`.
|
||||||
if first_error
|
if first_error
|
||||||
|
|
@ -396,10 +449,10 @@ async fn bring_up_secondary_with_retry(
|
||||||
"retrying secondary account bring-up after backoff"
|
"retrying secondary account bring-up after backoff"
|
||||||
);
|
);
|
||||||
tokio::time::sleep(std::time::Duration::from_secs(delay)).await;
|
tokio::time::sleep(std::time::Duration::from_secs(delay)).await;
|
||||||
match bring_up_account(cfg, tag.clone(), false).await {
|
match within_restore_timeout(&cfg.name, restore(cfg.clone(), false)).await {
|
||||||
Ok(Some((client, sync_loop))) => {
|
Ok(Some(client)) => {
|
||||||
tracing::info!(account = %cfg.name, "secondary matrix account recovered");
|
tracing::info!(account = %cfg.name, "secondary matrix account recovered");
|
||||||
return Some((client, sync_loop));
|
return Some(client);
|
||||||
}
|
}
|
||||||
Ok(None) => {
|
Ok(None) => {
|
||||||
tracing::warn!(account = %cfg.name, "secondary matrix account has no token; leaving it down");
|
tracing::warn!(account = %cfg.name, "secondary matrix account has no token; leaving it down");
|
||||||
|
|
@ -560,9 +613,7 @@ async fn retry_once<F, Fut>(
|
||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone());
|
let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone());
|
||||||
let restored = tokio::time::timeout(RESTORE_TIMEOUT, restore(cfg.clone()))
|
let restored = within_restore_timeout(&name, restore(cfg.clone())).await;
|
||||||
.await
|
|
||||||
.unwrap_or_else(|_| Err(anyhow::anyhow!("gave up after {RESTORE_TIMEOUT:?}")));
|
|
||||||
match restored {
|
match restored {
|
||||||
Ok(Some(client)) => {
|
Ok(Some(client)) => {
|
||||||
tracing::info!(account = %name, "secondary matrix account came up");
|
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<String>,
|
|
||||||
is_primary: bool,
|
|
||||||
) -> Result<Option<(Client, SyncLoop)>> {
|
|
||||||
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:
|
/// Restore one account's client, when it has a homeserver and a token:
|
||||||
/// everything in bringing an account up that talks to the network.
|
/// everything in bringing an account up that talks to the network.
|
||||||
async fn restore_account(cfg: &AccountCfg, is_primary: bool) -> Result<Option<Client>> {
|
async fn restore_account(cfg: &AccountCfg, is_primary: bool) -> Result<Option<Client>> {
|
||||||
|
|
@ -759,8 +794,8 @@ mod tests {
|
||||||
|
|
||||||
use super::{
|
use super::{
|
||||||
AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop,
|
AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop,
|
||||||
RESTORE_TIMEOUT, Registry, Restore, Retry, Token, client, retry_delay, retry_loop,
|
RESTORE_TIMEOUT, Registry, Restore, Retry, Token, bring_up_all, client, retry_delay,
|
||||||
retry_once, wake,
|
retry_loop, retry_once, wake,
|
||||||
};
|
};
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|
@ -1364,4 +1399,53 @@ mod tests {
|
||||||
assert_eq!(sent(&mut rx), 1);
|
assert_eq!(sent(&mut rx), 1);
|
||||||
std::fs::remove_dir_all(snap.parent().unwrap()).ok();
|
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")
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue