diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index d5449689..5acd7412 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -224,3 +224,17 @@ works correctly: - If a new public API is added to the core SDK (e.g. a method on `OrchestrationContext`), confirm it is accessible through the azuremanaged package and add a test or example if appropriate. + +## Release Coordination + +- Package versions and inter-package minimum dependencies record the latest + released compatibility contract. They are not an instruction to bump versions + in a feature PR. +- When a core `durabletask` API is consumed by `durabletask.azuremanaged` or + `azure-functions-durable`, do not update package versions or dependency + minimums in the feature PR. Document the affected package changelogs and + release coordination instead. +- Create a dedicated release PR after the core package has been released. The + pipeline will publish `durabletask` first, so bump all dependency minimums, + package versions, and release notes together so resolvers cannot select a + provider release with an incompatible older core package. diff --git a/CHANGELOG.md b/CHANGELOG.md index 572ff4a0..208b1068 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED +- Added optional `reason` parameters to `TaskHubGrpcClient` and +`AsyncTaskHubGrpcClient` suspend and resume operations. The reason is now sent +to the backend with the lifecycle request. - Added the optional `new_version` argument to `OrchestrationContext.continue_as_new()` so continued orchestrations can switch to a new version. diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index f5e4adf3..d7752196 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +> [!NOTE] +> Release this change only in coordination with a new `durabletask` release +> that contains the suspend/resume reason APIs. Publish `durabletask` first, +> then update this package's minimum dependency in its dedicated release PR. + +FIXED + +- Fixed deprecated `DurableFunctionsClient.suspend()` and `resume()` methods +discarding their `reason` arguments. Reasons are now forwarded to the Durable +Task backend. + ## v2.0.0b2 ADDED diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index d8f478ad..7c6cb039 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/azure-functions-durable/azure/durable_functions/client.py @@ -431,21 +431,13 @@ async def purge_instance_history(self, instance_id: str) -> PurgeHistoryResult: @deprecated("suspend is deprecated; use suspend_orchestration instead.") async def suspend(self, instance_id: str, reason: Optional[str] = None) -> None: - """Deprecated alias for :meth:`suspend_orchestration`. - - The v1 ``reason`` argument has no equivalent in durabletask and is - ignored. - """ - await self.suspend_orchestration(instance_id) + """Deprecated alias for :meth:`suspend_orchestration`.""" + await self.suspend_orchestration(instance_id, reason=reason) @deprecated("resume is deprecated; use resume_orchestration instead.") async def resume(self, instance_id: str, reason: Optional[str] = None) -> None: - """Deprecated alias for :meth:`resume_orchestration`. - - The v1 ``reason`` argument has no equivalent in durabletask and is - ignored. - """ - await self.resume_orchestration(instance_id) + """Deprecated alias for :meth:`resume_orchestration`.""" + await self.resume_orchestration(instance_id, reason=reason) @deprecated("restart is deprecated; use restart_orchestration instead.") async def restart( diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index dde10137..9a52d5ef 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- Added optional `reason` parameters to suspend and resume operations inherited +from `TaskHubGrpcClient` and `AsyncTaskHubGrpcClient`. + ## v1.9.0 CHANGED diff --git a/durabletask/client.py b/durabletask/client.py index a5974749..33ee4e13 100644 --- a/durabletask/client.py +++ b/durabletask/client.py @@ -786,13 +786,21 @@ def terminate_orchestration(self, instance_id: str, *, ) self._stub.TerminateInstance(req) - def suspend_orchestration(self, instance_id: str) -> None: - req = pb.SuspendRequest(instanceId=instance_id) + def suspend_orchestration(self, instance_id: str, *, + reason: str | None = None) -> None: + req = pb.SuspendRequest( + instanceId=instance_id, + reason=helpers.get_string_value(reason), + ) self._logger.info(f"Suspending instance '{instance_id}'.") self._stub.SuspendInstance(req) - def resume_orchestration(self, instance_id: str) -> None: - req = pb.ResumeRequest(instanceId=instance_id) + def resume_orchestration(self, instance_id: str, *, + reason: str | None = None) -> None: + req = pb.ResumeRequest( + instanceId=instance_id, + reason=helpers.get_string_value(reason), + ) self._logger.info(f"Resuming instance '{instance_id}'.") self._stub.ResumeInstance(req) @@ -1320,13 +1328,21 @@ async def terminate_orchestration(self, instance_id: str, *, ) await self._get_stub().TerminateInstance(req) - async def suspend_orchestration(self, instance_id: str) -> None: - req = pb.SuspendRequest(instanceId=instance_id) + async def suspend_orchestration(self, instance_id: str, *, + reason: str | None = None) -> None: + req = pb.SuspendRequest( + instanceId=instance_id, + reason=helpers.get_string_value(reason), + ) self._logger.info(f"Suspending instance '{instance_id}'.") await self._get_stub().SuspendInstance(req) - async def resume_orchestration(self, instance_id: str) -> None: - req = pb.ResumeRequest(instanceId=instance_id) + async def resume_orchestration(self, instance_id: str, *, + reason: str | None = None) -> None: + req = pb.ResumeRequest( + instanceId=instance_id, + reason=helpers.get_string_value(reason), + ) self._logger.info(f"Resuming instance '{instance_id}'.") await self._get_stub().ResumeInstance(req) diff --git a/durabletask/internal/helpers.py b/durabletask/internal/helpers.py index bd2f46cc..341d8064 100644 --- a/durabletask/internal/helpers.py +++ b/durabletask/internal/helpers.py @@ -234,19 +234,23 @@ def new_event_raised_event(name: str, encoded_input: str | None = None) -> pb.Hi ) -def new_suspend_event() -> pb.HistoryEvent: +def new_suspend_event(*, encoded_input: str | None = None) -> pb.HistoryEvent: return pb.HistoryEvent( eventId=-1, timestamp=timestamp_pb2.Timestamp(), - executionSuspended=pb.ExecutionSuspendedEvent() + executionSuspended=pb.ExecutionSuspendedEvent( + input=get_string_value(encoded_input) + ) ) -def new_resume_event() -> pb.HistoryEvent: +def new_resume_event(*, encoded_input: str | None = None) -> pb.HistoryEvent: return pb.HistoryEvent( eventId=-1, timestamp=timestamp_pb2.Timestamp(), - executionResumed=pb.ExecutionResumedEvent() + executionResumed=pb.ExecutionResumedEvent( + input=get_string_value(encoded_input) + ) ) diff --git a/durabletask/testing/in_memory_backend.py b/durabletask/testing/in_memory_backend.py index 0458266f..44aea1f7 100644 --- a/durabletask/testing/in_memory_backend.py +++ b/durabletask/testing/in_memory_backend.py @@ -386,7 +386,9 @@ def SuspendInstance(self, request: pb.SuspendRequest, context: grpc.ServicerCont if instance.status == pb.ORCHESTRATION_STATUS_SUSPENDED: return pb.SuspendResponse() - event = helpers.new_suspend_event() + event = helpers.new_suspend_event( + encoded_input=request.reason.value if request.HasField("reason") else None + ) instance.pending_events.append(event) instance.last_updated_at = datetime.now(timezone.utc) self._enqueue_orchestration(instance.instance_id) @@ -403,7 +405,9 @@ def ResumeInstance(self, request: pb.ResumeRequest, context: grpc.ServicerContex f"Orchestration instance '{request.instanceId}' not found") return pb.ResumeResponse() - event = helpers.new_resume_event() + event = helpers.new_resume_event( + encoded_input=request.reason.value if request.HasField("reason") else None + ) instance.pending_events.append(event) instance.last_updated_at = datetime.now(timezone.utc) self._enqueue_orchestration(instance.instance_id) diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index 80e68f5d..962ad8eb 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -472,13 +472,13 @@ async def test_suspend_resume_delegate(): new=AsyncMock()) as suspend_mock: with pytest.warns(DeprecationWarning): await client.suspend("abc", "reason") - suspend_mock.assert_awaited_once_with("abc") + suspend_mock.assert_awaited_once_with("abc", reason="reason") with patch.object(client, "resume_orchestration", new=AsyncMock()) as resume_mock: with pytest.warns(DeprecationWarning): await client.resume("abc", "reason") - resume_mock.assert_awaited_once_with("abc") + resume_mock.assert_awaited_once_with("abc", reason="reason") finally: await client.close() diff --git a/tests/durabletask/test_client.py b/tests/durabletask/test_client.py index 8a9cb5e3..ff9feada 100644 --- a/tests/durabletask/test_client.py +++ b/tests/durabletask/test_client.py @@ -1258,6 +1258,43 @@ def test_worker_stores_resiliency_options(): assert worker._resiliency_options is resiliency +def test_suspend_resume_orchestration_includes_optional_reason(): + stub = MagicMock() + + with patch('durabletask.client.shared.get_grpc_channel', return_value=MagicMock()), patch( + 'durabletask.client.stubs.TaskHubSidecarServiceStub', return_value=stub): + client = TaskHubGrpcClient() + client.suspend_orchestration('suspended', reason='maintenance') + client.resume_orchestration('resumed') + + suspend_request = stub.SuspendInstance.call_args.args[0] + assert suspend_request.instanceId == 'suspended' + assert suspend_request.reason.value == 'maintenance' + resume_request = stub.ResumeInstance.call_args.args[0] + assert resume_request.instanceId == 'resumed' + assert not resume_request.HasField('reason') + + +@pytest.mark.asyncio +async def test_async_suspend_resume_orchestration_includes_optional_reason(): + stub = MagicMock() + stub.SuspendInstance = AsyncMock() + stub.ResumeInstance = AsyncMock() + + with patch('durabletask.client.shared.get_async_grpc_channel', return_value=MagicMock()), patch( + 'durabletask.client.stubs.TaskHubSidecarServiceStub', return_value=stub): + client = AsyncTaskHubGrpcClient() + await client.suspend_orchestration('suspended') + await client.resume_orchestration('resumed', reason='maintenance complete') + + suspend_request = stub.SuspendInstance.call_args.args[0] + assert suspend_request.instanceId == 'suspended' + assert not suspend_request.HasField('reason') + resume_request = stub.ResumeInstance.call_args.args[0] + assert resume_request.instanceId == 'resumed' + assert resume_request.reason.value == 'maintenance complete' + + def test_get_orchestration_history_aggregates_chunks_and_deexternalizes_payloads(): store = FakePayloadStore() token = store.upload(b'history payload') diff --git a/tests/durabletask/test_orchestration_e2e.py b/tests/durabletask/test_orchestration_e2e.py index 278f35d6..df11754d 100644 --- a/tests/durabletask/test_orchestration_e2e.py +++ b/tests/durabletask/test_orchestration_e2e.py @@ -275,7 +275,7 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert state.serialized_output == json.dumps("timed out") -def test_suspend_and_resume(): +def test_suspend_and_resume_preserves_reasons_in_history(): def orchestrator(ctx: task.OrchestrationContext, _): result = yield ctx.wait_for_external_event("my_event") return result @@ -290,7 +290,7 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert state is not None # Suspend the orchestration and wait for it to go into the SUSPENDED state - task_hub_client.suspend_orchestration(id) + task_hub_client.suspend_orchestration(id, reason="maintenance") deadline = time.time() + 10 while state.runtime_status == client.OrchestrationStatus.RUNNING: assert time.time() < deadline, "Timed out waiting for SUSPENDED status" @@ -308,11 +308,18 @@ def orchestrator(ctx: task.OrchestrationContext, _): pass # Resume the orchestration and wait for it to complete - task_hub_client.resume_orchestration(id) + task_hub_client.resume_orchestration(id, reason="maintenance complete") state = task_hub_client.wait_for_orchestration_completion(id, timeout=30) + events = task_hub_client.get_orchestration_history(id) assert state is not None assert state.runtime_status == client.OrchestrationStatus.COMPLETED assert state.serialized_output == json.dumps(42) + suspended_event = next( + event for event in events if isinstance(event, history.ExecutionSuspendedEvent)) + resumed_event = next( + event for event in events if isinstance(event, history.ExecutionResumedEvent)) + assert suspended_event.input == "maintenance" + assert resumed_event.input == "maintenance complete" def test_terminate():