diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index 66bf831a..7e899d19 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -594,7 +594,14 @@ provisioning. A token in the store changes no file, so a store-backed agent also gets a timer that restarts the daemon five minutes after it last exited. A running daemon re-lists its linked accounts in the store every two minutes and, when the set changed, exits with status -75, which the unit's `RestartForceExitStatus` restarts. The same token watcher also drives avatar setting: on a +75, which the unit's `RestartForceExitStatus` restarts. On the same +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 +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 avatar (see below). diff --git a/docs/swarm/credentials.md b/docs/swarm/credentials.md index a31c1d42..ba520a0f 100644 --- a/docs/swarm/credentials.md +++ b/docs/swarm/credentials.md @@ -58,7 +58,7 @@ of the cell says how. | store path | minter | reader — pulls at runtime, holds in memory | automatic re-mint | automatic re-pull | | ----------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `swarm/agents//matrix/main` | `swarm-controller`, with the swarm's appservice token, at agent creation and in a five-minute pass | the agent container itself, under the certificate its hive passed in | ✅ the pass re-mints when the stored token is missing, unknown to the homeserver, or someone else's | ✅ `hive-matrix-daemon` exits when the homeserver rejects its token, and a five-minute timer restarts it, which reads the store again | -| `swarm/agents//matrix/` | `swarm-controller` | the agent container itself, under the certificate its hive passed in | must be stated | must be stated | +| `swarm/agents//matrix/` | `swarm-controller` | the agent container itself, under the certificate its hive passed in | must be stated | ✅ a secondary that didn't come up at start, or whose token the homeserver rejected, is read from the store again two minutes later, then at doubling waits up to sixteen minutes, without restarting the daemon or its other accounts; one that's up and then rejected ends its sync loop, the daemon exits and its timer restarts it | | `swarm/agents//matrix-recovery/` | `swarm-controller`, a random passphrase for every account listed under the agent's `matrix/`, in a five-minute pass | the agent container itself, under the certificate its hive passed in, when `hive-matrix-daemon` brings the account up | ❌ never replaced: the account's server-side key backup opens only with the passphrase it was enabled with, so the pass writes only to an empty path | ❌ read when the account comes up; an account whose passphrase lands later turns backup on at the next daemon start | | `swarm/controller/swarm-controller/matrix/appservice-token` | `swarm-matrix-ctl`, inside the `hive-matrix` container, once | `swarm-controller`, under its own certificate | ❌ `swarm-matrix-ctl` mints it once; the container keeps its copy and republishes it when the store's differs | ✅ the controller reads it on every five-minute matrix pass | | `swarm/controller/swarm-controller/oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | `swarm-controller`, under its own certificate, at start and on refusal | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ✅ read at start and held in memory; read again when authelia refuses it, at most once per backoff of 5 s doubling to 5 min while the store holds the refused value | 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/accounts.rs b/hive-matrix-mcp/src/accounts.rs index 7aff40f5..a02d6c86 100644 --- a/hive-matrix-mcp/src/accounts.rs +++ b/hive-matrix-mcp/src/accounts.rs @@ -26,7 +26,7 @@ use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{Arc, RwLock}; use anyhow::Context as _; use matrix_sdk::Client; @@ -276,11 +276,11 @@ pub struct AccountStatus { } /// Account name → live `Client` map plus the primary-account name used -/// when a request omits `account`. Built once at daemon startup from -/// the accounts that successfully restored a session. +/// when a request omits `account`. Built at daemon startup from the +/// accounts that restored a session; a secondary that restores later joins it. pub struct Registry { primary: String, - by_name: HashMap>, + by_name: RwLock>>, } /// Pick which account name a request resolves to, or an error. Pure (no @@ -314,19 +314,29 @@ impl Registry { pub fn new(primary: String) -> Self { Self { primary, - by_name: HashMap::new(), + by_name: RwLock::default(), } } /// Register a restored client under `name`. - pub fn insert(&mut self, name: String, client: Client) { - self.by_name.insert(name, Arc::new(client)); + pub fn insert(&self, name: String, client: Client) { + self.by_name + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(name, Arc::new(client)); + } + + /// The restored clients by account name. + fn clients(&self) -> std::sync::RwLockReadGuard<'_, HashMap>> { + self.by_name + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) } /// Whether any account restored successfully. #[must_use] pub fn is_empty(&self) -> bool { - self.by_name.is_empty() + self.clients().is_empty() } /// Snapshot every restored account: name, homeserver, user id, and @@ -338,7 +348,7 @@ impl Registry { #[must_use] pub fn list(&self) -> Vec { let mut out: Vec = self - .by_name + .clients() .iter() .map(|(name, client)| AccountStatus { name: name.clone(), @@ -370,11 +380,12 @@ impl Registry { /// configured, or when the named account (or the primary, if `None`) /// has no live client — e.g. an unknown name, or the primary failed to /// restore at startup. - pub fn resolve(&self, account: Option<&str>) -> Result<&Arc, String> { - let mut names: Vec<&str> = self.by_name.keys().map(String::as_str).collect(); + pub fn resolve(&self, account: Option<&str>) -> Result, String> { + let clients = self.clients(); + let mut names: Vec<&str> = clients.keys().map(String::as_str).collect(); names.sort_unstable(); let name = pick_name(account, &self.primary, &names)?; - self.by_name.get(name).ok_or_else(|| { + clients.get(name).cloned().ok_or_else(|| { format!( "unknown matrix account {name:?}; available accounts: [{}]", names.join(", ") @@ -385,7 +396,8 @@ impl Registry { /// Publish the live-account snapshot ([`Registry::list`]) to `path`. /// Thin wrapper over [`write_accounts_snapshot`] (split out so the /// atomic-write logic is unit-testable without a live `Client`). - /// Called once at daemon startup after all restores. + /// Called at daemon startup after all restores, and again each time a + /// secondary comes up later. /// /// # Errors /// Propagates filesystem errors from the atomic write. @@ -408,13 +420,14 @@ impl Registry { } /// Atomically write `accounts` as pretty-printed JSON to `path` -/// (`.tmp` + rename) so a dashboard reader never sees a partial +/// (a unique `.….tmp` + rename) so a dashboard reader never sees a partial /// file. Idempotent — skips the rewrite when the on-disk content already -/// matches, keeping the mtime stable. Used for the boot-time publish. +/// matches, keeping the mtime stable. Used for the boot-time publish and +/// when a secondary comes up later. /// /// The daemon rebuilds this file fresh on every boot, and restarts itself /// when the store's linked accounts change, so the file lists exactly the -/// accounts that restored at the last start. Real-time liveness is conveyed by the file mtime, +/// accounts that have restored since the last start. Real-time liveness is conveyed by the file mtime, /// which the daemon advances on a timer via /// [`heartbeat_accounts_snapshot`] — a stalled mtime means the daemon is /// down, so a reader can treat an old snapshot as stale. @@ -449,6 +462,9 @@ fn write_accounts_snapshot_inner( accounts: &[AccountStatus], force: bool, ) -> anyhow::Result<()> { + // Names a tmp file unique per write: the heartbeat and a late-joining + // account write concurrently. + static WRITES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); let body = serde_json::to_string_pretty(accounts).expect("Vec is always serialisable"); if !force && std::fs::read_to_string(path).ok().as_deref() == Some(body.as_str()) { @@ -457,7 +473,8 @@ fn write_accounts_snapshot_inner( if let Some(parent) = path.parent() { std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; } - let tmp = path.with_extension("json.tmp"); + let n = WRITES.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let tmp = path.with_extension(format!("json.{}.{n}.tmp", std::process::id())); std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?; std::fs::rename(&tmp, path).with_context(|| { format!( @@ -658,6 +675,28 @@ mod tests { ] } + #[test] + fn concurrent_writers_do_not_clobber_each_other_s_tmp_file() { + let dir = std::env::temp_dir().join(format!("hh-acct-race-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("matrix-accounts.json"); + let writers: Vec<_> = (0..4) + .map(|_| { + let path = path.clone(); + std::thread::spawn(move || { + for _ in 0..200 { + heartbeat_accounts_snapshot(&path, &sample()).unwrap(); + } + }) + }) + .collect(); + for w in writers { + w.join().unwrap(); + } + assert_eq!(std::fs::read_dir(&dir).unwrap().count(), 1); + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn snapshot_writes_json_and_cleans_up_tmp() { let dir = std::env::temp_dir().join(format!( @@ -681,7 +720,11 @@ mod tests { assert_eq!(parsed[1]["homeserver"], "https://matrix.org"); assert!(parsed[1]["user_id"].is_null()); // No leftover temp file after the atomic publish. - assert!(!path.with_extension("json.tmp").exists()); + assert_eq!( + std::fs::read_dir(&dir).unwrap().count(), + 1, + "only the snapshot is left" + ); std::fs::remove_dir_all(&dir).ok(); } @@ -737,7 +780,11 @@ mod tests { let written = std::fs::read_to_string(&path).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&written).unwrap(); assert_eq!(parsed[0]["name"], "main"); - assert!(!path.with_extension("json.tmp").exists()); + assert_eq!( + std::fs::read_dir(&dir).unwrap().count(), + 1, + "only the snapshot is left" + ); std::fs::remove_dir_all(&dir).ok(); } diff --git a/hive-matrix-mcp/src/client.rs b/hive-matrix-mcp/src/client.rs index 8b61e7ec..eff9a70d 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -43,8 +43,9 @@ use crate::credential::{Passphrase, Token}; /// Sentinel returned when `build_and_restore` detects that the token is /// permanently invalid (`M_UNKNOWN_TOKEN`). The token has already been -/// discarded ([`Token::discard_stale`]). Callers should NOT retry — the -/// account needs re-provisioning. +/// discarded ([`Token::discard_stale`]). Retrying the same token at once +/// won't help — the account needs a new one, which the daemon's periodic +/// retry of a down secondary picks up. /// /// Distinct from the general `anyhow::Error` path so callers can use /// `err.downcast_ref::()` to distinguish "retry @@ -130,13 +131,14 @@ pub async fn build_and_restore( } std::process::exit(0); } - // Secondary: token discarded (so it's cleanly skipped next boot - // rather than re-erroring); leave the sdk state in place in case - // the operator re-provisions a fresh token for the same device. + // Secondary: a file token is gone, a store token is untouched; the + // daemon tries the account again later and reads the token afresh. + // The sdk state stays in place in case the operator re-provisions a + // token for the same device. // Return a PermanentBringUpError so the caller can distinguish - // "don't retry" from a transient network/DNS failure. + // "retrying now won't help" from a transient network/DNS failure. return Err(anyhow::Error::new(PermanentBringUpError( - "matrix token rejected (M_UNKNOWN_TOKEN); discarded stale token, skipping account".into(), + "matrix token rejected (M_UNKNOWN_TOKEN); discarded stale token, leaving account down".into(), ))); } return Err(e); diff --git a/hive-matrix-mcp/src/credential.rs b/hive-matrix-mcp/src/credential.rs index 97324ca0..cc7b2ce2 100644 --- a/hive-matrix-mcp/src/credential.rs +++ b/hive-matrix-mcp/src/credential.rs @@ -172,7 +172,7 @@ impl Token { Origin::File(p) => { let _ = tokio::fs::remove_file(p).await; } - Origin::Store(path) => tracing::warn!( + Origin::Store(path) => tracing::debug!( credential = %path, "the stored matrix credential was rejected; leaving it alone (this agent may only read it)" ), @@ -543,8 +543,8 @@ mod tests { #[tokio::test] async fn an_unprovisioned_account_has_no_token_rather_than_an_error() { - // The daemon skips such an account (a secondary) or waits on the path - // watcher (the primary); neither is a failure. + // The daemon leaves such a secondary down and retries it, or waits on + // the path watcher (the primary); neither is a failure. let missing = std::env::temp_dir().join("hh-cred-definitely-not-here/matrix-token"); assert!(Token::from_file(&missing).await.unwrap().is_none()); } diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index ca9277b3..2c47d980 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -695,11 +695,8 @@ pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse { // Clear the invite todo so the accepted invite drops out of // `get_loose_ends` immediately (the sweep would also clear it // on its next tick, but this makes it instant). - let _ = crate::wake::send_todo_clear( - Some(&crate::timeline::invite_key(room.room_id())), - false, - ) - .await; + let _ = + crate::wake::send_todo_clear(&crate::timeline::invite_key(room.room_id())).await; DaemonResponse::ok(&serde_json::json!({ "joined": true, "room_id": room.room_id().to_string(), @@ -733,11 +730,9 @@ pub async fn resolve_invite( match room.leave().await { Ok(()) => { // Clear the invite todo immediately on reject. - let _ = crate::wake::send_todo_clear( - Some(&crate::timeline::invite_key(room.room_id())), - false, - ) - .await; + let _ = + crate::wake::send_todo_clear(&crate::timeline::invite_key(room.room_id())) + .await; DaemonResponse::ok(&serde_json::json!({ "rejected": true, "room_id": room.room_id().to_string(), diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 6200ebe9..e981f706 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -16,26 +16,28 @@ //! routes to its `account` arg (the primary account when omitted). //! 4. Every [`LINKED_REFRESH`], re-read the linked accounts; exit with //! [`ACCOUNTS_CHANGED_EXIT`] when the set changed, so systemd restarts us. +//! 5. On the same period, try again each secondary that did not come up +//! ([`retry_down`], backoff in [`retry_delay`]). One that comes up joins +//! the registry; the accounts already up are not restarted. //! -//! Standalone-degraded boot: the PRIMARY account having no token → -//! exit 0 cleanly; systemd restarts us once one exists (the path-watcher for -//! a token file, the store re-check timer for a stored token). A SECONDARY -//! account missing its token is skipped (the daemon still serves the others). -//! -//! Stale-token recovery (`M_UNKNOWN_TOKEN`): handled in -//! `client::build_and_restore`, and it mirrors the missing-token policy -//! above — a rejected PRIMARY token drops the stale token + sdk state and -//! exits 0 until a fresh token exists, while a rejected SECONDARY token -//! is removed and that one account is skipped so the daemon keeps serving -//! the primary and any other healthy account. +//! The PRIMARY account having no token, or one the homeserver rejects +//! (`M_UNKNOWN_TOKEN`, see `client::build_and_restore`, which also drops its +//! sdk state) → exit 0; systemd restarts us once a token exists (the +//! path-watcher for a token file, the store re-check timer for a stored +//! token). A SECONDARY account with no token, a rejected one, or a down +//! homeserver is left down and retried (step 5) while the others serve. -use std::collections::{BTreeSet, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::convert::Infallible; +use std::path::Path; use std::process::ExitCode; use std::sync::Arc; use anyhow::{Context, Result}; use clap::Parser; +use futures_util::stream::{FuturesUnordered, StreamExt}; use matrix_sdk::{Client, config::SyncSettings}; +use tokio::time::Instant; use hive_matrix_mcp::accounts::{AccountCfg, Registry}; use hive_matrix_mcp::client::PermanentBringUpError; @@ -57,15 +59,22 @@ struct Cli { /// `select_all`) rather than `tokio::spawn`/`JoinSet`. type SyncLoop = std::pin::Pin>>>; +/// A [`SyncLoop`] that yields its account's name when it ends. +type NamedSyncLoop = std::pin::Pin)>>>; + +/// What [`retry_down`] needs to bring a secondary up again: its config and +/// its todo tag. +type Retry = (AccountCfg, Option); + /// How often the daemon rewrites the accounts snapshot to advance its /// mtime, so the dashboard's `as_of` tracks daemon liveness. 30s keeps /// the staleness window small while the rewrite cost (one tmp + rename of /// a tiny file) is negligible. const ACCOUNTS_HEARTBEAT_SECS: u64 = 30; -/// Retry delays (seconds) for transient secondary-account failures. -/// After the initial attempt we back off through these before giving up -/// and skipping the account for the rest of the daemon lifetime. +/// Retry delays (seconds) for transient secondary-account failures at +/// start. After the initial attempt we back off through these before +/// leaving the account to [`retry_down`]. /// Transient = anything that is NOT a `PermanentBringUpError` (bad/expired /// token). A down homeserver at DNS-not-ready boot time is the typical /// case; the total wait is ~52s before we give up. @@ -75,6 +84,19 @@ const SECONDARY_RETRY_DELAYS_SECS: &[u64] = &[2, 5, 15, 30]; /// account comes up within this long of the swarm UI storing it. const LINKED_REFRESH: std::time::Duration = std::time::Duration::from_mins(2); +/// The longest wait between two tries of a secondary account that is down. +/// A rejected token is tried again at this pace until an operator links a +/// new one, so it bounds both the homeserver's load and how long a re-link +/// takes to come up. +const MAX_SECONDARY_RETRY: std::time::Duration = std::time::Duration::from_mins(16); + +/// Why a secondary is down when it had no token to try. +const NO_TOKEN: &str = "no token"; + +/// Why a secondary is down when start-up gave up on it; the reason was logged +/// there. +const FAILED_AT_START: &str = "failed at start"; + /// The exit status the daemon leaves with when the store's linked accounts /// changed. `nix/agent-modules/matrix.nix` names the same number in the unit's /// `RestartForceExitStatus` and `SuccessExitStatus`, so systemd restarts it @@ -96,6 +118,7 @@ async fn main() -> Result { .init(); let cli = Cli::parse(); + let todo_socket = wake::agent_socket(); let declared = accounts::configured().context("read matrix account config")?; accounts::remove_delivered_files(&declared); let mut warned = HashSet::new(); @@ -105,8 +128,9 @@ async fn main() -> Result { cfgs.extend(linked); let multi = cfgs.len() > 1; let primary = cfgs[0].name.clone(); - let mut registry = Registry::new(primary); - let mut sync_loops: Vec = Vec::new(); + 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; @@ -115,8 +139,8 @@ async fn main() -> Result { 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); - sync_loops.push(sync_loop); } // No token yet: for the primary that means the daemon isn't // useful — exit 0, and systemd restarts us once a token exists @@ -130,20 +154,28 @@ async fn main() -> Result { return Ok(ExitCode::SUCCESS); } Ok(None) => { - tracing::warn!(account = %cfg.name, "secondary matrix account has no token; skipping"); + 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 - // before being skipped for this daemon lifetime. + // here, then left to `retry_down`. Err(e) if is_primary => return Err(e.context("bring up primary matrix account")), - Err(e) => { - if let Some((client, sync_loop)) = bring_up_secondary_with_retry(&cfg, tag, e).await - { + 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); - sync_loops.push(sync_loop); } - } + None => { + down.failed( + cfg.name.clone(), + (cfg, tag), + Instant::now(), + FAILED_AT_START, + ); + } + }, } } @@ -152,6 +184,17 @@ async fn main() -> Result { return Ok(ExitCode::SUCCESS); } + // Startup cancel-and-recreate (loose-ends v2): wipe this agent's + // matrix todos so stale ones (rooms read / invites resolved while the + // daemon was down) don't linger, then let the first sweep rebuild the + // set to match current reality. Best-effort; the sweep converges. + // Once, before any sync loop runs: a secondary that comes up later must + // not wipe the todos the running accounts' sweeps already deduped, so + // `todo_socket` is passed to nothing that brings an account up. + if let Some(socket) = &todo_socket { + let _ = wake::clear_all_todos(socket).await; + } + // Publish the live-account snapshot (BE-4) for the dashboard: the // accounts that restored, with effective homeserver + user id. Lands // in the host-visible state dir so hive-c0re reads it to show live @@ -193,37 +236,54 @@ async fn main() -> Result { } }); - Ok(drive(sync_loops, wait_for_linked_change(declared, booted, warned)).await) + let (up_tx, up_rx) = tokio::sync::mpsc::unbounded_channel(); + let retry = Box::pin(retry_down(down, Arc::clone(®istry), up_tx)); + let changed = wait_for_linked_change(declared, booted, warned); + Ok(drive(sync_loops, up_rx, changed, retry).await) } /// Drive all per-account sync loops concurrently on this task (they aren't -/// `Send`, so no `tokio::spawn`) until one returns or `changed` resolves. +/// `Send`, so no `tokio::spawn`) until one returns or `changed` resolves, +/// adding each loop `up` delivers as `retry` brings a secondary up. /// matrix-sdk reconnects internally, so any loop returning is exceptional — /// log it and exit so systemd restarts the whole daemon cleanly. async fn drive( - sync_loops: Vec, + sync_loops: Vec, + mut up: tokio::sync::mpsc::UnboundedReceiver, changed: impl std::future::Future, + retry: impl std::future::Future, ) -> ExitCode { - tokio::select! { - (result, idx, _rest) = futures_util::future::select_all(sync_loops) => { - match result { - Ok(()) => tracing::warn!( - account_index = idx, - "a matrix sync loop exited cleanly; restarting daemon" - ), - Err(e) => { - tracing::error!(account_index = idx, error = %format!("{e:#}"), "a matrix sync loop errored; restarting daemon"); + let mut loops: FuturesUnordered = sync_loops.into_iter().collect(); + tokio::pin!(changed, retry); + loop { + tokio::select! { + Some((account, result)) = loops.next() => { + match result { + Ok(()) => tracing::warn!( + %account, + "a matrix sync loop exited cleanly; restarting daemon" + ), + Err(e) => { + tracing::error!(%account, error = %format!("{e:#}"), "a matrix sync loop errored; restarting daemon"); + } } + return ExitCode::SUCCESS; } - ExitCode::SUCCESS - } - () = changed => { - tracing::info!("the store's linked matrix accounts changed; exiting to be restarted onto them"); - ExitCode::from(ACCOUNTS_CHANGED_EXIT) + () = &mut changed => { + tracing::info!("the store's linked matrix accounts changed; exiting to be restarted onto them"); + return ExitCode::from(ACCOUNTS_CHANGED_EXIT); + } + never = &mut retry => match never {}, + Some(sync_loop) = up.recv() => loops.push(sync_loop), } } } +/// `sync_loop`, yielding `account` beside its result. +fn named(account: String, sync_loop: SyncLoop) -> NamedSyncLoop { + Box::pin(async move { (account, sync_loop.await) }) +} + /// The accounts the store links to this agent beyond `declared`, or `None` /// when the store could not be read. /// @@ -298,15 +358,15 @@ async fn wait_for_linked_change( /// 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 -/// that the account should be skipped for this daemon lifetime (permanent +/// 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( cfg: &AccountCfg, tag: Option, first_error: anyhow::Error, ) -> Option<(Client, SyncLoop)> { - // Permanent failure: the token was invalid/expired and has already been - // removed from disk. Retrying won't help — skip immediately. + // 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 .downcast_ref::() .is_some() @@ -314,7 +374,7 @@ async fn bring_up_secondary_with_retry( tracing::error!( account = %cfg.name, error = %format!("{first_error:#}"), - "secondary matrix account: permanent failure; skipping" + "secondary matrix account: permanent failure; leaving it down" ); return None; } @@ -337,14 +397,14 @@ async fn bring_up_secondary_with_retry( return Some((client, sync_loop)); } Ok(None) => { - tracing::warn!(account = %cfg.name, "secondary matrix account has no token; skipping"); + tracing::warn!(account = %cfg.name, "secondary matrix account has no token; leaving it down"); return None; } Err(re) if re.downcast_ref::().is_some() => { tracing::error!( account = %cfg.name, error = %format!("{re:#}"), - "secondary matrix account: permanent failure on retry; skipping" + "secondary matrix account: permanent failure on retry; leaving it down" ); return None; } @@ -360,11 +420,170 @@ async fn bring_up_secondary_with_retry( tracing::error!( account = %cfg.name, retries = SECONDARY_RETRY_DELAYS_SECS.len(), - "secondary matrix account failed after all retries; skipping for this daemon lifetime" + "secondary matrix account failed after all retries; leaving it down" ); None } +/// The wait before the next try of a secondary that has failed `failures` +/// times in a row: one [`LINKED_REFRESH`], doubled per further failure, up to +/// [`MAX_SECONDARY_RETRY`]. +fn retry_delay(failures: u32) -> std::time::Duration { + let doublings = failures.saturating_sub(1).min(16); + LINKED_REFRESH + .saturating_mul(1 << doublings) + .min(MAX_SECONDARY_RETRY) +} + +/// Secondary accounts that did not come up, each with what it takes to try it +/// again (`T`) and the earliest time to. +struct DownAccounts { + by_name: BTreeMap>, +} + +struct Down { + retry: T, + failures: u32, + not_before: Instant, + /// Why the last try failed. + reason: String, +} + +impl Default for DownAccounts { + fn default() -> Self { + Self { + by_name: BTreeMap::new(), + } + } +} + +impl DownAccounts { + /// Record that `name` failed to come up at `at`, for `reason`. `retry` is + /// kept from the first failure on. True when the failure is news — the + /// account's first, or for a different reason than the last — and so + /// worth a warning. + fn failed(&mut self, name: String, retry: T, at: Instant, reason: &str) -> bool { + let down = self.by_name.entry(name).or_insert(Down { + retry, + failures: 0, + not_before: at, + reason: String::new(), + }); + let news = down.failures == 0 || down.reason != reason; + reason.clone_into(&mut down.reason); + down.failures += 1; + down.not_before = at + retry_delay(down.failures); + news + } + + /// Record that `name` came up; it is never due again. + fn came_up(&mut self, name: &str) { + self.by_name.remove(name); + } + + /// The account to try at `now`: of those whose wait is over, the one + /// that has waited longest past it. + fn due(&self, now: Instant) -> Option<(&str, &T)> { + self.by_name + .iter() + .filter(|(_, d)| d.not_before <= now) + .min_by_key(|(_, d)| d.not_before) + .map(|(name, d)| (name.as_str(), &d.retry)) + } +} + +/// What a restore attempt yields; [`restore_account`]'s result. +type Restore = Result>; + +/// Every [`LINKED_REFRESH`], try again the secondary [`DownAccounts::due`] +/// names, one per tick, with [`restore_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| async move { restore_account(&cfg, false).await }, + ) + .await +} + +/// [`retry_down`] with the restore step and the snapshot path injected. +async fn retry_loop( + down: &mut DownAccounts, + registry: &Registry, + up: &tokio::sync::mpsc::UnboundedSender, + snapshot: &Path, + mut restore: F, +) -> Infallible +where + F: FnMut(AccountCfg) -> 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; + retry_once(down, registry, up, snapshot, now, &mut restore).await; + } +} + +/// 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. +async fn retry_once( + down: &mut DownAccounts, + registry: &Registry, + up: &tokio::sync::mpsc::UnboundedSender, + snapshot: &Path, + now: Instant, + restore: &mut F, +) where + F: FnMut(AccountCfg) -> 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 restore(cfg.clone()).await { + Ok(Some(client)) => { + tracing::info!(account = %name, "secondary matrix account came up"); + let sync_loop = account_sync_loop(&client, tag); + 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"); + } + // `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"); + } + } + } +} + /// Resolve `cfg`'s access token, preferring the copy this agent can fetch /// itself. /// @@ -430,16 +649,25 @@ async fn ensure_recovery(client: &Client, account: &str) { } } -/// Restore one account's client (when its token exists), install its -/// message handler, and build its sync loop. Returns -/// `Ok(Some((client, sync_loop)))` when the account came up, `Ok(None)` -/// when its token file is absent (caller decides primary-vs-secondary -/// handling). The returned sync loop is driven by the caller. +/// 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> { let Some(homeserver) = cfg.homeserver() else { // No homeserver for this account: the hive has none to offer (no // matrix vhost) or this agent's `services.hyperhive.agent.matrix.url` is null. Same @@ -474,7 +702,13 @@ async fn bring_up_account( client::ensure_cross_signing(&client, &cfg.name).await; // Best-effort, and after cross-signing: see `client::ensure_recovery`. ensure_recovery(&client, &cfg.name).await; + Ok(Some(client)) +} +/// `client`'s sync loop, sweeping invites and unread rooms into todos tagged +/// `tag` after each sync. Builds the loop only; nothing runs until it is +/// polled. Never wipes the todo set: that is `main`'s, once, at startup. +fn account_sync_loop(client: &Client, tag: Option) -> SyncLoop { let sync_client = client.clone(); let cb_client = client.clone(); // Separate dedup sets: one tracks invites already pushed as todos, the @@ -482,12 +716,7 @@ async fn bring_up_account( // each sweep (see the sweep fns) so re-invites / new messages re-push. let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); - // Startup cancel-and-recreate (loose-ends v2): wipe this agent's - // matrix todos so stale ones (rooms read / invites resolved while the - // daemon was down) don't linger, then let the first sweep rebuild the - // set to match current reality. Best-effort; the sweep converges. - let _ = wake::send_todo_clear(None, true).await; - let sync_loop: SyncLoop = Box::pin(async move { + Box::pin(async move { sync_client .sync_with_callback(SyncSettings::default(), move |_response| { let client = cb_client.clone(); @@ -503,6 +732,438 @@ async fn bring_up_account( .await .context("matrix-sdk sync loop exited")?; Ok(()) - }); - Ok(Some((client, sync_loop))) + }) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use tokio::time::Instant; + + 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, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, Registry, + Restore, Retry, retry_delay, retry_loop, retry_once, wake, + }; + + #[test] + fn the_wait_starts_at_one_refresh_and_doubles_up_to_the_cap() { + assert_eq!(retry_delay(1), LINKED_REFRESH); + assert_eq!(retry_delay(2), LINKED_REFRESH * 2); + assert_eq!(retry_delay(3), LINKED_REFRESH * 4); + assert_eq!(retry_delay(5), MAX_SECONDARY_RETRY); + assert_eq!(retry_delay(u32::MAX), MAX_SECONDARY_RETRY); + } + + #[test] + fn a_failed_account_is_due_once_its_wait_is_over_and_not_before() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + down.failed("ccc".to_owned(), 7, t0, "r"); + + assert_eq!(down.due(t0), None); + assert_eq!(down.due(t0 + LINKED_REFRESH - Duration::from_secs(1)), None); + assert_eq!(down.due(t0 + LINKED_REFRESH), Some(("ccc", &7))); + } + + #[test] + fn each_failure_in_a_row_pushes_the_next_try_further_out() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + down.failed("ccc".to_owned(), (), t0, "r"); + let t1 = t0 + LINKED_REFRESH; + down.failed("ccc".to_owned(), (), t1, "r"); + + assert_eq!(down.due(t1 + LINKED_REFRESH), None); + assert_eq!(down.due(t1 + LINKED_REFRESH * 2), Some(("ccc", &()))); + } + + #[test] + fn an_account_that_came_up_is_never_due_again() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + down.failed("ccc".to_owned(), (), t0, "r"); + down.came_up("ccc"); + + assert_eq!(down.due(t0 + MAX_SECONDARY_RETRY * 10), None); + } + + #[test] + fn the_longest_overdue_account_goes_first_and_the_others_keep_their_schedule() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + down.failed( + "late".to_owned(), + "late cfg", + t0 + Duration::from_secs(30), + "r", + ); + down.failed("early".to_owned(), "early cfg", t0, "r"); + let now = t0 + LINKED_REFRESH * 2; + + assert_eq!(down.due(now), Some(("early", &"early cfg"))); + down.failed("early".to_owned(), "early cfg", now, "r"); + assert_eq!(down.due(now), Some(("late", &"late cfg"))); + } + + #[test] + fn a_failure_is_news_the_first_time_and_when_its_reason_changes() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + assert!(down.failed("ccc".to_owned(), (), t0, "rejected")); + assert!(!down.failed("ccc".to_owned(), (), t0, "rejected")); + assert!(down.failed("ccc".to_owned(), (), t0, "no token")); + assert!(!down.failed("ccc".to_owned(), (), t0, "no token")); + assert!(down.failed("other".to_owned(), (), t0, "no token")); + } + + #[test] + fn the_first_retry_target_is_kept_across_failures() { + let t0 = Instant::now(); + let mut down = DownAccounts::default(); + down.failed("ccc".to_owned(), "first", t0, "r"); + down.failed("ccc".to_owned(), "second", t0, "r"); + + 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")), + } + } + + /// 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| { + let c = ccc.take().expect("tried once"); + async move { Ok(Some(c)) as Restore } + }; + + 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| { + let r = outcome.take().expect("tried once"); + async move { r as Restore } + }; + + 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 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 listener = tokio::net::UnixListener::bind(&socket).unwrap(); + let ops = Arc::new(std::sync::Mutex::new(Vec::new())); + let seen = Arc::clone(&ops); + let harness = tokio::spawn(async move { + loop { + let (mut conn, _) = listener.accept().await.unwrap(); + let mut op = String::new(); + conn.read_to_string(&mut op).await.unwrap(); + seen.lock().unwrap().push(op); + conn.write_all(b"{\"ok\":true}\n").await.unwrap(); + } + }); + // Startup's wipe, as `main` runs it: the one the harness should see. + 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 mut ccc = Some(client("ccc").await); + let mut restore = move |_: AccountCfg| { + let c = ccc.take().expect("tried once"); + async move { Ok(Some(c)) as Restore } + }; + retry_once( + &mut down, + ®, + &up, + &snap, + t0 + LINKED_REFRESH, + &mut restore, + ) + .await; + // 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(); + } + + #[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| { + tried += 1; + async { Ok(None) as Restore } + }; + + 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| { + let gate = gate_rx.take().expect("tried once"); + let c = ccc.take().expect("tried once"); + async move { + gate.await.unwrap(); + Ok(Some(c)) as Restore + } + }; + + // 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| async move { Ok(Some(client(&c.name).await)) as Restore }; + 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| { + calls.borrow_mut().push(t0.elapsed().as_secs() / 60); + async { Err(anyhow!("homeserver down")) as Restore } + }; + + 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"]); + } } diff --git a/hive-matrix-mcp/src/mcp.rs b/hive-matrix-mcp/src/mcp.rs index b94ab745..2fd51907 100644 --- a/hive-matrix-mcp/src/mcp.rs +++ b/hive-matrix-mcp/src/mcp.rs @@ -1,7 +1,7 @@ //! MCP tool surface for `hive-matrix-daemon`, served directly over //! streamable-http — no stdio bridge, no per-turn respawn, no //! round-trip unix socket. The daemon already owns the account -//! registry in-process (built at startup from the restored matrix-sdk +//! registry in-process (the restored matrix-sdk //! `Client`s), so each tool call resolves `account` against //! [`crate::accounts::Registry`] and calls straight into //! [`crate::handlers`]. Mirrors `hive-bash-mcp::mcp`'s shape (and, one @@ -257,10 +257,8 @@ impl MatrixMcp { /// account" / "ambiguous, pick one" error the same way a handler /// error would render (so a bad `account` arg and a bad room/event /// arg look the same to claude). - fn resolve(&self, account: Option<&str>) -> Result<&matrix_sdk::Client, String> { - self.registry - .resolve(account) - .map(std::convert::AsRef::as_ref) + fn resolve(&self, account: Option<&str>) -> Result, String> { + self.registry.resolve(account) } } @@ -278,7 +276,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::send_message(client, &args.room, &args.body).await) + render(handlers::send_message(&client, &args.room, &args.body).await) } #[tool(description = "Open (or reuse) a direct message room with `user_id` \ @@ -290,7 +288,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::send_dm(client, &args.user_id, &args.body).await) + render(handlers::send_dm(&client, &args.user_id, &args.body).await) } #[tool( @@ -305,7 +303,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::send_file(client, &args.room, &args.path, args.caption.as_deref()).await) + render(handlers::send_file(&client, &args.room, &args.path, args.caption.as_deref()).await) } #[tool(description = "Resolve (find-or-create) the DM room with `user_id` \ @@ -317,7 +315,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::open_dm(client, &args.user_id).await) + render(handlers::open_dm(&client, &args.user_id).await) } #[tool( @@ -330,7 +328,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::send_reaction(client, &args.room, &args.event_id, &args.key).await) + render(handlers::send_reaction(&client, &args.room, &args.event_id, &args.key).await) } #[tool(description = "Reply to a specific matrix event in a room, threaded \ @@ -342,7 +340,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::send_reply(client, &args.room, &args.event_id, &args.body).await) + render(handlers::send_reply(&client, &args.room, &args.event_id, &args.body).await) } #[tool(description = "Mark a specific event as read for this agent. Updates \ @@ -353,7 +351,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::mark_read(client, &args.room, &args.event_id).await) + render(handlers::mark_read(&client, &args.room, &args.event_id).await) } #[tool(description = "Redact (delete) a specific matrix event in a room — \ @@ -366,7 +364,8 @@ impl MatrixMcp { Err(e) => return format!("matrix error: {e}"), }; render( - handlers::send_redact(client, &args.room, &args.event_id, args.reason.as_deref()).await, + handlers::send_redact(&client, &args.room, &args.event_id, args.reason.as_deref()) + .await, ) } @@ -379,7 +378,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::list_rooms(client).await) + render(handlers::list_rooms(&client).await) } #[tool( @@ -393,7 +392,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::list_invites(client).await) + render(handlers::list_invites(&client).await) } #[tool( @@ -408,7 +407,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::join_room(client, &args.room).await) + render(handlers::join_room(&client, &args.room).await) } #[tool( @@ -431,7 +430,7 @@ impl MatrixMcp { ); } }; - render(handlers::resolve_invite(client, &args.room, action).await) + render(handlers::resolve_invite(&client, &args.room, action).await) } #[tool( @@ -445,7 +444,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::invite_user(client, &args.room, &args.user_id).await) + render(handlers::invite_user(&client, &args.room, &args.user_id).await) } #[tool( @@ -457,7 +456,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::list_room_members(client, &args.room).await) + render(handlers::list_room_members(&client, &args.room).await) } #[tool(description = "Read events from a matrix room (default 50, max 200), \ @@ -474,7 +473,7 @@ impl MatrixMcp { Ok(c) => c, Err(e) => return format!("matrix error: {e}"), }; - render(handlers::read_room(client, &args.room, args.limit, args.from, args.until).await) + render(handlers::read_room(&client, &args.room, args.limit, args.from, args.until).await) } #[tool( @@ -491,7 +490,7 @@ impl MatrixMcp { }; render( handlers::download_file( - client, + &client, &args.room, &args.event_id, args.dest_path.as_deref(), diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index e1447f8a..56b2decd 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -60,10 +60,7 @@ pub async fn sweep_unread( .collect() }; for id in stale { - if wake::send_todo_clear(Some(id.as_str()), false) - .await - .is_ok() - { + if wake::send_todo_clear(id.as_str()).await.is_ok() { notified.lock().await.remove(&id); } } @@ -121,10 +118,7 @@ pub async fn sweep_invites( .collect() }; for id in stale { - if wake::send_todo_clear(Some(&invite_key(&id)), false) - .await - .is_ok() - { + if wake::send_todo_clear(&invite_key(&id)).await.is_ok() { notified.lock().await.remove(&id); } } diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index ddf3c4d0..d8e84d58 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -23,7 +23,8 @@ const TODO_SOCKET_RETRY: Retry = Retry::None; /// go — distinct from the host-served control socket used by `send_wake`. /// `None` when unset/empty, in which case todo sends are a best-effort /// no-op (a standalone daemon without the harness socket). -fn agent_socket() -> Option { +#[must_use] +pub fn agent_socket() -> Option { std::env::var_os("HIVE_AGENT_SOCKET") .filter(|v| !v.is_empty()) .map(std::path::PathBuf::from) @@ -52,26 +53,40 @@ pub async fn send_todo_upsert(key: &str, summary: impl AsRef) -> Result<()> notify(&socket, &payload, TODO_SOCKET_RETRY).await } -/// Clear matrix-subsystem todos on the harness's in-agent socket. `key = -/// Some(room)` clears one room's todo (it was read); `all = true` wipes the -/// whole matrix set (cancel-and-recreate on daemon restart). Best-effort: -/// a no-op when `HIVE_AGENT_SOCKET` isn't configured. +/// Clear the matrix-subsystem todo `key` on the harness's in-agent socket +/// (its room was read, or its invite resolved). Best-effort: a no-op when +/// `HIVE_AGENT_SOCKET` isn't configured. /// /// # Errors /// /// Returns an error on socket connect failure, serialisation failure, /// or I/O error writing to or reading from the socket. -pub async fn send_todo_clear(key: Option<&str>, all: bool) -> Result<()> { +pub async fn send_todo_clear(key: &str) -> Result<()> { let Some(socket) = agent_socket() else { return Ok(()); }; + send_clear(&socket, Some(key), false).await +} + +/// Wipe the whole matrix todo set on `socket` (cancel-and-recreate on daemon +/// start). Takes the socket instead of reading `HIVE_AGENT_SOCKET`, so only +/// code handed the path can wipe the set. +/// +/// # Errors +/// +/// As [`send_todo_clear`]. +pub async fn clear_all_todos(socket: &std::path::Path) -> Result<()> { + send_clear(socket, None, true).await +} + +async fn send_clear(socket: &std::path::Path, key: Option<&str>, all: bool) -> Result<()> { let payload = serde_json::json!({ "cmd": "clear_todo", "subsystem": "matrix", "key": key, "all": all, }); - notify(&socket, &payload, TODO_SOCKET_RETRY).await + notify(socket, &payload, TODO_SOCKET_RETRY).await } /// Format a wake-message body from a list of per-room unread summaries. diff --git a/nix/agent-modules/matrix.nix b/nix/agent-modules/matrix.nix index 65276a8c..6cc1da06 100644 --- a/nix/agent-modules/matrix.nix +++ b/nix/agent-modules/matrix.nix @@ -99,7 +99,8 @@ in (for `main`, a file only when the store has no token: the swarm mints `main` into the store; an operator-supplied secret for an external one). The daemon - skips an extra account whose token file is absent, and + leaves an extra account whose token is absent down and tries + it again every few minutes, and exits cleanly to wait on the path watcher when `main`'s is. ''; }; @@ -148,8 +149,9 @@ in - runs `hive-matrix-daemon` as a systemd unit that holds a matrix-sdk Client + sync per account against that account's homeserver (`homeserver`, else - `services.hyperhive.agent.matrix.url`). The daemon auto-skips an - account whose URL or token file is missing, and a `systemd.paths` + `services.hyperhive.agent.matrix.url`). The daemon leaves an + account whose URL or token is missing down (an extra one is tried + again every few minutes), and a `systemd.paths` watcher restarts it the moment a token file appears (same path-trigger shape as `forge-avatar-sync`). - exposes the matrix tool surface (send_message, send_dm,