test(rtps): fault injection beyond loss and a heap soak in the QEMU matrix; best-effort readers reject duplicates

.github/workflows/rtps_qemu.yml

@@ -6,8 +6,9 @@ name: RTPS interop in QEMU (esp32 target)
 # log asserted on. The host-build matrix (rtps_interop.yml) covers the same
 # peers against the desktop build; this one covers what only the target can
 # show: the embedded pools and knobs, lwIP, FreeRTOS scheduling, fragmentation
-# over the real stack, delivery under injected loss, heap over time.
-# See components/rtps/interop/README.md.
+# over the real stack, delivery under injected loss / reordering / duplication
+# / delay / corruption, and the heap over minutes of churn (the soak).
+# One firmware build, two suites in parallel. See components/rtps/interop/README.md.
 
 permissions:
   contents: read
@@ -30,9 +31,9 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  qemu-interop:
+  firmware:
     runs-on: ubuntu-latest
-    timeout-minutes: 60
+    timeout-minutes: 20
     steps:
       - uses: actions/checkout@v7
         with:
@@ -47,10 +48,32 @@ jobs:
             idf.py -DSDKCONFIG_DEFAULTS="sdkconfig.defaults;sdkconfig.ci.defaults" build
             && cd build
             && python -m esptool --chip esp32 merge-bin --fill-flash-size 4MB -o qemu_flash.bin @flash_args
-      - name: Run the QEMU interop matrix
+      - uses: actions/upload-artifact@v4
+        with:
+          name: qemu-flash
+          path: components/rtps/interop/qemu/build/qemu_flash.bin
+          if-no-files-found: error
+
+  qemu:
+    needs: firmware
+    runs-on: ubuntu-latest
+    timeout-minutes: 75
+    strategy:
+      fail-fast: false
+      matrix:
+        suite: [matrix, soak]
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          submodules: "recursive"
+      - uses: actions/download-artifact@v4
+        with:
+          name: qemu-flash
+          path: components/rtps/interop/qemu/build
+      - name: Run the QEMU ${{ matrix.suite }} suite
         run: |
           cd components/rtps/interop
-          ./run_qemu.sh "$PWD/qemu/build/qemu_flash.bin"
+          ./run_qemu.sh "$PWD/qemu/build/qemu_flash.bin" ${{ matrix.suite }}
       - name: Device log (tail)
         if: always()
         run: |
@@ -60,6 +83,6 @@ jobs:
         if: always()
         uses: actions/upload-artifact@v4
         with:
-          name: qemu-device-log
+          name: qemu-device-log-${{ matrix.suite }}
           path: components/rtps/interop/qemu/build/qemu-serial.log
           if-no-files-found: ignore

components/rtps/README.md

@@ -10,7 +10,7 @@ the network over the standard RTPS wire protocol.
 
 It provides three messaging patterns, all validated against live ROS 2 (Jazzy):
 
-- **Pub/sub** — topic-based, best-effort or reliable (HEARTBEAT/ACKNACK).
+- **Pub/sub** — topic-based, best-effort or reliable (HEARTBEAT/ACKNACK). A best-effort reader never asks for a resend, but still rejects a duplicated or reordered sample (anything not above the highest sequence number already received from that writer, RTPS 2.5 8.4.12.1).
 - **Services (RMI)** — request/reply with correlated responses.
 - **Actions (AMI)** — long-running goals with feedback, result, and cancellation.
 
@@ -86,8 +86,10 @@ copying from `example/sdkconfig.defaults`: IP reassembly is **off**
 (`CONFIG_LWIP_IP4_REASSEMBLY`), and FastDDS / CycloneDDS send any sample under
 64 KB as one IP-fragmented datagram, so without it a sample bigger than one MTU
 never reaches the engine; and the socket mailboxes (`CONFIG_LWIP_UDP_RECVMBOX_SIZE`,
-6 by default) drop the tail of a burst of fragments. Both came out of the QEMU
-interop matrix (`interop/qemu`).
+6 by default) drop the tail of a burst of fragments. A third: lwIP does not
+verify UDP checksums on receive by default (`CONFIG_LWIP_CHECKSUM_CHECK_UDP`),
+so a corrupted datagram reaches the engine as valid data. All three came out
+of the QEMU interop matrix (`interop/qemu`).
 
 Services and actions are **pure library code** over pub/sub — the only wire
 addition is a `related_sample_identity` inline QoS on service request/reply (for

components/rtps/example/sdkconfig.defaults

@@ -39,3 +39,10 @@ CONFIG_LWIP_IP4_REASSEMBLY=y
 CONFIG_LWIP_IP_REASS_MAX_PBUFS=48
 CONFIG_LWIP_UDP_RECVMBOX_SIZE=64
 CONFIG_LWIP_TCPIP_RECVMBOX_SIZE=64
+
+# lwIP does not verify UDP (or IP header) checksums on receive by default
+# (found by the corruption case of the QEMU interop matrix: the device parsed
+# flipped bits as samples). Ethernet's frame check catches wire corruption,
+# this guards against the rest, for a software checksum per received datagram.
+CONFIG_LWIP_CHECKSUM_CHECK_IP=y
+CONFIG_LWIP_CHECKSUM_CHECK_UDP=y

components/rtps/include/rtps/entities/StatelessReader.hpp

@@ -28,6 +28,8 @@ Author: i11 - Embedded Software, RWTH Aachen University
 
 #include "rtps/entities/Reader.hpp"
 
+#include <mutex>
+
 namespace rtps {
 class StatelessReader final : public Reader {
 public:
@@ -36,6 +38,13 @@ class StatelessReader final : public Reader {
   bool onNewHeartbeat(const SubmessageHeartbeat &msg, const GuidPrefix_t &remotePrefix) override;
   bool addNewMatchedWriter(const WriterProxy &newProxy) override;
   bool onNewGapMessage(const SubmessageGap &msg, const GuidPrefix_t &remotePrefix) override;
+
+private:
+  /// Serializes newChange()'s claim of a sequence number and the callbacks it
+  /// delivers, so samples reach the application in claim order even when
+  /// receive workers for different sockets hand them in concurrently (see
+  /// StatefulReader::m_delivery_mutex).
+  std::mutex m_delivery_mutex;
 };
 
 } // namespace rtps

components/rtps/include/rtps/entities/WriterProxy.hpp

@@ -48,6 +48,9 @@ struct WriterProxy {
       , is_reliable(reliable)
       , remoteLocator(loc) {}
 
+  /// Sequence numbers one ACKNACK requests at most (see getMissing()).
+  static constexpr uint32_t kMaxNackSns = 32;
+
   // For now, we don't store any packets, so we just request all starting from
   // the next expected
   SequenceNumberSet getMissing(const SequenceNumber_t &firstAvail,
@@ -60,8 +63,21 @@ struct WriterProxy {
       set.base = expectedSN;
       SequenceNumber_t i;
       uint32_t bit;
-      for (bit = 0, i = expectedSN; i <= lastAvail && bit < SNS_MAX_NUM_BITS; i++, bit++) {
-        set.bitMap[0] |= uint32_t{1} << (31 - bit);
+      // One bit per missing SN, in the bucket the bit falls in (the set is
+      // SNS_MAX_NUM_BITS wide, 32 per bucket). This used to write bucket 0
+      // only, shifting by (31 - bit) for every bit: undefined past bit 31.
+      //
+      // At most kMaxNackSns per ACKNACK, deliberately: every SN requested is
+      // resent at once, and a reader behind by hundreds (joining a flood,
+      // say) that asked for the whole set on every heartbeat - there is one
+      // after every sample - raised a resend storm the receiving socket could
+      // not drain: the kernel dropped each new burst at its tail, where the
+      // one sample the reader was waiting for sat, and it never advanced
+      // (rtps_banded_churn starved once the bitmap was whole). 32 back to
+      // back fits the ESP32's socket mailbox and the emulated MAC's ring (64
+      // each); the next heartbeat requests the next 32.
+      for (bit = 0, i = expectedSN; i <= lastAvail && bit < kMaxNackSns; i++, bit++) {
+        set.bitMap[bit / 32] |= uint32_t{1} << (31 - (bit % 32));
         set.numBits++;
       }
     }

components/rtps/interop/README.md

@@ -62,8 +62,23 @@ cd .. && ./run_qemu.sh
 | espp_host_pub->device_sub, device_pub->espp_host_sub, espp_host_frag_30k->device_sub | the host engine against the embedded one, including its DATA_FRAG path |
 | ros2_pub->device_sub_frag_30k (wire summary) | tcpdump on the tap during the case: frames per 30 KB sample, fragments, multicast, the biggest burst (FastDDS 42 frames within 20 ms, CycloneDDS 23: why the RX ring is 64) |
 | netem_loss_reliable | 15 % loss + 10 ms jitter on the link (`tc netem`): the reliable reader still delivers |
+| netem_reorder_reliable, netem_duplicate_reliable, netem_corrupt_reliable | 30 % reordering, 30 % duplication, 2 % corruption on the link: the host engine's numbered samples reach the device's reliable reader every one exactly once and in order |
+| netem_reorder_best_effort, netem_duplicate_best_effort | the same against the best-effort reader: samples may be lost, but none arrives twice or out of order (RTPS 2.5 8.4.12.1, `rtps_best_effort_order` on the host) |
+| netem_delay_discovery | 300 ms (+-50) delay from before discovery: a ROS 2 publisher is still discovered, matched and delivered over a 600 ms round trip |
+| netem_loss_device_pub | 15 % loss towards the device: the device's reliable writer repairs for the host subscriber against lost ACKNACKs |
 | device_heap_stable, device_no_crash | free heap at the end within 96 KB of the start (two buffers grow once to the largest sample, 30 KB each); no panic / abort / assert |
 
+The **soak** suite (`./run_qemu.sh <image> soak`, the second CI job on the same
+firmware build) runs 20 cycles of churn instead: each cycle six ROS 2
+participants and two host-engine ones come, match the device's endpoints every
+way they can (reliable and best-effort in, the device's publishers out, a 30 KB
+sample, a service call) and go, FastDDS and CycloneDDS alternating, with the
+previous cycle's leases running out part way into the next. After a warm-up
+cycle (whatever grows once to the largest sample grows there) the free heap at
+the end must be within 8 KB of the baseline, the low-water mark within 16 KB,
+and the last cycle must still have delivered (`soak_heap_no_leak`,
+`soak_min_heap_stable`, `soak_still_delivering`). The per-cycle heap is printed.
+
 What only this can show: the embedded pools and knobs as built, lwIP's socket
 and reassembly behaviour, FreeRTOS priorities for the engine's threads,
 delivery under injected loss, heap over a run on the target allocator, and a

components/rtps/interop/qemu/sdkconfig.defaults

@@ -50,3 +50,11 @@ CONFIG_LWIP_TCPIP_RECVMBOX_SIZE=64
 CONFIG_RTPS_LIMITS_PROFILE_EMBEDDED=y
 CONFIG_RTPS_ENABLE_FRAGMENTATION=y
 CONFIG_RTPS_ENABLE_RPC=y
+
+# ESP-IDF's lwIP does not verify UDP (or IP header) checksums on receive by
+# default: a corrupted datagram reaches the engine as valid data (found by the
+# corruption case of the QEMU matrix, where the device parsed flipped bits as
+# samples). Ethernet's frame check catches wire corruption, so this guards
+# against the rest; it costs a software checksum per received datagram.
+CONFIG_LWIP_CHECKSUM_CHECK_IP=y
+CONFIG_LWIP_CHECKSUM_CHECK_UDP=y

components/rtps/interop/run_interop.sh

@@ -104,7 +104,7 @@ cmake -S lib -B lib/build -DCMAKE_BUILD_TYPE=Release -DESPP_INSTALL=ON -DCMAKE_I
        rtps_sedp_dedicated_locator rtps_banded_pubsub rtps_banded_deferred rtps_banded_ration \
        rtps_banded_churn rtps_service_rollback rtps_deferred_recovery rtps_guaranteed_submit \
        rtps_remove_reader_deadlock rtps_guaranteed_fairness rtps_writer_churn \
-       rtps_stateless_saturation rtps_callback_reentrancy \
+       rtps_stateless_saturation rtps_callback_reentrancy rtps_best_effort_order rtps_acknack_bitmap \
        rtps_interop_pub rtps_interop_sub > /tmp/build.log 2>&1
 build_rc=$?
 result "build" $build_rc
@@ -172,6 +172,8 @@ timeout 90 "$BIN"/rtps_stateless_saturation; result "stateless_saturation" $?
 # callback (and freeing its arg) must prevent the stale snapshot entry from
 # being invoked.
 timeout 30 "$BIN"/rtps_callback_reentrancy; result "callback_reentrancy" $?
+timeout 30 "$BIN"/rtps_best_effort_order; result "best_effort_order" $?
+timeout 30 "$BIN"/rtps_acknack_bitmap; result "acknack_bitmap" $?
 
 # The same saturation test against a STATIC-storage build of the engine with a
 # 2-slot best-effort history: this exercises the KEEP_LAST overwrite path (the

components/rtps/interop/run_qemu.sh

@@ -4,7 +4,8 @@
 # in QEMU inside the container, against FastDDS, CycloneDDS / ROS 2 and the espp
 # host engine on a tap device.
 #
-# Usage: ./run_qemu.sh [path/to/qemu_flash.bin]   (from components/rtps/interop)
+# Usage: ./run_qemu.sh [path/to/qemu_flash.bin] [matrix|soak]   (from components/rtps/interop)
+#   matrix (default): the interop cases + fault injection; soak: minutes of churn, heap asserted
 #
 # The firmware image defaults to qemu/build/qemu_flash.bin, i.e.
 #   cd qemu && idf.py -DSDKCONFIG_DEFAULTS="sdkconfig.defaults;sdkconfig.ci.defaults" build \
@@ -24,4 +25,4 @@ case "$FLASH" in "$REPO_ROOT"/*) FLASH_IN="/work${FLASH#"$REPO_ROOT"}" ;; *) ech
 docker build -t espp-rtps-interop .
 exec docker run --rm --network none --cap-add NET_ADMIN --device /dev/net/tun \
   -v "$REPO_ROOT":/work espp-rtps-interop \
-  bash /work/components/rtps/interop/run_qemu_interop.sh "$FLASH_IN"
+  bash /work/components/rtps/interop/run_qemu_interop.sh "$FLASH_IN" "${2:-matrix}"