fix(PrimaryClient): close stream before joining reconnect thread; interruptible producer backoff

Extends the RTDEClient teardown fix to the PrimaryClient pipeline, which is the
path used by UrDriver. ~PrimaryClient() could block indefinitely (up to the
unbounded reconnect timeout) when the robot dropped the primary connection at
teardown time.

Root cause: on connection loss TCPSocket::read() leaves the socket in
SocketState::Disconnected, so URProducer::tryGetImpl() enters its reconnect loop
(sleep backoff + stream_.connect() with unlimited retries). ~PrimaryClient() and
PrimaryClient::stop() joined the producer thread (pipeline_->stop()) without
first closing the stream, so the join blocked for the full reconnect duration.

Two related fixes:

1. ~PrimaryClient() and PrimaryClient::stop() now call stream_.close() BEFORE
   pipeline_->stop(). stream_.close() sets SocketState::Closed, waking a producer
   stuck in its reconnect path so the join returns promptly. Previously stop()
   closed the stream after the join (too late) and the destructor never closed it.

2. URProducer::tryGetImpl()'s reconnect backoff slept sleep_for(timeout_) (growing
   up to 120 s) with no cancellation point. It now sleeps in 100 ms slices and
   bails out as soon as running_ becomes false or the stream is closed, mirroring
   the TCPSocket::setup() interruptible-sleep fix.

test: add PrimaryClientReconnectTest.destructor_not_blocked_by_stuck_reconnect_thread
(test_primary_client_reconnect.cpp). The PrimaryClient counterpart of
RTDEClientTest.destructor_not_blocked_by_stuck_reconnect_thread, using the
in-process FakePrimaryServer so it runs without a robot (unlike the
INTEGRATION_TESTS-gated primary_client_test_headless). It starts a client against
the fake server, drops the server to drive the producer into its reconnect loop,
then asserts ~PrimaryClient() completes in < 2 s. Verified to hang without the fix
and pass in ~1.5 s with it.

Co-authored-by: Rune Søe-Knudsen <urrsk@users.noreply.github.com>

Author: urrsk

include/ur_client_library/comm/producer.h

@@ -85,7 +85,21 @@ class URProducer : public IProducer<T>
       }
 
       URCL_LOG_WARN("Failed to read from stream, reconnecting in %ld seconds...", timeout_.count());
-      std::this_thread::sleep_for(timeout_);
+      // Sleep in small slices so the producer can be stopped (running_ == false)
+      // or woken by a stream close (e.g. from a destructor calling stop()) within
+      // ~100 ms, instead of blocking for the full (exponentially growing) timeout.
+      // Without this, a thread joining the pipeline at teardown could block for up
+      // to 120 s while this thread sleeps here.
+      const auto sleep_slice = std::chrono::milliseconds(100);
+      const auto sleep_total = std::chrono::duration_cast<std::chrono::milliseconds>(timeout_);
+      for (auto slept = std::chrono::milliseconds(0); slept < sleep_total && running_ && !stream_.closed();
+           slept += sleep_slice)
+      {
+        std::this_thread::sleep_for(sleep_slice);
+      }
+
+      if (!running_ || stream_.closed())
+        return false;
 
       if (stream_.connect())
         continue;

src/primary/primary_client.cpp

@@ -67,6 +67,12 @@ PrimaryClient::PrimaryClient(const std::string& robot_ip, [[maybe_unused]] comm:
 PrimaryClient::~PrimaryClient()
 {
   URCL_LOG_INFO("Stopping primary client pipeline");
+  // Close the stream BEFORE stopping (joining) the pipeline. The pipeline's
+  // producer thread may be sleeping inside its reconnect backoff or blocked in
+  // TCPSocket::setup(); closing the stream sets SocketState::Closed, which wakes
+  // both paths so pipeline_->stop()'s join returns promptly instead of blocking
+  // until the (potentially unbounded) reconnect timeout expires.
+  stream_.close();
   pipeline_->stop();
 }
 
@@ -79,8 +85,10 @@ void PrimaryClient::start(const size_t max_num_tries, const std::chrono::millise
 
 void PrimaryClient::stop()
 {
-  pipeline_->stop();
+  // Close the stream before joining the pipeline so a producer thread stuck in
+  // its reconnect path is woken and the join returns promptly (see ~PrimaryClient).
   stream_.close();
+  pipeline_->stop();
 }
 
 void PrimaryClient::addPrimaryConsumer(std::shared_ptr<comm::IConsumer<PrimaryPackage>> primary_consumer)

tests/CMakeLists.txt

@@ -152,6 +152,14 @@ target_link_libraries(fake_primary_server_tests PRIVATE ur_client_library::urcl
 gtest_add_tests(TARGET      fake_primary_server_tests
 )
 
+# Robot-free regression test for ~PrimaryClient() not blocking on a stuck
+# reconnect thread. Uses the in-process FakePrimaryServer so it runs without a
+# robot (unlike the INTEGRATION_TESTS-gated primary_client_test_headless).
+add_executable(primary_client_reconnect_tests test_primary_client_reconnect.cpp fake_primary_server.cpp)
+target_link_libraries(primary_client_reconnect_tests PRIVATE ur_client_library::urcl GTest::gtest_main)
+gtest_add_tests(TARGET      primary_client_reconnect_tests
+)
+
 
 
 add_executable(rtde_data_package_tests test_rtde_data_package.cpp)

tests/test_primary_client_reconnect.cpp

@@ -0,0 +1,109 @@
+// -- BEGIN LICENSE BLOCK ----------------------------------------------
+// Copyright 2026 Universal Robots A/S
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+//
+//    * Redistributions of source code must retain the above copyright
+//      notice, this list of conditions and the following disclaimer.
+//
+//    * Redistributions in binary form must reproduce the above copyright
+//      notice, this list of conditions and the following disclaimer in the
+//      documentation and/or other materials provided with the distribution.
+//
+//    * Neither the name of the {copyright_holder} nor the names of its
+//      contributors may be used to endorse or promote products derived from
+//      this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+// POSSIBILITY OF SUCH DAMAGE.
+// -- END LICENSE BLOCK ------------------------------------------------
+
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <memory>
+#include <thread>
+
+#include <ur_client_library/primary/primary_client.h>
+
+#include "fake_primary_server.h"
+
+using namespace urcl;
+
+// Regression test for ~PrimaryClient() blocking indefinitely when the pipeline's
+// producer thread is stuck in its reconnect loop at teardown time.
+//
+// This is the PrimaryClient counterpart of
+// RTDEClientTest.destructor_not_blocked_by_stuck_reconnect_thread (test_rtde_client.cpp).
+//
+// Root cause: when the robot drops the primary connection, TCPSocket::read()
+// returns false and leaves the socket in SocketState::Disconnected. URProducer's
+// tryGetImpl() then enters its reconnect path: it sleeps an (exponentially
+// growing) backoff and calls stream_.connect(), which retries with no upper
+// bound (max_num_tries == 0), sleeping reconnection_time between attempts. If
+// ~PrimaryClient() simply called pipeline_->stop() (which joins the producer
+// thread) without first closing the stream, the join would block for the full
+// reconnect duration — effectively forever for an unreachable robot.
+//
+// Fix (two parts):
+//   1. ~PrimaryClient()/PrimaryClient::stop() close the stream BEFORE joining the
+//      pipeline. stream_.close() sets SocketState::Closed.
+//   2. URProducer's reconnect backoff sleeps in 100 ms slices and bails out as
+//      soon as the stream is closed (or the producer is stopped), and
+//      TCPSocket::setup()'s between-attempt sleep is likewise interruptible by
+//      SocketState::Closed.
+// Together these wake the producer within ~100 ms of the destructor closing the
+// stream, so the join — and therefore the destructor — returns promptly.
+//
+// Unlike test_primary_client.cpp's robot-dependent fixtures, this test uses the
+// in-process FakePrimaryServer, so it runs in the normal (non-INTEGRATION_TESTS)
+// build and needs no robot.
+TEST(PrimaryClientReconnectTest, destructor_not_blocked_by_stuck_reconnect_thread)
+{
+  comm::INotifier notifier;
+
+  auto server = std::make_unique<FakePrimaryServer>(primary_interface::UR_PRIMARY_PORT);
+  auto client = std::make_unique<primary_interface::PrimaryClient>("127.0.0.1", notifier);
+
+  // Unlimited reconnect attempts with a large reconnection time: if the fix is
+  // absent, the producer's reconnect path keeps the destructor blocked.
+  const std::chrono::milliseconds large_reconnect_timeout(5000);
+  ASSERT_NO_THROW(client->start(/*max_num_tries=*/0, large_reconnect_timeout));
+  ASSERT_TRUE(server->waitForClient()) << "PrimaryClient never connected to the fake server";
+
+  // Drop the server. The producer's read() fails, the socket transitions to
+  // SocketState::Disconnected, and the producer enters its reconnect loop.
+  server.reset();
+
+  // Give the producer time to detect the drop and reach its reconnect sleep
+  // (initial backoff is 1 s, after which it sleeps inside TCPSocket::setup()).
+  std::this_thread::sleep_for(std::chrono::milliseconds(1500));
+
+  // The destructor must return quickly: it closes the stream
+  // (SocketState::Closed), waking the producer, so the pipeline join completes in
+  // well under 2 s. Without the fix this blocks for at least the reconnect
+  // timeout (and indefinitely with unlimited retries against a dead port).
+  const auto t0 = std::chrono::steady_clock::now();
+  client.reset();
+  const auto elapsed = std::chrono::steady_clock::now() - t0;
+
+  EXPECT_LT(elapsed, std::chrono::seconds(2))
+      << "~PrimaryClient() blocked for " << std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count()
+      << " ms — the producer reconnect thread was not woken by the stream close";
+}
+
+int main(int argc, char* argv[])
+{
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}