diff --git a/esphome/components/api/api_connection.cpp b/esphome/components/api/api_connection.cpp index 73b4f3e5bd..6607e211da 100644 --- a/esphome/components/api/api_connection.cpp +++ b/esphome/components/api/api_connection.cpp @@ -417,15 +417,15 @@ void APIConnection::finalize_iterator_sync_() { } void APIConnection::process_iterator_batch_(ComponentIterator &iterator) { - size_t initial_size = this->deferred_batch_.size(); - size_t max_batch = MAX_INITIAL_PER_BATCH; - while (!iterator.completed() && (this->deferred_batch_.size() - initial_size) < max_batch) { - iterator.advance(); - } + // Budget by remaining batch capacity so a pass cannot overfill the batch; + // stops early on a refused send and resumes next loop pass + size_t batch_size = this->deferred_batch_.size(); + if (batch_size < MAX_INITIAL_BATCH_SIZE) + iterator.try_advance(MAX_INITIAL_BATCH_SIZE - batch_size); - // If the batch is full, process it immediately - // Note: iterator.advance() already calls schedule_batch_() via schedule_message_() - if (this->deferred_batch_.size() >= max_batch) { + // Flush immediately once enough is queued (not guaranteed every pass); + // partial batches go out via the batch timer or finalize_iterator_sync_() + if (this->deferred_batch_.size() >= MAX_INITIAL_BATCH_SIZE) { this->process_batch_(); } } diff --git a/esphome/components/api/api_connection.h b/esphome/components/api/api_connection.h index bb51a13000..1b47c23cfe 100644 --- a/esphome/components/api/api_connection.h +++ b/esphome/components/api/api_connection.h @@ -53,11 +53,11 @@ void log_dropped_message(const char *tag, int line, const LogString *what); // Keepalive timeout in milliseconds static constexpr uint32_t KEEPALIVE_TIMEOUT_MS = 60000; -// Maximum number of entities to process in a single batch during initial state/info sending -static constexpr size_t MAX_INITIAL_PER_BATCH = 34; +// Deferred batch size cap during initial state/info sync +static constexpr size_t MAX_INITIAL_BATCH_SIZE = 34; // Verify MAX_MESSAGES_PER_BATCH (defined in api_frame_helper.h) can hold the initial batch -static_assert(MAX_MESSAGES_PER_BATCH >= MAX_INITIAL_PER_BATCH, - "MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_PER_BATCH"); +static_assert(MAX_MESSAGES_PER_BATCH >= MAX_INITIAL_BATCH_SIZE, + "MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_BATCH_SIZE"); #ifdef USE_BENCHMARK class APIConnection; diff --git a/esphome/components/api/api_frame_helper.h b/esphome/components/api/api_frame_helper.h index 9c49956bbd..1c60bb87a5 100644 --- a/esphome/components/api/api_frame_helper.h +++ b/esphome/components/api/api_frame_helper.h @@ -36,7 +36,7 @@ static constexpr uint16_t MAX_MESSAGE_SIZE = 32768; // 32 KiB for ESP32 and oth static constexpr uint16_t RX_BUF_NULL_TERMINATOR = 1; // Maximum number of messages to batch in a single write operation -// Must be >= MAX_INITIAL_PER_BATCH in api_connection.h (enforced by static_assert there) +// Must be >= MAX_INITIAL_BATCH_SIZE in api_connection.h (enforced by static_assert there) static constexpr size_t MAX_MESSAGES_PER_BATCH = 34; // Max client name length (e.g., "Home Assistant 2026.1.0.dev0" = 28 chars) diff --git a/esphome/components/api/list_entities.cpp b/esphome/components/api/list_entities.cpp index f9e645b506..57ff616ca7 100644 --- a/esphome/components/api/list_entities.cpp +++ b/esphome/components/api/list_entities.cpp @@ -95,9 +95,17 @@ bool ListEntitiesIterator::on_end() { return this->client_->send_list_info_done( ListEntitiesIterator::ListEntitiesIterator(APIConnection *client) : client_(client) {} #ifdef USE_API_USER_DEFINED_ACTIONS +// Yield after every Nth service; bounds direct (non-batched) writes per loop pass +static constexpr uint8_t SERVICE_YIELD_INTERVAL = 3; + bool ListEntitiesIterator::on_service(UserServiceDescriptor *service) { auto resp = service->encode_list_service_response(); - return this->client_->send_message(resp); + if (!this->client_->send_message(resp)) + return false; + // at_ is this service's index + if ((this->at_ + 1) % SERVICE_YIELD_INTERVAL == 0) + this->yield_after_step_(); + return true; } #endif diff --git a/esphome/components/web_server/list_entities.h b/esphome/components/web_server/list_entities.h index 3edb84f555..dc32cbd1ad 100644 --- a/esphome/components/web_server/list_entities.h +++ b/esphome/components/web_server/list_entities.h @@ -35,7 +35,6 @@ class ListEntitiesIterator final : public ComponentIterator { #undef ENTITY_TYPE_ #undef ENTITY_CONTROLLER_TYPE_ // NOLINTEND(bugprone-macro-parentheses) - bool completed() { return this->state_ == IteratorState::NONE; } protected: const WebServer *web_server_; diff --git a/esphome/components/web_server/web_server.cpp b/esphome/components/web_server/web_server.cpp index 9e50b7a394..ec536910e5 100644 --- a/esphome/components/web_server/web_server.cpp +++ b/esphome/components/web_server/web_server.cpp @@ -214,8 +214,8 @@ void DeferredUpdateEventSource::process_deferred_queue_() { void DeferredUpdateEventSource::loop() { process_deferred_queue_(); - if (!this->entities_iterator_.completed()) - this->entities_iterator_.advance(); + // One step per loop; refusals retry next pass + this->entities_iterator_.try_advance(1); } void DeferredUpdateEventSource::deferrable_send_state(void *source, const char *event_type, @@ -321,12 +321,6 @@ void DeferredUpdateEventSourceList::on_client_connect_(DeferredUpdateEventSource #endif source->entities_iterator_.begin(ws->include_internal_); - - // just dump them all up-front and take advantage of the deferred queue - // on second thought that takes too long, but leaving the commented code here for debug purposes - // while(!source->entities_iterator_.completed()) { - // source->entities_iterator_.advance(); - //} }); } diff --git a/esphome/components/web_server_idf/web_server_idf.cpp b/esphome/components/web_server_idf/web_server_idf.cpp index 993fb6c035..9550570cdc 100644 --- a/esphome/components/web_server_idf/web_server_idf.cpp +++ b/esphome/components/web_server_idf/web_server_idf.cpp @@ -935,8 +935,8 @@ void AsyncEventSourceResponse::process_buffer_() { void AsyncEventSourceResponse::loop() { process_buffer_(); process_deferred_queue_(); - if (!this->entities_iterator_.completed()) - this->entities_iterator_.advance(); + // 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, diff --git a/esphome/core/component_iterator.cpp b/esphome/core/component_iterator.cpp index f4d3c05e19..3a497db741 100644 --- a/esphome/core/component_iterator.cpp +++ b/esphome/core/component_iterator.cpp @@ -22,23 +22,23 @@ void ComponentIterator::advance_platform_() { this->at_ = 0; } -void ComponentIterator::advance() { +bool ComponentIterator::advance_step_() { switch (this->state_) { case IteratorState::NONE: // not started - return; + return false; case IteratorState::BEGIN: if (this->on_begin()) { advance_platform_(); + return true; } - break; + return false; // Entity iterator cases (generated from entity_types.h) // NOLINTBEGIN(bugprone-macro-parentheses) #define ENTITY_TYPE_(type, singular, plural, count, upper) \ case IteratorState::upper: \ - this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular); \ - break; + return this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular); #define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \ ENTITY_TYPE_(type, singular, plural, count, upper) #include "esphome/core/entity_types.h" @@ -48,26 +48,29 @@ void ComponentIterator::advance() { #ifdef USE_API_USER_DEFINED_ACTIONS case IteratorState::SERVICE: - this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service); - break; + return this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service); #endif #ifdef USE_CAMERA case IteratorState::CAMERA: { camera::Camera *camera_instance = camera::Camera::instance(); - if (camera_instance != nullptr && (!camera_instance->is_internal() || this->include_internal_)) { - this->on_camera(camera_instance); + if (camera_instance != nullptr && (!camera_instance->is_internal() || this->include_internal_) && + !this->on_camera(camera_instance)) { + return false; } advance_platform_(); - } break; + return true; + } #endif case IteratorState::MAX: if (this->on_end()) { this->state_ = IteratorState::NONE; + return true; } - return; + return false; } + return false; } bool ComponentIterator::on_end() { return true; } diff --git a/esphome/core/component_iterator.h b/esphome/core/component_iterator.h index d271fcfed0..fac09e9e14 100644 --- a/esphome/core/component_iterator.h +++ b/esphome/core/component_iterator.h @@ -30,7 +30,23 @@ class RadioFrequency; class ComponentIterator { public: void begin(bool include_internal = false); - void advance(); + /// Run up to max_steps iteration steps; stops early when iteration + /// completes or a callback refuses (that step is retried on the next + /// call). Inline so an idle (completed) iterator costs one compare, no call. + ESPHOME_ALWAYS_INLINE void try_advance(size_t max_steps) { + size_t steps = 0; + while (steps < max_steps && !this->completed()) { + this->yield_requested_ = false; + if (!this->advance_step_()) + break; + steps++; + if (this->yield_requested_) + break; + } + } + // Remove before 2027.3.0 + ESPDEPRECATED("Use try_advance() instead. Removed in 2027.3.0", "2026.8.1") + void advance() { this->try_advance(1); } bool completed() const { return this->state_ == IteratorState::NONE; } virtual bool on_begin(); // Pure virtual entity callbacks (generated from entity_types.h) @@ -73,23 +89,34 @@ class ComponentIterator { #endif MAX, }; + /// End the current try_advance() pass after this step; lets callbacks + /// that write directly to the socket cap direct writes per pass. + void yield_after_step_() { this->yield_requested_ = true; } + uint16_t at_{0}; // Supports up to 65,535 entities per type IteratorState state_{IteratorState::NONE}; - bool include_internal_{false}; + bool yield_requested_ : 1 {false}; + bool include_internal_ : 1 {false}; template - void process_platform_item_(const Container &items, + bool process_platform_item_(const Container &items, bool (ComponentIterator::*on_item)(typename Container::value_type)) { if (this->at_ >= items.size()) { this->advance_platform_(); - } else { - typename Container::value_type item = items[this->at_]; - if ((item->is_internal() && !this->include_internal_) || (this->*on_item)(item)) { - this->at_++; - } + return true; } + typename Container::value_type item = items[this->at_]; + if ((item->is_internal() && !this->include_internal_) || (this->*on_item)(item)) { + this->at_++; + return true; + } + return false; } + /// One iteration step; false if no progress was made (callback refused + /// or iterator not running). + bool advance_step_(); + void advance_platform_(); }; diff --git a/tests/components/camera/__init__.py b/tests/components/camera/__init__.py new file mode 100644 index 0000000000..61104b0bb1 --- /dev/null +++ b/tests/components/camera/__init__.py @@ -0,0 +1,11 @@ +import esphome.codegen as cg +from tests.testing_helpers import ComponentManifestOverride + + +def override_manifest(manifest: ComponentManifestOverride) -> None: + # No host camera platform exists to emit USE_CAMERA; define it here so + # the iterator CAMERA state compiles into the test binary. + async def to_code_testing(config): + cg.add_define("USE_CAMERA") + + manifest.to_code = to_code_testing diff --git a/tests/components/camera/test_component_iterator_camera.cpp b/tests/components/camera/test_component_iterator_camera.cpp new file mode 100644 index 0000000000..efd8261554 --- /dev/null +++ b/tests/components/camera/test_component_iterator_camera.cpp @@ -0,0 +1,79 @@ +#include + +#include "esphome/core/component_iterator.h" + +#ifdef USE_CAMERA +#include "esphome/components/camera/camera.h" + +namespace esphome::testing { + +class StubCamera : public camera::Camera { + public: + void add_listener(camera::CameraListener *listener) override {} + camera::CameraImageReader *create_image_reader() override { return nullptr; } + void request_image(camera::CameraRequester requester) override {} + void start_stream(camera::CameraRequester requester) override {} + void stop_stream(camera::CameraRequester requester) override {} +}; + +// Iterator that accepts everything except the camera, which can refuse a +// configurable number of times. The CAMERA state is a singleton path +// distinct from process_platform_item_; this pins the same contract: +// a refused camera is re-offered, never skipped. +class CameraRefusingIterator : public ComponentIterator { + public: +// NOLINTBEGIN(bugprone-macro-parentheses) +#define ENTITY_TYPE_(type, singular, plural, count, upper) \ + bool on_##singular(type *obj) override { return true; } +#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \ + ENTITY_TYPE_(type, singular, plural, count, upper) +#include "esphome/core/entity_types.h" +#undef ENTITY_TYPE_ +#undef ENTITY_CONTROLLER_TYPE_ + // NOLINTEND(bugprone-macro-parentheses) + + bool on_camera(camera::Camera *obj) override { + this->camera_calls++; + if (this->camera_refusals > 0) { + this->camera_refusals--; + return false; + } + return true; + } + + int camera_calls{0}; + int camera_refusals{0}; +}; + +// Far above the fixed number of iterator states +static constexpr size_t BIG_BUDGET = 1000; + +class ComponentIteratorCameraTest : public ::testing::Test { + protected: + void SetUp() override { + // Constructing a Camera installs the process-wide singleton + static StubCamera stub_camera; + ASSERT_EQ(camera::Camera::instance(), &stub_camera); + } +}; + +TEST_F(ComponentIteratorCameraTest, RefusedCameraIsReofferedNotSkipped) { + CameraRefusingIterator it; + it.camera_refusals = 2; + it.begin(); + // Runs until the camera refuses, which stops the pass + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.camera_calls, 1); + EXPECT_FALSE(it.completed()); + // The camera is re-offered once per call, not skipped + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.camera_calls, 2); + EXPECT_FALSE(it.completed()); + // Once accepted, the iteration completes + it.try_advance(BIG_BUDGET); + EXPECT_TRUE(it.completed()); + EXPECT_EQ(it.camera_calls, 3); +} + +} // namespace esphome::testing +#endif // USE_CAMERA diff --git a/tests/components/core/benchmark.yaml b/tests/components/core/benchmark.yaml new file mode 100644 index 0000000000..063ac41eff --- /dev/null +++ b/tests/components/core/benchmark.yaml @@ -0,0 +1,11 @@ +# Pulls in sensor so entity iteration paths compile (USE_SENSOR); +# tests register their own instances. Plain yaml.safe_load, no ESPHome tags. +# An alphabetically-earlier component's sensor: block shadows this one in +# combined builds; the tests' sensor-count ASSERT catches a capacity drop. +sensor: + - platform: template + id: bench_sensor_a + name: "Bench A" + - platform: template + id: bench_sensor_b + name: "Bench B" diff --git a/tests/components/core/test_component_iterator.cpp b/tests/components/core/test_component_iterator.cpp new file mode 100644 index 0000000000..03d467c920 --- /dev/null +++ b/tests/components/core/test_component_iterator.cpp @@ -0,0 +1,195 @@ +#include + +#include "esphome/core/component_iterator.h" + +#ifdef USE_SENSOR +#include "esphome/components/sensor/sensor.h" +#include "esphome/core/application.h" +#endif + +namespace esphome::testing { + +// Iterator whose begin/end callbacks can refuse a configurable number of +// times; all entity callbacks accept (any registered entities are accepted). +class RefusingIterator : public ComponentIterator { + public: +// NOLINTBEGIN(bugprone-macro-parentheses) +#define ENTITY_TYPE_(type, singular, plural, count, upper) \ + bool on_##singular(type *obj) override { return true; } +#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \ + ENTITY_TYPE_(type, singular, plural, count, upper) +#include "esphome/core/entity_types.h" +#undef ENTITY_TYPE_ +#undef ENTITY_CONTROLLER_TYPE_ + // NOLINTEND(bugprone-macro-parentheses) + + bool on_begin() override { return step(this->begin_calls, this->begin_refusals); } + bool on_end() override { return step(this->end_calls, this->end_refusals); } + + int begin_calls{0}; + int end_calls{0}; + int begin_refusals{0}; + int end_refusals{0}; + + protected: + static bool step(int &calls, int &refusals) { + calls++; + if (refusals > 0) { + refusals--; + return false; + } + return true; + } +}; + +// Far above the fixed number of iterator states +static constexpr size_t BIG_BUDGET = 1000; + +TEST(ComponentIterator, NotRunningMakesNoProgress) { + RefusingIterator it; + it.try_advance(BIG_BUDGET); + EXPECT_TRUE(it.completed()); + EXPECT_EQ(it.begin_calls, 0); + EXPECT_EQ(it.end_calls, 0); +} + +TEST(ComponentIterator, CompletesInOneCallWithoutRefusals) { + RefusingIterator it; + it.begin(); + it.try_advance(BIG_BUDGET); + EXPECT_TRUE(it.completed()); + EXPECT_EQ(it.begin_calls, 1); + EXPECT_EQ(it.end_calls, 1); +} + +TEST(ComponentIterator, StepBudgetIsHonored) { + RefusingIterator it; + it.begin(); + it.try_advance(1); + EXPECT_EQ(it.begin_calls, 1); + EXPECT_EQ(it.end_calls, 0); + EXPECT_FALSE(it.completed()); +} + +TEST(ComponentIterator, RefusedStepStopsBatchAndRetriesSameStep) { + RefusingIterator it; + it.end_refusals = 3; + it.begin(); + // First call runs until the refused end step, which stops the pass + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.end_calls, 1); + EXPECT_FALSE(it.completed()); + // The refused step is retried once per call, not skipped + it.try_advance(BIG_BUDGET); + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.end_calls, 3); + EXPECT_FALSE(it.completed()); + // Once accepted, the iteration completes + it.try_advance(BIG_BUDGET); + EXPECT_TRUE(it.completed()); + EXPECT_EQ(it.end_calls, 4); +} + +TEST(ComponentIterator, RefusedBeginStopsBatchAndRetries) { + RefusingIterator it; + it.begin_refusals = 2; + it.begin(); + it.try_advance(BIG_BUDGET); + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.begin_calls, 2); + EXPECT_FALSE(it.completed()); + it.try_advance(BIG_BUDGET); + EXPECT_TRUE(it.completed()); + EXPECT_EQ(it.begin_calls, 3); +} + +// The deprecated advance() wrapper must keep the legacy once-per-loop +// pattern working during the deprecation window. +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" +TEST(ComponentIterator, DeprecatedAdvanceKeepsLegacyPatternWorking) { + RefusingIterator it; + it.end_refusals = 2; + it.begin(); + size_t guard = 0; + while (!it.completed() && guard++ < BIG_BUDGET) { + it.advance(); + } + EXPECT_TRUE(it.completed()); + // Two refused end steps were retried, then accepted + EXPECT_EQ(it.end_calls, 3); +} +#pragma GCC diagnostic pop + +#ifdef USE_SENSOR +// Iterator whose sensor callback can refuse or yield; pins the per-item +// contract: a refused item is re-offered with at_ unchanged, never skipped. +class ItemRefusingIterator : public RefusingIterator { + public: + bool on_sensor(sensor::Sensor *obj) override { + this->last_sensor = obj; + if (!step(this->sensor_calls, this->sensor_refusals)) + return false; + if (this->yield_on_sensor) + this->yield_after_step_(); + return true; + } + sensor::Sensor *last_sensor{nullptr}; + int sensor_calls{0}; + int sensor_refusals{0}; + bool yield_on_sensor{false}; +}; + +class ComponentIteratorSensorTest : public ::testing::Test { + protected: + void SetUp() override { + static sensor::Sensor sensor_a; + static sensor::Sensor sensor_b; + static bool registered = false; + if (!registered) { + App.register_sensor(&sensor_a); + App.register_sensor(&sensor_b); + registered = true; + } + // StaticVector drops silently when full; fail the fixture, not the contract + ASSERT_EQ(App.get_sensors().size(), 2u) << "benchmark.yaml sensor count too small"; + } +}; + +TEST_F(ComponentIteratorSensorTest, RefusedItemIsReofferedNotSkipped) { + ItemRefusingIterator it; + it.sensor_refusals = 2; + it.begin(); + // Runs until the first sensor refuses + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.sensor_calls, 1); + EXPECT_FALSE(it.completed()); + // The refused item is re-offered, not skipped + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.sensor_calls, 2); + sensor::Sensor *refused = it.last_sensor; + // Once accepted, iteration continues through the second sensor to the end + it.try_advance(BIG_BUDGET); + EXPECT_TRUE(it.completed()); + EXPECT_NE(it.last_sensor, refused); + EXPECT_EQ(it.sensor_calls, 4); +} + +TEST_F(ComponentIteratorSensorTest, YieldAfterStepEndsPassAndResumes) { + ItemRefusingIterator it; + it.yield_on_sensor = true; + it.begin(); + // The pass ends right after the first sensor despite a big budget + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.sensor_calls, 1); + EXPECT_FALSE(it.completed()); + // The next pass ends after the second sensor + it.try_advance(BIG_BUDGET); + EXPECT_EQ(it.sensor_calls, 2); + // Remaining states then run to completion in one pass + it.try_advance(BIG_BUDGET); + EXPECT_TRUE(it.completed()); +} +#endif // USE_SENSOR + +} // namespace esphome::testing diff --git a/tests/integration/README.md b/tests/integration/README.md index 44d9e0d644..790d9a3a11 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -7,6 +7,7 @@ This directory contains end-to-end integration tests for ESPHome, focusing on te - `conftest.py` - Common fixtures and utilities - `const.py` - Constants used throughout the integration tests - `types.py` - Type definitions for fixtures and functions +- `raw_api_client.py` - Minimal plaintext api client whose reads happen only on request (for backpressure tests) - `state_utils.py` - State handling utilities (e.g., `InitialStateHelper`, `find_entity`, `require_entity`) - `fixtures/` - YAML configuration files for tests - `test_*.py` - Individual test files @@ -347,6 +348,7 @@ Create C++ components in `fixtures/external_components/` for: - Custom entity behaviors - Scheduler testing - Memory management tests +- Deterministic network backpressure (`sndbuf_pin_component` pins socket send buffers; assert on its log line to prove the pin took effect) ##### Log Line Monitoring ```python diff --git a/tests/integration/fixtures/api_list_entities_backpressure.yaml b/tests/integration/fixtures/api_list_entities_backpressure.yaml new file mode 100644 index 0000000000..1e53af7b75 --- /dev/null +++ b/tests/integration/fixtures/api_list_entities_backpressure.yaml @@ -0,0 +1,23 @@ +esphome: + name: api-backpressure-test + +host: + +api: + # Smallest queue so a non-draining client blocks the send path quickly + max_send_queue: 1 + actions: +# GENERATED_ACTIONS + +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 diff --git a/tests/integration/fixtures/external_components/sndbuf_pin_component/__init__.py b/tests/integration/fixtures/external_components/sndbuf_pin_component/__init__.py new file mode 100644 index 0000000000..06dda5d1ac --- /dev/null +++ b/tests/integration/fixtures/external_components/sndbuf_pin_component/__init__.py @@ -0,0 +1,20 @@ +import esphome.codegen as cg +import esphome.config_validation as cv +from esphome.const import CONF_BUFFER_SIZE, CONF_ID + +DEPENDENCIES = ["api"] + +sndbuf_pin_ns = cg.esphome_ns.namespace("sndbuf_pin") +SndbufPinComponent = sndbuf_pin_ns.class_("SndbufPinComponent", cg.Component) + +CONFIG_SCHEMA = cv.Schema( + { + cv.GenerateID(): cv.declare_id(SndbufPinComponent), + cv.Required(CONF_BUFFER_SIZE): cv.int_range(min=1), + } +).extend(cv.COMPONENT_SCHEMA) + + +async def to_code(config): + var = cg.new_Pvariable(config[CONF_ID], config[CONF_BUFFER_SIZE]) + await cg.register_component(var, config) diff --git a/tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.cpp b/tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.cpp new file mode 100644 index 0000000000..430c5075c7 --- /dev/null +++ b/tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.cpp @@ -0,0 +1,55 @@ +#include "sndbuf_pin_component.h" + +#include +#include +#include + +#include "esphome/components/api/api_server.h" +#include "esphome/core/log.h" + +namespace esphome::sndbuf_pin { + +static const char *const TAG = "sndbuf_pin"; + +// Skip stdio; scan the low fd range where the listeners land +static constexpr int FIRST_USER_FD = 3; +static constexpr int MAX_FD_SCAN = 128; + +void SndbufPinComponent::setup() { + int pinned = 0; + for (int fd = FIRST_USER_FD; fd < MAX_FD_SCAN; fd++) { + int type = 0; + socklen_t len = sizeof(type); + if (::getsockopt(fd, SOL_SOCKET, SO_TYPE, &type, &len) != 0 || type != SOCK_STREAM) + continue; + struct sockaddr_in addr {}; + socklen_t addr_len = sizeof(addr); + if (::getsockname(fd, reinterpret_cast(&addr), &addr_len) != 0) { + ESP_LOGW(TAG, "fd %d: getsockname failed, errno %d", fd, errno); + continue; + } + if (ntohs(addr.sin_port) != api::global_api_server->get_port()) + continue; + if (::setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &this->buffer_size_, sizeof(this->buffer_size_)) != 0) { + ESP_LOGW(TAG, "fd %d: SO_SNDBUF pin failed, errno %d", fd, errno); + continue; + } + int applied = 0; + len = sizeof(applied); + if (::getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &applied, &len) != 0 || applied < this->buffer_size_) { + // Linux doubles the requested value; anything below it means clamped + ESP_LOGW(TAG, "fd %d: SO_SNDBUF readback %d below requested %d", fd, applied, this->buffer_size_); + continue; + } + // Tests assert on this line; accepted sockets inherit the pinned size + ESP_LOGD(TAG, "fd %d port %d: SO_SNDBUF pinned to %d (effective %d)", fd, ntohs(addr.sin_port), this->buffer_size_, + applied); + pinned++; + } + if (pinned == 0) { + ESP_LOGE(TAG, "api listener socket was not pinned"); + this->mark_failed(); + } +} + +} // namespace esphome::sndbuf_pin diff --git a/tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.h b/tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.h new file mode 100644 index 0000000000..b0af226d14 --- /dev/null +++ b/tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.h @@ -0,0 +1,21 @@ +#pragma once + +#include "esphome/core/component.h" + +namespace esphome::sndbuf_pin { + +// Test-only (host): pins SO_SNDBUF on every open TCP socket so integration +// tests get deterministic backpressure; an explicit SO_SNDBUF also disables +// kernel autotuning, and accepted sockets inherit it from the listener. +class SndbufPinComponent : public Component { + public: + explicit SndbufPinComponent(int buffer_size) : buffer_size_(buffer_size) {} + void setup() override; + // After the api server so its listening socket exists + float get_setup_priority() const override { return setup_priority::LATE; } + + protected: + int buffer_size_; +}; + +} // namespace esphome::sndbuf_pin diff --git a/tests/integration/raw_api_client.py b/tests/integration/raw_api_client.py new file mode 100644 index 0000000000..1dbe40933c --- /dev/null +++ b/tests/integration/raw_api_client.py @@ -0,0 +1,148 @@ +"""Minimal plaintext native-api client over a raw socket. + +Reads only when told to, so tests control when the TCP pipe backs up toward +the device; payloads are skipped and only message types are counted. +""" + +from __future__ import annotations + +import asyncio +from collections import Counter +import socket +from typing import Self + +from aioesphomeapi import api_pb2 +import aioesphomeapi.core as api_core +from google.protobuf import message + +from .const import LOCALHOST + +# Message type ids are protocol constants; derive them from aioesphomeapi so +# they cannot drift from the client library in use. +MESSAGE_TYPE_OF = {cls: num for num, cls in api_core.MESSAGE_TYPE_TO_PROTO.items()} + +_READ_CHUNK = 4096 + + +def encode_varint(value: int) -> bytes: + out = bytearray() + while True: + byte = value & 0x7F + value >>= 7 + if value: + out.append(byte | 0x80) + else: + out.append(byte) + return bytes(out) + + +def decode_varint(buf: bytearray, pos: int) -> tuple[int, int] | None: + """Decode one varint at pos; return (value, new_pos) or None if short.""" + value = shift = 0 + while pos < len(buf): + byte = buf[pos] + pos += 1 + value |= (byte & 0x7F) << shift + if not byte & 0x80: + return value, pos + shift += 7 + return None + + +def encode_frame(msg_type: int, payload: bytes) -> bytes: + """Encode one plaintext api frame: 0x00, payload length, message type.""" + return b"\x00" + encode_varint(len(payload)) + encode_varint(msg_type) + payload + + +class FrameParser: + """Incremental parser for the plaintext api frame stream.""" + + def __init__(self) -> None: + self._buf = bytearray() + + def feed(self, data: bytes) -> list[int]: + self._buf.extend(data) + types: list[int] = [] + while (msg_type := self._try_parse()) is not None: + types.append(msg_type) + return types + + def _try_parse(self) -> int | None: + buf = self._buf + if not buf: + return None + assert buf[0] == 0, f"expected plaintext frame, got indicator {buf[0]}" + if (size_decoded := decode_varint(buf, 1)) is None: + return None + size, pos = size_decoded + if (type_decoded := decode_varint(buf, pos)) is None: + return None + msg_type, pos = type_decoded + if len(buf) - pos < size: + return None + del buf[: pos + size] + return msg_type + + +class RawApiClient: + """Plaintext api client whose reads happen only on request.""" + + def __init__(self, port: int, recv_buffer_size: int | None = None) -> None: + self._port = port + self._parser = FrameParser() + self.bytes_received = 0 + self.frame_counts: Counter[int] = Counter() + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + if recv_buffer_size is not None: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, recv_buffer_size) + # Kernels may round up (Linux doubles) but must not clamp below + applied = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) + assert applied >= recv_buffer_size, ( + f"SO_RCVBUF clamped to {applied}, requested {recv_buffer_size}" + ) + sock.setblocking(False) + except Exception: + sock.close() + raise + self._sock = sock + + async def __aenter__(self) -> Self: + return self + + async def __aexit__(self, *exc_info: object) -> None: + self.close() + + async def connect(self, client_info: str = "raw-api-client") -> None: + """Connect and complete the Hello handshake (no auth step since 2026.1.0).""" + loop = asyncio.get_running_loop() + await loop.sock_connect(self._sock, (LOCALHOST, self._port)) + hello = api_pb2.HelloRequest() + hello.client_info = client_info + hello.api_version_major = 1 + hello.api_version_minor = 10 + await self.send_message(hello) + await self.read_until_frame(MESSAGE_TYPE_OF[api_pb2.HelloResponse]) + + async def send_message(self, msg: message.Message) -> None: + loop = asyncio.get_running_loop() + await loop.sock_sendall( + self._sock, + encode_frame(MESSAGE_TYPE_OF[type(msg)], msg.SerializeToString()), + ) + + async def read_until_frame(self, msg_type: int, timeout: float = 10.0) -> None: + """Read until at least one frame of msg_type has been received.""" + loop = asyncio.get_running_loop() + + async def _read_loop() -> None: + while not self.frame_counts[msg_type]: + data = await loop.sock_recv(self._sock, _READ_CHUNK) + assert data, "server closed the connection unexpectedly" + self.bytes_received += len(data) + self.frame_counts.update(self._parser.feed(data)) + + await asyncio.wait_for(_read_loop(), timeout) + + def close(self) -> None: + self._sock.close() diff --git a/tests/integration/test_api_list_entities_backpressure.py b/tests/integration/test_api_list_entities_backpressure.py new file mode 100644 index 0000000000..9df6c7a298 --- /dev/null +++ b/tests/integration/test_api_list_entities_backpressure.py @@ -0,0 +1,110 @@ +"""A client that stops reading the entity listing must not starve other clients. + +Service responses are sent directly (not via the deferred batch), so a full +TCP pipe makes the send path refuse; the drive loop now lives in +try_advance(), which stops on refusal instead of retrying forever. Not a +before/after regression test: pre-fix builds survive here because the +refusal path yields and pumps the socket each retry. + +The sndbuf_pin_component fixture pins the device's send buffers so the pipe +fills deterministically regardless of kernel autotuning; the test waits for +its log line before proceeding. +""" + +from __future__ import annotations + +import asyncio + +from aioesphomeapi import api_pb2 +import pytest + +from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient +from .types import APIClientConnectedFactory, RunCompiledFunction + +SERVICES_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesServicesResponse] +LIST_DONE_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesDoneResponse] + +# Both ends of the pipe are pinned small; only tens of KB fit in the kernel +RECV_BUFFER_SIZE = 4096 +SERVER_SNDBUF = 8192 # substituted into the fixture yaml +# Logged by the sndbuf_pin_component fixture when it pins a socket +SNDBUF_PIN_LOG = "SO_SNDBUF pinned to" +# One response (~6.4 KB) must stay smaller than the pinned send buffer; an +# oversized message parks in the overflow buffer and reports as sent. +ARGS_PER_SERVICE = 8 +ARG_NAME_LEN = 800 +# ~160 KB listing versus a tens-of-KB pipe guarantees a mid-services block +NUM_SERVICES = 25 +assert ARGS_PER_SERVICE * ARG_NAME_LEN < SERVER_SNDBUF +# The pipe fills in well under a second +STALL_SECONDS = 0.5 +# Well above pipe capacity, well below the listing size +MIN_DRAINED_BYTES = 60_000 + + +def _generated_actions() -> str: + """Build the api actions block: services with long argument names.""" + lines: list[str] = [] + for i in range(NUM_SERVICES): + lines.append(f" - action: backpressure_service_{i:04d}") + lines.append(" variables:") + for j in range(ARGS_PER_SERVICE): + prefix = f"arg_{i:04d}_{j:02d}_" + lines.append( + f" {prefix}{'x' * (ARG_NAME_LEN - len(prefix))}: string" + ) + lines.append(" then:") + lines.append(" - logger.log: service called") + return "\n".join(lines) + + +@pytest.mark.asyncio +async def test_api_list_entities_backpressure( + yaml_config: str, + run_compiled: RunCompiledFunction, + api_client_connected: APIClientConnectedFactory, + unused_tcp_port: int, +) -> None: + """A stalled reader mid-services must not block other api clients.""" + assert "# GENERATED_ACTIONS" in yaml_config + config = yaml_config.replace("# GENERATED_ACTIONS", _generated_actions()) + config = config.replace("SERVER_SNDBUF", str(SERVER_SNDBUF)) + + pin_applied = asyncio.Event() + + def _on_log_line(line: str) -> None: + if SNDBUF_PIN_LOG in line: + pin_applied.set() + + async with run_compiled(config, line_callback=_on_log_line): + # Fails loudly if the pin never applied + await asyncio.wait_for(pin_applied.wait(), 10) + + async with RawApiClient( + unused_tcp_port, recv_buffer_size=RECV_BUFFER_SIZE + ) as stalled: + await stalled.connect(client_info="backpressure-stall-client") + await stalled.send_message(api_pb2.ListEntitiesRequest()) + # The client now stops reading entirely. + + # Let the server run against the full pipe + await asyncio.sleep(STALL_SECONDS) + + # Other clients must still be served while the first is blocked + async with api_client_connected(timeout=20) as client: + device_info = await asyncio.wait_for(client.device_info(), 20) + assert device_info.name == "api-backpressure-test" + _, services = await asyncio.wait_for( + client.list_entities_services(), 30 + ) + assert len(services) == NUM_SERVICES + + # Fixture-size guard: the listing must dwarf the pinned pipe + before = stalled.bytes_received + await stalled.read_until_frame(LIST_DONE_RESPONSE, timeout=60) + drained = stalled.bytes_received - before + assert drained > MIN_DRAINED_BYTES, ( + f"only {drained} bytes drained; the listing never backed up" + ) + assert stalled.frame_counts[SERVICES_RESPONSE] == NUM_SERVICES + assert stalled.frame_counts[LIST_DONE_RESPONSE] == 1