Add Request/Response sample with update-based responses (fixes #6)

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.

pyproject.toml

@@ -120,6 +120,7 @@ packages = [
     "pydantic_converter_v1",
     "pyproject.toml",
     "replay",
+    "reqrespupdate",
     "schedules",
     "sentry",
     "sleep_for_days",

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.

reqrespupdate/__init__.py

@@ -0,0 +1,2 @@
+TASK_QUEUE = "reqrespupdate-task-queue"
+WORKFLOW_ID = "reqrespupdate-workflow-id"

reqrespupdate/activities.py

@@ -0,0 +1,6 @@
+from temporalio import activity
+
+
+@activity.defn
+async def uppercase(input: str) -> str:
+    return input.upper()

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())

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())

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())

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