[core] fast_select stats: spin-poll to bucket load-bearing hits by µs

Replaces the single load_bearing counter with three buckets measured
via a bounded (100µs) spin-poll of the task notification value after
the scan finds data:

  race  (<10µs)  — notification arrived within ~10µs: callback-ordering
                   race between rcvevent write and xTaskNotifyGive a few
                   instructions later. Scan is noise.
  micro (<100µs) — notification arrived within 100µs: still noise at
                   loop_interval scale.
  stall (>=100µs) — notification did not arrive within the poll window.
                    Only this case could represent a real latency spike
                    that the scan is rescuing.

The 100µs spin cap is intentional: if we are wrong and this IS a real
stall, we only add 100µs to that one unlucky loop iteration.

The immediate ESP_LOGW on each hit now includes gap_us and the bucket
label so individual events can be investigated.

esphome/core/application.cpp

@@ -450,6 +450,99 @@ void Application::enable_pending_loops_() {
 }
 
 #ifdef USE_LWIP_FAST_SELECT
+std::atomic<uint32_t> Application::fast_select_scan_total_{0};
+std::atomic<uint32_t> Application::fast_select_scan_found_data_{0};
+std::atomic<uint32_t> Application::fast_select_scan_load_bearing_{0};
+std::atomic<uint32_t> Application::fast_select_scan_load_bearing_race_{0};
+std::atomic<uint32_t> Application::fast_select_scan_load_bearing_micro_{0};
+std::atomic<uint32_t> Application::fast_select_scan_load_bearing_stall_{0};
+
+void Application::log_fast_select_scan_stats_() {
+  uint32_t total = fast_select_scan_total_.load(std::memory_order_relaxed);
+  uint32_t found = fast_select_scan_found_data_.load(std::memory_order_relaxed);
+  uint32_t load_bearing = fast_select_scan_load_bearing_.load(std::memory_order_relaxed);
+  uint32_t lb_race = fast_select_scan_load_bearing_race_.load(std::memory_order_relaxed);
+  uint32_t lb_micro = fast_select_scan_load_bearing_micro_.load(std::memory_order_relaxed);
+  uint32_t lb_stall = fast_select_scan_load_bearing_stall_.load(std::memory_order_relaxed);
+  ESP_LOGD(TAG,
+           "fast_select scan: total=%" PRIu32 " found_data=%" PRIu32 " load_bearing=%" PRIu32 " (race<10us=%" PRIu32
+           " micro<100us=%" PRIu32 " stall>100us=%" PRIu32 ")",
+           total, found, load_bearing, lb_race, lb_micro, lb_stall);
+}
+
+void Application::note_fast_select_load_bearing_(struct lwip_sock *sock, uint32_t delay_ms) {
+  uint32_t load_bearing = fast_select_scan_load_bearing_.fetch_add(1, std::memory_order_relaxed) + 1;
+
+  // Spin-poll the task notification value for a short bounded window to measure how long
+  // the counterfactual ulTaskNotifyTake would actually have blocked. This distinguishes
+  // three cases:
+  //   race  (<10µs)   — notification arrived within ~10µs of scan start: callback-ordering
+  //                     race between the lwip event_callback writing rcvevent and calling
+  //                     xTaskNotifyGive a few instructions later. Scan is noise.
+  //   micro (<100µs)  — notification arrived within 100µs: still noise at loop_interval scale.
+  //   stall (≥100µs)  — notification did not arrive within our polling window. This is the
+  //                     only case where the scan could be rescuing a real latency spike.
+  // Cap the spin at 100µs so that if we're wrong and this IS a real stall, we only add
+  // 100µs of extra work to that one unlucky loop iteration.
+  uint32_t t_start = micros();
+  uint32_t gap_us = UINT32_MAX;
+  while (true) {
+    if (ulTaskNotifyValueClear(nullptr, 0) != 0) {
+      gap_us = micros() - t_start;
+      break;
+    }
+    uint32_t elapsed = micros() - t_start;
+    if (elapsed >= 100) {
+      break;
+    }
+  }
+
+  const char *bucket;
+  if (gap_us == UINT32_MAX) {
+    fast_select_scan_load_bearing_stall_.fetch_add(1, std::memory_order_relaxed);
+    bucket = "STALL";
+  } else if (gap_us < 10) {
+    fast_select_scan_load_bearing_race_.fetch_add(1, std::memory_order_relaxed);
+    bucket = "race";
+  } else {
+    fast_select_scan_load_bearing_micro_.fetch_add(1, std::memory_order_relaxed);
+    bucket = "micro";
+  }
+
+  // Find the socket's index in monitored_sockets_ for easier correlation with registration order.
+  int index = -1;
+  for (size_t i = 0; i < this->monitored_sockets_.size(); i++) {
+    if (this->monitored_sockets_[i] == sock) {
+      index = static_cast<int>(i);
+      break;
+    }
+  }
+  // Read the rcvevent value directly. This is the same offset-based read used by
+  // esphome_lwip_socket_has_data(); value > 0 means unread data is queued.
+  int16_t rcvevent =
+      *reinterpret_cast<volatile int16_t *>(reinterpret_cast<char *>(sock) + ESPHOME_LWIP_SOCK_RCVEVENT_OFFSET);
+  // Count how many other sockets also had data at this scan (could reveal whether it's always
+  // the same socket or a burst across multiple).
+  size_t sockets_with_data = 0;
+  for (struct lwip_sock *s : this->monitored_sockets_) {
+    if (esphome_lwip_socket_has_data(s))
+      sockets_with_data++;
+  }
+  if (gap_us == UINT32_MAX) {
+    ESP_LOGW(TAG,
+             "fast_select LOAD-BEARING #%" PRIu32 " [%s]: sock=%p idx=%d/%u rcvevent=%d delay_ms=%" PRIu32
+             " sockets_with_data=%u gap_us=>100",
+             load_bearing, bucket, sock, index, static_cast<unsigned>(this->monitored_sockets_.size()), rcvevent,
+             delay_ms, static_cast<unsigned>(sockets_with_data));
+  } else {
+    ESP_LOGW(TAG,
+             "fast_select LOAD-BEARING #%" PRIu32 " [%s]: sock=%p idx=%d/%u rcvevent=%d delay_ms=%" PRIu32
+             " sockets_with_data=%u gap_us=%" PRIu32,
+             load_bearing, bucket, sock, index, static_cast<unsigned>(this->monitored_sockets_.size()), rcvevent,
+             delay_ms, static_cast<unsigned>(sockets_with_data), gap_us);
+  }
+}
+
 bool Application::register_socket(struct lwip_sock *sock) {
   // It modifies monitored_sockets_ without locking — must only be called from the main loop.
   if (sock == nullptr)

esphome/core/application.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <algorithm>
+#include <atomic>
 #include <ctime>
 #include <limits>
 #include <span>
@@ -655,6 +656,25 @@ class Application {
   FixedVector<Component *> looping_components_{};
 #ifdef USE_LWIP_FAST_SELECT
   std::vector<struct lwip_sock *> monitored_sockets_;  // Cached lwip_sock pointers for direct rcvevent read
+  // Stats to verify whether the pre-sleep socket scan in yield_with_select_() is ever load-bearing.
+  // If fast_select_scan_load_bearing_ stays 0 under real workloads, the scan can be removed.
+  // These are static because yield_with_select_() is inlined at every call site.
+  static std::atomic<uint32_t> fast_select_scan_total_;
+  static std::atomic<uint32_t> fast_select_scan_found_data_;
+  // Umbrella counter: pre-scan notify peek was 0 and scan found data.
+  // Broken down into three buckets based on the post-scan spin-poll result:
+  //   _race_  — notify arrived in < 10µs   (callback-ordering race, scan is noise)
+  //   _micro_ — notify arrived in 10..100µs (still noise at loop_interval scale)
+  //   _stall_ — notify did not arrive within 100µs (the only case that could be a real stall)
+  // If _stall_ stays 0, the scan is provably irrelevant under this workload.
+  static std::atomic<uint32_t> fast_select_scan_load_bearing_;
+  static std::atomic<uint32_t> fast_select_scan_load_bearing_race_;
+  static std::atomic<uint32_t> fast_select_scan_load_bearing_micro_;
+  static std::atomic<uint32_t> fast_select_scan_load_bearing_stall_;
+  uint32_t fast_select_scan_stats_last_log_{0};
+  void log_fast_select_scan_stats_();
+  // Non-inline, called only on the rare load-bearing event so the hot path stays unchanged.
+  void note_fast_select_load_bearing_(struct lwip_sock *sock, uint32_t delay_ms);
 #elif defined(USE_HOST)
   std::vector<int> socket_fds_;  // Vector of all monitored socket file descriptors
 #endif
@@ -889,6 +909,14 @@ inline void ESPHOME_ALWAYS_INLINE Application::loop() {
   this->yield_with_select_(delay_time);
   this->last_loop_ = last_op_end_time;
 
+#ifdef USE_LWIP_FAST_SELECT
+  // Periodic fast-select scan stats (debug). Remove once the scan is proven unneeded.
+  if (last_op_end_time - this->fast_select_scan_stats_last_log_ >= 30000) {
+    this->fast_select_scan_stats_last_log_ = last_op_end_time;
+    this->log_fast_select_scan_stats_();
+  }
+#endif
+
   if (this->dump_config_at_ < this->components_.size()) {
     this->process_dump_config_();
   }
@@ -909,8 +937,30 @@ inline void ESPHOME_ALWAYS_INLINE Application::yield_with_select_(uint32_t delay
   // If a socket still has unread data (rcvevent > 0) but the task notification was already
   // consumed, ulTaskNotifyTake would block until timeout — adding up to delay_ms latency.
   // This scan preserves select() semantics: return immediately when any fd is ready.
+  //
+  // Debug stats: peek the task notification value BEFORE scanning. This answers the
+  // counterfactual "if the scan did not exist and we called ulTaskNotifyTake right now,
+  // would it stall?". ulTaskNotifyValueClear(nullptr, 0) is a pure read — it returns the
+  // current value and clears zero bits, leaving the notification state untouched. Reading
+  // before the loop (rather than after finding data) makes the answer TOCTOU-free: the
+  // value we compare against is the value at the moment Take would have been called.
+  // LibreTiny's FreeRTOS port predates ulTaskNotifyValueClear (added in FreeRTOS 10.4.0),
+  // so we fall back to a pessimistic 0, which makes load_bearing an upper bound == found_data
+  // on that platform. Zero there is still a valid proof that the scan is unused.
+#ifdef USE_ESP32
+  uint32_t fast_select_notify_value_before_scan = ulTaskNotifyValueClear(nullptr, 0);
+#else
+  uint32_t fast_select_notify_value_before_scan = 0;
+#endif
+  fast_select_scan_total_.fetch_add(1, std::memory_order_relaxed);
   for (struct lwip_sock *sock : this->monitored_sockets_) {
     if (esphome_lwip_socket_has_data(sock)) {
+      fast_select_scan_found_data_.fetch_add(1, std::memory_order_relaxed);
+      if (fast_select_notify_value_before_scan == 0) {
+        // Scan was load-bearing: no notification pending, so Take would have stalled.
+        // Delegate to a non-inline helper so the hot path stays the same size.
+        this->note_fast_select_load_bearing_(sock, delay_ms);
+      }
       yield();
       return;
     }