Stop sensor subscribe-check timer once subscribed (fix #23 rmw_zenoh ERROR spam)

.agent/work-plans/issue-23/progress.md

@@ -0,0 +1,39 @@
+---
+issue: 23
+---
+
+# Issue #23 — subscribeCheck timer re-subscribes every second (ROS 1 one-shot semantics lost in ROS 2 port)
+
+## Issue Review
+**Status**: complete
+**When**: 2026-05-27 19:47 -04:00
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+
+**Issue**: #23
+**Comment**: https://github.com/rolker/mru_transform/issues/23#issuecomment-4559555297
+**Scope verdict**: well-scoped
+
+### Actions
+- [ ] Ship the regression test (subscribe() runs once, timer cancels) in the same PR — do not defer.
+- [ ] Confirm no package docs describe subscribe/retry behavior; update any that do.
+- [ ] Maintainer decision: approach A (minimal cancel) vs B (restructure to single repeating timer + cancel) before implementing.
+- [ ] Make subscribe-once semantics a conscious choice; if dynamic type-switching is ever wanted, open a separate issue (don't bury).
+- [ ] Add null-safety on the `subscribe_check_timer_->cancel()` call.
+- [ ] Decide whether to include the optional "already-subscribed" guard in `subscribe()` (defense-in-depth; keep change minimal if not).
+
+## Local Review (Pre-Push)
+**Status**: complete
+**When**: 2026-05-27 22:17 -04:00
+**By**: Claude Code Agent (Claude Opus 4.7 (1M context))
+**Verdict**: approved
+
+**Branch**: feature/issue-23 at `a8baec6`
+**Mode**: pre-push
+**Depth**: Standard (reason: shared sensor base header + subscription lifecycle, ~180 lines across 6 files)
+**Must-fix**: 0 | **Suggestions**: 1
+
+Specialists: Static Analysis (cppcheck; cpplint unavailable), Governance (lead), Claude Adversarial, Copilot Adversarial. Plan Drift skipped (no plan.md). Two Copilot "must-fix" claims verified as false positives (the test's explicit `ASSERT_EQ` after the wait loop catches "never subscribed"; `spin_for` gates on `steady_clock < deadline` so it runs the full duration). Cppcheck style hits were all on pre-existing context lines, not touched here.
+
+### Findings
+- [ ] (suggestion) Surface intentional loss of runtime type-switching in PR body — `mru_transform/include/mru_transform/sensor.hpp:111` (cross-confirmed by Claude Adversarial and the earlier Issue Review; subscribe-once semantics match the original ROS 1 intent).
+- [x] (suggestion-applied) Test Phase 1 discovery headroom raised 6 s → 15 s for CI flake insurance — `mru_transform/test/test_subscribe_once.cpp:102` (folded into a8baec6).

mru_transform/CMakeLists.txt

@@ -302,6 +302,17 @@ if(BUILD_TESTING)
   target_include_directories(test_twist_rotation PRIVATE
     "${CMAKE_CURRENT_SOURCE_DIR}/include"
   )
+
+  # Regression test for issue #23: subscribeCheck() subscribes once and stops.
+  ament_add_gtest(test_subscribe_once test/test_subscribe_once.cpp)
+  target_link_libraries(test_subscribe_once
+    ${PROJECT_NAME}
+    rclcpp::rclcpp
+    ${sensor_msgs_TARGETS}
+  )
+  target_include_directories(test_subscribe_once PRIVATE
+    "${CMAKE_CURRENT_SOURCE_DIR}/include"
+  )
 endif()
 
 ament_package()

mru_transform/include/mru_transform/sensor.hpp

@@ -87,8 +87,10 @@ class SensorBase
 
       RCLCPP_WARN_THROTTLE(logger_, *clock_,
           30 * 1000, msg.str().c_str()); // Throttle interval in milliseconds
+      return; // topic not advertised yet; the periodic timer keeps polling
     }
-    else if (!subscribe(topic_types)){
+
+    if (!subscribe(topic_types)){
       std::stringstream msg;
       msg <<"Unsupported " << sensor_type << " topic type for: " << topic_ << ", types: ";
       for(const auto &topic_type: topic_types)
@@ -97,11 +99,17 @@ class SensorBase
         30 * 1000,  // Throttle interval in milliseconds
         msg.str()
       );
+      return; // unsupported type for now; the periodic timer keeps polling
     }
-    else
-      return; // subscribed, so bail out before setting a new timer
 
-    subscribe_check_timer_ = rclcpp::create_wall_timer(1000ms, [this]{this->subscribeCheck();}, nullptr, node_.get_node_base_interface().get(), node_.get_node_timers_interface().get());
+    // Subscribed: stop the periodic check so subscribe() is not re-run on every
+    // tick. rclcpp wall timers always repeat (there is no one-shot option like
+    // the ROS 1 timer this logic was originally written for); without this
+    // cancel the timer keeps firing and each tick recreates the subscription,
+    // churning it ~1 Hz and racing in-flight samples — which rmw_zenoh surfaces
+    // as "SubscriberCallback triggered over ..." ERRORs (issue #23).
+    if(subscribe_check_timer_)
+      subscribe_check_timer_->cancel();
   }
 
   void call_callbacks_(const ValueType& value)

mru_transform/src/orientation_sensor.cpp

@@ -14,6 +14,11 @@ OrientationSensor::OrientationSensor(NodeInterfaces node, std::string name, Call
 
 bool OrientationSensor::subscribe(const std::vector<std::string>& topic_types)
 {
+  // Already subscribed — don't tear down and recreate a live subscription.
+  // Defense-in-depth for issue #23; SensorBase::subscribeCheck() also stops its
+  // timer once subscribed, so this is normally never re-entered.
+  if(subs_.imu || subs_.quaternion_stamped || subs_.geopose_stamped)
+    return true;
   for(const auto &topic_type: topic_types)
   {
     if (topic_type == "sensor_msgs/msg/Imu")

mru_transform/src/position_sensor.cpp

@@ -15,6 +15,11 @@ PositionSensor::PositionSensor(NodeInterfaces node, std::string name, CallbackTy
 
 bool PositionSensor::subscribe(const std::vector<std::string> &topic_types)
 {
+  // Already subscribed — don't tear down and recreate a live subscription.
+  // Defense-in-depth for issue #23; SensorBase::subscribeCheck() also stops its
+  // timer once subscribed, so this is normally never re-entered.
+  if(subs_.navsat_fix || subs_.geo_point_stamped || subs_.geo_pose_stamped)
+    return true;
   for(const auto &topic_type: topic_types)
   {
     if(topic_type == "sensor_msgs/msg/NavSatFix")

mru_transform/src/velocity_sensor.cpp

@@ -15,6 +15,11 @@ VelocitySensor::VelocitySensor(NodeInterfaces node, std::string name, CallbackTy
 
 bool VelocitySensor::subscribe(const std::vector<std::string> &topic_types)
 {
+  // Already subscribed — don't tear down and recreate a live subscription.
+  // Defense-in-depth for issue #23; SensorBase::subscribeCheck() also stops its
+  // timer once subscribed, so this is normally never re-entered.
+  if(subs_.twist_with_covariance_stamped || subs_.twist_stamped)
+    return true;
   for(const auto &topic_type: topic_types)
   {
     if (topic_type == "geometry_msgs/msg/TwistWithCovarianceStamped")

mru_transform/test/test_subscribe_once.cpp

@@ -0,0 +1,122 @@
+// Regression test for issue #23: SensorBase::subscribeCheck() must stop polling
+// once a sensor has subscribed. rclcpp wall timers always repeat (there is no
+// one-shot option like the ROS 1 timer this logic was ported from), so before
+// the fix the check timer kept firing every period and re-ran subscribe(),
+// tearing down and recreating the live subscription ~1 Hz. Under rmw_zenoh that
+// raced in-flight samples and produced "SubscriberCallback triggered over ..."
+// ERROR spam.
+
+#include <chrono>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include <gtest/gtest.h>
+#include <rclcpp/rclcpp.hpp>
+#include <sensor_msgs/msg/imu.hpp>
+
+#include "mru_transform/orientation_sensor.hpp"
+
+using namespace std::chrono_literals;
+
+namespace
+{
+
+// Counts how many times subscribeCheck() invokes subscribe() and exposes the
+// check timer's state. Overriding the (private, virtual) subscribe() is legal
+// for a derived class; subscribeCheck() dispatches to it through the base. The
+// override creates a real subscription so subscribeCheck() takes its
+// "subscribed" branch, exercising the actual base-class control flow.
+class CountingOrientationSensor : public mru_transform::OrientationSensor
+{
+public:
+  using mru_transform::OrientationSensor::OrientationSensor;
+
+  int subscribe_calls() const { return subscribe_calls_; }
+
+  bool timer_active() const
+  {
+    return subscribe_check_timer_ && !subscribe_check_timer_->is_canceled();
+  }
+
+private:
+  bool subscribe(const std::vector<std::string> & topic_types) override
+  {
+    ++subscribe_calls_;
+    for (const auto & topic_type : topic_types) {
+      if (topic_type == "sensor_msgs/msg/Imu") {
+        subs_.imu = rclcpp::create_subscription<sensor_msgs::msg::Imu>(
+          node_, topic_, rclcpp::SensorDataQoS(),
+          [](sensor_msgs::msg::Imu::SharedPtr) {});
+        return true;
+      }
+    }
+    return false;
+  }
+
+  int subscribe_calls_{0};
+};
+
+void spin_for(
+  rclcpp::executors::SingleThreadedExecutor & exec,
+  std::chrono::milliseconds duration)
+{
+  const auto deadline = std::chrono::steady_clock::now() + duration;
+  while (std::chrono::steady_clock::now() < deadline && rclcpp::ok()) {
+    exec.spin_some();
+    std::this_thread::sleep_for(10ms);
+  }
+}
+
+}  // namespace
+
+class SubscribeOnceTest : public ::testing::Test
+{
+protected:
+  void SetUp() override { rclcpp::init(0, nullptr); }
+  void TearDown() override { rclcpp::shutdown(); }
+};
+
+TEST_F(SubscribeOnceTest, SubscribesOnceThenStops)
+{
+  auto node = std::make_shared<rclcpp::Node>("subscribe_once_test");
+
+  // SensorBase reads this in its constructor and only arms the check timer when
+  // the topic is non-empty, so it must be set before the sensor is created.
+  node->declare_parameter("sensors.default.topics.orientation", "test_imu");
+
+  // Advertise the topic with a supported type so subscribeCheck() can resolve a
+  // type and take its subscribe() path.
+  auto pub = node->create_publisher<sensor_msgs::msg::Imu>(
+    "test_imu", rclcpp::SensorDataQoS());
+
+  auto sensor = std::make_shared<CountingOrientationSensor>(
+    *node, "default", mru_transform::OrientationSensor::CallbackType());
+
+  rclcpp::executors::SingleThreadedExecutor exec;
+  exec.add_node(node);
+
+  // Phase 1: wait for the first successful subscribe (timer period is 1 s).
+  // Generous headroom so a loaded CI box doesn't fail discovery before the
+  // first timer tick; the assertion below distinguishes "never subscribed".
+  const auto deadline = std::chrono::steady_clock::now() + 15s;
+  while (sensor->subscribe_calls() < 1 &&
+    std::chrono::steady_clock::now() < deadline && rclcpp::ok())
+  {
+    exec.spin_some();
+    std::this_thread::sleep_for(10ms);
+  }
+
+  ASSERT_EQ(sensor->subscribe_calls(), 1) << "sensor never subscribed";
+  EXPECT_FALSE(sensor->timer_active())
+    << "check timer should be cancelled after subscribing";
+
+  // Phase 2: spin past several more timer periods. subscribe() must not run
+  // again — the pre-fix bug re-subscribed every period.
+  spin_for(exec, 3500ms);
+
+  EXPECT_EQ(sensor->subscribe_calls(), 1)
+    << "subscribe() ran more than once — the check timer was not stopped";
+  EXPECT_FALSE(sensor->timer_active());
+}