hive-matrix-mcp: retry a secondary account that failed to come up
The daemon's two-minute refresh compared the store's linked account names with the names it booted with. A secondary that was named but did not come up - no token at bring-up, a token the homeserver rejected (M_UNKNOWN_TOKEN), or a homeserver still down after the ~52 s boot retries - kept its name in the set, so nothing ever noticed it again: it stayed down until the daemon restarted for some other reason, and an operator re-linking a rejected account changed no name. Each such secondary is now tried again on the same two-minute period, one per tick, after one refresh and then doubling waits capped at sixteen minutes. One that comes up joins the registry (now behind a lock) and its sync loop joins the running set; the primary and the other accounts are not restarted. The store is still only read. The startup todo wipe moves out of bring_up_account to run once before any sync loop: run per account, a late secondary would have cleared the todos the running accounts' sweeps had already deduped. Part of #5017 (slice M).
This commit is contained in:
atlas
mara
parent
2fb6b39f9f
commit
7f26238c87
7 changed files with 335 additions and 100 deletions
|
|
@ -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<String, Arc<Client>>,
|
||||
by_name: RwLock<HashMap<String, Arc<Client>>>,
|
||||
}
|
||||
|
||||
/// 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<String, Arc<Client>>> {
|
||||
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<AccountStatus> {
|
||||
let mut out: Vec<AccountStatus> = 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<Client>, String> {
|
||||
let mut names: Vec<&str> = self.by_name.keys().map(String::as_str).collect();
|
||||
pub fn resolve(&self, account: Option<&str>) -> Result<Arc<Client>, 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.
|
||||
|
|
@ -410,11 +422,12 @@ impl Registry {
|
|||
/// Atomically write `accounts` as pretty-printed JSON to `path`
|
||||
/// (`<path>.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.
|
||||
|
|
|
|||
|
|
@ -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::<PermanentBringUpError>()` to distinguish "retry
|
||||
|
|
@ -134,9 +135,9 @@ pub async fn build_and_restore(
|
|||
// rather than re-erroring); leave the sdk state in place in case
|
||||
// the operator re-provisions a fresh 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);
|
||||
|
|
|
|||
|
|
@ -16,26 +16,27 @@
|
|||
//! 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::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 +58,22 @@ struct Cli {
|
|||
/// `select_all`) rather than `tokio::spawn`/`JoinSet`.
|
||||
type SyncLoop = std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>>>>;
|
||||
|
||||
/// A [`SyncLoop`] that yields its account's name when it ends.
|
||||
type NamedSyncLoop = std::pin::Pin<Box<dyn std::future::Future<Output = (String, Result<()>)>>>;
|
||||
|
||||
/// What [`retry_down`] needs to bring a secondary up again: its config and
|
||||
/// its todo tag.
|
||||
type Retry = (AccountCfg, Option<String>);
|
||||
|
||||
/// 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 +83,12 @@ 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);
|
||||
|
||||
/// 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
|
||||
|
|
@ -105,8 +119,9 @@ async fn main() -> Result<ExitCode> {
|
|||
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<SyncLoop> = Vec::new();
|
||||
let registry = Registry::new(primary);
|
||||
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;
|
||||
|
|
@ -115,8 +130,8 @@ async fn main() -> Result<ExitCode> {
|
|||
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 +145,21 @@ async fn main() -> Result<ExitCode> {
|
|||
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());
|
||||
}
|
||||
// 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()),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -152,6 +168,14 @@ async fn main() -> Result<ExitCode> {
|
|||
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.
|
||||
let _ = wake::send_todo_clear(None, true).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 +217,54 @@ async fn main() -> Result<ExitCode> {
|
|||
}
|
||||
});
|
||||
|
||||
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<SyncLoop>,
|
||||
sync_loops: Vec<NamedSyncLoop>,
|
||||
mut up: tokio::sync::mpsc::UnboundedReceiver<NamedSyncLoop>,
|
||||
changed: impl std::future::Future<Output = ()>,
|
||||
retry: impl std::future::Future<Output = Infallible>,
|
||||
) -> 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<NamedSyncLoop> = 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 +339,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<String>,
|
||||
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::<PermanentBringUpError>()
|
||||
.is_some()
|
||||
|
|
@ -314,7 +355,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 +378,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::<PermanentBringUpError>().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 +401,115 @@ 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<T> {
|
||||
by_name: BTreeMap<String, Down<T>>,
|
||||
}
|
||||
|
||||
struct Down<T> {
|
||||
retry: T,
|
||||
failures: u32,
|
||||
not_before: Instant,
|
||||
}
|
||||
|
||||
impl<T> Default for DownAccounts<T> {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
by_name: BTreeMap::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> DownAccounts<T> {
|
||||
/// Record that `name` failed to come up at `at`. `retry` is kept from the
|
||||
/// first failure on.
|
||||
fn failed(&mut self, name: String, retry: T, at: Instant) {
|
||||
let down = self.by_name.entry(name).or_insert(Down {
|
||||
retry,
|
||||
failures: 0,
|
||||
not_before: at,
|
||||
});
|
||||
down.failures += 1;
|
||||
down.not_before = at + retry_delay(down.failures);
|
||||
}
|
||||
|
||||
/// 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))
|
||||
}
|
||||
}
|
||||
|
||||
/// Every [`LINKED_REFRESH`], try again the secondary [`DownAccounts::due`]
|
||||
/// names, one per tick. One that comes up joins `registry` and its sync loop
|
||||
/// goes to `up`; the accounts already up are not touched. Never returns.
|
||||
async fn retry_down(
|
||||
mut down: DownAccounts<Retry>,
|
||||
registry: Arc<Registry>,
|
||||
up: tokio::sync::mpsc::UnboundedSender<NamedSyncLoop>,
|
||||
) -> Infallible {
|
||||
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;
|
||||
let Some((name, (cfg, tag))) = down.due(now) else {
|
||||
continue;
|
||||
};
|
||||
let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone());
|
||||
match bring_up_account(&cfg, tag.clone(), false).await {
|
||||
Ok(Some((client, sync_loop))) => {
|
||||
tracing::info!(account = %name, "secondary matrix account came up");
|
||||
down.came_up(&name);
|
||||
registry.insert(name.clone(), client);
|
||||
if let Err(e) = registry.write_snapshot(&paths::accounts_file()) {
|
||||
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) => {
|
||||
tracing::warn!(account = %name, "secondary matrix account still has no token; will retry");
|
||||
down.failed(name, (cfg, tag), now);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
account = %name,
|
||||
error = %format!("{e:#}"),
|
||||
"secondary matrix account still failing; will retry"
|
||||
);
|
||||
down.failed(name, (cfg, tag), now);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve `cfg`'s access token, preferring the copy this agent can fetch
|
||||
/// itself.
|
||||
///
|
||||
|
|
@ -482,11 +627,6 @@ 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 {
|
||||
sync_client
|
||||
.sync_with_callback(SyncSettings::default(), move |_response| {
|
||||
|
|
@ -506,3 +646,77 @@ async fn bring_up_account(
|
|||
});
|
||||
Ok(Some((client, sync_loop)))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio::time::Instant;
|
||||
|
||||
use super::{DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, retry_delay};
|
||||
|
||||
#[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);
|
||||
|
||||
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);
|
||||
let t1 = t0 + LINKED_REFRESH;
|
||||
down.failed("ccc".to_owned(), (), t1);
|
||||
|
||||
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);
|
||||
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));
|
||||
down.failed("early".to_owned(), "early cfg", t0);
|
||||
let now = t0 + LINKED_REFRESH * 2;
|
||||
|
||||
assert_eq!(down.due(now), Some(("early", &"early cfg")));
|
||||
down.failed("early".to_owned(), "early cfg", now);
|
||||
assert_eq!(down.due(now), Some(("late", &"late cfg")));
|
||||
}
|
||||
|
||||
#[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);
|
||||
down.failed("ccc".to_owned(), "second", t0);
|
||||
|
||||
assert_eq!(down.due(t0 + MAX_SECONDARY_RETRY), Some(("ccc", &"first")));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Arc<matrix_sdk::Client>, 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(),
|
||||
|
|
|
|||
Loading…
Reference in a new issue