From 2c97b76b507c77c04995cd901713d9c908ed05a9 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:30:56 +0100 Subject: [PATCH 1/4] feat(pairing): wait reactively for bounded status transitions agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: openai-codex/gpt-6.1-sol --- CHANGELOG.md | 8 ++ docs/runtime-peer-api.md | 23 ++++- src/daemon/pairing.rs | 121 ++++++++++++++++++++------- src/daemon/runtime_control.rs | 100 +++++++++++++++++++++- src/runtime_peers.rs | 3 + src/runtime_peers/pairing.rs | 14 +++- src/runtime_peers/pairing/confirm.rs | 1 + 7 files changed, 232 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 37ba459..2df4951 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,14 @@ Deprecated, Fixed or Security. ## [Unreleased] +### Added + +- **Provisional:** restricted pairing `status` accepts optional + `wait: {after_revision: }` for notification-driven waits bounded by the + attempt deadline or 30 seconds. Advanced revisions and terminal states return + immediately; consumers reissue after an unchanged response. Joining-daemon + `pairing_remote` forwards waits with a 45-second total response deadline. + ### Fixed - **Provisional:** restricted pairing terminal replies use an independent diff --git a/docs/runtime-peer-api.md b/docs/runtime-peer-api.md index c688145..4fe43cb 100644 --- a/docs/runtime-peer-api.md +++ b/docs/runtime-peer-api.md @@ -82,9 +82,12 @@ Only a reply bound to the requested attempt and this daemon's NodeID is returned Each call makes one restricted connection using the daemon's current endpoint, with at most 32 outbound operations and one 30-second deadline covering connect, -request, and completion. On exchange reconnect, a small status frame first -authenticates the bound attempt on the same stream before the larger send/receive -frame. Pre-auth present/status frames remain capped at 4 KiB; replies are bounded +request, and completion for nonwaiting calls. A `status` request's optional `wait` +is forwarded unchanged; waiting calls have a 45-second total deadline to leave +time for the gateway's bounded wait and response. On exchange reconnect, a small +status frame first authenticates the bound attempt on the same stream before the +larger send/receive frame. Pre-auth present/status frames remain capped at 4 KiB; +replies are bounded to 1 MiB before allocation. The stream is half-closed after the final request; the gateway releases its attempt lease before closing, so a subsequent call can reconnect safely. Serialize calls for the same @@ -110,6 +113,7 @@ maximum encoded exchange request length 266,240 bytes after authentication): ```json {"operation":"present","attempt_id":"","secret":""} {"operation":"status","attempt_id":""} +{"operation":"status","attempt_id":"","wait":{"after_revision":2}} ``` Replies are length-prefixed pairing reply objects, without the local-control @@ -122,6 +126,19 @@ are accepted only for the authenticated bound node, including retained terminal status. Unknown, wrong-node, wrong-token and malformed requests close with a generic reason and expose no attempt status. +Omitting `wait` preserves immediate status reads. With +`wait: {"after_revision": }`, status returns immediately if +`attempt_revision` is greater than `after_revision` or the attempt is terminal +(`confirmed`, `expired`, or `rejected`). Otherwise it waits for a revision change +or terminal transition, bounded by the earlier of the original attempt deadline +and 30 seconds. The bound returns current status even if its revision is +unchanged; reissue `status` with the returned revision to keep waiting. Waiting +uses change notifications, not polling: bind, confirm, reject, cancel, expiry, +and recovery (`exchange_lost`) are observable. Opaque exchange send/receive +activity does not wake it. A pending wait is not an idle framing read. Waiting +does not extend the permit deadline or grant access; the same bound-node checks +and connection limits, including the 600-second connection lifetime, still apply. + There are at most 32 accepted pairing connections and one active connection per attempt. Unknown identities need global credit before parsing or runtime work: one request per second, burst 10, refunded after successful validation. Previously diff --git a/src/daemon/pairing.rs b/src/daemon/pairing.rs index 0dc30ff..3b12f6b 100644 --- a/src/daemon/pairing.rs +++ b/src/daemon/pairing.rs @@ -243,9 +243,13 @@ fn decode_request(bytes: &[u8]) -> Result { digest: Option, payload: Option, ack: Option, + wait: Option, } let frame: Frame = serde_json::from_slice(bytes).map_err(|_| refused())?; - let Frame { operation, attempt_id, secret, sequence, digest, payload, ack } = frame; + let Frame { operation, attempt_id, secret, sequence, digest, payload, ack, wait } = frame; + if wait.is_some() && !matches!(operation, Operation::Status) { + return Err(refused()); + } match operation { Operation::Present if sequence.is_none() && digest.is_none() && payload.is_none() && ack.is_none() => @@ -256,7 +260,7 @@ fn decode_request(bytes: &[u8]) -> Result { if secret.is_none() && sequence.is_none() && digest.is_none() && payload.is_none() && ack.is_none() => { - Ok(NetworkRequest::Status { attempt_id }) + Ok(NetworkRequest::Status { attempt_id, wait }) } Operation::Send if secret.is_none() && ack.is_none() => Ok(NetworkRequest::Send { attempt_id, @@ -390,7 +394,7 @@ pub(super) async fn remote( })?; let attempt = match &request { NetworkRequest::Present { attempt_id, .. } - | NetworkRequest::Status { attempt_id } + | NetworkRequest::Status { attempt_id, .. } | NetworkRequest::Send { attempt_id, .. } | NetworkRequest::Receive { attempt_id, .. } => attempt_id, }; @@ -403,7 +407,8 @@ pub(super) async fn remote( .into()); } let endpoint = state.current_endpoint(); - let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + let waiting = matches!(&request, NetworkRequest::Status { wait: Some(_), .. }); + let deadline = tokio::time::Instant::now() + Duration::from_secs(if waiting { 45 } else { 30 }); let connection = tokio::time::timeout_at( deadline, endpoint.connect( @@ -423,9 +428,7 @@ pub(super) async fn remote( let (mut send, mut recv) = connection.open_bi().await?; if matches!(&request, NetworkRequest::Send { .. } | NetworkRequest::Receive { .. }) { // Reconnect authenticates a small frame before a large exchange. - let prime = serde_json::to_vec(&NetworkRequest::Status { - attempt_id: attempt.clone(), - })?; + let prime = serde_json::to_vec(&NetworkRequest::Status { attempt_id: attempt.clone(), wait: None })?; send.write_all(&(prime.len() as u32).to_be_bytes()).await?; send.write_all(&prime).await?; let prime = read_remote_reply(&mut recv, attempt, endpoint.id()).await?; @@ -599,7 +602,7 @@ async fn serve_stream( } let attempt = match &request { NetworkRequest::Present { attempt_id, .. } - | NetworkRequest::Status { attempt_id } + | NetworkRequest::Status { attempt_id, .. } | NetworkRequest::Send { attempt_id, .. } | NetworkRequest::Receive { attempt_id, .. } => attempt_id, }; @@ -611,7 +614,12 @@ async fn serve_stream( lease = Some(state.pairing_gate.lease(attempt).ok_or_else(refused)?); } let exchange = matches!(&request, NetworkRequest::Send { .. } | NetworkRequest::Receive { .. }); - let reply = if !exchange && validated.is_some() { + let reply = if matches!(&request, NetworkRequest::Status { wait: Some(_), .. }) { + tokio::select! { + result = state.pairing_status_wait(connection.remote_id(), &request) => result?, + _ = connection.closed() => return Ok(()), + } + } else if !exchange && validated.is_some() { validated.expect("checked read-only reply") } else { // A timeout or disconnect must not cancel durable publication and @@ -794,9 +802,7 @@ mod tests { // Start a full refill interval at the barrier, independent of how long // endpoint startup took on the runner. state.pairing_gate.rates.lock().global.last = now; - let response = rpc(&mut send, &mut recv, NetworkRequest::Status { - attempt_id: "unknown".into(), - }); + let response = rpc(&mut send, &mut recv, NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }); let refused = tokio::time::timeout(Duration::from_millis(500), response).await; release.send(())?; blocker.await?; @@ -893,9 +899,7 @@ mod tests { let client = client().await?; let connection = client.connect(gateway.addr(), PAIR_ALPN).await?; let (mut send, _recv) = connection.open_bi().await?; - let request = serde_json::to_vec(&NetworkRequest::Status { - attempt_id: "unknown".into(), - })?; + let request = serde_json::to_vec(&NetworkRequest::Status { attempt_id: "unknown".into(), wait: None })?; send.write_all(&(request.len() as u32).to_be_bytes()).await?; send.write_all(&request).await?; let closed = tokio::time::timeout(Duration::from_secs(4), connection.closed()).await; @@ -933,7 +937,7 @@ mod tests { // the dedicated read semaphore: after each caller future is dropped, // its extra owners are the permits retained inside blocking workers. let owners_before = Arc::strong_count(&state.pairing_read_slots); - let request = NetworkRequest::Status { attempt_id: "unknown".into() }; + let request = NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }; let mut observations = Vec::new(); // The first pool's worth start real blocking workers. Later requests // time out waiting asynchronously for worker capacity, without spawning. @@ -1098,9 +1102,7 @@ mod tests { let same = rpc( &mut send, &mut recv, - NetworkRequest::Status { - attempt_id: attempt_id.clone(), - }, + NetworkRequest::Status { attempt_id: attempt_id.clone(), wait: None }, ) .await?; assert_eq!(same.status.unwrap().bound_node_id, Some(client.id())); @@ -1128,7 +1130,7 @@ mod tests { rpc( &mut other_send, &mut other_recv, - NetworkRequest::Status { attempt_id } + NetworkRequest::Status { attempt_id, wait: None } ) .await .is_err() @@ -1193,6 +1195,7 @@ mod tests { let (mut send, mut recv) = connection.open_bi().await?; let reply = rpc(&mut send, &mut recv, NetworkRequest::Status { attempt_id: attempt_id.clone(), + wait: None, }).await?; assert!(reply.error.is_none(), "{:?}", reply.error); let status = reply.status.context("missing retained terminal status")?; @@ -1210,6 +1213,61 @@ mod tests { Ok(()) } + #[tokio::test] + async fn status_wait_bridge_caps_expires_and_reads_restart_recovery() -> Result<()> { + use crate::runtime_peers::pairing::{Policy, Request, StatusWait}; + let owner_dir = tempfile::tempdir()?; + let owner_home = FabricHome::new(owner_dir.path()); + let mut owner = gateway(owner_home.clone()).await?; + let joining_dir = tempfile::tempdir()?; + let joining = gateway(FabricHome::new(joining_dir.path())).await?; + for mode in ["cap", "expiry", "restart"] { + let deadline = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)?.as_secs() + + if mode == "expiry" { 2 } else { 600 }; + let issued = owner.state().pairing_request(Request::Issue { + owner: "test".into(), creation_id: mode.into(), deadline, + policy: Policy { grants: vec![], name: None, + direction: crate::runtime_peers::Direction::Normal }, + }).await; + assert!(issued.error.is_none(), "{:?}", issued.error); + let attempt_id = issued.status.unwrap().attempt_id; + let pending = remote(&joining.state(), owner.id(), Some(owner.addr()), + NetworkRequest::Present { attempt_id: attempt_id.clone(), + secret: issued.secret.unwrap() }).await; + assert!(pending.error.is_none(), "{:?}", pending.error); + let pending = pending.status.unwrap(); + if mode == "restart" { + owner.shutdown().await?; + owner = gateway(owner_home.clone()).await?; + } + let started = tokio::time::Instant::now(); + let reply = remote(&joining.state(), owner.id(), Some(owner.addr()), + NetworkRequest::Status { attempt_id, + wait: Some(StatusWait { after_revision: pending.attempt_revision }) }).await; + assert!(reply.error.is_none(), "{mode}: {:?}", reply.error); + let status = reply.status.unwrap(); + match mode { + "cap" => { + assert_eq!(status, pending); + assert!(started.elapsed() >= Duration::from_secs(30)); + assert!(started.elapsed() < Duration::from_secs(40)); + } + "expiry" => { + assert_eq!(status.state, State::Expired); + assert!(status.attempt_revision > pending.attempt_revision); + } + _ => { + assert_eq!(status.state, State::Rejected); + assert_eq!(status.terminal_reason.as_deref(), Some("exchange_lost")); + assert!(started.elapsed() < Duration::from_secs(3)); + } + } + } + joining.shutdown().await?; + owner.shutdown().await?; + Ok(()) + } #[tokio::test] async fn disabled_pairing_and_oversized_frames_create_no_pending_record() -> Result<()> { use crate::runtime_peers::pairing::PAIR_ALPN; @@ -1239,7 +1297,9 @@ mod tests { attempt_id: "attempt".into(), secret: Secret::new("secret"), }, - NetworkRequest::Status { attempt_id: "attempt".into() }, + NetworkRequest::Status { attempt_id: "attempt".into(), wait: None }, + NetworkRequest::Status { attempt_id: "attempt".into(), + wait: Some(crate::runtime_peers::pairing::StatusWait { after_revision: 1 }) }, NetworkRequest::Send { attempt_id: "attempt".into(), sequence: 1, @@ -1256,6 +1316,9 @@ mod tests { br#"{"operation":"status","attempt_id":"a","payload":[1]}"#, br#"{"operation":"status","attempt_id":"a","unknown":1}"#, br#"{"operation":"status","attempt_id":"a","attempt_id":"b"}"#, + br#"{"operation":"receive","attempt_id":"a","wait":{"after_revision":1}}"#, + br#"{"operation":"status","attempt_id":"a","wait":{"after_revision":-1}}"#, + br#"{"operation":"status","attempt_id":"a","wait":{"after_revision":1,"unknown":1}}"#, ] { assert!(decode_request(bytes).is_err()); } @@ -1628,7 +1691,7 @@ mod tests { assert_eq!(&bytes, b"proof"); let unrelated = client.connect(gateway.addr(), b"not-granted").await?; tokio::time::timeout(Duration::from_secs(3), unrelated.closed()).await?; - let own_status = rpc(&mut send, &mut recv, NetworkRequest::Status { attempt_id }).await?; + let own_status = rpc(&mut send, &mut recv, NetworkRequest::Status { attempt_id, wait: None }).await?; assert!(own_status.access.is_empty()); let own_status = own_status.status.unwrap(); assert_eq!(own_status.state, State::Confirmed); @@ -1882,9 +1945,7 @@ mod tests { let joining_home = FabricHome::new(joining_dir.path()); let disabled = FabricNode::start(joining_home.clone()).await?; control_ready(&joining_home).await?; - let request = NetworkRequest::Status { - attempt_id: "unknown".into(), - }; + let request = NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }; let reply = remote_request(&joining_home, &owner, request).await?; assert_eq!(reply.error.unwrap().code, "runtime_peers_disabled"); disabled.shutdown().await?; @@ -1910,9 +1971,7 @@ mod tests { ControlRequest::PairingRemote { gateway: spoofed.id, addr: Some(spoofed), - request: NetworkRequest::Status { - attempt_id: "unknown".into(), - }, + request: NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }, }, ) .await? @@ -1954,7 +2013,7 @@ mod tests { anyhow::ensure!(length <= MAX_REQUEST, "oversized request"); let mut bytes = vec![0; length]; recv.read_exact(&mut bytes).await?; - let NetworkRequest::Status { attempt_id } = serde_json::from_slice(&bytes)? else { + let NetworkRequest::Status { attempt_id, .. } = serde_json::from_slice(&bytes)? else { bail!("expected status request"); }; let reply = Reply { @@ -2008,9 +2067,7 @@ mod tests { ControlRequest::PairingRemote { gateway: malicious.id(), addr: Some(malicious.addr()), - request: NetworkRequest::Status { - attempt_id: "bound-attempt".into(), - }, + request: NetworkRequest::Status { attempt_id: "bound-attempt".into(), wait: None }, }, ) .await? diff --git a/src/daemon/runtime_control.rs b/src/daemon/runtime_control.rs index d6e9e94..c337843 100644 --- a/src/daemon/runtime_control.rs +++ b/src/daemon/runtime_control.rs @@ -480,6 +480,41 @@ impl DaemonState { }).await } + pub(super) async fn pairing_status_wait( + &self, + node: EndpointId, + request: &runtime_peers::pairing::NetworkRequest, + ) -> Result { + use runtime_peers::pairing::{NetworkRequest, State}; + let NetworkRequest::Status { wait: Some(wait), .. } = request else { + anyhow::bail!("status wait requires a revision"); + }; + // Subscribe before reading: commits between the read and changed() must + // remain observable, including commits made by off-thread workers. + let mut changes = self.read_pairing_result(|runtime| { + Ok(runtime.pairing_changes.subscribe()) + }).await?; + let cap = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + changes.borrow_and_update(); + let reply = self.pairing_network_read(node, request).await? + .context("missing authenticated status")?; + let Some(status) = reply.status.as_ref() else { return Ok(reply); }; + if status.attempt_revision > wait.after_revision || status.state != State::Pending { + return Ok(reply); + } + let wall = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)?.as_secs(); + let deadline = cap.min(tokio::time::Instant::now() + + Duration::from_secs(status.deadline.saturating_sub(wall))); + if tokio::time::timeout_at(deadline, changes.changed()).await.is_err() { + // Read again to project expiry even before the expiry worker commits. + return self.pairing_network_read(node, request).await? + .context("missing authenticated status"); + } + } + } + async fn confirm_pairing( &self, confirmation: runtime_peers::pairing::ConfirmRequest, @@ -1522,6 +1557,69 @@ mod tests { .await; } + #[tokio::test] + async fn status_wait_observes_confirm_reject_cancel_and_immediate_revisions() -> Result<()> { + use runtime_peers::pairing::{NetworkRequest, RejectRequest, Request as PairRequest, State, StatusWait}; + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let node = FabricNode::start_with_daemon_options(FabricHome::new(dir.path()), options()).await?; + let state = node.state(); + let peer = iroh::SecretKey::from_bytes(&[86; 32]).public(); + for action in ["confirm", "reject", "cancel"] { + let confirmation = pending_confirmation(&state, action, peer, None).await?; + let request = NetworkRequest::Status { + attempt_id: confirmation.attempt_id.clone(), + wait: Some(StatusWait { after_revision: confirmation.expected_attempt_revision - 1 }), + }; + let advanced = tokio::time::timeout(Duration::from_secs(1), + state.pairing_status_wait(peer, &request)).await??; + assert_eq!(advanced.status.unwrap().state, State::Pending); + let request = NetworkRequest::Status { + attempt_id: confirmation.attempt_id.clone(), + wait: Some(StatusWait { after_revision: confirmation.expected_attempt_revision }), + }; + let waiting = state.pairing_status_wait(peer, &request); + tokio::pin!(waiting); + // Establish a real subscription before committing, without sleeps. + tokio::select! { + result = &mut waiting => panic!("wait completed before transition: {result:?}"), + _ = async { + loop { + if state.runtime_peers.as_ref().unwrap().lock().unwrap() + .pairing_changes.receiver_count() > 0 { break; } + tokio::task::yield_now().await; + } + } => {} + } + let transition = match action { + "confirm" => PairRequest::Confirm { confirmation }, + "reject" => PairRequest::Reject { rejection: RejectRequest { + owner: confirmation.owner, + attempt_id: confirmation.attempt_id, + expected_node_id: peer, + expected_attempt_revision: confirmation.expected_attempt_revision, + request_id: "reject-wait".into(), + }}, + _ => PairRequest::Cancel { owner: confirmation.owner, creation_id: action.into() }, + }; + let committed = state.pairing_request(transition).await; + assert!(committed.error.is_none(), "{:?}", committed.error); + let terminal = committed.status.unwrap(); + let observed = tokio::time::timeout(Duration::from_secs(1), &mut waiting).await??; + assert_eq!(observed.status.unwrap(), terminal); + // Terminal always returns, even if the client's revision is ahead. + let request = NetworkRequest::Status { + attempt_id: terminal.attempt_id, + wait: Some(StatusWait { after_revision: u64::MAX }), + }; + let observed = tokio::time::timeout(Duration::from_secs(1), + state.pairing_status_wait(peer, &request)).await??; + assert_eq!(observed.status.unwrap().state, terminal.state); + } + node.shutdown().await?; + Ok(()) + } + #[tokio::test] async fn concurrent_pairing_confirm_confirm_installs_once() -> Result<()> { use runtime_peers::pairing::Request as PairRequest; @@ -1754,7 +1852,7 @@ mod tests { for (id, expected_error) in [(attempt_id.clone(), false), ("unknown".into(), true)] { let reply = tokio::time::timeout( Duration::from_secs(1), - state.pairing_network_request(peer, NetworkRequest::Status { attempt_id: id }), + state.pairing_network_request(peer, NetworkRequest::Status { attempt_id: id, wait: None }), ).await?; assert_eq!(reply.error.is_some(), expected_error); } diff --git a/src/runtime_peers.rs b/src/runtime_peers.rs index 5100b2e..7392664 100644 --- a/src/runtime_peers.rs +++ b/src/runtime_peers.rs @@ -425,6 +425,7 @@ pub(crate) struct RuntimePeers { pairing_clock: (u64, std::time::Instant), expiry_clock: std::sync::Arc, exchanges: BTreeMap, + pub(crate) pairing_changes: tokio::sync::watch::Sender<()>, #[cfg(test)] fail_directory_sync: bool, #[cfg(test)] @@ -494,6 +495,7 @@ impl RuntimePeers { pairing_clock, expiry_clock: expiry_clock.unwrap_or_else(|| std::sync::Arc::new(ExpiryClock::new(pairing_clock))), exchanges: BTreeMap::new(), + pairing_changes: tokio::sync::watch::channel(()).0, #[cfg(test)] fail_directory_sync: false, #[cfg(test)] @@ -687,6 +689,7 @@ impl RuntimePeers { pairing_clock: self.pairing_clock, expiry_clock: self.expiry_clock.clone(), exchanges: BTreeMap::new(), + pairing_changes: tokio::sync::watch::channel(()).0, #[cfg(test)] fail_directory_sync: false, #[cfg(test)] diff --git a/src/runtime_peers/pairing.rs b/src/runtime_peers/pairing.rs index 36b7d65..53786de 100644 --- a/src/runtime_peers/pairing.rs +++ b/src/runtime_peers/pairing.rs @@ -23,6 +23,8 @@ pub enum NetworkRequest { }, Status { attempt_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + wait: Option, }, Send { attempt_id: String, @@ -36,6 +38,12 @@ pub enum NetworkRequest { }, } +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StatusWait { + pub after_revision: u64, +} + const MAX_PERMITS: usize = 64; const MAX_ATTEMPTS: usize = 1024; const MAX_OWNER_ATTEMPTS: usize = 128; @@ -443,7 +451,7 @@ impl RuntimePeers { let refused = || invalid("pairing request refused"); let (attempt_id, secret) = match request { NetworkRequest::Present { attempt_id, secret } => (attempt_id, Some(secret)), - NetworkRequest::Status { attempt_id } + NetworkRequest::Status { attempt_id, .. } | NetworkRequest::Send { attempt_id, .. } | NetworkRequest::Receive { attempt_id, .. } => (attempt_id, None), }; @@ -797,6 +805,7 @@ impl RuntimePeers { return Err(error); } self.stored = next; + self.pairing_changes.send_replace(()); } self.prune_exchanges(); Ok(()) @@ -1259,6 +1268,7 @@ mod tests { fn restricted_status(status: &Status) -> NetworkRequest { NetworkRequest::Status { attempt_id: status.attempt_id.clone(), + wait: None, } } @@ -2290,7 +2300,7 @@ mod tests { store.fail_directory_sync = true; assert!(store.pairing_network_read_at(node(1), &request, 101).unwrap().is_none()); for request in [ - NetworkRequest::Status { attempt_id: "unknown".into() }, + NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }, NetworkRequest::Present { attempt_id: "unknown".into(), secret: Secret::new("wrong") }, restricted_status(&status), presentation(&status, &Secret::new("wrong")), diff --git a/src/runtime_peers/pairing/confirm.rs b/src/runtime_peers/pairing/confirm.rs index 2324686..f885146 100644 --- a/src/runtime_peers/pairing/confirm.rs +++ b/src/runtime_peers/pairing/confirm.rs @@ -805,6 +805,7 @@ mod tests { node(1), NetworkRequest::Status { attempt_id: request.attempt_id.clone(), + wait: None, }, t + 1, ) From 98e578f126159c5a73610af5ccd464d652837827 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 20:25:55 +0100 Subject: [PATCH 2/4] fix(pairing): notify waits at confirmation and recovery commits agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: openai-codex/gpt-6.1-sol --- src/runtime_peers.rs | 1 + src/runtime_peers/pairing/confirm.rs | 1 + 2 files changed, 2 insertions(+) diff --git a/src/runtime_peers.rs b/src/runtime_peers.rs index 7392664..82001e0 100644 --- a/src/runtime_peers.rs +++ b/src/runtime_peers.rs @@ -574,6 +574,7 @@ impl RuntimePeers { let book = store.compose(&store.stored)?.0; if changed { store.persist(&store.stored)?; + store.pairing_changes.send_replace(()); } Ok(book) })(); diff --git a/src/runtime_peers/pairing/confirm.rs b/src/runtime_peers/pairing/confirm.rs index f885146..6d7319b 100644 --- a/src/runtime_peers/pairing/confirm.rs +++ b/src/runtime_peers/pairing/confirm.rs @@ -302,6 +302,7 @@ impl RuntimePeers { panic!("injected confirmation worker failure after persist, before stored update"); } self.stored = next; + self.pairing_changes.send_replace(()); self.prune_exchanges(); self.pairing_attempt_reply_at(&request.owner, &request.attempt_id, timestamp) } From 41118e4814cd11d35c68a6d54c890680b93ebebe Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:38:04 +0100 Subject: [PATCH 3/4] style(pairing): format only status wait additions agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: openai-codex/gpt-6.1-sol --- src/daemon/pairing.rs | 123 +++++++++++++++++++++++++++------- src/daemon/runtime_control.rs | 89 ++++++++++++++++-------- 2 files changed, 160 insertions(+), 52 deletions(-) diff --git a/src/daemon/pairing.rs b/src/daemon/pairing.rs index 1cda5a2..f5ea7d7 100644 --- a/src/daemon/pairing.rs +++ b/src/daemon/pairing.rs @@ -246,7 +246,16 @@ fn decode_request(bytes: &[u8]) -> Result { wait: Option, } let frame: Frame = serde_json::from_slice(bytes).map_err(|_| refused())?; - let Frame { operation, attempt_id, secret, sequence, digest, payload, ack, wait } = frame; + let Frame { + operation, + attempt_id, + secret, + sequence, + digest, + payload, + ack, + wait, + } = frame; if wait.is_some() && !matches!(operation, Operation::Status) { return Err(refused()); } @@ -408,7 +417,8 @@ pub(super) async fn remote( } let endpoint = state.current_endpoint(); let waiting = matches!(&request, NetworkRequest::Status { wait: Some(_), .. }); - let deadline = tokio::time::Instant::now() + Duration::from_secs(if waiting { 45 } else { 30 }); + let deadline = + tokio::time::Instant::now() + Duration::from_secs(if waiting { 45 } else { 30 }); let connection = tokio::time::timeout_at( deadline, endpoint.connect( @@ -428,7 +438,10 @@ pub(super) async fn remote( let (mut send, mut recv) = connection.open_bi().await?; if matches!(&request, NetworkRequest::Send { .. } | NetworkRequest::Receive { .. }) { // Reconnect authenticates a small frame before a large exchange. - let prime = serde_json::to_vec(&NetworkRequest::Status { attempt_id: attempt.clone(), wait: None })?; + let prime = serde_json::to_vec(&NetworkRequest::Status { + attempt_id: attempt.clone(), + wait: None, + })?; send.write_all(&(prime.len() as u32).to_be_bytes()).await?; send.write_all(&prime).await?; let prime = read_remote_reply(&mut recv, attempt, endpoint.id()).await?; @@ -802,7 +815,14 @@ mod tests { // Start a full refill interval at the barrier, independent of how long // endpoint startup took on the runner. state.pairing_gate.rates.lock().global.last = now; - let response = rpc(&mut send, &mut recv, NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }); + let response = rpc( + &mut send, + &mut recv, + NetworkRequest::Status { + attempt_id: "unknown".into(), + wait: None, + }, + ); let refused = tokio::time::timeout(Duration::from_millis(500), response).await; release.send(())?; blocker.await?; @@ -937,7 +957,10 @@ mod tests { // the dedicated read semaphore: after each caller future is dropped, // its extra owners are the permits retained inside blocking workers. let owners_before = Arc::strong_count(&state.pairing_read_slots); - let request = NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }; + let request = NetworkRequest::Status { + attempt_id: "unknown".into(), + wait: None, + }; let mut observations = Vec::new(); // The first pool's worth start real blocking workers. Later requests // time out waiting asynchronously for worker capacity, without spawning. @@ -1102,7 +1125,10 @@ mod tests { let same = rpc( &mut send, &mut recv, - NetworkRequest::Status { attempt_id: attempt_id.clone(), wait: None }, + NetworkRequest::Status { + attempt_id: attempt_id.clone(), + wait: None, + }, ) .await?; assert_eq!(same.status.unwrap().bound_node_id, Some(client.id())); @@ -1130,7 +1156,10 @@ mod tests { rpc( &mut other_send, &mut other_recv, - NetworkRequest::Status { attempt_id, wait: None } + NetworkRequest::Status { + attempt_id, + wait: None + } ) .await .is_err() @@ -1255,18 +1284,34 @@ mod tests { let joining = gateway(FabricHome::new(joining_dir.path())).await?; for mode in ["cap", "expiry", "restart"] { let deadline = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH)?.as_secs() + .duration_since(std::time::UNIX_EPOCH)? + .as_secs() + if mode == "expiry" { 2 } else { 600 }; - let issued = owner.state().pairing_request(Request::Issue { - owner: "test".into(), creation_id: mode.into(), deadline, - policy: Policy { grants: vec![], name: None, - direction: crate::runtime_peers::Direction::Normal }, - }).await; + let issued = owner + .state() + .pairing_request(Request::Issue { + owner: "test".into(), + creation_id: mode.into(), + deadline, + policy: Policy { + grants: vec![], + name: None, + direction: crate::runtime_peers::Direction::Normal, + }, + }) + .await; assert!(issued.error.is_none(), "{:?}", issued.error); let attempt_id = issued.status.unwrap().attempt_id; - let pending = remote(&joining.state(), owner.id(), Some(owner.addr()), - NetworkRequest::Present { attempt_id: attempt_id.clone(), - secret: issued.secret.unwrap() }).await; + let pending = remote( + &joining.state(), + owner.id(), + Some(owner.addr()), + NetworkRequest::Present { + attempt_id: attempt_id.clone(), + secret: issued.secret.unwrap(), + }, + ) + .await; assert!(pending.error.is_none(), "{:?}", pending.error); let pending = pending.status.unwrap(); if mode == "restart" { @@ -1274,9 +1319,18 @@ mod tests { owner = gateway(owner_home.clone()).await?; } let started = tokio::time::Instant::now(); - let reply = remote(&joining.state(), owner.id(), Some(owner.addr()), - NetworkRequest::Status { attempt_id, - wait: Some(StatusWait { after_revision: pending.attempt_revision }) }).await; + let reply = remote( + &joining.state(), + owner.id(), + Some(owner.addr()), + NetworkRequest::Status { + attempt_id, + wait: Some(StatusWait { + after_revision: pending.attempt_revision, + }), + }, + ) + .await; assert!(reply.error.is_none(), "{mode}: {:?}", reply.error); let status = reply.status.unwrap(); match mode { @@ -1329,9 +1383,14 @@ mod tests { attempt_id: "attempt".into(), secret: Secret::new("secret"), }, - NetworkRequest::Status { attempt_id: "attempt".into(), wait: None }, - NetworkRequest::Status { attempt_id: "attempt".into(), - wait: Some(crate::runtime_peers::pairing::StatusWait { after_revision: 1 }) }, + NetworkRequest::Status { + attempt_id: "attempt".into(), + wait: None, + }, + NetworkRequest::Status { + attempt_id: "attempt".into(), + wait: Some(crate::runtime_peers::pairing::StatusWait { after_revision: 1 }), + }, NetworkRequest::Send { attempt_id: "attempt".into(), sequence: 1, @@ -1723,7 +1782,15 @@ mod tests { assert_eq!(&bytes, b"proof"); let unrelated = client.connect(gateway.addr(), b"not-granted").await?; tokio::time::timeout(Duration::from_secs(3), unrelated.closed()).await?; - let own_status = rpc(&mut send, &mut recv, NetworkRequest::Status { attempt_id, wait: None }).await?; + let own_status = rpc( + &mut send, + &mut recv, + NetworkRequest::Status { + attempt_id, + wait: None, + }, + ) + .await?; assert!(own_status.access.is_empty()); let own_status = own_status.status.unwrap(); assert_eq!(own_status.state, State::Confirmed); @@ -1977,7 +2044,10 @@ mod tests { let joining_home = FabricHome::new(joining_dir.path()); let disabled = FabricNode::start(joining_home.clone()).await?; control_ready(&joining_home).await?; - let request = NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }; + let request = NetworkRequest::Status { + attempt_id: "unknown".into(), + wait: None, + }; let reply = remote_request(&joining_home, &owner, request).await?; assert_eq!(reply.error.unwrap().code, "runtime_peers_disabled"); disabled.shutdown().await?; @@ -2003,7 +2073,10 @@ mod tests { ControlRequest::PairingRemote { gateway: spoofed.id, addr: Some(spoofed), - request: NetworkRequest::Status { attempt_id: "unknown".into(), wait: None }, + request: NetworkRequest::Status { + attempt_id: "unknown".into(), + wait: None, + }, }, ) .await? diff --git a/src/daemon/runtime_control.rs b/src/daemon/runtime_control.rs index c337843..7694e60 100644 --- a/src/daemon/runtime_control.rs +++ b/src/daemon/runtime_control.rs @@ -486,30 +486,45 @@ impl DaemonState { request: &runtime_peers::pairing::NetworkRequest, ) -> Result { use runtime_peers::pairing::{NetworkRequest, State}; - let NetworkRequest::Status { wait: Some(wait), .. } = request else { + let NetworkRequest::Status { + wait: Some(wait), .. + } = request + else { anyhow::bail!("status wait requires a revision"); }; // Subscribe before reading: commits between the read and changed() must // remain observable, including commits made by off-thread workers. - let mut changes = self.read_pairing_result(|runtime| { - Ok(runtime.pairing_changes.subscribe()) - }).await?; + let mut changes = self + .read_pairing_result(|runtime| Ok(runtime.pairing_changes.subscribe())) + .await?; let cap = tokio::time::Instant::now() + Duration::from_secs(30); loop { changes.borrow_and_update(); - let reply = self.pairing_network_read(node, request).await? + let reply = self + .pairing_network_read(node, request) + .await? .context("missing authenticated status")?; - let Some(status) = reply.status.as_ref() else { return Ok(reply); }; + let Some(status) = reply.status.as_ref() else { + return Ok(reply); + }; if status.attempt_revision > wait.after_revision || status.state != State::Pending { return Ok(reply); } let wall = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH)?.as_secs(); - let deadline = cap.min(tokio::time::Instant::now() - + Duration::from_secs(status.deadline.saturating_sub(wall))); - if tokio::time::timeout_at(deadline, changes.changed()).await.is_err() { + .duration_since(std::time::UNIX_EPOCH)? + .as_secs(); + let deadline = cap.min( + tokio::time::Instant::now() + + Duration::from_secs(status.deadline.saturating_sub(wall)), + ); + if tokio::time::timeout_at(deadline, changes.changed()) + .await + .is_err() + { // Read again to project expiry even before the expiry worker commits. - return self.pairing_network_read(node, request).await? + return self + .pairing_network_read(node, request) + .await? .context("missing authenticated status"); } } @@ -1559,24 +1574,34 @@ mod tests { #[tokio::test] async fn status_wait_observes_confirm_reject_cancel_and_immediate_revisions() -> Result<()> { - use runtime_peers::pairing::{NetworkRequest, RejectRequest, Request as PairRequest, State, StatusWait}; + use runtime_peers::pairing::{ + NetworkRequest, RejectRequest, Request as PairRequest, State, StatusWait, + }; let _serial = SERIAL.lock().await; let dir = tempfile::tempdir()?; - let node = FabricNode::start_with_daemon_options(FabricHome::new(dir.path()), options()).await?; + let node = + FabricNode::start_with_daemon_options(FabricHome::new(dir.path()), options()).await?; let state = node.state(); let peer = iroh::SecretKey::from_bytes(&[86; 32]).public(); for action in ["confirm", "reject", "cancel"] { let confirmation = pending_confirmation(&state, action, peer, None).await?; let request = NetworkRequest::Status { attempt_id: confirmation.attempt_id.clone(), - wait: Some(StatusWait { after_revision: confirmation.expected_attempt_revision - 1 }), + wait: Some(StatusWait { + after_revision: confirmation.expected_attempt_revision - 1, + }), }; - let advanced = tokio::time::timeout(Duration::from_secs(1), - state.pairing_status_wait(peer, &request)).await??; + let advanced = tokio::time::timeout( + Duration::from_secs(1), + state.pairing_status_wait(peer, &request), + ) + .await??; assert_eq!(advanced.status.unwrap().state, State::Pending); let request = NetworkRequest::Status { attempt_id: confirmation.attempt_id.clone(), - wait: Some(StatusWait { after_revision: confirmation.expected_attempt_revision }), + wait: Some(StatusWait { + after_revision: confirmation.expected_attempt_revision, + }), }; let waiting = state.pairing_status_wait(peer, &request); tokio::pin!(waiting); @@ -1593,14 +1618,19 @@ mod tests { } let transition = match action { "confirm" => PairRequest::Confirm { confirmation }, - "reject" => PairRequest::Reject { rejection: RejectRequest { + "reject" => PairRequest::Reject { + rejection: RejectRequest { + owner: confirmation.owner, + attempt_id: confirmation.attempt_id, + expected_node_id: peer, + expected_attempt_revision: confirmation.expected_attempt_revision, + request_id: "reject-wait".into(), + }, + }, + _ => PairRequest::Cancel { owner: confirmation.owner, - attempt_id: confirmation.attempt_id, - expected_node_id: peer, - expected_attempt_revision: confirmation.expected_attempt_revision, - request_id: "reject-wait".into(), - }}, - _ => PairRequest::Cancel { owner: confirmation.owner, creation_id: action.into() }, + creation_id: action.into(), + }, }; let committed = state.pairing_request(transition).await; assert!(committed.error.is_none(), "{:?}", committed.error); @@ -1610,10 +1640,15 @@ mod tests { // Terminal always returns, even if the client's revision is ahead. let request = NetworkRequest::Status { attempt_id: terminal.attempt_id, - wait: Some(StatusWait { after_revision: u64::MAX }), + wait: Some(StatusWait { + after_revision: u64::MAX, + }), }; - let observed = tokio::time::timeout(Duration::from_secs(1), - state.pairing_status_wait(peer, &request)).await??; + let observed = tokio::time::timeout( + Duration::from_secs(1), + state.pairing_status_wait(peer, &request), + ) + .await??; assert_eq!(observed.status.unwrap().state, terminal.state); } node.shutdown().await?; From 1a823c30ee510848f32ebd35527ce40640420512 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 22:08:52 +0100 Subject: [PATCH 4/4] Bound every status wait read and preserve reconciliation notifications agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: openai-codex/gpt-6.1-sol --- CHANGELOG.md | 4 + crates/fabric-config/src/error.rs | 1 + docs/runtime-peer-api.md | 23 +- src/daemon.rs | 4 + src/daemon/pairing.rs | 46 +++- src/daemon/runtime_control.rs | 390 ++++++++++++++++++++++-------- src/runtime_peers.rs | 2 + src/runtime_peers/pairing.rs | 4 + 8 files changed, 362 insertions(+), 112 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2df4951..5ebfba0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,10 @@ Deprecated, Fixed or Security. attempt deadline or 30 seconds. Advanced revisions and terminal states return immediately; consumers reissue after an unchanged response. Joining-daemon `pairing_remote` forwards waits with a 45-second total response deadline. + The same absolute bound covers runtime reads and capacity waits; a stalled + read returns the last authenticated snapshot. Live reconciliation preserves + notifications. Ahead-of-current pending revisions receive the typed + `status-wait-revision-ahead` refusal with the current revision. ### Fixed diff --git a/crates/fabric-config/src/error.rs b/crates/fabric-config/src/error.rs index 57d68a3..07d887f 100644 --- a/crates/fabric-config/src/error.rs +++ b/crates/fabric-config/src/error.rs @@ -117,6 +117,7 @@ error_codes! { RuntimeCommitIndeterminate => "runtime_commit_indeterminate", Fault; RuntimeReceiptCapacity => "runtime_receipt_capacity", Fault; PairingExchangeInactive => "pairing_exchange_inactive", Refusal; + StatusWaitRevisionAhead => "status-wait-revision-ahead", Refusal; RequestIdConflict => "request_id_conflict", Usage; CancellationTimeout => "cancellation_timeout", Fault; IdentityMismatch => "identity_mismatch", Usage; diff --git a/docs/runtime-peer-api.md b/docs/runtime-peer-api.md index 4fe43cb..8797f4a 100644 --- a/docs/runtime-peer-api.md +++ b/docs/runtime-peer-api.md @@ -127,15 +127,22 @@ status. Unknown, wrong-node, wrong-token and malformed requests close with a generic reason and expose no attempt status. Omitting `wait` preserves immediate status reads. With -`wait: {"after_revision": }`, status returns immediately if +`wait: {"after_revision": }`, status returns immediately if the current `attempt_revision` is greater than `after_revision` or the attempt is terminal -(`confirmed`, `expired`, or `rejected`). Otherwise it waits for a revision change -or terminal transition, bounded by the earlier of the original attempt deadline -and 30 seconds. The bound returns current status even if its revision is -unchanged; reissue `status` with the returned revision to keep waiting. Waiting -uses change notifications, not polling: bind, confirm, reject, cancel, expiry, -and recovery (`exchange_lost`) are observable. Opaque exchange send/receive -activity does not wake it. A pending wait is not an idle framing read. Waiting +(`confirmed`, `expired`, or `rejected`). A pending attempt refuses a revision +ahead of its current revision with code `status-wait-revision-ahead`; the reply's +`status.attempt_revision` identifies the current revision. This is not a transition. +Otherwise it waits for a revision change or terminal transition. One absolute +deadline, the earlier of the original attempt deadline and 30 seconds, covers +subscription, read-capacity acquisition, every state read, and notification waits. +At the bound it returns the last authenticated status snapshot normally, with +an unchanged revision if no transition was observed, even if a concurrent disk +operation prevents a fresh read. That snapshot may still be pending at expiry; +reissue with the returned revision to read projected expiry or keep waiting. +This notification re-arm is not polling. Bind, confirm, reject, cancel, expiry, +and recovery (`exchange_lost`) are observable; live reconciliation preserves +subscriptions. Opaque exchange activity does not wake status. A pending wait is +not an idle framing read. Waiting does not extend the permit deadline or grant access; the same bound-node checks and connection limits, including the 600-second connection lifetime, still apply. diff --git a/src/daemon.rs b/src/daemon.rs index 4beb503..eaa79ee 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -504,6 +504,8 @@ pub struct DaemonState { /// Network read permits live inside blocking workers, not socket futures. /// A timed-out read waiting for the runtime mutex still consumes its permit. pairing_read_slots: Arc, + #[cfg(test)] + pairing_wait_read: tokio::sync::watch::Sender>, dial_slots: Arc, mux_stream_slots: Arc, peer_connections: Arc, @@ -1355,6 +1357,8 @@ impl DaemonState { handshake_slots: Arc::new(Semaphore::new(MAX_INCOMING_HANDLERS)), pairing_slots: Arc::new(Semaphore::new(pairing::MAX_CONNECTIONS)), pairing_read_slots: Arc::new(Semaphore::new(pairing::MAX_CONNECTIONS)), + #[cfg(test)] + pairing_wait_read: tokio::sync::watch::channel(None).0, dial_slots: Arc::new(Semaphore::new(MAX_DIAL_HANDLERS)), mux_stream_slots: Arc::new(Semaphore::new(MAX_INCOMING_HANDLERS)), peer_connections, diff --git a/src/daemon/pairing.rs b/src/daemon/pairing.rs index f5ea7d7..c2993f4 100644 --- a/src/daemon/pairing.rs +++ b/src/daemon/pairing.rs @@ -517,15 +517,27 @@ async fn read_remote_reply( ); let mut bytes = vec![0; length]; recv.read_exact(&mut bytes).await?; - let reply: Reply = serde_json::from_slice(&bytes)?; + let mut reply: Reply = serde_json::from_slice(&bytes)?; let status = reply.status.as_ref().context("missing pairing status")?; anyhow::ensure!( reply.secret.is_none() - && reply.error.is_none() + && reply + .error + .as_ref() + .is_none_or(|error| error.code == "status-wait-revision-ahead") && status.attempt_id == attempt && status.bound_node_id == Some(node), "unbound pairing reply" ); + if reply.error.is_some() { + reply.error = Some(crate::runtime_peers::PolicyError::of( + &FabricError::new( + ErrorCode::StatusWaitRevisionAhead, + "status wait revision is ahead of the current attempt_revision", + ) + .into(), + )); + } Ok(reply.restricted()) } @@ -629,7 +641,8 @@ async fn serve_stream( let exchange = matches!(&request, NetworkRequest::Send { .. } | NetworkRequest::Receive { .. }); let reply = if matches!(&request, NetworkRequest::Status { wait: Some(_), .. }) { tokio::select! { - result = state.pairing_status_wait(connection.remote_id(), &request) => result?, + result = state.pairing_status_wait(connection.remote_id(), &request, + validated.context("missing authenticated status")?) => result?, _ = connection.closed() => return Ok(()), } } else if !exchange && validated.is_some() { @@ -645,14 +658,18 @@ async fn serve_stream( tokio::time::timeout(FIRST_HEADER, commit).await?? }; // The restricted path never reports fault internals or any unbound attempt. - if reply.error.is_some() { + if reply + .error + .as_ref() + .is_some_and(|error| error.code != "status-wait-revision-ahead") + { return Err(refused()); } let status = reply.status.as_ref().ok_or_else(refused)?; if status.bound_node_id == Some(connection.remote_id()) { state.pairing_gate.remember_bound(connection.remote_id(), Instant::now()); } - let terminal = !matches!(status.state, State::Pending); + let terminal = reply.error.is_some() || !matches!(status.state, State::Pending); let wall = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH)? .as_secs(); @@ -1324,7 +1341,7 @@ mod tests { owner.id(), Some(owner.addr()), NetworkRequest::Status { - attempt_id, + attempt_id: attempt_id.clone(), wait: Some(StatusWait { after_revision: pending.attempt_revision, }), @@ -1332,7 +1349,22 @@ mod tests { ) .await; assert!(reply.error.is_none(), "{mode}: {:?}", reply.error); - let status = reply.status.unwrap(); + let mut status = reply.status.unwrap(); + if mode == "expiry" && status.state == State::Pending { + // A bounded read may return its last authenticated snapshot. + status = remote( + &joining.state(), + owner.id(), + Some(owner.addr()), + NetworkRequest::Status { + attempt_id, + wait: None, + }, + ) + .await + .status + .unwrap(); + } match mode { "cap" => { assert_eq!(status, pending); diff --git a/src/daemon/runtime_control.rs b/src/daemon/runtime_control.rs index 7694e60..da705c8 100644 --- a/src/daemon/runtime_control.rs +++ b/src/daemon/runtime_control.rs @@ -484,6 +484,7 @@ impl DaemonState { &self, node: EndpointId, request: &runtime_peers::pairing::NetworkRequest, + mut last: runtime_peers::pairing::Reply, ) -> Result { use runtime_peers::pairing::{NetworkRequest, State}; let NetworkRequest::Status { @@ -492,40 +493,77 @@ impl DaemonState { else { anyhow::bail!("status wait requires a revision"); }; - // Subscribe before reading: commits between the read and changed() must - // remain observable, including commits made by off-thread workers. - let mut changes = self - .read_pairing_result(|runtime| Ok(runtime.pairing_changes.subscribe())) - .await?; - let cap = tokio::time::Instant::now() + Duration::from_secs(30); + let status = last + .status + .as_ref() + .context("missing authenticated status")?; + if status.state != State::Pending || status.attempt_revision > wait.after_revision { + return Ok(last); + } + if wait.after_revision > status.attempt_revision { + last.error = Some(PolicyError::of( + &FabricError::new( + ErrorCode::StatusWaitRevisionAhead, + "status wait revision is ahead of the current attempt_revision", + ) + .into(), + )); + return Ok(last); + } + let wall = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?; + let remaining = Duration::from_secs(status.deadline).saturating_sub(wall); + let deadline = tokio::time::Instant::now() + remaining.min(Duration::from_secs(30)); + // Capacity and mutex acquisition share the notification deadline. + let subscription = async { + let permit = self.pairing_read_slots.clone().acquire_owned().await?; + self.read_pairing_result(move |runtime| { + let _permit = permit; // Abandoned workers retain their capacity. + Ok(runtime.pairing_changes.subscribe()) + }) + .await + }; + let mut changes = match tokio::time::timeout_at(deadline, subscription).await { + Ok(result) => result?, + Err(_) => return Ok(last), + }; + let mut closed = false; loop { + if tokio::time::Instant::now() >= deadline { + return Ok(last); + } changes.borrow_and_update(); - let reply = self - .pairing_network_read(node, request) - .await? + match tokio::time::timeout_at(deadline, self.pairing_network_read(node, request)).await + { + Ok(result) => last = result?.context("missing authenticated status")?, + Err(_) => return Ok(last), + } + let status = last + .status + .as_ref() .context("missing authenticated status")?; - let Some(status) = reply.status.as_ref() else { - return Ok(reply); - }; - if status.attempt_revision > wait.after_revision || status.state != State::Pending { - return Ok(reply); + if status.state != State::Pending || status.attempt_revision > wait.after_revision { + return Ok(last); } - let wall = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH)? - .as_secs(); - let deadline = cap.min( - tokio::time::Instant::now() - + Duration::from_secs(status.deadline.saturating_sub(wall)), - ); - if tokio::time::timeout_at(deadline, changes.changed()) - .await - .is_err() - { - // Read again to project expiry even before the expiry worker commits. - return self - .pairing_network_read(node, request) - .await? - .context("missing authenticated status"); + if wait.after_revision > status.attempt_revision { + last.error = Some(PolicyError::of( + &FabricError::new( + ErrorCode::StatusWaitRevisionAhead, + "status wait revision is ahead of the current attempt_revision", + ) + .into(), + )); + return Ok(last); + } + #[cfg(test)] + self.pairing_wait_read + .send_replace(Some(status.attempt_id.clone())); + if closed || tokio::time::Instant::now() >= deadline { + return Ok(last); + } + match tokio::time::timeout_at(deadline, changes.changed()).await { + Ok(Ok(())) => {} + Ok(Err(_)) => closed = true, // Reread once, never spin on closure. + Err(_) => return Ok(last), } } } @@ -1572,8 +1610,7 @@ mod tests { .await; } - #[tokio::test] - async fn status_wait_observes_confirm_reject_cancel_and_immediate_revisions() -> Result<()> { + async fn status_wait_transition(action: &str) -> Result<()> { use runtime_peers::pairing::{ NetworkRequest, RejectRequest, Request as PairRequest, State, StatusWait, }; @@ -1583,74 +1620,233 @@ mod tests { FabricNode::start_with_daemon_options(FabricHome::new(dir.path()), options()).await?; let state = node.state(); let peer = iroh::SecretKey::from_bytes(&[86; 32]).public(); - for action in ["confirm", "reject", "cancel"] { - let confirmation = pending_confirmation(&state, action, peer, None).await?; - let request = NetworkRequest::Status { - attempt_id: confirmation.attempt_id.clone(), - wait: Some(StatusWait { - after_revision: confirmation.expected_attempt_revision - 1, - }), - }; - let advanced = tokio::time::timeout( - Duration::from_secs(1), - state.pairing_status_wait(peer, &request), - ) - .await??; - assert_eq!(advanced.status.unwrap().state, State::Pending); - let request = NetworkRequest::Status { - attempt_id: confirmation.attempt_id.clone(), - wait: Some(StatusWait { - after_revision: confirmation.expected_attempt_revision, - }), - }; - let waiting = state.pairing_status_wait(peer, &request); - tokio::pin!(waiting); - // Establish a real subscription before committing, without sleeps. - tokio::select! { - result = &mut waiting => panic!("wait completed before transition: {result:?}"), - _ = async { - loop { - if state.runtime_peers.as_ref().unwrap().lock().unwrap() - .pairing_changes.receiver_count() > 0 { break; } - tokio::task::yield_now().await; - } - } => {} - } - let transition = match action { - "confirm" => PairRequest::Confirm { confirmation }, - "reject" => PairRequest::Reject { - rejection: RejectRequest { - owner: confirmation.owner, - attempt_id: confirmation.attempt_id, - expected_node_id: peer, - expected_attempt_revision: confirmation.expected_attempt_revision, - request_id: "reject-wait".into(), - }, - }, - _ => PairRequest::Cancel { + let confirmation = pending_confirmation(&state, action, peer, None).await?; + let request = NetworkRequest::Status { + attempt_id: confirmation.attempt_id.clone(), + wait: Some(StatusWait { + after_revision: confirmation.expected_attempt_revision - 1, + }), + }; + let advanced = + tokio::time::timeout(Duration::from_secs(1), wait_status(&state, peer, &request)) + .await??; + assert_eq!(advanced.status.unwrap().state, State::Pending); + let request = NetworkRequest::Status { + attempt_id: confirmation.attempt_id.clone(), + wait: Some(StatusWait { + after_revision: confirmation.expected_attempt_revision, + }), + }; + let mut reads = state.pairing_wait_read.subscribe(); + let waiting = wait_status(&state, peer, &request); + tokio::pin!(waiting); + tokio::select! { + result = &mut waiting => panic!("wait completed before transition: {result:?}"), + result = tokio::time::timeout(Duration::from_secs(5), async { + loop { + reads.changed().await.unwrap(); + if reads.borrow().as_ref() == Some(&confirmation.attempt_id) { break; } + } + }) => { result?; } + } + let transition = match action { + "confirm" => PairRequest::Confirm { confirmation }, + "reject" => PairRequest::Reject { + rejection: RejectRequest { owner: confirmation.owner, - creation_id: action.into(), + attempt_id: confirmation.attempt_id, + expected_node_id: peer, + expected_attempt_revision: confirmation.expected_attempt_revision, + request_id: "reject-wait".into(), }, - }; - let committed = state.pairing_request(transition).await; - assert!(committed.error.is_none(), "{:?}", committed.error); - let terminal = committed.status.unwrap(); - let observed = tokio::time::timeout(Duration::from_secs(1), &mut waiting).await??; - assert_eq!(observed.status.unwrap(), terminal); - // Terminal always returns, even if the client's revision is ahead. - let request = NetworkRequest::Status { - attempt_id: terminal.attempt_id, - wait: Some(StatusWait { - after_revision: u64::MAX, - }), - }; - let observed = tokio::time::timeout( - Duration::from_secs(1), - state.pairing_status_wait(peer, &request), - ) - .await??; - assert_eq!(observed.status.unwrap().state, terminal.state); + }, + _ => PairRequest::Cancel { + owner: confirmation.owner, + creation_id: action.into(), + }, + }; + let committed = if action == "expiry" { + let request = request.clone(); + state + .commit_pairing(move |runtime| { + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_secs() + + 601; + runtime.advance_pairing_clock_for_test(timestamp); + runtime.pairing_expire()?; + runtime + .pairing_network_read(peer, &request)? + .context("expired status") + }) + .await + } else { + state.pairing_request(transition).await + }; + assert!(committed.error.is_none(), "{:?}", committed.error); + let terminal = committed.status.unwrap(); + let observed = tokio::time::timeout(Duration::from_secs(1), &mut waiting).await??; + assert_eq!(observed.status.unwrap(), terminal); + // Terminal always returns, even if the client's revision is ahead. + let request = NetworkRequest::Status { + attempt_id: terminal.attempt_id, + wait: Some(StatusWait { + after_revision: u64::MAX, + }), + }; + let observed = + tokio::time::timeout(Duration::from_secs(1), wait_status(&state, peer, &request)) + .await??; + assert_eq!(observed.status.unwrap().state, terminal.state); + node.shutdown().await?; + Ok(()) + } + + async fn wait_status( + state: &DaemonState, + peer: EndpointId, + request: &runtime_peers::pairing::NetworkRequest, + ) -> Result { + let initial = state.pairing_network_read(peer, request).await?.unwrap(); + state.pairing_status_wait(peer, request, initial).await + } + + #[tokio::test] + async fn status_wait_confirm_notification() -> Result<()> { + status_wait_transition("confirm").await + } + #[tokio::test] + async fn status_wait_reject_notification() -> Result<()> { + status_wait_transition("reject").await + } + #[tokio::test] + async fn status_wait_cancel_notification() -> Result<()> { + status_wait_transition("cancel").await + } + #[tokio::test] + async fn status_wait_expiry_notification() -> Result<()> { + status_wait_transition("expiry").await + } + + #[tokio::test] + async fn status_wait_ahead_is_typed_and_terminal_is_normal() -> Result<()> { + use runtime_peers::pairing::{NetworkRequest, StatusWait}; + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let node = + FabricNode::start_with_daemon_options(FabricHome::new(dir.path()), options()).await?; + let state = node.state(); + let peer = iroh::SecretKey::from_bytes(&[87; 32]).public(); + let confirmation = pending_confirmation(&state, "ahead", peer, None).await?; + let request = NetworkRequest::Status { + attempt_id: confirmation.attempt_id, + wait: Some(StatusWait { + after_revision: u64::MAX, + }), + }; + let reply = + tokio::time::timeout(Duration::from_secs(5), wait_status(&state, peer, &request)) + .await??; + assert_eq!(reply.error.unwrap().code, "status-wait-revision-ahead"); + assert_eq!( + reply.status.unwrap().attempt_revision, + confirmation.expected_attempt_revision + ); + node.shutdown().await?; + Ok(()) + } + + #[tokio::test] + async fn status_wait_mutex_stall_keeps_absolute_cap_and_worker_permit() -> Result<()> { + use runtime_peers::pairing::{NetworkRequest, StatusWait}; + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let node = + FabricNode::start_with_daemon_options(FabricHome::new(dir.path()), options()).await?; + let state = node.state(); + let peer = iroh::SecretKey::from_bytes(&[88; 32]).public(); + let confirmation = pending_confirmation(&state, "stalled", peer, None).await?; + let request = NetworkRequest::Status { + attempt_id: confirmation.attempt_id, + wait: Some(StatusWait { + after_revision: confirmation.expected_attempt_revision, + }), + }; + let mut reads = state.pairing_wait_read.subscribe(); + let waiting = wait_status(&state, peer, &request); + tokio::pin!(waiting); + tokio::select! { + result = &mut waiting => panic!("premature reply: {result:?}"), + result = reads.changed() => { result?; } + } + let runtime = state.runtime_peers.as_ref().unwrap(); + let guard = runtime.lock().unwrap(); + guard.pairing_changes.send_replace(()); + tokio::time::pause(); + queue_policy_writer(waiting.as_mut()).await; + tokio::time::advance(Duration::from_secs(30)).await; + let reply = waiting.await?; + assert_eq!( + reply.status.unwrap().attempt_revision, + confirmation.expected_attempt_revision + ); + assert!(reply.error.is_none()); + assert_eq!( + state.pairing_read_slots.available_permits(), + pairing::MAX_CONNECTIONS - 1 + ); + drop(guard); + tokio::time::resume(); + tokio::time::timeout(Duration::from_secs(5), async { + while state.pairing_read_slots.available_permits() != pairing::MAX_CONNECTIONS { + tokio::task::yield_now().await; + } + }) + .await?; + node.shutdown().await?; + Ok(()) + } + + #[tokio::test] + async fn status_wait_live_reconcile_preserves_subscription() -> Result<()> { + use runtime_peers::pairing::{NetworkRequest, Request as PairRequest, State, StatusWait}; + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let node = FabricNode::start_with_daemon_options(home.clone(), options()).await?; + let state = node.state(); + let peer = iroh::SecretKey::from_bytes(&[89; 32]).public(); + let confirmation = pending_confirmation(&state, "reconcile-wait", peer, None).await?; + let request = NetworkRequest::Status { + attempt_id: confirmation.attempt_id, + wait: Some(StatusWait { + after_revision: confirmation.expected_attempt_revision, + }), + }; + let mut reads = state.pairing_wait_read.subscribe(); + let waiting = wait_status(&state, peer, &request); + tokio::pin!(waiting); + tokio::select! { + result = &mut waiting => panic!("premature reply: {result:?}"), + result = reads.changed() => { result?; } + } + reads.borrow_and_update(); + state + .read_pairing_result(move |runtime| runtime.reconcile(&home)) + .await?; + tokio::select! { + result = &mut waiting => panic!("reconcile ended pending wait: {result:?}"), + result = tokio::time::timeout(Duration::from_secs(5), reads.changed()) => { result??; } } + let committed = state + .pairing_request(PairRequest::Cancel { + owner: "controller".into(), + creation_id: "reconcile-wait".into(), + }) + .await; + assert!(committed.error.is_none()); + let reply = tokio::time::timeout(Duration::from_secs(5), waiting).await??; + assert_eq!(reply.status.unwrap().state, State::Rejected); node.shutdown().await?; Ok(()) } diff --git a/src/runtime_peers.rs b/src/runtime_peers.rs index 82001e0..8f243e8 100644 --- a/src/runtime_peers.rs +++ b/src/runtime_peers.rs @@ -664,6 +664,8 @@ impl RuntimePeers { loaded.pairing_clock = self.pairing_clock; loaded.exchanges = std::mem::take(&mut self.exchanges); loaded.prune_exchanges(); + loaded.pairing_changes = self.pairing_changes.clone(); + loaded.pairing_changes.send_replace(()); *self = loaded; Ok(book) } diff --git a/src/runtime_peers/pairing.rs b/src/runtime_peers/pairing.rs index 53786de..113ce7e 100644 --- a/src/runtime_peers/pairing.rs +++ b/src/runtime_peers/pairing.rs @@ -426,6 +426,10 @@ fn observe_clock( } impl RuntimePeers { + #[cfg(test)] + pub(crate) fn advance_pairing_clock_for_test(&mut self, timestamp: u64) { + self.pairing_clock = (timestamp, std::time::Instant::now()); + } pub(super) fn pairing_timestamp(&mut self) -> u64 { observe_clock(&mut self.pairing_clock, now(), std::time::Instant::now()) }