fix(shm): prevent watchdog invalidation of in-transit SHM chunks

commons/zenoh-shm/src/header/chunk_header.rs

@@ -29,6 +29,9 @@ pub struct ChunkHeaderType {
     */
     pub refcount: AtomicU32,
     pub watchdog_invalidated: AtomicBool,
+    /// Set by RX after installing its ConfirmedDescriptor; cleared on chunk recycle.
+    /// TX sweep checks this to release the pending lease early (before TTL expiry).
+    pub rx_ack: AtomicBool,
     pub generation: AtomicU32,
 
     /// Protocol identifier for particular SHM implementation

commons/zenoh-shm/src/lib.rs

@@ -174,13 +174,29 @@ impl ShmBufInner {
         self.metadata.owned.header().len()
     }
 
-    fn is_valid(&self) -> bool {
+    pub fn is_valid(&self) -> bool {
         let header = self.metadata.owned.header();
 
         !header.watchdog_invalidated.load(Ordering::SeqCst)
             && header.generation.load(Ordering::SeqCst) == self.info.generation
     }
 
+    /// Returns true if RX has installed its ConfirmedDescriptor for this buffer.
+    /// TX sweep uses this to release the pending lease before TTL expiry.
+    pub fn is_rx_acked(&self) -> bool {
+        self.metadata.owned.header().rx_ack.load(Ordering::Acquire)
+    }
+
+    /// Set rx_ack on this buffer. Called by `read_shmbuf` after the ConfirmedDescriptor
+    /// is installed; also available for testing the TX sweep path.
+    pub fn mark_rx_acked(&self) {
+        self.metadata
+            .owned
+            .header()
+            .rx_ack
+            .store(true, Ordering::Release);
+    }
+
     fn is_unique(&self) -> bool {
         self.ref_count() == 1
     }

commons/zenoh-shm/src/metadata/storage.rs

@@ -93,10 +93,14 @@ impl MetadataStorage {
 
     pub fn reclaim(&self, descriptor: OwnedMetadataDescriptor) {
         // header deallocated - increment it's generation to invalidate any existing references
-        descriptor
-            .header()
+        let header = descriptor.header();
+        header
             .generation
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        // clear rx_ack so the next use of this slot starts with a clean state
+        header
+            .rx_ack
+            .store(false, std::sync::atomic::Ordering::Relaxed);
         let mut guard = self.available.lock().unwrap();
         guard.push_back(descriptor);
     }

commons/zenoh-shm/src/reader.rs

@@ -55,6 +55,14 @@ impl ShmReader {
         // attach to the watchdog before doing other things
         let confirmed_metadata = GLOBAL_CONFIRMATOR.read().add(metadata);
 
+        // Signal TX that the ConfirmedDescriptor is installed — TX sweep can release the
+        // pending lease early. Must happen after add() returns (watchdog is kicking).
+        confirmed_metadata
+            .owned
+            .header()
+            .rx_ack
+            .store(true, std::sync::atomic::Ordering::Release);
+
         // retrieve data descriptor from metadata
         let data_descriptor = confirmed_metadata.owned.header().data_descriptor();
 

io/zenoh-transport/src/shm.rs

@@ -16,6 +16,7 @@ use std::{
     fmt::Debug,
     num::NonZeroUsize,
     sync::{Arc, Mutex},
+    time::{Duration, Instant},
 };
 
 use zenoh_buffers::{reader::HasReader, ZBuf, ZSlice, ZSliceKind};
@@ -46,6 +47,19 @@ use zenoh_shm::{
 
 use crate::unicast::establishment::ext::shm::AuthSegment;
 
+/// TTL for in-flight SHM buffer leases (Gray & Cheriton lease model).
+/// Must be significantly larger than the watchdog validator period (100 ms)
+/// to cover realistic RX thread stalls. 5× gives headroom for scheduler
+/// jitter without accumulating excessive memory.
+pub(crate) const SHM_PENDING_TTL: Duration = Duration::from_millis(500);
+
+pub(crate) struct PendingShmBuf {
+    // Held solely for its RAII Drop (keeps ConfirmedDescriptor alive); never read.
+    #[allow(dead_code)]
+    pub(crate) buf: ShmBufInner,
+    pub(crate) deadline: Instant,
+}
+
 #[derive(Debug)]
 struct ProviderInitCfg {
     shm_size: NonZeroUsize,
@@ -231,6 +245,51 @@ pub fn map_zmsg_to_partner<ShmCfg: PartnerShmConfig>(
     }
 }
 
+/// Clone every [`ShmBufInner`] from SHM-mapped ZSlices in `msg`.
+/// Returns an empty Vec if no SHM slices are present.
+/// The caller stores these clones in the connection's pending set so the
+/// [`ConfirmedDescriptor`] outlives this stack frame until the lease expires or
+/// the connection closes (Gray & Cheriton lease model).
+pub fn collect_shm_bufs(msg: &NetworkMessageMut) -> Vec<ShmBufInner> {
+    let mut out = Vec::new();
+    match &msg.body {
+        NetworkBodyMut::Push(Push { payload, .. }) => match payload {
+            PushBody::Put(b) => collect_from_zbuf(&b.payload, &mut out),
+            PushBody::Del(_) => {}
+        },
+        NetworkBodyMut::Request(Request { payload, .. }) => match payload {
+            RequestBody::Query(b) => {
+                if let Some(body) = &b.ext_body {
+                    collect_from_zbuf(&body.payload, &mut out);
+                }
+            }
+        },
+        NetworkBodyMut::Response(Response { payload, .. }) => match payload {
+            ResponseBody::Reply(b) => {
+                if let PushBody::Put(p) = &b.payload {
+                    collect_from_zbuf(&p.payload, &mut out);
+                }
+            }
+            ResponseBody::Err(b) => collect_from_zbuf(&b.payload, &mut out),
+        },
+        NetworkBodyMut::ResponseFinal(_)
+        | NetworkBodyMut::Interest(_)
+        | NetworkBodyMut::Declare(_)
+        | NetworkBodyMut::OAM(_) => {}
+    }
+    out
+}
+
+fn collect_from_zbuf(zbuf: &ZBuf, out: &mut Vec<ShmBufInner>) {
+    for zs in zbuf.zslices() {
+        if zs.kind == ZSliceKind::ShmPtr {
+            if let Some(shmb) = zs.downcast_ref::<ShmBufInner>() {
+                out.push(shmb.clone());
+            }
+        }
+    }
+}
+
 pub fn map_zmsg_to_shmbuf(msg: NetworkMessageMut, shmr: &ShmReader) -> ZResult<()> {
     match msg.body {
         NetworkBodyMut::Push(Push { payload, .. }) => match payload {
@@ -434,3 +493,174 @@ pub fn map_zslice_to_shmbuf(zslice: &mut ZSlice, shmr: &ShmReader) -> ZResult<()
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{
+        collections::HashMap,
+        time::{Duration, Instant},
+    };
+
+    use zenoh_buffers::ZBuf;
+    use zenoh_core::Wait;
+    use zenoh_shm::{api::provider::shm_provider::ShmProviderBuilder, ShmBufInner};
+
+    use super::{PendingShmBuf, SHM_PENDING_TTL};
+
+    /// Extract a cloned ShmBufInner from a ZShmMut.
+    /// The provider must be kept alive by the caller for the duration of use,
+    /// otherwise the provider drops and recycles the chunk (invalidating its generation).
+    fn shmbuf_from_zbuf(zbuf: &ZBuf) -> ShmBufInner {
+        let zslice = zbuf
+            .zslices()
+            .next()
+            .expect("ZBuf should have at least one ZSlice")
+            .clone();
+        zslice
+            .downcast_ref::<ShmBufInner>()
+            .expect("ZSlice should hold ShmBufInner")
+            .clone()
+    }
+
+    /// Invariant 1+3: a clone in the pending map keeps the chunk alive through
+    /// validator ticks; clearing pending lets the validator fire.
+    #[test]
+    fn pending_set_keeps_chunk_alive_and_clear_lets_validator_fire() {
+        // Provider must outlive all ShmBufInner references — drop order matters.
+        let provider = ShmProviderBuilder::default_backend(65536).wait().unwrap();
+        let zbuf: ZBuf = provider.alloc(64).wait().unwrap().into();
+        let shmb = shmbuf_from_zbuf(&zbuf);
+        assert!(shmb.is_valid(), "freshly allocated buffer should be valid");
+
+        // Clone into pending map (simulates TX collect_shm_bufs after push)
+        let now = Instant::now();
+        let deadline = now + SHM_PENDING_TTL;
+        let mut pending: HashMap<_, PendingShmBuf> = HashMap::new();
+        let key = shmb.info.metadata.clone();
+        let pending_clone = shmb.clone();
+        pending.insert(
+            key,
+            PendingShmBuf {
+                buf: pending_clone,
+                deadline,
+            },
+        );
+
+        // Drop zbuf (holds the original ZSlice/ShmBufInner) and shmb
+        // — simulates internal_schedule returning after do_push.
+        drop(zbuf);
+        drop(shmb);
+
+        // Wait > 2 validator ticks (each tick = 100 ms)
+        std::thread::sleep(Duration::from_millis(350));
+
+        // Invariant 1: pending clone holds ConfirmedDescriptor — chunk still valid
+        assert!(
+            pending.values().next().unwrap().buf.is_valid(),
+            "chunk should remain valid while held in shm_pending"
+        );
+
+        // Simulate transport delete(): clear the pending map
+        pending.clear();
+
+        // Invariant 3: after clear, ConfirmedDescriptor drops; validator fires within 200 ms
+        std::thread::sleep(Duration::from_millis(350));
+        // (We cannot call is_valid() after clear since we dropped the ShmBufInner.)
+        // The correctness here is validated end-to-end by Test C in unicast_shm.rs.
+
+        drop(provider);
+    }
+
+    /// Invariant 2: TTL sweep removes expired entries; live entry survives.
+    #[test]
+    fn ttl_sweep_removes_expired_entries() {
+        const SHORT_TTL: Duration = Duration::from_millis(50);
+
+        let provider = ShmProviderBuilder::default_backend(65536).wait().unwrap();
+        let zbuf1: ZBuf = provider.alloc(64).wait().unwrap().into();
+        let zbuf2: ZBuf = provider.alloc(64).wait().unwrap().into();
+        let shmb1 = shmbuf_from_zbuf(&zbuf1);
+        let shmb2 = shmbuf_from_zbuf(&zbuf2);
+        let key1 = shmb1.info.metadata.clone();
+        let key2 = shmb2.info.metadata.clone();
+        drop(zbuf1);
+        drop(zbuf2);
+
+        let t0 = Instant::now();
+        let expired_deadline = t0 + SHORT_TTL;
+        let mut pending: HashMap<_, PendingShmBuf> = HashMap::new();
+        pending.insert(
+            key1,
+            PendingShmBuf {
+                buf: shmb1,
+                deadline: expired_deadline,
+            },
+        );
+
+        // Wait for TTL to expire
+        std::thread::sleep(Duration::from_millis(100));
+
+        // Trigger sweep by inserting a second entry (mirrors the TX insert logic)
+        let now = Instant::now();
+        let live_deadline = now + SHM_PENDING_TTL;
+        pending.retain(|_, v| !v.buf.is_rx_acked() && v.deadline > now);
+        pending.insert(
+            key2,
+            PendingShmBuf {
+                buf: shmb2,
+                deadline: live_deadline,
+            },
+        );
+
+        // Invariant 2: first entry was swept, second entry remains
+        assert_eq!(
+            pending.len(),
+            1,
+            "expired entry should have been swept by TTL"
+        );
+        assert!(
+            pending.values().next().unwrap().buf.is_valid(),
+            "the live entry should still be valid"
+        );
+
+        drop(provider);
+    }
+
+    /// Invariant 4: rx_ack early release — entry is removed by retain before TTL expiry.
+    #[test]
+    fn rx_ack_releases_entry_before_ttl() {
+        let provider = ShmProviderBuilder::default_backend(65536).wait().unwrap();
+        let zbuf: ZBuf = provider.alloc(64).wait().unwrap().into();
+        let shmb = shmbuf_from_zbuf(&zbuf);
+        drop(zbuf);
+
+        let key = shmb.info.metadata.clone();
+        let now = Instant::now();
+        let deadline = now + SHM_PENDING_TTL; // TTL far in the future
+
+        let mut pending: HashMap<_, PendingShmBuf> = HashMap::new();
+        pending.insert(
+            key,
+            PendingShmBuf {
+                buf: shmb,
+                deadline,
+            },
+        );
+
+        assert_eq!(pending.len(), 1, "entry should be present before ack");
+
+        // Simulate RX setting rx_ack (as read_shmbuf does after GLOBAL_CONFIRMATOR.add)
+        pending.values().next().unwrap().buf.mark_rx_acked();
+
+        // Sweep: should remove the acked entry immediately, well before TTL
+        pending.retain(|_, v| !v.buf.is_rx_acked() && v.deadline > now);
+
+        assert_eq!(
+            pending.len(),
+            0,
+            "rx_acked entry should be removed by retain sweep"
+        );
+
+        drop(provider);
+    }
+}

io/zenoh-transport/src/unicast/lowlatency/transport.rs

@@ -13,6 +13,8 @@
 //
 #[cfg(feature = "stats")]
 use std::sync::OnceLock;
+#[cfg(feature = "shared-memory")]
+use std::{collections::HashMap, sync::Mutex};
 use std::{
     sync::{Arc, RwLock as SyncRwLock},
     time::Duration,
@@ -31,7 +33,11 @@ use zenoh_protocol::{
     },
 };
 use zenoh_result::{zerror, ZResult};
+#[cfg(feature = "shared-memory")]
+use zenoh_shm::metadata::descriptor::MetadataDescriptor;
 
+#[cfg(feature = "shared-memory")]
+use crate::shm::PendingShmBuf;
 #[cfg(feature = "shared-memory")]
 use crate::shm_context::UnicastTransportShmContext;
 use crate::{
@@ -71,6 +77,9 @@ pub(crate) struct TransportUnicastLowlatency {
 
     #[cfg(feature = "shared-memory")]
     pub(super) shm_context: Option<UnicastTransportShmContext>,
+    // Per-connection SHM lease set: keyed by MetadataDescriptor for O(1) rx_ack early release.
+    #[cfg(feature = "shared-memory")]
+    pub(super) shm_pending: Arc<Mutex<HashMap<MetadataDescriptor, PendingShmBuf>>>,
 }
 
 impl TransportUnicastLowlatency {
@@ -94,6 +103,8 @@ impl TransportUnicastLowlatency {
             tracker: TaskTracker::new(),
             #[cfg(feature = "shared-memory")]
             shm_context,
+            #[cfg(feature = "shared-memory")]
+            shm_pending: Arc::new(Mutex::new(HashMap::new())),
         }) as Arc<dyn TransportUnicastTrait>
     }
 
@@ -130,6 +141,10 @@ impl TransportUnicastLowlatency {
         // to avoid concurrent new_transport and closing/closed notifications
         let mut status_guard = self.get_status().await;
         *status_guard = TransportStatus::Closed;
+        // Release all in-flight SHM leases so ConfirmedDescriptors drop and the
+        // watchdog validator can reclaim chunks within ≤100 ms.
+        #[cfg(feature = "shared-memory")]
+        self.shm_pending.lock().expect("shm_pending lock").clear();
 
         // Close and drop the link
         self.token.cancel();