From cb546ff2e62b3019e9703d191109aa2505b00e81 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Tue, 30 Jun 2026 16:39:05 +0100 Subject: [PATCH 1/4] Add stateful-history delta delivery to the workflow worker durabletask-go adds the stateful-history optimization: a worker caches an instance's committed history between turns on its work-item stream, and the sidecar sends only the new events instead of the full history each turn. This implements the worker side in the vendored durabletask runtime; the backend is daprd, so there is no other production change. Regenerate the durabletask proto stubs (dapr/ext/workflow/_durabletask/internal) for the new contract: the CachedHistory message, optional WorkflowRequest.cachedHistory, GetWorkItemsRequest.capabilities, and WORKER_CAPABILITY_STATEFUL_HISTORY (the never-implemented HISTORY_STREAMING value is now reserved). worker.py (TaskHubGrpcWorker): - Add _WorkflowHistoryCache: a lock-guarded, per-stream cache of each instance's committed history, reclaimed by a sliding TTL, an instance-count cap, and an optional byte budget (LRU eviction). Eviction is always safe because a miss is recovered via the GetInstanceHistory RPC. - Advertise WORKER_CAPABILITY_STATEFUL_HISTORY on GetWorkItems. - Before replay, reconstruct the full history: for a delta work item (cachedHistory) prepend the cached prefix to the delta, falling back to GetInstanceHistory on any miss (cold stream, eviction, prefix-length mismatch); otherwise use the full pastEvents. - After a turn, cache the committed history (never the new events), and drop it when the turn ends the execution (a completeWorkflow action, whatever its status); reset the whole cache on stream reconnect; sweep TTL on a janitor thread. - Add a disable_stateful_history opt-out and history_cache_ttl / _max_instances / _max_bytes tuning to TaskHubGrpcWorker. Requires: https://github.com/dapr/durabletask-protobuf/pull/54 https://github.com/dapr/durabletask-go/pull/110 https://github.com/dapr/dapr/pull/10142 Signed-off-by: joshvanl --- .../internal/orchestrator_service_pb2.py | 120 ++++----- .../internal/orchestrator_service_pb2.pyi | 97 +++++++- dapr/ext/workflow/_durabletask/worker.py | 193 ++++++++++++++- .../durabletask/test_orchestration_e2e.py | 50 ++++ .../durabletask/test_worker_history_cache.py | 229 ++++++++++++++++++ 5 files changed, 616 insertions(+), 73 deletions(-) create mode 100644 tests/ext/workflow/durabletask/test_worker_history_cache.py diff --git a/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.py b/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.py index ad73d3262..443535186 100644 --- a/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.py +++ b/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.py @@ -30,7 +30,7 @@ from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1aorchestrator_service.proto\x1a\x13orchestration.proto\x1a\x14history_events.proto\x1a\x1aorchestrator_actions.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1egoogle/protobuf/wrappers.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xc6\x02\n\x0f\x41\x63tivityRequest\x12\x0c\n\x04name\x18\x01 \x01(\t\x12-\n\x07version\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x05input\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x10workflowInstance\x18\x04 \x01(\x0b\x32\x11.WorkflowInstance\x12\x0e\n\x06taskId\x18\x05 \x01(\x05\x12)\n\x12parentTraceContext\x18\x06 \x01(\x0b\x32\r.TraceContext\x12\x17\n\x0ftaskExecutionId\x18\x07 \x01(\t\x12\x32\n\x11propagatedHistory\x18\x08 \x01(\x0b\x32\x12.PropagatedHistoryH\x00\x88\x01\x01\x42\x14\n\x12_propagatedHistory\"\xaa\x01\n\x10\x41\x63tivityResponse\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x0e\n\x06taskId\x18\x02 \x01(\x05\x12,\n\x06result\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x0e\x66\x61ilureDetails\x18\x04 \x01(\x0b\x32\x13.TaskFailureDetails\x12\x17\n\x0f\x63ompletionToken\x18\x05 \x01(\t\"\xbc\x02\n\x0fWorkflowRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x31\n\x0b\x65xecutionId\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12!\n\npastEvents\x18\x03 \x03(\x0b\x32\r.HistoryEvent\x12 \n\tnewEvents\x18\x04 \x03(\x0b\x32\r.HistoryEvent\x12 \n\x18requiresHistoryStreaming\x18\x06 \x01(\x08\x12 \n\x06router\x18\x07 \x01(\x0b\x32\x0b.TaskRouterH\x00\x88\x01\x01\x12\x32\n\x11propagatedHistory\x18\x08 \x01(\x0b\x32\x12.PropagatedHistoryH\x01\x88\x01\x01\x42\t\n\x07_routerB\x14\n\x12_propagatedHistoryJ\x04\x08\x05\x10\x06\"\x82\x02\n\x10WorkflowResponse\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12 \n\x07\x61\x63tions\x18\x02 \x03(\x0b\x32\x0f.WorkflowAction\x12\x32\n\x0c\x63ustomStatus\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12\x17\n\x0f\x63ompletionToken\x18\x04 \x01(\t\x12\x37\n\x12numEventsProcessed\x18\x05 \x01(\x0b\x32\x1b.google.protobuf.Int32Value\x12&\n\x07version\x18\x06 \x01(\x0b\x32\x10.WorkflowVersionH\x00\x88\x01\x01\x42\n\n\x08_version\"\xaf\x03\n\x15\x43reateInstanceRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12-\n\x07version\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x05input\x18\x04 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12;\n\x17scheduledStartTimestamp\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x31\n\x0b\x65xecutionId\x18\x07 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12.\n\x04tags\x18\x08 \x03(\x0b\x32 .CreateInstanceRequest.TagsEntry\x12)\n\x12parentTraceContext\x18\t \x01(\x0b\x32\r.TraceContext\x1a+\n\tTagsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01J\x04\x08\x06\x10\x07R\x1aorchestrationIdReusePolicy\",\n\x16\x43reateInstanceResponse\x12\x12\n\ninstanceId\x18\x01 \x01(\t\"E\n\x12GetInstanceRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x1b\n\x13getInputsAndOutputs\x18\x02 \x01(\x08\"L\n\x13GetInstanceResponse\x12\x0e\n\x06\x65xists\x18\x01 \x01(\x08\x12%\n\rworkflowState\x18\x02 \x01(\x0b\x32\x0e.WorkflowState\"b\n\x11RaiseEventRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12+\n\x05input\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\"\x14\n\x12RaiseEventResponse\"g\n\x10TerminateRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12,\n\x06output\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12\x11\n\trecursive\x18\x03 \x01(\x08\"\x13\n\x11TerminateResponse\"R\n\x0eSuspendRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12,\n\x06reason\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\"\x11\n\x0fSuspendResponse\"Q\n\rResumeRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12,\n\x06reason\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\"\x10\n\x0eResumeResponse\"\x9e\x01\n\x15PurgeInstancesRequest\x12\x14\n\ninstanceId\x18\x01 \x01(\tH\x00\x12\x33\n\x13purgeInstanceFilter\x18\x02 \x01(\x0b\x32\x14.PurgeInstanceFilterH\x00\x12\x11\n\trecursive\x18\x03 \x01(\x08\x12\x12\n\x05\x66orce\x18\x04 \x01(\x08H\x01\x88\x01\x01\x42\t\n\x07requestB\x08\n\x06_force\"\xaa\x01\n\x13PurgeInstanceFilter\x12\x33\n\x0f\x63reatedTimeFrom\x18\x01 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x31\n\rcreatedTimeTo\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12+\n\rruntimeStatus\x18\x03 \x03(\x0e\x32\x14.OrchestrationStatus\"f\n\x16PurgeInstancesResponse\x12\x1c\n\x14\x64\x65letedInstanceCount\x18\x01 \x01(\x05\x12.\n\nisComplete\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.BoolValue\"-\n\x13GetWorkItemsRequestJ\x04\x08\x01\x10\x02J\x04\x08\x02\x10\x03J\x04\x08\x03\x10\x04J\x04\x08\n\x10\x0b\"\x9a\x01\n\x08WorkItem\x12+\n\x0fworkflowRequest\x18\x01 \x01(\x0b\x32\x10.WorkflowRequestH\x00\x12+\n\x0f\x61\x63tivityRequest\x18\x02 \x01(\x0b\x32\x10.ActivityRequestH\x00\x12\x17\n\x0f\x63ompletionToken\x18\n \x01(\tB\t\n\x07requestJ\x04\x08\x03\x10\x04J\x04\x08\x04\x10\x05J\x04\x08\x05\x10\x06\"\x16\n\x14\x43ompleteTaskResponse\"\x85\x02\n\x1dRerunWorkflowFromEventRequest\x12\x18\n\x10sourceInstanceID\x18\x01 \x01(\t\x12\x0f\n\x07\x65ventID\x18\x02 \x01(\r\x12\x1a\n\rnewInstanceID\x18\x03 \x01(\tH\x00\x88\x01\x01\x12+\n\x05input\x18\x04 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12\x16\n\x0eoverwriteInput\x18\x05 \x01(\x08\x12\'\n\x1anewChildWorkflowInstanceID\x18\x06 \x01(\tH\x01\x88\x01\x01\x42\x10\n\x0e_newInstanceIDB\x1d\n\x1b_newChildWorkflowInstanceID\"7\n\x1eRerunWorkflowFromEventResponse\x12\x15\n\rnewInstanceID\x18\x01 \x01(\t\"r\n\x16ListInstanceIDsRequest\x12\x1e\n\x11\x63ontinuationToken\x18\x01 \x01(\tH\x00\x88\x01\x01\x12\x15\n\x08pageSize\x18\x02 \x01(\rH\x01\x88\x01\x01\x42\x14\n\x12_continuationTokenB\x0b\n\t_pageSize\"d\n\x17ListInstanceIDsResponse\x12\x13\n\x0binstanceIds\x18\x01 \x03(\t\x12\x1e\n\x11\x63ontinuationToken\x18\x02 \x01(\tH\x00\x88\x01\x01\x42\x14\n\x12_continuationToken\"/\n\x19GetInstanceHistoryRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\";\n\x1aGetInstanceHistoryResponse\x12\x1d\n\x06\x65vents\x18\x01 \x03(\x0b\x32\r.HistoryEvent*^\n\x10WorkerCapability\x12!\n\x1dWORKER_CAPABILITY_UNSPECIFIED\x10\x00\x12\'\n#WORKER_CAPABILITY_HISTORY_STREAMING\x10\x01\x32\xe8\x08\n\x15TaskHubSidecarService\x12\x37\n\x05Hello\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\x12@\n\rStartInstance\x12\x16.CreateInstanceRequest\x1a\x17.CreateInstanceResponse\x12\x38\n\x0bGetInstance\x12\x13.GetInstanceRequest\x1a\x14.GetInstanceResponse\x12\x41\n\x14WaitForInstanceStart\x12\x13.GetInstanceRequest\x1a\x14.GetInstanceResponse\x12\x46\n\x19WaitForInstanceCompletion\x12\x13.GetInstanceRequest\x1a\x14.GetInstanceResponse\x12\x35\n\nRaiseEvent\x12\x12.RaiseEventRequest\x1a\x13.RaiseEventResponse\x12:\n\x11TerminateInstance\x12\x11.TerminateRequest\x1a\x12.TerminateResponse\x12\x34\n\x0fSuspendInstance\x12\x0f.SuspendRequest\x1a\x10.SuspendResponse\x12\x31\n\x0eResumeInstance\x12\x0e.ResumeRequest\x1a\x0f.ResumeResponse\x12\x41\n\x0ePurgeInstances\x12\x16.PurgeInstancesRequest\x1a\x17.PurgeInstancesResponse\x12\x31\n\x0cGetWorkItems\x12\x14.GetWorkItemsRequest\x1a\t.WorkItem0\x01\x12@\n\x14\x43ompleteActivityTask\x12\x11.ActivityResponse\x1a\x15.CompleteTaskResponse\x12I\n\x18\x43ompleteOrchestratorTask\x12\x11.WorkflowResponse\x1a\x15.CompleteTaskResponse\"\x03\x88\x02\x01\x12@\n\x14\x43ompleteWorkflowTask\x12\x11.WorkflowResponse\x1a\x15.CompleteTaskResponse\x12Y\n\x16RerunWorkflowFromEvent\x12\x1e.RerunWorkflowFromEventRequest\x1a\x1f.RerunWorkflowFromEventResponse\x12\x44\n\x0fListInstanceIDs\x12\x17.ListInstanceIDsRequest\x1a\x18.ListInstanceIDsResponse\x12M\n\x12GetInstanceHistory\x12\x1a.GetInstanceHistoryRequest\x1a\x1b.GetInstanceHistoryResponseBV\n+io.dapr.durabletask.implementation.protobufZ\x0b/api/protos\xaa\x02\x19\x44\x61pr.DurableTask.Protobufb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1aorchestrator_service.proto\x1a\x13orchestration.proto\x1a\x14history_events.proto\x1a\x1aorchestrator_actions.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1egoogle/protobuf/wrappers.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xc6\x02\n\x0f\x41\x63tivityRequest\x12\x0c\n\x04name\x18\x01 \x01(\t\x12-\n\x07version\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x05input\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x10workflowInstance\x18\x04 \x01(\x0b\x32\x11.WorkflowInstance\x12\x0e\n\x06taskId\x18\x05 \x01(\x05\x12)\n\x12parentTraceContext\x18\x06 \x01(\x0b\x32\r.TraceContext\x12\x17\n\x0ftaskExecutionId\x18\x07 \x01(\t\x12\x32\n\x11propagatedHistory\x18\x08 \x01(\x0b\x32\x12.PropagatedHistoryH\x00\x88\x01\x01\x42\x14\n\x12_propagatedHistory\"\xaa\x01\n\x10\x41\x63tivityResponse\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x0e\n\x06taskId\x18\x02 \x01(\x05\x12,\n\x06result\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x0e\x66\x61ilureDetails\x18\x04 \x01(\x0b\x32\x13.TaskFailureDetails\x12\x17\n\x0f\x63ompletionToken\x18\x05 \x01(\t\"#\n\rCachedHistory\x12\x12\n\neventCount\x18\x01 \x01(\x05\"\xfa\x02\n\x0fWorkflowRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x31\n\x0b\x65xecutionId\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12!\n\npastEvents\x18\x03 \x03(\x0b\x32\r.HistoryEvent\x12 \n\tnewEvents\x18\x04 \x03(\x0b\x32\r.HistoryEvent\x12 \n\x18requiresHistoryStreaming\x18\x06 \x01(\x08\x12 \n\x06router\x18\x07 \x01(\x0b\x32\x0b.TaskRouterH\x00\x88\x01\x01\x12\x32\n\x11propagatedHistory\x18\x08 \x01(\x0b\x32\x12.PropagatedHistoryH\x01\x88\x01\x01\x12*\n\rcachedHistory\x18\t \x01(\x0b\x32\x0e.CachedHistoryH\x02\x88\x01\x01\x42\t\n\x07_routerB\x14\n\x12_propagatedHistoryB\x10\n\x0e_cachedHistoryJ\x04\x08\x05\x10\x06\"\x82\x02\n\x10WorkflowResponse\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12 \n\x07\x61\x63tions\x18\x02 \x03(\x0b\x32\x0f.WorkflowAction\x12\x32\n\x0c\x63ustomStatus\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12\x17\n\x0f\x63ompletionToken\x18\x04 \x01(\t\x12\x37\n\x12numEventsProcessed\x18\x05 \x01(\x0b\x32\x1b.google.protobuf.Int32Value\x12&\n\x07version\x18\x06 \x01(\x0b\x32\x10.WorkflowVersionH\x00\x88\x01\x01\x42\n\n\x08_version\"\xaf\x03\n\x15\x43reateInstanceRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12-\n\x07version\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12+\n\x05input\x18\x04 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12;\n\x17scheduledStartTimestamp\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x31\n\x0b\x65xecutionId\x18\x07 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12.\n\x04tags\x18\x08 \x03(\x0b\x32 .CreateInstanceRequest.TagsEntry\x12)\n\x12parentTraceContext\x18\t \x01(\x0b\x32\r.TraceContext\x1a+\n\tTagsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01J\x04\x08\x06\x10\x07R\x1aorchestrationIdReusePolicy\",\n\x16\x43reateInstanceResponse\x12\x12\n\ninstanceId\x18\x01 \x01(\t\"E\n\x12GetInstanceRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x1b\n\x13getInputsAndOutputs\x18\x02 \x01(\x08\"L\n\x13GetInstanceResponse\x12\x0e\n\x06\x65xists\x18\x01 \x01(\x08\x12%\n\rworkflowState\x18\x02 \x01(\x0b\x32\x0e.WorkflowState\"b\n\x11RaiseEventRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12+\n\x05input\x18\x03 \x01(\x0b\x32\x1c.google.protobuf.StringValue\"\x14\n\x12RaiseEventResponse\"g\n\x10TerminateRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12,\n\x06output\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12\x11\n\trecursive\x18\x03 \x01(\x08\"\x13\n\x11TerminateResponse\"R\n\x0eSuspendRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12,\n\x06reason\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\"\x11\n\x0fSuspendResponse\"Q\n\rResumeRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\x12,\n\x06reason\x18\x02 \x01(\x0b\x32\x1c.google.protobuf.StringValue\"\x10\n\x0eResumeResponse\"\x9e\x01\n\x15PurgeInstancesRequest\x12\x14\n\ninstanceId\x18\x01 \x01(\tH\x00\x12\x33\n\x13purgeInstanceFilter\x18\x02 \x01(\x0b\x32\x14.PurgeInstanceFilterH\x00\x12\x11\n\trecursive\x18\x03 \x01(\x08\x12\x12\n\x05\x66orce\x18\x04 \x01(\x08H\x01\x88\x01\x01\x42\t\n\x07requestB\x08\n\x06_force\"\xaa\x01\n\x13PurgeInstanceFilter\x12\x33\n\x0f\x63reatedTimeFrom\x18\x01 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x31\n\rcreatedTimeTo\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12+\n\rruntimeStatus\x18\x03 \x03(\x0e\x32\x14.OrchestrationStatus\"f\n\x16PurgeInstancesResponse\x12\x1c\n\x14\x64\x65letedInstanceCount\x18\x01 \x01(\x05\x12.\n\nisComplete\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.BoolValue\"V\n\x13GetWorkItemsRequest\x12\'\n\x0c\x63\x61pabilities\x18\x04 \x03(\x0e\x32\x11.WorkerCapabilityJ\x04\x08\x01\x10\x02J\x04\x08\x02\x10\x03J\x04\x08\x03\x10\x04J\x04\x08\n\x10\x0b\"\x9a\x01\n\x08WorkItem\x12+\n\x0fworkflowRequest\x18\x01 \x01(\x0b\x32\x10.WorkflowRequestH\x00\x12+\n\x0f\x61\x63tivityRequest\x18\x02 \x01(\x0b\x32\x10.ActivityRequestH\x00\x12\x17\n\x0f\x63ompletionToken\x18\n \x01(\tB\t\n\x07requestJ\x04\x08\x03\x10\x04J\x04\x08\x04\x10\x05J\x04\x08\x05\x10\x06\"\x16\n\x14\x43ompleteTaskResponse\"\x85\x02\n\x1dRerunWorkflowFromEventRequest\x12\x18\n\x10sourceInstanceID\x18\x01 \x01(\t\x12\x0f\n\x07\x65ventID\x18\x02 \x01(\r\x12\x1a\n\rnewInstanceID\x18\x03 \x01(\tH\x00\x88\x01\x01\x12+\n\x05input\x18\x04 \x01(\x0b\x32\x1c.google.protobuf.StringValue\x12\x16\n\x0eoverwriteInput\x18\x05 \x01(\x08\x12\'\n\x1anewChildWorkflowInstanceID\x18\x06 \x01(\tH\x01\x88\x01\x01\x42\x10\n\x0e_newInstanceIDB\x1d\n\x1b_newChildWorkflowInstanceID\"7\n\x1eRerunWorkflowFromEventResponse\x12\x15\n\rnewInstanceID\x18\x01 \x01(\t\"r\n\x16ListInstanceIDsRequest\x12\x1e\n\x11\x63ontinuationToken\x18\x01 \x01(\tH\x00\x88\x01\x01\x12\x15\n\x08pageSize\x18\x02 \x01(\rH\x01\x88\x01\x01\x42\x14\n\x12_continuationTokenB\x0b\n\t_pageSize\"d\n\x17ListInstanceIDsResponse\x12\x13\n\x0binstanceIds\x18\x01 \x03(\t\x12\x1e\n\x11\x63ontinuationToken\x18\x02 \x01(\tH\x00\x88\x01\x01\x42\x14\n\x12_continuationToken\"/\n\x19GetInstanceHistoryRequest\x12\x12\n\ninstanceId\x18\x01 \x01(\t\";\n\x1aGetInstanceHistoryResponse\x12\x1d\n\x06\x65vents\x18\x01 \x03(\x0b\x32\r.HistoryEvent*\x88\x01\n\x10WorkerCapability\x12!\n\x1dWORKER_CAPABILITY_UNSPECIFIED\x10\x00\x12&\n\"WORKER_CAPABILITY_STATEFUL_HISTORY\x10\x02\"\x04\x08\x01\x10\x01*#WORKER_CAPABILITY_HISTORY_STREAMING2\xe8\x08\n\x15TaskHubSidecarService\x12\x37\n\x05Hello\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\x12@\n\rStartInstance\x12\x16.CreateInstanceRequest\x1a\x17.CreateInstanceResponse\x12\x38\n\x0bGetInstance\x12\x13.GetInstanceRequest\x1a\x14.GetInstanceResponse\x12\x41\n\x14WaitForInstanceStart\x12\x13.GetInstanceRequest\x1a\x14.GetInstanceResponse\x12\x46\n\x19WaitForInstanceCompletion\x12\x13.GetInstanceRequest\x1a\x14.GetInstanceResponse\x12\x35\n\nRaiseEvent\x12\x12.RaiseEventRequest\x1a\x13.RaiseEventResponse\x12:\n\x11TerminateInstance\x12\x11.TerminateRequest\x1a\x12.TerminateResponse\x12\x34\n\x0fSuspendInstance\x12\x0f.SuspendRequest\x1a\x10.SuspendResponse\x12\x31\n\x0eResumeInstance\x12\x0e.ResumeRequest\x1a\x0f.ResumeResponse\x12\x41\n\x0ePurgeInstances\x12\x16.PurgeInstancesRequest\x1a\x17.PurgeInstancesResponse\x12\x31\n\x0cGetWorkItems\x12\x14.GetWorkItemsRequest\x1a\t.WorkItem0\x01\x12@\n\x14\x43ompleteActivityTask\x12\x11.ActivityResponse\x1a\x15.CompleteTaskResponse\x12I\n\x18\x43ompleteOrchestratorTask\x12\x11.WorkflowResponse\x1a\x15.CompleteTaskResponse\"\x03\x88\x02\x01\x12@\n\x14\x43ompleteWorkflowTask\x12\x11.WorkflowResponse\x1a\x15.CompleteTaskResponse\x12Y\n\x16RerunWorkflowFromEvent\x12\x1e.RerunWorkflowFromEventRequest\x1a\x1f.RerunWorkflowFromEventResponse\x12\x44\n\x0fListInstanceIDs\x12\x17.ListInstanceIDsRequest\x1a\x18.ListInstanceIDsResponse\x12M\n\x12GetInstanceHistory\x12\x1a.GetInstanceHistoryRequest\x1a\x1b.GetInstanceHistoryResponseBV\n+io.dapr.durabletask.implementation.protobufZ\x0b/api/protos\xaa\x02\x19\x44\x61pr.DurableTask.Protobufb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -42,66 +42,68 @@ _globals['_CREATEINSTANCEREQUEST_TAGSENTRY']._serialized_options = b'8\001' _globals['_TASKHUBSIDECARSERVICE'].methods_by_name['CompleteOrchestratorTask']._loaded_options = None _globals['_TASKHUBSIDECARSERVICE'].methods_by_name['CompleteOrchestratorTask']._serialized_options = b'\210\002\001' - _globals['_WORKERCAPABILITY']._serialized_start=3673 - _globals['_WORKERCAPABILITY']._serialized_end=3767 + _globals['_WORKERCAPABILITY']._serialized_start=3814 + _globals['_WORKERCAPABILITY']._serialized_end=3950 _globals['_ACTIVITYREQUEST']._serialized_start=196 _globals['_ACTIVITYREQUEST']._serialized_end=522 _globals['_ACTIVITYRESPONSE']._serialized_start=525 _globals['_ACTIVITYRESPONSE']._serialized_end=695 - _globals['_WORKFLOWREQUEST']._serialized_start=698 - _globals['_WORKFLOWREQUEST']._serialized_end=1014 - _globals['_WORKFLOWRESPONSE']._serialized_start=1017 - _globals['_WORKFLOWRESPONSE']._serialized_end=1275 - _globals['_CREATEINSTANCEREQUEST']._serialized_start=1278 - _globals['_CREATEINSTANCEREQUEST']._serialized_end=1709 - _globals['_CREATEINSTANCEREQUEST_TAGSENTRY']._serialized_start=1632 - _globals['_CREATEINSTANCEREQUEST_TAGSENTRY']._serialized_end=1675 - _globals['_CREATEINSTANCERESPONSE']._serialized_start=1711 - _globals['_CREATEINSTANCERESPONSE']._serialized_end=1755 - _globals['_GETINSTANCEREQUEST']._serialized_start=1757 - _globals['_GETINSTANCEREQUEST']._serialized_end=1826 - _globals['_GETINSTANCERESPONSE']._serialized_start=1828 - _globals['_GETINSTANCERESPONSE']._serialized_end=1904 - _globals['_RAISEEVENTREQUEST']._serialized_start=1906 - _globals['_RAISEEVENTREQUEST']._serialized_end=2004 - _globals['_RAISEEVENTRESPONSE']._serialized_start=2006 - _globals['_RAISEEVENTRESPONSE']._serialized_end=2026 - _globals['_TERMINATEREQUEST']._serialized_start=2028 - _globals['_TERMINATEREQUEST']._serialized_end=2131 - _globals['_TERMINATERESPONSE']._serialized_start=2133 - _globals['_TERMINATERESPONSE']._serialized_end=2152 - _globals['_SUSPENDREQUEST']._serialized_start=2154 - _globals['_SUSPENDREQUEST']._serialized_end=2236 - _globals['_SUSPENDRESPONSE']._serialized_start=2238 - _globals['_SUSPENDRESPONSE']._serialized_end=2255 - _globals['_RESUMEREQUEST']._serialized_start=2257 - _globals['_RESUMEREQUEST']._serialized_end=2338 - _globals['_RESUMERESPONSE']._serialized_start=2340 - _globals['_RESUMERESPONSE']._serialized_end=2356 - _globals['_PURGEINSTANCESREQUEST']._serialized_start=2359 - _globals['_PURGEINSTANCESREQUEST']._serialized_end=2517 - _globals['_PURGEINSTANCEFILTER']._serialized_start=2520 - _globals['_PURGEINSTANCEFILTER']._serialized_end=2690 - _globals['_PURGEINSTANCESRESPONSE']._serialized_start=2692 - _globals['_PURGEINSTANCESRESPONSE']._serialized_end=2794 - _globals['_GETWORKITEMSREQUEST']._serialized_start=2796 - _globals['_GETWORKITEMSREQUEST']._serialized_end=2841 - _globals['_WORKITEM']._serialized_start=2844 - _globals['_WORKITEM']._serialized_end=2998 - _globals['_COMPLETETASKRESPONSE']._serialized_start=3000 - _globals['_COMPLETETASKRESPONSE']._serialized_end=3022 - _globals['_RERUNWORKFLOWFROMEVENTREQUEST']._serialized_start=3025 - _globals['_RERUNWORKFLOWFROMEVENTREQUEST']._serialized_end=3286 - _globals['_RERUNWORKFLOWFROMEVENTRESPONSE']._serialized_start=3288 - _globals['_RERUNWORKFLOWFROMEVENTRESPONSE']._serialized_end=3343 - _globals['_LISTINSTANCEIDSREQUEST']._serialized_start=3345 - _globals['_LISTINSTANCEIDSREQUEST']._serialized_end=3459 - _globals['_LISTINSTANCEIDSRESPONSE']._serialized_start=3461 - _globals['_LISTINSTANCEIDSRESPONSE']._serialized_end=3561 - _globals['_GETINSTANCEHISTORYREQUEST']._serialized_start=3563 - _globals['_GETINSTANCEHISTORYREQUEST']._serialized_end=3610 - _globals['_GETINSTANCEHISTORYRESPONSE']._serialized_start=3612 - _globals['_GETINSTANCEHISTORYRESPONSE']._serialized_end=3671 - _globals['_TASKHUBSIDECARSERVICE']._serialized_start=3770 - _globals['_TASKHUBSIDECARSERVICE']._serialized_end=4898 + _globals['_CACHEDHISTORY']._serialized_start=697 + _globals['_CACHEDHISTORY']._serialized_end=732 + _globals['_WORKFLOWREQUEST']._serialized_start=735 + _globals['_WORKFLOWREQUEST']._serialized_end=1113 + _globals['_WORKFLOWRESPONSE']._serialized_start=1116 + _globals['_WORKFLOWRESPONSE']._serialized_end=1374 + _globals['_CREATEINSTANCEREQUEST']._serialized_start=1377 + _globals['_CREATEINSTANCEREQUEST']._serialized_end=1808 + _globals['_CREATEINSTANCEREQUEST_TAGSENTRY']._serialized_start=1731 + _globals['_CREATEINSTANCEREQUEST_TAGSENTRY']._serialized_end=1774 + _globals['_CREATEINSTANCERESPONSE']._serialized_start=1810 + _globals['_CREATEINSTANCERESPONSE']._serialized_end=1854 + _globals['_GETINSTANCEREQUEST']._serialized_start=1856 + _globals['_GETINSTANCEREQUEST']._serialized_end=1925 + _globals['_GETINSTANCERESPONSE']._serialized_start=1927 + _globals['_GETINSTANCERESPONSE']._serialized_end=2003 + _globals['_RAISEEVENTREQUEST']._serialized_start=2005 + _globals['_RAISEEVENTREQUEST']._serialized_end=2103 + _globals['_RAISEEVENTRESPONSE']._serialized_start=2105 + _globals['_RAISEEVENTRESPONSE']._serialized_end=2125 + _globals['_TERMINATEREQUEST']._serialized_start=2127 + _globals['_TERMINATEREQUEST']._serialized_end=2230 + _globals['_TERMINATERESPONSE']._serialized_start=2232 + _globals['_TERMINATERESPONSE']._serialized_end=2251 + _globals['_SUSPENDREQUEST']._serialized_start=2253 + _globals['_SUSPENDREQUEST']._serialized_end=2335 + _globals['_SUSPENDRESPONSE']._serialized_start=2337 + _globals['_SUSPENDRESPONSE']._serialized_end=2354 + _globals['_RESUMEREQUEST']._serialized_start=2356 + _globals['_RESUMEREQUEST']._serialized_end=2437 + _globals['_RESUMERESPONSE']._serialized_start=2439 + _globals['_RESUMERESPONSE']._serialized_end=2455 + _globals['_PURGEINSTANCESREQUEST']._serialized_start=2458 + _globals['_PURGEINSTANCESREQUEST']._serialized_end=2616 + _globals['_PURGEINSTANCEFILTER']._serialized_start=2619 + _globals['_PURGEINSTANCEFILTER']._serialized_end=2789 + _globals['_PURGEINSTANCESRESPONSE']._serialized_start=2791 + _globals['_PURGEINSTANCESRESPONSE']._serialized_end=2893 + _globals['_GETWORKITEMSREQUEST']._serialized_start=2895 + _globals['_GETWORKITEMSREQUEST']._serialized_end=2981 + _globals['_WORKITEM']._serialized_start=2984 + _globals['_WORKITEM']._serialized_end=3138 + _globals['_COMPLETETASKRESPONSE']._serialized_start=3140 + _globals['_COMPLETETASKRESPONSE']._serialized_end=3162 + _globals['_RERUNWORKFLOWFROMEVENTREQUEST']._serialized_start=3165 + _globals['_RERUNWORKFLOWFROMEVENTREQUEST']._serialized_end=3426 + _globals['_RERUNWORKFLOWFROMEVENTRESPONSE']._serialized_start=3428 + _globals['_RERUNWORKFLOWFROMEVENTRESPONSE']._serialized_end=3483 + _globals['_LISTINSTANCEIDSREQUEST']._serialized_start=3485 + _globals['_LISTINSTANCEIDSREQUEST']._serialized_end=3599 + _globals['_LISTINSTANCEIDSRESPONSE']._serialized_start=3601 + _globals['_LISTINSTANCEIDSRESPONSE']._serialized_end=3701 + _globals['_GETINSTANCEHISTORYREQUEST']._serialized_start=3703 + _globals['_GETINSTANCEHISTORYREQUEST']._serialized_end=3750 + _globals['_GETINSTANCEHISTORYRESPONSE']._serialized_start=3752 + _globals['_GETINSTANCEHISTORYRESPONSE']._serialized_end=3811 + _globals['_TASKHUBSIDECARSERVICE']._serialized_start=3953 + _globals['_TASKHUBSIDECARSERVICE']._serialized_end=5081 # @@protoc_insertion_point(module_scope) diff --git a/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.pyi b/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.pyi index 7fa9ffde4..dcce421a4 100644 --- a/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.pyi +++ b/dapr/ext/workflow/_durabletask/internal/orchestrator_service_pb2.pyi @@ -33,21 +33,31 @@ class _WorkerCapability: class _WorkerCapabilityEnumTypeWrapper(_enum_type_wrapper._EnumTypeWrapper[_WorkerCapability.ValueType], _builtins.type): DESCRIPTOR: _descriptor.EnumDescriptor WORKER_CAPABILITY_UNSPECIFIED: _WorkerCapability.ValueType # 0 - WORKER_CAPABILITY_HISTORY_STREAMING: _WorkerCapability.ValueType # 1 - """Indicates that the worker is capable of streaming instance history as a more optimized - alternative to receiving the full history embedded in the workflow work-item. - When set, the service may return work items without any history events as an optimization. - It is strongly recommended that all SDKs support this capability. + WORKER_CAPABILITY_STATEFUL_HISTORY: _WorkerCapability.ValueType # 2 + """Indicates that the worker retains an instance's accumulated history in + memory between workflow turns on the same work-item stream, so that the + service can send only the new events (the delta) instead of the full + history each turn. When the service has dispatched a turn for an + instance to this stream and believes the stream is still warm for it, it + may set WorkflowRequest.cachedHistory and drop the committed-history + prefix the worker already holds from pastEvents, leaving only the delta + there. On a cache miss the worker recovers the full history via the + GetInstanceHistory RPC, so the optimization never affects correctness. """ class WorkerCapability(_WorkerCapability, metaclass=_WorkerCapabilityEnumTypeWrapper): ... WORKER_CAPABILITY_UNSPECIFIED: WorkerCapability.ValueType # 0 -WORKER_CAPABILITY_HISTORY_STREAMING: WorkerCapability.ValueType # 1 -"""Indicates that the worker is capable of streaming instance history as a more optimized -alternative to receiving the full history embedded in the workflow work-item. -When set, the service may return work items without any history events as an optimization. -It is strongly recommended that all SDKs support this capability. +WORKER_CAPABILITY_STATEFUL_HISTORY: WorkerCapability.ValueType # 2 +"""Indicates that the worker retains an instance's accumulated history in +memory between workflow turns on the same work-item stream, so that the +service can send only the new events (the delta) instead of the full +history each turn. When the service has dispatched a turn for an +instance to this stream and believes the stream is still warm for it, it +may set WorkflowRequest.cachedHistory and drop the committed-history +prefix the worker already holds from pastEvents, leaving only the delta +there. On a cache miss the worker recovers the full history via the +GetInstanceHistory RPC, so the optimization never affects correctness. """ Global___WorkerCapability: _TypeAlias = WorkerCapability # noqa: Y015 @@ -135,6 +145,42 @@ class ActivityResponse(_message.Message): Global___ActivityResponse: _TypeAlias = ActivityResponse # noqa: Y015 +@_typing.final +class CachedHistory(_message.Message): + """CachedHistory is set on a WorkflowRequest when the service has intentionally + omitted the committed history prefix the worker is expected to already hold + for this instance from a previous turn on the same stream (see + WORKER_CAPABILITY_STATEFUL_HISTORY). Its presence means pastEvents carries + only the delta since the worker was last brought up to date; its absence + means pastEvents is the full committed history. The worker reconstructs the + full past history by prepending its cached events to pastEvents. The service + only sets this for workers that advertised + WORKER_CAPABILITY_STATEFUL_HISTORY and that it believes to be warm for the + instance, so it is always safe for a worker to fall back to the + GetInstanceHistory RPC. + """ + + DESCRIPTOR: _descriptor.Descriptor + + EVENTCOUNT_FIELD_NUMBER: _builtins.int + eventCount: _builtins.int + """eventCount is the number of leading (committed) history events the + service believes the worker already holds, i.e. the length of the prefix + omitted from pastEvents. The worker's cached prefix must contain exactly + this many events; if it does not, the worker must treat this as a cache + miss and fetch the full history via GetInstanceHistory before applying + newEvents. + """ + def __init__( + self, + *, + eventCount: _builtins.int = ..., + ) -> None: ... + _ClearFieldArgType: _TypeAlias = _typing.Literal["eventCount", b"eventCount"] # noqa: Y015 + def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + +Global___CachedHistory: _TypeAlias = CachedHistory # noqa: Y015 + @_typing.final class WorkflowRequest(_message.Message): DESCRIPTOR: _descriptor.Descriptor @@ -146,6 +192,7 @@ class WorkflowRequest(_message.Message): REQUIRESHISTORYSTREAMING_FIELD_NUMBER: _builtins.int ROUTER_FIELD_NUMBER: _builtins.int PROPAGATEDHISTORY_FIELD_NUMBER: _builtins.int + CACHEDHISTORY_FIELD_NUMBER: _builtins.int instanceId: _builtins.str requiresHistoryStreaming: _builtins.bool @_builtins.property @@ -163,6 +210,14 @@ class WorkflowRequest(_message.Message): workflow function can access it via ctx. """ + @_builtins.property + def cachedHistory(self) -> Global___CachedHistory: + """cachedHistory, when present, signals that pastEvents holds only the + delta and the worker must reconstruct the omitted prefix from its own + cache (or fetch it via GetInstanceHistory on a miss). Absent for + full-history sends. + """ + def __init__( self, *, @@ -173,16 +228,21 @@ class WorkflowRequest(_message.Message): requiresHistoryStreaming: _builtins.bool = ..., router: _orchestration_pb2.TaskRouter | None = ..., propagatedHistory: _history_events_pb2.PropagatedHistory | None = ..., + cachedHistory: Global___CachedHistory | None = ..., ) -> None: ... - _HasFieldArgType: _TypeAlias = _typing.Literal["_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "executionId", b"executionId", "propagatedHistory", b"propagatedHistory", "router", b"router"] # noqa: Y015 + _HasFieldArgType: _TypeAlias = _typing.Literal["_cachedHistory", b"_cachedHistory", "_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "cachedHistory", b"cachedHistory", "executionId", b"executionId", "propagatedHistory", b"propagatedHistory", "router", b"router"] # noqa: Y015 def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "executionId", b"executionId", "instanceId", b"instanceId", "newEvents", b"newEvents", "pastEvents", b"pastEvents", "propagatedHistory", b"propagatedHistory", "requiresHistoryStreaming", b"requiresHistoryStreaming", "router", b"router"] # noqa: Y015 + _ClearFieldArgType: _TypeAlias = _typing.Literal["_cachedHistory", b"_cachedHistory", "_propagatedHistory", b"_propagatedHistory", "_router", b"_router", "cachedHistory", b"cachedHistory", "executionId", b"executionId", "instanceId", b"instanceId", "newEvents", b"newEvents", "pastEvents", b"pastEvents", "propagatedHistory", b"propagatedHistory", "requiresHistoryStreaming", b"requiresHistoryStreaming", "router", b"router"] # noqa: Y015 def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + _WhichOneofReturnType__cachedHistory: _TypeAlias = _typing.Literal["cachedHistory"] # noqa: Y015 + _WhichOneofArgType__cachedHistory: _TypeAlias = _typing.Literal["_cachedHistory", b"_cachedHistory"] # noqa: Y015 _WhichOneofReturnType__propagatedHistory: _TypeAlias = _typing.Literal["propagatedHistory"] # noqa: Y015 _WhichOneofArgType__propagatedHistory: _TypeAlias = _typing.Literal["_propagatedHistory", b"_propagatedHistory"] # noqa: Y015 _WhichOneofReturnType__router: _TypeAlias = _typing.Literal["router"] # noqa: Y015 _WhichOneofArgType__router: _TypeAlias = _typing.Literal["_router", b"_router"] # noqa: Y015 @_typing.overload + def WhichOneof(self, oneof_group: _WhichOneofArgType__cachedHistory) -> _WhichOneofReturnType__cachedHistory | None: ... + @_typing.overload def WhichOneof(self, oneof_group: _WhichOneofArgType__propagatedHistory) -> _WhichOneofReturnType__propagatedHistory | None: ... @_typing.overload def WhichOneof(self, oneof_group: _WhichOneofArgType__router) -> _WhichOneofReturnType__router | None: ... @@ -593,9 +653,22 @@ Global___PurgeInstancesResponse: _TypeAlias = PurgeInstancesResponse # noqa: Y0 class GetWorkItemsRequest(_message.Message): DESCRIPTOR: _descriptor.Descriptor + CAPABILITIES_FIELD_NUMBER: _builtins.int + @_builtins.property + def capabilities(self) -> _containers.RepeatedScalarFieldContainer[Global___WorkerCapability.ValueType]: + """capabilities advertises the optional protocol features this worker + supports, so the service can opt into optimizations on a per-stream + basis. Workers that leave this empty receive the default (fully + self-contained) behavior. + """ + def __init__( self, + *, + capabilities: _abc.Iterable[Global___WorkerCapability.ValueType] | None = ..., ) -> None: ... + _ClearFieldArgType: _TypeAlias = _typing.Literal["capabilities", b"capabilities"] # noqa: Y015 + def ClearField(self, field_name: _ClearFieldArgType) -> None: ... Global___GetWorkItemsRequest: _TypeAlias = GetWorkItemsRequest # noqa: Y015 diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index f7d0c04cc..bd42ec45d 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/dapr/ext/workflow/_durabletask/worker.py @@ -16,12 +16,13 @@ import os import random import threading +import time import warnings from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from threading import Event, Thread from types import GeneratorType -from typing import Any, Generator, Iterator, Optional, Sequence, TypeVar, Union +from typing import Any, Callable, Generator, Iterator, Optional, Sequence, TypeVar, Union import grpc from google.protobuf import empty_pb2, timestamp_pb2 @@ -223,6 +224,125 @@ def _is_message_too_large(rpc_error: grpc.RpcError) -> bool: # TODO: refactor this to closely match durabletask-go/client/worker_grpc.go instead of this. +_DEFAULT_HISTORY_CACHE_TTL = 3600.0 +_DEFAULT_HISTORY_CACHE_MAX_INSTANCES = 100_000 +_HISTORY_CACHE_SWEEP_INTERVAL = 60.0 + + +class _CachedHistory: + """One instance's cached committed history on a work-item stream.""" + + __slots__ = ('events', 'last_access', 'num_bytes') + + def __init__(self, events: list[pb.HistoryEvent], last_access: float, num_bytes: int): + self.events = events + self.last_access = last_access + self.num_bytes = num_bytes + + +class _WorkflowHistoryCache: + """Per-stream cache of each instance's committed history, enabling delta work items. + + A worker advertising WORKER_CAPABILITY_STATEFUL_HISTORY keeps the committed history it + replayed for each instance so the sidecar can send only the new events (the delta). + Entries are reclaimed by a sliding TTL, an instance-count cap, and an optional byte + budget (LRU eviction). Eviction is always safe: a miss is recovered via the + GetInstanceHistory RPC, so it only costs one extra fetch. + """ + + def __init__( + self, + *, + ttl: float = _DEFAULT_HISTORY_CACHE_TTL, + max_instances: int = _DEFAULT_HISTORY_CACHE_MAX_INSTANCES, + max_bytes: int = 0, + clock: Callable[[], float] = time.monotonic, + ): + self._ttl = ttl if ttl > 0 else _DEFAULT_HISTORY_CACHE_TTL + self._max_instances = ( + max_instances if max_instances > 0 else _DEFAULT_HISTORY_CACHE_MAX_INSTANCES + ) + self._max_bytes = max_bytes if max_bytes > 0 else 0 + self._clock = clock + self._lock = threading.Lock() + self._entries: dict[str, _CachedHistory] = {} + self._total_bytes = 0 + + def get(self, instance_id: str) -> Optional[list[pb.HistoryEvent]]: + """Returns an instance's cached committed history, refreshing its TTL.""" + with self._lock: + entry = self._entries.get(instance_id) + if entry is None: + return None + entry.last_access = self._clock() + return entry.events + + def put(self, instance_id: str, events: list[pb.HistoryEvent]) -> None: + """Caches an instance's committed history, evicting LRU entries to stay in bounds.""" + num_bytes = sum(event.ByteSize() for event in events) + with self._lock: + existing = self._entries.get(instance_id) + if existing is not None: + self._total_bytes -= existing.num_bytes + self._entries[instance_id] = _CachedHistory(events, self._clock(), num_bytes) + self._total_bytes += num_bytes + self._evict_to_fit(instance_id) + + def delete(self, instance_id: str) -> None: + """Drops an instance's cached history (e.g. once it completes).""" + with self._lock: + self._remove(instance_id) + + def reset(self) -> None: + """Clears the cache; used when the stream reconnects (and starts cold).""" + with self._lock: + self._entries.clear() + self._total_bytes = 0 + + def sweep_expired(self) -> None: + """Evicts entries whose last turn was longer ago than the TTL.""" + now = self._clock() + with self._lock: + expired = [ + instance_id + for instance_id, entry in self._entries.items() + if now - entry.last_access > self._ttl + ] + for instance_id in expired: + self._remove(instance_id) + + def _remove(self, instance_id: str) -> None: + entry = self._entries.pop(instance_id, None) + if entry is not None: + self._total_bytes -= entry.num_bytes + + def _evict_to_fit(self, keep: str) -> None: + """Evicts LRU entries until within the count and byte bounds. + + Always keeps the just-touched entry so the active working set is never evicted out + from under the current turn; a lone entry over the byte budget is kept (soft overage). + """ + while len(self._entries) > 1: + over_count = len(self._entries) > self._max_instances + over_bytes = self._max_bytes > 0 and self._total_bytes > self._max_bytes + if not over_count and not over_bytes: + return + victim = self._lru_except(keep) + if victim is None: + return + self._remove(victim) + + def _lru_except(self, keep: str) -> Optional[str]: + oldest_id: Optional[str] = None + oldest_access = 0.0 + for instance_id, entry in self._entries.items(): + if instance_id == keep: + continue + if oldest_id is None or entry.last_access < oldest_access: + oldest_id, oldest_access = instance_id, entry.last_access + return oldest_id + + class TaskHubGrpcWorker: """A gRPC-based worker for processing durable task orchestrations and activities. @@ -323,6 +443,10 @@ def __init__( channel_options: Optional[Sequence[tuple[str, Any]]] = None, stop_timeout: float = 30.0, keepalive_interval: float = 30.0, + disable_stateful_history: bool = False, + history_cache_ttl: float = _DEFAULT_HISTORY_CACHE_TTL, + history_cache_max_instances: int = _DEFAULT_HISTORY_CACHE_MAX_INSTANCES, + history_cache_max_bytes: int = 0, ): self._registry = _Registry() self._host_address = host_address if host_address else shared.get_default_host_address() @@ -355,6 +479,14 @@ def __init__( self._async_worker_manager = _AsyncWorkerManager(self._concurrency_options, self._logger) self._activity_executor = _ActivityExecutor(self._logger) + self._disable_stateful_history = disable_stateful_history + self._history_cache = _WorkflowHistoryCache( + ttl=history_cache_ttl, + max_instances=history_cache_max_instances, + max_bytes=history_cache_max_bytes, + ) + self._history_janitor: Optional[Thread] = None + @property def concurrency_options(self) -> ConcurrencyOptions: """Get the current concurrency options for this worker.""" @@ -392,6 +524,11 @@ def run_loop(): loop.run_until_complete(self._async_run_loop()) self._logger.info(f'Starting gRPC worker that connects to {self._host_address}') + if not self._disable_stateful_history: + self._history_janitor = Thread( + target=self._sweep_history_cache_loop, name='WorkerHistoryJanitor', daemon=True + ) + self._history_janitor.start() self._runLoop = Thread(target=run_loop, name='WorkerRunLoop') self._runLoop.start() while not self._stream_ready.wait(timeout=1): @@ -483,6 +620,9 @@ def invalidate_connection(): self._current_channel = None current_stub = None self._response_stream = None + # The sidecar drops this stream's warm set on disconnect, so start the + # next stream cold to stay in sync. + self._history_cache.reset() if current_reader_thread is not None: current_reader_thread.join(timeout=5) @@ -523,6 +663,10 @@ def should_invalidate_connection(rpc_error): assert current_stub is not None stub = current_stub get_work_items_request = pb.GetWorkItemsRequest() + if not self._disable_stateful_history: + get_work_items_request.capabilities.append( + pb.WORKER_CAPABILITY_STATEFUL_HISTORY + ) try: self._response_stream = stub.GetWorkItems(get_work_items_request) self._logger.info( @@ -885,6 +1029,49 @@ def _handle_grpc_execution_error(self, rpc_error: grpc.RpcError, request_type: s else: self._logger.exception(f'Failed to deliver {request_type} result: {rpc_error}') + def _sweep_history_cache_loop(self): + """Periodically reclaims TTL-expired history cache entries until shutdown.""" + while not self._shutdown.wait(_HISTORY_CACHE_SWEEP_INTERVAL): + self._history_cache.sweep_expired() + + def _resolve_history( + self, req: pb.WorkflowRequest, stub: stubs.TaskHubSidecarServiceStub + ) -> list[pb.HistoryEvent]: + """Resolves the full committed history to replay for a workflow work item. + + For a full send it is the request's pastEvents. For a delta send (cachedHistory) it + is the cached prefix plus the delta, recovered via GetInstanceHistory on any cache + miss (cold stream, eviction, or a prefix-length mismatch). + """ + if not req.HasField('cachedHistory'): + return list(req.pastEvents) + + cached = self._history_cache.get(req.instanceId) + if cached is not None and len(cached) == req.cachedHistory.eventCount: + return cached + list(req.pastEvents) + + response = stub.GetInstanceHistory(pb.GetInstanceHistoryRequest(instanceId=req.instanceId)) + return list(response.events) + + def _update_history_cache( + self, instance_id: str, committed_history: list[pb.HistoryEvent], actions + ) -> None: + """Refreshes the per-stream history cache after a turn. + + Caches only the committed history (never the not-yet-committed new events), and + drops the entry when the turn ends the execution. The current instance ends via a + completeWorkflow action whatever its status (completed/failed/terminated/ + continued-as-new); a terminateWorkflow action targets a different instance and is + deliberately not treated as a reset. + """ + if self._disable_stateful_history: + return + ended = any(a.WhichOneof('workflowActionType') == 'completeWorkflow' for a in actions) + if ended: + self._history_cache.delete(instance_id) + return + self._history_cache.put(instance_id, committed_history) + def _execute_orchestrator( self, req: pb.WorkflowRequest, @@ -898,9 +1085,11 @@ def _execute_orchestrator( if req.HasField('propagatedHistory') else None ) + old_events = self._resolve_history(req, stub) result = executor.execute( - req.instanceId, req.pastEvents, req.newEvents, propagated_history=propagated + req.instanceId, old_events, req.newEvents, propagated_history=propagated ) + self._update_history_cache(req.instanceId, old_events, result.actions) version = None if result.version_name: diff --git a/tests/ext/workflow/durabletask/test_orchestration_e2e.py b/tests/ext/workflow/durabletask/test_orchestration_e2e.py index 9aac1c221..07e5c9a82 100644 --- a/tests/ext/workflow/durabletask/test_orchestration_e2e.py +++ b/tests/ext/workflow/durabletask/test_orchestration_e2e.py @@ -121,6 +121,56 @@ def sequence(ctx: task.OrchestrationContext, start_val: int): assert state.serialized_custom_status is None +def _run_accumulate(disable_stateful_history: bool) -> Optional[client.WorkflowState]: + """Runs a 20-turn activity sequence and returns its terminal state. + + Each sequential activity is a distinct turn with a longer committed history, so a + stateful worker receives most turns as cached-history deltas. Requires a sidecar built + with the durabletask-go stateful-history change. + """ + + def plus_one(_: task.ActivityContext, value: int) -> int: + return value + 1 + + def accumulate(ctx: task.OrchestrationContext, start: int): + current = start + for _ in range(20): + current = yield ctx.call_activity(plus_one, input=current) + return current + + with worker.TaskHubGrpcWorker( + stop_timeout=2.0, disable_stateful_history=disable_stateful_history + ) as w: + w.add_orchestrator(accumulate) + w.add_activity(plus_one) + w.start() + + with client.TaskHubGrpcClient() as task_hub_client: + instance_id = task_hub_client.schedule_new_orchestration(accumulate, input=0) + return task_hub_client.wait_for_orchestration_completion(instance_id, timeout=30) + + +def test_stateful_history_multi_turn(): + """A multi-turn workflow completes correctly with the stateful-history delta path on. + + Deltas are exercised because the worker advertises WORKER_CAPABILITY_STATEFUL_HISTORY by + default; the reconstructed history must yield the same result a full-history run would. + Wire-level delta verification lives in the durabletask-go and dapr integration suites. + """ + state = _run_accumulate(disable_stateful_history=False) + assert state is not None + assert state.runtime_status == client.OrchestrationStatus.COMPLETED + assert state.serialized_output == json.dumps(20) + + +def test_stateful_history_disabled_matches(): + """With the optimization disabled (full history every turn) the result is identical.""" + state = _run_accumulate(disable_stateful_history=True) + assert state is not None + assert state.runtime_status == client.OrchestrationStatus.COMPLETED + assert state.serialized_output == json.dumps(20) + + def test_activity_error_handling(): def throw(_: task.ActivityContext, input: int) -> int: raise RuntimeError('Kah-BOOOOM!!!') diff --git a/tests/ext/workflow/durabletask/test_worker_history_cache.py b/tests/ext/workflow/durabletask/test_worker_history_cache.py new file mode 100644 index 000000000..b4f9e2372 --- /dev/null +++ b/tests/ext/workflow/durabletask/test_worker_history_cache.py @@ -0,0 +1,229 @@ +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the worker's stateful-history cache and history reconstruction. + +These mirror the Go reference (durabletask-go client/worker_history_test.go): the cache +bounds (TTL, instance count, byte budget) with LRU eviction, and the worker's resolution +of full vs delta work items with a GetInstanceHistory fallback on a cache miss. +""" + +from typing import cast + +import dapr.ext.workflow._durabletask.internal.orchestrator_service_pb2_grpc as stubs +import dapr.ext.workflow._durabletask.internal.protos as pb +from dapr.ext.workflow._durabletask.worker import TaskHubGrpcWorker, _WorkflowHistoryCache + + +class _Clock: + """A controllable monotonic clock for deterministic TTL tests.""" + + def __init__(self) -> None: + self.now = 0.0 + + def __call__(self) -> float: + return self.now + + +def _events(count: int) -> list[pb.HistoryEvent]: + """Events with non-zero serialized size (eventId 0 is the proto default → 0 bytes).""" + return [pb.HistoryEvent(eventId=i + 1) for i in range(count)] + + +class _FakeStub: + """A stub whose GetInstanceHistory returns a fixed history and counts calls.""" + + def __init__(self, events: list[pb.HistoryEvent]) -> None: + self._events = events + self.get_instance_history_calls = 0 + + def GetInstanceHistory(self, request: pb.GetInstanceHistoryRequest): + self.get_instance_history_calls += 1 + return pb.GetInstanceHistoryResponse(events=self._events) + + +# --- cache bounds ----------------------------------------------------------------- + + +def test_get_put_delete_reset(): + cache = _WorkflowHistoryCache() + + assert cache.get('a') is None + cache.put('a', _events(3)) + cached = cache.get('a') + assert cached is not None and len(cached) == 3 + + cache.delete('a') + assert cache.get('a') is None + + cache.put('b', _events(1)) + cache.reset() + assert cache.get('b') is None + + +def test_count_cap_evicts_lru(): + clock = _Clock() + cache = _WorkflowHistoryCache(max_instances=2, clock=clock) + + cache.put('a', _events(1)) + clock.now += 1 + cache.put('b', _events(1)) + clock.now += 1 + cache.put('c', _events(1)) # over the cap → evict LRU ('a') + + assert cache.get('a') is None + assert cache.get('b') is not None + assert cache.get('c') is not None + + +def test_byte_cap_evicts_lru(): + entry_bytes = sum(e.ByteSize() for e in _events(4)) + assert entry_bytes > 0 + clock = _Clock() + cache = _WorkflowHistoryCache(max_bytes=entry_bytes + 1, clock=clock) + + cache.put('a', _events(4)) + clock.now += 1 + cache.put('b', _events(4)) # two entries exceed the byte budget → evict LRU ('a') + + assert cache.get('a') is None + assert cache.get('b') is not None + assert cache._total_bytes <= entry_bytes + 1 + + +def test_single_oversized_entry_kept(): + cache = _WorkflowHistoryCache(max_bytes=1) + cache.put('big', _events(5)) + assert cache.get('big') is not None + + +def test_byte_accounting(): + cache = _WorkflowHistoryCache() + + cache.put('a', _events(3)) + cache.put('b', _events(2)) + assert cache._total_bytes == sum(e.ByteSize() for e in _events(3)) + sum( + e.ByteSize() for e in _events(2) + ) + + cache.put('a', _events(6)) # replace adjusts the running total to the new size + assert cache._total_bytes == sum(e.ByteSize() for e in _events(6)) + sum( + e.ByteSize() for e in _events(2) + ) + + cache.delete('a') + assert cache._total_bytes == sum(e.ByteSize() for e in _events(2)) + + cache.reset() + assert cache._total_bytes == 0 + + +def test_ttl_sweep_is_sliding(): + clock = _Clock() + cache = _WorkflowHistoryCache(ttl=60.0, clock=clock) + + cache.put('idle', _events(2)) + cache.put('active', _events(2)) + + clock.now += 120 # past the TTL... + assert cache.get('active') is not None # ...but a turn refreshes 'active' + + cache.sweep_expired() + assert cache.get('idle') is None + assert cache.get('active') is not None + + +def test_non_positive_config_uses_defaults(): + cache = _WorkflowHistoryCache(ttl=0, max_instances=-1, max_bytes=-5) + assert cache._ttl > 0 + assert cache._max_instances > 0 + assert cache._max_bytes == 0 # unlimited + + +# --- worker history resolution ---------------------------------------------------- + + +def _worker(**kwargs) -> TaskHubGrpcWorker: + return TaskHubGrpcWorker(host_address='localhost:0', **kwargs) + + +def _resolve( + worker: TaskHubGrpcWorker, req: pb.WorkflowRequest, stub: _FakeStub +) -> list[pb.HistoryEvent]: + return worker._resolve_history(req, cast(stubs.TaskHubSidecarServiceStub, stub)) + + +def test_resolve_full_send_returns_past_events(): + worker = _worker() + stub = _FakeStub(_events(99)) + req = pb.WorkflowRequest(instanceId='a', pastEvents=_events(4)) + + resolved = _resolve(worker, req, stub) + assert len(resolved) == 4 + assert stub.get_instance_history_calls == 0 + + +def test_resolve_cache_hit_reconstructs(): + worker = _worker() + worker._history_cache.put('a', _events(5)) + stub = _FakeStub(_events(99)) + req = pb.WorkflowRequest(instanceId='a', pastEvents=_events(3)) + req.cachedHistory.eventCount = 5 + + resolved = _resolve(worker, req, stub) + assert len(resolved) == 8 # cached prefix (5) + delta (3) + assert stub.get_instance_history_calls == 0 + + +def test_resolve_cache_miss_fetches_full_history(): + worker = _worker() + stub = _FakeStub(_events(7)) + req = pb.WorkflowRequest(instanceId='a', pastEvents=_events(3)) + req.cachedHistory.eventCount = 5 + + resolved = _resolve(worker, req, stub) + assert len(resolved) == 7 # recovered via GetInstanceHistory + assert stub.get_instance_history_calls == 1 + + +def test_resolve_length_mismatch_is_miss(): + worker = _worker() + worker._history_cache.put('a', _events(4)) # worker holds 4... + stub = _FakeStub(_events(7)) + req = pb.WorkflowRequest(instanceId='a', pastEvents=_events(3)) + req.cachedHistory.eventCount = 5 # ...but the sidecar expects 5 → fetch + + resolved = _resolve(worker, req, stub) + assert len(resolved) == 7 + assert stub.get_instance_history_calls == 1 + + +def test_update_cache_stores_then_evicts_on_complete(): + worker = _worker() + running = [pb.WorkflowAction(scheduleTask=pb.ScheduleTaskAction())] + worker._update_history_cache('a', _events(6), running) + assert worker._history_cache.get('a') is not None + + completed = [pb.WorkflowAction(completeWorkflow=pb.CompleteWorkflowAction())] + worker._update_history_cache('a', _events(6), completed) + assert worker._history_cache.get('a') is None + + +def test_disabled_worker_does_not_cache_and_passes_full_history(): + worker = _worker(disable_stateful_history=True) + worker._update_history_cache('a', _events(6), []) + assert worker._history_cache.get('a') is None + + stub = _FakeStub(_events(99)) + req = pb.WorkflowRequest(instanceId='a', pastEvents=_events(4)) + resolved = _resolve(worker, req, stub) + assert len(resolved) == 4 + assert stub.get_instance_history_calls == 0 From 429450f5eb886a51aba66bde65cf0ebd9557fe14 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Thu, 6 Aug 2026 11:45:32 -0300 Subject: [PATCH 2/4] Verify stateful-history delta delivery against dapr master in CI Signed-off-by: joshvanl --- .github/workflows/run-tests.yaml | 90 +++++++++++- dapr/ext/workflow/_durabletask/worker.py | 47 +++++- dapr/ext/workflow/workflow_runtime.py | 33 ++++- pyproject.toml | 2 + .../durabletask/test_worker_history_cache.py | 54 ++++++- tests/ext/workflow/test_workflow_runtime.py | 46 ++++++ .../test_workflow_stateful_history.py | 137 ++++++++++++++++++ tests/integration/workflow_observer.py | 121 ++++++++++++++++ 8 files changed, 521 insertions(+), 9 deletions(-) create mode 100644 tests/integration/test_workflow_stateful_history.py create mode 100644 tests/integration/workflow_observer.py diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index 7dde876a9..3909b4c49 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -113,19 +113,99 @@ jobs: nohup ollama serve > "$RUNNER_TEMP/ollama-serve.log" 2>&1 & timeout 60 bash -c 'until curl -fsS http://127.0.0.1:11434/api/tags > /dev/null; do sleep 1; done' ollama pull llama3.2:latest + # `dapr_head` covers behavior that only exists in daprd built from master; + # this job installs the latest release, so those tests run in + # validate-dapr-head instead. - name: Run integration tests run: | - uv run pytest tests/integration/ + uv run pytest tests/integration/ -m "not dapr_head" - name: Validate examples run: | uv run pytest tests/examples/ + # The stateful-history delta path (dapr/durabletask-go#110, reaching dapr via + # dapr/dapr#10142) is not in any release yet, so the matrix above cannot + # exercise it. This job builds daprd from master and runs the workflow suites + # against it. It also runs the vendored durabletask e2e tests, which nothing + # else does: build.yaml deselects `-m e2e`, and the validate job above only + # covers tests/integration and tests/examples. + validate-dapr-head: + runs-on: ubuntu-latest + env: + CHECKOUT_REPO: ${{ github.repository }} + CHECKOUT_REF: ${{ github.ref }} + # Distinct from the integration suite's 135xx block and the examples + # suite's 136xx block, and below the OS ephemeral range. See the "Port + # allocation" section in tests/integration/AGENTS.md. + E2E_HTTP_PORT: 13700 + E2E_GRPC_PORT: 13701 + steps: + - name: Parse repository_dispatch payload + if: github.event_name == 'repository_dispatch' + run: | + if [ ${{ github.event.client_payload.command }} = "ok-to-test" ]; then + echo "CHECKOUT_REPO=${{ github.event.client_payload.pull_head_repo }}" >> $GITHUB_ENV + echo "CHECKOUT_REF=${{ github.event.client_payload.pull_head_ref }}" >> $GITHUB_ENV + fi + + - name: Check out code + uses: actions/checkout@v7 + with: + repository: ${{ env.CHECKOUT_REPO }} + ref: ${{ env.CHECKOUT_REF }} + - name: Set up Python + uses: actions/setup-python@v7 + with: + python-version: "3.13" + - name: Install uv + uses: astral-sh/setup-uv@v8.3.0 + - name: Install dependencies + run: uv sync --frozen --all-packages --group tests + # setup-dapr-runtime builds daprd inside its own clone of dapr/dapr, so the + # toolchain has to satisfy that repo's `go` directive, not this one's. The + # action installs Go itself only when none is on PATH, and the runner ships + # one that may be older than dapr master requires, so pin it here instead. + - name: Set up Go + uses: actions/setup-go@v7 + with: + go-version: '1.26' + - name: Set up Dapr CLI + uses: dapr/.github/.github/actions/setup-dapr-cli@main + with: + github-token: ${{ secrets.GITHUB_TOKEN }} + # `dapr init` lays down the latest release, then the action overwrites + # ~/.dapr/bin/daprd with a build of `commit`. Placement and scheduler stay + # on release images, which is fine here: the whole server side of this + # feature lives in daprd's embedded durabletask-go grpcExecutor. + - name: Set up Dapr runtime from master + uses: dapr/.github/.github/actions/setup-dapr-runtime@main + with: + commit: ${{ github.event.inputs.daprdapr_commit || 'master' }} + github-token: ${{ secrets.GITHUB_TOKEN }} + - name: Run stateful-history integration tests + run: | + uv run pytest tests/integration/ -m dapr_head -v + # These drive a bare TaskHubGrpcWorker rather than the DaprTestEnvironment + # fixture, so they need a sidecar of their own and take its address from + # DAPR_GRPC_ENDPOINT. + - name: Run vendored durabletask e2e tests + env: + DAPR_GRPC_ENDPOINT: 127.0.0.1:${{ env.E2E_GRPC_PORT }} + run: | + dapr run --app-id dt-e2e \ + --dapr-grpc-port ${{ env.E2E_GRPC_PORT }} \ + --dapr-http-port ${{ env.E2E_HTTP_PORT }} \ + --dapr-internal-grpc-port 13702 \ + --metrics-port 13703 & + timeout 90 bash -c 'until curl -fsS http://127.0.0.1:${{ env.E2E_HTTP_PORT }}/v1.0/healthz/outbound > /dev/null; do sleep 1; done' + uv run pytest tests/ext/workflow/durabletask -m e2e -v + # Single stable check name to mark as required in branch protection, so the # required-checks list doesn't go stale when the Python version matrix # changes. Skipped required checks count as passing, hence `if: always()` # plus an explicit result check instead of relying on `needs` alone. validate-complete: - needs: validate + needs: [validate, validate-dapr-head] if: always() runs-on: ubuntu-latest steps: @@ -135,3 +215,9 @@ jobs: echo "validate matrix result: ${{ needs.validate.result }}" exit 1 fi + - name: Fail unless the dapr-head job succeeded + run: | + if [ "${{ needs.validate-dapr-head.result }}" != "success" ]; then + echo "validate-dapr-head result: ${{ needs.validate-dapr-head.result }}" + exit 1 + fi diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index bd42ec45d..76c167585 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/dapr/ext/workflow/_durabletask/worker.py @@ -229,6 +229,14 @@ def _is_message_too_large(rpc_error: grpc.RpcError) -> bool: _HISTORY_CACHE_SWEEP_INTERVAL = 60.0 +class _HistoryResolutionError(Exception): + """Raised when a delta work item's committed history could not be recovered. + + Distinct from an orchestrator failure: the workflow itself is healthy, we just could + not reconstruct what to replay, so the item must be re-dispatched rather than failed. + """ + + class _CachedHistory: """One instance's cached committed history on a work-item stream.""" @@ -675,6 +683,23 @@ def should_invalidate_connection(rpc_error): except Exception: raise + work_item_stream = self._response_stream + + def teardown_stream(): + """Cancels this specific work-item stream, forcing a reconnect. + + Bound to the stream the work item arrived on, so a late failure from a + superseded connection is a no-op instead of killing the current one. + Cancelling (rather than calling invalidate_connection, which mutates + listener-loop-local state) is safe from a worker thread: the reader + thread observes CANCELLED and the existing reconnect path runs, cache + reset included. + """ + try: + work_item_stream.cancel() + except Exception as cancel_error: + self._logger.debug(f'Failed to cancel work-item stream: {cancel_error}') + # Use a thread to read from the blocking gRPC stream and forward to asyncio import queue @@ -805,6 +830,7 @@ def stream_reader(): work_item.workflowRequest, stub, work_item.completionToken, + teardown_stream, ) elif work_item.HasField('activityRequest'): # Async user activities run on the event loop. Sync ones fall through @@ -1042,6 +1068,9 @@ def _resolve_history( For a full send it is the request's pastEvents. For a delta send (cachedHistory) it is the cached prefix plus the delta, recovered via GetInstanceHistory on any cache miss (cold stream, eviction, or a prefix-length mismatch). + + Raises: + _HistoryResolutionError: If the cache-miss fetch failed. """ if not req.HasField('cachedHistory'): return list(req.pastEvents) @@ -1050,7 +1079,13 @@ def _resolve_history( if cached is not None and len(cached) == req.cachedHistory.eventCount: return cached + list(req.pastEvents) - response = stub.GetInstanceHistory(pb.GetInstanceHistoryRequest(instanceId=req.instanceId)) + history_request = pb.GetInstanceHistoryRequest(instanceId=req.instanceId) + try: + response = stub.GetInstanceHistory(history_request) + except Exception as ex: + raise _HistoryResolutionError( + f"Failed to fetch the committed history for '{req.instanceId}': {ex}" + ) from ex return list(response.events) def _update_history_cache( @@ -1077,6 +1112,7 @@ def _execute_orchestrator( req: pb.WorkflowRequest, stub: stubs.TaskHubSidecarServiceStub, completionToken, + teardown_stream: Optional[Callable[[], None]] = None, ): try: executor = _OrchestrationExecutor(self._registry, self._logger) @@ -1106,6 +1142,15 @@ def _execute_orchestrator( completionToken=completionToken, version=version, ) + except _HistoryResolutionError as ex: + # The instance is healthy, we just cannot tell what to replay. Dropping the + # stream makes the sidecar cancel and re-dispatch its pending items, and the + # next (cold) stream sends a full history. Responding with a failure here + # would terminally kill the workflow over a transient fetch error. + self._logger.error(f'{ex}. Resetting the work-item stream to force a re-dispatch.') + if teardown_stream is not None: + teardown_stream() + return except Exception as ex: self._logger.exception( f"An error occurred while trying to execute instance '{req.instanceId}': {ex}" diff --git a/dapr/ext/workflow/workflow_runtime.py b/dapr/ext/workflow/workflow_runtime.py index 8d4e86505..def4da211 100644 --- a/dapr/ext/workflow/workflow_runtime.py +++ b/dapr/ext/workflow/workflow_runtime.py @@ -119,6 +119,10 @@ def __init__( maximum_thread_pool_workers: Optional[int] = None, worker_ready_timeout: Optional[float] = None, max_grpc_message_length: Optional[int] = None, + disable_stateful_history: bool = False, + history_cache_ttl: Optional[float] = None, + history_cache_max_instances: Optional[int] = None, + history_cache_max_bytes: Optional[int] = None, ): """Initializes the workflow runtime. @@ -146,6 +150,18 @@ def __init__( ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero), then the gRPC default (4 MiB). ``0`` in either source means "no opinion" and falls through to the next source. + disable_stateful_history: Stop advertising + ``WORKER_CAPABILITY_STATEFUL_HISTORY`` to the sidecar, so every + work item carries the instance's full committed history and no + history is cached. Defaults to False (the optimization is on). + history_cache_ttl: Seconds an idle instance's cached history is kept + before the janitor reclaims it. ``None`` uses the worker default + (1 hour). + history_cache_max_instances: Maximum number of instances to keep + cached histories for, LRU-evicted beyond that. ``None`` uses the + worker default (100,000). + history_cache_max_bytes: Total byte budget across cached histories, + LRU-evicted beyond that. ``None`` or ``0`` means unbounded. """ self._logger = Logger('WorkflowRuntime', logger_options) self._worker_ready_timeout = 30.0 if worker_ready_timeout is None else worker_ready_timeout @@ -166,6 +182,11 @@ def __init__( all_interceptors.extend(interceptors) all_interceptors.append(DaprClientTimeoutInterceptor()) channel_options = get_grpc_channel_options(max_grpc_message_length) + concurrency_options = worker.ConcurrencyOptions( + maximum_concurrent_activity_work_items=maximum_concurrent_activity_work_items, + maximum_concurrent_orchestration_work_items=maximum_concurrent_orchestration_work_items, + maximum_thread_pool_workers=maximum_thread_pool_workers, + ) self.__worker = worker.TaskHubGrpcWorker( host_address=uri.endpoint, metadata=metadata, @@ -174,11 +195,13 @@ def __init__( log_formatter=options.log_formatter, interceptors=all_interceptors, channel_options=channel_options, - concurrency_options=worker.ConcurrencyOptions( - maximum_concurrent_activity_work_items=maximum_concurrent_activity_work_items, - maximum_concurrent_orchestration_work_items=maximum_concurrent_orchestration_work_items, - maximum_thread_pool_workers=maximum_thread_pool_workers, - ), + concurrency_options=concurrency_options, + disable_stateful_history=disable_stateful_history, + # The history cache reads a non-positive bound as "use my default", + # so None collapses to 0 rather than duplicating the defaults here. + history_cache_ttl=history_cache_ttl or 0, + history_cache_max_instances=history_cache_max_instances or 0, + history_cache_max_bytes=history_cache_max_bytes or 0, ) def register_workflow(self, fn: Workflow, *, name: Optional[str] = None): diff --git a/pyproject.toml b/pyproject.toml index 003e54747..4bba23929 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -207,6 +207,8 @@ ignore_errors = true markers = [ 'example_dir(name): set the example directory for the dapr fixture', 'perf: timing-sensitive dispatch regression tests', + 'e2e: tests requiring a running Dapr sidecar', + 'dapr_head: tests requiring a daprd built from dapr/dapr master, not a release', ] pythonpath = ["."] asyncio_mode = "auto" diff --git a/tests/ext/workflow/durabletask/test_worker_history_cache.py b/tests/ext/workflow/durabletask/test_worker_history_cache.py index b4f9e2372..07723ca9d 100644 --- a/tests/ext/workflow/durabletask/test_worker_history_cache.py +++ b/tests/ext/workflow/durabletask/test_worker_history_cache.py @@ -18,9 +18,15 @@ from typing import cast +import pytest + import dapr.ext.workflow._durabletask.internal.orchestrator_service_pb2_grpc as stubs import dapr.ext.workflow._durabletask.internal.protos as pb -from dapr.ext.workflow._durabletask.worker import TaskHubGrpcWorker, _WorkflowHistoryCache +from dapr.ext.workflow._durabletask.worker import ( + TaskHubGrpcWorker, + _HistoryResolutionError, + _WorkflowHistoryCache, +) class _Clock: @@ -50,6 +56,19 @@ def GetInstanceHistory(self, request: pb.GetInstanceHistoryRequest): return pb.GetInstanceHistoryResponse(events=self._events) +class _FailingHistoryStub: + """A stub whose GetInstanceHistory always fails, recording any delivered response.""" + + def __init__(self) -> None: + self.responses: list[pb.WorkflowResponse] = [] + + def GetInstanceHistory(self, request: pb.GetInstanceHistoryRequest): + raise RuntimeError('sidecar unavailable') + + def CompleteOrchestratorTask(self, response: pb.WorkflowResponse) -> None: + self.responses.append(response) + + # --- cache bounds ----------------------------------------------------------------- @@ -217,6 +236,39 @@ def test_update_cache_stores_then_evicts_on_complete(): assert worker._history_cache.get('a') is None +def test_resolve_fetch_failure_raises_history_resolution_error(): + worker = _worker() + stub = _FailingHistoryStub() + req = pb.WorkflowRequest(instanceId='a', pastEvents=_events(3)) + req.cachedHistory.eventCount = 5 + + with pytest.raises(_HistoryResolutionError): + worker._resolve_history(req, cast(stubs.TaskHubSidecarServiceStub, stub)) + + +def test_fetch_failure_tears_down_stream_instead_of_failing_the_workflow(): + """A transient fetch error must re-dispatch the item, not terminally fail the instance.""" + worker = _worker() + stub = _FailingHistoryStub() + req = pb.WorkflowRequest(instanceId='a', pastEvents=_events(3)) + req.cachedHistory.eventCount = 5 + teardown_calls = 0 + + def teardown_stream() -> None: + nonlocal teardown_calls + teardown_calls += 1 + + worker._execute_orchestrator( + req, + cast(stubs.TaskHubSidecarServiceStub, stub), + 'completion-token', + teardown_stream, + ) + + assert teardown_calls == 1 + assert stub.responses == [] # no response at all, so the sidecar re-dispatches + + def test_disabled_worker_does_not_cache_and_passes_full_history(): worker = _worker(disable_stateful_history=True) worker._update_history_cache('a', _events(6), []) diff --git a/tests/ext/workflow/test_workflow_runtime.py b/tests/ext/workflow/test_workflow_runtime.py index 3d333fd77..35bb9e872 100644 --- a/tests/ext/workflow/test_workflow_runtime.py +++ b/tests/ext/workflow/test_workflow_runtime.py @@ -106,6 +106,52 @@ def test_timeout_interceptor_preserves_custom_interceptor_order(self): self.assertIsInstance(interceptors[2], DaprClientTimeoutInterceptor) +class WorkflowRuntimeStatefulHistoryTest(unittest.TestCase): + """The stateful-history options must reach the worker, not stop at the public API.""" + + def setUp(self): + listActivities.clear() + listOrchestrators.clear() + self._registry_patch = mock.patch( + 'dapr.ext.workflow._durabletask.worker._Registry', + return_value=FakeTaskHubGrpcWorker(), + ) + self._registry_patch.start() + + def tearDown(self): + mock.patch.stopall() + + def test_stateful_history_defaults_to_enabled_and_unconfigured(self): + with mock.patch( + 'dapr.ext.workflow._durabletask.worker.TaskHubGrpcWorker' + ) as mock_worker_cls: + WorkflowRuntime() + call_kwargs = mock_worker_cls.call_args[1] + + self.assertFalse(call_kwargs['disable_stateful_history']) + # 0 is the worker's "use your own default" sentinel for every bound. + self.assertEqual(call_kwargs['history_cache_ttl'], 0) + self.assertEqual(call_kwargs['history_cache_max_instances'], 0) + self.assertEqual(call_kwargs['history_cache_max_bytes'], 0) + + def test_stateful_history_options_are_forwarded(self): + with mock.patch( + 'dapr.ext.workflow._durabletask.worker.TaskHubGrpcWorker' + ) as mock_worker_cls: + WorkflowRuntime( + disable_stateful_history=True, + history_cache_ttl=120.0, + history_cache_max_instances=50, + history_cache_max_bytes=4096, + ) + call_kwargs = mock_worker_cls.call_args[1] + + self.assertTrue(call_kwargs['disable_stateful_history']) + self.assertEqual(call_kwargs['history_cache_ttl'], 120.0) + self.assertEqual(call_kwargs['history_cache_max_instances'], 50) + self.assertEqual(call_kwargs['history_cache_max_bytes'], 4096) + + class WorkflowRuntimeTest(unittest.TestCase): def setUp(self): listActivities.clear() diff --git a/tests/integration/test_workflow_stateful_history.py b/tests/integration/test_workflow_stateful_history.py new file mode 100644 index 000000000..aeacd7b9d --- /dev/null +++ b/tests/integration/test_workflow_stateful_history.py @@ -0,0 +1,137 @@ +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Wire-level verification of stateful-history (delta) work-item delivery. + +Requires a sidecar whose embedded durabletask-go grpcExecutor implements the feature +(dapr/durabletask-go#110, on dapr master since dapr/dapr#10142). Against an older sidecar +the capability is ignored and every turn arrives as a full send, which is exactly what +``test_delta_delivery_reduces_full_sends`` is written to catch. + +Asserting on workflow output alone would prove nothing here: a correct delta path and a +sidecar that never sends deltas produce identical results. The counts come from a gRPC +interceptor watching the real work-item stream. +""" + +import json + +import pytest + +from dapr.ext.workflow import ( + DaprWorkflowContext, + WorkflowActivityContext, + WorkflowRuntime, + WorkflowState, + WorkflowStatus, +) +from dapr.ext.workflow.dapr_workflow_client import DaprWorkflowClient +from tests.integration.workflow_observer import DeliveryCounts, WorkItemObserver + +# The released sidecar the default `validate` job installs predates the feature and would +# fail the delta assertions below, so only the dapr-head job runs this module. +pytestmark = pytest.mark.dapr_head + +HOST = '127.0.0.1' +GRPC_PORT = '13501' +WORKFLOW_TURNS = 20 +COMPLETION_TIMEOUT_S = 60 + +# The sidecar records how much history a stream holds only *after* rewriting a work item, +# so the first turn (empty past) leaves the watermark at zero and the second still fails +# the "worker holds something" check. Deltas therefore start at the third turn. +MAX_WARMUP_FULL_SENDS = 2 + + +@pytest.fixture(scope='module') +def sidecar(dapr_env): + return dapr_env.start_sidecar(app_id='test-workflow-stateful-history') + + +def _run_accumulate( + sidecar, *, disable_stateful_history: bool +) -> tuple[DeliveryCounts, WorkflowState]: + """Runs the chain on a fresh runtime and returns its delivery counts and final state. + + A new runtime per run means a new work-item stream, so the sidecar's warm set starts + empty and the counts describe this run alone. The workflow and activity are defined + per run because ``register_workflow`` stamps the function object, so the same callable + cannot be registered against a second runtime. + """ + + def plus_one(ctx: WorkflowActivityContext, value: int) -> int: + return value + 1 + + def accumulate(ctx: DaprWorkflowContext, start: int): + """A long sequential chain, so each activity result is its own turn. + + The committed history grows on every turn, which is what makes the omitted + prefix (and therefore the delta) large enough to be worth measuring. + """ + current = start + for _ in range(WORKFLOW_TURNS): + current = yield ctx.call_activity(plus_one, input=current) + return current + + observer = WorkItemObserver() + runtime = WorkflowRuntime( + host=HOST, + port=GRPC_PORT, + interceptors=[observer], + disable_stateful_history=disable_stateful_history, + ) + runtime.register_workflow(accumulate) + runtime.register_activity(plus_one) + runtime.start() + try: + workflow_client = DaprWorkflowClient(host=HOST, port=GRPC_PORT) + instance_id = workflow_client.schedule_new_workflow(accumulate, input=0) + state = workflow_client.wait_for_workflow_completion( + instance_id, timeout_in_seconds=COMPLETION_TIMEOUT_S + ) + finally: + runtime.shutdown() + + assert state is not None, 'workflow did not reach a terminal state' + return observer.counts_for(instance_id), state + + +def _assert_completed(state: WorkflowState) -> None: + assert state.runtime_status == WorkflowStatus.COMPLETED + assert state.serialized_output == json.dumps(WORKFLOW_TURNS) + + +def test_delta_delivery_reduces_full_sends(sidecar): + """Most turns must arrive as deltas, with only the unavoidable warm-up full sends.""" + counts, state = _run_accumulate(sidecar, disable_stateful_history=False) + + _assert_completed(state) + assert counts.deltas > 0, f'sidecar never sent a delta: {counts}' + assert counts.full_sends <= MAX_WARMUP_FULL_SENDS, f'too many full sends: {counts}' + assert counts.deltas >= WORKFLOW_TURNS - MAX_WARMUP_FULL_SENDS, ( + f'expected a delta for nearly every turn: {counts}' + ) + + +def test_warm_stream_never_misses_its_cache(sidecar): + """A steady stream should reconstruct every delta locally, never refetching history.""" + counts, state = _run_accumulate(sidecar, disable_stateful_history=False) + + _assert_completed(state) + assert counts.history_fetches == 0, f'unexpected GetInstanceHistory recovery: {counts}' + + +def test_disabled_receives_only_full_histories(sidecar): + """With the capability withheld, the sidecar must fall back to full sends throughout.""" + counts, state = _run_accumulate(sidecar, disable_stateful_history=True) + + _assert_completed(state) + assert counts.deltas == 0, f'delta sent to a worker that never advertised support: {counts}' + assert counts.full_sends >= WORKFLOW_TURNS, f'expected a full send per turn: {counts}' diff --git a/tests/integration/workflow_observer.py b/tests/integration/workflow_observer.py new file mode 100644 index 000000000..13c48db81 --- /dev/null +++ b/tests/integration/workflow_observer.py @@ -0,0 +1,121 @@ +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A gRPC interceptor that observes how the sidecar delivers workflow history. + +Mirrors the Go equivalent in dapr's integration framework +(``tests/integration/framework/process/workflow/worker.go``): every ``WorkflowRequest`` +arrives either as a delta (``cachedHistory`` set, ``pastEvents`` carrying only the events +since the worker was last brought up to date) or as a full send, and every +``GetInstanceHistory`` call is a cache miss the worker had to recover from. + +Without this, an e2e test cannot tell a working delta path from a sidecar that ignored +``WORKER_CAPABILITY_STATEFUL_HISTORY`` entirely: both produce identical workflow output. +""" + +import threading +from typing import Any, Iterator + +import grpc + +_GET_WORK_ITEMS = '/TaskHubSidecarService/GetWorkItems' +_GET_INSTANCE_HISTORY = '/TaskHubSidecarService/GetInstanceHistory' + + +class DeliveryCounts: + """How one instance's work items were delivered.""" + + __slots__ = ('deltas', 'full_sends', 'history_fetches') + + def __init__(self) -> None: + self.deltas = 0 + self.full_sends = 0 + self.history_fetches = 0 + + def __repr__(self) -> str: + return ( + f'DeliveryCounts(deltas={self.deltas}, full_sends={self.full_sends}, ' + f'history_fetches={self.history_fetches})' + ) + + +class _ObservedStream: + """Wraps a work-item stream so each item is counted as it is read. + + Delegates every other attribute to the underlying call: the worker's failure path + calls ``cancel()`` on the stream object, which a bare generator would not expose. + """ + + def __init__(self, call: Any, observer: 'WorkItemObserver') -> None: + self._call = call + self._observer = observer + + def __getattr__(self, name: str) -> Any: + return getattr(self._call, name) + + def __iter__(self) -> Iterator[Any]: + return self + + def __next__(self) -> Any: + work_item = next(self._call) + self._observer.record_work_item(work_item) + return work_item + + +class WorkItemObserver(grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor): + """Counts delta vs full work-item delivery, per instance and in total. + + Counters are read from the test's thread while the worker's reader thread writes them, + hence the lock. Pass an instance via ``WorkflowRuntime(interceptors=[observer])``. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._per_instance: dict[str, DeliveryCounts] = {} + + def counts_for(self, instance_id: str) -> DeliveryCounts: + """Returns a snapshot of one instance's delivery counts.""" + with self._lock: + counts = self._per_instance.get(instance_id) + snapshot = DeliveryCounts() + if counts is not None: + snapshot.deltas = counts.deltas + snapshot.full_sends = counts.full_sends + snapshot.history_fetches = counts.history_fetches + return snapshot + + def record_work_item(self, work_item: Any) -> None: + """Classifies a received work item as a delta or a full history send.""" + if not work_item.HasField('workflowRequest'): + return + request = work_item.workflowRequest + with self._lock: + counts = self._per_instance.setdefault(request.instanceId, DeliveryCounts()) + if request.HasField('cachedHistory'): + counts.deltas += 1 + return + counts.full_sends += 1 + + def _record_history_fetch(self, instance_id: str) -> None: + with self._lock: + counts = self._per_instance.setdefault(instance_id, DeliveryCounts()) + counts.history_fetches += 1 + + def intercept_unary_stream(self, continuation, client_call_details, request): + call = continuation(client_call_details, request) + if client_call_details.method != _GET_WORK_ITEMS: + return call + return _ObservedStream(call, self) + + def intercept_unary_unary(self, continuation, client_call_details, request): + if client_call_details.method == _GET_INSTANCE_HISTORY: + self._record_history_fetch(request.instanceId) + return continuation(client_call_details, request) From 8dd82dbfe971800624ce0eb5d55f2c1adbcac2e7 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Thu, 6 Aug 2026 12:08:28 -0300 Subject: [PATCH 3/4] Map deadline-expiry CANCELLED to TimeoutError in workflow clients test_orchestration_e2e_async.py::test_suspend_and_resume failed in the new validate-dapr-head job. The test asks for a bounded wait it expects to expire: state = await client.wait_for_orchestration_completion(id, timeout=3) assert False, 'Orchestration should not have completed' except TimeoutError: It got AioRpcError(CANCELLED, "Received RST_STREAM with error code 8") instead. Both clients map only DEADLINE_EXCEEDED to _TransientTimeout, so when the sidecar resets the stream marginally before gRPC raises the deadline locally, the same expiry escapes as a raw gRPC error. A caller that asked for a timeout should see TimeoutError under either code. _is_deadline_cancellation now covers that case in the sync and async retry helpers. It converts only once the caller's budget is spent, so a genuine cancellation, or a reset with time still on the clock, keeps propagating untouched. Unbounded waits (timeout=0) have no deadline to attribute a cancellation to and are left alone. This is a latent bug in shipped code rather than a regression. The vendored durabletask e2e suite ran in no CI job before this branch enabled it, so nothing had ever exercised the path. Also extracted the work-item stream teardown into _make_stream_teardown. It was an inline closure in the listener loop, which is unreachable from unit tests, leaving the cancel-failure branch uncovered. It now has direct tests for cancelling its own stream and for swallowing a failed cancel on an already-dead stream, plus one for the sweep loop's shutdown exit. Signed-off-by: joshvanl --- dapr/ext/workflow/_durabletask/aio/client.py | 3 ++ dapr/ext/workflow/_durabletask/client.py | 16 ++++++ dapr/ext/workflow/_durabletask/worker.py | 36 ++++++++------ .../workflow/durabletask/test_client_async.py | 49 ++++++++++++++++++- .../durabletask/test_orchestration_wait.py | 42 ++++++++++++++++ .../durabletask/test_worker_history_cache.py | 42 ++++++++++++++++ 6 files changed, 171 insertions(+), 17 deletions(-) diff --git a/dapr/ext/workflow/_durabletask/aio/client.py b/dapr/ext/workflow/_durabletask/aio/client.py index fe6a3ab53..a5da39b14 100644 --- a/dapr/ext/workflow/_durabletask/aio/client.py +++ b/dapr/ext/workflow/_durabletask/aio/client.py @@ -33,6 +33,7 @@ ) from dapr.ext.workflow._durabletask.client import ( OrchestrationStatus, + TaskHubGrpcClient, TInput, TOutput, WorkflowIdReusePolicy, @@ -229,6 +230,8 @@ async def _call_with_transient_retry(self, instance_id, timeout, call_fn): code = rpc_error.code() # type: ignore if code == grpc.StatusCode.DEADLINE_EXCEEDED: raise _TransientTimeout() + if TaskHubGrpcClient._is_deadline_cancellation(code, deadline): + raise _TransientTimeout() if code not in self._TRANSIENT_RPC_CODES: raise diff --git a/dapr/ext/workflow/_durabletask/client.py b/dapr/ext/workflow/_durabletask/client.py index 86c54fd47..07e63d4a7 100644 --- a/dapr/ext/workflow/_durabletask/client.py +++ b/dapr/ext/workflow/_durabletask/client.py @@ -307,6 +307,20 @@ def _call(grpc_timeout): # the long-poll), so its indefinite wait is preserved. _MAX_TRANSIENT_RETRY_SECONDS = 30.0 + @staticmethod + def _is_deadline_cancellation(code, deadline: Optional[float]) -> bool: + """Reports whether a CANCELLED status is really the caller's deadline expiring. + + As a bounded wait times out, the sidecar can reset the stream (RST_STREAM + CANCEL) marginally before gRPC raises DEADLINE_EXCEEDED locally, so the same + expiry surfaces as CANCELLED. Callers asked for a timeout and must see + TimeoutError either way. Requiring the budget to be spent keeps a genuine + cancellation, or a reset with time still on the clock, propagating as-is. + """ + if code != grpc.StatusCode.CANCELLED or deadline is None: + return False + return time.monotonic() >= deadline + def _call_with_transient_retry(self, instance_id, timeout, call_fn): """Run a gRPC wait call, retrying transient errors until the user timeout deadline. Re-raises non-transient errors immediately. @@ -334,6 +348,8 @@ def _call_with_transient_retry(self, instance_id, timeout, call_fn): code = rpc_error.code() # type: ignore if code == grpc.StatusCode.DEADLINE_EXCEEDED: raise _TransientTimeout() + if self._is_deadline_cancellation(code, deadline): + raise _TransientTimeout() if code not in self._TRANSIENT_RPC_CODES: raise diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index 76c167585..8b2d7fbb7 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/dapr/ext/workflow/_durabletask/worker.py @@ -683,22 +683,7 @@ def should_invalidate_connection(rpc_error): except Exception: raise - work_item_stream = self._response_stream - - def teardown_stream(): - """Cancels this specific work-item stream, forcing a reconnect. - - Bound to the stream the work item arrived on, so a late failure from a - superseded connection is a no-op instead of killing the current one. - Cancelling (rather than calling invalidate_connection, which mutates - listener-loop-local state) is safe from a worker thread: the reader - thread observes CANCELLED and the existing reconnect path runs, cache - reset included. - """ - try: - work_item_stream.cancel() - except Exception as cancel_error: - self._logger.debug(f'Failed to cancel work-item stream: {cancel_error}') + teardown_stream = self._make_stream_teardown(self._response_stream) # Use a thread to read from the blocking gRPC stream and forward to asyncio import queue @@ -1055,6 +1040,25 @@ def _handle_grpc_execution_error(self, rpc_error: grpc.RpcError, request_type: s else: self._logger.exception(f'Failed to deliver {request_type} result: {rpc_error}') + def _make_stream_teardown(self, stream) -> Callable[[], None]: + """Builds a callable that cancels one work-item stream, forcing a reconnect. + + Bound to the stream a work item arrived on, so a late failure from a superseded + connection is a no-op instead of killing the current one. Cancelling (rather than + invalidating the connection, which mutates listener-loop-local state) is safe from + a worker thread: the reader thread observes CANCELLED and the existing reconnect + path runs, history-cache reset included. Cancelling an already-dead stream is + expected and must not propagate. + """ + + def teardown_stream() -> None: + try: + stream.cancel() + except Exception as cancel_error: + self._logger.debug(f'Failed to cancel work-item stream: {cancel_error}') + + return teardown_stream + def _sweep_history_cache_loop(self): """Periodically reclaims TTL-expired history cache entries until shutdown.""" while not self._shutdown.wait(_HISTORY_CACHE_SWEEP_INTERVAL): diff --git a/tests/ext/workflow/durabletask/test_client_async.py b/tests/ext/workflow/durabletask/test_client_async.py index ec56c21b4..b23c69af6 100644 --- a/tests/ext/workflow/durabletask/test_client_async.py +++ b/tests/ext/workflow/durabletask/test_client_async.py @@ -9,7 +9,11 @@ # See the License for the specific language governing permissions and # limitations under the License. -from unittest.mock import patch +import asyncio +from unittest.mock import Mock, patch + +import grpc +import pytest from dapr.ext.workflow._durabletask.aio.client import AsyncTaskHubGrpcClient from dapr.ext.workflow._durabletask.aio.internal.grpc_interceptor import ( @@ -23,6 +27,13 @@ INTERCEPTORS_AIO = [DefaultClientInterceptorImpl(METADATA)] +def _make_async_rpc_error(code: grpc.StatusCode) -> grpc.RpcError: + err = grpc.RpcError() + err.code = lambda: code # type: ignore[method-assign] + err.details = lambda: f'simulated {code.name}' # type: ignore[method-assign] + return err + + def test_get_grpc_aio_channel_insecure(): with patch( 'dapr.ext.workflow._durabletask.aio.internal.shared.grpc_aio.insecure_channel' @@ -211,3 +222,39 @@ def test_aio_channel_passes_base_options_and_max_lengths(): assert ('grpc.max_send_message_length', 4321) in opts assert ('grpc.max_receive_message_length', 8765) in opts assert ('grpc.primary_user_agent', 'durabletask-aio-tests') in opts + + +async def test_cancelled_after_deadline_surfaces_as_timeout(): + """Async mirror of the sync deadline-cancellation mapping. + + This is the path that actually failed in CI against daprd from master + (test_orchestration_e2e_async.py::test_suspend_and_resume): the bounded wait + expired as CANCELLED rather than DEADLINE_EXCEEDED and escaped as a raw + AioRpcError instead of TimeoutError. + """ + + async def cancel_after_budget_spent(*args, **kwargs): + await asyncio.sleep(0.05) # outlast the caller's budget, as a real expiry would + raise _make_async_rpc_error(grpc.StatusCode.CANCELLED) + + client = AsyncTaskHubGrpcClient() + client._stub = Mock() + client._stub.WaitForInstanceCompletion = cancel_after_budget_spent + + with pytest.raises(TimeoutError): + await client.wait_for_orchestration_completion('test-instance', timeout=0.01) + + +async def test_cancelled_within_deadline_still_propagates(): + """A CANCELLED with budget remaining is a real cancellation, not a timeout.""" + + async def cancel_immediately(*args, **kwargs): + raise _make_async_rpc_error(grpc.StatusCode.CANCELLED) + + client = AsyncTaskHubGrpcClient() + client._stub = Mock() + client._stub.WaitForInstanceCompletion = cancel_immediately + + with pytest.raises(grpc.RpcError) as exc_info: + await client.wait_for_orchestration_completion('test-instance', timeout=300) + assert not isinstance(exc_info.value, TimeoutError) diff --git a/tests/ext/workflow/durabletask/test_orchestration_wait.py b/tests/ext/workflow/durabletask/test_orchestration_wait.py index c2f4f5e3e..99a7ce65b 100644 --- a/tests/ext/workflow/durabletask/test_orchestration_wait.py +++ b/tests/ext/workflow/durabletask/test_orchestration_wait.py @@ -187,3 +187,45 @@ def fake_monotonic(): # Retried at least once before giving up (proves it didn't fail-fast like the # non-transient path, and didn't loop forever). assert c._stub.WaitForInstanceStart.call_count >= 2 + + +def test_cancelled_after_deadline_surfaces_as_timeout(): + """A bounded wait must raise TimeoutError even if expiry arrives as CANCELLED. + + The sidecar can reset the stream (RST_STREAM CANCEL) just before gRPC raises + DEADLINE_EXCEEDED locally, so the same expiry reaches the client under either + code. Observed in CI against daprd from master; see + test_orchestration_e2e_async.py::test_suspend_and_resume. + """ + def cancel_after_budget_spent(*args, **kwargs): + time.sleep(0.05) # outlast the caller's budget, as a real expiry would + raise _make_rpc_error(grpc.StatusCode.CANCELLED) + + c = TaskHubGrpcClient() + c._stub = Mock() + c._stub.WaitForInstanceCompletion.side_effect = cancel_after_budget_spent + + with pytest.raises(TimeoutError): + c.wait_for_orchestration_completion('test-instance', timeout=0.01) + + +def test_cancelled_within_deadline_still_propagates(): + """A CANCELLED with budget remaining is a real cancellation, not a timeout.""" + c = TaskHubGrpcClient() + c._stub = Mock() + c._stub.WaitForInstanceCompletion.side_effect = _make_rpc_error(grpc.StatusCode.CANCELLED) + + with pytest.raises(grpc.RpcError) as exc_info: + c.wait_for_orchestration_completion('test-instance', timeout=300) + assert not isinstance(exc_info.value, TimeoutError) + + +def test_cancelled_unbounded_wait_still_propagates(): + """With no caller deadline there is nothing to attribute a CANCELLED to.""" + c = TaskHubGrpcClient() + c._stub = Mock() + c._stub.WaitForInstanceCompletion.side_effect = _make_rpc_error(grpc.StatusCode.CANCELLED) + + with pytest.raises(grpc.RpcError) as exc_info: + c.wait_for_orchestration_completion('test-instance', timeout=0) + assert not isinstance(exc_info.value, TimeoutError) diff --git a/tests/ext/workflow/durabletask/test_worker_history_cache.py b/tests/ext/workflow/durabletask/test_worker_history_cache.py index 07723ca9d..ec4b57e1b 100644 --- a/tests/ext/workflow/durabletask/test_worker_history_cache.py +++ b/tests/ext/workflow/durabletask/test_worker_history_cache.py @@ -279,3 +279,45 @@ def test_disabled_worker_does_not_cache_and_passes_full_history(): resolved = _resolve(worker, req, stub) assert len(resolved) == 4 assert stub.get_instance_history_calls == 0 + + +class _RecordingStream: + """A work-item stream that records cancellation, optionally failing it.""" + + def __init__(self, raises: bool = False) -> None: + self.cancelled = 0 + self._raises = raises + + def cancel(self) -> None: + self.cancelled += 1 + if self._raises: + raise RuntimeError('stream already dead') + + +def test_stream_teardown_cancels_its_own_stream(): + worker = _worker() + stream = _RecordingStream() + + worker._make_stream_teardown(stream)() + + assert stream.cancelled == 1 + + +def test_stream_teardown_swallows_cancel_failure(): + """Cancelling an already-dead stream is expected and must not propagate.""" + worker = _worker() + stream = _RecordingStream(raises=True) + + worker._make_stream_teardown(stream)() # must not raise + + assert stream.cancelled == 1 + + +def test_sweep_loop_exits_on_shutdown(): + worker = _worker() + worker._history_cache.put('a', _events(2)) + worker._shutdown.set() + + worker._sweep_history_cache_loop() # returns immediately rather than sleeping + + assert worker._history_cache.get('a') is not None From 72744ed02a709365f7860ddd6890ef2db7ebdcbd Mon Sep 17 00:00:00 2001 From: joshvanl Date: Thu, 6 Aug 2026 12:27:02 -0300 Subject: [PATCH 4/4] lint Signed-off-by: joshvanl --- tests/ext/workflow/durabletask/test_orchestration_wait.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/ext/workflow/durabletask/test_orchestration_wait.py b/tests/ext/workflow/durabletask/test_orchestration_wait.py index 99a7ce65b..2a2ab3560 100644 --- a/tests/ext/workflow/durabletask/test_orchestration_wait.py +++ b/tests/ext/workflow/durabletask/test_orchestration_wait.py @@ -197,6 +197,7 @@ def test_cancelled_after_deadline_surfaces_as_timeout(): code. Observed in CI against daprd from master; see test_orchestration_e2e_async.py::test_suspend_and_resume. """ + def cancel_after_budget_spent(*args, **kwargs): time.sleep(0.05) # outlast the caller's budget, as a real expiry would raise _make_rpc_error(grpc.StatusCode.CANCELLED)