[udp] Wake the main loop when a datagram arrives (#20585)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
Co-authored-by: J. Nick Koston <nick@koston.org>

esphome/components/udp/udp_component.cpp

@@ -45,7 +45,7 @@ void UDPComponent::setup() {
   // create listening socket if we either want to subscribe to providers, or need to listen
   // for ping key broadcasts.
   if (this->should_listen_) {
-    this->listen_socket_ = socket::socket_udp(AF_INET, IPPROTO_IP);
+    this->listen_socket_ = socket::socket_udp_loop_monitored(AF_INET, IPPROTO_IP);
     if (this->listen_socket_ == nullptr) {
       this->status_set_error(LOG_STR("Could not create socket"));
       this->mark_failed();
@@ -115,6 +115,8 @@ void UDPComponent::loop() {
   if (this->should_listen_) {
     std::array<uint8_t, MAX_PACKET_SIZE> buf;
     for (;;) {
+      if (!this->listen_socket_->ready())
+        break;
       auto len = this->listen_socket_->read(buf.data(), buf.size());
       if (len <= 0)
         break;

tests/integration/fixtures/socket_wake_gate_udp.yaml

@@ -0,0 +1,27 @@
+esphome:
+  name: socket-wake-gate-udp
+  on_boot:
+    priority: -100
+    then:
+      - lambda: |-
+          // Without a loop-monitored socket a datagram waits for loop_interval_ to expire.
+          App.set_loop_interval(2000);
+      - delay: 500ms
+      - lambda: |-
+          ESP_LOGI("test", "BOOT_DONE");
+
+host:
+
+api:
+
+logger:
+  level: INFO
+
+udp:
+  - id: test_udp
+    port: UDP_LISTEN_PORT_PLACEHOLDER
+    addresses: ["127.0.0.1"]
+    on_receive:
+      - logger.log:
+          format: "PONG"
+          level: INFO

tests/integration/test_socket_wake_gate_udp.py

@@ -0,0 +1,64 @@
+"""Test that an incoming UDP datagram wakes the main loop at once."""
+
+from __future__ import annotations
+
+import asyncio
+import socket
+import time
+
+import pytest
+
+from .types import RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_socket_wake_gate_udp(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+) -> None:
+    """A datagram must be handled well before loop_interval_ (2000ms) expires."""
+    # Hold the port with SO_REUSEADDR so nothing else claims it while the firmware builds. The
+    # device's listen socket sets SO_REUSEADDR too and binds the wildcard address alongside it.
+    hold_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    hold_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    hold_sock.bind(("127.0.0.1", 0))
+    listen_port = hold_sock.getsockname()[1]
+    config = yaml_config.replace("UDP_LISTEN_PORT_PLACEHOLDER", str(listen_port))
+
+    loop = asyncio.get_running_loop()
+    boot_done: asyncio.Future[None] = loop.create_future()
+    pong: asyncio.Future[None] = loop.create_future()
+
+    def on_log_line(line: str) -> None:
+        if "BOOT_DONE" in line and not boot_done.done():
+            boot_done.set_result(None)
+        if "PONG" in line and not pong.done():
+            pong.set_result(None)
+
+    try:
+        async with run_compiled(config, line_callback=on_log_line):
+            try:
+                await asyncio.wait_for(boot_done, timeout=15.0)
+            except TimeoutError:
+                pytest.fail("BOOT_DONE never appeared")
+
+            # Datagrams go to the more specific bind, so release the hold before sending
+            hold_sock.close()
+
+            # No API client is connected, so this datagram is the only thing that can wake select()
+            t_send = time.monotonic()
+            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as send_sock:
+                send_sock.sendto(b"PING", ("127.0.0.1", listen_port))
+
+            try:
+                await asyncio.wait_for(pong, timeout=5.0)
+            except TimeoutError:
+                pytest.fail("PONG never appeared")
+
+            elapsed_ms = (time.monotonic() - t_send) * 1000
+            # Without the wake the receive waits for loop_interval_; 500ms leaves CI headroom
+            assert elapsed_ms < 500, (
+                f"UDP receive took {elapsed_ms:.0f}ms with loop_interval_=2000ms"
+            )
+    finally:
+        hold_sock.close()