perf(webrtc): serve bounded concurrent browser RPCs (rpc-multiplex-4)

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.

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

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<u8>,
     /// Serialized proof-of-payment.
     pub payment_proof: Vec<u8>,
 }
 
+/// 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<u8>,
+    _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<PaidList>,
     config: &ReplicationConfig,
     send_semaphore: &Arc<Semaphore>,
+    pending_offer: OwnedSemaphorePermit,
 ) -> Vec<PeerId> {
     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,

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<Semaphore>,
+    /// Bounds how many encoded fresh offers can wait behind `send_semaphore`;
+    /// see [`MAX_PENDING_FRESH_OFFERS`].
+    pending_offer_semaphore: Arc<Semaphore>,
     /// 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");

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<u8>` 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() {