Merge remote-tracking branch 'origin/dev' into HEAD

esphome/components/api/api_connection.h

@@ -802,8 +802,9 @@ class APIConnection final : public APIServerConnectionBase {
   }
 
   // Helper function to schedule a high priority message at the front of the batch
-  // Out-of-line: callers (on_shutdown, check_keepalive_) are cold paths
+  // Out-of-line: callers (send_shutdown_disconnect_, check_keepalive_) are cold paths
   bool schedule_message_front_(EntityBase *entity, uint16_t message_type, uint8_t estimated_size);
+  void send_shutdown_disconnect_();
 
   // Helper function to log client messages with name and peername
   void log_client_(int level, const LogString *message);

esphome/components/api/api_server.cpp

@@ -4,6 +4,7 @@
 #include <cinttypes>
 #include <cstring>
 #include "api_connection.h"
+#include "api_connection_buffer.h"
 #include "esphome/components/network/util.h"
 #include "esphome/core/application.h"
 #include "esphome/core/defines.h"
@@ -806,6 +807,31 @@ void APIServer::on_camera_image(const std::shared_ptr<camera::CameraImage> &imag
 }
 #endif
 
+// Kept out of api_connection.cpp so GCC inlines its send path as before
+void APIConnection::send_shutdown_disconnect_() {
+#ifndef USE_DEEP_SLEEP
+  // Queued states first: Home Assistant mishandles states after DisconnectRequest (#22610).
+  // Not with deep sleep: its entities stay available, and the disconnect must not wait on a slow link
+  this->flags_.state_subscription = false;
+  this->destroy_active_iterator_();
+  // process_batch_ first flushes any overflow without blocking; stop once the queue stops shrinking
+  for (size_t left = this->deferred_batch_.size(); left != 0;) {
+    this->process_batch_();
+    const size_t now_left = this->deferred_batch_.size();
+    if (now_left >= left)
+      break;
+    left = now_left;
+  }
+#endif
+  DisconnectRequest req;
+  if (!this->send_message(req)) {
+    // tx buffer full: first in the queue, so a slow link still delivers it before the teardown timeout.
+    // Due at once: teardown does not advance the cached loop time the batch_delay gate reads
+    this->schedule_message_front_(nullptr, DisconnectRequest::MESSAGE_TYPE, DisconnectRequest::ESTIMATED_SIZE);
+    this->make_batch_due_now_();
+  }
+}
+
 void APIServer::on_shutdown() {
   this->shutting_down_ = true;
 
@@ -818,15 +844,8 @@ void APIServer::on_shutdown() {
   // Change batch delay to 5ms for quick flushing during shutdown
   this->batch_delay_ = 5;
 
-  // Send disconnect requests to all connected clients
-  for (auto &c : this->active_clients()) {
-    DisconnectRequest req;
-    if (!c->send_message(req)) {
-      // If we can't send the disconnect request directly (tx_buffer full),
-      // schedule it at the front of the batch so it will be sent with priority
-      c->schedule_message_front_(nullptr, DisconnectRequest::MESSAGE_TYPE, DisconnectRequest::ESTIMATED_SIZE);
-    }
-  }
+  for (auto &c : this->active_clients())
+    c->send_shutdown_disconnect_();
 }
 
 bool APIServer::teardown() {

tests/integration/fixtures/api_shutdown_disconnect_backpressure.yaml

@@ -0,0 +1,65 @@
+esphome:
+  name: api-shutdown-bp-test
+
+host:
+
+api:
+  # One queued message, so DisconnectRequest cannot join the overflow buffer once it holds data
+  max_send_queue: 1
+  actions:
+    - action: fill_then_restart
+      then:
+        - lambda: &fill_pipe |-
+            // The client is not reading: log lines fill the TCP pipe, then the overflow buffer. Keep going
+            // until the overflow buffer holds data with the client's window closed, so nothing drains it
+            auto *client = sndbuf_pin::api_log_client();
+            const std::string padding(250, 'x');
+            const uint32_t start = millis();
+            bool full = false;
+            while (client != nullptr && millis() - start < 5000) {
+              const bool window_closed = sndbuf_pin::api_peer_window_closed();
+              if (client->try_to_clear_buffer(false)) {
+                ESP_LOGD("fill", "fill line %s", padding.c_str());
+              } else if (window_closed) {
+                full = true;
+                break;
+              } else {
+                delay(1);
+              }
+            }
+            if (!full)
+              ESP_LOGE("fill", "fill timed out");
+        # The test reads again once this line is on the console
+        - logger.log: fill done
+        # DisconnectRequest cannot be sent now, so it waits in an otherwise empty batch
+        - button.press: restart_button
+    - action: fill_publish_then_restart
+      then:
+        - lambda: *fill_pipe
+        # Queued behind the blocked socket when the shutdown starts
+        - lambda: id(queued_sensor).publish_state(42);
+        - logger.log: fill done
+        - button.press: restart_button
+
+button:
+  - platform: restart
+    id: restart_button
+    name: "Restart"
+
+sensor:
+  - platform: template
+    id: queued_sensor
+    name: "Queued sensor"
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+    components: [sndbuf_pin_component]
+
+# Pins the device's socket send buffers for deterministic TCP backpressure
+sndbuf_pin_component:
+  buffer_size: SERVER_SNDBUF
+
+logger:
+  level: DEBUG

tests/integration/fixtures/api_shutdown_sends_states_first.yaml

@@ -0,0 +1,114 @@
+esphome:
+  name: api-shutdown-order
+
+host:
+
+logger:
+
+api:
+
+switch:
+  - platform: template
+    id: test_switch
+    name: "Test switch"
+    optimistic: true
+
+button:
+  - platform: restart
+    id: restart_button
+    name: "Restart"
+  - platform: template
+    name: "Toggle then restart"
+    on_press:
+      - switch.toggle: test_switch
+      - button.press: restart_button
+  - platform: template
+    name: "Publish then restart"
+    on_press:
+      - lambda: |-
+          for (auto *sensor : App.get_sensors())
+            sensor->publish_state(1);
+      - button.press: restart_button
+
+sensor:
+  # More states than one batch packet holds (MAX_MESSAGES_PER_BATCH is 34)
+  - platform: template
+    name: "Burst 0"
+  - platform: template
+    name: "Burst 1"
+  - platform: template
+    name: "Burst 2"
+  - platform: template
+    name: "Burst 3"
+  - platform: template
+    name: "Burst 4"
+  - platform: template
+    name: "Burst 5"
+  - platform: template
+    name: "Burst 6"
+  - platform: template
+    name: "Burst 7"
+  - platform: template
+    name: "Burst 8"
+  - platform: template
+    name: "Burst 9"
+  - platform: template
+    name: "Burst 10"
+  - platform: template
+    name: "Burst 11"
+  - platform: template
+    name: "Burst 12"
+  - platform: template
+    name: "Burst 13"
+  - platform: template
+    name: "Burst 14"
+  - platform: template
+    name: "Burst 15"
+  - platform: template
+    name: "Burst 16"
+  - platform: template
+    name: "Burst 17"
+  - platform: template
+    name: "Burst 18"
+  - platform: template
+    name: "Burst 19"
+  - platform: template
+    name: "Burst 20"
+  - platform: template
+    name: "Burst 21"
+  - platform: template
+    name: "Burst 22"
+  - platform: template
+    name: "Burst 23"
+  - platform: template
+    name: "Burst 24"
+  - platform: template
+    name: "Burst 25"
+  - platform: template
+    name: "Burst 26"
+  - platform: template
+    name: "Burst 27"
+  - platform: template
+    name: "Burst 28"
+  - platform: template
+    name: "Burst 29"
+  - platform: template
+    name: "Burst 30"
+  - platform: template
+    name: "Burst 31"
+  - platform: template
+    name: "Burst 32"
+  - platform: template
+    name: "Burst 33"
+  - platform: template
+    name: "Burst 34"
+  - platform: template
+    name: "Burst 35"
+  - platform: template
+    name: "Burst 36"
+  - platform: template
+    name: "Burst 37"
+  - platform: template
+    name: "Burst 38"
+  - platform: template
+    name: "Burst 39"

tests/integration/test_api_shutdown_disconnect_backpressure.py

@@ -0,0 +1,112 @@
+"""A client whose send path is full at shutdown still gets DisconnectRequest.
+
+The disconnect then waits in an otherwise empty deferred batch. Teardown does not
+advance the cached loop time that the batch_delay gate reads, so unless the batch
+is made due at once it is never sent before the reboot.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+
+from aioesphomeapi import api_pb2
+import pytest
+
+from .log_utils import LineWaiter
+from .raw_api_client import (
+    MESSAGE_TYPE_OF,
+    RECV_BUFFER_SIZE,
+    SNDBUF_PIN_LOG,
+    RawApiClient,
+    pin_server_sndbuf,
+)
+from .types import RunCompiledFunction
+
+DISCONNECT_REQUEST = MESSAGE_TYPE_OF[api_pb2.DisconnectRequest]
+SERVICES_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesServicesResponse]
+LIST_DONE_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesDoneResponse]
+SENSOR_STATE = MESSAGE_TYPE_OF[api_pb2.SensorStateResponse]
+
+LINUX_ONLY = pytest.mark.skipif(
+    sys.platform != "linux",
+    reason="other kernels grow loopback buffers past the pinned size, so the pipe never fills",
+)
+
+
+async def _read_until_disconnect_or_closed(client: RawApiClient) -> None:
+    while not client.frame_counts[DISCONNECT_REQUEST]:
+        try:
+            await client._recv_once()
+        except (AssertionError, ConnectionError):
+            return
+
+
+async def _fill_then_restart(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    port: int,
+    action: str,
+    subscribe_states: bool,
+) -> RawApiClient:
+    """Run ``action`` with the client not reading, then read again during teardown."""
+    waiter = LineWaiter()
+    async with run_compiled(
+        pin_server_sndbuf(yaml_config), line_callback=waiter.callback
+    ):
+        await waiter.wait_for(SNDBUF_PIN_LOG, timeout=10)
+
+        async with RawApiClient(port, recv_buffer_size=RECV_BUFFER_SIZE) as client:
+            await client.connect(client_info="shutdown-backpressure-client")
+            await client.send_message(api_pb2.ListEntitiesRequest())
+            await client.read_until_frame(LIST_DONE_RESPONSE)
+            services = [
+                api_pb2.ListEntitiesServicesResponse.FromString(payload)
+                for payload in client.frames(SERVICES_RESPONSE)
+            ]
+            (action_key,) = (svc.key for svc in services if svc.name == action)
+            if subscribe_states:
+                await client.send_message(api_pb2.SubscribeStatesRequest())
+                # Take the initial state so nothing is queued before the fill
+                await client.read_until_frame(SENSOR_STATE)
+            await client.send_message(
+                api_pb2.SubscribeLogsRequest(level=api_pb2.LOG_LEVEL_DEBUG)
+            )
+
+            # The client stops reading here; the device console tells when the pipe is full
+            await client.send_message(api_pb2.ExecuteServiceRequest(key=action_key))
+            await waiter.wait_for("fill done", timeout=10)
+            # Logged just before the shutdown hooks; reading earlier would let the direct send succeed
+            await waiter.wait_for("Rebooting safely", timeout=10)
+            assert not any("fill timed out" in line for line in waiter.lines), (
+                "the client's receive window never closed while the overflow buffer held data"
+            )
+
+            # Read again during the 1 s teardown; the disconnect must arrive before the reboot
+            await asyncio.wait_for(_read_until_disconnect_or_closed(client), 10)
+            return client
+
+
+@LINUX_ONLY
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("action", "subscribe_states"),
+    [
+        ("fill_then_restart", False),
+        # A state queued behind the blocked socket when the shutdown starts
+        ("fill_publish_then_restart", True),
+    ],
+)
+async def test_api_shutdown_disconnect_backpressure(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    unused_tcp_port: int,
+    action: str,
+    subscribe_states: bool,
+) -> None:
+    client = await _fill_then_restart(
+        yaml_config, run_compiled, unused_tcp_port, action, subscribe_states
+    )
+    assert client.frame_counts[DISCONNECT_REQUEST] == 1, (
+        "the backed up client never got DisconnectRequest"
+    )

tests/integration/test_api_shutdown_sends_states_first.py

@@ -0,0 +1,82 @@
+"""States published right before a shutdown reach the client before DisconnectRequest."""
+
+from __future__ import annotations
+
+from aioesphomeapi import (
+    ButtonInfo,
+    EntityState,
+    SensorInfo,
+    SensorState,
+    SwitchInfo,
+    SwitchState,
+)
+import pytest
+
+from .state_utils import require_entity, subscribe_after_initial_states
+from .types import APIClientConnectedWithDisconnectFactory, RunCompiledFunction
+from .wait_utils import wait_or_fail
+
+# Sensors in the fixture: more than MAX_MESSAGES_PER_BATCH (34) states
+BURST_SENSORS = 40
+
+
+@pytest.mark.asyncio
+@pytest.mark.shared_yaml("api_shutdown_sends_states_first")
+async def test_api_shutdown_sends_states_first(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected_with_disconnect: APIClientConnectedWithDisconnectFactory,
+) -> None:
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected_with_disconnect() as (client, disconnect_event),
+    ):
+        entities, _ = await client.list_entities_services()
+        switch = require_entity(entities, "test_switch", SwitchInfo)
+        button = require_entity(entities, "toggle_then_restart", ButtonInfo)
+
+        states: list[EntityState] = []
+        sub = await subscribe_after_initial_states(client, entities, states.append)
+        assert sub.initial_states[switch.key].state is False
+
+        client.button_command(button.key)
+        await wait_or_fail(
+            disconnect_event, msg="device never disconnected", states=states
+        )
+
+        assert [s.state for s in states if isinstance(s, SwitchState)] == [True], (
+            "the toggled state must arrive before the device disconnects"
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.shared_yaml("api_shutdown_sends_states_first")
+async def test_api_shutdown_sends_burst_before_disconnect(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected_with_disconnect: APIClientConnectedWithDisconnectFactory,
+) -> None:
+    """More states than one batch packet holds all arrive before the disconnect."""
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected_with_disconnect() as (client, disconnect_event),
+    ):
+        entities, _ = await client.list_entities_services()
+        button = require_entity(entities, "publish_then_restart", ButtonInfo)
+        burst_keys = {e.key for e in entities if isinstance(e, SensorInfo)}
+        assert len(burst_keys) == BURST_SENSORS
+
+        states: list[EntityState] = []
+        await subscribe_after_initial_states(client, entities, states.append)
+
+        client.button_command(button.key)
+        await wait_or_fail(
+            disconnect_event, msg="device never disconnected", states=states
+        )
+
+        received = {
+            s.key for s in states if isinstance(s, SensorState) and s.key in burst_keys
+        }
+        assert received == burst_keys, (
+            f"{len(burst_keys - received)} of {BURST_SENSORS} states missed the disconnect"
+        )