diff --git a/CHANGELOG.md b/CHANGELOG.md index 37ba459..5ebfba0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,18 @@ 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. + 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 - **Provisional:** restricted pairing terminal replies use an independent 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 c688145..8797f4a 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,26 @@ 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 the current +`attempt_revision` is greater than `after_revision` or the attempt is terminal +(`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. + 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.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 10b4165..c2993f4 100644 --- a/src/daemon/pairing.rs +++ b/src/daemon/pairing.rs @@ -243,9 +243,22 @@ 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 +269,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 +403,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 +416,9 @@ 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( @@ -425,6 +440,7 @@ pub(super) async fn remote( // Reconnect authenticates a small frame before a large exchange. 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?; @@ -501,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()) } @@ -599,7 +627,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 +639,13 @@ 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, + validated.context("missing authenticated status")?) => 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 @@ -624,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(); @@ -794,9 +832,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(), - }); + 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 +936,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 +974,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() }; + 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. @@ -1100,6 +1144,7 @@ mod tests { &mut recv, NetworkRequest::Status { attempt_id: attempt_id.clone(), + wait: None, }, ) .await?; @@ -1128,7 +1173,10 @@ mod tests { rpc( &mut other_send, &mut other_recv, - NetworkRequest::Status { attempt_id } + NetworkRequest::Status { + attempt_id, + wait: None + } ) .await .is_err() @@ -1219,6 +1267,7 @@ mod tests { &mut recv, NetworkRequest::Status { attempt_id: attempt_id.clone(), + wait: None, }, ) .await?; @@ -1242,6 +1291,101 @@ 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: attempt_id.clone(), + wait: Some(StatusWait { + after_revision: pending.attempt_revision, + }), + }, + ) + .await; + assert!(reply.error.is_none(), "{mode}: {:?}", reply.error); + 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); + 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; @@ -1271,7 +1415,14 @@ 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, @@ -1288,6 +1439,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()); } @@ -1660,7 +1814,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 }).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); @@ -1916,6 +2078,7 @@ mod tests { control_ready(&joining_home).await?; 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"); @@ -1944,6 +2107,7 @@ mod tests { addr: Some(spoofed), request: NetworkRequest::Status { attempt_id: "unknown".into(), + wait: None, }, }, ) @@ -1986,7 +2150,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 { @@ -2040,9 +2204,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..da705c8 100644 --- a/src/daemon/runtime_control.rs +++ b/src/daemon/runtime_control.rs @@ -480,6 +480,94 @@ impl DaemonState { }).await } + pub(super) async fn pairing_status_wait( + &self, + node: EndpointId, + request: &runtime_peers::pairing::NetworkRequest, + mut last: runtime_peers::pairing::Reply, + ) -> Result { + use runtime_peers::pairing::{NetworkRequest, State}; + let NetworkRequest::Status { + wait: Some(wait), .. + } = request + else { + anyhow::bail!("status wait requires a revision"); + }; + 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(); + 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")?; + 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); + } + #[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), + } + } + } + async fn confirm_pairing( &self, confirmation: runtime_peers::pairing::ConfirmRequest, @@ -1522,6 +1610,247 @@ mod tests { .await; } + async fn status_wait_transition(action: &str) -> 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(); + 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, + 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 = 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(()) + } + #[tokio::test] async fn concurrent_pairing_confirm_confirm_installs_once() -> Result<()> { use runtime_peers::pairing::Request as PairRequest; @@ -1754,7 +2083,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..8f243e8 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)] @@ -572,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) })(); @@ -661,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) } @@ -687,6 +692,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..113ce7e 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; @@ -418,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()) } @@ -443,7 +455,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 +809,7 @@ impl RuntimePeers { return Err(error); } self.stored = next; + self.pairing_changes.send_replace(()); } self.prune_exchanges(); Ok(()) @@ -1259,6 +1272,7 @@ mod tests { fn restricted_status(status: &Status) -> NetworkRequest { NetworkRequest::Status { attempt_id: status.attempt_id.clone(), + wait: None, } } @@ -2290,7 +2304,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..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) } @@ -805,6 +806,7 @@ mod tests { node(1), NetworkRequest::Status { attempt_id: request.attempt_id.clone(), + wait: None, }, t + 1, )