test(pubsub): assert delivery ordering, not completeness, on the bounded path

transient_local_delivery_preserves_order published 500 samples through a
KeepLast(10) subscriber and asserted all 500 arrived. That asserts a
promise no RMW makes: rmw_zenoh_cpp's add_new_message drops the oldest
once message_queue_.size() >= adapted_qos_profile.depth, for every
arriving sample, with no TransientLocal exemption -- the check reads the
history policy only.

The test's stated property is ordering, and its own doc says so. Assert
that instead: a strictly increasing subsequence of what was published.
That catches reordering whether or not anything was dropped, where an
equality check conflated the two failures.

Losslessness is still covered, on the profile that actually promises it,
by keep_all_delivers_every_local_sample.

Also records in the dispatcher docs that both implementations drop
silently w.r.t. the ROS event API: upstream's MESSAGE_LOST comes from
sequence-number gaps among arriving messages, which a depth-drop cannot
produce, so bounding introduces no reporting gap.

Author: YuanYuYuan

crates/hiroz-tests/tests/reentrant_publish.rs

@@ -613,6 +613,19 @@ fn intra_closed_loop_runs_iteratively() {
 /// The shim enqueues from inside `handle_sample` — i.e. under zenoh-ext's state
 /// mutex, in exactly the order zenoh-ext chose to deliver — and a single thread
 /// pops FIFO, so the observed order must be the publish order.
+///
+/// **Ordering, not completeness.** The burst is far deeper than the declared
+/// `KeepLast` depth, so the queue drops the oldest — exactly as
+/// `rmw_zenoh_cpp`'s `add_new_message` does (`rmw_subscription_data.cpp`:
+/// `size() >= adapted_qos_profile.depth` → `pop_front()`, with no
+/// `TransientLocal` exemption). Asserting the delivered values were *all*
+/// published would therefore assert a promise no RMW makes.
+///
+/// Asserting a strictly increasing subsequence is the stronger test anyway: it
+/// catches reordering whether or not anything was dropped, whereas an equality
+/// check conflates the two failures. `keep_all_delivers_every_local_sample` in
+/// `dispatch_backpressure.rs` covers losslessness on the profile that promises
+/// it.
 #[test]
 #[serial]
 fn transient_local_delivery_preserves_order() {
@@ -655,14 +668,35 @@ fn transient_local_delivery_preserves_order() {
                 .expect("publish failed");
         }
 
-        await_deliveries(&seen, COUNT as usize);
+        // Settle rather than wait for a fixed count: with a bounded queue the
+        // delivered total is a property of scheduling, not of the publish count.
+        let mut last = 0usize;
+        let mut stable_since = Instant::now();
+        let deadline = Instant::now() + DELIVERY_TIMEOUT;
+        loop {
+            let len = seen.load(Ordering::SeqCst);
+            if len != last {
+                last = len;
+                stable_since = Instant::now();
+            } else if stable_since.elapsed() >= Duration::from_millis(300) {
+                break;
+            }
+            assert!(Instant::now() < deadline, "delivery never settled");
+            thread::sleep(Duration::from_millis(25));
+        }
 
         let received = received.lock().unwrap();
-        let expected: Vec<u64> = (0..COUNT).collect();
-        assert_eq!(
-            &received[..COUNT as usize],
-            &expected[..],
-            "delivery thread reordered samples"
+        assert!(
+            !received.is_empty(),
+            "nothing was delivered — the scenario proved nothing"
+        );
+        assert!(
+            received.windows(2).all(|w| w[0] < w[1]),
+            "delivery thread reordered samples: {received:?}"
+        );
+        assert!(
+            received.iter().all(|&c| c < COUNT),
+            "delivered a counter that was never published: {received:?}"
         );
     });
 }

crates/hiroz/src/pubsub.rs

@@ -342,15 +342,24 @@ impl DispatchQueue {
 ///   survives, so the ordering objection that applies to the advanced path does
 ///   not apply here.
 ///
-/// * **Advanced path — same bound, and the same expression.** A `TransientLocal`
-///   subscriber exists to replay history and to recover samples flagged as
-///   missed, so dropping discards data zenoh-ext went out of its way to fetch
-///   and breaks the reordering contract mid-flight — a single
-///   `deliver_and_flush` enqueues several back-to-back samples whose contiguity
-///   is the point. That argument is why `KeepAll` maps to
-///   [`DISPATCH_UNBOUNDED`]. It is *not* a reason to ignore a declared
-///   `KeepLast(depth)`, which is what an unbounded capacity on every profile
-///   amounted to.
+/// * **Advanced path — same bound, and the same expression.** This matches
+///   `rmw_zenoh_cpp`: `SubscriptionData::add_new_message` drops the oldest once
+///   `message_queue_.size() >= adapted_qos_profile.depth`, for every arriving
+///   sample, with **no `TransientLocal` exemption** — the check reads the
+///   history policy only. Its advanced-subscriber cache is sized the same way
+///   (`adv_sub_opts.history->max_samples = qos_.depth`).
+///
+///   An earlier revision left this path unbounded on *every* profile, reasoning
+///   that dropping discards what miss-detection recovered. That is why `KeepAll`
+///   maps to [`DISPATCH_UNBOUNDED`] — but applied to a declared
+///   `KeepLast(depth)` it ignores the QoS, and since [`Self::always_shim`]
+///   enqueues remote samples too, it also traded zenoh's transport backpressure
+///   for unbounded growth.
+///
+///   Both implementations drop **silently** as far as the ROS event API is
+///   concerned: upstream's `MESSAGE_LOST` is raised from *sequence-number gaps*
+///   among arriving messages, which a depth-drop cannot produce. The escalating
+///   `warn!` below is strictly more visible than upstream's debug log.
 ///
 /// Two consequences are worth stating rather than discovering.
 ///