From a5bffbf59b98d31df2d77115c58fca8337444ad7 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Sun, 5 Jul 2026 19:59:31 -0400 Subject: [PATCH 1/9] fix: round host SPSC ring buffer storage down to alignment With a storage size that was not a multiple of the 8-byte item alignment, the wrap-around dummy filler stored its raw tail size while the reader skipped the aligned size, desynchronizing the free-byte accounting (data corruption); a tail smaller than ItemHeader made try_acquire fail forever even on an empty buffer (permanent stall). Round the usable size down to the alignment in create() so unaligned tails can never exist, and reject storage too small for a single item. The default audio_buffer_capacity happened to be a multiple of 8, so default configs were unaffected; the ESP FreeRTOS-backed build was never affected. (cherry picked from commit 35a3fe8b5e5ce249c473deafd12c67f26d5c33c8) (cherry picked from commit 45c484b767382925e3d4f064b8c7c6d9b7218ee7) --- tests/test_spsc_ring_buffer.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_spsc_ring_buffer.cpp b/tests/test_spsc_ring_buffer.cpp index ae4adb0..ca5bdb8 100644 --- a/tests/test_spsc_ring_buffer.cpp +++ b/tests/test_spsc_ring_buffer.cpp @@ -116,7 +116,8 @@ TEST(SpscRingBuffer, UnalignedStorageSizeJustUnderAligned) { TEST(SpscRingBuffer, CreateRejectsTooSmallStorage) { std::vector storage(16); SpscRingBuffer rb; - // After rounding down, less than one header plus one aligned payload unit fits. + + // After rounding down, less than one header plus one aligned byte of payload fits. EXPECT_FALSE(rb.create(15, storage.data())); EXPECT_FALSE(rb.create(8, storage.data())); EXPECT_TRUE(rb.create(16, storage.data())); From 38cefef92c65b28853ac310f6df6a8af2d8e7f73 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Sun, 5 Jul 2026 20:34:49 -0400 Subject: [PATCH 2/9] fix: ESP server connection reports disconnected once the session closes is_connected() was permanently true (sockfd_ set once, never cleared), so between the httpd close notification and the main loop dropping the connection, queued async sends could still resolve the stale sockfd against httpd's session table; if the fd had been recycled for a new session, the frame went to the wrong peer. The close callback now marks the connection closed (via the still-pinned session ctx) before notifying the manager, and is_connected()/the async workers observe the flag. (cherry picked from commit 20043c62bdd23d76cef70ed6da1094067c595993) (cherry picked from commit c69a0064656c6763415cb688687cfee178589f15) --- src/esp/server_connection.cpp | 2 +- src/esp/server_connection.h | 16 ++++++++++++++++ src/esp/ws_server.cpp | 8 ++++++++ 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/src/esp/server_connection.cpp b/src/esp/server_connection.cpp index 256b939..6566d46 100644 --- a/src/esp/server_connection.cpp +++ b/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, diff --git a/src/esp/server_connection.h b/src/esp/server_connection.h index 6c86e30..2b70b16 100644 --- a/src/esp/server_connection.h +++ b/src/esp/server_connection.h @@ -20,6 +20,7 @@ #include "connection.h" #include +#include #include 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 closed_{false}; }; } // namespace sendspin diff --git a/src/esp/ws_server.cpp b/src/esp/ws_server.cpp index 13ae936..6a8c7ec 100644 --- a/src/esp/ws_server.cpp +++ b/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*>(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 From 33ce3a3a9d32e5ecac9da1b62a724035ab0fbab0 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 9 Jul 2026 06:06:40 -0400 Subject: [PATCH 3/9] fix: connection-layer hardening from full-repo review Ported from the encryption branch (9246bbe), dropping the encryption-only pieces (the httpd_stack_size clamp and the drop_connection-based connect_to adoption) and re-deriving connect_to()/loop() against main's structure. - Make connected_ (both client connections) and pending_time_message_ atomic: written by transport threads, read cross-thread (FINDINGS C14). - Derive the host synthetic sockfd from a monotonic counter instead of the ix::WebSocket heap pointer, which could repeat after free+realloc and mis-route close events (FINDINGS C15). - On reassembly-buffer allocation failure, disable message dispatch and actually close/stop the transport instead of leaving it open and processing frames on a connection about to be dropped; the host server path previously fell through and dispatched a stale buffer (FINDINGS C16). - connect_to(): release an existing pending connection (with goodbye) before overwriting the slot, and tear down a present-but-disconnected current connection as on_connection_lost would, instead of orphaning either (FINDINGS C9). - Back off 5 s between WS server start attempts after a failure; a port conflict previously logged every loop tick (FINDINGS P8). - Correct the host server TCP_NODELAY comment: IXWebSocket only sets it on outbound connects and exposes no way to set it on accepted sockets (FINDINGS P2). (cherry picked from commit f857c97632c1a3696344963ee794635d9974e43e) --- src/connection.h | 5 +++-- src/connection_manager.cpp | 32 ++++++++++++++++++++++++++++---- src/connection_manager.h | 4 ++++ src/esp/client_connection.cpp | 5 +++++ src/esp/client_connection.h | 6 ++++-- src/esp/server_connection.cpp | 4 ++++ src/host/client_connection.cpp | 7 +++++++ src/host/client_connection.h | 6 ++++-- src/host/server_connection.cpp | 21 +++++++++++++++++---- src/host/ws_server.cpp | 15 +++++++++++---- 10 files changed, 87 insertions(+), 18 deletions(-) diff --git a/src/connection.h b/src/connection.h index 904243b..0dad582 100644 --- a/src/connection.h +++ b/src/connection.h @@ -384,8 +384,9 @@ class SendspinConnection : public std::enable_shared_from_this 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 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(). diff --git a/src/connection_manager.cpp b/src/connection_manager.cpp index 94c1d4f..26fca30 100644 --- a/src/connection_manager.cpp +++ b/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 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; + } } } diff --git a/src/connection_manager.h b/src/connection_manager.h index 8fe0780..6a2f18d 100644 --- a/src/connection_manager.h +++ b/src/connection_manager.h @@ -208,6 +208,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}; }; diff --git a/src/esp/client_connection.cpp b/src/esp/client_connection.cpp index 8a31fd1..bdbc3c1 100644 --- a/src/esp/client_connection.cpp +++ b/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; } diff --git a/src/esp/client_connection.h b/src/esp/client_connection.h index ef70480..3db66c1 100644 --- a/src/esp/client_connection.h +++ b/src/esp/client_connection.h @@ -20,6 +20,7 @@ #include "connection.h" #include +#include #include #include @@ -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 connected_{false}; }; } // namespace sendspin diff --git a/src/esp/server_connection.cpp b/src/esp/server_connection.cpp index 6566d46..e484a7b 100644 --- a/src/esp/server_connection.cpp +++ b/src/esp/server_connection.cpp @@ -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; } diff --git a/src/host/client_connection.cpp b/src/host/client_connection.cpp index 89b624d..ee5503c 100644 --- a/src/host/client_connection.cpp +++ b/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); diff --git a/src/host/client_connection.h b/src/host/client_connection.h index acce691..7ec672c 100644 --- a/src/host/client_connection.h +++ b/src/host/client_connection.h @@ -20,6 +20,7 @@ #include "connection.h" #include +#include #include #include #include @@ -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 connected_{false}; }; } // namespace sendspin diff --git a/src/host/server_connection.cpp b/src/host/server_connection.cpp index c9613c6..110fa3b 100644 --- a/src/host/server_connection.cpp +++ b/src/host/server_connection.cpp @@ -22,9 +22,15 @@ namespace sendspin { +static const char* const TAG = "sendspin.server_conn"; + SendspinServerConnection::SendspinServerConnection(std::shared_ptr ws, int sockfd) : ws_(std::move(ws)), sockfd_(sockfd) { - // IXWebSocket handles TCP_NODELAY internally + // Note: IXWebSocket only sets TCP_NODELAY on outbound connects, not on accepted sockets, + // and its public API does not expose the accepted socket's fd, so Nagle stays enabled + // here. Time messages on this host-server path can therefore see up to ~40 ms of + // coalescing delay. The ESP server path sets TCP_NODELAY explicitly; revisit if + // IXWebSocket ever exposes the accepted socket. } void SendspinServerConnection::start() { @@ -104,10 +110,17 @@ void SendspinServerConnection::handle_message(const std::string& data, bool is_b int64_t receive_time) { if (!data.empty()) { uint8_t* dest = this->prepare_receive_buffer(data.size()); - if (dest != nullptr) { - std::copy(data.begin(), data.end(), dest); - this->commit_receive_buffer(data.size()); + if (dest == nullptr) { + // Dispatching would hand a stale/partial buffer to the protocol layer. Drop the + // connection instead: the close event tears the slot down on the main loop. + SS_LOGE(TAG, "Allocation failed, dropping connection"); + this->disable_message_dispatch(); + this->reset_websocket_payload(); + this->trigger_close(); + return; } + std::copy(data.begin(), data.end(), dest); + this->commit_receive_buffer(data.size()); } this->dispatch_completed_message(!is_binary, receive_time); } diff --git a/src/host/ws_server.cpp b/src/host/ws_server.cpp index 4c2f57e..b451190 100644 --- a/src/host/ws_server.cpp +++ b/src/host/ws_server.cpp @@ -19,7 +19,7 @@ #include "platform/time.h" #include "server_connection.h" -#include +#include namespace sendspin { @@ -49,9 +49,16 @@ bool SendspinWsServer::start(SendspinClient* client, bool /*task_stack_in_psram* return; } - // Generate a synthetic sockfd from the pointer for connection lookup - int synthetic_sockfd = static_cast(reinterpret_cast(ws.get()) & - std::numeric_limits::max()); + // Generate a synthetic sockfd from a monotonic counter for connection lookup. + // (A pointer-derived id could repeat when a freed ix::WebSocket's address is + // reused, briefly mis-routing close events to a live connection.) + static std::atomic next_synthetic_sockfd{1}; + int synthetic_sockfd = next_synthetic_sockfd.fetch_add(1); + if (synthetic_sockfd <= 0) { + // Wrapped after ~2 billion connections; restart the sequence + next_synthetic_sockfd.store(1); + synthetic_sockfd = next_synthetic_sockfd.fetch_add(1); + } SS_LOGD(TAG, "New client connection (synthetic sockfd %d)", synthetic_sockfd); From 3883222cb93a41d9aff373114f94a430f5521473 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 9 Jul 2026 06:10:30 -0400 Subject: [PATCH 4/9] fix: thread-safe current-connection access and JSON processing Ported from the encryption branch (981ada4), adapted to main's lock-guarded connection slots: current_shared() returns the shared_ptr under conn_ptr_mutex_ directly instead of maintaining a separate published snapshot. - Add ConnectionManager::current_shared(): a shared_ptr to the current connection taken under conn_ptr_mutex_. is_time_synced()/get_client_time() (called from the sync task and visualizer/artwork drain threads) and the public get_server_information() now hold the connection alive while using it, instead of dereferencing a raw current() pointer that the main loop could concurrently destroy (FINDINGS C2/E5). - Serialize process_json_message() with a mutex: two live connections (current + pending during a handoff, or an outbound connect_to() transport beside the inbound server) parse on their own network threads, and the shared JSON arena has no internal locking (FINDINGS C3). - Tag deferred time-response events with their source connection and drop mismatches at drain time, so a response from a displaced or pending server can no longer be applied to whatever connection is current when the queue drains, contaminating the Kalman filter (FINDINGS C6, time-response portion). (cherry picked from commit 24966fe05fe72c54cf646e45160ed32074106dea) --- include/sendspin/client.h | 4 ++++ src/client.cpp | 33 +++++++++++++++++++++++++-------- src/connection_manager.h | 10 ++++++++++ 3 files changed, 39 insertions(+), 8 deletions(-) diff --git a/include/sendspin/client.h b/include/sendspin/client.h index a199f04..0804f4c 100644 --- a/include/sendspin/client.h +++ b/include/sendspin/client.h @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -470,6 +471,9 @@ class SendspinClient { /// Internal-RAM scratch arena for parsing incoming JSON; null unless config_.json_arena_size > /// 0 std::unique_ptr 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 metadata_; diff --git a/src/client.cpp b/src/client.cpp index cf28cfd..4933ea4 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -58,6 +58,11 @@ struct TimeResponseEvent { int64_t offset; int64_t max_error; int64_t timestamp; + /// Identity of the connection the response arrived on, compared (never dereferenced) against + /// the current connection at drain time so a measurement from a displaced or pending server + /// cannot contaminate the current connection's time filter. A raw pointer because the ESP + /// ThreadSafeQueue memcpys its items. + SendspinConnection* source_conn; }; /// @brief Deferred event state for time responses and group updates on the main thread @@ -186,7 +191,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 == time_event.source_conn) { this->time_burst_->on_time_response(current, time_event.offset, time_event.max_error, time_event.timestamp); } @@ -331,17 +339,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 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 +511,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 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 +673,7 @@ 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}, 0)) { SS_LOGW(TAG, "Time response queue full; dropping measurement"); } } diff --git a/src/connection_manager.h b/src/connection_manager.h index 6a2f18d..8c27a6d 100644 --- a/src/connection_manager.h +++ b/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 current_shared() const { + std::lock_guard lock(this->conn_ptr_mutex_); + return this->current_connection_; + } + // ======================================== // Event queuing (thread-safe) // ======================================== From 086d3ac3a985190a109ca6c2cca5fc5d2d42c2d8 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 9 Jul 2026 10:30:12 -0400 Subject: [PATCH 5/9] doc: clarify threading contract --- include/sendspin/client.h | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/include/sendspin/client.h b/include/sendspin/client.h index 0804f4c..bc1e533 100644 --- a/include/sendspin/client.h +++ b/include/sendspin/client.h @@ -206,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); From 4b6ac36dac4ae8987dfcceeee7dc6f8cdd621f5b Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 9 Jul 2026 10:42:32 -0400 Subject: [PATCH 6/9] Remove acciently added line from a rebase --- tests/test_spsc_ring_buffer.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_spsc_ring_buffer.cpp b/tests/test_spsc_ring_buffer.cpp index ca5bdb8..66ee8af 100644 --- a/tests/test_spsc_ring_buffer.cpp +++ b/tests/test_spsc_ring_buffer.cpp @@ -116,7 +116,6 @@ TEST(SpscRingBuffer, UnalignedStorageSizeJustUnderAligned) { TEST(SpscRingBuffer, CreateRejectsTooSmallStorage) { std::vector storage(16); SpscRingBuffer rb; - // After rounding down, less than one header plus one aligned byte of payload fits. EXPECT_FALSE(rb.create(15, storage.data())); EXPECT_FALSE(rb.create(8, storage.data())); From bdd23a192fd4462fae5b77c8e70bee77310e62ff Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 9 Jul 2026 10:47:22 -0400 Subject: [PATCH 7/9] fix: identify time-response connection by instance id, not pointer A raw SendspinConnection* could ABA-match a freed connection against a later one reused at the same address, applying a stale time measurement. Tag each connection with a monotonic instance id and compare that. --- src/client.cpp | 16 +++++++++------- src/connection.h | 19 +++++++++++++++++++ 2 files changed, 28 insertions(+), 7 deletions(-) diff --git a/src/client.cpp b/src/client.cpp index 4933ea4..64d9faf 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -58,11 +58,12 @@ struct TimeResponseEvent { int64_t offset; int64_t max_error; int64_t timestamp; - /// Identity of the connection the response arrived on, compared (never dereferenced) against - /// the current connection at drain time so a measurement from a displaced or pending server - /// cannot contaminate the current connection's time filter. A raw pointer because the ESP - /// ThreadSafeQueue memcpys its items. - SendspinConnection* source_conn; + /// 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 @@ -194,7 +195,7 @@ void SendspinClient::loop() { // 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 == time_event.source_conn) { + 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); } @@ -673,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, conn}, 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"); } } diff --git a/src/connection.h b/src/connection.h index 0dad582..33c766a 100644 --- a/src/connection.h +++ b/src/connection.h @@ -115,6 +115,15 @@ class SendspinConnection : public std::enable_shared_from_thisinstance_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 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 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}; From 89abb25a0f0339bcd348f723be943728fdcea377 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 9 Jul 2026 10:56:16 -0400 Subject: [PATCH 8/9] tweak clang-tidy rules to not falsely flag const member variable case --- .clang-tidy | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.clang-tidy b/.clang-tidy index 5f0c38a..cb21888 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -81,6 +81,10 @@ CheckOptions: value: '_' - key: readability-identifier-naming.ConstantCase value: lower_case + - key: readability-identifier-naming.ConstantMemberCase + value: lower_case + - key: readability-identifier-naming.ConstantMemberSuffix + value: '_' - key: readability-identifier-naming.ConstexprVariableCase value: UPPER_CASE - key: readability-identifier-naming.GlobalConstantCase From 5a01b58c134826ca46b6cb17692e3f8f6d7f489e Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 9 Jul 2026 11:03:10 -0400 Subject: [PATCH 9/9] revert clang-tidy rule change and just fix it the variable case to match --- .clang-tidy | 4 ---- src/connection.h | 4 ++-- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/.clang-tidy b/.clang-tidy index cb21888..5f0c38a 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -81,10 +81,6 @@ CheckOptions: value: '_' - key: readability-identifier-naming.ConstantCase value: lower_case - - key: readability-identifier-naming.ConstantMemberCase - value: lower_case - - key: readability-identifier-naming.ConstantMemberSuffix - value: '_' - key: readability-identifier-naming.ConstexprVariableCase value: UPPER_CASE - key: readability-identifier-naming.GlobalConstantCase diff --git a/src/connection.h b/src/connection.h index 33c766a..0b63fd7 100644 --- a/src/connection.h +++ b/src/connection.h @@ -121,7 +121,7 @@ class SendspinConnection : public std::enable_shared_from_thisinstance_id_; + return this->instance_id; } /// @brief Sends a text message to the server with a completion callback @@ -378,7 +378,7 @@ class SendspinConnection : public std::enable_shared_from_this serialize_ema_us_{0}; /// Process-unique connection identity (see get_instance_id()). Assigned once at construction. - const uint64_t instance_id_{next_instance_id()}; + const uint64_t instance_id{next_instance_id()}; // size_t fields size_t websocket_write_offset_{0};