Connection-layer hardening

include/sendspin/client.h

@@ -24,6 +24,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <memory>
+#include <mutex>
 #include <optional>
 #include <string>
 #include <vector>
@@ -205,6 +206,10 @@ class SendspinClient {
     bool start_server();
 
     /// @brief Initiates a client connection to a Sendspin server at the given URL
+    ///
+    /// Must be called from the main loop thread: it tears down and replaces connection state
+    /// (time filter, dispatch, client state) directly rather than deferring to loop(), so calling
+    /// it concurrently with loop() would race those mutations.
     /// @param url WebSocket server URL (e.g., "ws://server.local:8927/sendspin")
     void connect_to(const std::string& url);
 
@@ -470,6 +475,9 @@ class SendspinClient {
     /// Internal-RAM scratch arena for parsing incoming JSON; null unless config_.json_arena_size >
     /// 0
     std::unique_ptr<SendspinArenaAllocator> json_arena_;
+    /// Serializes process_json_message() (and its use of json_arena_) across the network threads
+    /// of concurrently live connections (current + pending during a handoff).
+    std::mutex json_processing_mutex_;
     SendspinClientListener* listener_{nullptr};
 #ifdef SENDSPIN_ENABLE_METADATA
     std::unique_ptr<MetadataRole> metadata_;

src/client.cpp

@@ -58,6 +58,12 @@ struct TimeResponseEvent {
     int64_t offset;
     int64_t max_error;
     int64_t timestamp;
+    /// Instance id of the connection the response arrived on, compared against the current
+    /// connection at drain time so a measurement from a displaced or pending server cannot
+    /// contaminate the current connection's time filter. An id (not a pointer) so a since-freed
+    /// connection cannot ABA-match a later connection reusing its address; 0 never matches a live
+    /// connection (ids start at 1).
+    uint64_t source_id;
 };
 
 /// @brief Deferred event state for time responses and group updates on the main thread
@@ -186,7 +192,10 @@ void SendspinClient::loop() {
         TimeResponseEvent time_event{};
         while (this->event_state_->time_queue.receive(time_event, 0)) {
             auto* current = this->connection_manager_->current();
-            if (current != nullptr) {
+            // Apply only measurements from the connection that is still current; a response queued
+            // by a since-displaced (or pending) server carries that server's clock and would
+            // contaminate this connection's Kalman filter.
+            if (current != nullptr && current->get_instance_id() == time_event.source_id) {
                 this->time_burst_->on_time_response(current, time_event.offset,
                                                     time_event.max_error, time_event.timestamp);
             }
@@ -331,17 +340,21 @@ bool SendspinClient::is_connected() const {
 }
 
 bool SendspinClient::is_time_synced() const {
-    auto* conn = this->connection_manager_->current();
+    // current_shared(): called from role threads (sync task, drain threads), so the shared_ptr
+    // must keep the connection alive while it is dereferenced.
+    auto conn = this->connection_manager_->current_shared();
     return conn != nullptr && conn->is_time_synced();
 }
 
 int64_t SendspinClient::get_client_time(int64_t server_time) const {
-    auto* conn = this->connection_manager_->current();
+    // current_shared(): called from role threads; see is_time_synced().
+    auto conn = this->connection_manager_->current_shared();
     return conn != nullptr ? conn->get_client_time(server_time) : 0;
 }
 
 std::optional<ServerInformationObject> SendspinClient::get_server_information() const {
-    auto* conn = this->connection_manager_->current();
+    // current_shared(): public accessor, callable from any thread.
+    auto conn = this->connection_manager_->current_shared();
     if (conn == nullptr || !conn->is_handshake_complete()) {
         return std::nullopt;
     }
@@ -499,9 +512,14 @@ std::string SendspinClient::build_hello_message() {
 
 void SendspinClient::process_json_message(SendspinConnection* conn, const char* data, size_t len,
                                           int64_t timestamp) {
-    // Reuse the internal-RAM scratch arena if configured. Safe to reset here: this runs only on
-    // the network task, and the JsonDocument from the previous call was already destroyed when
-    // that call returned.
+    // Two connections can deliver JSON concurrently on their own network threads (current +
+    // pending during a handoff, or an outbound connect_to() transport alongside the inbound
+    // server). Serialize the shared arena and the parse itself; JSON control messages are
+    // infrequent, so contention is negligible.
+    std::lock_guard<std::mutex> lock(this->json_processing_mutex_);
+
+    // Reuse the internal-RAM scratch arena if configured. Safe to reset here: the JsonDocument
+    // from the previous call was already destroyed when that call returned.
     if (this->json_arena_) {
         this->json_arena_->reset();
     }
@@ -656,7 +674,8 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char*
             int64_t offset{0};
             int64_t max_error{0};
             if (process_server_time_message(root, timestamp, &offset, &max_error)) {
-                if (!this->event_state_->time_queue.send({offset, max_error, timestamp}, 0)) {
+                if (!this->event_state_->time_queue.send(
+                        {offset, max_error, timestamp, conn->get_instance_id()}, 0)) {
                     SS_LOGW(TAG, "Time response queue full; dropping measurement");
                 }
             }

src/connection.h

@@ -115,6 +115,15 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
         return -1;
     }
 
+    /// @brief Returns this connection's process-unique instance id
+    /// @return A monotonic id assigned at construction, never reused for the lifetime of the
+    ///         process. Used to identify a connection across a thread-safe queue without a raw
+    ///         pointer, which could ABA-collide with a later connection allocated at the same
+    ///         address. Ids start at 1, so 0 is a safe "no connection" sentinel.
+    uint64_t get_instance_id() const {
+        return this->instance_id;
+    }
+
     /// @brief Sends a text message to the server with a completion callback
     /// @param message The message string to send.
     /// @param cb Callback invoked with the send result. On asynchronous transports it is not
@@ -342,6 +351,13 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     /// @param receive_time Timestamp when the data was received (microseconds).
     void dispatch_completed_message(bool is_text, int64_t receive_time);
 
+    /// @brief Returns the next process-unique connection id (starts at 1, monotonic).
+    /// @note The function-local atomic gives thread-safe, ordering-independent uniqueness.
+    static uint64_t next_instance_id() {
+        static std::atomic<uint64_t> counter{1};
+        return counter.fetch_add(1, std::memory_order_relaxed);
+    }
+
     // Struct fields
 
     /// Message buffering (for websocket frame assembly).
@@ -361,6 +377,9 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     /// server worker thread updates it while the hub thread reads it for logging.
     std::atomic<int64_t> serialize_ema_us_{0};
 
+    /// Process-unique connection identity (see get_instance_id()). Assigned once at construction.
+    const uint64_t instance_id{next_instance_id()};
+
     // size_t fields
     size_t websocket_write_offset_{0};
 
@@ -384,8 +403,9 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     /// Set to false on the main thread before cleanup; checked on the network thread.
     std::atomic<bool> message_dispatch_enabled_{true};
 
-    /// Time message state.
-    bool pending_time_message_{false};
+    /// Time message state. Written by the main-loop time burst and cleared by the transport
+    /// thread's disconnect handler, hence atomic.
+    std::atomic<bool> pending_time_message_{false};
 
     /// Atomic for the same reason as client_hello_sent_: written on network threads, read from the
     /// main loop via is_handshake_complete().

src/connection_manager.cpp

@@ -31,6 +31,8 @@ static constexpr uint32_t FNV_OFFSET_BASIS = 2166136261UL;
 static constexpr uint32_t FNV_PRIME = 16777619UL;
 static constexpr int64_t HELLO_INITIAL_DELAY_MS = 100LL;
 static constexpr int64_t HELLO_INITIAL_DELAY_US = HELLO_INITIAL_DELAY_MS * US_PER_MS;
+static constexpr int64_t WS_SERVER_START_RETRY_MS = 5000LL;
+static constexpr int64_t WS_SERVER_START_RETRY_US = WS_SERVER_START_RETRY_MS * US_PER_MS;
 
 // ============================================================================
 // Constructor / Destructor
@@ -72,9 +74,27 @@ void ConnectionManager::connect_to(const std::string& url) {
         std::lock_guard<std::mutex> lock(this->conn_ptr_mutex_);
         if (this->current_connection_ != nullptr && this->current_connection_->is_connected()) {
             SS_LOGD(TAG, "Existing connection active, new connection will go through handoff");
+            // Release any existing pending connection (e.g. an inbound handoff candidate) before
+            // overwriting the slot, mirroring on_new_connection and complete_handoff; otherwise it
+            // is dropped with no goodbye and, on ESP, leaves its httpd session pinned.
+            if (this->pending_connection_ != nullptr) {
+                this->remove_hello_retry(this->pending_connection_.get());
+                this->disconnect_and_release(std::move(this->pending_connection_),
+                                             SendspinGoodbyeReason::ANOTHER_SERVER);
+            }
             this->pending_connection_ = std::move(client_conn);
             this->pending_connection_->start();
         } else {
+            // A present-but-disconnected current connection (its close event not yet processed) is
+            // being replaced. Tear its state down as on_connection_lost would, instead of
+            // overwriting the slot and orphaning dispatch/time-filter/client state and its retry.
+            if (this->current_connection_ != nullptr) {
+                this->current_connection_->disable_message_dispatch();
+                this->client_->time_burst_->reset();
+                this->client_->cleanup_connection_state();
+                this->remove_hello_retry(this->current_connection_.get());
+                this->current_connection_.reset();
+            }
             this->current_connection_ = std::move(client_conn);
             this->current_connection_->start();
         }
@@ -132,12 +152,16 @@ void ConnectionManager::init_server(SendspinClient* client) {
 }
 
 void ConnectionManager::loop() {
-    // Start WS server when network becomes ready
+    // Start WS server when network becomes ready. A persistent failure (e.g. the server port is
+    // already in use) is retried with backoff instead of on every tick, which would spam the log.
     if (this->ws_server_ != nullptr && !this->ws_server_->is_started()) {
-        if (this->client_->network_provider_ &&
+        const int64_t now_us = platform_time_us();
+        if (now_us >= this->ws_server_start_retry_time_us_ && this->client_->network_provider_ &&
             this->client_->network_provider_->is_network_ready()) {
-            this->ws_server_->start(this->client_, this->client_->config_.httpd_psram_stack,
-                                    this->client_->config_.httpd_priority);
+            if (!this->ws_server_->start(this->client_, this->client_->config_.httpd_psram_stack,
+                                         this->client_->config_.httpd_priority)) {
+                this->ws_server_start_retry_time_us_ = now_us + WS_SERVER_START_RETRY_US;
+            }
         }
     }
 

src/connection_manager.h

@@ -120,6 +120,16 @@ class ConnectionManager {
         return this->current_connection_.get();
     }
 
+    /// @brief Returns a shared_ptr to the current connection. Thread-safe.
+    /// Role threads (sync task, artwork/visualizer drains) must use this instead of current():
+    /// the shared_ptr keeps the connection alive for the duration of the caller's use even if the
+    /// main loop concurrently drops or replaces the current connection.
+    /// @return Shared pointer to the current connection, or nullptr if none.
+    std::shared_ptr<SendspinConnection> current_shared() const {
+        std::lock_guard<std::mutex> lock(this->conn_ptr_mutex_);
+        return this->current_connection_;
+    }
+
     // ========================================
     // Event queuing (thread-safe)
     // ========================================
@@ -208,6 +218,10 @@ class ConnectionManager {
     // 32-bit fields
     uint32_t last_played_server_hash_{0};
 
+    // 64-bit fields
+    /// Earliest time (us) to attempt another WS server start after a failure. Main-loop only.
+    int64_t ws_server_start_retry_time_us_{0};
+
     // 8-bit fields
     bool has_last_played_server_{false};
 };

src/esp/client_connection.cpp

@@ -258,6 +258,11 @@ void SendspinClientConnection::handle_data(const esp_websocket_event_data_t* dat
         uint8_t* dest = this->prepare_receive_buffer(prepare_len);
         if (dest == nullptr) {
             SS_LOGE(TAG, "Allocation failed, dropping connection");
+            // Stop processing frames that keep arriving on the still-open transport; the
+            // manager reacts to the disconnect callback by dropping the connection, whose
+            // destructor stops the transport (esp_websocket_client_stop cannot be called
+            // from the websocket task's own event handler).
+            this->disable_message_dispatch();
             this->handle_disconnected();
             return;
         }

src/esp/client_connection.h

@@ -20,6 +20,7 @@
 #include "connection.h"
 #include <esp_websocket_client.h>
 
+#include <atomic>
 #include <functional>
 #include <string>
 
@@ -162,8 +163,9 @@ class SendspinClientConnection : public SendspinConnection {
     /// @brief Whether to automatically reconnect after connection loss
     bool auto_reconnect_{true};
 
-    /// @brief Whether the websocket is currently connected
-    bool connected_{false};
+    /// @brief Whether the websocket is currently connected. Written by the transport task,
+    /// read cross-thread via is_connected(), hence atomic.
+    std::atomic<bool> connected_{false};
 };
 
 }  // namespace sendspin

src/esp/server_connection.cpp

@@ -110,7 +110,7 @@ void SendspinServerConnection::disconnect(SendspinGoodbyeReason reason,
 }
 
 bool SendspinServerConnection::is_connected() const {
-    return this->sockfd_ >= 0;
+    return this->sockfd_ >= 0 && !this->closed_.load(std::memory_order_acquire);
 }
 
 SsErr SendspinServerConnection::send_text_message(const std::string& message,
@@ -211,6 +211,10 @@ SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t
     // Allocate/grow directly into the websocket payload buffer (zero-copy)
     uint8_t* dest = this->prepare_receive_buffer(ws_pkt.len);
     if (dest == nullptr) {
+        // Returning an error makes httpd close the session, which tears the slot down via
+        // the close notification on the main loop
+        SS_LOGE(TAG, "Allocation failed, dropping connection");
+        this->disable_message_dispatch();
         return ESP_ERR_NO_MEM;
     }
 

src/esp/server_connection.h

@@ -20,6 +20,7 @@
 #include "connection.h"
 #include <esp_http_server.h>
 
+#include <atomic>
 #include <functional>
 
 namespace sendspin {
@@ -85,6 +86,16 @@ class SendspinServerConnection : public SendspinConnection {
     /// @return true if connected, false otherwise.
     bool is_connected() const override;
 
+    /// @brief Marks the connection closed after the httpd session ends
+    ///
+    /// Called from the ws server's close notification (httpd thread). Without this,
+    /// is_connected() stayed true until the manager dropped the connection on the main loop,
+    /// and a queued async send in that window could resolve the stale sockfd against a
+    /// recycled httpd session and write the frame to the wrong peer.
+    void mark_closed() {
+        this->closed_.store(true, std::memory_order_release);
+    }
+
     /// @brief Sends a text message to the server with a completion callback
     /// @param message The message string to send.
     /// @param on_complete Callback invoked after send completes.
@@ -150,6 +161,11 @@ class SendspinServerConnection : public SendspinConnection {
 
     /// @brief The socket file descriptor for this connection
     int sockfd_{-1};
+
+    // 8-bit fields
+
+    /// @brief Set once the httpd session has closed (see mark_closed())
+    std::atomic<bool> closed_{false};
 };
 
 }  // namespace sendspin

src/esp/ws_server.cpp

@@ -151,6 +151,14 @@ esp_err_t SendspinWsServer::open_callback(httpd_handle_t handle, int sockfd) {
 void SendspinWsServer::close_callback(httpd_handle_t handle, int sockfd) {
     SS_LOGD(TAG, "Client closed connection on socket %d", sockfd);
 
+    // Flip the connection to disconnected immediately: queued async sends check is_connected()
+    // and must not resolve this sockfd once httpd may recycle it for a new session.
+    auto* slot =
+        static_cast<std::shared_ptr<SendspinServerConnection>*>(httpd_sess_get_ctx(handle, sockfd));
+    if (slot != nullptr && *slot != nullptr) {
+        (*slot)->mark_closed();
+    }
+
     SendspinWsServer* server = (SendspinWsServer*)httpd_get_global_user_ctx(handle);
 
     // Notify ConnectionManager so it can drop its observer shared_ptr. The session slot (set in

src/host/client_connection.cpp

@@ -173,10 +173,17 @@ void SendspinClientConnection::setup_callbacks() {
                     uint8_t* dest = this->prepare_receive_buffer(data.size());
                     if (dest == nullptr) {
                         SS_LOGE(TAG, "Allocation failed, dropping connection");
+                        // Stop processing further frames and initiate a real transport close
+                        // (stop() would join IX's thread and deadlock here; close() is async
+                        // and safe). The later Close event repeats the disconnect callback,
+                        // which the manager tolerates (drop of an unmanaged connection is a
+                        // no-op).
+                        this->disable_message_dispatch();
                         this->connected_ = false;
                         if (this->on_disconnected_cb) {
                             this->on_disconnected_cb(this);
                         }
+                        this->ws_->close();
                         return;
                     }
                     std::copy(data.begin(), data.end(), dest);

src/host/client_connection.h

@@ -20,6 +20,7 @@
 #include "connection.h"
 #include <ixwebsocket/IXWebSocket.h>
 
+#include <atomic>
 #include <functional>
 #include <memory>
 #include <string>
@@ -128,8 +129,9 @@ class SendspinClientConnection : public SendspinConnection {
     /// @brief Whether to automatically reconnect after connection loss
     bool auto_reconnect_{true};
 
-    /// @brief Whether the websocket is currently connected
-    bool connected_{false};
+    /// @brief Whether the websocket is currently connected. Written by the IXWebSocket
+    /// callback thread, read cross-thread via is_connected(), hence atomic.
+    std::atomic<bool> connected_{false};
 };
 
 }  // namespace sendspin