Add scheduled tasks feature and delayed entity signals

CHANGELOG.md

@@ -9,6 +9,18 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ADDED
 
+- Added `durabletask.scheduled`, a recurring schedule feature built on durable
+  entities. Use `configure_scheduled_tasks(worker)` to enable it on a worker,
+  then manage schedules from the client via `ScheduledTaskClient` (and the
+  per-schedule `ScheduleClient`). Supports creating, describing, listing,
+  updating, pausing, resuming, and deleting schedules with configurable
+  `interval`, `start_at`, `end_at`, and `start_immediately_if_late` options.
+- Added an optional `signal_time` parameter to `EntityContext.signal_entity`
+  and `DurableEntity.signal_entity`, allowing an entity signal to be scheduled
+  for future delivery.
+- Added an optional `signal_time` parameter to `OrchestrationContext.signal_entity`
+  and to the client `signal_entity` methods (sync and async), allowing entity
+  signals to be scheduled for future delivery from orchestrations and clients.
 - Added a pluggable `DataConverter` (`durabletask.serialization`) accepted by
   `TaskHubGrpcWorker`, `TaskHubGrpcClient`, and `AsyncTaskHubGrpcClient` via a
   `data_converter` argument. Every payload boundary (inputs, outputs, events,

durabletask/client.py

@@ -825,8 +825,10 @@ def purge_orchestrations_by(self,
     def signal_entity(self,
                       entity_instance_id: EntityInstanceId,
                       operation_name: str,
-                      input: Any | None = None) -> None:
-        req = build_signal_entity_req(entity_instance_id, operation_name, input, self._data_converter)
+                      input: Any | None = None,
+                      signal_time: datetime | None = None) -> None:
+        req = build_signal_entity_req(
+            entity_instance_id, operation_name, input, signal_time, self._data_converter)
         self._logger.info(f"Signaling entity '{entity_instance_id}' operation '{operation_name}'.")
         if self._payload_store is not None:
             payload_helpers.externalize_payloads(
@@ -1308,8 +1310,10 @@ async def purge_orchestrations_by(self,
     async def signal_entity(self,
                             entity_instance_id: EntityInstanceId,
                             operation_name: str,
-                            input: Any | None = None) -> None:
-        req = build_signal_entity_req(entity_instance_id, operation_name, input, self._data_converter)
+                            input: Any | None = None,
+                            signal_time: datetime | None = None) -> None:
+        req = build_signal_entity_req(
+            entity_instance_id, operation_name, input, signal_time, self._data_converter)
         self._logger.info(f"Signaling entity '{entity_instance_id}' operation '{operation_name}'.")
         if self._payload_store is not None:
             await payload_helpers.externalize_payloads_async(

durabletask/entities/durable_entity.py

@@ -2,6 +2,7 @@
 # Licensed under the MIT License.
 
 from typing import Any, TypeVar, overload
+from datetime import datetime
 
 from durabletask.entities.entity_context import EntityContext
 from durabletask.entities.entity_instance_id import EntityInstanceId
@@ -52,7 +53,9 @@ def set_state(self, state: Any):
         """
         self.entity_context.set_state(state)
 
-    def signal_entity(self, entity_instance_id: EntityInstanceId, operation: str, input: Any | None = None) -> None:
+    def signal_entity(self, entity_instance_id: EntityInstanceId, operation: str,
+                      input: Any | None = None,
+                      signal_time: datetime | None = None) -> None:
         """Signal another entity to perform an operation.
 
         Parameters
@@ -63,8 +66,12 @@ def signal_entity(self, entity_instance_id: EntityInstanceId, operation: str, in
             The operation to perform on the entity.
         input : Any, optional
             The input to provide to the entity for the operation.
+        signal_time : datetime, optional
+            The time at which the signal should be delivered. If None, the signal is
+            delivered as soon as possible. Use this to schedule a future operation,
+            for example to have an entity wake itself up at a later time.
         """
-        self.entity_context.signal_entity(entity_instance_id, operation, input)
+        self.entity_context.signal_entity(entity_instance_id, operation, input, signal_time)
 
     def schedule_new_orchestration(self, orchestration_name: str, input: Any | None = None, instance_id: str | None = None) -> str:
         """Schedule a new orchestration instance.

durabletask/entities/entity_context.py

@@ -1,8 +1,10 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+from datetime import datetime
 from typing import TYPE_CHECKING, Any, TypeVar, overload
 import uuid
+from google.protobuf import timestamp_pb2
 from durabletask.entities.entity_instance_id import EntityInstanceId
 from durabletask.internal import helpers
 from durabletask.internal.entity_state_shim import StateShim
@@ -88,7 +90,9 @@ def set_state(self, new_state: Any) -> None:
         """
         self._state.set_state(new_state)
 
-    def signal_entity(self, entity_instance_id: EntityInstanceId, operation: str, input: Any | None = None) -> None:
+    def signal_entity(self, entity_instance_id: EntityInstanceId, operation: str,
+                      input: Any | None = None,
+                      signal_time: datetime | None = None) -> None:
         """Signal another entity to perform an operation.
 
         Parameters
@@ -99,15 +103,23 @@ def signal_entity(self, entity_instance_id: EntityInstanceId, operation: str, in
             The operation to perform on the entity.
         input : Any, optional
             The input to provide to the entity for the operation.
+        signal_time : datetime, optional
+            The time at which the signal should be delivered. If None, the signal is
+            delivered as soon as possible. Use this to schedule a future operation,
+            for example to have an entity wake itself up at a later time.
         """
         encoded_input: str | None = self._data_converter.serialize(input)
+        scheduled_time: timestamp_pb2.Timestamp | None = None
+        if signal_time is not None:
+            scheduled_time = timestamp_pb2.Timestamp()
+            scheduled_time.FromDatetime(signal_time)
         self._state.add_operation_action(
             pb.OperationAction(
                 sendSignal=pb.SendSignalAction(
                     instanceId=str(entity_instance_id),
                     name=operation,
                     input=helpers.get_string_value(encoded_input),
-                    scheduledTime=None,
+                    scheduledTime=scheduled_time,
                     requestTime=None,
                     parentTraceContext=None,
                 )

durabletask/internal/client_helpers.py

@@ -205,14 +205,16 @@ def build_signal_entity_req(
         entity_instance_id: EntityInstanceId,
         operation_name: str,
         input: Any | None = None,
+        signal_time: datetime | None = None,
         data_converter: DataConverter | None = None) -> pb.SignalEntityRequest:
     """Build a SignalEntityRequest for signaling an entity."""
+    scheduled_time = helpers.new_timestamp(signal_time) if signal_time is not None else None
     return pb.SignalEntityRequest(
         instanceId=str(entity_instance_id),
         name=operation_name,
         input=helpers.get_string_value(_serialize(input, data_converter)),
         requestId=str(uuid.uuid4()),
-        scheduledTime=None,
+        scheduledTime=scheduled_time,
         parentTraceContext=None,
         requestTime=helpers.new_timestamp(datetime.now(timezone.utc))
     )

durabletask/internal/helpers.py

@@ -2,7 +2,7 @@
 # Licensed under the MIT License.
 
 import traceback
-from datetime import datetime
+from datetime import datetime, timezone
 
 from google.protobuf import timestamp_pb2, wrappers_pb2
 
@@ -255,11 +255,13 @@ def new_signal_entity_action(id: int,
                              entity_id: EntityInstanceId,
                              operation: str,
                              encoded_input: str | None,
-                             request_id: str) -> pb.OrchestratorAction:
+                             request_id: str,
+                             signal_time: datetime | None = None) -> pb.OrchestratorAction:
+    scheduled_time = new_timestamp(signal_time) if signal_time is not None else None
     return pb.OrchestratorAction(id=id, sendEntityMessage=pb.SendEntityMessageAction(entityOperationSignaled=pb.EntityOperationSignaledEvent(
         requestId=request_id,
         operation=operation,
-        scheduledTime=None,
+        scheduledTime=scheduled_time,
         input=get_string_value(encoded_input),
         targetInstanceId=get_string_value(str(entity_id)),
     )))
@@ -300,6 +302,21 @@ def new_timestamp(dt: datetime) -> timestamp_pb2.Timestamp:
     return ts
 
 
+def ensure_aware(value: datetime | None) -> datetime | None:
+    """Return ``value`` as a timezone-aware datetime, assuming UTC when naive.
+
+    A naive datetime is tagged as UTC; an already-aware datetime is returned
+    unchanged. Useful before comparing user-supplied datetimes against the
+    SDK's always-aware-UTC timestamps to avoid "can't compare offset-naive and
+    offset-aware datetimes".
+    """
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value
+
+
 def new_create_sub_orchestration_action(
         id: int,
         name: str,

durabletask/scheduled/__init__.py

@@ -0,0 +1,38 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Scheduled tasks support for the Durable Task SDK.
+
+This package provides a recurring schedule feature built on top of durable
+entities and a helper orchestrator. Register the entity and orchestrator with a
+worker via :func:`configure_scheduled_tasks`, then manage schedules from the
+client via :class:`ScheduledTaskClient`.
+"""
+
+from durabletask.scheduled.client import ScheduleClient, ScheduledTaskClient
+from durabletask.scheduled.exceptions import (ScheduleClientValidationError,
+                                              ScheduleError,
+                                              ScheduleInvalidTransitionError,
+                                              ScheduleNotFoundError)
+from durabletask.scheduled.models import (ScheduleCreationOptions,
+                                          ScheduleDescription, ScheduleQuery,
+                                          ScheduleUpdateOptions)
+from durabletask.scheduled.registration import configure_scheduled_tasks
+from durabletask.scheduled.schedule_status import ScheduleStatus
+
+__all__ = [
+    "ScheduledTaskClient",
+    "ScheduleClient",
+    "ScheduleCreationOptions",
+    "ScheduleUpdateOptions",
+    "ScheduleDescription",
+    "ScheduleQuery",
+    "ScheduleStatus",
+    "ScheduleError",
+    "ScheduleNotFoundError",
+    "ScheduleClientValidationError",
+    "ScheduleInvalidTransitionError",
+    "configure_scheduled_tasks",
+]
+
+PACKAGE_NAME = "durabletask.scheduled"

durabletask/scheduled/client.py

@@ -0,0 +1,154 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import logging
+
+from durabletask.client import (EntityQuery, OrchestrationStatus,
+                                TaskHubGrpcClient)
+from durabletask.entities import EntityInstanceId
+from durabletask.internal.helpers import ensure_aware
+from durabletask.scheduled import transitions
+from durabletask.scheduled.exceptions import ScheduleNotFoundError
+from durabletask.scheduled.models import (ScheduleCreationOptions,
+                                          ScheduleDescription, ScheduleQuery,
+                                          ScheduleState, ScheduleUpdateOptions)
+from durabletask.scheduled.orchestrator import (
+    ScheduleOperationRequest, execute_schedule_operation_orchestrator)
+from durabletask.scheduled.schedule_entity import (DELETE_OPERATION,
+                                                   ENTITY_NAME)
+
+logger = logging.getLogger("durabletask.scheduled")
+
+
+class ScheduleClient:
+    """Client for managing a single schedule instance."""
+
+    def __init__(self, client: TaskHubGrpcClient, schedule_id: str,
+                 *, operation_timeout: float = 60):
+        if not schedule_id:
+            raise ValueError("schedule_id cannot be empty.")
+        self._client = client
+        self._schedule_id = schedule_id
+        self._entity_id = EntityInstanceId(ENTITY_NAME, schedule_id)
+        self._operation_timeout = operation_timeout
+
+    @property
+    def schedule_id(self) -> str:
+        """Gets the ID of this schedule."""
+        return self._schedule_id
+
+    def _run_operation(self, operation_name: str, input: object | None = None) -> None:
+        request = ScheduleOperationRequest(
+            entity_id=str(self._entity_id),
+            operation_name=operation_name,
+            input=input,
+        )
+        instance_id = self._client.schedule_new_orchestration(
+            execute_schedule_operation_orchestrator, input=request)
+        state = self._client.wait_for_orchestration_completion(
+            instance_id, timeout=self._operation_timeout)
+        if state is None or state.runtime_status != OrchestrationStatus.COMPLETED:
+            failure = state.failure_details if state else None
+            message = failure.message if failure else "unknown error"
+            raise RuntimeError(
+                f"Failed to '{operation_name}' schedule '{self._schedule_id}': {message}")
+
+    def create(self, options: ScheduleCreationOptions) -> None:
+        """Create or update this schedule with the given configuration."""
+        self._run_operation(transitions.CREATE_SCHEDULE, options)
+
+    def update(self, options: ScheduleUpdateOptions) -> None:
+        """Update this schedule's configuration."""
+        self._run_operation(transitions.UPDATE_SCHEDULE, options)
+
+    def pause(self) -> None:
+        """Pause this schedule."""
+        self._run_operation(transitions.PAUSE_SCHEDULE)
+
+    def resume(self) -> None:
+        """Resume this schedule."""
+        self._run_operation(transitions.RESUME_SCHEDULE)
+
+    def delete(self) -> None:
+        """Delete this schedule."""
+        self._run_operation(DELETE_OPERATION)
+
+    def describe(self) -> ScheduleDescription:
+        """Retrieve the current details of this schedule."""
+        metadata = self._client.get_entity(self._entity_id, include_state=True)
+        if metadata is None:
+            raise ScheduleNotFoundError(self._schedule_id)
+        state = metadata.get_typed_state(ScheduleState)
+        if state is None:
+            raise ScheduleNotFoundError(self._schedule_id)
+        return state.to_description()
+
+
+class ScheduledTaskClient:
+    """Client for managing scheduled tasks in a Durable Task application."""
+
+    def __init__(self, client: TaskHubGrpcClient, *, operation_timeout: float = 60):
+        self._client = client
+        self._operation_timeout = operation_timeout
+
+    def get_schedule_client(self, schedule_id: str) -> ScheduleClient:
+        """Get a handle to manage a specific schedule."""
+        return ScheduleClient(self._client, schedule_id,
+                              operation_timeout=self._operation_timeout)
+
+    def create_schedule(self, options: ScheduleCreationOptions) -> ScheduleClient:
+        """Create a new schedule and return a client for managing it."""
+        schedule_client = self.get_schedule_client(options.schedule_id)
+        schedule_client.create(options)
+        return schedule_client
+
+    def get_schedule(self, schedule_id: str) -> ScheduleDescription | None:
+        """Get a schedule description by ID, or None if it does not exist."""
+        try:
+            return self.get_schedule_client(schedule_id).describe()
+        except ScheduleNotFoundError:
+            return None
+
+    def list_schedules(self, schedule_query: ScheduleQuery | None = None) -> list[ScheduleDescription]:
+        """List schedules matching the given filter criteria.
+
+        > [!NOTE]
+        > The ``status`` and ``created_from``/``created_to`` filters are applied
+        > client-side after each page of entities is fetched, so an individual
+        > page may contain fewer than ``page_size`` matches (or none) even when
+        > more matching schedules exist. This mirrors the .NET implementation.
+        """
+        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,
+            page_size=page_size,
+        )
+        results: list[ScheduleDescription] = []
+        for metadata in self._client.get_all_entities(query):
+            state = metadata.get_typed_state(ScheduleState)
+            if state is None or state.schedule_configuration is None:
+                continue
+            if not self._matches_filter(state, schedule_query):
+                continue
+            results.append(state.to_description())
+        return results
+
+    @staticmethod
+    def _matches_filter(state: ScheduleState, schedule_query: ScheduleQuery | None) -> bool:
+        if schedule_query is None:
+            return True
+        if schedule_query.status is not None and state.status != schedule_query.status:
+            return False
+        # ``ScheduleQuery`` normalizes its bounds to aware UTC; defensively
+        # normalize the stored timestamp too (a payload could in principle carry
+        # a naive value) so the comparison can never raise on naive-vs-aware.
+        # Bounds are exclusive, matching the .NET ScheduledTasks implementation.
+        created_at = ensure_aware(state.schedule_created_at)
+        if schedule_query.created_from is not None and not (created_at and created_at > schedule_query.created_from):
+            return False
+        if schedule_query.created_to is not None and not (created_at and created_at < schedule_query.created_to):
+            return False
+        return True