From 1c4dd0eede5a3600c17a753b79e563b18d7b3b89 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:28:25 +0100 Subject: [PATCH 1/2] fix(pairing): preserve retained terminal reply write budget 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 | 6 +++ docs/runtime-peer-api.md | 4 +- src/daemon/pairing.rs | 80 +++++++++++++++++++++++++++++++++++++++- 3 files changed, 88 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d740d10..37ba459 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,12 @@ Deprecated, Fixed or Security. ## [Unreleased] +### Fixed + +- **Provisional:** restricted pairing terminal replies use an independent + ten-second write budget, allowing retained expired status to be returned after + the original attempt deadline. + ## [0.9.4] - 2026-11-03 ### Fixed diff --git a/docs/runtime-peer-api.md b/docs/runtime-peer-api.md index 8f82fde..c688145 100644 --- a/docs/runtime-peer-api.md +++ b/docs/runtime-peer-api.md @@ -129,7 +129,9 @@ validated bound identities use an independent bucket (burst 8, refill one/second selected by a bounded expiring hint that never authorizes an operation. Ordinary traffic uses neither pool nor bucket. Pairing availability is best effort under attack. First-header waits are limited to three seconds; subsequent idle framing -waits are ten seconds, and connection lifetime is bounded by the attempt deadline. +waits are ten seconds. Active-attempt work is bounded by the attempt deadline; +terminal replies, including retained expired status, have a separate ten-second +write budget so the original deadline cannot suppress the response. Post-ALPN transport settings allow one bidi stream, zero uni streams and an 8 KiB receive window; already-advertised QUIC credits cannot be retracted. diff --git a/src/daemon/pairing.rs b/src/daemon/pairing.rs index 9fd2e4f..0dc30ff 100644 --- a/src/daemon/pairing.rs +++ b/src/daemon/pairing.rs @@ -638,8 +638,14 @@ async fn serve_stream( deadline = tokio::time::Instant::now() + Duration::from_secs(status.deadline.saturating_sub(wall).min(600)); let bytes = serde_json::to_vec(&reply.restricted())?; + // Retained terminal receipts outlive the permit's original deadline. + let write_deadline = if terminal { + tokio::time::Instant::now() + IDLE + } else { + deadline.min(tokio::time::Instant::now() + IDLE) + }; tokio::time::timeout_at( - deadline.min(tokio::time::Instant::now() + IDLE), + write_deadline, async { send.write_all(&(bytes.len() as u32).to_be_bytes()).await?; send.write_all(&bytes).await @@ -1132,6 +1138,78 @@ mod tests { gateway.shutdown().await?; Ok(()) } + #[tokio::test] + async fn retained_terminal_status_writes_after_original_deadline() -> Result<()> { + use crate::runtime_peers::pairing::{PAIR_ALPN, Policy, Request}; + let dir = tempfile::tempdir()?; + let gateway = gateway(FabricHome::new(dir.path())).await?; + let state = gateway.state(); + let deadline = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)?.as_secs() + 5; + let issued = state.pairing_request(Request::Issue { + owner: "test".into(), + creation_id: "retained-terminal".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.context("missing issued status")?.attempt_id; + // Force the reply to await QUIC flow-control credit instead of fitting + // entirely in a ready write, which can mask an already-expired timeout. + let client = Endpoint::builder(presets::N0) + .transport_config(iroh::endpoint::QuicTransportConfig::builder() + .stream_receive_window(VarInt::from_u32(64)) + .build()) + .bind().await?; + let connection = client.connect(gateway.addr(), PAIR_ALPN).await?; + let (mut send, mut recv) = connection.open_bi().await?; + let pending = rpc(&mut send, &mut recv, NetworkRequest::Present { + attempt_id: attempt_id.clone(), + secret: issued.secret.context("missing permit secret")?, + }).await?; + assert_eq!(pending.status.context("missing pending status")?.state, State::Pending); + let cancelled = state.pairing_request(Request::Cancel { + owner: "test".into(), + creation_id: "retained-terminal".into(), + }).await; + assert!(cancelled.error.is_none(), "{:?}", cancelled.error); + let retained = cancelled.status.context("missing cancelled status")?; + assert_eq!(retained.state, State::Rejected); + connection.close(VarInt::from_u32(0), b"reconnect after deadline"); + tokio::time::timeout(Duration::from_secs(3), async { + while !state.pairing_gate.attempts.lock().is_empty() { + tokio::task::yield_now().await; + } + }).await?; + let expired_at = std::time::UNIX_EPOCH + Duration::from_secs(deadline); + tokio::time::sleep(expired_at.duration_since(std::time::SystemTime::now()) + .unwrap_or_default() + Duration::from_millis(50)).await; + assert!(std::time::SystemTime::now() >= expired_at); + let connection = client.connect(gateway.addr(), PAIR_ALPN).await?; + let (mut send, mut recv) = connection.open_bi().await?; + let reply = rpc(&mut send, &mut recv, NetworkRequest::Status { + attempt_id: attempt_id.clone(), + }).await?; + assert!(reply.error.is_none(), "{:?}", reply.error); + let status = reply.status.context("missing retained terminal status")?; + assert_eq!(status.attempt_id, attempt_id); + assert_eq!(status.state, State::Rejected); + assert_eq!(status.deadline, deadline); + assert_eq!(status.attempt_revision, retained.attempt_revision); + assert_eq!(status.bound_node_id, Some(client.id())); + assert_eq!(status.terminal_at, retained.terminal_at); + let mut trailing = [0; 1]; + assert!(tokio::time::timeout(Duration::from_secs(3), recv.read(&mut trailing)) + .await??.is_none(), "terminal reply must finish the stream"); + client.close().await; + gateway.shutdown().await?; + Ok(()) + } + #[tokio::test] async fn disabled_pairing_and_oversized_frames_create_no_pending_record() -> Result<()> { use crate::runtime_peers::pairing::PAIR_ALPN; From 25bb3c444a98747f347d0d210330674b4c7fbbb7 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:35:15 +0100 Subject: [PATCH 2/2] style(pairing): format retained terminal regression 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 | 96 ++++++++++++++++++++++++++++--------------- 1 file changed, 64 insertions(+), 32 deletions(-) diff --git a/src/daemon/pairing.rs b/src/daemon/pairing.rs index 0dc30ff..10b4165 100644 --- a/src/daemon/pairing.rs +++ b/src/daemon/pairing.rs @@ -1145,37 +1145,54 @@ mod tests { let gateway = gateway(FabricHome::new(dir.path())).await?; let state = gateway.state(); let deadline = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH)?.as_secs() + 5; - let issued = state.pairing_request(Request::Issue { - owner: "test".into(), - creation_id: "retained-terminal".into(), - deadline, - policy: Policy { - grants: vec![], - name: None, - direction: crate::runtime_peers::Direction::Normal, - }, - }).await; + .duration_since(std::time::UNIX_EPOCH)? + .as_secs() + + 5; + let issued = state + .pairing_request(Request::Issue { + owner: "test".into(), + creation_id: "retained-terminal".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.context("missing issued status")?.attempt_id; // Force the reply to await QUIC flow-control credit instead of fitting // entirely in a ready write, which can mask an already-expired timeout. let client = Endpoint::builder(presets::N0) - .transport_config(iroh::endpoint::QuicTransportConfig::builder() - .stream_receive_window(VarInt::from_u32(64)) - .build()) - .bind().await?; + .transport_config( + iroh::endpoint::QuicTransportConfig::builder() + .stream_receive_window(VarInt::from_u32(64)) + .build(), + ) + .bind() + .await?; let connection = client.connect(gateway.addr(), PAIR_ALPN).await?; let (mut send, mut recv) = connection.open_bi().await?; - let pending = rpc(&mut send, &mut recv, NetworkRequest::Present { - attempt_id: attempt_id.clone(), - secret: issued.secret.context("missing permit secret")?, - }).await?; - assert_eq!(pending.status.context("missing pending status")?.state, State::Pending); - let cancelled = state.pairing_request(Request::Cancel { - owner: "test".into(), - creation_id: "retained-terminal".into(), - }).await; + let pending = rpc( + &mut send, + &mut recv, + NetworkRequest::Present { + attempt_id: attempt_id.clone(), + secret: issued.secret.context("missing permit secret")?, + }, + ) + .await?; + assert_eq!( + pending.status.context("missing pending status")?.state, + State::Pending + ); + let cancelled = state + .pairing_request(Request::Cancel { + owner: "test".into(), + creation_id: "retained-terminal".into(), + }) + .await; assert!(cancelled.error.is_none(), "{:?}", cancelled.error); let retained = cancelled.status.context("missing cancelled status")?; assert_eq!(retained.state, State::Rejected); @@ -1184,16 +1201,27 @@ mod tests { while !state.pairing_gate.attempts.lock().is_empty() { tokio::task::yield_now().await; } - }).await?; + }) + .await?; let expired_at = std::time::UNIX_EPOCH + Duration::from_secs(deadline); - tokio::time::sleep(expired_at.duration_since(std::time::SystemTime::now()) - .unwrap_or_default() + Duration::from_millis(50)).await; + tokio::time::sleep( + expired_at + .duration_since(std::time::SystemTime::now()) + .unwrap_or_default() + + Duration::from_millis(50), + ) + .await; assert!(std::time::SystemTime::now() >= expired_at); let connection = client.connect(gateway.addr(), PAIR_ALPN).await?; let (mut send, mut recv) = connection.open_bi().await?; - let reply = rpc(&mut send, &mut recv, NetworkRequest::Status { - attempt_id: attempt_id.clone(), - }).await?; + let reply = rpc( + &mut send, + &mut recv, + NetworkRequest::Status { + attempt_id: attempt_id.clone(), + }, + ) + .await?; assert!(reply.error.is_none(), "{:?}", reply.error); let status = reply.status.context("missing retained terminal status")?; assert_eq!(status.attempt_id, attempt_id); @@ -1203,8 +1231,12 @@ mod tests { assert_eq!(status.bound_node_id, Some(client.id())); assert_eq!(status.terminal_at, retained.terminal_at); let mut trailing = [0; 1]; - assert!(tokio::time::timeout(Duration::from_secs(3), recv.read(&mut trailing)) - .await??.is_none(), "terminal reply must finish the stream"); + assert!( + tokio::time::timeout(Duration::from_secs(3), recv.read(&mut trailing)) + .await?? + .is_none(), + "terminal reply must finish the stream" + ); client.close().await; gateway.shutdown().await?; Ok(())