From 9309cf96b9710ad81d3e01a07a6f76637ba117a1 Mon Sep 17 00:00:00 2001 From: Jeroen Date: Thu, 17 Sep 2026 17:38:56 +0200 Subject: [PATCH] [web_server_idf] Close stalled EventSource sessions via HTTPD (#17800) Co-authored-by: Jeroen Jansen Co-authored-by: J. Nick Koston Co-authored-by: J. Nick Koston --- esphome/components/web_server/web_server.cpp | 8 +- .../web_server_idf/web_server_idf.cpp | 112 +++++++++++++++--- .../web_server_idf/web_server_idf.h | 20 +++- 3 files changed, 115 insertions(+), 25 deletions(-) diff --git a/esphome/components/web_server/web_server.cpp b/esphome/components/web_server/web_server.cpp index 8b0dfe166f..bad3d050f3 100644 --- a/esphome/components/web_server/web_server.cpp +++ b/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 diff --git a/esphome/components/web_server_idf/web_server_idf.cpp b/esphome/components/web_server_idf/web_server_idf.cpp index 9550570cdc..0062556063 100644 --- a/esphome/components/web_server_idf/web_server_idf.cpp +++ b/esphome/components/web_server_idf/web_server_idf.cpp @@ -6,6 +6,7 @@ #include #include +#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(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(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(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 diff --git a/esphome/components/web_server_idf/web_server_idf.h b/esphome/components/web_server_idf/web_server_idf.h index 743d296d73..7894206102 100644 --- a/esphome/components/web_server_idf/web_server_idf.h +++ b/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 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;