Axiomatic: reconnect on broken socket indefinitely

axiomatic_adapter/src/axiomatic_adapter.cpp

@@ -16,6 +16,7 @@
 
 #include <linux/can.h>
 #include <linux/can/raw.h>
+#include <sys/socket.h>
 
 #include <algorithm>
 #include <atomic>
@@ -167,6 +168,15 @@ class AxiomaticAdapter::AxiomaticAdapterImpl
 
     tcp_receive_thread_ = std::thread([this]() {
       while (!stop_thread_requested_) {
+        // a failed send() flags the link as lost; recover it here (on this
+        // thread, so it never races the read below). reconnect() returns false
+        // only on shutdown.
+        if (connection_lost_) {
+          if (!reconnect()) {
+            break;
+          }
+        }
+
         polymath::socketcan::CanFrame frame = polymath::socketcan::CanFrame();
         std::optional<AxiomaticAdapter::socket_error_string_t> error = receive(frame);
 
@@ -198,6 +208,41 @@ class AxiomaticAdapter::AxiomaticAdapterImpl
     return false;
   }
 
+  // Tear down a broken socket and retry openSocket() every 100 ms until it
+  // reconnects or shutdown is requested. Runs ONLY on the reception thread, so
+  // it never races receive(). Returns false if a stop was requested first.
+  bool reconnect()
+  {
+    std::cerr << "[Axiomatic] Connection lost — reconnecting to " << ip_address_ << ":" << port_ << "..." << std::endl;
+
+    // force any in-flight send() to unblock and error out so it releases
+    // write_mutex_; without this a send() blocked on a wedged socket would
+    // deadlock the teardown below
+    ::shutdown(tcp_socket_.native_handle(), SHUT_RDWR);
+
+    while (!stop_thread_requested_) {
+      {
+        // serialize the teardown against a concurrent send() on the other thread
+        std::lock_guard<std::mutex> guard(write_mutex_);
+        std::cerr << "[Axiomatic DBG] reconnect: closing socket..." << std::endl;
+        closeSocket();
+        std::cerr << "[Axiomatic DBG] reconnect: closed" << std::endl;
+      }
+
+      std::cerr << "[Axiomatic DBG] reconnect: attempting openSocket()..." << std::endl;
+      if (openSocket()) {
+        connection_lost_ = false;
+        std::cerr << "[Axiomatic] Reconnected to " << ip_address_ << ":" << port_ << std::endl;
+        return true;
+      }
+
+      std::cerr << "[Axiomatic DBG] reconnect: openSocket() failed, retrying in " << RECONNECT_RETRY_INTERVAL_MS.count()
+                << " ms" << std::endl;
+      std::this_thread::sleep_for(RECONNECT_RETRY_INTERVAL_MS);
+    }
+    return false;
+  }
+
   std::optional<AxiomaticAdapter::socket_error_string_t> receive(polymath::socketcan::CanFrame & can_frame)
   {
     // A previous TCP read may have decoded several CAN frames out of a single
@@ -411,9 +456,19 @@ class AxiomaticAdapter::AxiomaticAdapterImpl
     // insert the can frame data
     full_message.insert(full_message.end(), frame_data.begin(), frame_data.end());
 
+    // don't touch the socket while the reception thread is reconnecting it
+    if (connection_lost_) {
+      return std::optional<AxiomaticAdapter::socket_error_string_t>("TCP Send Failed: connection lost (reconnecting)");
+    }
+
     try {
+      // lock out the reconnect teardown for the duration of the write
+      std::lock_guard<std::mutex> guard(write_mutex_);
       boost::asio::write(tcp_socket_, boost::asio::buffer(full_message.data(), full_message.size()));
     } catch (const std::exception & e) {
+      // flag the link lost; the reception thread owns the actual reconnect
+      connection_lost_ = true;
+      std::cerr << "[Axiomatic DBG] send: write threw: " << e.what() << std::endl;
       return std::optional<AxiomaticAdapter::socket_error_string_t>(std::string("TCP Send Failed: ") + e.what());
     }
     return std::nullopt;
@@ -433,6 +488,9 @@ class AxiomaticAdapter::AxiomaticAdapterImpl
   static constexpr std::array<uint8_t, 7> AXIOMATIC_CAN_MESSAGE_HEADER = {'A', 'X', 'I', 'O', 0xBA, 0x36, 0x01};
   static constexpr std::chrono::milliseconds TCP_IP_CONNECTION_TIMEOUT_MS{3000};
 
+  // after a send() fails, retry openSocket() this often until reconnected
+  static constexpr std::chrono::milliseconds RECONNECT_RETRY_INTERVAL_MS{100};
+
   // receive buffer size for each async_receive call. Larger than the
   // protocol's per-message cap (256 bytes) by a wide margin
   static constexpr size_t RECEIVE_BUFFER_SIZE = 65536;
@@ -517,6 +575,13 @@ class AxiomaticAdapter::AxiomaticAdapterImpl
   boost::asio::ip::tcp::socket tcp_socket_;
   TCPSocketState socket_state_{TCPSocketState::CLOSED};
 
+  // set by send() when a write fails; the reception thread reads it, reconnects,
+  // and clears it. Atomic because send() runs on a different thread.
+  std::atomic<bool> connection_lost_{false};
+
+  // guards send()'s write against the reception thread's reconnect teardown
+  std::mutex write_mutex_;
+
   std::thread tcp_receive_thread_;
   std::atomic<bool> thread_running_;
   std::atomic<bool> stop_thread_requested_;