From a7709115ccf32252a5a68f250f03a1a9a426734d Mon Sep 17 00:00:00 2001 From: eigger Date: Mon, 26 Oct 2026 13:13:56 +0900 Subject: [PATCH 1/3] feat(ws_bridge): add camera snapshot uploads --- components/ws_bridge/README.md | 47 +++ components/ws_bridge/__init__.py | 81 ++++ components/ws_bridge/const.py | 3 + components/ws_bridge/event/__init__.py | 10 +- components/ws_bridge/ws_bridge.cpp | 12 + components/ws_bridge/ws_bridge.h | 13 + components/ws_bridge/ws_bridge_snapshot.cpp | 351 ++++++++++++++++++ components/ws_bridge/ws_bridge_snapshot.h | 111 ++++++ components/ws_bridge/ws_protocol.cpp | 25 ++ components/ws_bridge/ws_protocol.h | 4 + .../ws_bridge/test.snapshot.esp32-idf.yaml | 55 +++ .../ws_bridge/test.ssl-false.esp32-idf.yaml | 24 ++ 12 files changed, 731 insertions(+), 5 deletions(-) create mode 100644 components/ws_bridge/ws_bridge_snapshot.cpp create mode 100644 components/ws_bridge/ws_bridge_snapshot.h create mode 100644 tests/components/ws_bridge/test.snapshot.esp32-idf.yaml create mode 100644 tests/components/ws_bridge/test.ssl-false.esp32-idf.yaml diff --git a/components/ws_bridge/README.md b/components/ws_bridge/README.md index b3fd297a..d1318ffc 100644 --- a/components/ws_bridge/README.md +++ b/components/ws_bridge/README.md @@ -222,6 +222,7 @@ update: | `reannounce_interval` | | `60s` | How often to resend `ws_bridge/connect` + all entity/state declarations while nominally connected. Guards against the HA-side integration losing track of this gateway (e.g. its config entry reloaded) while the raw connection and ping/pong stay healthy — that scenario is otherwise invisible, since HA answers pings regardless of our integration's state. If a re-announce goes unanswered, forces a full reconnect rather than repeating the same no-op | | `trackers` | | - | List of GPS `device_tracker` entities — see [GPS device trackers](#gps-device-trackers) below | | `entities` | | - | Existing ESPHome entities to expose without a parallel `platform: ws_bridge` entity — see [Exposing existing entities](#exposing-existing-entities) | +| `snapshots` | | - | Optional ESP32 camera snapshots uploaded as HA `camera` or `image` entities — see [Camera snapshots](#camera-snapshots) | "Templatable" fields accept a plain string or a `!lambda return std::string(...);` — the lambda is re-evaluated on every `ws_bridge/connect` (the initial connect and each `reannounce_interval`), not just once at boot. @@ -270,6 +271,52 @@ left out. `datetime_id:` must point at an entity of the same `type:`. strings (`2026-08-13`, `21:30:00`, `2026-08-13T21:30:00`); Home Assistant attaches its own local timezone to them rather than reading them as UTC. +## Camera snapshots + +`snapshots:` exposes JPEG frames from an `esp32_camera` as either a Home +Assistant `camera` (the default) or `image`. It requires **ESPHome 2026.10+ +and ESP-IDF** because it uses ESPHome's generic `camera::Camera` listener API. +This requirement applies only to configurations that use `snapshots:`; normal +`ws_bridge` entity configurations keep their existing compatibility. + +```yaml +esp32_camera: + id: front_sensor + # ...normal esp32_camera pins and settings... + +ws_bridge: + host: !secret ha_host + token: !secret ha_token + snapshots: + - camera_id: front_sensor + unique_id: front_camera + name: "Front camera" + update_interval: 30s # request a fresh JPEG periodically + - camera_id: front_sensor + unique_id: front_thumbnail + name: "Front thumbnail" + platform: image # default is camera + update_interval: 2min +``` + +Each list item has a separate, ESP-initiated media WebSocket because Home +Assistant binds one media connection to exactly one declared entity. The +firmware first uses the existing authenticated control connection to request +`ws_bridge/media_open`, then authenticates a second standard `/api/websocket` +connection with the same token and sends `ws_bridge/media_connect`. Its JPEG +is one logical binary WebSocket message: the assigned handler byte immediately +followed by the JPEG bytes. This remains compatible with Home Assistant remote +access/reverse proxies: Home Assistant never connects back to the ESP device. + +The worker retains a camera frame until its complete send finishes and keeps at +most one pending/latest frame per configured snapshot. A failed fragment closes +that media connection and obtains a fresh one-use session rather than trying to +continue an incomplete binary message. Frames larger than 1 MiB are discarded. +The media socket is recreated after a real control reconnect; routine control +re-announces leave an established media socket alone. See the HA-side +[wire protocol](https://github.com/eigger/hass-ws-bridge/blob/main/docs/PROTOCOL.md#321-authenticated-jpeg-upload-ws_bridgemedia_open--ws_bridgemedia_connect) +for authentication and frame limits. + ## Exposing existing entities Two ways to put an already-declared ESPHome entity onto Home Assistant. diff --git a/components/ws_bridge/__init__.py b/components/ws_bridge/__init__.py index 43f164a7..02d1c9fc 100644 --- a/components/ws_bridge/__init__.py +++ b/components/ws_bridge/__init__.py @@ -59,6 +59,9 @@ CONF_REANNOUNCE_INTERVAL, CONF_ENTITIES, CONF_SOURCE_ID, + CONF_SNAPSHOTS, + CONF_CAMERA_ID, + CONF_PLATFORM, ) CODEOWNERS = ["@eigger"] @@ -70,6 +73,10 @@ ws_bridge_ns = cg.esphome_ns.namespace("ws_bridge") WsBridgeComponent = ws_bridge_ns.class_("WsBridgeComponent", cg.Component) WsBridgeDevice = ws_bridge_ns.class_("WsBridgeDevice") +# Camera is deliberately not imported here. A normal ws_bridge configuration +# must not auto-load the camera integration; snapshots import/check it only +# inside to_code when their optional list is actually present. +WsBridgeSnapshot = ws_bridge_ns.class_("WsBridgeSnapshot", cg.PollingComponent) ConnectedTrigger = ws_bridge_ns.class_("ConnectedTrigger", automation.Trigger.template()) DisconnectedTrigger = ws_bridge_ns.class_("DisconnectedTrigger", automation.Trigger.template()) @@ -145,6 +152,20 @@ } ) +SNAPSHOT_SCHEMA = cv.Schema( + { + cv.GenerateID(): cv.declare_id(WsBridgeSnapshot), + # Kept loosely typed until to_code so importing this component alone + # does not auto-load camera. The conditional check there still + # rejects non-camera ids with a useful message. + cv.Required(CONF_CAMERA_ID): cv.use_id(cg.Component), + cv.Required(CONF_UNIQUE_ID): cv.string_strict, + cv.Required(CONF_NAME): cv.string_strict, + # HA's image and camera entities share the same JPEG upload transport. + cv.Optional(CONF_PLATFORM, default="camera"): cv.one_of("camera", "image", lower=True), + } +).extend(cv.polling_component_schema("60s")) + async def to_code_entity_ref(hub_var, conf): source_id = conf[CONF_SOURCE_ID] @@ -185,6 +206,14 @@ def _validate_esp_idf(config): return config +def _validate_snapshot_version(config): + # Generic camera::CameraListener landed in ESPHome 2026.10. Keep this + # opt-in check separate from the hub's long-standing configuration. + if config.get(CONF_SNAPSHOTS): + cv.require_esphome_version(2026, 10, 0)(config) + return config + + CONFIG_SCHEMA = cv.All( cv.Schema( { @@ -261,9 +290,14 @@ def _validate_esp_idf(config): # `platform: ws_bridge` entity per value — see ENTITY_SCHEMA above # and the README's "Exposing existing entities" section. cv.Optional(CONF_ENTITIES, default=[]): cv.ensure_list(ENTITY_SCHEMA), + # Optional snapshot media. This is intentionally not AUTO_LOADed: + # only configs using it need a camera platform and ESPHome 2026.10 + # (the generic camera::Camera listener API used by this feature). + cv.Optional(CONF_SNAPSHOTS, default=[]): cv.ensure_list(SNAPSHOT_SCHEMA), } ).extend(cv.COMPONENT_SCHEMA), _validate_esp_idf, + _validate_snapshot_version, ) # Shared schema every ws_bridge platform must extend, mirroring uartex's @@ -291,6 +325,25 @@ async def register_ws_bridge_device(var, config): async def to_code(config): esp32.add_idf_component(name="espressif/esp_websocket_client", ref=ESP_WEBSOCKET_CLIENT_VERSION) + # The managed WebSocket client links ESP-TLS even for ws:// builds, so it + # needs ESPHome's trimmed mbedTLS client symbols in every configuration. + if hasattr(esp32, "request_tls"): + esp32.request_tls() + else: + # ESPHome before 2026.10 has no request_tls helper. + esp32.include_builtin_idf_component("esp-tls") + esp32.add_idf_sdkconfig_option("CONFIG_MBEDTLS_TLS_ENABLED", True) + if config[CONF_SSL]: + # esp_websocket_client itself does not request ESPHome's trimmed TLS + # pieces. Explicitly retain the TLS stack and public CA bundle used by + # esp_crt_bundle_attach; otherwise a minimal ws_bridge firmware cannot + # even compile its TLS configuration. + cg.add_define("USE_WS_BRIDGE_CERT_BUNDLE") + if hasattr(esp32, "require_certificate_bundle"): + esp32.require_certificate_bundle() + else: + # Older ESPHome's full-bundle request also enables the bundle. + esp32.require_full_certificate_bundle() var = cg.new_Pvariable(config[CONF_ID]) await cg.register_component(var, config) @@ -356,3 +409,31 @@ async def to_code(config): for conf in config.get(CONF_ENTITIES, []): await to_code_entity_ref(var, conf) + + snapshots = config.get(CONF_SNAPSHOTS, []) + if snapshots: + # Keep ws_bridge's existing minimum unchanged; snapshot users require + # the newer generic camera listener API (documented in README). + # ESPHome's generic C++ camera::Camera has no Python MockObjClass; + # esp32_camera is the currently supported source of that interface. + # Import it only here so ordinary ws_bridge configurations stay free + # of a camera dependency. + from esphome.components import esp32_camera + + cg.add_define("USE_WS_BRIDGE_SNAPSHOT") + for conf in snapshots: + camera_full_id, camera_var = await cg.get_variable_with_full_id(conf[CONF_CAMERA_ID]) + if not isinstance(camera_full_id.type, cg.MockObjClass) or not camera_full_id.type.inherits_from( + esp32_camera.ESP32Camera + ): + raise cv.Invalid( + f"'{conf[CONF_CAMERA_ID].id}' is not an esp32_camera id; " + "snapshot currently supports esp32_camera" + ) + snapshot = cg.new_Pvariable(conf[CONF_ID]) + await cg.register_component(snapshot, conf) + cg.add(snapshot.set_parent(var)) + cg.add(snapshot.set_camera(camera_var)) + cg.add(snapshot.set_unique_id(conf[CONF_UNIQUE_ID])) + cg.add(snapshot.set_name(conf[CONF_NAME])) + cg.add(snapshot.set_platform(conf[CONF_PLATFORM])) diff --git a/components/ws_bridge/const.py b/components/ws_bridge/const.py index 29b1b9e2..851ca4b4 100644 --- a/components/ws_bridge/const.py +++ b/components/ws_bridge/const.py @@ -10,6 +10,9 @@ CONF_HW_VERSION = "hw_version" CONF_KEEP_LAST_STATE_ON_DISCONNECT = "keep_last_state_on_disconnect" CONF_SYNC_ENTITIES = "sync_entities" +CONF_SNAPSHOTS = "snapshots" +CONF_CAMERA_ID = "camera_id" +CONF_PLATFORM = "platform" CONF_UNIQUE_ID = "unique_id" # Prefixed with ws_ to avoid colliding with ESPHome's own reserved diff --git a/components/ws_bridge/event/__init__.py b/components/ws_bridge/event/__init__.py index ba3d4b92..b2e4e26a 100644 --- a/components/ws_bridge/event/__init__.py +++ b/components/ws_bridge/event/__init__.py @@ -10,11 +10,11 @@ WsBridgeEvent = ws_bridge_ns.class_("WsBridgeEvent", event.Event, cg.Component, ws_bridge.WsBridgeDevice) # An empty Python list renders as a bare `{}`, which is ambiguous across -# Event::set_event_types()'s initializer_list / FixedVector / vector overloads -# (including the =delete'd std::string ones that exist to catch mistakes). Name -# the type so the vector overload is an exact match. The wrapper's real list is -# copied off the source in setup() — see WsBridgeEvent::setup(). -_NO_EVENT_TYPES = cg.RawExpression("std::vector{}") +# Event::set_event_types() has a FixedVector overload in both supported older +# ESPHome and 2026.10+, while its std::vector overload was removed in 2026.10. +# Explicitly select the durable overload for an event_id: wrapper whose real +# list is copied from its source during setup(). +_NO_EVENT_TYPES = cg.RawExpression("esphome::FixedVector{}") def validate_event_types(config): diff --git a/components/ws_bridge/ws_bridge.cpp b/components/ws_bridge/ws_bridge.cpp index b0d8afad..995f1d71 100644 --- a/components/ws_bridge/ws_bridge.cpp +++ b/components/ws_bridge/ws_bridge.cpp @@ -1,6 +1,8 @@ #include "ws_bridge.h" #include +#ifdef USE_WS_BRIDGE_CERT_BUNDLE #include "esp_crt_bundle.h" +#endif #include "esp_transport_ws.h" #include "esphome/components/network/util.h" #include "esphome/core/application.h" @@ -23,7 +25,9 @@ void WsBridgeComponent::setup() { config.path = "/api/websocket"; config.transport = this->ssl_ ? WEBSOCKET_TRANSPORT_OVER_SSL : WEBSOCKET_TRANSPORT_OVER_TCP; if (this->ssl_) { +#ifdef USE_WS_BRIDGE_CERT_BUNDLE config.crt_bundle_attach = esp_crt_bundle_attach; +#endif } config.disable_auto_reconnect = false; config.reconnect_timeout_ms = 10000; @@ -501,6 +505,14 @@ void WsBridgeComponent::handle_message_(const std::string &raw) { this->awaiting_connect_result_ = false; } } + this->message_cb_.call(msg); +} + +uint32_t WsBridgeComponent::open_media_session(const std::string &unique_id) { + if (!this->is_connected()) + return 0; + uint32_t id = this->next_id_(); + return this->send_raw_(build_media_open(id, unique_id)) ? id : 0; } void WsBridgeComponent::route_command_(const WsCommand &command) { diff --git a/components/ws_bridge/ws_bridge.h b/components/ws_bridge/ws_bridge.h index e98897e1..05becbe4 100644 --- a/components/ws_bridge/ws_bridge.h +++ b/components/ws_bridge/ws_bridge.h @@ -77,6 +77,10 @@ class WsBridgeComponent : public Component { // declarations must hook this rather than on_connected, so that a // re-announce which heals a lost HA-side registration re-declares them too. void add_on_declare_callback(std::function &&cb) { this->declare_cb_.add(std::move(cb)); } + // Receives complete, parsed control-channel messages on the ESPHome loop. + // Optional features (such as snapshot media) use this for their own result + // IDs without handling messages in the websocket client task. + void add_message_callback(std::function &&cb) { this->message_cb_.add(std::move(cb)); } void setup() override; void loop() override; @@ -89,6 +93,14 @@ class WsBridgeComponent : public Component { // atomic. See ws_event_handler_ for why the disconnect-family transitions // in particular can't wait for the main loop to process a queued event. bool is_connected() const { return this->state_.load(std::memory_order_acquire) == WS_BRIDGE_CONNECTED; } + const std::string &get_host() const { return this->host_; } + uint16_t get_port() const { return this->port_; } + bool get_ssl() const { return this->ssl_; } + const std::string &get_token() const { return this->token_; } + + // Opens a one-use, HA-authenticated media session for an already declared + // entity. The caller correlates the returned control-channel message ID. + uint32_t open_media_session(const std::string &unique_id); // Called by platform entities (via WsBridgeDevice helpers) to push state // and declarations. No-ops while not connected; the next (re)connect will @@ -314,6 +326,7 @@ class WsBridgeComponent : public Component { CallbackManager connected_cb_{}; CallbackManager disconnected_cb_{}; CallbackManager declare_cb_{}; + CallbackManager message_cb_{}; }; } // namespace ws_bridge diff --git a/components/ws_bridge/ws_bridge_snapshot.cpp b/components/ws_bridge/ws_bridge_snapshot.cpp new file mode 100644 index 00000000..c0372a0a --- /dev/null +++ b/components/ws_bridge/ws_bridge_snapshot.cpp @@ -0,0 +1,351 @@ +#include "ws_bridge_snapshot.h" + +#ifdef USE_WS_BRIDGE_SNAPSHOT + +#ifdef USE_WS_BRIDGE_CERT_BUNDLE +#include "esp_crt_bundle.h" +#endif +#include "esphome/core/log.h" +#include "freertos/task.h" +#include "ws_bridge.h" + +namespace esphome::ws_bridge { + +static const char *const TAG = "ws_bridge.snapshot"; +static constexpr UBaseType_t WORK_QUEUE_LENGTH = 12; + +void WsBridgeSnapshot::setup() { + if (parent_ == nullptr || camera_ == nullptr) { + ESP_LOGE(TAG, "Snapshot requires ws_bridge and camera"); + mark_failed(); + return; + } + worker_queue_ = xQueueCreate(WORK_QUEUE_LENGTH, sizeof(WorkItem *)); + if (worker_queue_ == nullptr || + xTaskCreate(media_task_, "ws_bridge_media", 8192, this, tskIDLE_PRIORITY + 1, &worker_task_) != pdPASS) { + ESP_LOGE(TAG, "Unable to start media worker"); + mark_failed(); + return; + } + camera_->add_listener(this); + parent_->add_on_declare_callback([this]() { + parent_->send_entity_declare(unique_id_, platform_, name_, "", "", nullptr); + }); + parent_->add_on_connected_callback([this]() { on_control_connected_(); }); + parent_->add_on_disconnected_callback([this]() { on_control_disconnected_(); }); + parent_->add_message_callback([this](const ParsedMessage &message) { on_control_message_(message); }); +} + +void WsBridgeSnapshot::dump_config() { + ESP_LOGCONFIG(TAG, "WS Bridge snapshot:"); + ESP_LOGCONFIG(TAG, " Entity: %s (%s)", unique_id_.c_str(), platform_.c_str()); + ESP_LOGCONFIG(TAG, " Capture interval: %u ms", static_cast(get_update_interval())); +} + +void WsBridgeSnapshot::loop() { + if (needs_media_open_.exchange(false, std::memory_order_acq_rel)) + open_media_(); + if (media_open_id_ != 0 && millis() - media_open_sent_ms_ >= MEDIA_OPEN_TIMEOUT_MS) { + media_open_id_ = 0; + open_media_(); + } +} + +void WsBridgeSnapshot::update() { request_frame_(); } + +void WsBridgeSnapshot::on_control_connected_() { + media_open_id_ = 0; + media_ready_.store(false, std::memory_order_release); + handler_id_.store(0, std::memory_order_release); + enqueue_(new WorkItem{WorkType::STOP}); + open_media_(); +} + +void WsBridgeSnapshot::on_control_disconnected_() { + media_open_id_ = 0; + media_ready_.store(false, std::memory_order_release); + handler_id_.store(0, std::memory_order_release); + { + std::lock_guard lock(frame_mutex_); + capture_requested_ = false; + pending_image_.reset(); + } + enqueue_(new WorkItem{WorkType::STOP}); +} + +void WsBridgeSnapshot::open_media_() { + if (parent_ == nullptr || !parent_->is_connected() || media_open_id_ != 0) + return; + media_open_id_ = parent_->open_media_session(unique_id_); + if (media_open_id_ != 0) + media_open_sent_ms_ = millis(); +} + +void WsBridgeSnapshot::on_control_message_(const ParsedMessage &message) { + if (media_open_id_ == 0 || message.type != "result" || message.id != media_open_id_) + return; + media_open_id_ = 0; + if (!message.success || message.session_id.empty()) { + needs_media_open_.store(true, std::memory_order_release); + return; + } + auto *item = new WorkItem{WorkType::START}; + item->session_id = message.session_id; // opaque one-use session; never logged + enqueue_(item); +} + +void WsBridgeSnapshot::request_frame_() { + if (!media_ready_.load(std::memory_order_acquire) || camera_ == nullptr) + return; + std::lock_guard lock(frame_mutex_); + if (capture_requested_) + return; + capture_requested_ = true; + camera_->request_image(camera::IDLE); +} + +void WsBridgeSnapshot::on_camera_image(const std::shared_ptr &image) { + if (image == nullptr || !image->was_requested_by(camera::IDLE)) + return; // do not steal webserver/API camera requests + { + std::lock_guard lock(frame_mutex_); + if (!capture_requested_) + return; + capture_requested_ = false; + if (image->get_data_length() == 0 || image->get_data_length() > MAX_JPEG_BYTES) + return; + // One latest-frame slot bounds RAM and retains the camera buffer until TX. + pending_image_ = image; + } + enqueue_(new WorkItem{WorkType::WAKE}); +} + +void WsBridgeSnapshot::enqueue_(WorkItem *item) { + if (item == nullptr) + return; + if (worker_queue_ == nullptr || xQueueSend(worker_queue_, &item, 0) != pdPASS) { + ESP_LOGW(TAG, "Media worker queue full; dropping work item"); + delete item; + } +} + +void WsBridgeSnapshot::media_task_(void *arg) { static_cast(arg)->worker_loop_(); } + +void WsBridgeSnapshot::worker_loop_() { + while (true) { + WorkItem *item = nullptr; + if (xQueueReceive(worker_queue_, &item, pdMS_TO_TICKS(100)) == pdPASS && item != nullptr) { + process_item_(item); + delete item; + } + send_pending_(); + } +} + +void WsBridgeSnapshot::process_item_(WorkItem *item) { + switch (item->type) { + case WorkType::START: + start_media_(item->session_id); + break; + case WorkType::STOP: + stop_media_(); + break; + case WorkType::WAKE: + break; + case WorkType::EVENT: + handle_media_event_(item); + break; + } +} + +void WsBridgeSnapshot::start_media_(const std::string &session_id) { + stop_media_(); + if (parent_ == nullptr || session_id.empty()) { + needs_media_open_.store(true, std::memory_order_release); + return; + } + worker_session_id_ = session_id; + esp_websocket_client_config_t config = {}; + config.host = parent_->get_host().c_str(); + config.port = parent_->get_port(); + config.path = "/api/websocket"; + config.transport = parent_->get_ssl() ? WEBSOCKET_TRANSPORT_OVER_SSL : WEBSOCKET_TRANSPORT_OVER_TCP; + config.disable_auto_reconnect = true; + config.network_timeout_ms = 10000; + config.buffer_size = 1024; + config.task_stack = parent_->get_ssl() ? 8192 : 4096; + if (parent_->get_ssl()) { +#ifdef USE_WS_BRIDGE_CERT_BUNDLE + config.crt_bundle_attach = esp_crt_bundle_attach; +#endif + } + media_ = esp_websocket_client_init(&config); + active_generation_ = ++media_generation_; + callback_context_ = std::make_unique(CallbackContext{this, active_generation_}); + if (media_ == nullptr || + esp_websocket_register_events(media_, WEBSOCKET_EVENT_ANY, media_event_, callback_context_.get()) != ESP_OK || + esp_websocket_client_start(media_) != ESP_OK) { + ESP_LOGW(TAG, "Failed to start media WebSocket"); + stop_media_(); + needs_media_open_.store(true, std::memory_order_release); + } +} + +void WsBridgeSnapshot::stop_media_() { + media_ready_.store(false, std::memory_order_release); + handler_id_.store(0, std::memory_order_release); + worker_session_id_.clear(); + media_connect_id_ = 0; + rx_accum_.clear(); + rx_text_frame_ = false; + rx_drop_message_ = false; + if (media_ != nullptr) { + stopping_.store(true, std::memory_order_release); + esp_websocket_unregister_events(media_, WEBSOCKET_EVENT_ANY, media_event_); + esp_websocket_client_stop(media_); + esp_websocket_client_destroy(media_); + media_ = nullptr; + stopping_.store(false, std::memory_order_release); + } + active_generation_ = 0; + // Work items have already copied the generation. Retain a few detached + // callback contexts while stop/unregister completes, then bound reconnect + // churn. They are never used to decide ownership after active_generation_ + // has changed. + if (callback_context_ != nullptr) { + retired_callback_contexts_.push_back(std::move(callback_context_)); + if (retired_callback_contexts_.size() > 4) + retired_callback_contexts_.pop_front(); + } +} + +void WsBridgeSnapshot::transport_failed_() { + media_ready_.store(false, std::memory_order_release); + handler_id_.store(0, std::memory_order_release); + // A failed partial frame cannot be followed by another binary message: the + // peer may still await its FIN. Stop/destroy on the owner worker and obtain + // a fresh one-use HA session through the control channel. + stop_media_(); + needs_media_open_.store(true, std::memory_order_release); +} + +void WsBridgeSnapshot::media_event_(void *arg, esp_event_base_t, int32_t event_id, void *event_data) { + auto *context = static_cast(arg); + if (context == nullptr || context->snapshot == nullptr) + return; + auto *self = context->snapshot; + auto *data = static_cast(event_data); + auto *item = new WorkItem{WorkType::EVENT}; + item->client = data == nullptr ? nullptr : data->client; + item->event_id = event_id; + item->generation = context->generation; + if (event_id == WEBSOCKET_EVENT_DATA && data != nullptr) { + item->opcode = data->op_code; + item->payload_len = data->payload_len; + item->payload_offset = data->payload_offset; + item->fin = data->fin; + if (data->data_ptr != nullptr && data->data_len > 0) + item->text.assign(data->data_ptr, data->data_len); + } + self->enqueue_(item); +} + +void WsBridgeSnapshot::handle_media_event_(WorkItem *item) { + if (item->generation != active_generation_ || item->client != media_) + return; // stale queued event, including an address-reused client handle + if (item->event_id == WEBSOCKET_EVENT_DISCONNECTED || item->event_id == WEBSOCKET_EVENT_ERROR || + item->event_id == WEBSOCKET_EVENT_CLOSED) { + if (!stopping_.load(std::memory_order_acquire)) + transport_failed_(); + return; + } + if (item->event_id != WEBSOCKET_EVENT_DATA) + return; + uint8_t opcode = item->opcode & 0x0F; + if (opcode != WS_TRANSPORT_OPCODES_TEXT && opcode != WS_TRANSPORT_OPCODES_CONT) + return; + // payload_len/offset describe buffer chunks within one WS frame; fin is the + // separate RFC 6455 signal that the logical TEXT+CONT message is complete. + bool frame_complete = + item->payload_len == 0 || item->payload_offset + static_cast(item->text.size()) >= item->payload_len; + if (rx_drop_message_) { + if (frame_complete && item->fin) { + rx_drop_message_ = false; + rx_text_frame_ = false; + } + return; + } + if (opcode == WS_TRANSPORT_OPCODES_TEXT) + rx_text_frame_ = true; + else if (!rx_text_frame_) + return; + if (rx_accum_.size() + item->text.size() > RX_ACCUM_MAX) { + ESP_LOGW(TAG, "Media response exceeded %u bytes; dropping", static_cast(RX_ACCUM_MAX)); + rx_accum_.clear(); + rx_text_frame_ = false; + rx_drop_message_ = !(frame_complete && item->fin); + return; + } + rx_accum_.append(item->text); + if (!frame_complete || !item->fin) + return; + rx_text_frame_ = false; + if (!rx_accum_.empty()) { + std::string message = std::move(rx_accum_); + rx_accum_.clear(); + handle_media_message_(message); + } +} + +void WsBridgeSnapshot::handle_media_message_(const std::string &raw) { + ParsedMessage message = parse_message(raw); + if (message.type == "auth_required") { + std::string auth = build_auth(parent_->get_token()); + if (media_ == nullptr || esp_websocket_client_send_text(media_, auth.c_str(), auth.size(), pdMS_TO_TICKS(10000)) < 0) + transport_failed_(); + } else if (message.type == "auth_ok") { + media_connect_id_ = 1; + std::string connect = build_media_connect(media_connect_id_, worker_session_id_); + if (media_ == nullptr || esp_websocket_client_send_text(media_, connect.c_str(), connect.size(), pdMS_TO_TICKS(10000)) < 0) + transport_failed_(); + } else if (message.type == "auth_invalid") { + ESP_LOGE(TAG, "Home Assistant rejected the media connection token"); + transport_failed_(); + } else if (message.type == "result" && message.id == media_connect_id_ && message.success && message.handler_id != 0) { + handler_id_.store(message.handler_id, std::memory_order_release); + media_ready_.store(true, std::memory_order_release); + } else if (message.type == "result" && message.id == media_connect_id_) { + ESP_LOGW(TAG, "Home Assistant rejected media session"); + transport_failed_(); + } +} + +void WsBridgeSnapshot::send_pending_() { + if (!media_ready_.load(std::memory_order_acquire) || media_ == nullptr || !esp_websocket_client_is_connected(media_)) + return; + std::shared_ptr image; + { + std::lock_guard lock(frame_mutex_); + image = std::move(pending_image_); + } + if (image == nullptr) + return; + uint8_t handler_id = handler_id_.load(std::memory_order_acquire); + const char prefix = static_cast(handler_id); + int first = esp_websocket_client_send_bin_partial(media_, &prefix, 1, pdMS_TO_TICKS(10000)); + int body = first == 1 + ? esp_websocket_client_send_cont_msg(media_, reinterpret_cast(image->get_data_buffer()), + static_cast(image->get_data_length()), pdMS_TO_TICKS(10000)) + : -1; + // send_fin() returns a status/byte count for the empty FIN frame, not the + // JPEG length. A non-negative result is the documented success condition. + int finish = body == static_cast(image->get_data_length()) ? esp_websocket_client_send_fin(media_, pdMS_TO_TICKS(10000)) : -1; + if (first != 1 || body != static_cast(image->get_data_length()) || finish < 0) { + ESP_LOGW(TAG, "JPEG upload failed; reopening media connection"); + transport_failed_(); + } +} + +} // namespace esphome::ws_bridge + +#endif // USE_WS_BRIDGE_SNAPSHOT diff --git a/components/ws_bridge/ws_bridge_snapshot.h b/components/ws_bridge/ws_bridge_snapshot.h new file mode 100644 index 00000000..58bddc3c --- /dev/null +++ b/components/ws_bridge/ws_bridge_snapshot.h @@ -0,0 +1,111 @@ +#pragma once + +// Snapshot support is optional. Keeping camera behind this generated define +// lets a normal ws_bridge configuration build without the camera integration. +#ifdef USE_WS_BRIDGE_SNAPSHOT + +#include +#include +#include +#include +#include + +#include "esphome/components/camera/camera.h" +#include "esphome/core/component.h" +#include "esp_websocket_client.h" +#include "freertos/FreeRTOS.h" +#include "freertos/queue.h" + +#include "ws_protocol.h" + +namespace esphome::ws_bridge { + +class WsBridgeComponent; + +// The ESPHome loop owns camera/control decisions. A dedicated worker owns all +// esp_websocket_client calls, including start/stop/destroy and binary writes. +class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener { + public: + void set_parent(WsBridgeComponent *parent) { parent_ = parent; } + void set_camera(camera::Camera *camera) { camera_ = camera; } + void set_unique_id(const std::string &id) { unique_id_ = id; } + void set_name(const std::string &name) { name_ = name; } + void set_platform(const std::string &platform) { platform_ = platform; } + + void setup() override; + void loop() override; + void update() override; + void on_camera_image(const std::shared_ptr &image) override; + void dump_config() override; + + protected: + enum class WorkType : uint8_t { START, STOP, WAKE, EVENT }; + struct WorkItem { + WorkType type; + esp_websocket_client_handle_t client{nullptr}; + int32_t event_id{0}; + uint8_t opcode{0}; + int payload_len{0}; + int payload_offset{0}; + bool fin{false}; + uint32_t generation{0}; + std::string text; + std::string session_id; + }; + struct CallbackContext { + WsBridgeSnapshot *snapshot; + uint32_t generation; + }; + + static void media_task_(void *arg); + static void media_event_(void *arg, esp_event_base_t base, int32_t event_id, void *event_data); + void worker_loop_(); + void enqueue_(WorkItem *item); + void process_item_(WorkItem *item); + void start_media_(const std::string &session_id); + void stop_media_(); + void transport_failed_(); + void handle_media_event_(WorkItem *item); + void handle_media_message_(const std::string &raw); + void send_pending_(); + void on_control_connected_(); + void on_control_disconnected_(); + void on_control_message_(const ParsedMessage &message); + void open_media_(); + void request_frame_(); + + WsBridgeComponent *parent_{nullptr}; + camera::Camera *camera_{nullptr}; + std::string unique_id_; + std::string name_; + std::string platform_{"camera"}; + QueueHandle_t worker_queue_{nullptr}; + TaskHandle_t worker_task_{nullptr}; + esp_websocket_client_handle_t media_{nullptr}; // worker task only + std::unique_ptr callback_context_; // worker task only + std::deque> retired_callback_contexts_; + uint32_t media_generation_{0}; // worker task only, never reset + uint32_t active_generation_{0}; // worker task only + std::string worker_session_id_; // worker task only + uint32_t media_connect_id_{0}; // worker task only + std::string rx_accum_; // worker task only + bool rx_text_frame_{false}; // worker task only + bool rx_drop_message_{false}; // worker task only + + std::mutex frame_mutex_; + std::shared_ptr pending_image_; + bool capture_requested_{false}; + std::atomic handler_id_{0}; + std::atomic media_ready_{false}; + std::atomic needs_media_open_{false}; + std::atomic stopping_{false}; + uint32_t media_open_id_{0}; // ESPHome loop only + uint32_t media_open_sent_ms_{0}; + static constexpr size_t MAX_JPEG_BYTES = 1024 * 1024; + static constexpr size_t RX_ACCUM_MAX = 16384; + static constexpr uint32_t MEDIA_OPEN_TIMEOUT_MS = 15000; +}; + +} // namespace esphome::ws_bridge + +#endif // USE_WS_BRIDGE_SNAPSHOT diff --git a/components/ws_bridge/ws_protocol.cpp b/components/ws_bridge/ws_protocol.cpp index 9ddd6a99..a7144ab6 100644 --- a/components/ws_bridge/ws_protocol.cpp +++ b/components/ws_bridge/ws_protocol.cpp @@ -47,6 +47,15 @@ ParsedMessage parse_message(const std::string &raw) { if (msg.type == "result") { msg.success = root["success"].is() && root["success"].as(); if (!root["id"].isNull()) msg.id = root["id"].as(); + JsonObject result = root["result"].as(); + if (!result.isNull()) { + if (result["session_id"].is()) msg.session_id = result["session_id"].as(); + if (result["handler_id"].is()) { + uint32_t handler_id = result["handler_id"].as(); + // HA binary handler IDs are one byte; 0 is never a usable handler. + if (handler_id > 0 && handler_id <= 255) msg.handler_id = static_cast(handler_id); + } + } return true; } @@ -178,6 +187,22 @@ std::string build_ping(uint32_t id) { }); } +std::string build_media_open(uint32_t id, const std::string &unique_id) { + return json::build_json([&](JsonObject root) { + root["id"] = id; + root["type"] = "ws_bridge/media_open"; + root["unique_id"] = unique_id; + }); +} + +std::string build_media_connect(uint32_t id, const std::string &session_id) { + return json::build_json([&](JsonObject root) { + root["id"] = id; + root["type"] = "ws_bridge/media_connect"; + root["session_id"] = session_id; + }); +} + std::string build_state_float(uint32_t id, const std::string &unique_id, float value) { return json::build_json([&](JsonObject root) { root["id"] = id; diff --git a/components/ws_bridge/ws_protocol.h b/components/ws_bridge/ws_protocol.h index 7920aed0..1701a6bd 100644 --- a/components/ws_bridge/ws_protocol.h +++ b/components/ws_bridge/ws_protocol.h @@ -57,6 +57,8 @@ struct ParsedMessage { std::string type; // "auth_required" | "auth_ok" | "auth_invalid" | "result" | "event" | "" (unrecognized) bool success{false}; // for "result" uint32_t id{0}; // for "result": echoes the id of the command it answers + std::string session_id; // ws_bridge/media_open result + uint8_t handler_id{0}; // ws_bridge/media_connect result WsCommand command; // populated when type == "event" and event.kind == "command" }; @@ -75,6 +77,8 @@ std::string build_connect(uint32_t id, const std::string &gateway_id, const std: // layer itself doesn't notice (e.g. HA process killed without a clean WS // close, so the socket never sees a FIN/RST). std::string build_ping(uint32_t id); +std::string build_media_open(uint32_t id, const std::string &unique_id); +std::string build_media_connect(uint32_t id, const std::string &session_id); // Tells HA the complete set of unique_ids this gateway currently provides, so // it can drop entities we no longer declare (PROTOCOL.md §3.5). Entities that diff --git a/tests/components/ws_bridge/test.snapshot.esp32-idf.yaml b/tests/components/ws_bridge/test.snapshot.esp32-idf.yaml new file mode 100644 index 00000000..03d2dabd --- /dev/null +++ b/tests/components/ws_bridge/test.snapshot.esp32-idf.yaml @@ -0,0 +1,55 @@ +esphome: + name: test-ws-bridge-snapshot + +esp32: + board: esp32dev + framework: + type: esp-idf + +external_components: + - source: + type: local + path: ../../../components + components: [ws_bridge] + +wifi: + ssid: "test_ssid" + password: "test_password" + +psram: + mode: quad + speed: 80MHz + +# Compile-only camera wiring: these common ESP32-CAM pins are not accessed by +# this fixture, but ensure the generic camera::Camera listener path is built. +esp32_camera: + id: test_camera + name: "Test Camera" + external_clock: + pin: GPIO0 + frequency: 20MHz + i2c_pins: + sda: GPIO26 + scl: GPIO27 + data_pins: [GPIO5, GPIO18, GPIO19, GPIO21, GPIO36, GPIO39, GPIO34, GPIO35] + vsync_pin: GPIO25 + href_pin: GPIO23 + pixel_clock_pin: GPIO22 + resolution: 320x240 + +ws_bridge: + id: bridge + host: 192.0.2.1 + token: "test_token" + snapshots: + - camera_id: test_camera + unique_id: front_camera + name: "Front Camera" + # Default is camera; image proves both HA entity types use the same + # separate media WebSocket upload path. + update_interval: 30s + - camera_id: test_camera + unique_id: front_thumbnail + name: "Front Thumbnail" + platform: image + update_interval: 2min diff --git a/tests/components/ws_bridge/test.ssl-false.esp32-idf.yaml b/tests/components/ws_bridge/test.ssl-false.esp32-idf.yaml new file mode 100644 index 00000000..df85c966 --- /dev/null +++ b/tests/components/ws_bridge/test.ssl-false.esp32-idf.yaml @@ -0,0 +1,24 @@ +esphome: + name: test-ws-bridge-plain + +esp32: + board: esp32dev + framework: + type: esp-idf + +external_components: + - source: + type: local + path: ../../../components + components: [ws_bridge] + +wifi: + ssid: "test_ssid" + password: "test_password" + +# Exercises the no-certificate-bundle compile path. This is only appropriate +# for local development; normal deployments should keep ws_bridge's ssl:true. +ws_bridge: + host: 192.0.2.1 + ssl: false + token: "test_token" From a70983ef74832971b4f7afbabee1115ec061b05b Mon Sep 17 00:00:00 2001 From: eigger Date: Mon, 26 Oct 2026 15:53:49 +0900 Subject: [PATCH 2/3] fix(ws_bridge): recover snapshot media failures --- components/ws_bridge/__init__.py | 11 ++- components/ws_bridge/ws_bridge_snapshot.cpp | 91 +++++++++++++++++---- components/ws_bridge/ws_bridge_snapshot.h | 14 +++- 3 files changed, 94 insertions(+), 22 deletions(-) diff --git a/components/ws_bridge/__init__.py b/components/ws_bridge/__init__.py index 02d1c9fc..aa8ac8eb 100644 --- a/components/ws_bridge/__init__.py +++ b/components/ws_bridge/__init__.py @@ -329,10 +329,13 @@ async def to_code(config): # needs ESPHome's trimmed mbedTLS client symbols in every configuration. if hasattr(esp32, "request_tls"): esp32.request_tls() - else: - # ESPHome before 2026.10 has no request_tls helper. + elif hasattr(esp32, "include_builtin_idf_component"): + # ESPHome releases with component trimming but before request_tls(). esp32.include_builtin_idf_component("esp-tls") - esp32.add_idf_sdkconfig_option("CONFIG_MBEDTLS_TLS_ENABLED", True) + if hasattr(esp32, "add_idf_sdkconfig_option"): + esp32.add_idf_sdkconfig_option("CONFIG_MBEDTLS_TLS_ENABLED", True) + # Older ESPHome releases did not trim these TLS symbols, so no helper is + # needed (and their esp32 module does not expose the newer APIs above). if config[CONF_SSL]: # esp_websocket_client itself does not request ESPHome's trimmed TLS # pieces. Explicitly retain the TLS stack and public CA bundle used by @@ -341,7 +344,7 @@ async def to_code(config): cg.add_define("USE_WS_BRIDGE_CERT_BUNDLE") if hasattr(esp32, "require_certificate_bundle"): esp32.require_certificate_bundle() - else: + elif hasattr(esp32, "require_full_certificate_bundle"): # Older ESPHome's full-bundle request also enables the bundle. esp32.require_full_certificate_bundle() diff --git a/components/ws_bridge/ws_bridge_snapshot.cpp b/components/ws_bridge/ws_bridge_snapshot.cpp index c0372a0a..947ac8e9 100644 --- a/components/ws_bridge/ws_bridge_snapshot.cpp +++ b/components/ws_bridge/ws_bridge_snapshot.cpp @@ -2,6 +2,7 @@ #ifdef USE_WS_BRIDGE_SNAPSHOT +#include #ifdef USE_WS_BRIDGE_CERT_BUNDLE #include "esp_crt_bundle.h" #endif @@ -43,12 +44,15 @@ void WsBridgeSnapshot::dump_config() { } void WsBridgeSnapshot::loop() { - if (needs_media_open_.exchange(false, std::memory_order_acq_rel)) - open_media_(); + if (media_ready_.load(std::memory_order_acquire)) { + media_open_backoff_ms_ = 2000; + next_media_open_ms_ = 0; + } if (media_open_id_ != 0 && millis() - media_open_sent_ms_ >= MEDIA_OPEN_TIMEOUT_MS) { - media_open_id_ = 0; - open_media_(); + schedule_open_retry_(); } + if (needs_media_open_.load(std::memory_order_acquire) && media_open_id_ == 0) + open_media_(); } void WsBridgeSnapshot::update() { request_frame_(); } @@ -57,8 +61,9 @@ void WsBridgeSnapshot::on_control_connected_() { media_open_id_ = 0; media_ready_.store(false, std::memory_order_release); handler_id_.store(0, std::memory_order_release); - enqueue_(new WorkItem{WorkType::STOP}); - open_media_(); + stop_requested_.store(true, std::memory_order_release); + next_media_open_ms_ = 0; + needs_media_open_.store(true, std::memory_order_release); } void WsBridgeSnapshot::on_control_disconnected_() { @@ -70,15 +75,33 @@ void WsBridgeSnapshot::on_control_disconnected_() { capture_requested_ = false; pending_image_.reset(); } - enqueue_(new WorkItem{WorkType::STOP}); + stop_requested_.store(true, std::memory_order_release); } void WsBridgeSnapshot::open_media_() { if (parent_ == nullptr || !parent_->is_connected() || media_open_id_ != 0) return; + uint32_t now = millis(); + if (next_media_open_ms_ != 0 && static_cast(now - next_media_open_ms_) < 0) + return; media_open_id_ = parent_->open_media_session(unique_id_); - if (media_open_id_ != 0) + if (media_open_id_ != 0) { media_open_sent_ms_ = millis(); + // A later denial, worker failure, or dropped START waits at least this + // long before reserving another one-use HA session. + next_media_open_ms_ = now + media_open_backoff_ms_; + needs_media_open_.store(false, std::memory_order_release); + } else { + // The control TX queue can reject admission without a result callback. + schedule_open_retry_(); + } +} + +void WsBridgeSnapshot::schedule_open_retry_() { + media_open_id_ = 0; + needs_media_open_.store(true, std::memory_order_release); + next_media_open_ms_ = millis() + media_open_backoff_ms_; + media_open_backoff_ms_ = std::min(media_open_backoff_ms_ * 2, MEDIA_OPEN_BACKOFF_MAX_MS); } void WsBridgeSnapshot::on_control_message_(const ParsedMessage &message) { @@ -86,12 +109,13 @@ void WsBridgeSnapshot::on_control_message_(const ParsedMessage &message) { return; media_open_id_ = 0; if (!message.success || message.session_id.empty()) { - needs_media_open_.store(true, std::memory_order_release); + schedule_open_retry_(); return; } auto *item = new WorkItem{WorkType::START}; item->session_id = message.session_id; // opaque one-use session; never logged - enqueue_(item); + if (!enqueue_(item)) + schedule_open_retry_(); } void WsBridgeSnapshot::request_frame_() { @@ -120,24 +144,29 @@ void WsBridgeSnapshot::on_camera_image(const std::shared_ptr(arg)->worker_loop_(); } void WsBridgeSnapshot::worker_loop_() { while (true) { + if (stop_requested_.exchange(false, std::memory_order_acq_rel)) + stop_media_(); WorkItem *item = nullptr; if (xQueueReceive(worker_queue_, &item, pdMS_TO_TICKS(100)) == pdPASS && item != nullptr) { process_item_(item); delete item; } + check_worker_deadline_(); send_pending_(); } } @@ -147,9 +176,6 @@ void WsBridgeSnapshot::process_item_(WorkItem *item) { case WorkType::START: start_media_(item->session_id); break; - case WorkType::STOP: - stop_media_(); - break; case WorkType::WAKE: break; case WorkType::EVENT: @@ -165,6 +191,8 @@ void WsBridgeSnapshot::start_media_(const std::string &session_id) { return; } worker_session_id_ = session_id; + media_stage_ = MediaStage::WAIT_TRANSPORT; + media_stage_started_ms_ = millis(); esp_websocket_client_config_t config = {}; config.host = parent_->get_host().c_str(); config.port = parent_->get_port(); @@ -196,9 +224,18 @@ void WsBridgeSnapshot::stop_media_() { handler_id_.store(0, std::memory_order_release); worker_session_id_.clear(); media_connect_id_ = 0; + media_stage_ = MediaStage::IDLE; + media_stage_started_ms_ = 0; rx_accum_.clear(); rx_text_frame_ = false; rx_drop_message_ = false; + { + std::lock_guard lock(frame_mutex_); + // A response from a request issued before this media session died must + // not retain the camera buffer or become a frame for a replacement. + capture_requested_ = false; + pending_image_.reset(); + } if (media_ != nullptr) { stopping_.store(true, std::memory_order_release); esp_websocket_unregister_events(media_, WEBSOCKET_EVENT_ANY, media_event_); @@ -229,6 +266,15 @@ void WsBridgeSnapshot::transport_failed_() { needs_media_open_.store(true, std::memory_order_release); } +void WsBridgeSnapshot::check_worker_deadline_() { + if (media_ == nullptr || media_stage_ == MediaStage::IDLE || media_stage_ == MediaStage::READY) + return; + if (millis() - media_stage_started_ms_ >= MEDIA_STAGE_TIMEOUT_MS) { + ESP_LOGW(TAG, "Media %s timed out", media_stage_ == MediaStage::WAIT_MEDIA_CONNECT ? "connect" : "authentication"); + transport_failed_(); + } +} + void WsBridgeSnapshot::media_event_(void *arg, esp_event_base_t, int32_t event_id, void *event_data) { auto *context = static_cast(arg); if (context == nullptr || context->snapshot == nullptr) @@ -247,7 +293,15 @@ void WsBridgeSnapshot::media_event_(void *arg, esp_event_base_t, int32_t event_i if (data->data_ptr != nullptr && data->data_len > 0) item->text.assign(data->data_ptr, data->data_len); } - self->enqueue_(item); + if (!self->enqueue_(item) && + (event_id == WEBSOCKET_EVENT_DISCONNECTED || event_id == WEBSOCKET_EVENT_ERROR || + event_id == WEBSOCKET_EVENT_CLOSED)) { + // A terminal transport event must not be lost behind a full queue. + self->media_ready_.store(false, std::memory_order_release); + self->handler_id_.store(0, std::memory_order_release); + self->stop_requested_.store(true, std::memory_order_release); + self->needs_media_open_.store(true, std::memory_order_release); + } } void WsBridgeSnapshot::handle_media_event_(WorkItem *item) { @@ -301,11 +355,15 @@ void WsBridgeSnapshot::handle_media_message_(const std::string &raw) { ParsedMessage message = parse_message(raw); if (message.type == "auth_required") { std::string auth = build_auth(parent_->get_token()); + media_stage_ = MediaStage::WAIT_AUTH; + media_stage_started_ms_ = millis(); if (media_ == nullptr || esp_websocket_client_send_text(media_, auth.c_str(), auth.size(), pdMS_TO_TICKS(10000)) < 0) transport_failed_(); } else if (message.type == "auth_ok") { media_connect_id_ = 1; std::string connect = build_media_connect(media_connect_id_, worker_session_id_); + media_stage_ = MediaStage::WAIT_MEDIA_CONNECT; + media_stage_started_ms_ = millis(); if (media_ == nullptr || esp_websocket_client_send_text(media_, connect.c_str(), connect.size(), pdMS_TO_TICKS(10000)) < 0) transport_failed_(); } else if (message.type == "auth_invalid") { @@ -314,6 +372,7 @@ void WsBridgeSnapshot::handle_media_message_(const std::string &raw) { } else if (message.type == "result" && message.id == media_connect_id_ && message.success && message.handler_id != 0) { handler_id_.store(message.handler_id, std::memory_order_release); media_ready_.store(true, std::memory_order_release); + media_stage_ = MediaStage::READY; } else if (message.type == "result" && message.id == media_connect_id_) { ESP_LOGW(TAG, "Home Assistant rejected media session"); transport_failed_(); diff --git a/components/ws_bridge/ws_bridge_snapshot.h b/components/ws_bridge/ws_bridge_snapshot.h index 58bddc3c..933fbfbc 100644 --- a/components/ws_bridge/ws_bridge_snapshot.h +++ b/components/ws_bridge/ws_bridge_snapshot.h @@ -39,7 +39,8 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener void dump_config() override; protected: - enum class WorkType : uint8_t { START, STOP, WAKE, EVENT }; + enum class WorkType : uint8_t { START, WAKE, EVENT }; + enum class MediaStage : uint8_t { IDLE, WAIT_TRANSPORT, WAIT_AUTH, WAIT_MEDIA_CONNECT, READY }; struct WorkItem { WorkType type; esp_websocket_client_handle_t client{nullptr}; @@ -60,11 +61,12 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener static void media_task_(void *arg); static void media_event_(void *arg, esp_event_base_t base, int32_t event_id, void *event_data); void worker_loop_(); - void enqueue_(WorkItem *item); + bool enqueue_(WorkItem *item); void process_item_(WorkItem *item); void start_media_(const std::string &session_id); void stop_media_(); void transport_failed_(); + void check_worker_deadline_(); void handle_media_event_(WorkItem *item); void handle_media_message_(const std::string &raw); void send_pending_(); @@ -72,6 +74,7 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener void on_control_disconnected_(); void on_control_message_(const ParsedMessage &message); void open_media_(); + void schedule_open_retry_(); void request_frame_(); WsBridgeComponent *parent_{nullptr}; @@ -88,6 +91,8 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener uint32_t active_generation_{0}; // worker task only std::string worker_session_id_; // worker task only uint32_t media_connect_id_{0}; // worker task only + MediaStage media_stage_{MediaStage::IDLE}; // worker task only + uint32_t media_stage_started_ms_{0}; // worker task only std::string rx_accum_; // worker task only bool rx_text_frame_{false}; // worker task only bool rx_drop_message_{false}; // worker task only @@ -98,12 +103,17 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener std::atomic handler_id_{0}; std::atomic media_ready_{false}; std::atomic needs_media_open_{false}; + std::atomic stop_requested_{false}; std::atomic stopping_{false}; uint32_t media_open_id_{0}; // ESPHome loop only uint32_t media_open_sent_ms_{0}; + uint32_t next_media_open_ms_{0}; + uint32_t media_open_backoff_ms_{2000}; static constexpr size_t MAX_JPEG_BYTES = 1024 * 1024; static constexpr size_t RX_ACCUM_MAX = 16384; static constexpr uint32_t MEDIA_OPEN_TIMEOUT_MS = 15000; + static constexpr uint32_t MEDIA_STAGE_TIMEOUT_MS = 15000; + static constexpr uint32_t MEDIA_OPEN_BACKOFF_MAX_MS = 30000; }; } // namespace esphome::ws_bridge From 17ff440ec3e616ba0e17f7cf646ef4d0d3b8ef04 Mon Sep 17 00:00:00 2001 From: eigger Date: Mon, 26 Oct 2026 15:57:19 +0900 Subject: [PATCH 3/3] fix(ws_bridge): ignore stale media queue failures --- components/ws_bridge/ws_bridge_snapshot.cpp | 16 +++++++++++----- components/ws_bridge/ws_bridge_snapshot.h | 4 ++++ 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/components/ws_bridge/ws_bridge_snapshot.cpp b/components/ws_bridge/ws_bridge_snapshot.cpp index 947ac8e9..6557ac10 100644 --- a/components/ws_bridge/ws_bridge_snapshot.cpp +++ b/components/ws_bridge/ws_bridge_snapshot.cpp @@ -159,6 +159,9 @@ void WsBridgeSnapshot::media_task_(void *arg) { static_cast( void WsBridgeSnapshot::worker_loop_() { while (true) { + uint32_t terminal_generation = terminal_event_generation_.exchange(0, std::memory_order_acq_rel); + if (terminal_generation != 0 && terminal_generation == active_generation_) + transport_failed_(); if (stop_requested_.exchange(false, std::memory_order_acq_rel)) stop_media_(); WorkItem *item = nullptr; @@ -296,11 +299,14 @@ void WsBridgeSnapshot::media_event_(void *arg, esp_event_base_t, int32_t event_i if (!self->enqueue_(item) && (event_id == WEBSOCKET_EVENT_DISCONNECTED || event_id == WEBSOCKET_EVENT_ERROR || event_id == WEBSOCKET_EVENT_CLOSED)) { - // A terminal transport event must not be lost behind a full queue. - self->media_ready_.store(false, std::memory_order_release); - self->handler_id_.store(0, std::memory_order_release); - self->stop_requested_.store(true, std::memory_order_release); - self->needs_media_open_.store(true, std::memory_order_release); + // A terminal transport event must not be lost behind a full queue. Do + // not mutate global connection state here: this callback can belong to a + // retired client. The worker validates the recorded immutable generation. + uint32_t observed = self->terminal_event_generation_.load(std::memory_order_acquire); + while (observed < context->generation && + !self->terminal_event_generation_.compare_exchange_weak( + observed, context->generation, std::memory_order_release, std::memory_order_acquire)) { + } } } diff --git a/components/ws_bridge/ws_bridge_snapshot.h b/components/ws_bridge/ws_bridge_snapshot.h index 933fbfbc..c93365b9 100644 --- a/components/ws_bridge/ws_bridge_snapshot.h +++ b/components/ws_bridge/ws_bridge_snapshot.h @@ -104,6 +104,10 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener std::atomic media_ready_{false}; std::atomic needs_media_open_{false}; std::atomic stop_requested_{false}; + // Event callbacks cannot read worker-only active_generation_. On a full + // queue they record only their immutable generation; the worker compares it + // before deciding whether the current client may be stopped. + std::atomic terminal_event_generation_{0}; std::atomic stopping_{false}; uint32_t media_open_id_{0}; // ESPHome loop only uint32_t media_open_sent_ms_{0};