diff --git a/README.md b/README.md index d39428dc..e8800843 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,7 @@ without wrapping them in a workflow. * [pydantic_converter](pydantic_converter) - Data converter for using Pydantic models. * [pydantic_converter_v1](pydantic_converter_v1) - Data converter for Pydantic v1 models (prefer pydantic_converter for v2). * [replay](replay) - Verify that workflow code changes are compatible with existing histories. +* [reqrespupdate](reqrespupdate) - Send a request to a long-running workflow and get a response back via an update, with continue-as-new to bound history. * [resource_pool](resource_pool) - Allocate a pool of shared resources across workflows. * [schedules](schedules) - Demonstrates a Workflow Execution that occurs according to a schedule. * [sentry](sentry) - Report errors to Sentry. diff --git a/pyproject.toml b/pyproject.toml index e8f3ebd8..076dd2ec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -120,6 +120,7 @@ packages = [ "pydantic_converter_v1", "pyproject.toml", "replay", + "reqrespupdate", "schedules", "sentry", "sleep_for_days", diff --git a/reqrespupdate/README.md b/reqrespupdate/README.md new file mode 100644 index 00000000..ecdbcf53 --- /dev/null +++ b/reqrespupdate/README.md @@ -0,0 +1,56 @@ +# Request/Response Sample with Update-Based Responses + +This sample shows how to send a request to a running workflow and get a response +back, using [Workflow Update](https://docs.temporal.io/encyclopedia/workflow-message-passing#updates). +The update handler runs an activity and returns its result directly to the caller, +so there is no response task queue, no callback activity and no request IDs to +correlate. + +It is a port of the Go [reqrespupdate](https://github.com/temporalio/samples-go/tree/main/reqrespupdate) +sample. + +To run, open three terminals. + +Run the worker in the first: + + uv run reqrespupdate/worker.py + +Start the long-running workflow in the second: + + uv run reqrespupdate/starter.py + +Then request an uppercasing every second in the third: + + uv run reqrespupdate/requester.py + +Several requesters can be run at once, in separate terminals, to confirm they are +independent of one another. + +### Comparison with the other request/response approaches + +The Go samples also show request/response via +[activities](https://github.com/temporalio/samples-go/tree/main/reqrespactivity) and via +[queries](https://github.com/temporalio/samples-go/tree/main/reqrespquery). Both predate +Workflow Update. Update is the recommended approach and is the only one ported here. + +### Continue-as-new and backpressure + +Workflow history cannot grow without limit, so a workflow that fields requests +indefinitely has to continue-as-new periodically. To avoid losing work, it must do so +only when no handler is still running, which means there has to be a moment where the +workflow is idle. + +`workflow.all_handlers_finished()` reports exactly that: whether any update or signal +handler is still executing, including one waiting on an activity retry. The workflow +waits on it before continuing as new, so no in-flight request is interrupted. + +If requests arrive faster than they are handled, that idle moment may never come and +history keeps growing. The workflow therefore rejects requests from the update +validator once it is draining toward a continue-as-new. A validator rejection is not +written to history, which is what makes it the right tool when the problem is history +size. The requester sees the rejection and retries, and because the retry targets the +same workflow ID it lands on the fresh run. + +The retry policy and timeout on the activity affect how long a handler can stay in +flight, and therefore how long the workflow can be kept from continuing as new. Set +them balancing resilience against the need for a period of idleness. diff --git a/reqrespupdate/__init__.py b/reqrespupdate/__init__.py new file mode 100644 index 00000000..c2ede455 --- /dev/null +++ b/reqrespupdate/__init__.py @@ -0,0 +1,2 @@ +TASK_QUEUE = "reqrespupdate-task-queue" +WORKFLOW_ID = "reqrespupdate-workflow-id" diff --git a/reqrespupdate/activities.py b/reqrespupdate/activities.py new file mode 100644 index 00000000..4759d46e --- /dev/null +++ b/reqrespupdate/activities.py @@ -0,0 +1,6 @@ +from temporalio import activity + + +@activity.defn +async def uppercase(input: str) -> str: + return input.upper() diff --git a/reqrespupdate/requester.py b/reqrespupdate/requester.py new file mode 100644 index 00000000..8db55baa --- /dev/null +++ b/reqrespupdate/requester.py @@ -0,0 +1,42 @@ +import asyncio + +from temporalio.client import Client, WorkflowUpdateFailedError +from temporalio.envconfig import ClientConfig +from temporalio.exceptions import ApplicationError + +from reqrespupdate import WORKFLOW_ID +from reqrespupdate.workflow import BACKOFF_ERROR_TYPE, Request, UppercaseWorkflow + + +async def main(): + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**config) + handle = client.get_workflow_handle(WORKFLOW_ID) + + # Request an uppercasing every second. Several of these can be run at once, + # in separate terminals, to confirm the requesters are independent of each + # other. + i = 0 + while True: + request = Request(input=f"foo{i}") + try: + response = await handle.execute_update(UppercaseWorkflow.uppercase, request) + print(f"Requested uppercase of {request.input}, got {response.output}") + i += 1 + except WorkflowUpdateFailedError as err: + # The run we sent to is draining toward a continue-as-new and asked + # us to back off. Retrying sends to the same workflow ID, which by + # then is the new run. Any other failure is a real one. + if ( + isinstance(err.cause, ApplicationError) + and err.cause.type == BACKOFF_ERROR_TYPE + ): + print("Rejected while the workflow continues as new, retrying") + else: + raise + await asyncio.sleep(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/reqrespupdate/starter.py b/reqrespupdate/starter.py new file mode 100644 index 00000000..d0c08609 --- /dev/null +++ b/reqrespupdate/starter.py @@ -0,0 +1,25 @@ +import asyncio + +from temporalio.client import Client +from temporalio.envconfig import ClientConfig + +from reqrespupdate import TASK_QUEUE, WORKFLOW_ID +from reqrespupdate.workflow import UppercaseWorkflow, UppercaseWorkflowInput + + +async def main(): + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**config) + + handle = await client.start_workflow( + UppercaseWorkflow.run, + UppercaseWorkflowInput(), + id=WORKFLOW_ID, + task_queue=TASK_QUEUE, + ) + print(f"Started workflow with ID {handle.id}, now run the requester") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/reqrespupdate/worker.py b/reqrespupdate/worker.py new file mode 100644 index 00000000..43cea313 --- /dev/null +++ b/reqrespupdate/worker.py @@ -0,0 +1,38 @@ +import asyncio +import logging + +from temporalio.client import Client +from temporalio.envconfig import ClientConfig +from temporalio.worker import Worker + +from reqrespupdate import TASK_QUEUE +from reqrespupdate.activities import uppercase +from reqrespupdate.workflow import UppercaseWorkflow + +interrupt_event = asyncio.Event() + + +async def main(): + config = ClientConfig.load_client_connect_config() + config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**config) + + async with Worker( + client, + task_queue=TASK_QUEUE, + workflows=[UppercaseWorkflow], + activities=[uppercase], + ): + logging.info("UppercaseWorkflow worker started, ctrl+c to exit") + await interrupt_event.wait() + logging.info("Shutting down") + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(main()) + except KeyboardInterrupt: + interrupt_event.set() + loop.run_until_complete(loop.shutdown_asyncgens()) diff --git a/reqrespupdate/workflow.py b/reqrespupdate/workflow.py new file mode 100644 index 00000000..405bec62 --- /dev/null +++ b/reqrespupdate/workflow.py @@ -0,0 +1,107 @@ +from dataclasses import dataclass +from datetime import timedelta + +from temporalio import workflow +from temporalio.common import RetryPolicy +from temporalio.exceptions import ApplicationError + +with workflow.unsafe.imports_passed_through(): + from reqrespupdate.activities import uppercase + + +# Error type used when the workflow rejects a request because it is continuing +# as new. The requester matches on this to tell "back off and retry" apart from +# a genuine failure. +BACKOFF_ERROR_TYPE = "ContinuingAsNew" + + +# Be in the habit of storing message inputs and outputs in serializable +# structures. This makes it easier to add more over time in a +# backward-compatible way. +@dataclass +class Request: + input: str + + +@dataclass +class Response: + output: str + + +@dataclass +class UppercaseWorkflowInput: + # Workflows cannot have infinitely-sized history, so a workflow that fields + # requests forever has to continue-as-new periodically. We bound the number + # of requests each run accepts. + requests_before_continue_as_new: int = 500 + # If requests arrive faster than they are handled, the workflow may never + # get an idle moment in which to continue-as-new, and history keeps growing. + # Rejecting requests from the update validator while a continue-as-new is + # pending gives the workflow that idle moment and tells the requester to + # back off. A validator rejection is not written to history, which is what + # makes it a useful tool when the problem is history size. + reject_update_on_pending_continue_as_new: bool = True + + +@workflow.defn +class UppercaseWorkflow: + """A long-running workflow that uppercases strings on request. + + The response is returned directly from the update handler, so there is no + response task queue, no callback activity and no request IDs to correlate. + """ + + @workflow.init + def __init__(self, input: UppercaseWorkflowInput) -> None: + # A run has to accept at least one request, otherwise it continues as + # new the moment it starts and the chain never does any work. + if input.requests_before_continue_as_new < 1: + raise ApplicationError( + "requests_before_continue_as_new must be at least 1", + non_retryable=True, + ) + self.input = input + self.request_count = 0 + + @workflow.run + async def run(self, input: UppercaseWorkflowInput) -> None: + # Wait until this run has taken its share of requests and no handler is + # still running. all_handlers_finished() accounts for in-flight update + # handlers, including ones waiting on an activity retry, so there is no + # need to track pending handlers by hand. + await workflow.wait_condition( + lambda: self.continue_as_new_pending() and workflow.all_handlers_finished() + ) + workflow.continue_as_new(input) + + @workflow.update + async def uppercase(self, request: Request) -> Response: + self.request_count += 1 + # WARNING: the timeout and retry policy affect how long this handler can + # stay in flight, and therefore how long the workflow can be prevented + # from continuing as new. Set them balancing resilience against the need + # for a period of idleness. + output = await workflow.execute_activity( + uppercase, + request.input, + schedule_to_close_timeout=timedelta(seconds=5), + retry_policy=RetryPolicy(maximum_attempts=2), + ) + return Response(output=output) + + @uppercase.validator + def validate_uppercase(self, request: Request) -> None: + # Raising from a validator rejects the update without writing it to + # history, and without the workflow having to handle it. Note that a + # validator must be synchronous. + if ( + self.input.reject_update_on_pending_continue_as_new + and self.continue_as_new_pending() + ): + raise ApplicationError( + "Workflow is continuing as new, please retry", + type=BACKOFF_ERROR_TYPE, + ) + + def continue_as_new_pending(self) -> bool: + return self.request_count >= self.input.requests_before_continue_as_new diff --git a/tests/reqrespupdate/__init__.py b/tests/reqrespupdate/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/reqrespupdate/workflow_test.py b/tests/reqrespupdate/workflow_test.py new file mode 100644 index 00000000..5084adb9 --- /dev/null +++ b/tests/reqrespupdate/workflow_test.py @@ -0,0 +1,103 @@ +import asyncio +import uuid + +import pytest +from temporalio.client import Client, WorkflowUpdateFailedError +from temporalio.exceptions import ApplicationError +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import Worker + +from reqrespupdate.activities import uppercase +from reqrespupdate.workflow import ( + BACKOFF_ERROR_TYPE, + Request, + UppercaseWorkflow, + UppercaseWorkflowInput, +) + + +async def request_uppercase(handle, text: str, max_attempts: int = 10) -> str: + """Request an uppercasing, retrying if the workflow is continuing as new. + + This is the same backoff the requester in this sample performs, bounded so + that a workflow which never continues as new fails the test instead of + retrying forever. + """ + for _ in range(max_attempts): + try: + response = await handle.execute_update( + UppercaseWorkflow.uppercase, Request(input=text) + ) + return response.output + except WorkflowUpdateFailedError as err: + if ( + isinstance(err.cause, ApplicationError) + and err.cause.type == BACKOFF_ERROR_TYPE + ): + await asyncio.sleep(0.1) + continue + raise + raise AssertionError( + f"Request for {text} still rejected after {max_attempts} attempts" + ) + + +async def test_uppercase(client: Client, env: WorkflowEnvironment): + if env.supports_time_skipping: + pytest.skip( + "Java test server: https://github.com/temporalio/sdk-java/issues/1903" + ) + task_queue = f"tq-{uuid.uuid4()}" + async with Worker( + client, + task_queue=task_queue, + workflows=[UppercaseWorkflow], + activities=[uppercase], + ): + handle = await client.start_workflow( + UppercaseWorkflow.run, + UppercaseWorkflowInput(), + id=f"reqrespupdate-{uuid.uuid4()}", + task_queue=task_queue, + ) + try: + for text in ["foo", "bar", "baz"]: + assert await request_uppercase(handle, text) == text.upper() + finally: + await handle.terminate() + + +async def test_continues_as_new_without_losing_requests( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping: + pytest.skip( + "Java test server: https://github.com/temporalio/sdk-java/issues/1903" + ) + task_queue = f"tq-{uuid.uuid4()}" + async with Worker( + client, + task_queue=task_queue, + workflows=[UppercaseWorkflow], + activities=[uppercase], + ): + handle = await client.start_workflow( + UppercaseWorkflow.run, + # Low enough that the run continues as new several times over the + # requests below. + UppercaseWorkflowInput(requests_before_continue_as_new=2), + id=f"reqrespupdate-{uuid.uuid4()}", + task_queue=task_queue, + ) + try: + # Whether any individual request is rejected depends on how the + # continue-as-new interleaves with it, so we assert the contract the + # requester actually relies on: every request eventually succeeds, + # and the workflow does continue as new underneath. + for i in range(6): + assert await request_uppercase(handle, f"foo{i}") == f"FOO{i}" + + description = await handle.describe() + assert description.run_id != handle.first_execution_run_id + finally: + await handle.terminate()