From fb52e1679dad530ff56c0edeb53713baa8441092 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 14 Oct 2026 10:37:26 -1000 Subject: [PATCH] Grow the AsyncMqttClient topic buffer at connect instead of rejecting long topics --- esphome/components/mqtt/__init__.py | 110 -------------- .../components/mqtt/mqtt_backend_esp8266.h | 2 + .../components/mqtt/mqtt_backend_libretiny.h | 2 + esphome/components/mqtt/mqtt_client.cpp | 19 +++ esphome/components/mqtt/mqtt_client.h | 4 + .../components/test_mqtt_topic_length.py | 140 ------------------ 6 files changed, 27 insertions(+), 250 deletions(-) delete mode 100644 tests/unit_tests/components/test_mqtt_topic_length.py diff --git a/esphome/components/mqtt/__init__.py b/esphome/components/mqtt/__init__.py index 24ea300ceeb..fe426827ce6 100644 --- a/esphome/components/mqtt/__init__.py +++ b/esphome/components/mqtt/__init__.py @@ -1,5 +1,3 @@ -from collections.abc import Iterator - from esphome import automation import esphome.codegen as cg from esphome.components import logger, socket @@ -25,7 +23,6 @@ from esphome.const import ( CONF_CLIENT_ID, CONF_COMMAND_RETAIN, CONF_COMMAND_TOPIC, - CONF_DEVICE_ID, CONF_DISCOVER_IP, CONF_DISCOVERY, CONF_DISCOVERY_OBJECT_ID_GENERATOR, @@ -33,17 +30,10 @@ from esphome.const import ( CONF_DISCOVERY_RETAIN, CONF_DISCOVERY_UNIQUE_ID_GENERATOR, CONF_ENABLE_ON_BOOT, - CONF_ESPHOME, CONF_ID, - CONF_INTERNAL, CONF_KEEPALIVE, CONF_LEVEL, CONF_LOG_TOPIC, - CONF_MODE_COMMAND_TOPIC, - CONF_MQTT as CONF_MQTT, - CONF_MQTT_ID, - CONF_NAME, - CONF_NAME_ADD_MAC_SUFFIX, CONF_ON_CONNECT, CONF_ON_DISCONNECT, CONF_ON_JSON_MESSAGE, @@ -52,7 +42,6 @@ from esphome.const import ( CONF_PAYLOAD, CONF_PAYLOAD_AVAILABLE, CONF_PAYLOAD_NOT_AVAILABLE, - CONF_PLATFORM, CONF_PORT, CONF_PUBLISH_NAN_AS_NONE, CONF_QOS, @@ -64,7 +53,6 @@ from esphome.const import ( CONF_SUBSCRIBE_QOS, CONF_TOPIC, CONF_TOPIC_PREFIX, - CONF_TYPE, CONF_USE_ABBREVIATIONS, CONF_USERNAME, CONF_WILL_MESSAGE, @@ -76,8 +64,6 @@ from esphome.const import ( PlatformFramework, ) from esphome.core import CORE, CoroPriority, coroutine_with_priority -from esphome.core.entity_helpers import get_base_entity_object_id -import esphome.final_validate as fv from esphome.types import ConfigType DEPENDENCIES = ["network"] @@ -322,102 +308,6 @@ CONFIG_SCHEMA = cv.All( _consume_mqtt_sockets, ) -# AsyncMqttClient, the ESP8266 and LibreTiny backend, drops any inbound message whose -# topic is longer than this (its default maxTopicLength) -ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN = 128 -# name_add_mac_suffix appends "-" plus 6 hex digits to the name at runtime -MAC_SUFFIX = "-xxxxxx" -# Entity domains whose MQTT component always subscribes to its command topic -_COMMAND_TOPIC_DOMAINS = ( - "alarm_control_panel", - "button", - "climate", - "cover", - "datetime", - "fan", - "light", - "lock", - "number", - "select", - "switch", - "text", - "update", - "valve", -) - - -def _entity_command_topic( - domain: str, config: ConfigType, topic_prefix: str, mac_suffix: bool -) -> str | None: - """Return the topic an entity's MQTT component always subscribes to, if known.""" - if domain == "climate": - suffix, custom_key = "mode/command", CONF_MODE_COMMAND_TOPIC - else: - suffix, custom_key = "command", CONF_COMMAND_TOPIC - if (custom := config.get(custom_key)) is not None: - # A lambda topic is only known at runtime - return custom if isinstance(custom, str) else None - # Mirrors MQTTComponent::compute_is_internal_(): internal components never subscribe - if CONF_STATE_TOPIC in config: - internal = config[CONF_STATE_TOPIC] == "" - else: - internal = config.get(CONF_INTERNAL, False) - if internal: - return None - device = config.get(CONF_DEVICE_ID) - object_id = get_base_entity_object_id( - config.get(CONF_NAME) or "", CORE.friendly_name, device.id if device else None - ) - if mac_suffix and not config.get(CONF_NAME): - object_id += MAC_SUFFIX - component_type = config[CONF_TYPE].lower() if domain == "datetime" else domain - return f"{topic_prefix}/{component_type}/{object_id}/{suffix}" - - -def _subscribed_topics(config: ConfigType) -> Iterator[str]: - """Yield every subscribed topic that is known at build time.""" - full_config = fv.full_config.get() - mac_suffix = full_config[CONF_ESPHOME].get(CONF_NAME_ADD_MAC_SUFFIX, False) - if topic_prefix := config[CONF_TOPIC_PREFIX]: - if mac_suffix and topic_prefix == CORE.name: - topic_prefix += MAC_SUFFIX - for domain in _COMMAND_TOPIC_DOMAINS: - for entity in full_config.get(domain) or []: - if CONF_MQTT_ID in entity and ( - topic := _entity_command_topic( - domain, entity, topic_prefix, mac_suffix - ) - ): - yield topic - for key in (CONF_ON_MESSAGE, CONF_ON_JSON_MESSAGE): - for trigger in config.get(key, []): - yield trigger[CONF_TOPIC] - for domain in ("sensor", "text_sensor"): - for entity in full_config.get(domain) or []: - if entity[CONF_PLATFORM] == "mqtt_subscribe": - yield entity[CONF_TOPIC] - - -def _validate_topic_lengths(config: ConfigType) -> ConfigType: - """Reject topics that the ESP8266 and LibreTiny MQTT client would silently drop.""" - if not (CORE.is_esp8266 or CORE.is_libretiny): - return config - for topic in _subscribed_topics(config): - # A wildcard filter matches topics of any length, so only literals are checked - if ( - len(topic) > ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN - and "+" not in topic - and "#" not in topic - ): - raise cv.Invalid( - f"MQTT topic '{topic}' is {len(topic)} characters; the MQTT client on " - f"this platform drops topics longer than {ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN}" - ) - return config - - -FINAL_VALIDATE_SCHEMA = _validate_topic_lengths - def exp_mqtt_message(config): if config is None: diff --git a/esphome/components/mqtt/mqtt_backend_esp8266.h b/esphome/components/mqtt/mqtt_backend_esp8266.h index 0bf5b510a4c..5b73f201f40 100644 --- a/esphome/components/mqtt/mqtt_backend_esp8266.h +++ b/esphome/components/mqtt/mqtt_backend_esp8266.h @@ -49,6 +49,8 @@ class MQTTBackendESP8266 final : public MQTTBackend { bool connected() const final { return mqtt_client_.connected(); } void connect() final { mqtt_client_.connect(); } + /// Only call while disconnected: the parser writes inbound topics into this buffer. + void set_max_topic_length(uint16_t length) { mqtt_client_.setMaxTopicLength(length); } void disconnect() final { mqtt_client_.disconnect(true); } bool subscribe(const char *topic, uint8_t qos) final { return mqtt_client_.subscribe(topic, qos) != 0; } bool unsubscribe(const char *topic) final { return mqtt_client_.unsubscribe(topic) != 0; } diff --git a/esphome/components/mqtt/mqtt_backend_libretiny.h b/esphome/components/mqtt/mqtt_backend_libretiny.h index 5fa3406193c..8fc890ef3ce 100644 --- a/esphome/components/mqtt/mqtt_backend_libretiny.h +++ b/esphome/components/mqtt/mqtt_backend_libretiny.h @@ -49,6 +49,8 @@ class MQTTBackendLibreTiny final : public MQTTBackend { bool connected() const final { return mqtt_client_.connected(); } void connect() final { mqtt_client_.connect(); } + /// Only call while disconnected: the parser writes inbound topics into this buffer. + void set_max_topic_length(uint16_t length) { mqtt_client_.setMaxTopicLength(length); } void disconnect() final { mqtt_client_.disconnect(true); } bool subscribe(const char *topic, uint8_t qos) final { return mqtt_client_.subscribe(topic, qos) != 0; } bool unsubscribe(const char *topic) final { return mqtt_client_.unsubscribe(topic) != 0; } diff --git a/esphome/components/mqtt/mqtt_client.cpp b/esphome/components/mqtt/mqtt_client.cpp index 803f486333a..50d6b5ffa69 100644 --- a/esphome/components/mqtt/mqtt_client.cpp +++ b/esphome/components/mqtt/mqtt_client.cpp @@ -2,6 +2,7 @@ #ifdef USE_MQTT +#include #include #include "esphome/components/network/util.h" #include "esphome/core/application.h" @@ -312,6 +313,17 @@ void MQTTClientComponent::start_connect_() { // Force disconnect first this->mqtt_backend_.disconnect(); +#if defined(USE_ESP8266) || defined(USE_LIBRETINY) + // Grow the topic buffer to the longest subscription. Only here, while no connection parses into it. + size_t longest = this->max_topic_length_; + for (const auto &subscription : this->subscriptions_) + longest = std::max(longest, subscription.topic.size()); + if (longest > this->max_topic_length_ && longest <= UINT16_MAX) { + this->max_topic_length_ = longest; + this->mqtt_backend_.set_max_topic_length(longest); + } +#endif + this->mqtt_backend_.set_client_id(this->credentials_.client_id.c_str()); this->mqtt_backend_.set_clean_session(this->credentials_.clean_session); const char *username = nullptr; @@ -472,6 +484,13 @@ void MQTTClientComponent::add_subscription_(std::string &&topic, mqtt_subscripti .callback = callback, .resubscribe_timeout = 0, }); +#if defined(USE_ESP8266) || defined(USE_LIBRETINY) + if (subscription.topic.size() > this->max_topic_length_ && + (this->state_ == MQTT_CLIENT_CONNECTING || this->state_ == MQTT_CLIENT_CONNECTED)) { + ESP_LOGW(TAG, "Topic '%s' is longer than %u characters; its messages arrive after the next reconnect", + subscription.topic.c_str(), this->max_topic_length_); + } +#endif this->resubscribe_subscription_(&subscription); } diff --git a/esphome/components/mqtt/mqtt_client.h b/esphome/components/mqtt/mqtt_client.h index 30edb812975..8dcda8c4366 100644 --- a/esphome/components/mqtt/mqtt_client.h +++ b/esphome/components/mqtt/mqtt_client.h @@ -370,6 +370,10 @@ class MQTTClientComponent final : public Component { bool wait_for_connection_{false}; bool dispatching_{false}; bool removed_while_dispatching_{false}; +#if defined(USE_ESP8266) || defined(USE_LIBRETINY) + // AsyncMqttClient drops inbound topics longer than this; it starts at the library's default + uint16_t max_topic_length_{128}; +#endif }; extern MQTTClientComponent *global_mqtt_client; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables) diff --git a/tests/unit_tests/components/test_mqtt_topic_length.py b/tests/unit_tests/components/test_mqtt_topic_length.py deleted file mode 100644 index fc36e998b67..00000000000 --- a/tests/unit_tests/components/test_mqtt_topic_length.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Tests for the mqtt check that subscribed topics fit the AsyncMqttClient limit.""" - -import pytest - -from esphome.components.mqtt import ( - ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN, - _validate_topic_lengths, -) -import esphome.config_validation as cv -from esphome.const import ( - CONF_ESPHOME, - CONF_MQTT_ID, - CONF_NAME, - CONF_NAME_ADD_MAC_SUFFIX, - KEY_CORE, - KEY_TARGET_PLATFORM, - PLATFORM_BK72XX, - PLATFORM_ESP32, - PLATFORM_ESP8266, -) -from esphome.core import CORE, Lambda -import esphome.final_validate as fv -from esphome.types import ConfigType - -# "dev/switch//command" is 19 characters around the object id -FITS = "a" * (ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN - 19) -TOO_LONG = FITS + "a" -LONG_LITERAL = "t" * (ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN + 1) - - -def _switch(name: str, **extra) -> dict: - return {"switch": [{CONF_NAME: name, CONF_MQTT_ID: "mqtt_switch", **extra}]} - - -def _run( - monkeypatch: pytest.MonkeyPatch, - full_config: dict, - mqtt: ConfigType | None = None, - platform: str = PLATFORM_ESP8266, -) -> ConfigType: - monkeypatch.setitem(CORE.data, KEY_CORE, {KEY_TARGET_PLATFORM: platform}) - monkeypatch.setattr(CORE, "name", "dev") - monkeypatch.setattr(CORE, "friendly_name", None) - config = {"topic_prefix": "dev", **(mqtt or {})} - token = fv.full_config.set({CONF_ESPHOME: {}, **full_config}) - try: - return _validate_topic_lengths(config) - finally: - fv.full_config.reset(token) - - -@pytest.mark.parametrize( - ("full_config", "mqtt", "platform"), - [ - pytest.param(_switch(TOO_LONG), None, PLATFORM_ESP8266, id="command_topic"), - pytest.param(_switch(TOO_LONG), None, PLATFORM_BK72XX, id="libretiny"), - pytest.param( - {CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: True}, **_switch(FITS)}, - None, - PLATFORM_ESP8266, - id="mac_suffix", - ), - pytest.param( - _switch("s", command_topic=LONG_LITERAL), - None, - PLATFORM_ESP8266, - id="custom_command_topic", - ), - pytest.param( - { - "climate": [ - # "dev/climate//mode/command" is 25 characters around it - {CONF_NAME: "a" * 104, CONF_MQTT_ID: "mqtt_climate"} - ] - }, - None, - PLATFORM_ESP8266, - id="climate_mode_command", - ), - pytest.param( - {}, - {"on_message": [{"topic": LONG_LITERAL}]}, - PLATFORM_ESP8266, - id="trigger", - ), - pytest.param( - {"sensor": [{"platform": "mqtt_subscribe", "topic": LONG_LITERAL}]}, - None, - PLATFORM_ESP8266, - id="mqtt_subscribe", - ), - ], -) -def test_rejected( - monkeypatch: pytest.MonkeyPatch, full_config: dict, mqtt: dict | None, platform: str -) -> None: - with pytest.raises(cv.Invalid, match="characters"): - _run(monkeypatch, full_config, mqtt, platform) - - -@pytest.mark.parametrize( - ("full_config", "mqtt", "platform"), - [ - pytest.param(_switch(FITS), None, PLATFORM_ESP8266, id="at_the_limit"), - pytest.param(_switch(TOO_LONG), None, PLATFORM_ESP32, id="esp32"), - pytest.param( - _switch(TOO_LONG, command_topic="short/cmd"), - None, - PLATFORM_ESP8266, - id="short_custom_topic", - ), - pytest.param( - _switch(TOO_LONG, command_topic=Lambda('return "x";')), - None, - PLATFORM_ESP8266, - id="lambda_topic", - ), - pytest.param( - _switch(TOO_LONG, internal=True), None, PLATFORM_ESP8266, id="internal" - ), - pytest.param( - {"switch": [{CONF_NAME: TOO_LONG}]}, None, PLATFORM_ESP8266, id="no_mqtt" - ), - pytest.param( - {}, - {"on_message": [{"topic": LONG_LITERAL + "/#"}]}, - PLATFORM_ESP8266, - id="wildcard", - ), - ], -) -def test_accepted( - monkeypatch: pytest.MonkeyPatch, full_config: dict, mqtt: dict | None, platform: str -) -> None: - _run(monkeypatch, full_config, mqtt, platform) - - -def test_message_names_the_topic_and_length(monkeypatch: pytest.MonkeyPatch) -> None: - with pytest.raises(cv.Invalid, match=rf"'dev/switch/{TOO_LONG}/command' is 129"): - _run(monkeypatch, _switch(TOO_LONG))