From a215b82bd9bae32f88ed19e23d9960cd58b3a6bc Mon Sep 17 00:00:00 2001
From: Raman Gupta <7243222+raman325@users.noreply.github.com>
Date: Tue, 12 May 2026 06:32:58 -0400
Subject: [PATCH] Cancel previous Debouncer timer handle in _schedule_timer
 (#170339)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
---
 homeassistant/helpers/debounce.py | 11 ++---
 tests/helpers/test_debounce.py    | 68 +++++++++++++++++++++++++++++++
 2 files changed, 74 insertions(+), 5 deletions(-)

diff --git a/homeassistant/helpers/debounce.py b/homeassistant/helpers/debounce.py
index 4941b4960467..190947636ccd 100644
--- a/homeassistant/helpers/debounce.py
+++ b/homeassistant/helpers/debounce.py
@@ -193,8 +193,9 @@ class Debouncer[_R_co]:
 
     @callback
     def _schedule_timer(self) -> None:
-        """Schedule a timer."""
-        if not self._shutdown_requested:
-            self._timer_task = self.hass.loop.call_later(
-                self.cooldown, self._on_debounce
-            )
+        """Schedule a timer, cancelling any previously-scheduled handle."""
+        if self._shutdown_requested:
+            return
+        if self._timer_task is not None:
+            self._timer_task.cancel()
+        self._timer_task = self.hass.loop.call_later(self.cooldown, self._on_debounce)
diff --git a/tests/helpers/test_debounce.py b/tests/helpers/test_debounce.py
index 35a48e0963e1..860458def00b 100644
--- a/tests/helpers/test_debounce.py
+++ b/tests/helpers/test_debounce.py
@@ -578,3 +578,71 @@ async def test_shutdown_releases_parent_class(hass: HomeAssistant) -> None:
     # Debouncer shutdown releases the class
     debouncer.async_shutdown()
     assert my_class_weak_ref() is None
+
+
+async def test_schedule_timer_cancels_previous_handle(hass: HomeAssistant) -> None:
+    """Ensure _schedule_timer cancels any previously-scheduled handle."""
+    # Use a large cooldown so the scheduled timer can't fire mid-test on a slow
+    # event loop; the timer is only inspected and cancelled, never awaited.
+    debouncer = debounce.Debouncer(
+        hass,
+        _LOGGER,
+        cooldown=3600.0,
+        immediate=True,
+        function=AsyncMock(),
+    )
+
+    debouncer._schedule_timer()
+    first_handle = debouncer._timer_task
+    assert first_handle is not None
+    assert not first_handle.cancelled()
+
+    debouncer._schedule_timer()
+    second_handle = debouncer._timer_task
+    assert second_handle is not None
+    assert second_handle is not first_handle
+    assert first_handle.cancelled()
+
+    debouncer.async_shutdown()
+
+
+async def test_concurrent_async_call_does_not_orphan_timer(
+    hass: HomeAssistant,
+) -> None:
+    """Concurrent async_call during in-flight execution must not orphan a timer."""
+    started = asyncio.Event()
+    can_finish = asyncio.Event()
+
+    async def slow_function() -> None:
+        started.set()
+        await can_finish.wait()
+
+    # Use a large cooldown so the T1 timer scheduled below can't fire before
+    # the in-flight call completes; cancellation is verified deterministically.
+    debouncer = debounce.Debouncer(
+        hass,
+        _LOGGER,
+        cooldown=3600.0,
+        immediate=True,
+        function=slow_function,
+    )
+
+    in_flight = hass.async_create_task(debouncer.async_call())
+    await started.wait()
+    assert debouncer._timer_task is None
+
+    # The concurrent call hits the locked-immediate branch and schedules T1.
+    await debouncer.async_call()
+    first_timer = debouncer._timer_task
+    assert first_timer is not None
+    assert not first_timer.cancelled()
+
+    # Letting the in-flight call complete schedules T2.
+    can_finish.set()
+    await in_flight
+    second_timer = debouncer._timer_task
+    assert second_timer is not None
+    assert second_timer is not first_timer
+    assert first_timer.cancelled()
+
+    debouncer.async_shutdown()