PR Feedback

Author: andystaples

durabletask/scheduled/client.py

@@ -108,10 +108,11 @@ def get_schedule(self, schedule_id: str) -> ScheduleDescription | None:
         except ScheduleNotFoundError:
             return None
 
-    def list_schedules(self, filter: ScheduleQuery | None = None) -> list[ScheduleDescription]:
+    def list_schedules(self, schedule_query: ScheduleQuery | None = None) -> list[ScheduleDescription]:
         """List schedules matching the given filter criteria."""
-        prefix = filter.schedule_id_prefix if filter and filter.schedule_id_prefix else ""
-        page_size = filter.page_size if filter and filter.page_size else ScheduleQuery.DEFAULT_PAGE_SIZE
+        prefix = schedule_query.schedule_id_prefix if schedule_query and schedule_query.schedule_id_prefix else ""
+        page_size = (schedule_query.page_size if schedule_query and schedule_query.page_size
+                     else ScheduleQuery.DEFAULT_PAGE_SIZE)
         query = EntityQuery(
             instance_id_starts_with=f"@{ENTITY_NAME}@{prefix}",
             include_state=True,
@@ -122,20 +123,20 @@ def list_schedules(self, filter: ScheduleQuery | None = None) -> list[ScheduleDe
             state = metadata.get_typed_state(ScheduleState)
             if state is None or state.schedule_configuration is None:
                 continue
-            if not self._matches_filter(state, filter):
+            if not self._matches_filter(state, schedule_query):
                 continue
             results.append(state.to_description())
         return results
 
     @staticmethod
-    def _matches_filter(state: ScheduleState, filter: ScheduleQuery | None) -> bool:
-        if filter is None:
+    def _matches_filter(state: ScheduleState, schedule_query: ScheduleQuery | None) -> bool:
+        if schedule_query is None:
             return True
-        if filter.status is not None and state.status != filter.status:
+        if schedule_query.status is not None and state.status != schedule_query.status:
             return False
         created_at = state.schedule_created_at
-        if filter.created_from is not None and not (created_at and created_at > filter.created_from):
+        if schedule_query.created_from is not None and not (created_at and created_at > schedule_query.created_from):
             return False
-        if filter.created_to is not None and not (created_at and created_at < filter.created_to):
+        if schedule_query.created_to is not None and not (created_at and created_at < schedule_query.created_to):
             return False
         return True

durabletask/scheduled/registration.py

@@ -2,8 +2,7 @@
 # Licensed under the MIT License.
 
 from durabletask.worker import TaskHubGrpcWorker
-from durabletask.scheduled.orchestrator import \
-    execute_schedule_operation_orchestrator
+from durabletask.scheduled.orchestrator import execute_schedule_operation_orchestrator
 from durabletask.scheduled.schedule_entity import ENTITY_NAME, Schedule
 
 

durabletask/testing/in_memory_backend.py

@@ -153,6 +153,13 @@ def __init__(self, max_history_size: int = 10000, port: int = 50051):
         self._logger = logging.getLogger(__name__)
         self._shutdown_event = threading.Event()
         self._work_available = threading.Event()
+        # Monotonic lifecycle counter, bumped on every stop()/reset(). Background
+        # timers (e.g. delayed entity signals) capture it when scheduled and bail
+        # if it has changed by the time they fire, so a timer created before a
+        # stop/reset cannot mutate a subsequently-restarted or cleared backend.
+        # Unlike ``_shutdown_event`` (which reset() clears to allow restart), this
+        # only ever moves forward, so it reliably invalidates stale timers.
+        self._generation = 0
 
     def start(self) -> str:
         """
@@ -183,6 +190,7 @@ def stop(self, grace: float | None = None):
         """
         self._shutdown_event.set()
         self._work_available.set()  # Unblock GetWorkItems loops
+        self._generation += 1
         if self._server:
             stop_future = self._server.stop(grace)
             stop_future.wait()
@@ -208,6 +216,7 @@ def reset(self):
             self._state_waiters.clear()
             self._shutdown_event.clear()
             self._work_available.clear()
+            self._generation += 1
 
     # gRPC Service Methods
 
@@ -1799,11 +1808,14 @@ def _schedule_delayed_entity_operation(self, entity_id: str, event: pb.HistoryEv
         in-memory backend.
         """
         delay = max(0.0, (fire_at - datetime.now(timezone.utc)).total_seconds())
+        # Capture the lifecycle generation so a timer outliving a stop()/reset()
+        # does not enqueue into a restarted or cleared backend.
+        scheduled_generation = self._generation
 
         def fire() -> None:
             time.sleep(delay)
             with self._lock:
-                if self._shutdown_event.is_set():
+                if self._shutdown_event.is_set() or self._generation != scheduled_generation:
                     return
                 self._queue_entity_operation(entity_id, event)
 

tests/durabletask/test_delayed_signal_normalization.py

@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Unit tests for delayed-signal datetime normalization and in-memory backend
+timer lifecycle (regression coverage for PR #160 review feedback)."""
+
+import time
+from datetime import datetime, timedelta, timezone
+
+import durabletask.internal.orchestrator_service_pb2 as pb
+from durabletask.entities import EntityInstanceId
+from durabletask.internal import helpers
+from durabletask.internal.client_helpers import build_signal_entity_req
+from durabletask.testing import create_test_backend
+
+from tests.durabletask._port_utils import find_free_port
+
+
+class TestSignalTimeNormalization:
+    def test_build_signal_entity_req_naive_matches_aware(self):
+        entity_id = EntityInstanceId("Recorder", "k")
+        naive = datetime(2030, 1, 1, 12, 0, 0)
+        aware = datetime(2030, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+
+        req_naive = build_signal_entity_req(entity_id, "ping", signal_time=naive)
+        req_aware = build_signal_entity_req(entity_id, "ping", signal_time=aware)
+
+        assert req_naive.scheduledTime.seconds == req_aware.scheduledTime.seconds
+
+    def test_new_signal_entity_action_naive_matches_aware(self):
+        entity_id = EntityInstanceId("Recorder", "k")
+        naive = datetime(2030, 1, 1, 12, 0, 0)
+        aware = datetime(2030, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+
+        a_naive = helpers.new_signal_entity_action(1, entity_id, "ping", None, "r1", naive)
+        a_aware = helpers.new_signal_entity_action(1, entity_id, "ping", None, "r2", aware)
+
+        naive_ts = a_naive.sendEntityMessage.entityOperationSignaled.scheduledTime
+        aware_ts = a_aware.sendEntityMessage.entityOperationSignaled.scheduledTime
+        assert naive_ts.seconds == aware_ts.seconds
+
+
+class TestDelayedTimerLifecycle:
+    def test_delayed_operation_does_not_fire_into_reset_backend(self):
+        backend = create_test_backend(port=find_free_port())
+        try:
+            entity_id = "@recorder@k"
+            event = pb.HistoryEvent(
+                eventId=-1,
+                entityOperationSignaled=pb.EntityOperationSignaledEvent(operation="ping"),
+            )
+            # Schedule the op to fire shortly in the future, then reset before it
+            # fires so the timer wakes into a reset (new-generation) backend.
+            fire_at = datetime.now(timezone.utc) + timedelta(seconds=0.3)
+            backend._schedule_delayed_entity_operation(  # pyright: ignore[reportPrivateUsage]
+                entity_id, event, fire_at)
+            backend.reset()
+
+            # Give the timer thread time to wake; it must observe the new
+            # generation and refuse to recreate state in the reset backend.
+            time.sleep(0.6)
+            assert backend._entities == {}  # pyright: ignore[reportPrivateUsage]
+        finally:
+            backend.stop()
+            backend.reset()