diff --git a/docs/adr/ADR-0015-direct-browser-clients-over-webrtc-direct.md b/docs/adr/ADR-0015-direct-browser-clients-over-webrtc-direct.md index 3b6742ef..02cfd67a 100644 --- a/docs/adr/ADR-0015-direct-browser-clients-over-webrtc-direct.md +++ b/docs/adr/ADR-0015-direct-browser-clients-over-webrtc-direct.md @@ -1058,3 +1058,46 @@ The decision advances beyond PoC only after all of the following are covered: AI tools may help draft this ADR, but **must not mark it Accepted without human review**. Accepted ADRs are immutable: create a new superseding ADR rather than editing an Accepted ADR. + +## Bounded concurrent RPCs and cancellation (2026-09-17) + +Nodes advertise `rpc-multiplex-4` in authenticated HELLO. Supporting clients may +have at most four uncompleted RPCs on each ordered DataChannel; older nodes keep +one. This ceiling is a protocol resource limit, not a device-specific throughput +setting. Global read memory/CPU admission and server source/rate/byte limits still +apply. Request IDs correlate out-of-order completions. One writer serializes whole +frames and AEAD sealing; one reader authenticates frames in wire sequence before +dispatch. Frames are not interleaved, and no cryptographic checks are removed. + +Cancelling an admitted caller abandons its result, not its encrypted exchange. +The transport drains the response under ordinary deadlines, retaining its RPC +slot and any physical-read reservation. Unsent queued work remains cancellable. +Cancelled PUTs are not interpreted as failed payments or rolled back storage. +Explicit client closure and genuine transport/protocol failure close the session. +For a response already transferring, size-derived frame deadlines still apply; +first-response timers cannot cut an authenticated bulk frame short. + +The server keeps receiving while work executes and while replies are written. +A channel reset retires channel I/O independently of tracked storage/payment +workers and their reservations. An excess replacement channel is rejected without +tearing down a healthy sibling channel. Admission remains bounded per channel, +connection, source, and listener. Wire request/response formats and native QUIC +behavior are unchanged; the capability is additive and old clients remain valid. + +Draining preserves authentication but does not stop server work or reclaim bytes +already sent. A full response frame can still delay later responses on its lane. +Further wire-level cancellation or fragment interleaving would be a separate +protocol decision, justified by measured benefit and resource accounting. + +Validation covers delayed work, out-of-order replies, cancelled active and queued +callers, authenticated-session reuse, old-node fallback, ingress limits, transfer +deadlines, and a real WebRTC blocked-storage regression. Live benchmarks compare +unchanged file bytes and integrity checks, elapsed time and received bytes. + +Bulk response preparation additionally queues behind per-source and global +permits derived from the existing byte budgets and worst-case response copies. +Those permits remain held through writing. At default limits, a single source +pipelines GETs while preparing one large reply at a time; control operations and +PUT processing remain independently admitted. This avoids four prepared records +consuming the space needed to encode and send any one of them. Exact byte +accounting remains authoritative; configured byte limits are unchanged. diff --git a/src/replication/config.rs b/src/replication/config.rs index 9150d584..1781aba7 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -170,6 +170,17 @@ pub const SELF_LOOKUP_INTERVAL_MAX: Duration = Duration::from_secs(SELF_LOOKUP_I /// at most ~12 MB queued for the upload link at any instant. pub const MAX_CONCURRENT_REPLICATION_SENDS: usize = 3; +/// Maximum number of encoded fresh-replication offers held in memory. +/// +/// Each accepted write is encoded once (chunk plus proof, up to ~4 MB) and +/// that buffer stays alive until the last of its per-peer sends completes. +/// With only `MAX_CONCURRENT_REPLICATION_SENDS` transfers in flight, a write +/// rate above the network's send rate would otherwise queue an unbounded +/// number of encoded offers behind the send permits. The fresh-write drainer +/// takes one of these permits before it reads and encodes a chunk, so the +/// backlog waits as small queued events instead of chunk-sized buffers. +pub const MAX_PENDING_FRESH_OFFERS: usize = 8; + /// Maximum number of concurrent in-flight audit-responder tasks. /// /// The LIGHT audit-responder handlers — responsible-chunk audits and subtree diff --git a/src/replication/fresh.rs b/src/replication/fresh.rs index 80e9766e..d41e283d 100644 --- a/src/replication/fresh.rs +++ b/src/replication/fresh.rs @@ -11,7 +11,7 @@ use crate::logging::{debug, warn}; use rand::Rng; use saorsa_core::identity::PeerId; use saorsa_core::P2PNode; -use tokio::sync::Semaphore; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use crate::ant_protocol::XorName; use crate::replication::config::{ @@ -26,16 +26,27 @@ use crate::replication::protocol::{ /// /// Sent from the chunk PUT handler to the replication engine via an /// unbounded channel so that the PUT response is not blocked by -/// replication fan-out. +/// replication fan-out. The event deliberately carries no chunk bytes: the +/// chunk is already on disk, and the drainer reads it back only once it +/// holds a pending-offer permit, so a replication backlog queues as small +/// events rather than chunk-sized buffers. pub struct FreshWriteEvent { /// Content-address of the stored chunk. pub key: XorName, - /// The chunk data. - pub data: Vec, /// Serialized proof-of-payment. pub payment_proof: Vec, } +/// An encoded fresh offer shared by the per-peer send tasks. +/// +/// The pending-offer permit is released together with the buffer, once the +/// last send task drops its reference, which caps how many encoded offers +/// can wait behind the send permits at `MAX_PENDING_FRESH_OFFERS`. +struct EncodedOffer { + bytes: Vec, + _pending: OwnedSemaphorePermit, +} + /// Execute fresh replication for a newly accepted record. /// /// Sends fresh offers to close group members (with bounded delivery retries, @@ -45,7 +56,10 @@ pub struct FreshWriteEvent { /// /// The `send_semaphore` limits how many outbound chunk transfers can be /// in-flight concurrently across the entire replication engine, preventing -/// bandwidth saturation on home broadband connections. +/// bandwidth saturation on home broadband connections. `pending_offer` is the +/// caller's permit from the pending-offer semaphore; it is held with the +/// encoded offer until the last per-peer send finishes. +#[allow(clippy::too_many_arguments)] pub async fn replicate_fresh( key: &XorName, data: &[u8], @@ -54,6 +68,7 @@ pub async fn replicate_fresh( paid_list: &Arc, config: &ReplicationConfig, send_semaphore: &Arc, + pending_offer: OwnedSemaphorePermit, ) -> Vec { let self_id = *p2p_node.peer_id(); @@ -95,11 +110,15 @@ pub async fn replicate_fresh( }; // Share one encoded copy across the per-peer send tasks so a retry only // re-materialises the buffer for the (consuming) send call, keeping the - // common single-attempt path at one clone per peer. - let encoded = Arc::new(encoded); + // common single-attempt path at one clone per peer. The pending-offer + // permit travels with the buffer. + let encoded = Arc::new(EncodedOffer { + bytes: encoded, + _pending: pending_offer, + }); for peer in &target_peers { let p2p = Arc::clone(p2p_node); - let data = Arc::clone(&encoded); + let offer = Arc::clone(&encoded); let peer_id = *peer; let sem = Arc::clone(send_semaphore); tokio::spawn(async move { @@ -117,12 +136,7 @@ pub async fn replicate_fresh( let mut attempt = 0u32; loop { match p2p - .send_message( - &peer_id, - REPLICATION_PROTOCOL_ID, - data.as_ref().clone(), - &[], - ) + .send_message(&peer_id, REPLICATION_PROTOCOL_ID, offer.bytes.clone(), &[]) .await { Ok(()) => break, diff --git a/src/replication/mod.rs b/src/replication/mod.rs index e245bd9d..9cb2263e 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -79,7 +79,7 @@ use crate::replication::commitment_state::{ use crate::replication::config::{ max_parallel_fetch, storage_admission_width, ReplicationConfig, MAX_AUDIT_RESPONSES_PER_PEER, MAX_CONCURRENT_AUDIT_RESPONSES, MAX_CONCURRENT_REPLICATION_SENDS, - MAX_DIGEST_AUDIT_RESPONSES_PER_PEER, MAX_INCOMING_VERIFICATION_KEYS, + MAX_DIGEST_AUDIT_RESPONSES_PER_PEER, MAX_INCOMING_VERIFICATION_KEYS, MAX_PENDING_FRESH_OFFERS, MAX_SUBTREE_ROUND1_PER_PEER, MAX_SUBTREE_SESSIONS, MAX_VERIFICATION_KEYS_PER_CYCLE, REPLICATION_PROTOCOL_ID, SUBTREE_AUDIT_PROTOCOL_ID, SUBTREE_ROUND1_WORK_BURST_BYTES, SUBTREE_ROUND1_WORK_REFILL_BYTES_PER_SEC, SUBTREE_SESSION_TTL, @@ -1746,6 +1746,9 @@ pub struct ReplicationEngine { /// Limits concurrent outbound replication sends to prevent bandwidth /// saturation on home broadband connections. send_semaphore: Arc, + /// Bounds how many encoded fresh offers can wait behind `send_semaphore`; + /// see [`MAX_PENDING_FRESH_OFFERS`]. + pending_offer_semaphore: Arc, /// Bounds concurrent IN-FLIGHT LIGHT audit-responder tasks (responsible-chunk /// audits + subtree slice round 2). The heavy subtree round 1 has its own /// tighter pool ([`SubtreeRound1Limiter`]). Those are spawned off the serial @@ -1911,6 +1914,7 @@ impl ReplicationEngine { recent_provers: Arc::new(RwLock::new(RecentProvers::new())), sig_verify_attempts: Arc::new(RwLock::new(HashMap::new())), send_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_REPLICATION_SENDS)), + pending_offer_semaphore: Arc::new(Semaphore::new(MAX_PENDING_FRESH_OFFERS)), audit_responder_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_AUDIT_RESPONSES)), audit_responder_inflight: Arc::new(RwLock::new(HashMap::new())), audit_responder_metrics: Arc::new(AuditResponderMetrics::default()), @@ -2369,6 +2373,13 @@ impl ReplicationEngine { /// drainer; this direct entry point schedules here so callers (and tests) /// that drive replication directly still get the possession check. pub async fn replicate_fresh(&self, key: &XorName, data: &[u8], proof_of_payment: &[u8]) { + // The semaphore is never closed, so this only fails at shutdown. + let Ok(pending_offer) = Arc::clone(&self.pending_offer_semaphore) + .acquire_owned() + .await + else { + return; + }; let peers = fresh::replicate_fresh( key, data, @@ -2377,6 +2388,7 @@ impl ReplicationEngine { &self.paid_list, &self.config, &self.send_semaphore, + pending_offer, ) .await; if !peers.is_empty() { @@ -2398,37 +2410,62 @@ impl ReplicationEngine { }; let p2p = Arc::clone(&self.p2p_node); let paid_list = Arc::clone(&self.paid_list); + let storage = Arc::clone(&self.storage); let config = Arc::clone(&self.config); let send_semaphore = Arc::clone(&self.send_semaphore); + let pending_offer_semaphore = Arc::clone(&self.pending_offer_semaphore); let possession_tx = self.possession_check_tx.clone(); let shutdown = self.shutdown.clone(); let handle = tokio::spawn(async move { loop { - tokio::select! { + let event = tokio::select! { () = shutdown.cancelled() => break, event = rx.recv() => { let Some(event) = event else { break }; - let peers = fresh::replicate_fresh( - &event.key, - &event.data, - &event.payment_proof, - &p2p, - &paid_list, - &config, - &send_semaphore, - ) - .await; - // Schedule the delayed possession check (ADR-0003) for - // the responsible close-group peers. A closed receiver - // (engine shutting down) is ignored. - if !peers.is_empty() { - let _ = possession_tx.send(possession::PossessionCheckEvent { - key: event.key, - peers, - }); - } + event + } + }; + // Wait for a pending-offer permit before touching the chunk so a + // send backlog holds queued events, not encoded chunk buffers. + let pending_offer = tokio::select! { + () = shutdown.cancelled() => break, + permit = Arc::clone(&pending_offer_semaphore).acquire_owned() => { + let Ok(permit) = permit else { break }; + permit + } + }; + let key_hex = hex::encode(event.key); + let data = match storage.get(&event.key).await { + Ok(Some(data)) => data, + Ok(None) => { + debug!("Chunk {key_hex} no longer stored, skipping fresh replication"); + continue; } + Err(e) => { + warn!("Failed to read chunk {key_hex} for fresh replication: {e}"); + continue; + } + }; + let peers = fresh::replicate_fresh( + &event.key, + &data, + &event.payment_proof, + &p2p, + &paid_list, + &config, + &send_semaphore, + pending_offer, + ) + .await; + // Schedule the delayed possession check (ADR-0003) for + // the responsible close-group peers. A closed receiver + // (engine shutting down) is ignored. + if !peers.is_empty() { + let _ = possession_tx.send(possession::PossessionCheckEvent { + key: event.key, + peers, + }); } } debug!("Fresh-write drainer shut down"); diff --git a/src/storage/handler.rs b/src/storage/handler.rs index bedffa11..3f6a4fa5 100644 --- a/src/storage/handler.rs +++ b/src/storage/handler.rs @@ -733,14 +733,11 @@ impl AntProtocol { // fall back to the original proof rather than dropping the // replication entirely. let proof = Self::strip_commitment_sidecars(proof); - // `request.content` is now `bytes::Bytes`; FreshWriteEvent - // still carries the chunk as `Vec` for compatibility - // with the replication wire format, so materialise once - // here. Done only on the success path, where storage has - // already accepted the chunk. + // Storage has already accepted the chunk on this path, so + // the event carries only the key; the replication drainer + // reads the chunk back when it is ready to send it. let event = FreshWriteEvent { key: address, - data: request.content.to_vec(), payment_proof: proof, }; if tx.send(event).is_err() { diff --git a/src/web_rtc.rs b/src/web_rtc.rs index c9c7b139..8c421ca3 100644 --- a/src/web_rtc.rs +++ b/src/web_rtc.rs @@ -22,6 +22,7 @@ use certificate::load_or_generate_certificate; use errors::{error_response, public_error, sanitize_response}; use evmlib::common::{Amount, TxHash}; use evmlib::{EncodedPeerId, PaymentQuote, ProofOfPayment, RewardsAddress}; +use futures::{stream::FuturesUnordered, StreamExt}; use parking_lot::{Mutex, RwLock}; use saorsa_core::identity::NodeIdentity; use saorsa_core::{AddressType, DHTNode, MultiAddr, P2PNode, PeerId}; @@ -39,7 +40,7 @@ use saorsa_transport::webrtc_direct::{ WebRtcAdmissionLimits, WebRtcDataChannel, WebRtcDiagnostics, WebRtcDiagnosticsSnapshot, WebRtcDirectConnection, WebRtcDirectListener, }; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet, VecDeque}; use std::future::Future; use std::net::{IpAddr, SocketAddr}; use std::path::{Path, PathBuf}; @@ -47,12 +48,16 @@ use std::str::FromStr; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime}; -use tokio::sync::{OwnedSemaphorePermit, Semaphore}; +use tokio::sync::{watch, OwnedSemaphorePermit, Semaphore}; use tokio::task::{JoinHandle, JoinSet}; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; const MAX_FIND_NODE_RESULTS: usize = 20; +// Advertised by HELLO. This is a protocol resource ceiling, not a throughput +// target; clients still obey their global memory/CPU admission budgets. +const MAX_CHANNEL_REQUESTS: usize = 4; +const RPC_MULTIPLEX_CAPABILITY: &str = "rpc-multiplex-4"; // Browser dials use a 10-second channel-open timeout. Give successful clients // modest server-side headroom while bounding associations that never open one. const FIRST_DATA_CHANNEL_TIMEOUT: Duration = Duration::from_secs(15); @@ -258,6 +263,7 @@ impl TrackedBytes { } struct SourceQuota { + bulk_responses: Arc, request_rate: Mutex, bytes: Arc, } @@ -288,6 +294,7 @@ struct ListenerResources { connection_limit: Arc, channel_limit: Arc, request_limit: Arc, + bulk_responses: Arc, request_tasks: Mutex, global_request_rate: Mutex, global_bytes: Arc, @@ -316,6 +323,9 @@ impl ListenerResources { connection_limit: Arc::new(Semaphore::new(config.max_connections)), channel_limit: Arc::new(Semaphore::new(config.max_channels)), request_limit: Arc::new(Semaphore::new(config.max_concurrent_requests)), + bulk_responses: Arc::new(Semaphore::new(bulk_response_slots( + config.max_in_flight_bytes, + ))), request_tasks: Mutex::new(TaskTracker::new()), global_request_rate: Mutex::new(RequestRateBucket::new(config.max_requests_per_second)), global_bytes: Arc::new(ByteBudget::with_rejections( @@ -374,6 +384,9 @@ impl ListenerResources { active_connections: 0, last_seen: Instant::now(), quota: Arc::new(SourceQuota { + bulk_responses: Arc::new(Semaphore::new(bulk_response_slots( + self.max_in_flight_bytes_per_ip, + ))), request_rate: Mutex::new(RequestRateBucket::new( self.max_requests_per_second_per_ip, )), @@ -1090,7 +1103,9 @@ async fn start_data_channel_task( if let Err(error) = channel.close().await { debug!(remote = %remote_addr, %error, "Failed to close excess DataChannel"); } - return Err("per-connection DataChannel capacity exhausted".to_string()); + // A reset channel's handler may still be scheduled for retirement. + // Reject only this excess stream, never its healthy sibling association. + return Ok(()); } let Ok(channel_permit) = Arc::clone(&resources.listener.channel_limit).try_acquire_owned() else { @@ -1141,110 +1156,114 @@ async fn handle_webrtc_channel( )); } - let mut pq_session = tokio::select! { + let pq_session = tokio::select! { biased; () = shutdown.cancelled() => return Ok(()), result = establish_pq_session(channel, &state, &resources) => result?, }; + // Receiving stays live while application work or response writes are pending. + // In particular, a channel reset releases its handler slot immediately; the + // listener's separately tracked storage/payment work retains its charges. + let session = Mutex::new(pq_session); let mut hello_completed = false; + let mut ids = HashSet::new(); + let mut pending = FuturesUnordered::new(); + let (activity, active_requests) = watch::channel(0usize); + let mut reading = Box::pin(read_webrtc_request( + channel, + &session, + &resources, + &active_requests, + )); + let mut writing = None; + let mut ready = VecDeque::::new(); loop { - let admitted_result = tokio::select! { - biased; - () = shutdown.cancelled() => return Ok(()), - result = read_webrtc_request( - channel, - &mut pq_session, - &resources, - ) => result, - }; - let admitted = match admitted_result { - Ok(request) => request, - Err(error) if is_quiet_channel_close(&error) => return Ok(()), - Err(error) => { - let response = Response::error(0, "invalid_request", error); - tokio::select! { - biased; - () = shutdown.cancelled() => return Ok(()), - result = write_webrtc_response( - channel, - &mut pq_session, - &response, - None, - &resources, - ) => result?, + if writing.is_none() { + if let Some(completed) = ready.pop_front() { + let id = completed.response.request_id; + if matches!(completed.response.body, ResponseBody::Hello { .. }) + && completed.response.status == ResponseStatus::Ok + { + hello_completed = true; } - return Ok(()); - } - }; - let request = &admitted.request; - if request.version != BROWSER_PROTOCOL_VERSION { - let response = Response::error( - request.request_id, - "unsupported_version", - format!( - "protocol version {} is unsupported; expected {BROWSER_PROTOCOL_VERSION}", - request.version - ), - ); - tokio::select! { - biased; - () = shutdown.cancelled() => return Ok(()), - result = write_webrtc_response( - channel, - &mut pq_session, - &response, - None, - &resources, - ) => result?, - } - continue; - } - - let is_hello = matches!(&request.body, RequestBody::Hello); - if !is_hello && !hello_completed { - let response = Response::error( - request.request_id, - "authentication_required", - "HELLO must initialize this encrypted WebRTC session first".to_string(), - ); - tokio::select! { - biased; - () = shutdown.cancelled() => return Ok(()), - result = write_webrtc_response( - channel, - &mut pq_session, - &response, - None, - &resources, - ) => result?, + let session = &session; + let resources = &resources; + // Only one complete encrypted frame is written at a time. AEAD + // sequence numbers follow wire order, not request completion order. + writing = Some(Box::pin(async move { + write_webrtc_response( + channel, + session, + &completed.response, + completed.content.as_ref(), + resources, + ) + .await?; + drop(completed); + Ok::<_, String>(id) + })); } - continue; - } - - let CompletedRequest { - response, - content, - _request_permit, - _in_flight_bytes, - _resources, - } = tokio::select! { - biased; - () = shutdown.cancelled() => return Ok(()), - result = start_request(admitted, Arc::clone(&state), Arc::clone(&resources)) => result?, - }; - if is_hello && matches!(&response.status, ResponseStatus::Ok) { - hello_completed = true; } tokio::select! { biased; () = shutdown.cancelled() => return Ok(()), - result = write_webrtc_response( - channel, - &mut pq_session, - &response, - content.as_ref(), - &resources, - ) => result?, + result = async { match writing.as_mut() { Some(writer) => writer.await, None => std::future::pending().await } }, if writing.is_some() => { + ids.remove(&result?); + activity.send_replace(ids.len()); + writing = None; + } + result = &mut reading => { + let admitted = match result { + Ok(request) => request, + Err(error) if is_quiet_channel_close(&error) => return Ok(()), + Err(error) => { + if ids.is_empty() { + let response = Response::error(0, "invalid_request", error); + tokio::select! { + () = shutdown.cancelled() => return Ok(()), + result = write_webrtc_response(channel, &session, &response, None, &resources) => result?, + } + return Ok(()); + } + return Err(error); + }, + }; + let request = &admitted.request; + if ids.len() >= MAX_CHANNEL_REQUESTS || !ids.insert(request.request_id) { + return Err("duplicate request ID or per-channel RPC capacity exhausted".into()); + } + activity.send_replace(ids.len()); + let error = if request.version != BROWSER_PROTOCOL_VERSION { + Some(Response::error(request.request_id, "unsupported_version", + format!("protocol version {} is unsupported; expected {BROWSER_PROTOCOL_VERSION}", request.version))) + } else if !hello_completed && !matches!(request.body, RequestBody::Hello) { + Some(Response::error(request.request_id, "authentication_required", + "HELLO must initialize this encrypted WebRTC session first")) + } else { None }; + let request_state = Arc::clone(&state); + let request_resources = Arc::clone(&resources); + pending.push(async move { + match error { + Some(response) => { + let AdmittedRequest { _request_permit: request_permit, _in_flight_bytes: in_flight_bytes, .. } = admitted; + Ok(CompletedRequest { + _bulk_permits: None, + response, content: None, + _request_permit: request_permit, + _in_flight_bytes: in_flight_bytes, + _resources: request_resources, + }) + }, + None => start_request(admitted, request_state, request_resources).await, + } + }); + reading = Box::pin(read_webrtc_request(channel, &session, &resources, &active_requests)); + } + completed = pending.next(), if !pending.is_empty() => { + let Some(completed) = completed else { continue }; + let completed = completed?; + ready.push_back(completed); + } } } } @@ -1302,6 +1321,8 @@ struct AdmittedRequest { } struct CompletedRequest { + // Retain preparation capacity until the response is written or discarded. + _bulk_permits: Option<(OwnedSemaphorePermit, OwnedSemaphorePermit)>, response: Response, content: Option, _request_permit: OwnedSemaphorePermit, @@ -1310,6 +1331,21 @@ struct CompletedRequest { _resources: Arc, } +// A GET may hold record bytes, encoded plaintext and ciphertext together. +// This queue complements exact byte accounting; it never raises those limits. +fn bulk_response_slots(bytes: usize) -> usize { + (bytes / (3 * ant_protocol::MAX_WIRE_MESSAGE_SIZE + 2 * MAX_BROWSER_HEADER_BYTES)).max(1) +} + +fn request_is_get(request: &Request, content: &[u8]) -> bool { + matches!(request.body, RequestBody::GetChunk { .. }) + || (matches!(request.body, RequestBody::ChunkProtocol) + // A canonical binary GET is small. Avoid decoding/copying PUTs just + // to classify scheduling; the ordinary decoder still validates all. + && content.len() <= 128 + && ChunkMessage::decode(content).is_ok_and(|message| matches!(message.body, ChunkMessageBody::GetRequest(_)))) +} + async fn start_request( admitted: AdmittedRequest, state: Arc, @@ -1323,8 +1359,25 @@ async fn start_request( _request_permit: request_permit, _in_flight_bytes: in_flight_bytes, } = admitted; + // Reserve enough preparation capacity for one complete response to + // make progress under the existing byte limits. Acquire source first: + // queued work from one source cannot occupy every global bulk slot. + let bulk_permits = if request_is_get(&request, &content) { + let source = Arc::clone(&resources.source.bulk_responses) + .acquire_owned() + .await + .map_err(|_| "bulk response admission closed".to_string())?; + let global = Arc::clone(&resources.listener.bulk_responses) + .acquire_owned() + .await + .map_err(|_| "bulk response admission closed".to_string())?; + Some((source, global)) + } else { + None + }; let (response, content) = process_request(request, content, &state, &resources).await?; Ok(CompletedRequest { + _bulk_permits: bulk_permits, response, content, _request_permit: request_permit, @@ -1340,10 +1393,20 @@ async fn start_request( async fn read_webrtc_request( channel: &WebRtcDataChannel, - pq_session: &mut PqSession, + pq_session: &Mutex, resources: &ConnectionResources, + active_requests: &watch::Receiver, ) -> ServerResult { - let first_message = receive_first_message(channel, "request idle timeout").await?; + let first_message = receive_while_active( + channel.receive(), + active_requests.clone(), + REQUEST_IDLE_TIMEOUT, + ) + .await? + .map_err(|error| format!("DataChannel message read failed: {error}"))?; + if first_message.is_empty() { + return Err("DataChannel closed".into()); + } // Admission happens as soon as a client starts a frame. Idle persistent // channels consume neither request-rate tokens nor request worker slots. let request_permit = resources.try_admit_request()?; @@ -1361,6 +1424,7 @@ async fn read_webrtc_request( // second buffer before asking the cryptographic layer to allocate it. encrypted.reservation.try_grow(encrypted_len)?; let frame = pq_session + .lock() .open(&encrypted.bytes) .map_err(|error| format!("PQ session: {error}"))?; let TrackedBytes { @@ -1389,6 +1453,29 @@ async fn read_webrtc_request( }) } +// Observe resets even while work is pending. Only an idle channel spends the +// inactivity budget; a slow disk operation or permitted bulk transfer must not +// be interrupted by the request-idle timer. Keep the receive future pinned so +// activity changes never discard a partially polled transport read. +async fn receive_while_active( + receive: impl Future, + mut activity: watch::Receiver, + idle_timeout: Duration, +) -> ServerResult { + tokio::pin!(receive); + loop { + let idle = *activity.borrow_and_update() == 0; + tokio::select! { + biased; + result = &mut receive => return Ok(result), + changed = activity.changed() => { + changed.map_err(|_| "channel activity closed".to_string())?; + } + () = tokio::time::sleep(idle_timeout), if idle => return Err("request idle timeout".into()), + } + } +} + async fn receive_first_message( channel: &WebRtcDataChannel, idle_timeout_message: &str, @@ -1489,7 +1576,7 @@ async fn read_pq_payload_after_first( async fn write_webrtc_response( channel: &WebRtcDataChannel, - pq_session: &mut PqSession, + pq_session: &Mutex, response: &Response, content: Option<&TrackedBytes>, resources: &ConnectionResources, @@ -1511,6 +1598,7 @@ async fn write_webrtc_response( .ok_or_else(|| "encrypted response length overflow".to_string())?; reservation.try_grow(encrypted_len)?; let encrypted = pq_session + .lock() .seal(&plaintext) .map_err(|error| format!("PQ session: {error}"))?; drop(plaintext); @@ -1690,6 +1778,7 @@ fn hello_response(request_id: u64, state: &ServerState) -> Response { payment: state.payment.clone(), capabilities: vec![ "chunk_protocol".into(), + RPC_MULTIPLEX_CAPABILITY.into(), "find_node".into(), ant_protocol::transport::ADDRESS_V2_CAPABILITY.into(), "get_chunk".into(), @@ -2399,6 +2488,43 @@ mod tests { .expect("released prefix slot"); } + #[tokio::test] + async fn active_requests_suspend_idle_timeout_but_not_channel_closure() { + let (activity, receiver) = watch::channel(1usize); + let (closed, receive) = tokio::sync::oneshot::channel::<()>(); + let mut waiting = Box::pin(receive_while_active( + receive, + receiver, + Duration::from_millis(15), + )); + assert!( + tokio::time::timeout(Duration::from_millis(40), &mut waiting) + .await + .is_err() + ); + closed.send(()).unwrap(); + assert!(waiting.await.unwrap().is_ok()); + drop(activity); + } + + #[tokio::test] + async fn idle_budget_starts_after_the_last_active_response() { + let (activity, receiver) = watch::channel(1usize); + let waiting = receive_while_active( + std::future::pending::<()>(), + receiver, + Duration::from_millis(15), + ); + tokio::pin!(waiting); + assert!( + tokio::time::timeout(Duration::from_millis(30), &mut waiting) + .await + .is_err() + ); + activity.send_replace(0); + assert_eq!(waiting.await.unwrap_err(), "request idle timeout"); + } + #[test] fn request_token_bucket_refills_without_growing_state() { let mut bucket = RequestRateBucket::new(2); diff --git a/tests/webrtc_direct_devnet.rs b/tests/webrtc_direct_devnet.rs index 77556433..033124d2 100644 --- a/tests/webrtc_direct_devnet.rs +++ b/tests/webrtc_direct_devnet.rs @@ -41,6 +41,23 @@ struct MockChainRpc { #[allow(clippy::await_holding_lock)] // Deliberately stall a blocking disk write across awaits. async fn disconnected_put_retains_admission_and_allows_small_binary_quotes( ) -> Result<(), Box> { + blocked_put_lifecycle(false).await +} + +#[cfg(feature = "test-utils")] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial_test::serial] +async fn multiplexed_get_completes_before_blocked_put() -> Result<(), Box> { + blocked_put_lifecycle(true).await +} + +#[cfg(feature = "test-utils")] +#[allow( + clippy::await_holding_lock, + clippy::future_not_send, + clippy::too_many_lines +)] +async fn blocked_put_lifecycle(probe_multiplex: bool) -> Result<(), Box> { use std::time::Duration; let rpc = @@ -84,6 +101,15 @@ async fn disconnected_put_retains_admission_and_allows_small_binary_quotes( } }).await?; + // This GET must finish on the same authenticated channel while the + // earlier PUT is blocked. Replies are correlated by ID, not send order. + if probe_multiplex { + let (reply, _) = tokio::time::timeout(Duration::from_secs(2), client.rpc( + json!({"version":BROWSER_PROTOCOL_VERSION,"request_id":99,"type":"get_chunk","address":hex::encode([9;32])}), &[] + )).await??; + assert_eq!(reply["request_id"], 99); + } + // Under the old fixed 10 MiB allowance, this second client's QUOTE // closed its channel even though both operations fit the source budget. let mut other = BrowserRpcClient::connect(&endpoint).await?; @@ -106,6 +132,15 @@ async fn disconnected_put_retains_admission_and_allows_small_binary_quotes( }; assert!(matches!(response, ant_protocol::ChunkMessageBody::QuoteResponse(ant_protocol::ChunkQuoteResponse::Error(ref error)) if *error == expected)); } + // Reset only the busy stream, leaving the association itself alive. + // Its handler slot must retire before the blocked application work. + client.client.data_channel().close().await?; + tokio::time::timeout(Duration::from_secs(2), async { + while diagnostics.snapshot().active_channels > 1 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await?; + assert_eq!(diagnostics.snapshot().active_requests, 1); client.close().await?; other.close().await?; tokio::time::timeout(Duration::from_secs(20), async { @@ -147,6 +182,86 @@ async fn disconnected_put_retains_admission_and_allows_small_binary_quotes( Ok(()) } +/// Four full GETs must queue within the default 16 MiB source budget, retain +/// authenticated frame order, and return every independent request correctly. +#[cfg(feature = "test-utils")] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial_test::serial] +async fn multiplexed_full_chunks_respect_source_budget() -> Result<(), Box> { + use std::time::Duration; + let rpc = + MockChainRpc::new(json!({"jsonrpc":"2.0", "id":1, "result":"0x7a69"}).to_string()).await?; + let temp = tempfile::tempdir()?; + let mut config = DevnetConfig::minimal(); + config.node_count = 2; + config.bootstrap_count = 1; + config.base_port = 0; + config.webrtc_direct = true; + config.data_dir = temp.path().join("bulk-devnet"); + config.spawn_delay = Duration::from_millis(20); + config.evm_network = Some(rpc.network()); + let mut devnet = Devnet::new(config).await?; + devnet.start().await?; + let (protocol, diagnostics) = devnet.test_browser_node(0).ok_or("missing node")?; + let bytes = vec![42; ant_protocol::MAX_CHUNK_SIZE]; + let address = *blake3::hash(&bytes).as_bytes(); + protocol.storage().put(&address, &bytes).await?; + let mut client = BrowserRpcClient::connect(&devnet.browser_endpoints()[0].endpoint).await?; + let (hello, _) = client + .rpc( + json!({"version":BROWSER_PROTOCOL_VERSION,"request_id":1,"type":"hello"}), + &[], + ) + .await?; + assert!(hello["capabilities"] + .as_array() + .ok_or("capabilities")? + .contains(&json!("rpc-multiplex-4"))); + for id in 2..6 { + let message = ant_protocol::ChunkMessage { + request_id: id, + body: ant_protocol::ChunkMessageBody::GetRequest(ant_protocol::ChunkGetRequest::new( + address, + )), + } + .encode()?; + let mut frame = serde_json::to_vec( + &json!({"version":BROWSER_PROTOCOL_VERSION,"request_id":id,"type":"chunk_protocol","content_length":message.len()}), + )?; + frame.extend_from_slice(&message); + send_pq_payload( + client.client.data_channel(), + &client.pq_session.seal(&frame)?, + ) + .await?; + } + let mut ids = std::collections::HashSet::new(); + for _ in 0..4 { + let encrypted = tokio::time::timeout( + Duration::from_secs(10), + read_pq_payload( + client.client.data_channel(), + saorsa_transport::webrtc::MAX_BROWSER_FRAME_BYTES + PQ_ENCRYPTED_OVERHEAD_BYTES, + ), + ) + .await??; + let plaintext = client.pq_session.open(&encrypted)?; + let frame = saorsa_transport::webrtc::parse_response_frame(&plaintext)?; + assert!(ids.insert(frame.header.request_id)); + let response = ant_protocol::ChunkMessage::decode(&frame.content)?; + assert_eq!(response.request_id, frame.header.request_id); + assert!( + matches!(response.body, ant_protocol::ChunkMessageBody::GetResponse( + ant_protocol::ChunkGetResponse::Success { content, .. }) if content.as_slice() == bytes.as_slice()) + ); + } + assert_eq!(ids, (2..6).collect()); + assert_eq!(diagnostics.snapshot().byte_rejections, 0); + client.close().await?; + devnet.shutdown().await?; + Ok(()) +} + impl MockChainRpc { async fn new(body: String) -> io::Result { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;