Merge branch 'main' into andystaples/add-replay-safe-logging

Author: andystaples

.flake8

@@ -1,5 +1,5 @@
 [flake8]
-ignore = E501,C901
+ignore = E501,C901,W503
 exclude = 
     .git
     *_pb2*

docs/features.md

@@ -352,3 +352,69 @@ with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=secure_cha
 
 > [!NOTE]
 > The worker and client output many logs at the `DEBUG` level that will be useful when understanding orchestration flow and diagnosing issues with Durable applications. Before submitting issues, please attempt a repro of the issue with debug logging enabled.
+
+### Work item filtering
+
+By default a worker receives **all** work items from the backend,
+regardless of which orchestrations, activities, or entities are
+registered. Work item filtering lets you explicitly tell the backend
+which work items a worker can handle so that only matching items are
+dispatched. This is useful when running multiple specialized workers
+against the same task hub.
+
+Work item filtering is **opt-in**. Call `use_work_item_filters()` on
+the worker before starting it.
+
+#### Auto-generated filters
+
+Calling `use_work_item_filters()` with no arguments builds filters
+automatically from the worker's registry at start time:
+
+```python
+with DurableTaskSchedulerWorker(...) as w:
+    w.add_orchestrator(my_orchestrator)
+    w.add_activity(my_activity)
+    w.use_work_item_filters()  # auto-generate from registry
+    w.start()
+```
+
+When versioning is configured with `VersionMatchStrategy.STRICT`,
+the worker's version is included in every filter so the backend
+only dispatches work items that match that exact version.
+
+#### Explicit filters
+
+Pass a `WorkItemFilters` instance for fine-grained control:
+
+```python
+from durabletask.worker import (
+    WorkItemFilters,
+    OrchestrationWorkItemFilter,
+    ActivityWorkItemFilter,
+    EntityWorkItemFilter,
+)
+
+w.use_work_item_filters(WorkItemFilters(
+    orchestrations=[
+        OrchestrationWorkItemFilter(name="my_orch", versions=["2.0.0"]),
+    ],
+    activities=[
+        ActivityWorkItemFilter(name="my_activity"),
+    ],
+    entities=[
+        EntityWorkItemFilter(name="my_entity"),
+    ],
+))
+```
+
+#### Clearing filters
+
+Pass `None` to clear any previously configured filters and return
+to the default behaviour of processing all work items:
+
+```python
+w.use_work_item_filters(None)
+```
+
+See the full
+[work item filtering sample](../examples/work_item_filtering.py).

docs/supported-patterns.md

@@ -64,12 +64,12 @@ def purchase_order_workflow(ctx: task.OrchestrationContext, order: Order):
     # Orders of $1000 or more require manager approval
     yield ctx.call_activity(send_approval_request, input=order)
 
-    # Approvals must be received within 24 hours or they will be canceled.
+    # Approvals must be received within 24 hours or they will be cancelled.
     approval_event = ctx.wait_for_external_event("approval_received")
     timeout_event = ctx.create_timer(timedelta(hours=24))
     winner = yield task.when_any([approval_event, timeout_event])
     if winner == timeout_event:
-        return "Canceled"
+        return "Cancelled"
 
     # The order was approved
     yield ctx.call_activity(place_order, input=order)
@@ -120,6 +120,51 @@ def my_orchestrator(ctx: task.OrchestrationContext, order: Order):
 
 See the full [version-aware orchestrator sample](../examples/version_aware_orchestrator.py)
 
+### Work item filtering
+
+When running multiple workers against the same task hub, each
+worker can declare which work items it handles. The backend then
+dispatches only the matching orchestrations, activities, and
+entities, avoiding unnecessary round-trips. Filtering is opt-in
+and supports both auto-generated and explicit filter sets.
+
+The simplest approach auto-generates filters from the worker's
+registry:
+
+```python
+with DurableTaskSchedulerWorker(...) as w:
+    w.add_orchestrator(greeting_orchestrator)
+    w.add_activity(greet)
+    w.use_work_item_filters()  # auto-generate from registry
+    w.start()
+```
+
+For more control you can provide explicit filters, including
+version constraints:
+
+```python
+from durabletask.worker import (
+    WorkItemFilters,
+    OrchestrationWorkItemFilter,
+    ActivityWorkItemFilter,
+)
+
+w.use_work_item_filters(WorkItemFilters(
+    orchestrations=[
+        OrchestrationWorkItemFilter(
+            name="greeting_orchestrator",
+            versions=["2.0.0"],
+        ),
+    ],
+    activities=[
+        ActivityWorkItemFilter(name="greet"),
+    ],
+))
+```
+
+See the full
+[work item filtering sample](../examples/work_item_filtering.py).
+
 ### Large payload externalization
 
 When orchestrations work with very large inputs, outputs, or event

durabletask-azuremanaged/durabletask/azuremanaged/worker.py

@@ -85,4 +85,7 @@ def __init__(self, *,
             log_formatter=log_formatter,
             interceptors=interceptors,
             concurrency_options=concurrency_options,
-            payload_store=payload_store)
+            # DTS natively supports long timers so chunking is unnecessary
+            maximum_timer_interval=None,
+            payload_store=payload_store
+        )

durabletask/__init__.py

@@ -4,13 +4,24 @@
 """Durable Task SDK for Python"""
 
 from durabletask.payload.store import LargePayloadStorageOptions, PayloadStore
-from durabletask.worker import ConcurrencyOptions, VersioningOptions
+from durabletask.worker import (
+    ActivityWorkItemFilter,
+    ConcurrencyOptions,
+    EntityWorkItemFilter,
+    OrchestrationWorkItemFilter,
+    VersioningOptions,
+    WorkItemFilters,
+)
 
 __all__ = [
+    "ActivityWorkItemFilter",
     "ConcurrencyOptions",
+    "EntityWorkItemFilter",
     "LargePayloadStorageOptions",
+    "OrchestrationWorkItemFilter",
     "PayloadStore",
     "VersioningOptions",
+    "WorkItemFilters",
 ]
 
 PACKAGE_NAME = "durabletask"

durabletask/task.py

@@ -99,7 +99,7 @@ def set_custom_status(self, custom_status: Any) -> None:
         pass
 
     @abstractmethod
-    def create_timer(self, fire_at: Union[datetime, timedelta]) -> Task:
+    def create_timer(self, fire_at: Union[datetime, timedelta]) -> CancellableTask:
         """Create a Timer Task to fire after at the specified deadline.
 
         Parameters
@@ -229,10 +229,10 @@ def call_sub_orchestrator(self, orchestrator: Union[Orchestrator[TInput, TOutput
         """
         pass
 
-    # TOOD: Add a timeout parameter, which allows the task to be canceled if the event is
+    # TOOD: Add a timeout parameter, which allows the task to be cancelled if the event is
     # not received within the specified timeout. This requires support for task cancellation.
     @abstractmethod
-    def wait_for_external_event(self, name: str) -> CompletableTask:
+    def wait_for_external_event(self, name: str) -> CancellableTask:
         """Wait asynchronously for an event to be raised with the name `name`.
 
         Parameters
@@ -370,6 +370,10 @@ class OrchestrationStateError(Exception):
     pass
 
 
+class TaskCancelledError(Exception):
+    """Exception type for cancelled orchestration tasks."""
+
+
 class Task(ABC, Generic[T]):
     """Abstract base class for asynchronous tasks in a durable orchestration."""
     _result: T
@@ -481,6 +485,48 @@ def fail(self, message: str, details: Union[Exception, pb.TaskFailureDetails]):
             self._parent.on_child_completed(self)
 
 
+class CancellableTask(CompletableTask[T]):
+    """A completable task that can be cancelled before it finishes."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._is_cancelled = False
+        self._cancel_handler: Optional[Callable[[], None]] = None
+
+    @property
+    def is_cancelled(self) -> bool:
+        """Returns True if the task was cancelled, False otherwise."""
+        return self._is_cancelled
+
+    def get_result(self) -> T:
+        if self._is_cancelled:
+            raise TaskCancelledError('The task was cancelled.')
+        return super().get_result()
+
+    def set_cancel_handler(self, cancel_handler: Callable[[], None]) -> None:
+        self._cancel_handler = cancel_handler
+
+    def cancel(self) -> bool:
+        """Attempts to cancel this task.
+
+        Returns
+        -------
+        bool
+            True if cancellation was applied, False if the task had already completed.
+        """
+        if self._is_complete:
+            return False
+
+        if self._cancel_handler is not None:
+            self._cancel_handler()
+
+        self._is_cancelled = True
+        self._is_complete = True
+        if self._parent is not None:
+            self._parent.on_child_completed(self)
+        return True
+
+
 class RetryableTask(CompletableTask[T]):
     """A task that can be retried according to a retry policy."""
 
@@ -520,14 +566,29 @@ def compute_next_delay(self) -> Optional[timedelta]:
         return None
 
 
-class TimerTask(CompletableTask[T]):
-
-    def __init__(self) -> None:
+class TimerTask(CancellableTask[None]):
+    def __init__(self, final_fire_at: Optional[datetime] = None,
+                 maximum_timer_interval: Optional[timedelta] = None):
         super().__init__()
+        self._final_fire_at = final_fire_at
+        self._maximum_timer_interval = maximum_timer_interval
 
     def set_retryable_parent(self, retryable_task: RetryableTask):
         self._retryable_parent = retryable_task
 
+    def _handle_timer_fired(self, current_utc_datetime: datetime) -> Optional[datetime]:
+        if (self._final_fire_at is not None
+                and self._maximum_timer_interval is not None
+                and current_utc_datetime < self._final_fire_at):
+            return self._get_next_fire_at(current_utc_datetime)
+        super().complete(None)
+        return None
+
+    def _get_next_fire_at(self, current_utc_datetime: datetime) -> datetime:
+        if current_utc_datetime + self._maximum_timer_interval < self._final_fire_at:
+            return current_utc_datetime + self._maximum_timer_interval
+        return self._final_fire_at
+
 
 class WhenAnyTask(CompositeTask[Task]):
     """A task that completes when any of its child tasks complete."""