diff --git a/crates/maestro-cli/src/http.rs b/crates/maestro-cli/src/http.rs index 51cc355..5ad291b 100644 --- a/crates/maestro-cli/src/http.rs +++ b/crates/maestro-cli/src/http.rs @@ -60,7 +60,6 @@ impl HttpError { pub fn is_unreachable(&self) -> bool { matches!(self, HttpError::Reqwest(e) if e.is_connect()) } - } /// HTTP client bound to a single registry. Persists the auth cookie back to diff --git a/crates/maestro-daemon/src/broker.rs b/crates/maestro-daemon/src/broker.rs index 98e63a1..254ecbf 100644 --- a/crates/maestro-daemon/src/broker.rs +++ b/crates/maestro-daemon/src/broker.rs @@ -190,7 +190,9 @@ impl BrokerHandle { } match tokio::time::timeout(RESOLVE_TIMEOUT, rx).await { Ok(Ok(RpcResult::Ok { value })) => Ok(value), - Ok(Ok(RpcResult::Err { error })) => Err(BrokerError::Registry(error.code, error.message)), + Ok(Ok(RpcResult::Err { error })) => { + Err(BrokerError::Registry(error.code, error.message)) + } Ok(Err(_)) => Err(BrokerError::ReplyClosed), Err(_) => Err(BrokerError::Timeout(RESOLVE_TIMEOUT)), } @@ -222,7 +224,9 @@ impl BrokerHandle { } match tokio::time::timeout(RESOLVE_TIMEOUT, rx).await { Ok(Ok(RpcResult::Ok { .. })) => Ok(()), - Ok(Ok(RpcResult::Err { error })) => Err(BrokerError::Registry(error.code, error.message)), + Ok(Ok(RpcResult::Err { error })) => { + Err(BrokerError::Registry(error.code, error.message)) + } Ok(Err(_)) => Err(BrokerError::ReplyClosed), Err(_) => Err(BrokerError::Timeout(RESOLVE_TIMEOUT)), } @@ -391,7 +395,12 @@ mod tests { created_at: chrono::Utc::now(), }; handle - .complete_forward_create(reply_to, RpcResult::Ok { value: session.clone() }) + .complete_forward_create( + reply_to, + RpcResult::Ok { + value: session.clone(), + }, + ) .await; let got = task.await.unwrap().expect("create ok"); @@ -423,7 +432,8 @@ mod tests { handle.set_connected(true); let h = handle.clone(); - let task = tokio::spawn(async move { h.forward_kill("konishi".into(), Uuid::new_v4()).await }); + let task = + tokio::spawn(async move { h.forward_kill("konishi".into(), Uuid::new_v4()).await }); let reply_to = match out_rx.recv().await.expect("ForwardKill went out") { DaemonOutbound::ForwardKill { reply_to, .. } => reply_to, diff --git a/crates/maestro-e2e/tests/e2e.rs b/crates/maestro-e2e/tests/e2e.rs index 7dcd21d..2f66bd5 100644 --- a/crates/maestro-e2e/tests/e2e.rs +++ b/crates/maestro-e2e/tests/e2e.rs @@ -192,51 +192,66 @@ async fn home_daemon_brokers_cross_daemon_create_and_kill() { assert_eq!(s.daemon_id, "beta", "session should be owned by beta"); // The registry catalog (and thus every client) sees it on beta. - wait_for_async("beta session visible in catalog", Duration::from_secs(5), || { - let client = registry_client.clone(); - let want = s.id; - async move { - client - .list_sessions() - .await - .map(|ss| ss.iter().any(|x| x.id == want && x.daemon_id == "beta")) - .unwrap_or(false) - } - }) + wait_for_async( + "beta session visible in catalog", + Duration::from_secs(5), + || { + let client = registry_client.clone(); + let want = s.id; + async move { + client + .list_sessions() + .await + .map(|ss| ss.iter().any(|x| x.id == want && x.daemon_id == "beta")) + .unwrap_or(false) + } + }, + ) .await; // alpha's mirror must reflect beta's session before it can broker the kill. - wait_for_async("alpha mirrors beta's session", Duration::from_secs(5), || { - let base = alpha_base.clone(); - let want = s.id; - async move { - let resp = match reqwest::get(format!("{base}/api/sessions")).await { - Ok(r) => r, - Err(_) => return false, - }; - let list: Vec = match resp.json().await { - Ok(v) => v, - Err(_) => return false, - }; - list.iter().any(|x| x.id == want) - } - }) + wait_for_async( + "alpha mirrors beta's session", + Duration::from_secs(5), + || { + let base = alpha_base.clone(); + let want = s.id; + async move { + let resp = match reqwest::get(format!("{base}/api/sessions")).await { + Ok(r) => r, + Err(_) => return false, + }; + let list: Vec = match resp.json().await { + Ok(v) => v, + Err(_) => return false, + }; + list.iter().any(|x| x.id == want) + } + }, + ) .await; // Kill the beta session via alpha — brokered to beta. - alpha.kill_session(s.id).await.expect("alpha brokers kill to beta"); + alpha + .kill_session(s.id) + .await + .expect("alpha brokers kill to beta"); - wait_for_async("beta session gone from catalog", Duration::from_secs(5), || { - let client = registry_client.clone(); - let gone = s.id; - async move { - client - .list_sessions() - .await - .map(|ss| !ss.iter().any(|x| x.id == gone)) - .unwrap_or(false) - } - }) + wait_for_async( + "beta session gone from catalog", + Duration::from_secs(5), + || { + let client = registry_client.clone(); + let gone = s.id; + async move { + client + .list_sessions() + .await + .map(|ss| !ss.iter().any(|x| x.id == gone)) + .unwrap_or(false) + } + }, + ) .await; }