test(rtps): interop cases for lease liveliness, many participants, and the CycloneDDS addressing fixes

.github/workflows/rtps_interop.yml

@@ -28,7 +28,7 @@ concurrency:
 jobs:
   interop:
     runs-on: ubuntu-latest
-    timeout-minutes: 30
+    timeout-minutes: 40
     steps:
       - uses: actions/checkout@v7
         with:

components/rtps/interop/README.md

@@ -28,7 +28,20 @@ touched.
 | espp_pub->ros2_echo | espp reliable writer -> ROS 2 subscriber (`rt/chatter`, `std_msgs::msg::dds_::String_`) |
 | ros2_pub->espp_sub | ROS 2 publisher -> espp reliable reader |
 | espp_be_pub->ros2_be_echo | best-effort pairing |
-| cyclone: ... | the ROS 2 pub/sub pairs (reliable and best-effort, both directions) again on `rmw_cyclonedds_cpp`. Services are not covered: rmw_cyclonedds_cpp puts a `{guid, seq}` request header in the payload instead of `related_sample_identity` |
+| ros2_pub->espp_sub_liveliness, espp_pub->ros2_sub_liveliness | samples keep arriving 22 s in, i.e. past a remote participant lease (CycloneDDS: 10 s). Only samples after that point count (`rtps_interop_sub ... count_after_s`, `ros2_count_subscriber.py`), so a peer whose SPDP re-announcements were not delivered - and whose lease therefore expired, dropping every matched endpoint - fails. CycloneDDS addresses those re-announcements to readerId ENTITYID_UNKNOWN; the FastDDS pass is the control. Only the espp-publisher leg detects a dropped peer: a HEARTBEAT that reaches a matched espp reader also refreshes the sending participant's liveliness (`MessageReceiver::processHeartbeat`), so a CycloneDDS *writer* keeps its participant alive through its heartbeats even when its SPDP re-announcements are lost, while an espp writer only gets ACKNACKs back, which do not refresh it, and the peer expires |
+| many_participants_late_joiner | twelve unrelated ROS 2 participants are on the domain before espp starts, then the publisher espp needs joins last and must still be matched. Guards the remote-participant pool (`SPDP_MAX_NUMBER_FOUND_PARTICIPANTS`, 32 on the host profile built here, 10 on the embedded profile - raise it via menuconfig for a domain this busy): a participant beyond the pool is ignored silently |
+| cyclone: ... | the ROS 2 pub/sub pairs (reliable and best-effort, both directions) and the two cases above again on `rmw_cyclonedds_cpp`. Services are not covered: rmw_cyclonedds_cpp puts a `{guid, seq}` request header in the payload instead of `related_sample_identity` |
+
+## Not covered (yet): a multi-homed peer
+
+A FastDDS or CycloneDDS peer on several networks announces one unicast locator
+per interface, and espp must pick the one on its own network (`locator_subnet`
+checks the selection logic on canned announcements). Reproducing it live needs
+a peer address espp cannot reach, which a single shared network namespace cannot
+provide: every local address is reachable there. The design would be a second
+Docker network, the ROS 2 side attached to both and espp to one, so the ROS 2
+participant announces a locator on the network espp is not on; that is a
+two-container `run.sh` and is left for a follow-up.
 
 ## Notes
 

components/rtps/interop/ros2_count_subscriber.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""Count std_msgs/String samples from a ROS 2 subscriber for a fixed time.
+
+Used by the interop harness as the ROS 2 side of the liveliness cases:
+`ros2 topic echo --once` proves one sample arrived, this proves samples KEEP
+arriving - samples received earlier than `count_after_s` seconds after start
+are reported but not counted, so a peer that stops being served once its
+participant lease has expired cannot pass.
+
+Usage: ros2_count_subscriber.py TOPIC [--seconds S] [--required N] [--count-after S]
+       [--reliability reliable|best_effort]
+Exits 0 iff at least REQUIRED samples arrived after COUNT_AFTER seconds, 2 on
+bad arguments.
+"""
+
+import argparse
+import sys
+import time
+
+import rclpy
+from rclpy.qos import QoSProfile, ReliabilityPolicy
+from std_msgs.msg import String
+
+
+def non_negative(text: str) -> float:
+    value = float(text)
+    if value < 0:
+        raise argparse.ArgumentTypeError(f"must be >= 0, got {text}")
+    return value
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Count std_msgs/String samples on a topic for a fixed time; "
+        "samples before --count-after are reported but not counted."
+    )
+    parser.add_argument("topic", nargs="?", default="/chatter")
+    parser.add_argument("--seconds", type=non_negative, default=25.0, help="how long to listen")
+    parser.add_argument("--required", type=int, default=10, help="counted samples needed to pass")
+    parser.add_argument(
+        "--count-after", type=non_negative, default=0.0, help="seconds before counting starts"
+    )
+    parser.add_argument(
+        "--reliability", choices=("reliable", "best_effort"), default="reliable"
+    )
+    args = parser.parse_args(argv)
+    if args.required < 0:
+        parser.error(f"--required must be >= 0, got {args.required}")
+    return args
+
+
+def main() -> int:
+    args = parse_args(sys.argv[1:])
+    topic = args.topic
+    seconds = args.seconds
+    required = args.required
+    count_after_s = args.count_after
+    reliable = args.reliability == "reliable"
+
+    rclpy.init()
+    node = rclpy.create_node("countsub")
+    qos = QoSProfile(depth=10)
+    qos.reliability = ReliabilityPolicy.RELIABLE if reliable else ReliabilityPolicy.BEST_EFFORT
+    start = time.time()
+    counts = {"early": 0, "counted": 0}
+
+    def on_sample(_msg: String) -> None:
+        if time.time() - start < count_after_s:
+            counts["early"] += 1
+        else:
+            counts["counted"] += 1
+            if counts["counted"] <= 3 or counts["counted"] % 25 == 0:
+                print(f"received {counts['counted']} (t={time.time() - start:.1f}s)", flush=True)
+
+    node.create_subscription(String, topic, on_sample, qos)
+    while time.time() - start < seconds:
+        rclpy.spin_once(node, timeout_sec=0.1)
+
+    ok = counts["counted"] >= required
+    print(
+        f"{'PASS' if ok else 'FAIL'} received={counts['counted']} required={required} "
+        f"(not counted, before {count_after_s:.0f}s: {counts['early']})",
+        flush=True,
+    )
+    node.destroy_node()
+    rclpy.shutdown()
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

components/rtps/interop/run_interop.sh

@@ -22,6 +22,73 @@ result() { # name exit_code
   else echo "RESULT FAIL: $1"; FAIL=$((FAIL+1)); fi
 }
 
+# --- Liveliness past the remote participant lease ----------------------------
+# A peer keeps re-announcing itself over SPDP; if those re-announcements are not
+# delivered to the SPDP reader, espp lets the peer's lease expire (10 s for
+# CycloneDDS, which addresses the repeats to readerId ENTITYID_UNKNOWN) and
+# drops every matched endpoint with it - discovery looked fine, data stopped
+# after the lease. `ros2 topic echo --once` cannot see that, so both legs here
+# only count samples that arrive from 22 s in: discovery (a few seconds), the
+# 10 s lease and the liveliness check cadence all fit before that, so a
+# dropped peer contributes nothing to the count (count_after_s, see
+# rtps_interop_sub / ros2_count_subscriber.py). The FastDDS pass is the
+# control: it names the SPDP reader in every announcement.
+lease_liveliness_cases() { # prefix log-tag
+  local prefix="$1" tag="$2"
+  note "${prefix}ROS 2 publisher -> espp subscriber, samples still arriving 22 s in (liveliness past the lease)"
+  "$BIN"/rtps_interop_sub rt/lease std_msgs::msg::dds_::String_ 1 10 34 "" 0 2 22 > "/tmp/${tag}leasesub.log" 2>&1 &
+  local sub=$!
+  sleep 3
+  timeout 36 ros2 topic pub -r 5 /lease std_msgs/msg/String "data: 'still alive'" > "/tmp/${tag}leaserospub.log" 2>&1 &
+  local rospub=$!
+  wait $sub
+  local sub_rc=$?
+  kill $rospub 2>/dev/null; wait $rospub 2>/dev/null || true
+  tail -2 "/tmp/${tag}leasesub.log"
+  result "${prefix}ros2_pub->espp_sub_liveliness" $sub_rc
+
+  note "${prefix}espp publisher -> ROS 2 subscriber, samples still arriving 22 s in (liveliness past the lease)"
+  "$BIN"/rtps_interop_pub rt/lease2 std_msgs::msg::dds_::String_ 1 220 200 > "/tmp/${tag}leasepub.log" 2>&1 &
+  local pub=$!
+  sleep 3
+  timeout 40 python3 /work/components/rtps/interop/ros2_count_subscriber.py /lease2 --seconds 34 --required 10 --count-after 22 --reliability reliable \
+    > "/tmp/${tag}leasecount.log" 2>&1
+  local count_rc=$?
+  kill $pub 2>/dev/null; wait $pub 2>/dev/null || true
+  tail -2 "/tmp/${tag}leasecount.log"
+  result "${prefix}espp_pub->ros2_sub_liveliness" $count_rc
+}
+
+# --- A dozen participants on the domain, the publisher joins last -------------
+# Every ROS 2 process is a DDS participant, and espp tracks remote participants
+# in a fixed pool (SPDP_MAX_NUMBER_FOUND_PARTICIPANTS: 32 on the host profile
+# this harness builds, 10 on the embedded profile). A participant beyond the
+# pool is ignored silently - never matched, its data dropped. With twelve
+# unrelated participants already on the domain, the publisher espp actually
+# needs joins last and must still be matched. On an ESP32 the same scenario
+# needs the pool raised (menuconfig RTPS -> Custom capacity overrides).
+many_participants_case() { # prefix log-tag
+  local prefix="$1" tag="$2"
+  note "${prefix}espp subscriber with twelve other participants on the domain; the publisher joins last"
+  local crowd=()
+  local i
+  for i in $(seq 1 12); do
+    timeout 45 ros2 topic pub -r 1 "/crowd$i" std_msgs/msg/String "data: 'crowd'" > /dev/null 2>&1 &
+    crowd+=("$!")
+  done
+  sleep 8 # let the crowd come up (each is its own participant) before espp starts
+  "$BIN"/rtps_interop_sub rt/crowd std_msgs::msg::dds_::String_ 1 3 30 > "/tmp/${tag}crowdsub.log" 2>&1 &
+  local sub=$!
+  sleep 3
+  timeout 32 ros2 topic pub -r 5 /crowd std_msgs/msg/String "data: 'late joiner'" > "/tmp/${tag}crowdpub.log" 2>&1 &
+  local pub=$!
+  wait $sub
+  local sub_rc=$?
+  kill $pub "${crowd[@]}" 2>/dev/null; wait $pub "${crowd[@]}" 2>/dev/null || true
+  tail -2 "/tmp/${tag}crowdsub.log"
+  result "${prefix}many_participants_late_joiner" $sub_rc
+}
+
 note "Build espp lib + host binaries (linux)"
 cmake -S lib -B lib/build -DCMAKE_BUILD_TYPE=Release -DESPP_INSTALL=ON -DCMAKE_INSTALL_PREFIX=/tmp/espp/install > /tmp/cmake_lib.log 2>&1 \
   && cmake --build lib/build -j"$(nproc)" --target install > /tmp/build_lib.log 2>&1 \
@@ -233,6 +300,9 @@ kill $ROSBAND_PID 2>/dev/null; wait $ROSBAND_PID 2>/dev/null
 cat /tmp/subband.log
 result "ros2_pub->espp_banded_sub" $subband_rc
 
+lease_liveliness_cases "" "f"
+many_participants_case "" "f"
+
 note "espp best-effort publisher -> ROS 2 best-effort subscriber"
 "$BIN"/rtps_interop_pub rt/chatter std_msgs::msg::dds_::String_ 0 60 200 > /tmp/pub2.log 2>&1 &
 ESPP_PID=$!
@@ -354,6 +424,10 @@ result "espp_action_client->ros2_server" $actclient_rc
 # that espp's services (and rmw_fastrtps_cpp) use.
 note "CycloneDDS (rmw_cyclonedds_cpp): ROS 2 pub/sub pairs"
 export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
+# CycloneDDS caps the participants one host may create on a domain through its
+# participant-index range (MaxAutoParticipantIndex, default 9 = ten
+# participants); the busy-domain case below creates thirteen, so lift it.
+export CYCLONEDDS_URI='<CycloneDDS><Domain><Discovery><MaxAutoParticipantIndex>60</MaxAutoParticipantIndex></Discovery></Domain></CycloneDDS>'
 ros2 daemon stop > /dev/null 2>&1
 
 "$BIN"/rtps_interop_pub rt/chatter std_msgs::msg::dds_::String_ 1 60 200 > /tmp/cpub1.log 2>&1 &
@@ -395,10 +469,14 @@ kill $ROS_PID 2>/dev/null; wait $ROS_PID 2>/dev/null
 tail -2 /tmp/csub2.log
 result "cyclone: ros2_be_pub->espp_be_sub" $csub2_rc
 
+lease_liveliness_cases "cyclone: " "c"
+many_participants_case "cyclone: " "c"
+
 if grep -l "malformed packet received from vendor" /tmp/c*.log > /dev/null 2>&1; then
   echo "CycloneDDS reported malformed packets from espp:"; grep -h "malformed packet" /tmp/c*.log | head -3
 fi
 export RMW_IMPLEMENTATION=rmw_fastrtps_cpp
+unset CYCLONEDDS_URI
 
 # --- Python bindings: functional round-trip of every RMI/AMI mechanism --------
 # Build the espp wheel in-container and run the demo (multicast discovery works

pc/tests/rtps_interop_sub.cpp

@@ -6,7 +6,7 @@
 // std_msgs/String on /chatter.
 //
 // Usage: rtps_interop_sub [topic] [type] [reliable(0|1)] [required] [timeout_s]
-// [interface_ip] [payload_bytes] [band]
+// [interface_ip] [payload_bytes] [band] [count_after_s]
 // When payload_bytes > 0, each received String is verified byte-exact against the
 // deterministic payload_bytes-long pattern (proving fragmented >64 KB samples are
 // reassembled correctly); only byte-exact receptions count toward `required`.
@@ -15,7 +15,15 @@
 // announced via its SEDP per-endpoint unicast locator - the interop matrix uses
 // this to prove a FastDDS/ROS 2 peer honors that locator and delivers the
 // topic's traffic to the dedicated port.
-// Exits 0 once `required` samples arrive within `timeout_s`.
+// count_after_s (default 0, must be >= 0 - a negative value is a usage error
+// and exits 1): samples that arrive earlier than this many seconds after start
+// are reported (and, in large-payload mode, still verified byte-exact) but
+// NOT counted toward `required`. The interop
+// matrix uses this as a liveliness gate: a peer whose participant we dropped
+// (e.g. its lease expired because its re-announcements were not delivered)
+// stops being matched, so nothing counts after the lease - only samples that
+// keep arriving past that point prove the peer stayed alive.
+// Exits 0 once `required` (counted) samples arrive within `timeout_s`.
 
 #include <atomic>
 #include <chrono>
@@ -61,6 +69,11 @@ int main(int argc, char **argv) {
   const char *interface_ip = (argc > 6) ? argv[6] : ""; // "" -> auto-detect
   const std::size_t payload_bytes = (argc > 7) ? std::strtoul(argv[7], nullptr, 10) : 0;
   const int band_arg = (argc > 8) ? std::atoi(argv[8]) : static_cast<int>(espp::QosBand::Normal);
+  const int count_after_s = (argc > 9) ? std::atoi(argv[9]) : 0;
+  if (count_after_s < 0) {
+    std::printf("FAIL: count_after_s must be >= 0, got %d\n", count_after_s);
+    return 1;
+  }
   // Validate before casting: an out-of-range value would index band arrays.
   if (band_arg < static_cast<int>(espp::QosBand::Critical) ||
       band_arg > static_cast<int>(espp::QosBand::Low)) {
@@ -72,6 +85,9 @@ int main(int argc, char **argv) {
   const std::string expected = payload_bytes > 0 ? make_pattern(payload_bytes) : std::string{};
 
   std::atomic<int> received{0};
+  std::atomic<int> early{0}; // arrived before count_after_s; reported, not counted
+  const auto start = std::chrono::steady_clock::now();
+  const auto count_from = start + std::chrono::seconds(count_after_s);
 
   espp::RtpsParticipant participant({
       .interface_address = interface_ip,
@@ -87,11 +103,15 @@ int main(int argc, char **argv) {
           .type_name = type,
           .reliability = reliable ? Reliability::RELIABLE : Reliability::BEST_EFFORT,
           .on_sample =
-              [&received, payload_bytes, &expected](std::span<const uint8_t> cdr_payload) {
+              [&received, &early, count_from, payload_bytes,
+               &expected](std::span<const uint8_t> cdr_payload) {
                 auto msg = cdr::deserialize<StringMsg>(std::as_bytes(cdr_payload));
                 if (!msg) {
                   return;
                 }
+                // Early samples (before count_after_s) are verified and
+                // reported exactly like counted ones; only the count differs.
+                const bool counts = std::chrono::steady_clock::now() >= count_from;
                 if (payload_bytes > 0) {
                   // Large-payload mode: only a byte-exact reassembly counts.
                   const bool exact = msg->data.size() == payload_bytes && msg->data == expected;
@@ -101,10 +121,23 @@ int main(int argc, char **argv) {
                     std::fflush(stdout);
                     return;
                   }
+                  if (!counts) {
+                    const int n = early.fetch_add(1) + 1;
+                    std::printf("received (not counted yet) %d: %zu bytes byte-exact\n", n,
+                                msg->data.size());
+                    std::fflush(stdout);
+                    return;
+                  }
                   const int n = received.fetch_add(1) + 1;
                   std::printf("received %d: %zu bytes byte-exact\n", n, msg->data.size());
                   std::fflush(stdout);
                 } else {
+                  if (!counts) {
+                    const int n = early.fetch_add(1) + 1;
+                    std::printf("received (not counted yet) %d: '%s'\n", n, msg->data.c_str());
+                    std::fflush(stdout);
+                    return;
+                  }
                   const int n = received.fetch_add(1) + 1;
                   std::printf("received %d: '%s'\n", n, msg->data.c_str());
                   std::fflush(stdout);
@@ -115,17 +148,18 @@ int main(int argc, char **argv) {
     std::printf("FAIL: add_reader\n");
     return 1;
   }
-  std::printf("interop_sub: topic=%s type=%s reliable=%d required=%d timeout=%ds band=%d\n", topic,
-              type, reliable ? 1 : 0, required, timeout_s, band_arg);
+  std::printf("interop_sub: topic=%s type=%s reliable=%d required=%d timeout=%ds band=%d "
+              "count_after=%ds\n",
+              topic, type, reliable ? 1 : 0, required, timeout_s, band_arg, count_after_s);
 
-  const auto start = std::chrono::steady_clock::now();
   while (received.load() < required &&
          std::chrono::steady_clock::now() - start < std::chrono::seconds(timeout_s)) {
     std::this_thread::sleep_for(100ms);
   }
 
   const int n = received.load();
-  std::printf("%s received=%d required=%d\n", n >= required ? "PASS" : "FAIL", n, required);
+  std::printf("%s received=%d required=%d (not counted, before %ds: %d)\n",
+              n >= required ? "PASS" : "FAIL", n, required, count_after_s, early.load());
   participant.stop();
   return n >= required ? 0 : 1;
 }