Merge remote-tracking branch 'origin/dev' into esp32-log-v2-idf61

esphome/components/api/api_connection.cpp

@@ -1,4 +1,7 @@
 #include "api_connection.h"
+#if defined(USE_LOGGER) && defined(USE_CRASH_HANDLER)
+#include "esphome/components/logger/logger.h"
+#endif
 #ifdef USE_API
 #include "api_connection_buffer.h"  // for the APIServer-dependent APIConnection inlines
 #ifdef USE_API_NOISE
@@ -451,6 +454,15 @@ bool APIConnection::send_disconnect_response_() {
   DisconnectResponse resp;
   return this->send_message(resp);
 }
+void APIConnection::on_subscribe_logs_request(const SubscribeLogsRequest &msg) {
+  this->flags_.log_subscription = msg.level;
+  if (msg.dump_config)
+    App.schedule_dump_config();
+#if defined(USE_LOGGER) && defined(USE_CRASH_HANDLER)
+  logger::log_crash_to_client();
+#endif
+}
+
 void APIConnection::on_disconnect_response() {
   // Don't close socket here, let APIServer::loop() do it
   // so getpeername() still works for the disconnect trigger

esphome/components/api/api_connection.h

@@ -15,15 +15,6 @@
 #include "subscribe_state.h"
 #include "esphome/core/application.h"
 #include "esphome/core/component.h"
-#ifdef USE_ESP32_CRASH_HANDLER
-#include "esphome/components/esp32/crash_handler.h"
-#endif
-#ifdef USE_RP2_CRASH_HANDLER
-#include "esphome/components/rp2/crash_handler.h"
-#endif
-#ifdef USE_ESP8266_CRASH_HANDLER
-#include "esphome/components/esp8266/crash_handler.h"
-#endif
 #include "esphome/core/entity_base.h"
 #include "esphome/core/log.h"
 #include "esphome/core/string_ref.h"
@@ -298,21 +289,7 @@ class APIConnection final : public APIServerConnectionBase {
       this->begin_iterator_(ActiveIterator::INITIAL_STATE);
     }
   }
-  void on_subscribe_logs_request(const SubscribeLogsRequest &msg) {
-    this->flags_.log_subscription = msg.level;
-    if (msg.dump_config)
-      App.schedule_dump_config();
-#ifdef USE_ESP32_CRASH_HANDLER
-    esp32::crash_handler_log();
-    esp32::crash_handler_clear();
-#endif
-#ifdef USE_RP2_CRASH_HANDLER
-    rp2::crash_handler_log();
-#endif
-#ifdef USE_ESP8266_CRASH_HANDLER
-    esp8266::crash_handler_log();
-#endif
-  }
+  void on_subscribe_logs_request(const SubscribeLogsRequest &msg);
 #ifdef USE_API_HOMEASSISTANT_SERVICES
   void on_subscribe_homeassistant_services_request() { this->flags_.service_call_subscription = true; }
 #endif

esphome/components/esp32/__init__.py

@@ -2920,6 +2920,7 @@ async def _finalize_arduino_aware_flags():
     if not arduino_linked:
         cg.add_build_flag("-Wl,--wrap=esp_panic_handler")
         cg.add_define("USE_ESP32_CRASH_HANDLER")
+        cg.add_define("USE_CRASH_HANDLER")
     elif not CORE.using_arduino:
         cg.add_build_flag("-DUSE_ARDUINO")
 

esphome/components/esp32/crash_handler.cpp

@@ -161,7 +161,7 @@ static bool s_crash_data_valid = false;  // NOLINT(cppcoreguidelines-avoid-non-c
 
 namespace esphome::esp32 {
 
-ESPHOME_LOG_TAG(TAG, "esp32.crash");
+ESPHOME_LOG_TAG(TAG, CRASH_LOG_TAG);
 
 // RAM copy of the build timestamp. The generated constant lives in flash,
 // which the panic handler must not read (cache may be disabled during
@@ -175,7 +175,8 @@ static uint32_t s_current_build_time = static_cast<uint32_t>(ESPHOME_BUILD_TIME)
 
 // Validate the NOINIT record. Runs on every has_data() call; re-running is
 // harmless and the magic is left alone so the record survives an OTA
-// rollback reboot, crash_handler_clear() drops it once an API client has it.
+// rollback reboot, crash_handler_clear() drops it once a log client has it
+// (logger::log_crash_to_client(), for the API and MQTT).
 static void read_crash_data() {
   if (s_raw_crash_data.magic == CRASH_MAGIC && s_raw_crash_data.version == CRASH_DATA_VERSION) {
     s_crash_data_valid = true;

esphome/components/esp32/crash_handler.h

@@ -4,6 +4,9 @@
 
 namespace esphome::esp32 {
 
+/// The tag the crash record is logged under, so a client can check the level in force for it
+inline constexpr const char *CRASH_LOG_TAG = "esp32.crash";
+
 /// Log crash data if a crash was detected on previous boot.
 void crash_handler_log();
 

esphome/components/esp8266/__init__.py

@@ -420,6 +420,7 @@ async def to_code(config: ConfigType) -> None:
     cg.add_define("ESPHOME_VARIANT", "ESP8266")
     cg.add_define(ThreadModel.SINGLE)
     cg.add_define("USE_ESP8266_CRASH_HANDLER")
+    cg.add_define("USE_CRASH_HANDLER")
 
     enable_scanf_float = config.get(CONF_ENABLE_SCANF_FLOAT)
     if enable_scanf_float is None and lambdas_use_scanf_float(CORE.config):

esphome/components/logger/logger.cpp

@@ -142,6 +142,8 @@ void Logger::log_vprintf_(uint8_t level, const char *tag, int line, const __Flas
 }
 #endif  // USE_ESP8266
 
+uint8_t Logger::get_log_level(const char *tag) { return this->level_for(tag); }
+
 inline uint8_t Logger::level_for(const char *tag) {
 #ifdef USE_LOGGER_RUNTIME_TAG_LEVELS
   // Skip the lookup when no per-tag levels are set

esphome/components/logger/logger.h

@@ -178,6 +178,8 @@ class Logger final : public Component {
   void set_log_level(const char *tag, uint8_t log_level);
 #endif
   uint8_t get_log_level() { return this->current_level_; }
+  /// The level in force for a tag: its own when one is set, else the default
+  uint8_t get_log_level(const char *tag);
 
   // ========== INTERNAL METHODS ==========
   // (In most use cases you won't need these)
@@ -511,6 +513,14 @@ class LoggerMessageTrigger final : public Trigger<uint8_t, const char *, const c
 #endif
 };
 
+#ifdef USE_CRASH_HANDLER
+/// Whether a crash record from before this boot is waiting to be reported
+bool crash_log_pending();
+/// Log the previous crash for a client that just subscribed to the logs. On the ESP32 the record
+/// is kept until a client has seen it, so this call also clears it there.
+void log_crash_to_client();
+#endif
+
 }  // namespace esphome::logger
 
 // Platform-specific inline implementations of write_msg_()

esphome/components/logger/logger_esp32.cpp

@@ -162,6 +162,17 @@ const LogString *Logger::get_uart_selection_() {
   }
 }
 
+#ifdef USE_CRASH_HANDLER
+bool crash_log_pending() { return esp32::crash_handler_has_data(); }
+void log_crash_to_client() {
+  // A level that drops the lines would throw the record away unseen; keep it for a client that can see it
+  if (global_logger->get_log_level(esp32::CRASH_LOG_TAG) < ESPHOME_LOG_LEVEL_ERROR)
+    return;
+  esp32::crash_handler_log();
+  esp32::crash_handler_clear();
+}
+#endif
+
 }  // namespace esphome::logger
 
 // The version guard keeps builds that define USE_ESP32_LOG_V2 unconditionally (static analysis) off

esphome/components/logger/logger_esp8266.cpp

@@ -48,6 +48,11 @@ const LogString *Logger::get_uart_selection_() {
 #endif
 }
 
+#ifdef USE_CRASH_HANDLER
+bool crash_log_pending() { return esp8266::crash_handler_has_data(); }
+void log_crash_to_client() { esp8266::crash_handler_log(); }
+#endif
+
 }  // namespace esphome::logger
 
 #if !defined(USE_ESP8266_LOGGER_SERIAL) && !defined(USE_ESP8266_LOGGER_SERIAL1)

esphome/components/logger/logger_rp2.cpp

@@ -49,5 +49,10 @@ const LogString *Logger::get_uart_selection_() {
   }
 }
 
+#ifdef USE_CRASH_HANDLER
+bool crash_log_pending() { return rp2::crash_handler_has_data(); }
+void log_crash_to_client() { rp2::crash_handler_log(); }
+#endif
+
 }  // namespace esphome::logger
 #endif  // USE_RP2

esphome/components/mqtt/mqtt_client.cpp

@@ -16,6 +16,7 @@
 #include "lwip/dns.h"
 #include "lwip/err.h"
 #include "mqtt_component.h"
+#include "mqtt_topic_match.h"
 
 #ifdef USE_API
 #include "esphome/components/api/api_server.h"
@@ -49,6 +50,13 @@ MQTTClientComponent::MQTTClientComponent() {
 }
 
 // Connection
+// Room for "esphome/<kind>/<name>" with the longest kind, "discover"
+static constexpr size_t ESPHOME_TOPIC_BUFFER_SIZE = sizeof("esphome/discover/") + ESPHOME_DEVICE_NAME_MAX_LEN;
+
+static void format_esphome_topic(char (&topic)[ESPHOME_TOPIC_BUFFER_SIZE], const char *kind) {
+  buf_append_printf(topic, sizeof(topic), 0, "esphome/%s/%s", kind, App.get_name().c_str());
+}
+
 void MQTTClientComponent::setup() {
   this->mqtt_backend_.set_on_message(
       [this](const char *topic, const char *payload, size_t len, size_t index, size_t total) {
@@ -78,6 +86,13 @@ void MQTTClientComponent::setup() {
         this, [](void *self, uint8_t level, const char *tag, const char *message, size_t message_len) {
           static_cast<MQTTClientComponent *>(self)->on_log(level, tag, message, message_len);
         });
+#ifdef USE_MQTT_CRASH_LOG
+    // `esphome logs` announces itself here, like an API client's log subscription, and the record stays
+    // loggable until the next boot, so the offer does too. A log topic that drops errors would swallow the
+    // replay, so it is not offered. The topic is rebuilt from the device name when needed, so nothing is
+    // stored for it.
+    this->crash_log_offered_ = logger::crash_log_pending() && this->log_level_ >= ESPHOME_LOG_LEVEL_ERROR;
+#endif
   }
 #endif
 
@@ -87,10 +102,8 @@ void MQTTClientComponent::setup() {
         2);
 
     // Format topic on stack - subscribe() copies it
-    // "esphome/ping/" (13) + name (ESPHOME_DEVICE_NAME_MAX_LEN) + null (1)
-    constexpr size_t ping_topic_buffer_size = 13 + ESPHOME_DEVICE_NAME_MAX_LEN + 1;
-    char ping_topic[ping_topic_buffer_size];
-    buf_append_printf(ping_topic, sizeof(ping_topic), 0, "esphome/ping/%s", App.get_name().c_str());
+    char ping_topic[ESPHOME_TOPIC_BUFFER_SIZE];
+    format_esphome_topic(ping_topic, "ping");
     this->subscribe(
         ping_topic, [this](const std::string &topic, const std::string &payload) { this->send_device_info_(); }, 2);
   }
@@ -105,10 +118,8 @@ void MQTTClientComponent::send_device_info_() {
     return;
   }
   // Format topic on stack to avoid heap allocation
-  // "esphome/discover/" (17) + name (ESPHOME_DEVICE_NAME_MAX_LEN) + null (1)
-  constexpr size_t topic_buffer_size = 17 + ESPHOME_DEVICE_NAME_MAX_LEN + 1;
-  char topic[topic_buffer_size];
-  buf_append_printf(topic, sizeof(topic), 0, "esphome/discover/%s", App.get_name().c_str());
+  char topic[ESPHOME_TOPIC_BUFFER_SIZE];
+  format_esphome_topic(topic, "discover");
 
   // NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson
   this->publish_json(
@@ -354,6 +365,14 @@ void MQTTClientComponent::check_connected() {
   delay(100);  // NOLINT
 
   this->resubscribe_subscriptions_();
+#ifdef USE_MQTT_CRASH_LOG
+  // Best effort: a refused subscription is tried again at the next connection
+  if (this->crash_log_offered_) {
+    char topic[ESPHOME_TOPIC_BUFFER_SIZE];
+    format_esphome_topic(topic, "logs");
+    this->subscribe_(topic, 0);
+  }
+#endif
   this->send_device_info_();
 
   for (MQTTComponent *component : this->children_)
@@ -461,6 +480,19 @@ void MQTTClientComponent::resubscribe_subscriptions_() {
   }
 }
 
+#ifdef USE_MQTT_CRASH_LOG
+void MQTTClientComponent::handle_crash_log_request_(const std::string &topic, const std::string &payload) {
+  // Only the CLI's empty request counts: a log_topic pointing at this topic would otherwise replay the crash
+  // log for every line it logs
+  if (!this->crash_log_offered_ || !payload.empty())
+    return;
+  char logs_topic[ESPHOME_TOPIC_BUFFER_SIZE];
+  format_esphome_topic(logs_topic, "logs");
+  if (topic == logs_topic)
+    logger::log_crash_to_client();
+}
+#endif
+
 void MQTTClientComponent::add_subscription_(std::string &&topic, mqtt_subscription_callback_t callback, bool boxed,
                                             uint8_t qos) {
   auto &subscription = this->subscriptions_.emplace_back(MQTTSubscription{
@@ -580,63 +612,6 @@ void MQTTClientComponent::disable() {
   this->on_shutdown();
 }
 
-/** Check if the message topic matches the given subscription topic
- *
- * INFO: MQTT spec mandates that topics must not be empty and must be valid NULL-terminated UTF-8 strings.
- *
- * @param message The message topic that was received from the MQTT server. Note: this must not contain
- *                wildcard characters as mandated by the MQTT spec.
- * @param subscription The subscription topic we are matching against.
- * @param is_normal Is this a "normal" topic - Does the message topic not begin with a "$".
- * @param past_separator Are we past the first '/' topic separator.
- * @return true if the subscription topic matches the message topic, false otherwise.
- */
-static bool topic_match(const char *message, const char *subscription, bool is_normal, bool past_separator) {
-  // Reached end of both strings at the same time, this means we have a successful match
-  if (*message == '\0' && *subscription == '\0')
-    return true;
-
-  // Either the message or the subscribe are at the end. This means they don't match.
-  if (*message == '\0' || *subscription == '\0')
-    return false;
-
-  bool do_wildcards = is_normal || past_separator;
-
-  if (*subscription == '+' && do_wildcards) {
-    // single level wildcard
-    // consume + from subscription
-    subscription++;
-    // consume everything from message until '/' found or end of string
-    while (*message != '\0' && *message != '/') {
-      message++;
-    }
-    // after this, both pointers will point to a '/' or to the end of the string
-
-    return topic_match(message, subscription, is_normal, true);
-  }
-
-  if (*subscription == '#' && do_wildcards) {
-    // multilevel wildcard - MQTT mandates that this must be at end of subscribe topic
-    return true;
-  }
-
-  // this handles '/' and normal characters at the same time.
-  if (*message != *subscription)
-    return false;
-
-  past_separator = past_separator || *subscription == '/';
-
-  // consume characters
-  subscription++;
-  message++;
-
-  return topic_match(message, subscription, is_normal, past_separator);
-}
-
-static bool topic_match(const char *message, const char *subscription) {
-  return topic_match(message, subscription, *message != '\0' && *message != '$', false);
-}
-
 void MQTTClientComponent::on_message(const std::string &topic, const std::string &payload) {
 #ifdef USE_ESP8266
   // IMPORTANT: This defer is REQUIRED to prevent stack overflow crashes on ESP8266.
@@ -658,6 +633,9 @@ void MQTTClientComponent::on_message(const std::string &topic, const std::string
   // DO NOT REMOVE THIS DEFER without understanding the above. It may appear to work
   // in simple tests but will cause crashes with complex automations.
   this->defer([this, topic, payload]() {
+#endif
+#ifdef USE_MQTT_CRASH_LOG
+    this->handle_crash_log_request_(topic, payload);
 #endif
     // Indexed with the count taken up front: a callback may subscribe, which can reallocate the
     // vector, and the new entry waits for the next message.

esphome/components/mqtt/mqtt_client.h

@@ -24,6 +24,11 @@
 
 #include <vector>
 
+// The crash log replay needs the logger, which mqtt does not depend on
+#if defined(USE_LOGGER) && defined(USE_CRASH_HANDLER)
+#define USE_MQTT_CRASH_LOG
+#endif
+
 namespace esphome::mqtt {
 
 /** Callback for MQTT events.
@@ -316,6 +321,9 @@ class MQTTClientComponent final : public Component {
     callback(topic, doc.as<JsonObject>());
   }
   void erase_removed_subscriptions_();
+#ifdef USE_MQTT_CRASH_LOG
+  void handle_crash_log_request_(const std::string &topic, const std::string &payload);
+#endif
   void add_subscription_(std::string &&topic, mqtt_subscription_callback_t callback, bool boxed, uint8_t qos);
   void resubscribe_subscriptions_();
 
@@ -359,6 +367,9 @@ class MQTTClientComponent final : public Component {
   bool dns_resolved_{false};
   bool dns_resolve_error_{false};
   bool enable_on_boot_{true};
+#ifdef USE_MQTT_CRASH_LOG
+  bool crash_log_offered_{false};
+#endif
   std::vector<MQTTComponent *> children_;
   uint32_t reboot_timeout_{300000};
   uint32_t connect_begin_;

esphome/components/mqtt/mqtt_topic_match.h

@@ -0,0 +1,28 @@
+#pragma once
+
+namespace esphome::mqtt {
+
+/// True if the message topic matches the subscription, with + and # wildcards.
+/// A loop on purpose: recursing per character overflowed the ESP32 loop task's stack on long topics.
+inline bool topic_match(const char *message, const char *subscription) {
+  // First-level wildcards do not match topics that begin with '$'
+  bool wildcards = *message != '$';
+  while (*message != '\0' && *subscription != '\0') {
+    if (wildcards && *subscription == '#')
+      return true;
+    if (wildcards && *subscription == '+') {
+      subscription++;
+      while (*message != '\0' && *message != '/')
+        message++;
+      continue;
+    }
+    if (*message != *subscription)
+      return false;
+    wildcards |= *subscription == '/';
+    message++;
+    subscription++;
+  }
+  return *message == *subscription;  // a match only if both ended
+}
+
+}  // namespace esphome::mqtt

esphome/components/rp2/__init__.py

@@ -392,6 +392,7 @@ async def to_code(config: ConfigType) -> None:
         "USE_RP2040_WATCHDOG_TIMEOUT", config[CONF_WATCHDOG_TIMEOUT]
     )  # back-compat alias
     cg.add_define("USE_RP2_CRASH_HANDLER")
+    cg.add_define("USE_CRASH_HANDLER")
     cg.add_define("USE_RP2040_CRASH_HANDLER")  # back-compat alias
 
     _configure_lwip()

esphome/core/defines.h

@@ -58,6 +58,7 @@
 #define USE_CLIMATE_VISUAL_OVERRIDES
 #define USE_CONTROLLER_REGISTRY
 #define USE_COVER
+#define USE_CRASH_HANDLER
 #define USE_DATETIME
 #define USE_DATETIME_DATE
 #define USE_DATETIME_DATETIME

esphome/mqtt.py

@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING
 import paho.mqtt.client as mqtt
 
 from esphome.const import (
+    CONF_BIRTH_MESSAGE,
     CONF_BROKER,
     CONF_CERTIFICATE_AUTHORITY,
     CONF_CLIENT_CERTIFICATE,
@@ -21,7 +22,9 @@ from esphome.const import (
     CONF_MQTT,
     CONF_NAME,
     CONF_PASSWORD,
+    CONF_PAYLOAD,
     CONF_PORT,
+    CONF_RETAIN,
     CONF_SKIP_CERT_CN_CHECK,
     CONF_TOPIC,
     CONF_TOPIC_PREFIX,
@@ -319,13 +322,36 @@ def show_logs(config, topic=None, username=None, password=None, client_id=None):
         return 1
     _LOGGER.info("Starting log output from %s", topic)
 
-    def on_message(client, userdata, msg):
-        time_ = datetime.now().astimezone().time().strftime("[%H:%M:%S]")
-        payload = msg.payload.decode(errors="backslashreplace")
-        message = time_ + payload
-        safe_print(message)
+    logs_topic = f"esphome/logs/{config[CONF_ESPHOME][CONF_NAME]}"
+    # Ask when the retained birth message reports online: a request sent while the device is down is
+    # lost, and only a retained message tells a CLI that connects later that the device is already up.
+    # An empty payload deletes the retained message instead, so it counts as no birth message.
+    birth = config.get(CONF_MQTT, {}).get(CONF_BIRTH_MESSAGE) or {}
+    birth_payload = birth.get(CONF_PAYLOAD)
+    birth_topic = birth.get(CONF_TOPIC) if birth_payload else None
+    ask_on_connect = birth_topic is None or not birth[CONF_RETAIN]
 
-    return initialize(config, [topic], on_message, None, username, password, client_id)
+    def request_crash_log(client):
+        _LOGGER.debug("Requesting the crash log via topic: %s", logs_topic)
+        client.publish(logs_topic, None, retain=False)
+
+    def on_message(client, userdata, msg):
+        payload = msg.payload.decode(errors="backslashreplace")
+        if msg.topic == birth_topic and payload == birth_payload:
+            request_crash_log(client)
+        # The birth topic may sit inside the log subscription, so print by what was asked for
+        if mqtt.topic_matches_sub(topic, msg.topic):
+            time_ = datetime.now().astimezone().time().strftime("[%H:%M:%S]")
+            safe_print(time_ + payload)
+
+    def on_connect(client, userdata, flags, return_code):
+        if ask_on_connect:
+            request_crash_log(client)
+
+    subscriptions = [topic] if birth_topic is None else [topic, birth_topic]
+    return initialize(
+        config, subscriptions, on_message, on_connect, username, password, client_id
+    )
 
 
 def clear_topic(config, topic, username=None, password=None, client_id=None):

tests/components/friedrich/friedrich_test.cpp

@@ -4,6 +4,7 @@
 #include "esphome/components/friedrich/friedrich.h"
 #include "esphome/components/remote_base/aeha_protocol.h"
 #include "esphome/core/helpers.h"
+#include "esphome/core/preferences.h"
 
 namespace esphome::friedrich::testing {
 
@@ -41,7 +42,12 @@ struct Fixture {
   TestFriedrich sut;
   CapturingTransmitter transmitter;
 
-  Fixture() { this->sut.set_transmitter(&this->transmitter); }
+  Fixture() {
+    // Climate tests share one preference key, so clear what an earlier test saved or setup restores it
+    if (global_preferences != nullptr)
+      global_preferences->reset();
+    this->sut.set_transmitter(&this->transmitter);
+  }
 
   // The last transmission decoded, or an empty frame if it could not be decoded
   AEHAData sent() { return AEHAProtocol().decode(rx_of(this->transmitter.last)).value_or(AEHAData{}); }

tests/components/mqtt/test_topic_match.cpp

@@ -0,0 +1,103 @@
+#include <gtest/gtest.h>
+
+#include <random>
+#include <string>
+
+#include "esphome/components/mqtt/mqtt_topic_match.h"
+
+namespace esphome::mqtt::testing {
+
+// The previous recursive implementation, kept to check the loop matches it
+static bool reference_match(const char *message, const char *subscription, bool is_normal, bool past_separator) {
+  if (*message == '\0' && *subscription == '\0')
+    return true;
+  if (*message == '\0' || *subscription == '\0')
+    return false;
+  bool do_wildcards = is_normal || past_separator;
+  if (*subscription == '+' && do_wildcards) {
+    subscription++;
+    while (*message != '\0' && *message != '/')
+      message++;
+    return reference_match(message, subscription, is_normal, true);
+  }
+  if (*subscription == '#' && do_wildcards)
+    return true;
+  if (*message != *subscription)
+    return false;
+  past_separator = past_separator || *subscription == '/';
+  return reference_match(message + 1, subscription + 1, is_normal, past_separator);
+}
+
+static bool reference_match(const char *message, const char *subscription) {
+  return reference_match(message, subscription, *message != '\0' && *message != '$', false);
+}
+
+TEST(MQTTTopicMatch, Exact) {
+  EXPECT_TRUE(topic_match("a/b/c", "a/b/c"));
+  EXPECT_FALSE(topic_match("a/b/c", "a/b"));
+  EXPECT_FALSE(topic_match("a/b", "a/b/c"));
+  EXPECT_FALSE(topic_match("a/b/c", "a/b/d"));
+}
+
+TEST(MQTTTopicMatch, SingleLevelWildcard) {
+  EXPECT_TRUE(topic_match("a/b/c", "a/+/c"));
+  EXPECT_TRUE(topic_match("a/b/c", "+/+/+"));
+  EXPECT_TRUE(topic_match("a//c", "a/+/c"));
+  EXPECT_FALSE(topic_match("a/b/c", "a/+"));
+  EXPECT_FALSE(topic_match("a/b/c/d", "a/+/c"));
+}
+
+TEST(MQTTTopicMatch, MultiLevelWildcard) {
+  EXPECT_TRUE(topic_match("a/b/c", "a/#"));
+  EXPECT_TRUE(topic_match("a/b/c", "#"));
+  EXPECT_TRUE(topic_match("a/b/c", "a/+/#"));
+  EXPECT_FALSE(topic_match("b/c", "a/#"));
+}
+
+TEST(MQTTTopicMatch, DollarTopicsSkipFirstLevelWildcards) {
+  EXPECT_FALSE(topic_match("$SYS/broker", "#"));
+  EXPECT_FALSE(topic_match("$SYS/broker", "+/broker"));
+  EXPECT_TRUE(topic_match("$SYS/broker", "$SYS/#"));
+  EXPECT_TRUE(topic_match("$SYS/broker", "$SYS/+"));
+}
+
+TEST(MQTTTopicMatch, EdgeCases) {
+  EXPECT_TRUE(topic_match("", ""));
+  EXPECT_FALSE(topic_match("a", ""));
+  // Kept from the previous code; the MQTT spec has "#" match the parent level and "+" an empty last level
+  EXPECT_FALSE(topic_match("a/b", "a/b/#"));
+  EXPECT_FALSE(topic_match("a/", "a/+"));
+  EXPECT_TRUE(topic_match("a/$x", "a/+"));
+}
+
+// Recursion overflowed the ESP32 loop stack at about 250 characters
+TEST(MQTTTopicMatch, LongTopic) {
+  std::string topic(4096, 'x');
+  topic[100] = '/';
+  EXPECT_TRUE(topic_match(topic.c_str(), topic.c_str()));
+  std::string other = topic;
+  other.back() = 'y';
+  EXPECT_FALSE(topic_match(topic.c_str(), other.c_str()));
+}
+
+TEST(MQTTTopicMatch, MatchesPreviousImplementation) {
+  // Small alphabet so wildcards, separators and '$' meet often
+  static constexpr char MESSAGE_CHARS[] = {'a', 'b', '/', '$'};
+  static constexpr char SUBSCRIPTION_CHARS[] = {'a', 'b', '/', '$', '+', '#'};
+  std::mt19937 rng(1234);  // NOLINT(cert-msc32-c,cert-msc51-cpp,bugprone-random-generator-seed) reproducible
+  auto random_string = [&rng](const char *chars, size_t count) {
+    std::string s(rng() % 8, ' ');
+    for (char &c : s)
+      c = chars[rng() % count];
+    return s;
+  };
+  for (int i = 0; i < 20000; i++) {
+    std::string message = random_string(MESSAGE_CHARS, sizeof(MESSAGE_CHARS));
+    std::string subscription = random_string(SUBSCRIPTION_CHARS, sizeof(SUBSCRIPTION_CHARS));
+    ASSERT_EQ(topic_match(message.c_str(), subscription.c_str()),
+              reference_match(message.c_str(), subscription.c_str()))
+        << "message='" << message << "' subscription='" << subscription << "'";
+  }
+}
+
+}  // namespace esphome::mqtt::testing

tests/unit_tests/test_mqtt.py

@@ -9,9 +9,112 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
-from esphome.const import CONF_BROKER, CONF_ESPHOME, CONF_MQTT, CONF_NAME
+from esphome.const import (
+    CONF_BIRTH_MESSAGE,
+    CONF_BROKER,
+    CONF_ESPHOME,
+    CONF_MQTT,
+    CONF_NAME,
+    CONF_PAYLOAD,
+    CONF_RETAIN,
+    CONF_TOPIC,
+)
 from esphome.core import EsphomeError
-from esphome.mqtt import get_esphome_device_ip
+from esphome.mqtt import get_esphome_device_ip, show_logs
+
+
+@pytest.mark.parametrize(
+    "birth",
+    [
+        None,
+        {CONF_TOPIC: "test-device/status", CONF_PAYLOAD: "online", CONF_RETAIN: False},
+        {CONF_TOPIC: "test-device/status", CONF_PAYLOAD: "", CONF_RETAIN: True},
+    ],
+    ids=["no birth message", "unretained birth message", "empty birth payload"],
+)
+@patch("esphome.mqtt.initialize")
+def test_show_logs_asks_on_connect(
+    mock_initialize: MagicMock, birth: dict | None
+) -> None:
+    """Without a retained birth message nothing tells a later CLI the device is up, so it asks on connect."""
+    config = _discovery_config()
+    if birth is not None:
+        config[CONF_MQTT][CONF_BIRTH_MESSAGE] = birth
+
+    show_logs(config)
+
+    on_connect = mock_initialize.call_args.args[3]
+    client = MagicMock()
+    on_connect(client, None, None, 0)
+    client.publish.assert_called_once_with(
+        "esphome/logs/test-device", None, retain=False
+    )
+    if birth is None or not birth[CONF_PAYLOAD]:
+        assert mock_initialize.call_args.args[1] == ["test-device/debug"]
+
+
+@patch("esphome.mqtt.safe_print")
+@patch("esphome.mqtt.initialize")
+def test_show_logs_requests_crash_log_when_device_reports_online(
+    mock_initialize: MagicMock, mock_print: MagicMock
+) -> None:
+    """With a retained birth message the request follows the device coming online, not the CLI connecting."""
+    config = _discovery_config()
+    config[CONF_MQTT][CONF_BIRTH_MESSAGE] = {
+        CONF_TOPIC: "test-device/status",
+        CONF_PAYLOAD: "online",
+        CONF_RETAIN: True,
+    }
+
+    show_logs(config)
+
+    assert mock_initialize.call_args.args[1] == [
+        "test-device/debug",
+        "test-device/status",
+    ]
+    on_message = mock_initialize.call_args.args[2]
+    on_connect = mock_initialize.call_args.args[3]
+    client = MagicMock()
+    on_connect(client, None, None, 0)
+    client.publish.assert_not_called()
+
+    on_message(client, None, MagicMock(topic="test-device/status", payload=b"offline"))
+    client.publish.assert_not_called()
+    on_message(client, None, MagicMock(topic="test-device/status", payload=b"online"))
+    client.publish.assert_called_once_with(
+        "esphome/logs/test-device", None, retain=False
+    )
+    mock_print.assert_not_called()
+
+    on_message(
+        client, None, MagicMock(topic="test-device/debug", payload=b"[I][app]: hi")
+    )
+    assert mock_print.call_args.args[0].endswith("[I][app]: hi")
+    client.publish.assert_called_once()
+
+
+@patch("esphome.mqtt.safe_print")
+@patch("esphome.mqtt.initialize")
+def test_show_logs_prints_a_birth_message_inside_the_log_subscription(
+    mock_initialize: MagicMock, mock_print: MagicMock
+) -> None:
+    """A birth topic covered by the log topic is still printed, and still requests the crash log."""
+    config = _discovery_config()
+    config[CONF_MQTT][CONF_BIRTH_MESSAGE] = {
+        CONF_TOPIC: "test-device/status",
+        CONF_PAYLOAD: "online",
+        CONF_RETAIN: True,
+    }
+
+    show_logs(config, topic="test-device/#")
+
+    on_message = mock_initialize.call_args.args[2]
+    client = MagicMock()
+    on_message(client, None, MagicMock(topic="test-device/status", payload=b"online"))
+    client.publish.assert_called_once_with(
+        "esphome/logs/test-device", None, retain=False
+    )
+    assert mock_print.call_args.args[0].endswith("online")
 
 
 def test_get_esphome_device_ip_empty_broker() -> None: