From 1905e107c524058bfc4e586755f61503894ed773 Mon Sep 17 00:00:00 2001
From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com>
Date: Mon, 21 Sep 2026 19:16:21 +0200
Subject: [PATCH 1/6] fix(replication): bound encoded fresh offers waiting
 behind send permits

Under sustained client writes on a real network, nodes grew to 1-2 GiB
within an hour. Heap profiles on the testnet attribute ~80% of live
memory to encoded FreshReplicationOffer messages queued in the
fresh-write drainer: every accepted PUT was encoded immediately (chunk
plus proof, ~4-5 MiB) and its per-peer send tasks then waited for one
of MAX_CONCURRENT_REPLICATION_SENDS (3) permits while pinning that
buffer. When WAN sends hold permits longer than writes arrive, nothing
bounded the backlog, so the number of encoded offers kept growing.

- FreshWriteEvent no longer carries the chunk bytes; the chunk is on
  disk already and the drainer reads it back when it is ready to send.
- The drainer acquires a pending-offer permit (MAX_PENDING_FRESH_OFFERS)
  before reading and encoding, and the permit lives with the encoded
  buffer until the last per-peer send drops it. A backlog now waits as
  small queued events instead of chunk-sized buffers.
- The direct ReplicationEngine::replicate_fresh entry point takes the
  same permit so tests and callers share the bound.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 src/replication/config.rs | 11 ++++++
 src/replication/fresh.rs  | 42 ++++++++++++++-------
 src/replication/mod.rs    | 79 ++++++++++++++++++++++++++++-----------
 src/storage/handler.rs    |  9 ++---
 4 files changed, 100 insertions(+), 41 deletions(-)

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<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,
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<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");
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<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() {

From 90f25135e3438bd5ebf83dbe81beca5f2e7d35e6 Mon Sep 17 00:00:00 2001
From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com>
Date: Tue, 22 Sep 2026 08:58:28 +0200
Subject: [PATCH 2/6] perf(replication): cut copies of fresh offers on the send
 path

Two of the copies each queued fresh offer carried were avoidable inside
this crate:

- The chunk read from storage now moves into FreshReplicationOffer
  instead of being copied, and the offer is dropped as soon as it has
  been encoded, so only the encoded bytes stay alive while sends queue.
- ReplicationMessage::encode serializes into a buffer sized from
  postcard's serialized_size. A doubling Vec left chunk-sized messages
  with up to twice their length in capacity, retained by every queued
  offer for as long as it waited for a send permit.

The remaining copies per in-flight send live in saorsa-core (payload
clone per channel attempt, signing re-serialization, wire frame) and
saorsa-transport (stream buffer copy in SendStream::write).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 src/replication/fresh.rs    | 13 +++++++++----
 src/replication/mod.rs      |  4 ++--
 src/replication/protocol.rs | 26 +++++++++++++++++++++++++-
 3 files changed, 36 insertions(+), 7 deletions(-)

diff --git a/src/replication/fresh.rs b/src/replication/fresh.rs
index d41e283d..9d05cbeb 100644
--- a/src/replication/fresh.rs
+++ b/src/replication/fresh.rs
@@ -58,11 +58,12 @@ struct EncodedOffer {
 /// in-flight concurrently across the entire replication engine, preventing
 /// 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.
+/// encoded offer until the last per-peer send finishes. `data` is taken by
+/// value so the chunk moves into the offer instead of being copied.
 #[allow(clippy::too_many_arguments)]
 pub async fn replicate_fresh(
     key: &XorName,
-    data: &[u8],
+    data: Vec<u8>,
     proof_of_payment: &[u8],
     p2p_node: &Arc<P2PNode>,
     paid_list: &Arc<PaidList>,
@@ -92,7 +93,7 @@ pub async fn replicate_fresh(
 
     let offer = FreshReplicationOffer {
         key: *key,
-        data: data.to_vec(),
+        data,
         proof_of_payment: proof_of_payment.to_vec(),
     };
     let request_id = rand::thread_rng().gen::<u64>();
@@ -101,7 +102,11 @@ pub async fn replicate_fresh(
         body: ReplicationMessageBody::FreshReplicationOffer(offer),
     };
 
-    let Ok(encoded) = offer_msg.encode() else {
+    let encoded = offer_msg.encode();
+    // Only the encoded bytes are needed from here on; release the chunk now
+    // rather than holding it alongside the encoding while sends are queued.
+    drop(offer_msg);
+    let Ok(encoded) = encoded else {
         warn!(
             "Failed to encode FreshReplicationOffer for {}",
             hex::encode(key),
diff --git a/src/replication/mod.rs b/src/replication/mod.rs
index 9cb2263e..29c52f99 100644
--- a/src/replication/mod.rs
+++ b/src/replication/mod.rs
@@ -2382,7 +2382,7 @@ impl ReplicationEngine {
         };
         let peers = fresh::replicate_fresh(
             key,
-            data,
+            data.to_vec(),
             proof_of_payment,
             &self.p2p_node,
             &self.paid_list,
@@ -2449,7 +2449,7 @@ impl ReplicationEngine {
                 };
                 let peers = fresh::replicate_fresh(
                     &event.key,
-                    &data,
+                    data,
                     &event.payment_proof,
                     &p2p,
                     &paid_list,
diff --git a/src/replication/protocol.rs b/src/replication/protocol.rs
index 3ef7d0fa..40935fdb 100644
--- a/src/replication/protocol.rs
+++ b/src/replication/protocol.rs
@@ -46,7 +46,13 @@ impl ReplicationMessage {
     /// Returns [`ReplicationProtocolError::SerializationFailed`] if postcard
     /// serialization fails.
     pub fn encode(&self) -> Result<Vec<u8>, ReplicationProtocolError> {
-        let bytes = postcard::to_stdvec(self)
+        // Size the buffer exactly up front. Chunk-carrying bodies run to
+        // several MiB, and a growing `Vec` would otherwise end up with up to
+        // twice the needed capacity, retained for as long as the encoded
+        // message is queued for sending.
+        let size = postcard::experimental::serialized_size(self)
+            .map_err(|e| ReplicationProtocolError::SerializationFailed(e.to_string()))?;
+        let bytes = postcard::to_extend(self, Vec::with_capacity(size))
             .map_err(|e| ReplicationProtocolError::SerializationFailed(e.to_string()))?;
 
         // The same family ceiling the decoder applies, from the same table and
@@ -2429,6 +2435,24 @@ mod tests {
         assert_eq!(decoded.request_id, 7);
     }
 
+    #[test]
+    fn encode_allocates_exactly_the_serialized_size() {
+        // A chunk-sized offer must not carry growth slack: the encoded buffer is
+        // shared by every per-peer send task for as long as it is queued.
+        let msg = ReplicationMessage {
+            request_id: 7,
+            body: ReplicationMessageBody::FreshReplicationOffer(FreshReplicationOffer {
+                key: [3; 32],
+                data: vec![0xAB; 3 * 1024 * 1024 + 123],
+                proof_of_payment: vec![1, 2, 3],
+            }),
+        };
+        let encoded = msg.encode().unwrap();
+        assert_eq!(encoded.capacity(), encoded.len());
+        let decoded = ReplicationMessage::decode(&encoded).unwrap();
+        assert_eq!(decoded.request_id, 7);
+    }
+
     #[test]
     fn encode_rejects_oversized_message() {
         // Build a message whose serialized form exceeds the limit.

From f55b7df8d69b60b34b60b72e0cb4b57a4dd967ac Mon Sep 17 00:00:00 2001
From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com>
Date: Tue, 22 Sep 2026 09:35:07 +0200
Subject: [PATCH 3/6] perf(replication): share fresh offers with the transport
 as Bytes

With saorsa-core accepting `impl Into<Bytes>` on `send_message`, the
encoded fresh offer is now held as `Bytes` and each per-peer send
attempt hands out a reference-counted handle instead of cloning the
multi-MiB buffer. Together with the exactly-sized frame and the
transport's owned-buffer write, an in-flight send now costs one frame
instead of the previous four copies.

Adds ADR-0016 describing the bounded fresh-offer backlog and the
copy-free send path across ant-node, saorsa-core and saorsa-transport.

Pins: ant-protocol ac25b717, saorsa-core e1f1ef92, saorsa-transport
3bd451d2 (all on perf/replication-send-path).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 Cargo.lock                                    |  16 +--
 Cargo.toml                                    |   6 +-
 ...ounded-fresh-offers-and-copy-free-sends.md | 128 ++++++++++++++++++
 src/replication/fresh.rs                      |  16 ++-
 4 files changed, 148 insertions(+), 18 deletions(-)
 create mode 100644 docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md

diff --git a/Cargo.lock b/Cargo.lock
index 3ee7e5a8..9f5d2fd9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -883,7 +883,7 @@ dependencies = [
 [[package]]
 name = "ant-protocol"
 version = "2.4.0"
-source = "git+https://github.com/WithAutonomi/ant-protocol?rev=1764950d7af880aa13af679ac8c2d0fcbcde0776#1764950d7af880aa13af679ac8c2d0fcbcde0776"
+source = "git+https://github.com/WithAutonomi/ant-protocol?rev=ac25b717d99e70468d934b66c32d6e6219a37350#ac25b717d99e70468d934b66c32d6e6219a37350"
 dependencies = [
  "blake3",
  "bytes",
@@ -3294,7 +3294,7 @@ dependencies = [
  "libc",
  "percent-encoding",
  "pin-project-lite",
- "socket2 0.6.5",
+ "socket2 0.5.10",
  "tokio",
  "tower-service",
  "tracing",
@@ -4655,7 +4655,7 @@ dependencies = [
  "quinn-udp 0.5.15",
  "rustc-hash",
  "rustls",
- "socket2 0.6.5",
+ "socket2 0.5.10",
  "thiserror 2.0.20",
  "tokio",
  "tracing",
@@ -4694,7 +4694,7 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2 0.6.5",
+ "socket2 0.5.10",
  "tracing",
  "windows-sys 0.61.2",
 ]
@@ -4707,7 +4707,7 @@ checksum = "76150b617afc75e6e21ac5f39bc196e80b65415ae48d62dbef8e2519d040ce42"
 dependencies = [
  "cfg_aliases",
  "libc",
- "socket2 0.6.5",
+ "socket2 0.5.10",
  "tracing",
  "windows-sys 0.61.2",
 ]
@@ -5324,7 +5324,7 @@ dependencies = [
 [[package]]
 name = "saorsa-core"
 version = "0.27.4"
-source = "git+https://github.com/WithAutonomi/saorsa-core?rev=02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c#02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c"
+source = "git+https://github.com/WithAutonomi/saorsa-core?rev=e1f1ef92aacb0dab60e6c3dded5da8bae3b63d07#e1f1ef92aacb0dab60e6c3dded5da8bae3b63d07"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -5392,7 +5392,7 @@ dependencies = [
 [[package]]
 name = "saorsa-transport"
 version = "0.36.4"
-source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=3ed66a9c0880583ceab1e1550dce756e2716d111#3ed66a9c0880583ceab1e1550dce756e2716d111"
+source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=3bd451d2d2c86b01df1dd355a88e81bcb854d815#3bd451d2d2c86b01df1dd355a88e81bcb854d815"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -7121,7 +7121,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.48.0",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index f3610c9f..7a4be10a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -231,9 +231,9 @@ webrtc-direct = [
 [patch.crates-io]
 saorsa-pqc = { git = "https://github.com/saorsa-labs/saorsa-pqc", rev = "29a2b272c4ee8b72c17102da00033dbfe1ddcd37" }
 evmlib = { git = "https://github.com/WithAutonomi/evmlib", rev = "ccd65f18bf3750af8b8c3cfe07548bc77e3c368d" }
-ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "1764950d7af880aa13af679ac8c2d0fcbcde0776" }
-saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c" }
-saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "3ed66a9c0880583ceab1e1550dce756e2716d111" }
+ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "ac25b717d99e70468d934b66c32d6e6219a37350" }
+saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "e1f1ef92aacb0dab60e6c3dded5da8bae3b63d07" }
+saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "3bd451d2d2c86b01df1dd355a88e81bcb854d815" }
 
 [profile.release]
 lto = true
diff --git a/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
new file mode 100644
index 00000000..ec20f974
--- /dev/null
+++ b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
@@ -0,0 +1,128 @@
+# ADR-0016: Bounded fresh-replication offers and copy-free message sends
+
+- **Status:** Proposed
+- **Date:** 2026-09-22
+- **Decision owners:** <pending>
+- **Reviewers:** <pending>
+- **Supersedes:** none
+- **Superseded by:** none
+- **Related:** [ADR-0003](ADR-0003-full-node-detection-and-eviction.md)
+  (best-effort fresh delivery and possession checks),
+  [ADR-0005](ADR-0005-replication-repair-hardening.md),
+  ant-node `perf/replication-send-path`, saorsa-core `perf/replication-send-path`,
+  saorsa-transport `perf/replication-send-path`,
+  `ant-testnet/state/comparisons/web-support-memory-diag-0921/` (heap profiles
+  and per-minute live/RSS reports from the diagnosis)
+
+## Context
+
+Under sustained client uploads on a 60-node DigitalOcean testnet, individual
+nodes grew from ~100 MiB to 1–2 GiB of resident memory within an hour and
+kept growing for as long as writes continued. Heap profiles taken on the
+running nodes attributed roughly 80% of live memory to one site: encoded
+`FreshReplicationOffer` messages queued by the fresh-write drainer.
+
+The mechanism was structural rather than a leak:
+
+- Every accepted PUT was pushed to the drainer with the full chunk, and
+  `replicate_fresh` encoded the offer (chunk plus proof, up to ~4–5 MiB)
+  immediately, before any send permit was held.
+- One send task per close-group peer then waited for one of
+  `MAX_CONCURRENT_REPLICATION_SENDS` (3) permits while pinning that encoded
+  buffer. Nothing bounded how many chunks could be waiting in that state.
+- On a real network each send holds its permit for seconds (QUIC delivery
+  acknowledgement, retries, unreachable NAT peers), so a write rate above
+  the send rate grew the queue without limit. Loopback devnets never showed
+  it because sends complete instantly.
+
+Once the backlog was bounded, the profile showed the remaining cost per
+in-flight send: the same frame existed as the caller's serialized message
+*and* as the QUIC stream's copy for the whole transfer, plus transient
+copies made while framing (payload clone per channel attempt, owned wire
+message for signing, and a doubling `Vec` that left chunk-sized frames with
+up to twice their length in capacity).
+
+## Decision Drivers
+
+- Node memory must stay bounded under any client write rate; replication
+  may be delayed by backpressure but must not be dropped.
+- The change must not alter the wire format, storage format or payment
+  logic, so it can ship as a behavioural fix.
+- Existing callers of the send APIs in saorsa-core and saorsa-transport must
+  keep compiling and behaving the same.
+
+## Considered Options
+
+1. Bound the fresh-write channel and drop or block PUT handling when full.
+   Rejected: either silently loses replication or blocks client responses
+   on network conditions.
+2. Raise `MAX_CONCURRENT_REPLICATION_SENDS`. Rejected: only moves the
+   knee of the curve and increases bandwidth pressure on home links; the
+   queue behind the permits would still be unbounded.
+3. Keep events small and take a bounded permit before materialising an
+   offer; separately remove the avoidable copies on the send path. Chosen.
+
+## Decision
+
+We will bound the number of encoded fresh offers that can exist at once and
+make the send path hand a single owned buffer down to the QUIC stream:
+
+- `FreshWriteEvent` carries only the key and the payment proof. The drainer
+  acquires a `MAX_PENDING_FRESH_OFFERS` (8) permit before it reads the chunk
+  back from storage and encodes it; the permit lives with the encoded offer
+  until the last per-peer send drops it. A backlog therefore waits as small
+  queued events, and at most ~40 MiB of encoded offers exist per node.
+- The chunk moves into the offer rather than being copied, and
+  `ReplicationMessage::encode` serializes into an exactly-sized buffer.
+- The encoded offer is shared as `Bytes`; saorsa-core's `send_message`
+  accepts `impl Into<Bytes>`, frames the payload through a borrowing
+  `WireMessageRef` (byte-identical to `WireMessage` on the wire) into an
+  exactly-sized frame, and passes that frame as `Bytes` to
+  saorsa-transport's new `send_bytes`, where the QUIC stream takes ownership
+  via `write_chunks` instead of copying it.
+
+## Consequences
+
+### Positive
+
+- Memory under write load is bounded by configuration: pending offers plus
+  the three in-flight sends, each held once, instead of growing with the
+  backlog. On the diagnostic fleets peak live memory fell from 1068 MiB to
+  298 MiB (mimalloc build) and from 674 MiB to 262 MiB (jemalloc build)
+  after the backpressure change alone.
+- Every large send node-wide (chunk GET responses included) stops paying
+  for a second copy of its frame during the transfer.
+- No wire, storage or API break: `send(&[u8])` remains and copies once as
+  before; `Vec<u8>` callers of `send_message` convert without copying.
+
+### Negative / Trade-offs
+
+- Replication of a burst of writes is spread out in time rather than
+  encoded eagerly; the delayed possession check is scheduled after each
+  offer's sends are dispatched, so it shifts by the same amount.
+- The drainer re-reads each chunk from disk when its permit arrives, one
+  extra read per accepted write.
+
+### Neutral / Operational
+
+- `MAX_PENDING_FRESH_OFFERS` and `MAX_CONCURRENT_REPLICATION_SENDS` are
+  the two knobs; raising the first trades memory for burst absorption.
+- The signing step still serializes the payload once to produce the signed
+  bytes; changing that would alter the signature input and is out of scope.
+
+## Validation
+
+- Unit tests: exact-capacity encoding of chunk-sized offers (ant-node) and
+  byte-for-byte equivalence of `WireMessageRef` with `WireMessage`
+  (saorsa-core); replication unit and e2e fresh-replication scenarios pass.
+- Testnet evidence (2026-09-21): with the backpressure change, the node
+  that had reached 1051 MiB live memory stayed flat at 0.0 MiB/min with a
+  150 MiB peak, and the worst bootstrap's queued offers dropped from 101
+  (463 MiB) to 6 (17.7 MiB) in heap profiles.
+- Review trigger: any change to fresh replication fan-out, send permits, or
+  the wire-message framing must re-run the memory diagnostics under
+  sustained uploads and confirm live memory stays bounded.
+
+## Notes for AI-assisted work
+
+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.
diff --git a/src/replication/fresh.rs b/src/replication/fresh.rs
index 9d05cbeb..8fd3cdc6 100644
--- a/src/replication/fresh.rs
+++ b/src/replication/fresh.rs
@@ -8,6 +8,7 @@
 use std::sync::Arc;
 
 use crate::logging::{debug, warn};
+use bytes::Bytes;
 use rand::Rng;
 use saorsa_core::identity::PeerId;
 use saorsa_core::P2PNode;
@@ -41,9 +42,11 @@ pub struct FreshWriteEvent {
 ///
 /// 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`.
+/// can wait behind the send permits at `MAX_PENDING_FRESH_OFFERS`. The bytes
+/// are shared with the transport as well: each send attempt hands out a
+/// reference-counted handle rather than a copy.
 struct EncodedOffer {
-    bytes: Vec<u8>,
+    bytes: Bytes,
     _pending: OwnedSemaphorePermit,
 }
 
@@ -113,12 +116,11 @@ pub async fn replicate_fresh(
         );
         return Vec::new();
     };
-    // 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. The pending-offer
-    // permit travels with the buffer.
+    // One encoded copy serves every per-peer send task and every retry; the
+    // transport borrows it through `Bytes` instead of taking a copy. The
+    // pending-offer permit travels with the buffer.
     let encoded = Arc::new(EncodedOffer {
-        bytes: encoded,
+        bytes: Bytes::from(encoded),
         _pending: pending_offer,
     });
     for peer in &target_peers {

From 07a11e091fc1c4d0ef4883d077a4e0dc7836f43d Mon Sep 17 00:00:00 2001
From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com>
Date: Tue, 22 Sep 2026 15:09:20 +0200
Subject: [PATCH 4/6] fix(replication): send PaidNotify before waiting for an
 offer permit

PaidNotify carries the paid-list evidence the paid close group needs to
repair a key later. Since the pending-offer permit was introduced it was
sent from replicate_fresh, i.e. only after the drainer had waited for a
permit, so a chunk backlog also delayed the evidence. Send it as soon as
a write is dequeued (and from the direct replicate_fresh entry point),
before any permit wait; only the bulk chunk offers are back-pressured.

Nothing is dropped by the permit: the fresh-write queue is unbounded and
FIFO, permits are released whenever a send terminates, and each offer
keeps the same fan-out, retries and delayed possession check. ADR-0016
now says so explicitly and records the measured download-latency cost.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 ...ounded-fresh-offers-and-copy-free-sends.md | 20 ++++++++++++----
 src/replication/fresh.rs                      | 23 ++++++++++---------
 src/replication/mod.rs                        |  5 ++++
 3 files changed, 32 insertions(+), 16 deletions(-)

diff --git a/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
index ec20f974..fddee579 100644
--- a/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
+++ b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
@@ -68,10 +68,15 @@ We will bound the number of encoded fresh offers that can exist at once and
 make the send path hand a single owned buffer down to the QUIC stream:
 
 - `FreshWriteEvent` carries only the key and the payment proof. The drainer
-  acquires a `MAX_PENDING_FRESH_OFFERS` (8) permit before it reads the chunk
-  back from storage and encodes it; the permit lives with the encoded offer
-  until the last per-peer send drops it. A backlog therefore waits as small
-  queued events, and at most ~40 MiB of encoded offers exist per node.
+  sends `PaidNotify` to the paid close group as soon as it dequeues a write,
+  so the paid-list evidence that later repair depends on is never delayed by
+  chunk back-pressure. It then acquires a `MAX_PENDING_FRESH_OFFERS` (8)
+  permit before it reads the chunk back from storage and encodes it; the
+  permit lives with the encoded offer until the last per-peer send drops it.
+  A backlog therefore waits as small queued events, and at most ~40 MiB of
+  encoded offers exist per node. Nothing is dropped: the queue is unbounded
+  and FIFO, and every offer is still dispatched with the same fan-out,
+  retries and delayed possession check.
 - The chunk moves into the offer rather than being copied, and
   `ReplicationMessage::encode` serializes into an exactly-sized buffer.
 - The encoded offer is shared as `Bytes`; saorsa-core's `send_message`
@@ -99,7 +104,12 @@ make the send path hand a single owned buffer down to the QUIC stream:
 
 - Replication of a burst of writes is spread out in time rather than
   encoded eagerly; the delayed possession check is scheduled after each
-  offer's sends are dispatched, so it shifts by the same amount.
+  offer's sends are dispatched, so it shifts by the same amount. A chunk
+  fetched seconds after its upload can therefore have fewer replicas than
+  before (the 2026-09-22 comparison measured downloads of just-uploaded
+  files 8% slower). Paid-list evidence is not affected, and the previous
+  unbounded fan-out lost that evidence outright under load (2,795
+  "paid notify dropped at admission" in one hour on the baseline fleet).
 - The drainer re-reads each chunk from disk when its permit arrives, one
   extra read per accepted write.
 
diff --git a/src/replication/fresh.rs b/src/replication/fresh.rs
index 8fd3cdc6..9414a98f 100644
--- a/src/replication/fresh.rs
+++ b/src/replication/fresh.rs
@@ -53,9 +53,12 @@ struct EncodedOffer {
 /// Execute fresh replication for a newly accepted record.
 ///
 /// Sends fresh offers to close group members (with bounded delivery retries,
-/// ADR-0003) and `PaidNotify` to `PaidCloseGroup`. Returns the close-group
-/// peers responsible for the key (excluding self) so the caller can schedule
-/// the delayed possession check; `PaidNotify` remains fire-and-forget.
+/// ADR-0003). Returns the close-group peers responsible for the key
+/// (excluding self) so the caller can schedule the delayed possession check.
+/// `PaidNotify` is deliberately not sent here: it carries the paid-list
+/// evidence peers need to repair the key later, so callers send it with
+/// [`send_paid_notify`] as soon as the write is accepted, before waiting for
+/// a pending-offer permit.
 ///
 /// The `send_semaphore` limits how many outbound chunk transfers can be
 /// in-flight concurrently across the entire replication engine, preventing
@@ -166,13 +169,8 @@ pub async fn replicate_fresh(
         });
     }
 
-    // Rule 7-8: Send PaidNotify to every member of PaidCloseGroup(K).
-    // PaidNotify messages are small metadata (no chunk data), so they don't
-    // need semaphore gating.
-    send_paid_notify(key, proof_of_payment, p2p_node, config).await;
-
     debug!(
-        "Fresh replication initiated for {} to {} peers + PaidNotify",
+        "Fresh replication initiated for {} to {} peers",
         hex::encode(key),
         target_peers.len()
     );
@@ -182,8 +180,11 @@ pub async fn replicate_fresh(
 
 /// Send `PaidNotify(K)` to every peer in `PaidCloseGroup(K)` (fire-and-forget).
 ///
-/// Per Invariant 16: sender MUST attempt delivery to every member.
-async fn send_paid_notify(
+/// Per Invariant 16: sender MUST attempt delivery to every member. The
+/// message is small metadata (no chunk data), so it is neither gated by the
+/// send semaphore nor by the pending-offer permit: rules 7-8 run as soon as
+/// the write is accepted, even when chunk offers are backed up.
+pub(crate) async fn send_paid_notify(
     key: &XorName,
     proof_of_payment: &[u8],
     p2p_node: &Arc<P2PNode>,
diff --git a/src/replication/mod.rs b/src/replication/mod.rs
index 29c52f99..97e4030d 100644
--- a/src/replication/mod.rs
+++ b/src/replication/mod.rs
@@ -2373,6 +2373,7 @@ 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]) {
+        fresh::send_paid_notify(key, proof_of_payment, &self.p2p_node, &self.config).await;
         // The semaphore is never closed, so this only fails at shutdown.
         let Ok(pending_offer) = Arc::clone(&self.pending_offer_semaphore)
             .acquire_owned()
@@ -2426,6 +2427,10 @@ impl ReplicationEngine {
                         event
                     }
                 };
+                // Paid-list evidence goes out immediately: it is what lets the
+                // paid close group repair the key later, so it must never wait
+                // behind chunk offers.
+                fresh::send_paid_notify(&event.key, &event.payment_proof, &p2p, &config).await;
                 // 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! {

From ac8348e644decee3d4353f310d2c560706094f2d Mon Sep 17 00:00:00 2001
From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com>
Date: Tue, 22 Sep 2026 17:27:02 +0200
Subject: [PATCH 5/6] fix(replication): two-stage fresh replication with
 un-gated paid-list evidence

Review follow-up. Sending PaidNotify before the permit wait only helped
the head-of-line event: the drainer was one serial loop, so every write
queued behind a blocked permit still had its PaidNotify and PaidForList
insert delayed by chunk back-pressure.

- The fresh-write drainer now never waits for a permit. For every event,
  at arrival rate, it records PaidForList(self) and sends PaidNotify,
  then forwards the event to a new offer dispatcher, the only stage that
  takes a pending-offer permit.
- The dispatcher reads the chunk back with `get_raw` (it was
  content-checked when stored), retries a failed read up to
  MAX_FRESH_READ_ATTEMPTS times with the permit released in between,
  and skips only a chunk that is no longer stored.
- The offer pipeline is one function (`dispatch_fresh_offer`) shared by
  the dispatcher and the direct `replicate_fresh` entry point; the
  8-argument helper and its clippy allow are gone.
- Chunk-carrying protocol fields are encoded as byte strings
  (`serde_bytes`), which has the same postcard layout as a u8 sequence
  (unit-tested) but sizes and serializes in one memcpy pass; an
  oversized body is now refused before anything is allocated.
- PaidNotify shares one `Bytes` buffer across its recipients.
- A new e2e test drives the PUT pipeline through the real channel with
  a missing-chunk event queued ahead of a real one; the harness keeps
  the fresh-write sender so the drainer stays alive in tests.

Pins: ant-protocol 79a68a80, saorsa-core 9baba9c5, saorsa-transport
0f55bfc7 (all perf/replication-send-path).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 Cargo.lock                                    |   7 +-
 Cargo.toml                                    |   9 +-
 ...ounded-fresh-offers-and-copy-free-sends.md |  26 +--
 src/replication/config.rs                     |  11 ++
 src/replication/fresh.rs                      | 120 ++++++++-----
 src/replication/mod.rs                        | 158 +++++++++++++-----
 src/replication/protocol.rs                   |  71 +++++++-
 src/storage/handler.rs                        |   7 +
 tests/e2e/replication.rs                      |  77 +++++++++
 tests/e2e/testnet.rs                          |   9 +-
 10 files changed, 386 insertions(+), 109 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 9f5d2fd9..727ca90b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -862,6 +862,7 @@ dependencies = [
  "self_encryption",
  "semver 1.0.28",
  "serde",
+ "serde_bytes",
  "serde_json",
  "serial_test",
  "sha2",
@@ -883,7 +884,7 @@ dependencies = [
 [[package]]
 name = "ant-protocol"
 version = "2.4.0"
-source = "git+https://github.com/WithAutonomi/ant-protocol?rev=ac25b717d99e70468d934b66c32d6e6219a37350#ac25b717d99e70468d934b66c32d6e6219a37350"
+source = "git+https://github.com/WithAutonomi/ant-protocol?rev=79a68a80c9b35393fc36c8b69d174567e6a3cda4#79a68a80c9b35393fc36c8b69d174567e6a3cda4"
 dependencies = [
  "blake3",
  "bytes",
@@ -5324,7 +5325,7 @@ dependencies = [
 [[package]]
 name = "saorsa-core"
 version = "0.27.4"
-source = "git+https://github.com/WithAutonomi/saorsa-core?rev=e1f1ef92aacb0dab60e6c3dded5da8bae3b63d07#e1f1ef92aacb0dab60e6c3dded5da8bae3b63d07"
+source = "git+https://github.com/WithAutonomi/saorsa-core?rev=9baba9c5ba1e0992a4855892db759a312f27414a#9baba9c5ba1e0992a4855892db759a312f27414a"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -5392,7 +5393,7 @@ dependencies = [
 [[package]]
 name = "saorsa-transport"
 version = "0.36.4"
-source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=3bd451d2d2c86b01df1dd355a88e81bcb854d815#3bd451d2d2c86b01df1dd355a88e81bcb854d815"
+source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=0f55bfc779fd367c2b1f1a01a12b13d6d559989d#0f55bfc779fd367c2b1f1a01a12b13d6d559989d"
 dependencies = [
  "anyhow",
  "async-trait",
diff --git a/Cargo.toml b/Cargo.toml
index 7a4be10a..d7456a76 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -107,6 +107,9 @@ page_size = "0.6"
 
 # Protocol serialization
 postcard = { version = "1.1.3", features = ["use-std"] }
+# Byte-string encoding for chunk payloads in replication messages (same postcard
+# wire layout as a u8 sequence, but serialized and sized in one memcpy pass).
+serde_bytes = "0.11"
 bao = "0.13.1"
 
 # Shared portable browser profile. The native listener is enabled separately
@@ -231,9 +234,9 @@ webrtc-direct = [
 [patch.crates-io]
 saorsa-pqc = { git = "https://github.com/saorsa-labs/saorsa-pqc", rev = "29a2b272c4ee8b72c17102da00033dbfe1ddcd37" }
 evmlib = { git = "https://github.com/WithAutonomi/evmlib", rev = "ccd65f18bf3750af8b8c3cfe07548bc77e3c368d" }
-ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "ac25b717d99e70468d934b66c32d6e6219a37350" }
-saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "e1f1ef92aacb0dab60e6c3dded5da8bae3b63d07" }
-saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "3bd451d2d2c86b01df1dd355a88e81bcb854d815" }
+ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "79a68a80c9b35393fc36c8b69d174567e6a3cda4" }
+saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "9baba9c5ba1e0992a4855892db759a312f27414a" }
+saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "0f55bfc779fd367c2b1f1a01a12b13d6d559989d" }
 
 [profile.release]
 lto = true
diff --git a/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
index fddee579..ba3741c8 100644
--- a/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
+++ b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md
@@ -67,16 +67,22 @@ up to twice their length in capacity).
 We will bound the number of encoded fresh offers that can exist at once and
 make the send path hand a single owned buffer down to the QUIC stream:
 
-- `FreshWriteEvent` carries only the key and the payment proof. The drainer
-  sends `PaidNotify` to the paid close group as soon as it dequeues a write,
-  so the paid-list evidence that later repair depends on is never delayed by
-  chunk back-pressure. It then acquires a `MAX_PENDING_FRESH_OFFERS` (8)
-  permit before it reads the chunk back from storage and encodes it; the
-  permit lives with the encoded offer until the last per-peer send drops it.
-  A backlog therefore waits as small queued events, and at most ~40 MiB of
-  encoded offers exist per node. Nothing is dropped: the queue is unbounded
-  and FIFO, and every offer is still dispatched with the same fan-out,
-  retries and delayed possession check.
+- `FreshWriteEvent` carries only the key and the payment proof, and fresh
+  replication runs as two stages. The fresh-write drainer never waits for
+  chunk back-pressure: for every event, at arrival rate, it records the key
+  in `PaidForList(self)` and sends `PaidNotify` to the paid close group —
+  the evidence later repair depends on — then forwards the event to the
+  offer dispatcher. The dispatcher is the only permit-gated stage: it
+  acquires a `MAX_PENDING_FRESH_OFFERS` (8) permit before it reads the
+  chunk back from storage (`get_raw`; the chunk was content-checked when
+  stored) and encodes it; the permit lives with the encoded offer until the
+  last per-peer send drops it. A backlog therefore waits as small queued
+  events, and at most ~40 MiB of encoded offers exist per node. Nothing is
+  dropped by back-pressure: both queues are unbounded and FIFO, every offer
+  is dispatched with the same fan-out, retries and delayed possession check,
+  and a failed read-back is retried `MAX_FRESH_READ_ATTEMPTS` times with the
+  permit released in between; only a chunk that is no longer stored is
+  skipped.
 - The chunk moves into the offer rather than being copied, and
   `ReplicationMessage::encode` serializes into an exactly-sized buffer.
 - The encoded offer is shared as `Bytes`; saorsa-core's `send_message`
diff --git a/src/replication/config.rs b/src/replication/config.rs
index 1781aba7..6aaec7b3 100644
--- a/src/replication/config.rs
+++ b/src/replication/config.rs
@@ -181,6 +181,17 @@ pub const MAX_CONCURRENT_REPLICATION_SENDS: usize = 3;
 /// backlog waits as small queued events instead of chunk-sized buffers.
 pub const MAX_PENDING_FRESH_OFFERS: usize = 8;
 
+/// How many times the offer dispatcher tries to read an accepted chunk back
+/// from storage before giving up on its fresh offers.
+///
+/// The chunk was stored moments earlier, so a failed read is a transient
+/// fault (exhausted descriptors, an I/O hiccup) far more often than a lost
+/// chunk; a lost chunk reports `None` and is skipped without retry.
+pub const MAX_FRESH_READ_ATTEMPTS: u32 = 3;
+
+/// Pause before retrying a failed chunk read-back in the offer dispatcher.
+pub const FRESH_READ_RETRY_DELAY: Duration = Duration::from_secs(1);
+
 /// 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 9414a98f..1b3b156d 100644
--- a/src/replication/fresh.rs
+++ b/src/replication/fresh.rs
@@ -2,8 +2,11 @@
 //!
 //! When a node accepts a newly written record with valid `PoP`:
 //! 1. Store locally (already done by chunk handler).
-//! 2. Send fresh offers to `CLOSE_GROUP_SIZE` nearest peers (excluding self).
-//! 3. Send `PaidNotify` to all peers in `PaidCloseGroup(K)`.
+//! 2. Record the key in `PaidForList(self)` and send `PaidNotify` to every
+//!    peer in `PaidCloseGroup(K)` — immediately, never behind back-pressure.
+//! 3. Send fresh offers to `CLOSE_GROUP_SIZE` nearest peers (excluding self),
+//!    bounded by the pending-offer permits so a write burst cannot pile up
+//!    chunk-sized buffers.
 
 use std::sync::Arc;
 
@@ -12,13 +15,14 @@ use bytes::Bytes;
 use rand::Rng;
 use saorsa_core::identity::PeerId;
 use saorsa_core::P2PNode;
-use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore};
 
 use crate::ant_protocol::XorName;
 use crate::replication::config::{
     ReplicationConfig, FRESH_REPLICATION_DELIVERY_MAX_RETRIES, REPLICATION_PROTOCOL_ID,
 };
 use crate::replication::paid_list::PaidList;
+use crate::replication::possession::PossessionCheckEvent;
 use crate::replication::protocol::{
     FreshReplicationOffer, PaidNotify, ReplicationMessage, ReplicationMessageBody,
 };
@@ -28,9 +32,9 @@ 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. 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.
+/// chunk is already on disk, and the offer dispatcher 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,
@@ -38,6 +42,28 @@ pub struct FreshWriteEvent {
     pub payment_proof: Vec<u8>,
 }
 
+/// A write whose paid-list evidence has been announced and whose chunk offer
+/// is waiting for a pending-offer permit. Carries no chunk bytes.
+pub(crate) struct FreshOfferEvent {
+    pub(crate) key: XorName,
+    pub(crate) payment_proof: Vec<u8>,
+    /// Storage read-backs attempted so far; see `MAX_FRESH_READ_ATTEMPTS`.
+    pub(crate) read_attempts: u32,
+}
+
+/// Handles shared by everything that dispatches fresh offers, so the offer
+/// dispatcher task and the direct entry point run one pipeline.
+#[derive(Clone)]
+pub(crate) struct FreshOfferContext {
+    pub(crate) p2p_node: Arc<P2PNode>,
+    pub(crate) config: Arc<ReplicationConfig>,
+    /// Limits concurrent outbound chunk transfers across the engine.
+    pub(crate) send_semaphore: Arc<Semaphore>,
+    /// Delayed possession checks (ADR-0003) are scheduled here once an
+    /// offer's sends are dispatched.
+    pub(crate) possession_check_tx: mpsc::UnboundedSender<PossessionCheckEvent>,
+}
+
 /// An encoded fresh offer shared by the per-peer send tasks.
 ///
 /// The pending-offer permit is released together with the buffer, once the
@@ -50,46 +76,49 @@ struct EncodedOffer {
     _pending: OwnedSemaphorePermit,
 }
 
-/// Execute fresh replication for a newly accepted record.
-///
-/// Sends fresh offers to close group members (with bounded delivery retries,
-/// ADR-0003). Returns the close-group peers responsible for the key
-/// (excluding self) so the caller can schedule the delayed possession check.
-/// `PaidNotify` is deliberately not sent here: it carries the paid-list
-/// evidence peers need to repair the key later, so callers send it with
-/// [`send_paid_notify`] as soon as the write is accepted, before waiting for
-/// a pending-offer permit.
+/// Rules 6-8: record the paid key locally and announce it to
+/// `PaidCloseGroup(K)`.
 ///
-/// 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. `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. `data` is taken by
-/// value so the chunk moves into the offer instead of being copied.
-#[allow(clippy::too_many_arguments)]
-pub async fn replicate_fresh(
+/// This is the evidence peers need to repair the key later, so it runs the
+/// moment a write is accepted and is never gated by the pending-offer permit
+/// or the send semaphore; both messages are small metadata.
+pub(crate) async fn announce_paid_write(
     key: &XorName,
-    data: Vec<u8>,
     proof_of_payment: &[u8],
+    paid_list: &PaidList,
     p2p_node: &Arc<P2PNode>,
-    paid_list: &Arc<PaidList>,
     config: &ReplicationConfig,
-    send_semaphore: &Arc<Semaphore>,
-    pending_offer: OwnedSemaphorePermit,
-) -> Vec<PeerId> {
-    let self_id = *p2p_node.peer_id();
-
+) {
     // Rule 6: Node that validates PoP adds K to PaidForList(self).
     if let Err(e) = paid_list.insert(key).await {
         warn!("Failed to add key {} to PaidForList: {e}", hex::encode(key));
     }
+    // Rules 7-8: PaidNotify to every member of PaidCloseGroup(K).
+    send_paid_notify(key, proof_of_payment, p2p_node, config).await;
+}
 
-    // Rule 2-3: Send fresh offers to CLOSE_GROUP_SIZE nearest peers
-    // (excluding self). Use self-inclusive query to get the true close group,
-    // then filter self out.
-    let closest = p2p_node
+/// Rules 2-3: send fresh offers to the close group and schedule the delayed
+/// possession check (ADR-0003) for the responsible peers.
+///
+/// `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.
+/// `data` is taken by value so the chunk moves into the offer instead of
+/// being copied.
+pub(crate) async fn dispatch_fresh_offer(
+    ctx: &FreshOfferContext,
+    key: &XorName,
+    data: Vec<u8>,
+    proof_of_payment: &[u8],
+    pending_offer: OwnedSemaphorePermit,
+) {
+    let self_id = *ctx.p2p_node.peer_id();
+
+    // Use the self-inclusive query to get the true close group, then filter
+    // self out.
+    let closest = ctx
+        .p2p_node
         .dht_manager()
-        .find_closest_nodes_local_with_self(key, config.close_group_size)
+        .find_closest_nodes_local_with_self(key, ctx.config.close_group_size)
         .await;
     let target_peers: Vec<PeerId> = closest
         .iter()
@@ -117,7 +146,7 @@ pub async fn replicate_fresh(
             "Failed to encode FreshReplicationOffer for {}",
             hex::encode(key),
         );
-        return Vec::new();
+        return;
     };
     // One encoded copy serves every per-peer send task and every retry; the
     // transport borrows it through `Bytes` instead of taking a copy. The
@@ -127,10 +156,10 @@ pub async fn replicate_fresh(
         _pending: pending_offer,
     });
     for peer in &target_peers {
-        let p2p = Arc::clone(p2p_node);
+        let p2p = Arc::clone(&ctx.p2p_node);
         let offer = Arc::clone(&encoded);
         let peer_id = *peer;
-        let sem = Arc::clone(send_semaphore);
+        let sem = Arc::clone(&ctx.send_semaphore);
         tokio::spawn(async move {
             // Acquire a permit before sending — this caps the number of
             // concurrent outbound replication transfers across the engine.
@@ -175,15 +204,21 @@ pub async fn replicate_fresh(
         target_peers.len()
     );
 
-    target_peers
+    // Schedule the delayed possession check (ADR-0003) for the responsible
+    // close-group peers. A closed receiver (engine shutting down) is ignored.
+    if !target_peers.is_empty() {
+        let _ = ctx.possession_check_tx.send(PossessionCheckEvent {
+            key: *key,
+            peers: target_peers,
+        });
+    }
 }
 
 /// Send `PaidNotify(K)` to every peer in `PaidCloseGroup(K)` (fire-and-forget).
 ///
 /// Per Invariant 16: sender MUST attempt delivery to every member. The
 /// message is small metadata (no chunk data), so it is neither gated by the
-/// send semaphore nor by the pending-offer permit: rules 7-8 run as soon as
-/// the write is accepted, even when chunk offers are backed up.
+/// send semaphore nor by the pending-offer permit.
 pub(crate) async fn send_paid_notify(
     key: &XorName,
     proof_of_payment: &[u8],
@@ -210,7 +245,8 @@ pub(crate) async fn send_paid_notify(
         warn!("Failed to encode PaidNotify for {}", hex::encode(key));
         return;
     };
-
+    // One buffer for every recipient; the sends only take handles.
+    let encoded = Bytes::from(encoded);
     for node in &paid_group {
         if node.peer_id == self_id {
             continue;
diff --git a/src/replication/mod.rs b/src/replication/mod.rs
index 97e4030d..0347bf2b 100644
--- a/src/replication/mod.rs
+++ b/src/replication/mod.rs
@@ -77,12 +77,12 @@ use crate::replication::commitment_state::{
     PeerCommitmentRecord, PersistedRetention, ResponderCommitmentState, GOSSIP_ANSWERABILITY_TTL,
 };
 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_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,
+    max_parallel_fetch, storage_admission_width, ReplicationConfig, FRESH_READ_RETRY_DELAY,
+    MAX_AUDIT_RESPONSES_PER_PEER, MAX_CONCURRENT_AUDIT_RESPONSES, MAX_CONCURRENT_REPLICATION_SENDS,
+    MAX_DIGEST_AUDIT_RESPONSES_PER_PEER, MAX_FRESH_READ_ATTEMPTS, 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,
 };
 use crate::replication::paid_list::PaidList;
 use crate::replication::protocol::{
@@ -1815,9 +1815,15 @@ pub struct ReplicationEngine {
     subtree_round1: SubtreeRound1Limiter,
     /// Receiver for fresh-write events from the chunk PUT handler.
     ///
-    /// When present, `start()` spawns a drainer task that calls
-    /// `replicate_fresh` for each event.
+    /// When present, `start()` spawns the fresh-write drainer, which records
+    /// paid-list evidence for each event immediately and forwards it to the
+    /// offer dispatcher.
     fresh_write_rx: Option<mpsc::UnboundedReceiver<fresh::FreshWriteEvent>>,
+    /// Hand-off from the fresh-write drainer to the offer dispatcher, the only
+    /// permit-gated stage. Unbounded and FIFO, holding key + proof only.
+    fresh_offer_tx: mpsc::UnboundedSender<fresh::FreshOfferEvent>,
+    /// Receiver paired with `fresh_offer_tx`; taken by the dispatcher task.
+    fresh_offer_rx: Option<mpsc::UnboundedReceiver<fresh::FreshOfferEvent>>,
     /// Sender for delayed possession-check events (ADR-0003). The fresh-write
     /// drainer pushes the responsible close-group peers here after each fresh
     /// replication; the possession-check scheduler drains the paired receiver.
@@ -1881,6 +1887,7 @@ impl ReplicationEngine {
         let initial_neighbors = NeighborSyncState::new_cycle(Vec::new());
         let config = Arc::new(config);
         let (possession_check_tx, possession_check_rx) = mpsc::unbounded_channel();
+        let (fresh_offer_tx, fresh_offer_rx) = mpsc::unbounded_channel();
 
         // ADR-0004: monetized-pin channel (verifier -> first-audit drainer).
         // Bounded (Amendment 2): every stage of the first-audit pipeline is
@@ -1952,6 +1959,8 @@ impl ReplicationEngine {
                 config.subtree_round1_max_concurrent,
             ),
             fresh_write_rx: Some(fresh_write_rx),
+            fresh_offer_tx,
+            fresh_offer_rx: Some(fresh_offer_rx),
             possession_check_tx,
             possession_check_rx: Some(possession_check_rx),
             monetized_pin_tx,
@@ -2235,6 +2244,7 @@ impl ReplicationEngine {
         self.start_verification_worker();
         self.start_bootstrap_sync(dht_events);
         self.start_fresh_write_drainer();
+        self.start_fresh_offer_dispatcher();
         self.start_possession_check_scheduler();
         // ADR-0004: deterministic first audit of commitments that backed a
         // payment (surfaced by the verifier cross-check).
@@ -2373,7 +2383,14 @@ 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]) {
-        fresh::send_paid_notify(key, proof_of_payment, &self.p2p_node, &self.config).await;
+        fresh::announce_paid_write(
+            key,
+            proof_of_payment,
+            &self.paid_list,
+            &self.p2p_node,
+            &self.config,
+        )
+        .await;
         // The semaphore is never closed, so this only fails at shutdown.
         let Ok(pending_offer) = Arc::clone(&self.pending_offer_semaphore)
             .acquire_owned()
@@ -2381,21 +2398,23 @@ impl ReplicationEngine {
         else {
             return;
         };
-        let peers = fresh::replicate_fresh(
+        fresh::dispatch_fresh_offer(
+            &self.fresh_offer_context(),
             key,
             data.to_vec(),
             proof_of_payment,
-            &self.p2p_node,
-            &self.paid_list,
-            &self.config,
-            &self.send_semaphore,
             pending_offer,
         )
         .await;
-        if !peers.is_empty() {
-            let _ = self
-                .possession_check_tx
-                .send(possession::PossessionCheckEvent { key: *key, peers });
+    }
+
+    /// Handles the offer dispatcher and the direct entry point share.
+    fn fresh_offer_context(&self) -> fresh::FreshOfferContext {
+        fresh::FreshOfferContext {
+            p2p_node: Arc::clone(&self.p2p_node),
+            config: Arc::clone(&self.config),
+            send_semaphore: Arc::clone(&self.send_semaphore),
+            possession_check_tx: self.possession_check_tx.clone(),
         }
     }
 
@@ -2411,11 +2430,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 offer_tx = self.fresh_offer_tx.clone();
+        let shutdown = self.shutdown.clone();
+
+        let handle = tokio::spawn(async move {
+            loop {
+                let event = tokio::select! {
+                    () = shutdown.cancelled() => break,
+                    event = rx.recv() => {
+                        let Some(event) = event else { break };
+                        event
+                    }
+                };
+                // Stage one never waits for chunk back-pressure: the paid-list
+                // entry and PaidNotify are what let peers repair the key later,
+                // so every queued write gets them at arrival rate.
+                fresh::announce_paid_write(
+                    &event.key,
+                    &event.payment_proof,
+                    &paid_list,
+                    &p2p,
+                    &config,
+                )
+                .await;
+                if offer_tx
+                    .send(fresh::FreshOfferEvent {
+                        key: event.key,
+                        payment_proof: event.payment_proof,
+                        read_attempts: 0,
+                    })
+                    .is_err()
+                {
+                    break;
+                }
+            }
+            debug!("Fresh-write drainer shut down");
+        });
+        self.task_handles.push(handle);
+    }
+
+    /// Spawn the fresh-offer dispatcher: the only stage of fresh replication
+    /// that waits for a pending-offer permit.
+    ///
+    /// For each forwarded write it acquires a permit, reads the chunk back
+    /// from storage and dispatches the offers. A missing chunk is skipped; a
+    /// failed read is retried up to `MAX_FRESH_READ_ATTEMPTS` times with the
+    /// permit released in between, so a transient I/O fault does not lose the
+    /// write's replication.
+    fn start_fresh_offer_dispatcher(&mut self) {
+        let Some(mut rx) = self.fresh_offer_rx.take() else {
+            return;
+        };
+        let storage = Arc::clone(&self.storage);
         let pending_offer_semaphore = Arc::clone(&self.pending_offer_semaphore);
-        let possession_tx = self.possession_check_tx.clone();
+        let offer_tx = self.fresh_offer_tx.clone();
+        let ctx = self.fresh_offer_context();
         let shutdown = self.shutdown.clone();
 
         let handle = tokio::spawn(async move {
@@ -2427,10 +2497,6 @@ impl ReplicationEngine {
                         event
                     }
                 };
-                // Paid-list evidence goes out immediately: it is what lets the
-                // paid close group repair the key later, so it must never wait
-                // behind chunk offers.
-                fresh::send_paid_notify(&event.key, &event.payment_proof, &p2p, &config).await;
                 // 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! {
@@ -2441,39 +2507,47 @@ impl ReplicationEngine {
                     }
                 };
                 let key_hex = hex::encode(event.key);
-                let data = match storage.get(&event.key).await {
+                // The chunk was content-checked when it was stored; the
+                // receiver validates the offer against its address anyway.
+                let data = match storage.get_raw(&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}");
+                        drop(pending_offer);
+                        let attempts = event.read_attempts + 1;
+                        if attempts >= MAX_FRESH_READ_ATTEMPTS {
+                            warn!(
+                                "Giving up fresh replication of {key_hex} after {attempts} failed reads: {e}"
+                            );
+                            continue;
+                        }
+                        warn!(
+                            "Failed to read chunk {key_hex} for fresh replication (attempt {attempts}): {e}"
+                        );
+                        tokio::select! {
+                            () = shutdown.cancelled() => break,
+                            () = tokio::time::sleep(FRESH_READ_RETRY_DELAY) => {}
+                        }
+                        let _ = offer_tx.send(fresh::FreshOfferEvent {
+                            read_attempts: attempts,
+                            ..event
+                        });
                         continue;
                     }
                 };
-                let peers = fresh::replicate_fresh(
+                fresh::dispatch_fresh_offer(
+                    &ctx,
                     &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");
+            debug!("Fresh-offer dispatcher shut down");
         });
         self.task_handles.push(handle);
     }
diff --git a/src/replication/protocol.rs b/src/replication/protocol.rs
index 40935fdb..55b1fd0f 100644
--- a/src/replication/protocol.rs
+++ b/src/replication/protocol.rs
@@ -52,6 +52,12 @@ impl ReplicationMessage {
         // message is queued for sending.
         let size = postcard::experimental::serialized_size(self)
             .map_err(|e| ReplicationProtocolError::SerializationFailed(e.to_string()))?;
+        // The size is known before anything is allocated, so an oversized body
+        // is refused without serializing it first.
+        let max_size = ceiling_for(family_of_variant(self.body.variant_index()));
+        if size > max_size {
+            return Err(ReplicationProtocolError::MessageTooLarge { size, max_size });
+        }
         let bytes = postcard::to_extend(self, Vec::with_capacity(size))
             .map_err(|e| ReplicationProtocolError::SerializationFailed(e.to_string()))?;
 
@@ -72,13 +78,6 @@ impl ReplicationMessage {
         // the largest is a round-1 proof at the commitment
         // key-count cap, pinned under it with headroom by
         // `max_round1_proof_fits_the_audit_family_ceiling`.
-        let max_size = ceiling_for(family_of_variant(self.body.variant_index()));
-        if bytes.len() > max_size {
-            return Err(ReplicationProtocolError::MessageTooLarge {
-                size: bytes.len(),
-                max_size,
-            });
-        }
 
         // V2-623: cumulative per-variant tx accounting. Every replication send
         // funnels through here, so this is the single tx choke point.
@@ -803,9 +802,13 @@ pub(crate) fn log_served_peers_summary() {
 pub struct FreshReplicationOffer {
     /// The record key.
     pub key: XorName,
-    /// The record data.
+    /// The record data. Encoded as a byte string, which postcard lays out
+    /// exactly like a `u8` sequence, so the wire format is unchanged while
+    /// serialization and sizing copy the payload in one pass.
+    #[serde(with = "serde_bytes")]
     pub data: Vec<u8>,
     /// Proof of Payment (required, validated by receiver).
+    #[serde(with = "serde_bytes")]
     pub proof_of_payment: Vec<u8>,
 }
 
@@ -835,6 +838,7 @@ pub struct PaidNotify {
     /// The record key.
     pub key: XorName,
     /// Proof of Payment for receiver-side verification.
+    #[serde(with = "serde_bytes")]
     pub proof_of_payment: Vec<u8>,
 }
 
@@ -2435,6 +2439,57 @@ mod tests {
         assert_eq!(decoded.request_id, 7);
     }
 
+    /// `serde_bytes` must not change the wire layout: postcard encodes a byte
+    /// string and a `u8` sequence identically (varint length + raw bytes).
+    #[test]
+    fn byte_string_fields_encode_like_u8_sequences() {
+        #[derive(Serialize)]
+        struct PlainOffer {
+            key: XorName,
+            data: Vec<u8>,
+            proof_of_payment: Vec<u8>,
+        }
+        #[derive(Serialize)]
+        struct PlainNotify {
+            key: XorName,
+            proof_of_payment: Vec<u8>,
+        }
+        let data: Vec<u8> = (0..=255u8).cycle().take(70_000).collect();
+        let proof = vec![9u8; 300];
+
+        let offer = FreshReplicationOffer {
+            key: [5; 32],
+            data: data.clone(),
+            proof_of_payment: proof.clone(),
+        };
+        let plain_offer = PlainOffer {
+            key: [5; 32],
+            data: data.clone(),
+            proof_of_payment: proof.clone(),
+        };
+        assert_eq!(
+            postcard::to_stdvec(&offer).unwrap(),
+            postcard::to_stdvec(&plain_offer).unwrap()
+        );
+        let decoded: FreshReplicationOffer =
+            postcard::from_bytes(&postcard::to_stdvec(&plain_offer).unwrap()).unwrap();
+        assert_eq!(decoded.data, data);
+        assert_eq!(decoded.proof_of_payment, proof);
+
+        let notify = PaidNotify {
+            key: [6; 32],
+            proof_of_payment: proof.clone(),
+        };
+        let plain_notify = PlainNotify {
+            key: [6; 32],
+            proof_of_payment: proof,
+        };
+        assert_eq!(
+            postcard::to_stdvec(&notify).unwrap(),
+            postcard::to_stdvec(&plain_notify).unwrap()
+        );
+    }
+
     #[test]
     fn encode_allocates_exactly_the_serialized_size() {
         // A chunk-sized offer must not carry growth slack: the encoded buffer is
diff --git a/src/storage/handler.rs b/src/storage/handler.rs
index 3f6a4fa5..d2726e4a 100644
--- a/src/storage/handler.rs
+++ b/src/storage/handler.rs
@@ -357,6 +357,13 @@ impl AntProtocol {
         self.fresh_write_tx = Some(tx);
     }
 
+    /// The fresh-write sender, if one was set. Lets tests drive the
+    /// replication engine's fresh-write pipeline exactly as a PUT would.
+    #[cfg(any(test, feature = "test-utils"))]
+    pub fn fresh_write_sender(&self) -> Option<mpsc::UnboundedSender<FreshWriteEvent>> {
+        self.fresh_write_tx.clone()
+    }
+
     /// Get the protocol identifier.
     #[must_use]
     pub fn protocol_id(&self) -> &'static str {
diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs
index abf5f92b..08cffaad 100644
--- a/tests/e2e/replication.rs
+++ b/tests/e2e/replication.rs
@@ -13,6 +13,7 @@ use ant_node::replication::commitment_state::{BuiltCommitment, ResponderCommitme
 use ant_node::replication::config::{
     storage_admission_width, K_BUCKET_SIZE, REPLICATION_PROTOCOL_ID,
 };
+use ant_node::replication::fresh::FreshWriteEvent;
 use ant_node::replication::protocol::{
     compute_audit_digest, AuditChallenge, AuditResponse, FetchRequest, FetchResponse,
     FreshReplicationOffer, FreshReplicationResponse, NeighborSyncRequest, ReplicationMessage,
@@ -256,6 +257,82 @@ async fn test_fresh_replication_propagates_to_close_group() {
     harness.teardown().await.expect("teardown");
 }
 
+/// The PUT-driven pipeline (fresh-write drainer → offer dispatcher) replicates
+/// a queued write, and a write whose chunk is no longer stored is skipped
+/// without stalling the pipeline or leaking a pending-offer permit.
+#[tokio::test]
+async fn fresh_write_pipeline_replicates_queued_writes_and_skips_missing_chunks() {
+    let harness = TestHarness::setup_minimal().await.expect("setup");
+    harness.warmup_dht().await.expect("warmup");
+
+    let source_idx = 3; // first regular node
+    let source = harness.test_node(source_idx).expect("source node");
+    let source_protocol = source.ant_protocol.as_ref().expect("protocol");
+    let fresh_tx = source
+        .fresh_write_tx
+        .clone()
+        .expect("fresh-write sender wired by the harness");
+
+    let content = b"queued write through the fresh-write pipeline";
+    let address = compute_address(content);
+    source_protocol
+        .storage()
+        .put(&address, content)
+        .await
+        .expect("put");
+    for i in 0..harness.node_count() {
+        if let Some(node) = harness.test_node(i) {
+            if let Some(protocol) = &node.ant_protocol {
+                protocol.payment_verifier().cache_insert(address);
+            }
+        }
+    }
+
+    let dummy_pop = vec![0x01u8; 64];
+    // A write whose chunk was never stored goes first: the dispatcher must
+    // skip it and carry on with the next event.
+    let missing = compute_address(b"never stored anywhere");
+    fresh_tx
+        .send(FreshWriteEvent {
+            key: missing,
+            payment_proof: dummy_pop.clone(),
+        })
+        .expect("queue missing write");
+    fresh_tx
+        .send(FreshWriteEvent {
+            key: address,
+            payment_proof: dummy_pop,
+        })
+        .expect("queue write");
+
+    let deadline = tokio::time::Instant::now() + PROPAGATION_TIMEOUT;
+    let mut found_on_other = false;
+    while tokio::time::Instant::now() < deadline {
+        for i in 0..harness.node_count() {
+            if i == source_idx {
+                continue;
+            }
+            if let Some(node) = harness.test_node(i) {
+                if let Some(protocol) = &node.ant_protocol {
+                    if protocol.storage().exists(&address).unwrap_or(false) {
+                        found_on_other = true;
+                    }
+                }
+            }
+        }
+        if found_on_other {
+            break;
+        }
+        tokio::time::sleep(PROPAGATION_POLL_INTERVAL).await;
+    }
+    assert!(
+        found_on_other,
+        "queued write should have replicated through the fresh-write pipeline"
+    );
+
+    harness.teardown().await.expect("teardown");
+}
+
 /// ADR-0003: the delayed possession check penalises a responsible peer that
 /// does NOT hold the chunk, and leaves a peer that DOES hold it unpenalised.
 ///
diff --git a/tests/e2e/testnet.rs b/tests/e2e/testnet.rs
index a082efd0..a06344c6 100644
--- a/tests/e2e/testnet.rs
+++ b/tests/e2e/testnet.rs
@@ -23,6 +23,7 @@ use ant_node::payment::{
     QuotingMetricsTracker,
 };
 use ant_node::replication::config::MAX_REPLICATION_MESSAGE_SIZE;
+use ant_node::replication::fresh::FreshWriteEvent;
 use ant_node::storage::{AntProtocol, ChunkStore, ChunkStoreConfig};
 use ant_node::{ReplicationConfig, ReplicationEngine};
 use bytes::Bytes;
@@ -425,6 +426,10 @@ pub struct TestNode {
 
     /// Shutdown token for the replication engine.
     pub replication_shutdown: Option<CancellationToken>,
+
+    /// Sender feeding the replication engine's fresh-write pipeline, kept so
+    /// tests can queue writes exactly as the PUT handler does.
+    pub fresh_write_tx: Option<tokio::sync::mpsc::UnboundedSender<FreshWriteEvent>>,
 }
 
 impl TestNode {
@@ -1090,6 +1095,7 @@ impl TestNetwork {
             protocol_task: None,
             replication_engine: None,
             replication_shutdown: None,
+            fresh_write_tx: None,
         })
     }
 
@@ -1333,7 +1339,8 @@ impl TestNetwork {
         {
             let shutdown = CancellationToken::new();
             let repl_config = self.config.replication_config.clone().unwrap_or_default();
-            let (_fresh_tx, fresh_rx) = tokio::sync::mpsc::unbounded_channel();
+            let (fresh_tx, fresh_rx) = tokio::sync::mpsc::unbounded_channel();
+            node.fresh_write_tx = Some(fresh_tx);
             let node_identity = Arc::clone(id);
             match ReplicationEngine::new(
                 repl_config,

From b0263b324c418a5732d1727d7a66df4c15946559 Mon Sep 17 00:00:00 2001
From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com>
Date: Wed, 23 Sep 2026 09:43:28 +0200
Subject: [PATCH 6/6] test(replication): saturation and read-retry coverage for
 the fresh-write pipeline

Review "patch first" follow-up for the two-stage fresh replication.

Two e2e tests over the real harness:

- `fresh_write_pipeline_drains_a_burst_larger_than_the_offer_budget`
  queues 3 x MAX_PENDING_FRESH_OFFERS writes in one go, asserts every one
  replicates to another node, and asserts all pending-offer permits are
  back afterwards, so a burst neither loses a write to back-pressure nor
  leaks a permit.
- `fresh_write_pipeline_retries_a_transient_read_failure` makes the chunk
  file unreadable on disk, waits until the dispatcher's failed read has
  marked the chunk suspect (observable through `exists`), restores it
  inside FRESH_READ_RETRY_DELAY, and asserts the write replicates and the
  suspect mark clears on the source.

Adds the test-utils accessor `ReplicationEngine::pending_offer_permits_available`
and shares the store/poll helpers with the existing pipeline test.

Pins: saorsa-transport 1766037e (legacy `send` copies only on the QUIC
path, `send_bytes` regression tests), saorsa-core 53f1fc6f and
ant-protocol a66ddcfb (pin bumps only).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 Cargo.lock               |  16 +-
 Cargo.toml               |   6 +-
 src/replication/mod.rs   |  10 ++
 tests/e2e/replication.rs | 318 ++++++++++++++++++++++++++++++++++-----
 4 files changed, 299 insertions(+), 51 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 727ca90b..18969f86 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -884,7 +884,7 @@ dependencies = [
 [[package]]
 name = "ant-protocol"
 version = "2.4.0"
-source = "git+https://github.com/WithAutonomi/ant-protocol?rev=79a68a80c9b35393fc36c8b69d174567e6a3cda4#79a68a80c9b35393fc36c8b69d174567e6a3cda4"
+source = "git+https://github.com/WithAutonomi/ant-protocol?rev=a66ddcfb60d50a9ec07b9aee0d063a730a96b330#a66ddcfb60d50a9ec07b9aee0d063a730a96b330"
 dependencies = [
  "blake3",
  "bytes",
@@ -3295,7 +3295,7 @@ dependencies = [
  "libc",
  "percent-encoding",
  "pin-project-lite",
- "socket2 0.5.10",
+ "socket2 0.6.5",
  "tokio",
  "tower-service",
  "tracing",
@@ -4656,7 +4656,7 @@ dependencies = [
  "quinn-udp 0.5.15",
  "rustc-hash",
  "rustls",
- "socket2 0.5.10",
+ "socket2 0.6.5",
  "thiserror 2.0.20",
  "tokio",
  "tracing",
@@ -4695,7 +4695,7 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2 0.5.10",
+ "socket2 0.6.5",
  "tracing",
  "windows-sys 0.61.2",
 ]
@@ -4708,7 +4708,7 @@ checksum = "76150b617afc75e6e21ac5f39bc196e80b65415ae48d62dbef8e2519d040ce42"
 dependencies = [
  "cfg_aliases",
  "libc",
- "socket2 0.5.10",
+ "socket2 0.6.5",
  "tracing",
  "windows-sys 0.61.2",
 ]
@@ -5325,7 +5325,7 @@ dependencies = [
 [[package]]
 name = "saorsa-core"
 version = "0.27.4"
-source = "git+https://github.com/WithAutonomi/saorsa-core?rev=9baba9c5ba1e0992a4855892db759a312f27414a#9baba9c5ba1e0992a4855892db759a312f27414a"
+source = "git+https://github.com/WithAutonomi/saorsa-core?rev=53f1fc6fca13109f583dd951ee3eab42d7a37121#53f1fc6fca13109f583dd951ee3eab42d7a37121"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -5393,7 +5393,7 @@ dependencies = [
 [[package]]
 name = "saorsa-transport"
 version = "0.36.4"
-source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=0f55bfc779fd367c2b1f1a01a12b13d6d559989d#0f55bfc779fd367c2b1f1a01a12b13d6d559989d"
+source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=1766037ef159d80495515de65f96b3fd8fc94319#1766037ef159d80495515de65f96b3fd8fc94319"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -7122,7 +7122,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index d7456a76..e189a326 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -234,9 +234,9 @@ webrtc-direct = [
 [patch.crates-io]
 saorsa-pqc = { git = "https://github.com/saorsa-labs/saorsa-pqc", rev = "29a2b272c4ee8b72c17102da00033dbfe1ddcd37" }
 evmlib = { git = "https://github.com/WithAutonomi/evmlib", rev = "ccd65f18bf3750af8b8c3cfe07548bc77e3c368d" }
-ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "79a68a80c9b35393fc36c8b69d174567e6a3cda4" }
-saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "9baba9c5ba1e0992a4855892db759a312f27414a" }
-saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "0f55bfc779fd367c2b1f1a01a12b13d6d559989d" }
+ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "a66ddcfb60d50a9ec07b9aee0d063a730a96b330" }
+saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "53f1fc6fca13109f583dd951ee3eab42d7a37121" }
+saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "1766037ef159d80495515de65f96b3fd8fc94319" }
 
 [profile.release]
 lto = true
diff --git a/src/replication/mod.rs b/src/replication/mod.rs
index 0347bf2b..cac1bbeb 100644
--- a/src/replication/mod.rs
+++ b/src/replication/mod.rs
@@ -2218,6 +2218,16 @@ impl ReplicationEngine {
         })
     }
 
+    /// Test-only: pending-offer permits not currently held by an encoded
+    /// fresh offer. Equals [`MAX_PENDING_FRESH_OFFERS`] when no fresh
+    /// replication is in flight, which is how tests prove a burst of writes
+    /// drained without leaking a permit.
+    #[cfg(any(test, feature = "test-utils"))]
+    #[must_use]
+    pub fn pending_offer_permits_available(&self) -> usize {
+        self.pending_offer_semaphore.available_permits()
+    }
+
     /// Start all background tasks.
     ///
     /// `dht_events` must be subscribed **before** `P2PNode::start()` so that
diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs
index 08cffaad..814931bf 100644
--- a/tests/e2e/replication.rs
+++ b/tests/e2e/replication.rs
@@ -11,7 +11,8 @@ use ant_node::client::compute_address;
 use ant_node::replication::audit_coordinator::AuditChallengeCoordinator;
 use ant_node::replication::commitment_state::{BuiltCommitment, ResponderCommitmentState};
 use ant_node::replication::config::{
-    storage_admission_width, K_BUCKET_SIZE, REPLICATION_PROTOCOL_ID,
+    storage_admission_width, FRESH_READ_RETRY_DELAY, K_BUCKET_SIZE, MAX_PENDING_FRESH_OFFERS,
+    REPLICATION_PROTOCOL_ID,
 };
 use ant_node::replication::fresh::FreshWriteEvent;
 use ant_node::replication::protocol::{
@@ -22,11 +23,17 @@ use ant_node::replication::protocol::{
 use ant_node::replication::pruning;
 use ant_node::replication::scheduling::ReplicationQueues;
 use ant_node::replication::types::{NeighborSyncState, RepairProofs};
+use ant_node::storage::file_store::CHUNKS_DIR_NAME;
+use ant_node::storage::XorName;
 use ant_node::ReplicationConfig;
 use saorsa_core::identity::PeerId;
 use saorsa_core::{P2PNode, TrustEvent};
 use serial_test::serial;
 use std::collections::HashSet;
+use std::fs;
+#[cfg(unix)]
+use std::os::unix::fs::PermissionsExt;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
@@ -49,6 +56,17 @@ const FULL_NODE_SHUN_POSSESSION_DELAY_MAX: Duration = Duration::from_millis(500)
 const DUMMY_PAYMENT_PROOF_LEN: usize = 64;
 /// Dummy proof byte used when a test only needs to reach pre-payment gates.
 const DUMMY_PAYMENT_PROOF_BYTE: u8 = 0x01;
+/// First regular (non-bootstrap) node of the minimal harness; source of the
+/// fresh-write pipeline tests.
+const FRESH_PIPELINE_SOURCE_INDEX: usize = 3;
+/// Writes queued at once by the saturation test: three times the pending-offer
+/// budget, so the dispatcher must block on and recycle permits to drain it.
+const FRESH_BURST_WRITES: usize = 3 * MAX_PENDING_FRESH_OFFERS;
+/// Wait budget for the whole burst to replicate and release its permits.
+const FRESH_BURST_TIMEOUT: Duration = Duration::from_secs(45);
+/// File mode that makes a chunk unreadable, injecting a transient read fault.
+#[cfg(unix)]
+const UNREADABLE_FILE_MODE: u32 = 0o000;
 /// Minimal paid-list repair close group used by the deterministic repair e2e.
 const PAID_REPAIR_GROUP_SIZE: usize = 5;
 /// Storage threshold configured above majority so one holder is below quorum.
@@ -265,74 +283,294 @@ async fn fresh_write_pipeline_replicates_queued_writes_and_skips_missing_chunks(
     let harness = TestHarness::setup_minimal().await.expect("setup");
     harness.warmup_dht().await.expect("warmup");
 
-    let source_idx = 3; // first regular node
-    let source = harness.test_node(source_idx).expect("source node");
-    let source_protocol = source.ant_protocol.as_ref().expect("protocol");
+    let source = harness
+        .test_node(FRESH_PIPELINE_SOURCE_INDEX)
+        .expect("source node");
     let fresh_tx = source
         .fresh_write_tx
         .clone()
         .expect("fresh-write sender wired by the harness");
+    let address = store_paid_chunk(
+        &harness,
+        FRESH_PIPELINE_SOURCE_INDEX,
+        b"queued write through the fresh-write pipeline",
+    )
+    .await;
 
-    let content = b"queued write through the fresh-write pipeline";
-    let address = compute_address(content);
-    source_protocol
-        .storage()
-        .put(&address, content)
-        .await
-        .expect("put");
-    for i in 0..harness.node_count() {
-        if let Some(node) = harness.test_node(i) {
-            if let Some(protocol) = &node.ant_protocol {
-                protocol.payment_verifier().cache_insert(address);
-            }
-        }
-    }
-
-    let dummy_pop = vec![0x01u8; 64];
     // A write whose chunk was never stored goes first: the dispatcher must
     // skip it and carry on with the next event.
     let missing = compute_address(b"never stored anywhere");
     fresh_tx
         .send(FreshWriteEvent {
             key: missing,
-            payment_proof: dummy_pop.clone(),
+            payment_proof: dummy_payment_proof(),
         })
         .expect("queue missing write");
     fresh_tx
         .send(FreshWriteEvent {
             key: address,
-            payment_proof: dummy_pop,
+            payment_proof: dummy_payment_proof(),
         })
         .expect("queue write");
 
-    let deadline = tokio::time::Instant::now() + PROPAGATION_TIMEOUT;
-    let mut found_on_other = false;
-    while tokio::time::Instant::now() < deadline {
-        for i in 0..harness.node_count() {
-            if i == source_idx {
-                continue;
-            }
-            if let Some(node) = harness.test_node(i) {
-                if let Some(protocol) = &node.ant_protocol {
-                    if protocol.storage().exists(&address).unwrap_or(false) {
-                        found_on_other = true;
-                    }
-                }
+    assert!(
+        wait_until_replicated(
+            &harness,
+            FRESH_PIPELINE_SOURCE_INDEX,
+            &address,
+            PROPAGATION_TIMEOUT
+        )
+        .await,
+        "queued write should have replicated through the fresh-write pipeline"
+    );
+
+    harness.teardown().await.expect("teardown");
+}
+
+/// Saturation: a burst of writes three times larger than the pending-offer
+/// budget, queued in one go, all replicate. The dispatcher has to block on the
+/// `MAX_PENDING_FRESH_OFFERS` semaphore and recycle permits to get through it,
+/// and once the burst has drained every permit is back — no write was lost to
+/// back-pressure and no permit leaked.
+#[tokio::test]
+async fn fresh_write_pipeline_drains_a_burst_larger_than_the_offer_budget() {
+    let harness = TestHarness::setup_minimal().await.expect("setup");
+    harness.warmup_dht().await.expect("warmup");
+
+    let source = harness
+        .test_node(FRESH_PIPELINE_SOURCE_INDEX)
+        .expect("source node");
+    let engine = source
+        .replication_engine
+        .as_ref()
+        .expect("replication engine");
+    let fresh_tx = source
+        .fresh_write_tx
+        .clone()
+        .expect("fresh-write sender wired by the harness");
+
+    let mut addresses = Vec::with_capacity(FRESH_BURST_WRITES);
+    for i in 0..FRESH_BURST_WRITES {
+        let content = format!("fresh-write burst chunk {i}");
+        addresses.push(
+            store_paid_chunk(&harness, FRESH_PIPELINE_SOURCE_INDEX, content.as_bytes()).await,
+        );
+    }
+    assert_eq!(
+        engine.pending_offer_permits_available(),
+        MAX_PENDING_FRESH_OFFERS,
+        "all permits must be free before the burst"
+    );
+
+    for &address in &addresses {
+        fresh_tx
+            .send(FreshWriteEvent {
+                key: address,
+                payment_proof: dummy_payment_proof(),
+            })
+            .expect("queue write");
+    }
+
+    let deadline = tokio::time::Instant::now() + FRESH_BURST_TIMEOUT;
+    let mut replicated: HashSet<XorName> = HashSet::new();
+    while tokio::time::Instant::now() < deadline && replicated.len() < addresses.len() {
+        for address in &addresses {
+            if !replicated.contains(address)
+                && stored_on_another_node(&harness, FRESH_PIPELINE_SOURCE_INDEX, address)
+            {
+                replicated.insert(*address);
             }
         }
-        if found_on_other {
-            break;
-        }
         tokio::time::sleep(PROPAGATION_POLL_INTERVAL).await;
     }
+    assert_eq!(
+        replicated.len(),
+        addresses.len(),
+        "only {} of {} burst writes replicated within {FRESH_BURST_TIMEOUT:?}",
+        replicated.len(),
+        addresses.len()
+    );
+
+    // A permit is released when the last per-peer send of its offer finishes,
+    // which can trail the chunk landing on a peer by a moment.
+    while tokio::time::Instant::now() < deadline
+        && engine.pending_offer_permits_available() < MAX_PENDING_FRESH_OFFERS
+    {
+        tokio::time::sleep(PROPAGATION_POLL_INTERVAL).await;
+    }
+    assert_eq!(
+        engine.pending_offer_permits_available(),
+        MAX_PENDING_FRESH_OFFERS,
+        "the burst leaked a pending-offer permit"
+    );
+
+    harness.teardown().await.expect("teardown");
+}
+
+/// Retry: a chunk whose file cannot be read when its offer permit arrives is
+/// not dropped. The dispatcher releases the permit, waits
+/// `FRESH_READ_RETRY_DELAY` and reads again; once the fault has cleared the
+/// write replicates. The fault is injected by making the chunk file
+/// unreadable on disk and restored inside the retry window.
+#[cfg(unix)]
+#[tokio::test]
+async fn fresh_write_pipeline_retries_a_transient_read_failure() {
+    let harness = TestHarness::setup_minimal().await.expect("setup");
+    harness.warmup_dht().await.expect("warmup");
+
+    let source = harness
+        .test_node(FRESH_PIPELINE_SOURCE_INDEX)
+        .expect("source node");
+    let storage = source.ant_protocol.as_ref().expect("protocol").storage();
+    let fresh_tx = source
+        .fresh_write_tx
+        .clone()
+        .expect("fresh-write sender wired by the harness");
+    let address = store_paid_chunk(
+        &harness,
+        FRESH_PIPELINE_SOURCE_INDEX,
+        b"chunk whose first read-back fails",
+    )
+    .await;
+
+    let chunk_file = chunk_file_path(storage.root_dir(), &address).expect("chunk file on disk");
+    let readable = fs::metadata(&chunk_file)
+        .expect("chunk metadata")
+        .permissions();
+    fs::set_permissions(
+        &chunk_file,
+        fs::Permissions::from_mode(UNREADABLE_FILE_MODE),
+    )
+    .expect("make chunk unreadable");
+    if fs::File::open(&chunk_file).is_ok() {
+        // File modes are not enforced for this user (root): the fault cannot
+        // be injected, so there is nothing to test here.
+        eprintln!("skipping: file modes are not enforced for this user");
+        fs::set_permissions(&chunk_file, readable).expect("restore chunk mode");
+        harness.teardown().await.expect("teardown");
+        return;
+    }
     assert!(
-        found_on_other,
-        "queued write should have replicated through the fresh-write pipeline"
+        storage.exists(&address).unwrap_or(false),
+        "chunk is indexed before the fault is hit"
+    );
+
+    fresh_tx
+        .send(FreshWriteEvent {
+            key: address,
+            payment_proof: dummy_payment_proof(),
+        })
+        .expect("queue write");
+
+    // A failed read marks the chunk suspect, which hides it from `exists`:
+    // that is the observable proof the dispatcher's first attempt hit the
+    // fault. Only then clear it, inside the retry delay.
+    assert!(
+        wait_until(
+            || !storage.exists(&address).unwrap_or(true),
+            PROPAGATION_TIMEOUT
+        )
+        .await,
+        "dispatcher never attempted the faulty read"
+    );
+    fs::set_permissions(&chunk_file, readable).expect("restore chunk mode");
+
+    assert!(
+        wait_until_replicated(
+            &harness,
+            FRESH_PIPELINE_SOURCE_INDEX,
+            &address,
+            FRESH_READ_RETRY_DELAY + PROPAGATION_TIMEOUT
+        )
+        .await,
+        "write should have replicated on the retried read"
+    );
+    assert!(
+        storage.exists(&address).unwrap_or(false),
+        "the successful retry clears the suspect mark on the source"
     );
 
     harness.teardown().await.expect("teardown");
 }
 
+/// Proof bytes for tests that only need to reach the pre-payment gates.
+fn dummy_payment_proof() -> Vec<u8> {
+    vec![DUMMY_PAYMENT_PROOF_BYTE; DUMMY_PAYMENT_PROOF_LEN]
+}
+
+/// Store `content` on `source_idx` and mark it paid on every node, so a fresh
+/// offer for it is accepted wherever it lands. Returns the chunk's address.
+async fn store_paid_chunk(harness: &TestHarness, source_idx: usize, content: &[u8]) -> XorName {
+    let address = compute_address(content);
+    harness
+        .test_node(source_idx)
+        .expect("source node")
+        .ant_protocol
+        .as_ref()
+        .expect("protocol")
+        .storage()
+        .put(&address, content)
+        .await
+        .expect("put");
+    for i in 0..harness.node_count() {
+        if let Some(protocol) = harness
+            .test_node(i)
+            .and_then(|node| node.ant_protocol.as_ref())
+        {
+            protocol.payment_verifier().cache_insert(address);
+        }
+    }
+    address
+}
+
+/// Whether any node other than `source_idx` currently stores `address`.
+fn stored_on_another_node(harness: &TestHarness, source_idx: usize, address: &XorName) -> bool {
+    (0..harness.node_count())
+        .filter(|&i| i != source_idx)
+        .filter_map(|i| harness.test_node(i))
+        .filter_map(|node| node.ant_protocol.as_ref())
+        .any(|protocol| protocol.storage().exists(address).unwrap_or(false))
+}
+
+/// Poll until `address` is stored on a node other than `source_idx`, or
+/// `budget` runs out.
+async fn wait_until_replicated(
+    harness: &TestHarness,
+    source_idx: usize,
+    address: &XorName,
+    budget: Duration,
+) -> bool {
+    wait_until(
+        || stored_on_another_node(harness, source_idx, address),
+        budget,
+    )
+    .await
+}
+
+/// Poll `condition` every `PROPAGATION_POLL_INTERVAL` until it holds or
+/// `budget` runs out.
+async fn wait_until(mut condition: impl FnMut() -> bool, budget: Duration) -> bool {
+    let deadline = tokio::time::Instant::now() + budget;
+    while tokio::time::Instant::now() < deadline {
+        if condition() {
+            return true;
+        }
+        tokio::time::sleep(PROPAGATION_POLL_INTERVAL).await;
+    }
+    false
+}
+
+/// On-disk file of a stored chunk: the store shards `root/chunks/` into
+/// subdirectories, so look through them for the file named after the address.
+fn chunk_file_path(root_dir: &Path, address: &XorName) -> Option<PathBuf> {
+    let file_name = hex::encode(address);
+    fs::read_dir(root_dir.join(CHUNKS_DIR_NAME))
+        .ok()?
+        .filter_map(Result::ok)
+        .map(|shard| shard.path().join(&file_name))
+        .find(|candidate| candidate.is_file())
+}
+
 /// ADR-0003: the delayed possession check penalises a responsible peer that
 /// does NOT hold the chunk, and leaves a peer that DOES hold it unpenalised.
 ///