Add an outbound binary send primitive to all transports

src/binary_send_registry.h

@@ -0,0 +1,90 @@
+// Copyright 2026 Sendspin Contributors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// @file binary_send_registry.h
+/// @brief Owner-scoped keep-alive registry for work queued into a queue with no cancellation
+/// hook (ESP httpd_queue_work); platform-free so its transitions are unit-tested on host
+
+#pragma once
+
+#include <algorithm>
+#include <memory>
+#include <mutex>
+#include <vector>
+
+namespace sendspin {
+
+/// @brief Keeps queued work blocks alive until their worker claims them or their owner's queue
+/// is reclaimed
+///
+/// A Block must carry a `std::shared_ptr<Block> self` member. engage() parks the block's
+/// self-reference (keeping it alive independently of everything else) and records it under an
+/// opaque owner; claim() atomically takes the self-reference back for the worker that is about
+/// to run; reclaim() breaks the self-references of ONE owner's still-engaged blocks. Scoping
+/// discipline is the safety contract: reclaim an owner only once its queue can no longer run
+/// workers (after httpd_stop for that handle), so a claim() against a reclaimed block never
+/// happens with a dangling pointer, and other owners' queued work is never touched.
+template <typename Block>
+class BinarySendRegistry {
+public:
+    /// @brief Parks the block's self-reference and records it under the owner
+    void engage(const std::shared_ptr<Block>& block, const void* owner) {
+        std::lock_guard<std::mutex> lock(this->mutex_);
+        block->self = block;
+        this->entries_.push_back(Entry{owner, block});
+    }
+
+    /// @brief Takes the self-reference back for the worker about to run
+    /// @return The keep-alive reference, or empty if the block is no longer engaged (already
+    ///         claimed, or its owner was reclaimed).
+    std::shared_ptr<Block> claim(Block* raw) {
+        std::lock_guard<std::mutex> lock(this->mutex_);
+        auto it = std::find_if(this->entries_.begin(), this->entries_.end(),
+                               [&](const Entry& e) { return e.block.get() == raw; });
+        if (it == this->entries_.end()) {
+            return {};
+        }
+        std::shared_ptr<Block> keep = std::move(it->block->self);
+        this->entries_.erase(it);
+        return keep;
+    }
+
+    /// @brief Breaks the self-references of every block engaged under the owner
+    /// @return Number of blocks reclaimed.
+    size_t reclaim(const void* owner) {
+        std::lock_guard<std::mutex> lock(this->mutex_);
+        size_t count = 0;
+        for (auto it = this->entries_.begin(); it != this->entries_.end();) {
+            if (it->owner == owner) {
+                it->block->self.reset();
+                it = this->entries_.erase(it);
+                ++count;
+            } else {
+                ++it;
+            }
+        }
+        return count;
+    }
+
+private:
+    struct Entry {
+        const void* owner;
+        std::shared_ptr<Block> block;
+    };
+
+    std::mutex mutex_;
+    std::vector<Entry> entries_;
+};
+
+}  // namespace sendspin

src/connection.h

@@ -152,6 +152,30 @@ class SendspinConnection : public std::enable_shared_from_this<SendspinConnectio
     virtual SsErr send_text_message(const std::string& message, SendCompleteCallback cb,
                                     bool allow_before_hello = false) = 0;
 
+    /// @brief Sends a binary message to the server with a completion callback
+    ///
+    /// Callable from role task threads via ConnectionManager::current_shared(). The payload
+    /// must stay valid until @p cb fires or the call returns an error (queuing transports copy
+    /// it first). Binary frames are gated behind the client/hello like the default text path:
+    /// asynchronous transports enforce the gate themselves; synchronous transports send inline
+    /// and rely on caller ordering, exactly as their text sends do.
+    ///
+    /// Unlike the text path's best-effort callback, @p cb fires exactly once for EVERY call --
+    /// on success, every failure path, and connection teardown -- because single-in-flight
+    /// transports release their send slot from the completion path. Its execution context is
+    /// transport-dependent: inline on the calling thread for synchronous transports and for
+    /// immediate failures, from the httpd worker for a queued ESP-server send, and from the
+    /// destructor's thread for work that can never run -- so the callback must not perform
+    /// thread-affine work or call back into the connection.
+    ///
+    /// @param data Pointer to the message bytes (type byte first).
+    /// @param len Length of the message in bytes.
+    /// @param cb Callback invoked with the send result.
+    /// @return SsErr::OK if sent/queued successfully; SsErr::NOT_FINISHED if a previous binary
+    ///         send is still in flight on a single-in-flight transport (the caller treats this
+    ///         as "drop this chunk" and owns logging that drop); other codes on failure.
+    virtual SsErr send_binary_message(const uint8_t* data, size_t len, SendCompleteCallback cb) = 0;
+
     /// @brief Sends a client/time synchronization message
     ///
     /// The transport implementation captures `client_transmitted` as close to the actual wire

src/esp/client_connection.cpp

@@ -157,6 +157,34 @@ SsErr SendspinClientConnection::send_text_message(const std::string& message,
     return SsErr::OK;
 }
 
+SsErr SendspinClientConnection::send_binary_message(const uint8_t* data, size_t len,
+                                                    SendCompleteCallback cb) {
+    if (!this->is_connected()) {
+        if (cb) {
+            cb(false);
+        }
+        return SsErr::INVALID_STATE;
+    }
+
+    // esp_websocket_client_send_bin is synchronous in the current task, like the text path
+    int sent = esp_websocket_client_send_bin(this->client_, reinterpret_cast<const char*>(data),
+                                             static_cast<int>(len),
+                                             pdMS_TO_TICKS(WEBSOCKET_SEND_TIMEOUT_MS));
+
+    bool success = (sent >= 0);
+
+    if (cb) {
+        cb(success);
+    }
+
+    if (!success) {
+        SS_LOGE(TAG, "Failed to send binary message (timeout or error): %d", sent);
+        return SsErr::FAIL;
+    }
+
+    return SsErr::OK;
+}
+
 bool SendspinClientConnection::send_time_message() {
     if (!this->is_connected()) {
         return false;

src/esp/client_connection.h

@@ -90,6 +90,13 @@ class SendspinClientConnection : public SendspinConnection {
     SsErr send_text_message(const std::string& message, SendCompleteCallback cb,
                             bool allow_before_hello) override;
 
+    /// @brief Sends a binary message to the server, synchronously like the text path
+    /// @param data Pointer to the message bytes.
+    /// @param len Length of the message in bytes.
+    /// @param cb Callback invoked inline in the calling thread with the send result.
+    /// @return SsErr::OK if sent successfully, error code otherwise.
+    SsErr send_binary_message(const uint8_t* data, size_t len, SendCompleteCallback cb) override;
+
     /// @brief Sends a client/time message, capturing the timestamp just before send
     /// @return true if the message was sent successfully, false otherwise.
     bool send_time_message() override;

src/esp/server_connection.cpp

@@ -14,6 +14,7 @@
 
 #include "server_connection.h"
 
+#include "binary_send_registry.h"
 #include "lwip/sockets.h"  // for setsockopt, IPPROTO_TCP, NODELAY
 #include "platform/compiler.h"
 #include "platform/logging.h"
@@ -55,19 +56,56 @@ struct SessionLookup {
     std::weak_ptr<SendspinServerConnection> conn;
 };
 
+/// @brief Once-per-connection identity block for queued binary send work (a reusable
+/// SessionLookup: the binary path runs per chunk and must not allocate in steady state)
+///
+/// While a work item is queued, `self` keeps the block alive independently of the connection;
+/// the worker claims it back before resolving `conn`, so teardown with work in flight makes the
+/// worker a clean no-op. httpd_queue_work has no cancellation hook, so work discarded by
+/// httpd_stop would strand the engaged `self` cycle; engaged blocks are therefore tracked in a
+/// registry scoped by their owning httpd handle and reclaimed by
+/// reclaim_orphaned_binary_send_work(handle) once THAT server is stopped -- another live
+/// server's queued work is never touched. The destructor still fails the pending completion.
+struct BinarySendLookup {
+    std::weak_ptr<SendspinServerConnection> conn;
+    std::shared_ptr<BinarySendLookup> self;
+};
+
+// Engage/claim/reclaim transitions live in the host-tested registry; this file only decides
+// when to call them. One process-wide instance, keyed by httpd handle.
+namespace {
+BinarySendRegistry<BinarySendLookup> g_binary_send_registry;
+}  // namespace
+
+void reclaim_orphaned_binary_send_work(httpd_handle_t server) {
+    g_binary_send_registry.reclaim(server);
+}
+
 // ============================================================================
 // SendspinConnection interface implementation
 // ============================================================================
 
 SendspinServerConnection::SendspinServerConnection(httpd_handle_t server, int sockfd)
     : server_(server), sockfd_(sockfd) {
+    // Allocated here, off the send path; the weak self-reference is bound on the first send
+    // (shared_from_this is unusable inside a constructor)
+    this->binary_send_lookup_ = std::make_shared<BinarySendLookup>();
     // Disabling Nagle's algorithm significantly improves the time syncing accuracy
     int nodelay = 1;
     if (setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay)) < 0) {
         SS_LOGW(TAG, "Failed to turn on TCP_NODELAY, syncing may be inaccurate");
     }
 }
 
+SendspinServerConnection::~SendspinServerConnection() {
+    // A still-queued worker can never touch this connection again (weak_ptr lock fails), so the
+    // pending completion is failed here; a worker that DID lock blocks destruction until done
+    if (this->binary_send_in_flight_.load(std::memory_order_acquire) && this->binary_send_cb_) {
+        SendCompleteCallback pending = std::move(this->binary_send_cb_);
+        pending(false);
+    }
+}
+
 void SendspinServerConnection::start() {
     // Time filter is initialized by the hub when it sets up the connection.
 }
@@ -172,6 +210,107 @@ SsErr SendspinServerConnection::send_text_message(const std::string& message,
     return SsErr::OK;
 }
 
+SsErr SendspinServerConnection::send_binary_message(const uint8_t* data, size_t len,
+                                                    SendCompleteCallback on_complete) {
+    if (!this->is_connected()) {
+        if (on_complete) {
+            on_complete(false);
+        }
+        return SsErr::INVALID_STATE;
+    }
+
+    // Single-in-flight slot: a chunk arriving while the previous is still queued is rejected
+    // and the caller drops it (the spec's stall policy)
+    if (this->binary_send_in_flight_.exchange(true, std::memory_order_acq_rel)) {
+        if (on_complete) {
+            on_complete(false);
+        }
+        return SsErr::NOT_FINISHED;
+    }
+
+    // Grow-only buffer sized by the first payload: chunks are near-constant size, so steady
+    // state allocates nothing and any growth is loud. SPIRAM-preferred like the receive buffer.
+    if (this->binary_send_payload_.size() < len) {
+        bool grown;
+        if (this->binary_send_payload_.data() == nullptr) {
+            grown = this->binary_send_payload_.allocate(len, MemoryLocation::PREFER_EXTERNAL);
+        } else {
+            SS_LOGW(TAG, "Growing binary send slot %zu -> %zu bytes",
+                    this->binary_send_payload_.size(), len);
+            grown = this->binary_send_payload_.realloc(len);
+        }
+        if (!grown) {
+            SS_LOGE(TAG, "Failed to allocate %zu bytes for binary send slot", len);
+            this->binary_send_in_flight_.store(false, std::memory_order_release);
+            if (on_complete) {
+                on_complete(false);
+            }
+            return SsErr::NO_MEM;
+        }
+    }
+
+    std::memcpy(this->binary_send_payload_.data(), data, len);
+    this->binary_send_len_ = len;
+    this->binary_send_cb_ = std::move(on_complete);
+
+    if (this->binary_send_lookup_->conn.expired()) {
+        this->binary_send_lookup_->conn =
+            std::static_pointer_cast<SendspinServerConnection>(this->shared_from_this());
+    }
+    // Engage the keep-alive reference for the queued worker, scoped to this connection's httpd
+    // handle for reclamation at that server's stop (see BinarySendLookup).
+    g_binary_send_registry.engage(this->binary_send_lookup_, this->server_);
+
+    if (httpd_queue_work(this->server_, async_send_binary, this->binary_send_lookup_.get()) !=
+        ESP_OK) {
+        SS_LOGE(TAG, "httpd_queue_work failed for binary message");
+        g_binary_send_registry.claim(this->binary_send_lookup_.get());
+        SendCompleteCallback pending = std::move(this->binary_send_cb_);
+        this->binary_send_in_flight_.store(false, std::memory_order_release);
+        if (pending) {
+            pending(false);
+        }
+        return SsErr::FAIL;
+    }
+    return SsErr::OK;
+}
+
+void SendspinServerConnection::async_send_binary(void* arg) {
+    auto* lookup = static_cast<BinarySendLookup*>(arg);
+    // Claim the keep-alive back under the registry lock (serialized against reclaim); a
+    // successful conn.lock() then blocks destruction until return.
+    std::shared_ptr<BinarySendLookup> keep = g_binary_send_registry.claim(lookup);
+    if (keep == nullptr) {
+        return;  // Reclaimed or already claimed; nothing here is safe to touch
+    }
+    auto conn = lookup->conn.lock();
+    if (conn == nullptr) {
+        return;  // Torn down with work queued: the destructor already failed the completion
+    }
+
+    bool success = false;
+    // Same identity and hello gating as async_send_text
+    if (conn->is_connected() && conn->client_hello_sent_) {
+        httpd_ws_frame_t ws_pkt;
+        memset(&ws_pkt, 0, sizeof(httpd_ws_frame_t));
+        ws_pkt.payload = conn->binary_send_payload_.data();
+        ws_pkt.len = conn->binary_send_len_;
+        ws_pkt.type = HTTPD_WS_TYPE_BINARY;
+        success = httpd_ws_send_frame_async(conn->server_, conn->sockfd_, &ws_pkt) == ESP_OK;
+    }
+
+    // The completion fires on every exit path with a live connection (sent, send failed, gated,
+    // or already disconnected) — the slot would wedge otherwise. The callback is moved out and
+    // the slot released before invoking it, so the source task, once woken by the completion,
+    // finds the slot free for its next send. The callback itself does not re-enter the
+    // connection (the interface forbids it); it only records the result and wakes the task.
+    SendCompleteCallback pending = std::move(conn->binary_send_cb_);
+    conn->binary_send_in_flight_.store(false, std::memory_order_release);
+    if (pending) {
+        pending(success);
+    }
+}
+
 void SendspinServerConnection::trigger_close() {
     // Gate on is_connected(): once close_callback has marked this connection closed, httpd may
     // recycle the fd onto a freshly-accepted session, and closing by the stale fd would kill the