[web_server_idf] Close stalled EventSource sessions via HTTPD (#17800)

Co-authored-by: Jeroen Jansen <jeroen@Jeroens-MacBook-Air.local>
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>

esphome/components/web_server/web_server.cpp

@@ -200,10 +200,10 @@ void DeferredUpdateEventSource::process_deferred_queue_() {
       deferred_queue_.erase(deferred_queue_.begin());
       this->consecutive_send_failures_ = 0;  // Reset failure count on successful send
     } else {
-      // NOTE: Similar logic exists in web_server_idf/web_server_idf.cpp in AsyncEventSourceResponse::process_buffer_()
-      // The implementations differ due to platform-specific APIs (DISCARDED vs HTTPD_SOCK_ERR_TIMEOUT, close() vs
-      // fd_.store(0)), but the failure counting and timeout logic should be kept in sync. If you change this logic,
-      // also update the ESP-IDF implementation.
+      // NOTE: Similar logic exists in web_server_idf/web_server_idf.cpp in AsyncEventSourceResponse::process_buffer_().
+      // The close mechanisms are platform-specific (this path calls close() directly; the IDF path is time-based and
+      // closes through HTTPD to preserve session ownership), but both drop a client after roughly 20 seconds without
+      // send progress. Keep that stall policy in sync when changing either side.
       this->consecutive_send_failures_++;
       if (this->consecutive_send_failures_ >= MAX_CONSECUTIVE_SEND_FAILURES) {
         // Too many failures, connection is likely dead

esphome/components/web_server_idf/web_server_idf.cpp

@@ -6,6 +6,7 @@
 #include <cctype>
 #include <cinttypes>
 
+#include "esphome/core/application.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
@@ -728,7 +729,7 @@ bool AsyncEventSource::loop() {
   for (size_t i = 0; i < this->sessions_.size();) {
     auto *ses = this->sessions_[i];
     // If the session has a dead socket (marked by destroy callback)
-    if (ses->fd_.load() == 0) {
+    if (ses->safe_to_delete_()) {
       // destroy() already logged the close with the fd; don't double-log here.
       delete ses;  // NOLINT(cppcoreguidelines-owning-memory)
       // Remove by swapping with last element (O(1) removal, order doesn't matter for sessions)
@@ -751,7 +752,7 @@ void AsyncEventSource::adopt_pending_sessions_main_loop_() {
   }
   for (auto *rsp : incoming) {
     // Already disconnected? Drop it; skip on_connect_/session start on a dead session.
-    if (rsp->fd_.load() == 0) {
+    if (rsp->safe_to_delete_()) {
       delete rsp;  // NOLINT(cppcoreguidelines-owning-memory)
       continue;
     }
@@ -865,10 +866,16 @@ void AsyncEventSourceResponse::deq_push_back_with_dedup_(void *source, message_g
 }
 
 void AsyncEventSourceResponse::process_deferred_queue_() {
+  if (this->close_requested_) {
+    return;
+  }
   while (!deferred_queue_.empty()) {
     DeferredEvent &de = deferred_queue_.front();
     auto message = de.message_generator_(web_server_, de.source_);
     if (this->try_send_nodefer(message.c_str(), message.size(), "state")) {
+      if (this->close_requested_ || deferred_queue_.empty()) {
+        return;
+      }
       // O(n) but memory efficiency is more important than speed here which is why std::vector was chosen
       deferred_queue_.erase(deferred_queue_.begin());
     } else {
@@ -877,8 +884,64 @@ void AsyncEventSourceResponse::process_deferred_queue_() {
   }
 }
 
+void AsyncEventSourceResponse::request_close_() {
+  if (!this->close_requested_) {
+    this->close_requested_ = true;
+    this->deferred_queue_.clear();
+    this->event_buffer_.clear();
+    this->event_bytes_sent_ = 0;
+    this->next_close_attempt_ms_ = App.get_loop_component_start_time();
+  }
+
+  this->process_close_();
+}
+
+void AsyncEventSourceResponse::process_close_() {
+  if (!this->close_requested_ || this->close_work_queued_.load(std::memory_order_acquire)) {
+    return;
+  }
+  const int fd = this->fd_.load();
+  if (fd == 0) {
+    return;
+  }
+
+  const uint32_t now = App.get_loop_component_start_time();
+  if (static_cast<int32_t>(now - this->next_close_attempt_ms_) < 0) {
+    return;
+  }
+
+  // Queue an identity-checked shutdown on the HTTPD task. The public
+  // httpd_sess_trigger_close() queues only a reusable fd/session slot and can
+  // therefore close a new client if the original peer disconnects meanwhile.
+  this->close_work_queued_.store(true, std::memory_order_release);
+  const esp_err_t err = httpd_queue_work(this->hd_, &AsyncEventSourceResponse::close_session_work, this);
+  this->next_close_attempt_ms_ = now + (err == ESP_OK ? CLOSE_CONFIRM_INTERVAL_MS : CLOSE_RETRY_INTERVAL_MS);
+  if (err == ESP_OK) {
+    return;
+  }
+
+  this->close_work_queued_.store(false, std::memory_order_release);
+  if (!this->close_retry_warning_logged_) {
+    ESP_LOGW(TAG, "Failed to queue EventSource close (%s); retrying", esp_err_to_name(err));
+    this->close_retry_warning_logged_ = true;
+  }
+}
+
+void AsyncEventSourceResponse::close_session_work(void *arg) {
+  auto *response = static_cast<AsyncEventSourceResponse *>(arg);
+  const int fd = response->fd_.load();
+  if (fd != 0 && httpd_sess_get_ctx(response->hd_, fd) == response) {
+    // The HTTPD task remains the session owner. Shutting the socket down makes
+    // its next select/recv path delete the session and invoke destroy().
+    shutdown(fd, SHUT_RDWR);
+  }
+
+  // Release self only after the HTTPD-task callback has finished every access.
+  response->close_work_queued_.store(false, std::memory_order_release);
+}
+
 void AsyncEventSourceResponse::process_buffer_() {
-  if (event_buffer_.empty()) {
+  if (this->close_requested_ || event_buffer_.empty()) {
     return;
   }
   if (event_bytes_sent_ == event_buffer_.size()) {
@@ -892,32 +955,33 @@ void AsyncEventSourceResponse::process_buffer_() {
       httpd_socket_send(this->hd_, this->fd_.load(), event_buffer_.c_str() + event_bytes_sent_, remaining, 0);
   if (bytes_sent == HTTPD_SOCK_ERR_TIMEOUT) {
     // EAGAIN/EWOULDBLOCK - socket buffer full, try again later
-    // NOTE: Similar logic exists in web_server/web_server.cpp in DeferredUpdateEventSource::process_deferred_queue_()
-    // The implementations differ due to platform-specific APIs (HTTPD_SOCK_ERR_TIMEOUT vs DISCARDED, fd_.store(0) vs
-    // close()), but the failure counting and timeout logic should be kept in sync. If you change this logic, also
-    // update the Arduino implementation.
-    this->consecutive_send_failures_++;
-    if (this->consecutive_send_failures_ >= MAX_CONSECUTIVE_SEND_FAILURES) {
-      // Too many failures, connection is likely dead
-      ESP_LOGW(TAG, "Closing stuck EventSource connection after %" PRIu16 " failed sends",
-               this->consecutive_send_failures_);
-      this->fd_.store(0);  // Mark for cleanup
-      this->deferred_queue_.clear();
+    // NOTE: Similar logic exists in web_server/web_server.cpp in DeferredUpdateEventSource::process_deferred_queue_().
+    // The IDF path is intentionally time-based and closes through HTTPD to preserve session ownership.
+    const uint32_t now = App.get_loop_component_start_time();
+    if (this->send_failure_started_ms_ == 0) {
+      this->send_failure_started_ms_ = now != 0 ? now : 1;  // Reserve zero for no stall.
+    }
+    if (static_cast<int32_t>(now - (this->send_failure_started_ms_ + SEND_STALL_TIMEOUT_MS)) >= 0) {
+      ESP_LOGW(TAG, "Closing stuck EventSource connection after %" PRIu32 " ms without send progress",
+               now - this->send_failure_started_ms_);
+      this->request_close_();
     }
     return;
   }
   if (bytes_sent == HTTPD_SOCK_ERR_FAIL) {
-    // Real socket error - connection will be closed by httpd and destroy callback will be called
+    // Low-level asynchronous sends do not make HTTPD close the session automatically.
+    this->request_close_();
     return;
   }
   if (bytes_sent <= 0) {
     // Unexpected error or zero bytes sent
     ESP_LOGW(TAG, "Unexpected send result: %d", bytes_sent);
+    this->request_close_();
     return;
   }
 
-  // Successful send - reset failure counter
-  this->consecutive_send_failures_ = 0;
+  // Successful send - reset stall tracking
+  this->send_failure_started_ms_ = 0;
   event_bytes_sent_ += bytes_sent;
 
   // Log partial sends for debugging
@@ -933,20 +997,26 @@ void AsyncEventSourceResponse::process_buffer_() {
 }
 
 void AsyncEventSourceResponse::loop() {
+  if (this->close_requested_) {
+    this->process_close_();
+    return;
+  }
   process_buffer_();
   process_deferred_queue_();
+  if (this->close_requested_)
+    return;
   // One step per loop; refusals retry next pass
   this->entities_iterator_.try_advance(1);
 }
 
 bool AsyncEventSourceResponse::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
                                                 uint32_t reconnect) {
-  if (this->fd_.load() == 0) {
+  if (this->fd_.load() == 0 || this->close_requested_) {
     return false;
   }
 
   process_buffer_();
-  if (!event_buffer_.empty()) {
+  if (this->close_requested_ || !event_buffer_.empty()) {
     // there is still pending event data to send first
     return false;
   }
@@ -1098,6 +1168,10 @@ void AsyncEventSourceResponse::deferrable_send_state(void *source, const char *e
   process_buffer_();
   process_deferred_queue_();
 
+  if (this->close_requested_) {
+    return;
+  }
+
   if (!event_buffer_.empty() || !deferred_queue_.empty()) {
     // outgoing event buffer or deferred queue still not empty which means downstream tcp send buffer full, no point
     // trying to send first

esphome/components/web_server_idf/web_server_idf.h

@@ -301,6 +301,14 @@ class AsyncEventSourceResponse {
   void deq_push_back_with_dedup_(void *source, message_generator_t *message_generator);
   void process_deferred_queue_();
   void process_buffer_();
+  void request_close_();
+  void process_close_();
+  static void close_session_work(void *arg);
+
+  // Deletable only after destroy() zeroed fd_ and no queued HTTPD close work still references this object.
+  bool safe_to_delete_() const {
+    return this->fd_.load() == 0 && !this->close_work_queued_.load(std::memory_order_acquire);
+  }
 
   static void destroy(void *p);
   AsyncEventSource *server_;
@@ -311,8 +319,16 @@ class AsyncEventSourceResponse {
   esphome::web_server::ListEntitiesIterator entities_iterator_;
   std::string event_buffer_;
   size_t event_bytes_sent_;
-  uint16_t consecutive_send_failures_{0};
-  static constexpr uint16_t MAX_CONSECUTIVE_SEND_FAILURES = 2500;  // ~20 seconds at 125Hz loop rate
+  uint32_t send_failure_started_ms_{0};  // Zero means no send stall in progress.
+  uint32_t next_close_attempt_ms_{0};
+  // Main-loop only; the HTTPD task never reads or writes this flag.
+  bool close_requested_{false};
+  bool close_retry_warning_logged_{false};
+  // Set on the main loop before queueing close work, cleared by the HTTPD-task callback when done.
+  std::atomic<bool> close_work_queued_{false};
+  static constexpr uint32_t SEND_STALL_TIMEOUT_MS = 20000;
+  static constexpr uint32_t CLOSE_RETRY_INTERVAL_MS = 250;
+  static constexpr uint32_t CLOSE_CONFIRM_INTERVAL_MS = 1000;
 };
 
 using AsyncEventSourceClient = AsyncEventSourceResponse;