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 f621a334..e981f706 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -118,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(); @@ -188,8 +189,11 @@ async fn main() -> Result { // 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; + // 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 @@ -488,11 +492,11 @@ impl DownAccounts { } } -/// What a bring-up attempt yields; [`bring_up_account`]'s result. -type BringUp = Result>; +/// 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 [`bring_up_account`]. Never returns. +/// names, one per tick, with [`restore_account`]. Never returns. async fn retry_down( mut down: DownAccounts, registry: Arc, @@ -503,22 +507,22 @@ async fn retry_down( ®istry, &up, &paths::accounts_file(), - |cfg: AccountCfg, tag| async move { bring_up_account(&cfg, tag, false).await }, + |cfg: AccountCfg| async move { restore_account(&cfg, false).await }, ) .await } -/// [`retry_down`] with the bring-up step and the snapshot path injected. +/// [`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 bring_up: F, + mut restore: F, ) -> Infallible where - F: FnMut(AccountCfg, Option) -> Fut, - Fut: std::future::Future, + 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); @@ -526,33 +530,34 @@ where tick.tick().await; loop { let now = tick.tick().await; - retry_once(down, registry, up, snapshot, now, &mut bring_up).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 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 -/// `bring_up` runs. +/// 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, - bring_up: &mut F, + restore: &mut F, ) where - F: FnMut(AccountCfg, Option) -> Fut, - Fut: std::future::Future, + 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 bring_up(cfg.clone(), tag.clone()).await { - Ok(Some((client, sync_loop))) => { + 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) { @@ -644,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 @@ -688,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 @@ -696,7 +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())); - 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(); @@ -712,8 +732,7 @@ async fn bring_up_account( .await .context("matrix-sdk sync loop exited")?; Ok(()) - }); - Ok(Some((client, sync_loop))) + }) } #[cfg(test)] @@ -730,8 +749,8 @@ mod tests { use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel}; use super::{ - AccountCfg, BringUp, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, - Registry, Retry, SyncLoop, retry_delay, retry_loop, retry_once, + AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, Registry, + Restore, Retry, retry_delay, retry_loop, retry_once, wake, }; #[test] @@ -833,10 +852,6 @@ mod tests { } } - fn pending_loop() -> SyncLoop { - Box::pin(std::future::pending()) - } - /// A registry holding the primary `main` and a running secondary `other`. async fn registry() -> Registry { let r = Registry::new("main".to_owned()); @@ -877,9 +892,9 @@ mod tests { let (up, mut rx) = unbounded_channel(); let snap = snapshot_path("up"); let mut ccc = Some(client("ccc").await); - let mut bring_up = move |_: AccountCfg, _: Option| { + let mut bring_up = move |_: AccountCfg| { let c = ccc.take().expect("tried once"); - async move { Ok(Some((c, pending_loop()))) as BringUp } + async move { Ok(Some(c)) as Restore } }; let now = t0 + LINKED_REFRESH; @@ -915,9 +930,9 @@ mod tests { let (up, mut rx) = unbounded_channel(); let snap = snapshot_path("fail"); let mut outcome = Some(outcome); - let mut bring_up = move |_: AccountCfg, _: Option| { + let mut bring_up = move |_: AccountCfg| { let r = outcome.take().expect("tried once"); - async move { r as BringUp } + async move { r as Restore } }; let now = t0 + LINKED_REFRESH; @@ -940,6 +955,56 @@ mod tests { } } + #[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(); @@ -947,9 +1012,9 @@ mod tests { let reg = registry().await; let (up, _rx) = unbounded_channel(); let mut tried = 0; - let mut bring_up = |_: AccountCfg, _: Option| { + let mut bring_up = |_: AccountCfg| { tried += 1; - async { Ok(None) as BringUp } + async { Ok(None) as Restore } }; let early = t0 + LINKED_REFRESH - Duration::from_secs(1); @@ -976,12 +1041,12 @@ mod tests { 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, _: Option| { + 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, pending_loop()))) as BringUp + Ok(Some(c)) as Restore } }; @@ -1050,9 +1115,8 @@ mod tests { .collect(); let (up, _rx) = unbounded_channel(); let snap = snapshot_path("readers"); - let mut bring_up = |c: AccountCfg, _: Option| async move { - Ok(Some((client(&c.name).await, pending_loop()))) as BringUp - }; + 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; @@ -1081,9 +1145,9 @@ mod tests { let reg = registry().await; let (up, _rx) = unbounded_channel(); let calls = std::cell::RefCell::new(Vec::new()); - let bring_up = |_: AccountCfg, _: Option| { + let bring_up = |_: AccountCfg| { calls.borrow_mut().push(t0.elapsed().as_secs() / 60); - async { Err(anyhow!("homeserver down")) as BringUp } + async { Err(anyhow!("homeserver down")) as Restore } }; let _ = tokio::time::timeout( 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.