[api] Reuse overflow buffer storage instead of allocating per stalled write (#19093)

esphome/components/api/__init__.py

@@ -350,10 +350,9 @@ CONFIG_SCHEMA = cv.All(
                 ln882x=5,  # Moderate RAM
                 nrf52=4,  # ~256KB RAM, BSD sockets, Thread (single HA controller)
             ): cv.int_range(min=1, max=20),
-            # Maximum queued send buffers per connection before dropping connection
-            # Each buffer uses ~8-12 bytes overhead plus actual message size
+            # Max queued messages per connection, and 2 KB of backlog per slot up
+            # to 64 KB (a lone message is exempt), before the connection is dropped
             # Platform defaults based on available RAM and typical message rates:
-            # CONF_MAX_SEND_QUEUE defaults are power of 2 for efficient modulo
             cv.SplitDefault(
                 CONF_MAX_SEND_QUEUE,
                 esp8266=4,  # Limited RAM, need to fail fast

esphome/components/api/api_buffer.cpp

@@ -1,20 +1,37 @@
 #include "api_buffer.h"
-#include <new>
+#ifdef ESPHOME_DEBUG_API
+#include "esphome/core/log.h"
+#endif
 
 namespace esphome::api {
 
+#ifdef ESPHOME_DEBUG_API
+void APIBuffer::debug_check_drop_(size_t drop) const {
+  if (drop > this->size_) {
+    ESP_LOGE("api.buffer", "drop_front: drop=%zu size=%u", drop, this->size_);
+    abort();
+  }
+}
+#endif
+
 bool APIBuffer::grow_(size_t n) {
-  // nothrow (no zero-fill) so OOM is reportable; plain new aborts instead
-  // (NEW_OOM_ABORT on ESP8266 Arduino, exception stub on ESP-IDF).
-  // RAMAllocator is no fit here: unique_ptr needs delete[]-compatible memory.
-  std::unique_ptr<uint8_t[]> new_data(new (std::nothrow) uint8_t[n]);
-  if (new_data == nullptr)
+  if (n > MAX_SIZE)
     return false;
-  if (this->size_)
-    std::memcpy(new_data.get(), this->data_.get(), this->size_);
-  this->data_ = std::move(new_data);
+  // realloc extends in place when it can, avoiding the copy
+  uint8_t *grown = RAMAllocator<uint8_t>().reallocate(this->data_.get(), n);
+  if (grown == nullptr)
+    return false;
+  (void) this->data_.release();  // realloc already freed or reused the old block
+  this->data_.reset(grown);
   this->capacity_ = n;
   return true;
 }
 
+uint8_t *APIBuffer::append(size_t n) {
+  const size_t old_size = this->size_;
+  if (!this->resize(old_size + n))
+    return nullptr;
+  return this->data_.get() + old_size;
+}
+
 }  // namespace esphome::api

esphome/components/api/api_buffer.h

@@ -25,6 +25,7 @@ namespace esphome::api {
 /// writes in debug builds.
 class APIBuffer {
  public:
+  static constexpr size_t MAX_SIZE = UINT16_MAX;  // API frames carry 16 bit lengths
   void clear() { this->size_ = 0; }
   /// Returns false if allocation fails; the buffer is left unchanged.
   [[nodiscard]] inline bool reserve(size_t n) ESPHOME_ALWAYS_INLINE { return n <= this->capacity_ || this->grow_(n); }
@@ -36,9 +37,19 @@ class APIBuffer {
   [[nodiscard]] inline bool reserve_and_resize(size_t reserve_size, size_t new_size) ESPHOME_ALWAYS_INLINE {
     if (!this->reserve(std::max(reserve_size, new_size)))
       return false;
-    this->size_ = new_size;
+    this->size_ = static_cast<uint16_t>(new_size);
     return true;
   }
+  /// Grow by n bytes; returns the new bytes, or nullptr on allocation failure.
+  [[nodiscard]] uint8_t *append(size_t n);
+  /// Drop the first `drop` bytes, sliding the rest down. Precondition: drop <= size().
+  void drop_front(size_t drop) {
+#ifdef ESPHOME_DEBUG_API
+    this->debug_check_drop_(drop);
+#endif
+    this->size_ -= drop;
+    std::memmove(this->data_.get(), this->data_.get() + drop, this->size_);
+  }
   uint8_t *data() { return this->data_.get(); }
   const uint8_t *data() const { return this->data_.get(); }
   size_t size() const { return this->size_; }
@@ -55,9 +66,14 @@ class APIBuffer {
 
  protected:
   bool grow_(size_t n);
-  std::unique_ptr<uint8_t[]> data_;
-  size_t size_{0};
-  size_t capacity_{0};
+#ifdef ESPHOME_DEBUG_API
+  void debug_check_drop_(size_t drop) const;
+#endif
+  // RAMAllocator: PSRAM when available, and it reports failure where
+  // new (std::nothrow) still aborts on ESP-IDF without exceptions
+  RAMUniquePtr<uint8_t[]> data_;
+  uint16_t size_{0};
+  uint16_t capacity_{0};
 };
 
 }  // namespace esphome::api

esphome/components/api/api_connection.cpp

@@ -364,7 +364,10 @@ void APIConnection::check_keepalive_(uint32_t now) {
     ESP_LOGVV(TAG, "Sending keepalive PING");
     PingRequest req;
     this->flags_.sent_ping = this->send_message(req);
-    if (!this->flags_.sent_ping) {
+    if (this->flags_.sent_ping) {
+      // Quiet for a keepalive period and the ping is on its way: a one-off stall's storage can go
+      this->helper_->release_overflow_buffer();
+    } else {
       // If we can't send the ping request directly (tx_buffer full),
       // schedule it at the front of the batch so it will be sent with priority
       ESP_LOGW(TAG, "Buffer full, ping queued");

esphome/components/api/api_frame_helper.cpp

@@ -171,7 +171,7 @@ APIError APIFrameHelper::write_raw_iov_(const struct iovec *iov, int iovcnt, uin
     return APIError::OK;
 
   // Queue unsent data into overflow buffer
-  if (!this->overflow_buf_.enqueue_iov(iov, iovcnt, total_write_len, static_cast<uint16_t>(sent))) {
+  if (!this->overflow_buf_.enqueue_iov(iov, iovcnt, total_write_len, sent)) {
     HELPER_LOG("Overflow buffer full or out of memory, dropping connection");
     this->state_ = State::FAILED;
     return APIError::SOCKET_WRITE_FAILED;

esphome/components/api/api_frame_helper.h

@@ -219,7 +219,10 @@ class APIFrameHelper {
     if (this->rx_buf_len_ == 0) {
       this->rx_buf_.release();
     }
+    this->release_overflow_buffer();
   }
+  // Free the send backlog storage once it has drained
+  void release_overflow_buffer() { this->overflow_buf_.release(); }
 
  protected:
   // Drain backlogged overflow data to the socket and handle errors.

esphome/components/api/api_frame_helper_noise.cpp

@@ -67,15 +67,15 @@ APIError APINoiseFrameHelper::init() {
   }
 
   // init prologue
-  size_t old_size = prologue_.size();
-  if (!prologue_.resize(old_size + PROLOGUE_INIT_LEN)) [[unlikely]] {
+  uint8_t *dst = prologue_.append(PROLOGUE_INIT_LEN);
+  if (dst == nullptr) [[unlikely]] {
     state_ = State::FAILED;
     return APIError::OUT_OF_MEMORY;
   }
 #ifdef USE_ESP8266
-  memcpy_P(prologue_.data() + old_size, PROLOGUE_INIT, PROLOGUE_INIT_LEN);
+  memcpy_P(dst, PROLOGUE_INIT, PROLOGUE_INIT_LEN);
 #else
-  std::memcpy(prologue_.data() + old_size, PROLOGUE_INIT, PROLOGUE_INIT_LEN);
+  std::memcpy(dst, PROLOGUE_INIT, PROLOGUE_INIT_LEN);
 #endif
 
   state_ = State::CLIENT_HELLO;
@@ -272,17 +272,17 @@ APIError APINoiseFrameHelper::state_action_client_hello_() {
     return handle_handshake_frame_error_(aerr);
   }
   // ignore contents, may be used in future for flags
-  // Resize for: existing prologue + 2 size bytes + frame data
-  size_t old_size = this->prologue_.size();
+  // Append 2 size bytes + frame data to the prologue
   size_t rx_size = this->rx_buf_.size();
-  if (!this->prologue_.resize(old_size + 2 + rx_size)) [[unlikely]] {
+  uint8_t *dst = this->prologue_.append(2 + rx_size);
+  if (dst == nullptr) [[unlikely]] {
     state_ = State::FAILED;
     return APIError::OUT_OF_MEMORY;
   }
-  this->prologue_[old_size] = (uint8_t) (rx_size >> 8);
-  this->prologue_[old_size + 1] = (uint8_t) rx_size;
+  dst[0] = (uint8_t) (rx_size >> 8);
+  dst[1] = (uint8_t) rx_size;
   if (rx_size > 0) {
-    std::memcpy(this->prologue_.data() + old_size + 2, this->rx_buf_.data(), rx_size);
+    std::memcpy(dst + 2, this->rx_buf_.data(), rx_size);
   }
 
   state_ = State::SERVER_HELLO;

esphome/components/api/api_overflow_buffer.cpp

@@ -1,98 +1,91 @@
 #include "api_overflow_buffer.h"
 #ifdef USE_API
 #include <cstring>
-#include <new>
 
 namespace esphome::api {
 
-APIOverflowBuffer::~APIOverflowBuffer() {
-  for (auto *entry : this->queue_) {
-    if (entry != nullptr)
-      Entry::destroy(entry);
-  }
-}
-
 ssize_t APIOverflowBuffer::try_drain(socket::Socket *socket) {
-  // socket->write() can re-enter this function: a log message emitted from an
-  // lwip callback during the write goes out over the API and lands back in the
-  // frame helper's write/drain path. If a nested drain ran here it would send
-  // and free the entry the outer drain is still holding, causing a double free.
-  // Report "no progress" instead; the outer drain keeps draining, and the
-  // nested send is enqueued behind the existing backlog.
+  // Nested call from inside socket->write(); see draining_
   if (this->draining_)
     return 0;
 
-  // RAII so the flag is cleared on every return path
   struct DrainGuard {
-    explicit DrainGuard(bool &flag) : flag_(flag) { flag_ = true; }
-    ~DrainGuard() { this->flag_ = false; }
-    bool &flag_;
-  } guard(this->draining_);
+    APIOverflowBuffer &owner;
+    ~DrainGuard() { this->owner.draining_ = false; }
+  } guard{*this};
+  this->draining_ = true;
 
   while (this->count_ > 0) {
-    Entry *front = this->queue_[this->head_];
+    uint8_t *msg = this->buf_.data() + this->head_;
+    size_t len = msg[0] | (msg[1] << 8);
 
-    ssize_t sent = socket->write(front->current_data(), front->remaining());
-
-    if (sent <= 0) {
-      // -1 = error (caller checks errno for EWOULDBLOCK vs hard error)
-      // 0 = nothing sent (treat as no progress)
+    ssize_t sent = socket->write(msg + LEN_PREFIX, len);
+    if (sent <= 0)
+      return sent;
+    if (static_cast<size_t>(sent) < len) {
+      // Step past the sent bytes and rewrite the prefix there; it lands on bytes already sent
+      this->head_ += sent;
+      len -= sent;
+      msg += sent;
+      msg[0] = len;
+      msg[1] = len >> 8;
       return sent;
     }
-
-    if (static_cast<uint16_t>(sent) < front->remaining()) {
-      // Partially sent, update offset and stop
-      front->offset += static_cast<uint16_t>(sent);
-      return sent;
-    }
-
-    // Entry fully sent — unlink it before freeing so a freed pointer is never
-    // reachable from the queue
-    this->queue_[this->head_] = nullptr;
-    this->head_ = (this->head_ + 1) % API_MAX_SEND_QUEUE;
+    this->head_ += LEN_PREFIX + len;
     this->count_--;
-    Entry::destroy(front);
   }
 
-  return 0;  // All drained
+  this->head_ = 0;
+  if (this->release_when_drained_) {
+    this->release_when_drained_ = false;
+    this->buf_.release();
+  } else {
+    this->buf_.clear();
+  }
+  return 0;
 }
 
-bool APIOverflowBuffer::enqueue_iov(const struct iovec *iov, int iovcnt, uint16_t total_len, uint16_t skip) {
+bool APIOverflowBuffer::enqueue_iov(const struct iovec *iov, int iovcnt, size_t total_len, size_t skip) {
   if (this->count_ >= API_MAX_SEND_QUEUE)
     return false;
 
-  uint16_t buffer_size = total_len - skip;
-  // nothrow: a failed allocation returns nullptr so the connection is dropped
-  // cleanly instead of plain new's crash or abort on OOM
-  // NOLINTNEXTLINE(cppcoreguidelines-owning-memory)
-  auto *data = new (std::nothrow) uint8_t[buffer_size];
-  if (data == nullptr)
-    return false;
-  // NOLINTNEXTLINE(cppcoreguidelines-owning-memory)
-  auto *entry = new (std::nothrow) Entry{data, buffer_size, 0};
-  if (entry == nullptr) {
-    delete[] data;
+  const size_t new_len = total_len - skip;
+  const size_t new_bytes = LEN_PREFIX + new_len;
+  const size_t live = this->buf_.size() - this->head_;
+  // A lone message is only bound by the buffer; refusing it would just drop the connection
+  if (live + new_bytes > (this->count_ > 0 ? MAX_BYTES : MAX_LONE_BYTES))
     return false;
+
+  if (this->buf_.size() + new_bytes > this->buf_.capacity()) {
+    // Storage would move under an outer drain's write()
+    if (this->draining_)
+      return false;
+    if (this->head_ > 0) {
+      // Reclaim the sent prefix before growing
+      this->buf_.drop_front(this->head_);
+      this->head_ = 0;
+    }
+    if (!this->buf_.reserve(reserve_for(live + new_bytes)))
+      return false;
   }
 
-  uint16_t to_skip = skip;
-  uint16_t write_pos = 0;
-
-  for (int i = 0; i < iovcnt; i++) {
-    if (to_skip >= iov[i].iov_len) {
-      to_skip -= static_cast<uint16_t>(iov[i].iov_len);
+  uint8_t *dst = this->buf_.append(new_bytes);
+  if (dst == nullptr)
+    return false;
+  dst[0] = new_len;
+  dst[1] = new_len >> 8;
+  dst += LEN_PREFIX;
+  for (const struct iovec *end = iov + iovcnt; iov != end; iov++) {
+    if (skip >= iov->iov_len) {
+      skip -= iov->iov_len;
     } else {
-      const uint8_t *src = reinterpret_cast<uint8_t *>(iov[i].iov_base) + to_skip;
-      uint16_t len = static_cast<uint16_t>(iov[i].iov_len) - to_skip;
-      std::memcpy(entry->data + write_pos, src, len);
-      write_pos += len;
-      to_skip = 0;
+      const size_t len = iov->iov_len - skip;
+      std::memcpy(dst, static_cast<const uint8_t *>(iov->iov_base) + skip, len);
+      dst += len;
+      skip = 0;
     }
   }
 
-  // Publish only after the copy completes so a half-built entry is never reachable
-  this->queue_[this->tail_] = entry;
-  this->tail_ = (this->tail_ + 1) % API_MAX_SEND_QUEUE;
   this->count_++;
   return true;
 }

esphome/components/api/api_overflow_buffer.h

@@ -1,5 +1,6 @@
 #pragma once
-#include <array>
+#include <algorithm>
+#include <cstddef>
 #include <cstdint>
 #include <sys/types.h>
 
@@ -8,71 +9,57 @@
 
 #include "esphome/components/socket/headers.h"
 #include "esphome/components/socket/socket.h"
-#include "esphome/core/helpers.h"
+#include "api_buffer.h"
 
 namespace esphome::api {
 
-/// Circular queue of heap-allocated byte buffers used as a TCP send backlog.
-///
-/// Under normal operation this buffer is **never used** — data goes straight
-/// from the frame helper to the socket.  It only fills when the LWIP TCP
-/// send buffer is full (slow client, congested network, heavy logging).
-/// The queue drains automatically on subsequent write/loop calls once the
-/// socket becomes writable again.
-///
-/// Capacity is compile-time-fixed via API_MAX_SEND_QUEUE (set from Python
-/// config).  If the queue fills completely the connection is marked failed.
+/// TCP send backlog, only used when the socket send buffer is full.
+/// One contiguous buffer per connection, allocated on the first stall and
+/// kept at its high-water mark so a lossy link does not churn the heap.
+/// Messages are stored as a 2 byte length prefix plus payload.
+/// API_MAX_SEND_QUEUE bounds queued messages and, at 2 KB per slot, queued
+/// bytes; exceeding either fails the connection.
 class APIOverflowBuffer {
  public:
-  /// A single heap-allocated send-backlog entry.
-  /// Lifetime is manually managed — see destroy().
-  struct Entry {
-    uint8_t *data;
-    uint16_t size;    // Total size of the buffer
-    uint16_t offset;  // Current send offset within the buffer
-
-    uint16_t remaining() const { return this->size - this->offset; }
-    const uint8_t *current_data() const { return this->data + this->offset; }
-
-    /// Free this entry and its data buffer.
-    static ESPHOME_ALWAYS_INLINE void destroy(Entry *entry) {
-      delete[] entry->data;
-      delete entry;  // NOLINT(cppcoreguidelines-owning-memory)
-    }
-  };
-
-  ~APIOverflowBuffer();
-
   /// True when no backlogged data is waiting.
   bool empty() const { return this->count_ == 0; }
 
-  /// True when the queue has no room for another entry.
-  bool full() const { return this->count_ >= API_MAX_SEND_QUEUE; }
-
-  /// Number of entries currently queued.
-  uint8_t count() const { return this->count_; }
-
-  /// Try to drain queued data to the socket.
-  /// Returns bytes-written > 0 on success/partial, 0 if all drained or no progress,
-  /// -1 on error (caller must check errno to distinguish EWOULDBLOCK from hard errors).
-  /// Callers only need to act on -1; 0 and positive values both mean "no error".
-  /// Frees entries as they are fully sent.
+  /// Drain queued messages to the socket.
+  /// Returns bytes written, 0 for a re-entrant call, -1 on error (check errno
+  /// for EWOULDBLOCK); callers only need to act on -1.
   ssize_t try_drain(socket::Socket *socket);
 
-  /// Enqueue unsent IOV data into the backlog.
-  /// Copies iov data starting at byte offset `skip` into a new entry.
-  /// Returns false if the queue is full or allocation fails (caller should fail the connection).
-  bool enqueue_iov(const struct iovec *iov, int iovcnt, uint16_t total_len, uint16_t skip);
+  /// Queue iov data from byte offset `skip` as one message.
+  /// Returns false when a limit is hit, allocation fails, or storage would move
+  /// during a drain; the caller should fail the connection.
+  bool enqueue_iov(const struct iovec *iov, int iovcnt, size_t total_len, size_t skip);
+
+  /// Free the retained storage, now if empty, otherwise once it has drained.
+  void release() {
+    if (this->count_ == 0) {
+      this->buf_.release();
+    } else {
+      this->release_when_drained_ = true;
+    }
+  }
 
  protected:
-  std::array<Entry *, API_MAX_SEND_QUEUE> queue_{};
-  uint8_t head_{0};
-  uint8_t tail_{0};
+  static constexpr size_t LEN_PREFIX = 2;
+  static constexpr size_t BYTES_PER_SLOT = 2048;
+  // Reserve in 256 byte steps so a creeping high-water mark settles quickly
+  static constexpr size_t GROW_QUANTUM = 256;
+  // Lone message ceiling, rounded down so reserve_for() never exceeds the buffer limit
+  static constexpr size_t MAX_LONE_BYTES = APIBuffer::MAX_SIZE & ~(GROW_QUANTUM - 1);
+  static constexpr size_t MAX_BYTES = std::min(API_MAX_SEND_QUEUE * BYTES_PER_SLOT, MAX_LONE_BYTES);
+  static constexpr size_t reserve_for(size_t want) { return (want + GROW_QUANTUM - 1) & ~(GROW_QUANTUM - 1); }
+
+  APIBuffer buf_;
+  uint16_t head_{0};  // offset of the front message's length prefix; bytes before it are sent
   uint8_t count_{0};
-  // Guards against re-entrant drains: socket->write() can re-enter the API
-  // send path (e.g. a log message emitted from an lwip callback), and a nested
-  // drain would free the entry the outer drain is still holding.
-  bool draining_{false};
+  // socket->write() can re-enter the send path (log from an lwip callback):
+  // a nested drain makes no progress and a nested enqueue never moves storage
+  bool draining_ : 1 {false};
+  bool release_when_drained_ : 1 {false};
 };
 
 }  // namespace esphome::api

tests/components/api/__init__.py

@@ -0,0 +1,17 @@
+import esphome.codegen as cg
+from esphome.core import CORE
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    # USE_API compiles every api source, so emit what they need. No socket
+    # override: an __init__.py there makes pytest import its conftest as socket.conftest.
+    async def to_code_testing(config):
+        cg.add_define("USE_API")
+        cg.add_define("USE_API_PLAINTEXT")
+        cg.add_define("API_MAX_SEND_QUEUE", 8)
+        cg.add_define("MAX_API_CONNECTIONS", 1)
+        cg.add_define("USE_SOCKET_IMPL_BSD_SOCKETS")
+        CORE.register_controller()  # api_server registers with the controller registry
+
+    manifest.to_code = to_code_testing

tests/components/api/test_api_buffer.cpp

@@ -0,0 +1,65 @@
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <cstring>
+
+#include "esphome/components/api/api_buffer.h"
+
+namespace esphome::api::testing {
+
+// Pointer plus two 16 bit sizes
+static_assert(sizeof(APIBuffer) <= 2 * sizeof(void *));
+
+TEST(APIBuffer, RefusesSizesAbove16Bits) {
+  APIBuffer buf;
+  ASSERT_TRUE(buf.resize(16));
+  EXPECT_FALSE(buf.reserve(UINT16_MAX + 1));
+  EXPECT_EQ(buf.size(), 16u);
+  EXPECT_EQ(buf.capacity(), 16u);
+  EXPECT_TRUE(buf.reserve(UINT16_MAX));
+  EXPECT_EQ(buf.capacity(), UINT16_MAX);
+}
+
+static const uint8_t BYTES[] = {1, 2, 3, 4, 5, 6};
+
+TEST(APIBuffer, AppendReturnsTheNewBytes) {
+  APIBuffer buf;
+  ASSERT_TRUE(buf.reserve(8));
+  uint8_t *first = buf.append(3);
+  ASSERT_NE(first, nullptr);
+  std::memcpy(first, BYTES, 3);
+  EXPECT_EQ(buf.size(), 3u);
+  EXPECT_EQ(buf.capacity(), 8u);
+
+  // Grows through realloc and keeps what was there
+  uint8_t *second = buf.append(6);
+  ASSERT_EQ(second, buf.data() + 3);
+  std::memcpy(second, BYTES + 3, 3);
+  EXPECT_EQ(buf.size(), 9u);
+  EXPECT_EQ(buf.capacity(), 9u);
+  EXPECT_EQ(std::memcmp(buf.data(), BYTES, 6), 0);
+}
+
+TEST(APIBuffer, DropFrontSlidesTheRestDown) {
+  APIBuffer buf;
+  uint8_t *bytes = buf.append(6);
+  ASSERT_NE(bytes, nullptr);
+  std::memcpy(bytes, BYTES, 6);
+
+  buf.drop_front(2);
+  EXPECT_EQ(buf.size(), 4u);
+  EXPECT_EQ(buf.capacity(), 6u);
+  EXPECT_EQ(std::memcmp(buf.data(), BYTES + 2, 4), 0);
+
+  // Growing afterwards keeps the slid bytes
+  ASSERT_TRUE(buf.reserve(64));
+  EXPECT_EQ(buf.size(), 4u);
+  EXPECT_EQ(std::memcmp(buf.data(), BYTES + 2, 4), 0);
+
+  // Dropping everything leaves an empty buffer with its capacity
+  buf.drop_front(4);
+  EXPECT_EQ(buf.size(), 0u);
+  EXPECT_EQ(buf.capacity(), 64u);
+}
+
+}  // namespace esphome::api::testing

tests/components/api/test_overflow_buffer.cpp

@@ -0,0 +1,510 @@
+#include <gtest/gtest.h>
+
+#include <fcntl.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <algorithm>
+#include <cerrno>
+#include <cstdint>
+#include <cstring>
+#include <initializer_list>
+#include <memory>
+#include <vector>
+
+#include "esphome/components/api/api_overflow_buffer.h"
+
+#ifdef USE_HOST
+namespace esphome::api::testing {
+
+// Idle cost is the buffer plus one word of bookkeeping
+static_assert(sizeof(APIOverflowBuffer) <= sizeof(APIBuffer) + sizeof(void *));
+
+// Exposes storage so tests can check it is reused, not reallocated
+class TestOverflowBuffer : public APIOverflowBuffer {
+ public:
+  using APIOverflowBuffer::LEN_PREFIX;
+  using APIOverflowBuffer::MAX_BYTES;
+  using APIOverflowBuffer::MAX_LONE_BYTES;
+  struct Storage {
+    size_t capacity;
+    const uint8_t *data;
+    bool operator==(const Storage &) const = default;
+  };
+  size_t capacity() const { return this->buf_.capacity(); }
+  Storage storage() const { return {this->buf_.capacity(), this->buf_.data()}; }
+  uint8_t count() const { return this->count_; }
+  size_t live() const { return this->buf_.size() - this->head_; }
+  /// Simulates a socket write inside try_drain() re-entering the send path
+  void set_draining(bool draining) { this->draining_ = draining; }
+};
+
+static std::vector<uint8_t> make_message(size_t len, uint8_t seed) {
+  std::vector<uint8_t> msg(len);
+  for (size_t i = 0; i < len; i++)
+    msg[i] = static_cast<uint8_t>(seed + i);
+  return msg;
+}
+
+static bool enqueue(TestOverflowBuffer &buf, const std::vector<uint8_t> &msg, uint16_t skip = 0) {
+  struct iovec iov = {const_cast<uint8_t *>(msg.data()), msg.size()};
+  return buf.enqueue_iov(&iov, 1, static_cast<uint16_t>(msg.size()), skip);
+}
+
+static void append(std::vector<uint8_t> &dst, const std::vector<uint8_t> &src, size_t skip = 0) {
+  dst.insert(dst.end(), src.begin() + skip, src.end());
+}
+
+static std::vector<uint8_t> concat(std::initializer_list<std::vector<uint8_t>> parts) {
+  std::vector<uint8_t> out;
+  for (const auto &part : parts)
+    append(out, part);
+  return out;
+}
+
+/// The pipe delivers the filler first, then the drained messages.
+static void expect_after_filler(const std::vector<uint8_t> &received, size_t filler,
+                                const std::vector<uint8_t> &expected) {
+  ASSERT_EQ(received.size(), filler + expected.size());
+  EXPECT_TRUE(std::equal(expected.begin(), expected.end(), received.begin() + filler));
+}
+
+// Non-blocking socket pair with small buffers, so the writer fills like a stalled TCP connection
+class OverflowBufferTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    int fds[2];
+    ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+    int size = 4096;
+    ASSERT_EQ(::setsockopt(fds[0], SOL_SOCKET, SO_SNDBUF, &size, sizeof(size)), 0);
+    ASSERT_EQ(::setsockopt(fds[1], SOL_SOCKET, SO_RCVBUF, &size, sizeof(size)), 0);
+    ASSERT_EQ(::fcntl(fds[1], F_SETFL, O_NONBLOCK), 0);
+    this->reader_ = fds[1];
+    this->sock_ = std::make_unique<socket::Socket>(fds[0]);
+    ASSERT_EQ(this->sock_->setblocking(false), 0);
+  }
+  void TearDown() override { ::close(this->reader_); }
+
+  /// Write filler until the socket refuses; returns the bytes accepted
+  size_t fill_pipe_() {
+    uint8_t junk[512];
+    std::memset(junk, 0xEE, sizeof(junk));
+    size_t total = 0;
+    for (;;) {
+      ssize_t written = this->sock_->write(junk, sizeof(junk));
+      if (written <= 0)
+        break;
+      total += static_cast<size_t>(written);
+    }
+    return total;
+  }
+
+  /// Append whatever the pipe currently holds.
+  void read_into_(std::vector<uint8_t> &out) {
+    uint8_t tmp[1024];
+    for (;;) {
+      ssize_t n = ::read(this->reader_, tmp, sizeof(tmp));
+      if (n <= 0)
+        break;
+      out.insert(out.end(), tmp, tmp + n);
+    }
+  }
+
+  /// Drain once; a refusal must be a would-block, never a hard error.
+  ssize_t drain_(TestOverflowBuffer &buf) {
+    ssize_t sent = buf.try_drain(this->sock_.get());
+    if (sent == -1) {
+      EXPECT_TRUE(errno == EWOULDBLOCK || errno == EAGAIN);
+    }
+    return sent;
+  }
+
+  /// Read and drain until the backlog is empty; returns all bytes received
+  std::vector<uint8_t> drain_all_(TestOverflowBuffer &buf) {
+    std::vector<uint8_t> received;
+    for (int i = 0; i < 10000 && !buf.empty(); i++) {
+      this->read_into_(received);
+      // A hard socket error would never clear the backlog; stop instead of spinning
+      if (this->drain_(buf) == -1 && errno != EWOULDBLOCK && errno != EAGAIN)
+        break;
+    }
+    EXPECT_TRUE(buf.empty());
+    this->read_into_(received);
+    return received;
+  }
+
+  struct Stall {
+    size_t filler;
+    std::vector<uint8_t> first, second, received;
+    TestOverflowBuffer::Storage before;
+  };
+  /// Park two messages, then drain the first fully and the second part way
+  void stall_mid_message_(TestOverflowBuffer &buf, Stall &s) {
+    s.filler = this->fill_pipe_();
+    s.first = make_message(1500, 20);
+    ASSERT_GT(s.filler, s.first.size());  // the first message must drain in one go
+    // Larger than the whole pipe, so a drain always stops inside it
+    s.second = make_message(std::max<size_t>(s.filler + 1, std::min<size_t>(s.filler * 3, 12000)), 60);
+    ASSERT_GT(s.second.size(), s.filler);
+    ASSERT_TRUE(enqueue(buf, s.first));
+    ASSERT_TRUE(enqueue(buf, s.second));
+    s.before = buf.storage();
+    this->read_into_(s.received);
+    ASSERT_GT(this->drain_(buf), 0);
+    ASSERT_EQ(buf.count(), 1);
+  }
+
+  int reader_{-1};
+  std::unique_ptr<socket::Socket> sock_;
+};
+
+TEST_F(OverflowBufferTest, IdleBufferOwnsNoStorage) {
+  TestOverflowBuffer buf;
+  EXPECT_TRUE(buf.empty());
+  EXPECT_EQ(buf.capacity(), 0u);
+  EXPECT_EQ(buf.storage().data, nullptr);
+}
+
+TEST_F(OverflowBufferTest, StorageIsReusedAcrossStalls) {
+  TestOverflowBuffer buf;
+  auto msg = make_message(1000, 1);
+
+  size_t filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, msg));
+  const auto storage = buf.storage();
+  EXPECT_GE(storage.capacity, msg.size() + TestOverflowBuffer::LEN_PREFIX);
+
+  for (int stall = 0; stall < 5; stall++) {
+    expect_after_filler(this->drain_all_(buf), filler, msg);
+    EXPECT_TRUE(buf.empty());
+    // Same allocation every time: no free, no new allocation
+    EXPECT_EQ(buf.storage(), storage);
+
+    filler = this->fill_pipe_();
+    ASSERT_TRUE(enqueue(buf, msg));
+    EXPECT_EQ(buf.storage(), storage);
+  }
+}
+
+TEST_F(OverflowBufferTest, ReleaseWhileQueuedFreesOnceDrained) {
+  TestOverflowBuffer buf;
+  auto msg = make_message(1000, 7);
+
+  size_t filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, msg));
+  const size_t capacity = buf.capacity();
+
+  // Requested while the backlog still holds data: storage must stay until sent
+  buf.release();
+  EXPECT_FALSE(buf.empty());
+  EXPECT_EQ(buf.capacity(), capacity);
+
+  expect_after_filler(this->drain_all_(buf), filler, msg);
+  EXPECT_TRUE(buf.empty());
+  EXPECT_EQ(buf.capacity(), 0u);
+  EXPECT_EQ(buf.storage().data, nullptr);
+
+  // A later stall allocates again and keeps it, since nobody asked for a release
+  filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, msg));
+  EXPECT_GT(buf.capacity(), 0u);
+  this->drain_all_(buf);
+  EXPECT_GT(buf.capacity(), 0u);
+}
+
+TEST_F(OverflowBufferTest, ReleaseWhenEmptyFreesImmediately) {
+  TestOverflowBuffer buf;
+  auto msg = make_message(100, 3);
+
+  this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, msg));
+  this->drain_all_(buf);
+  EXPECT_GT(buf.capacity(), 0u);
+
+  buf.release();
+  EXPECT_EQ(buf.capacity(), 0u);
+  EXPECT_EQ(buf.storage().data, nullptr);
+}
+
+TEST_F(OverflowBufferTest, PreservesOrderAndSkipsSentPrefix) {
+  TestOverflowBuffer buf;
+  auto first = make_message(700, 10);
+  auto second_a = make_message(300, 50);
+  auto second_b = make_message(400, 90);
+  auto third = make_message(200, 130);
+
+  size_t filler = this->fill_pipe_();
+  // 100 bytes of the first message were already accepted by the socket
+  ASSERT_TRUE(enqueue(buf, first, 100));
+  // Two iovecs with the skip covering all of the first one plus part of the second
+  struct iovec iov[2] = {{second_a.data(), second_a.size()}, {second_b.data(), second_b.size()}};
+  const uint16_t second_skip = static_cast<uint16_t>(second_a.size() + 5);
+  ASSERT_TRUE(buf.enqueue_iov(iov, 2, static_cast<uint16_t>(second_a.size() + second_b.size()), second_skip));
+  ASSERT_TRUE(enqueue(buf, third));
+  EXPECT_EQ(buf.count(), 3);
+
+  // Nothing can go out while the pipe is full
+  EXPECT_EQ(this->drain_(buf), -1);
+  EXPECT_EQ(buf.count(), 3);
+
+  std::vector<uint8_t> expected;
+  append(expected, first, 100);
+  append(expected, second_b, 5);
+  append(expected, third);
+  expect_after_filler(this->drain_all_(buf), filler, expected);
+}
+
+TEST_F(OverflowBufferTest, RefusesWhenQueueIsFull) {
+  TestOverflowBuffer buf;
+  auto msg = make_message(16, 1);
+
+  size_t filler = this->fill_pipe_();
+  for (int i = 0; i < API_MAX_SEND_QUEUE; i++) {
+    ASSERT_TRUE(enqueue(buf, msg)) << "message " << i;
+  }
+  EXPECT_FALSE(enqueue(buf, msg));
+  EXPECT_EQ(buf.count(), API_MAX_SEND_QUEUE);
+
+  // Draining frees the slots again
+  std::vector<uint8_t> expected;
+  for (int i = 0; i < API_MAX_SEND_QUEUE; i++)
+    append(expected, msg);
+  expect_after_filler(this->drain_all_(buf), filler, expected);
+  this->fill_pipe_();
+  EXPECT_TRUE(enqueue(buf, msg));
+  EXPECT_EQ(buf.count(), 1);
+}
+
+TEST_F(OverflowBufferTest, SkipAtIovecBoundary) {
+  TestOverflowBuffer buf;
+  auto sent = make_message(300, 50);
+  auto unsent = make_message(400, 90);
+
+  size_t filler = this->fill_pipe_();
+  // The skip covers the first iovec exactly, so only the second is copied
+  struct iovec iov[2] = {{sent.data(), sent.size()}, {unsent.data(), unsent.size()}};
+  ASSERT_TRUE(
+      buf.enqueue_iov(iov, 2, static_cast<uint16_t>(sent.size() + unsent.size()), static_cast<uint16_t>(sent.size())));
+  EXPECT_EQ(buf.live(), unsent.size() + TestOverflowBuffer::LEN_PREFIX);
+  expect_after_filler(this->drain_all_(buf), filler, unsent);
+}
+
+TEST_F(OverflowBufferTest, AppendsBehindSentPrefixWhenItFits) {
+  TestOverflowBuffer buf;
+  size_t filler = this->fill_pipe_();
+  auto first = make_message(200, 20);
+  // Size the second message so the two land half way into a 256 byte step,
+  // leaving exactly 128 bytes of slack whatever the pipe accepted
+  const size_t base = std::max<size_t>(filler + 1, std::min<size_t>(filler * 3, 12000));
+  const size_t second_len = (base / 256 + 1) * 256 + 128 - first.size() - 2 * TestOverflowBuffer::LEN_PREFIX;
+  auto second = make_message(second_len, 60);
+  ASSERT_GT(second.size(), filler);
+  ASSERT_TRUE(enqueue(buf, first));
+  ASSERT_TRUE(enqueue(buf, second));
+  const auto storage = buf.storage();
+  const size_t slack = storage.capacity - first.size() - second.size() - 2 * TestOverflowBuffer::LEN_PREFIX;
+  ASSERT_EQ(slack, 128u);
+  auto third = make_message(slack - TestOverflowBuffer::LEN_PREFIX, 200);
+
+  std::vector<uint8_t> received;
+  this->read_into_(received);
+  ASSERT_GT(this->drain_(buf), 0);
+  ASSERT_EQ(buf.count(), 1);
+  const size_t live = buf.live();
+
+  // Fits in the tail, so the sent prefix is left alone
+  ASSERT_TRUE(enqueue(buf, third));
+  EXPECT_EQ(buf.storage(), storage);
+  EXPECT_EQ(buf.live(), live + third.size() + TestOverflowBuffer::LEN_PREFIX);
+
+  append(received, this->drain_all_(buf));
+  expect_after_filler(received, filler, concat({first, second, third}));
+}
+
+TEST_F(OverflowBufferTest, ReleaseSurvivesFurtherEnqueues) {
+  TestOverflowBuffer buf;
+  auto first = make_message(300, 7);
+  auto second = make_message(300, 70);
+
+  size_t filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, first));
+  buf.release();
+  ASSERT_TRUE(enqueue(buf, second));
+  EXPECT_GT(buf.capacity(), 0u);
+
+  expect_after_filler(this->drain_all_(buf), filler, concat({first, second}));
+  EXPECT_EQ(buf.capacity(), 0u);
+}
+
+TEST_F(OverflowBufferTest, RefusesWhenByteLimitIsExceeded) {
+  TestOverflowBuffer buf;
+  // Two of these fill the byte budget exactly, well before the slot count is reached
+  static_assert(API_MAX_SEND_QUEUE >= 3);
+  auto msg = make_message(TestOverflowBuffer::MAX_BYTES / 2 - TestOverflowBuffer::LEN_PREFIX, 1);
+
+  this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, msg));
+  ASSERT_TRUE(enqueue(buf, msg));
+  EXPECT_FALSE(enqueue(buf, msg));
+  EXPECT_EQ(buf.count(), 2);
+}
+
+TEST_F(OverflowBufferTest, LoneMessageMayExceedByteLimit) {
+  TestOverflowBuffer buf;
+  // The oversized message must still fit under the lone message ceiling
+  static_assert(TestOverflowBuffer::MAX_BYTES + 100 + TestOverflowBuffer::LEN_PREFIX <=
+                TestOverflowBuffer::MAX_LONE_BYTES);
+  auto big = make_message(TestOverflowBuffer::MAX_BYTES + 100, 5);
+  auto small = make_message(16, 9);
+
+  // Refusing the only message would drop the connection for nothing
+  size_t filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, big));
+  EXPECT_EQ(buf.count(), 1);
+  // With a backlog present the byte limit applies again
+  EXPECT_FALSE(enqueue(buf, small));
+  EXPECT_EQ(buf.count(), 1);
+
+  expect_after_filler(this->drain_all_(buf), filler, big);
+}
+
+TEST_F(OverflowBufferTest, LoneMessageAboveOffsetLimitIsRefused) {
+  TestOverflowBuffer buf;
+  // Payload plus prefix is past the lone message ceiling
+  auto msg = make_message(TestOverflowBuffer::MAX_LONE_BYTES, 3);
+
+  this->fill_pipe_();
+  EXPECT_FALSE(enqueue(buf, msg));
+  EXPECT_TRUE(buf.empty());
+  EXPECT_EQ(buf.capacity(), 0u);
+}
+
+TEST_F(OverflowBufferTest, HardSocketErrorLeavesBacklogIntact) {
+  TestOverflowBuffer buf;
+  auto msg = make_message(300, 40);
+
+  this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, msg));
+  // A closed socket fails every write outright, unlike a full one
+  ASSERT_EQ(this->sock_->close(), 0);
+
+  errno = 0;
+  EXPECT_EQ(buf.try_drain(this->sock_.get()), -1);
+  EXPECT_NE(errno, EWOULDBLOCK);
+  EXPECT_NE(errno, EAGAIN);
+  EXPECT_EQ(buf.count(), 1);
+  EXPECT_EQ(buf.live(), msg.size() + TestOverflowBuffer::LEN_PREFIX);
+}
+
+TEST_F(OverflowBufferTest, GrowsWhileReclaimingSentPrefix) {
+  TestOverflowBuffer buf;
+  Stall s;
+  ASSERT_NO_FATAL_FAILURE(this->stall_mid_message_(buf, s));
+
+  // One byte too many to fit even after the sent prefix is reclaimed: grows in one copy
+  auto third = make_message(s.before.capacity - buf.live() + 1, 200);
+  ASSERT_TRUE(enqueue(buf, third));
+  EXPECT_GT(buf.capacity(), s.before.capacity);
+  EXPECT_EQ(buf.count(), 2);
+
+  append(s.received, this->drain_all_(buf));
+  expect_after_filler(s.received, s.filler, concat({s.first, s.second, third}));
+}
+
+TEST_F(OverflowBufferTest, NestedDrainMakesNoProgress) {
+  TestOverflowBuffer buf;
+  auto msg = make_message(300, 40);
+
+  size_t filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, msg));
+  std::vector<uint8_t> received;
+  this->read_into_(received);
+
+  // Room is available, but a nested drain must leave the outer one's message alone
+  buf.set_draining(true);
+  EXPECT_EQ(this->drain_(buf), 0);
+  EXPECT_EQ(buf.count(), 1);
+  std::vector<uint8_t> nothing;
+  this->read_into_(nothing);
+  EXPECT_TRUE(nothing.empty());
+
+  buf.set_draining(false);
+  append(received, this->drain_all_(buf));
+  expect_after_filler(received, filler, msg);
+}
+
+TEST_F(OverflowBufferTest, NestedEnqueueAppendsWithinCapacity) {
+  TestOverflowBuffer buf;
+  auto first = make_message(500, 10);
+  auto second = make_message(4, 90);
+
+  size_t filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, first));
+  const auto storage = buf.storage();
+  ASSERT_GE(storage.capacity, first.size() + second.size() + 2 * TestOverflowBuffer::LEN_PREFIX);
+
+  buf.set_draining(true);
+  EXPECT_TRUE(enqueue(buf, second));
+  EXPECT_EQ(buf.count(), 2);
+  EXPECT_EQ(buf.storage(), storage);
+  buf.set_draining(false);
+
+  expect_after_filler(this->drain_all_(buf), filler, concat({first, second}));
+}
+
+TEST_F(OverflowBufferTest, NestedEnqueueRefusesToGrow) {
+  TestOverflowBuffer buf;
+  auto first = make_message(500, 10);
+  auto second = make_message(100, 90);
+
+  size_t filler = this->fill_pipe_();
+  ASSERT_TRUE(enqueue(buf, first));
+  const auto storage = buf.storage();
+  ASSERT_LT(storage.capacity, first.size() + second.size() + 2 * TestOverflowBuffer::LEN_PREFIX);
+
+  // Growing would free the bytes the outer write() is sending from
+  buf.set_draining(true);
+  EXPECT_FALSE(enqueue(buf, second));
+  EXPECT_EQ(buf.count(), 1);
+  EXPECT_EQ(buf.storage(), storage);
+  buf.set_draining(false);
+
+  expect_after_filler(this->drain_all_(buf), filler, first);
+}
+
+TEST_F(OverflowBufferTest, NestedEnqueueRefusesToCompact) {
+  TestOverflowBuffer buf;
+  Stall s;
+  ASSERT_NO_FATAL_FAILURE(this->stall_mid_message_(buf, s));
+  auto third = make_message(1000, 200);
+
+  // Sliding the remainder down would move the bytes the outer write() points at
+  buf.set_draining(true);
+  EXPECT_FALSE(enqueue(buf, third));
+  EXPECT_EQ(buf.count(), 1);
+  EXPECT_EQ(buf.storage(), s.before);
+  buf.set_draining(false);
+
+  // Once the drain is over the same enqueue compacts and succeeds
+  ASSERT_TRUE(enqueue(buf, third));
+  EXPECT_EQ(buf.storage(), s.before);
+  append(s.received, this->drain_all_(buf));
+  expect_after_filler(s.received, s.filler, concat({s.first, s.second, third}));
+}
+
+TEST_F(OverflowBufferTest, CompactsInsteadOfGrowingAfterPartialDrain) {
+  TestOverflowBuffer buf;
+  Stall s;
+  ASSERT_NO_FATAL_FAILURE(this->stall_mid_message_(buf, s));
+  auto third = make_message(1000, 200);
+
+  // The sent first message is reclaimed by sliding the remainder down, not by reallocating
+  ASSERT_TRUE(enqueue(buf, third));
+  EXPECT_EQ(buf.storage(), s.before);
+
+  append(s.received, this->drain_all_(buf));
+  expect_after_filler(s.received, s.filler, concat({s.first, s.second, third}));
+}
+
+}  // namespace esphome::api::testing
+#endif  // USE_HOST