cleanup

esphome/core/scheduler.cpp

@@ -386,7 +386,7 @@ void HOT Scheduler::process_to_add() {
 void HOT Scheduler::cleanup_() {
   // Fast path: if nothing to remove, just return
   // Reading to_remove_ without lock is safe because:
-  // 1. It's volatile, ensuring we read the latest value
+  // 1. We only call this from the main thread during call()
   // 2. If it's 0, there's definitely nothing to cleanup
   // 3. If it becomes non-zero after we check, cleanup will happen next time
   if (this->to_remove_ == 0)

tests/integration/fixtures/scheduler_bulk_cleanup.yaml

@@ -0,0 +1,23 @@
+esphome:
+  name: scheduler-bulk-cleanup
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+host:
+
+logger:
+  level: DEBUG
+
+api:
+  services:
+    - service: trigger_bulk_cleanup
+      then:
+        - lambda: |-
+            auto component = id(bulk_cleanup_component);
+            component->trigger_bulk_cleanup();
+
+scheduler_bulk_cleanup_component:
+  id: bulk_cleanup_component

tests/integration/test_scheduler_bulk_cleanup.py

@@ -0,0 +1,110 @@
+"""Test that triggers the bulk cleanup path when to_remove_ > MAX_LOGICALLY_DELETED_ITEMS."""
+
+import asyncio
+from pathlib import Path
+import re
+
+from aioesphomeapi import UserService
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_scheduler_bulk_cleanup(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that bulk cleanup path is triggered when many items are cancelled."""
+
+    # Get the absolute path to the external components directory
+    external_components_path = str(
+        Path(__file__).parent / "fixtures" / "external_components"
+    )
+
+    # Replace the placeholder in the YAML config with the actual path
+    yaml_config = yaml_config.replace(
+        "EXTERNAL_COMPONENT_PATH", external_components_path
+    )
+
+    # Create a future to signal test completion
+    loop = asyncio.get_event_loop()
+    test_complete_future: asyncio.Future[None] = loop.create_future()
+    bulk_cleanup_triggered = False
+    cleanup_stats: dict[str, int] = {
+        "removed": 0,
+        "before": 0,
+        "after": 0,
+    }
+
+    def on_log_line(line: str) -> None:
+        nonlocal bulk_cleanup_triggered
+
+        # Look for logs indicating bulk cleanup was triggered
+        # The actual cleanup happens silently, so we track the cancel operations
+        if "Successfully cancelled" in line and "timeouts" in line:
+            match = re.search(r"Successfully cancelled (\d+) timeouts", line)
+            if match and int(match.group(1)) > 10:
+                bulk_cleanup_triggered = True
+
+        # Track cleanup statistics
+        match = re.search(r"Bulk cleanup triggered: removed (\d+) items", line)
+        if match:
+            cleanup_stats["removed"] = int(match.group(1))
+
+        match = re.search(r"Items before cleanup: (\d+), after: (\d+)", line)
+        if match:
+            cleanup_stats["before"] = int(match.group(1))
+            cleanup_stats["after"] = int(match.group(2))
+
+        # Check for test completion
+        if "Bulk cleanup test complete" in line and not test_complete_future.done():
+            test_complete_future.set_result(None)
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_log_line),
+        api_client_connected() as client,
+    ):
+        # Verify we can connect
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "scheduler-bulk-cleanup"
+
+        # List entities and services
+        _, services = await asyncio.wait_for(
+            client.list_entities_services(), timeout=5.0
+        )
+
+        # Find our test service
+        trigger_bulk_cleanup_service: UserService | None = None
+        for service in services:
+            if service.name == "trigger_bulk_cleanup":
+                trigger_bulk_cleanup_service = service
+                break
+
+        assert trigger_bulk_cleanup_service is not None, (
+            "trigger_bulk_cleanup service not found"
+        )
+
+        # Execute the test
+        client.execute_service(trigger_bulk_cleanup_service, {})
+
+        # Wait for test completion
+        try:
+            await asyncio.wait_for(test_complete_future, timeout=10.0)
+        except asyncio.TimeoutError:
+            pytest.fail("Bulk cleanup test timed out")
+
+        # Verify bulk cleanup was triggered
+        assert bulk_cleanup_triggered, (
+            "Bulk cleanup path was not triggered - MAX_LOGICALLY_DELETED_ITEMS threshold not reached"
+        )
+
+        # Verify cleanup statistics if available
+        if cleanup_stats.get("removed", 0) > 0:
+            assert cleanup_stats.get("removed", 0) > 10, (
+                f"Expected more than 10 items removed, got {cleanup_stats.get('removed', 0)}"
+            )
+            # Note: We're not tracking before/after counts in this test
+            # The important thing is that >10 items were cancelled triggering bulk cleanup