From 391455c2fdfac622b13358e27b0c1f85b0fc7c39 Mon Sep 17 00:00:00 2001 From: Alice Lin Date: Tue, 8 Sep 2026 16:59:50 -0700 Subject: [PATCH] Add Nexus Signal-with-Start example --- nexus_messaging/callerpattern/README.md | 18 ++++---- .../callerpattern/handler/service_handler.py | 11 +++-- nexus_messaging/ondemandpattern/README.md | 36 ++++++++++------ .../ondemandpattern/caller/workflows.py | 41 +++++++++++++++++++ .../handler/service_handler.py | 37 ++++++++++++++--- .../ondemandpattern/handler/workflows.py | 22 +++++++++- nexus_messaging/ondemandpattern/service.py | 9 ++++ tests/conftest.py | 6 +++ tests/nexus_messaging/ondemandpattern_test.py | 18 +++++++- 9 files changed, 164 insertions(+), 34 deletions(-) diff --git a/nexus_messaging/callerpattern/README.md b/nexus_messaging/callerpattern/README.md index 8be001cbc..925f07faa 100644 --- a/nexus_messaging/callerpattern/README.md +++ b/nexus_messaging/callerpattern/README.md @@ -17,19 +17,23 @@ The caller Workflow: ### Running -Start a Temporal server: +This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download the compatible +binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support). + +Start the Temporal dev server with the required namespaces pre-created and Workflow Update callbacks enabled: ```bash -temporal server start-dev +./temporal server start-dev \ + --dynamic-config-value history.enableUpdateCallbacks=true \ + --dynamic-config-value history.enableCHASMSignalBacklinks=true \ + --namespace nexus-messaging-handler-namespace \ + --namespace nexus-messaging-caller-namespace ``` -Create the namespaces and Nexus endpoint: +Create the Nexus endpoint: ```bash -temporal operator namespace create --namespace nexus-messaging-handler-namespace -temporal operator namespace create --namespace nexus-messaging-caller-namespace - -temporal operator nexus endpoint create \ +./temporal operator nexus endpoint create \ --name nexus-messaging-nexus-endpoint \ --target-namespace nexus-messaging-handler-namespace \ --target-task-queue nexus-messaging-handler-task-queue diff --git a/nexus_messaging/callerpattern/handler/service_handler.py b/nexus_messaging/callerpattern/handler/service_handler.py index c9f384a74..d9cce860a 100644 --- a/nexus_messaging/callerpattern/handler/service_handler.py +++ b/nexus_messaging/callerpattern/handler/service_handler.py @@ -1,7 +1,8 @@ """ Nexus operation handler implementation for the entity pattern. Each operation receives a -user_id, which is mapped to a workflow ID. The operations are synchronous because queries -and updates against a running workflow complete quickly. +user_id, which is mapped to a Workflow ID. The Query and Signal operations are synchronous +because they complete quickly against a running Workflow; set_language is async because it +is backed by a Workflow Update that may call an activity. """ from __future__ import annotations @@ -77,13 +78,11 @@ async def set_language( client: nexus.TemporalNexusClient, input: SetLanguageInput, ) -> nexus.TemporalOperationResult[Language]: - result = await self._get_workflow_handle( - client.client, input.user_id - ).execute_update( + return await client.start_workflow_update( + get_workflow_id(input.user_id), GreetingWorkflow.set_language_using_activity, input, ) - return nexus.TemporalOperationResult.sync(result) @nexus.temporal_operation async def approve( diff --git a/nexus_messaging/ondemandpattern/README.md b/nexus_messaging/ondemandpattern/README.md index b0c6145a0..07394b2ea 100644 --- a/nexus_messaging/ondemandpattern/README.md +++ b/nexus_messaging/ondemandpattern/README.md @@ -6,28 +6,36 @@ operations. `NexusRemoteGreetingService` adds a `run_from_remote` operation that instance to target. The caller Workflow: -1. Starts two remote `GreetingWorkflow` instances via `run_from_remote` (backed by `temporal_operation`) -2. Queries each for supported languages -3. Changes the language on each (Arabic and Hindi) -4. Confirms the changes via queries -5. Approves both Workflows -6. Waits for each to complete and returns their results +1. Attaches approval context for the first user via `attach_approval_context`, before anything has + started that user's Workflow +2. Starts two remote `GreetingWorkflow` instances via `run_from_remote` (backed by `temporal_operation`) +3. Attaches approval context for the second user, whose Workflow now already exists +4. Queries each for supported languages +5. Changes the language on each (Arabic and Hindi) +6. Confirms the changes via queries +7. Approves both Workflows +8. Waits for each to complete and returns their results ### Running -Start a Temporal server: +This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download the compatible +binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support). + +Start the Temporal dev server with the required namespaces pre-created and Workflow Update callbacks enabled: ```bash -temporal server start-dev +./temporal server start-dev \ + --dynamic-config-value history.enableUpdateCallbacks=true \ + --dynamic-config-value history.enableCHASMSignalBacklinks=true \ + --dynamic-config-value history.enableSignalWithStartFromWorkflow=true \ + --namespace nexus-messaging-handler-namespace \ + --namespace nexus-messaging-caller-namespace ``` -Create the namespaces and Nexus endpoint: +Create the Nexus endpoint: ```bash -temporal operator namespace create --namespace nexus-messaging-handler-namespace -temporal operator namespace create --namespace nexus-messaging-caller-namespace - -temporal operator nexus endpoint create \ +./temporal operator nexus endpoint create \ --name nexus-messaging-nexus-endpoint \ --target-namespace nexus-messaging-handler-namespace \ --target-task-queue nexus-messaging-handler-task-queue @@ -48,8 +56,10 @@ uv run python -m nexus_messaging.ondemandpattern.caller.app Expected output: ``` +Attached approval context before the workflow existed: UserId One started remote greeting workflow: UserId One started remote greeting workflow: UserId Two +Attached approval context to the running workflow: UserId Two Supported languages for UserId One: [, ] Supported languages for UserId Two: [, ] UserId One changed language: ENGLISH -> ARABIC diff --git a/nexus_messaging/ondemandpattern/caller/workflows.py b/nexus_messaging/ondemandpattern/caller/workflows.py index 4e1e86bb3..01bde9196 100644 --- a/nexus_messaging/ondemandpattern/caller/workflows.py +++ b/nexus_messaging/ondemandpattern/caller/workflows.py @@ -8,6 +8,7 @@ from nexus_messaging.ondemandpattern.service import ( ApproveInput, + AttachApprovalContextInput, GetLanguageInput, GetLanguagesInput, Language, @@ -38,10 +39,33 @@ async def run(self) -> list[str]: # users we want to process. The first calls start two workflows, one for each # user. Subsequent calls perform different actions between the two users. + # Attach information before the Workflow exists. Because attach_approval_context + # is backed by Signal-with-Start on the handler, this call creates the Workflow + # and delivers the note to it. + await self.nexus_client.execute_operation( + NexusRemoteGreetingService.attach_approval_context, + AttachApprovalContextInput( + note="queued for localization review by the nightly batch", + user_id=REMOTE_WORKFLOW_ONE, + ), + ) + log.append( + f"Attached approval context before the workflow existed: " + f"{REMOTE_WORKFLOW_ONE}" + ) + workflow.logger.info( + "attached approval context for %s, creating the workflow", + REMOTE_WORKFLOW_ONE, + ) + # This is an async Nexus operation -- starts a workflow on the handler and # returns a handle. Unlike the sync operations below, this does not block # until the workflow completes. It is backed by temporal_operation on the # handler side. + + # The Workflow for this user is already running due to the call above. The + # handler sets the conflict policy to USE_EXISTING, so this call attaches the + # operation's completion callback to the running execution. handle_one = await self.nexus_client.start_operation( NexusRemoteGreetingService.run_from_remote, RunFromRemoteInput(user_id=REMOTE_WORKFLOW_ONE), @@ -56,6 +80,23 @@ async def run(self) -> list[str]: log.append(f"started remote greeting workflow: {REMOTE_WORKFLOW_TWO}") workflow.logger.info("started remote greeting workflow %s", REMOTE_WORKFLOW_TWO) + # This user's Workflow was created by run_from_remote just above, so here + # Signal-with-Start skips the start and only delivers the Signal. + await self.nexus_client.execute_operation( + NexusRemoteGreetingService.attach_approval_context, + AttachApprovalContextInput( + note="translation approved by the localization team", + user_id=REMOTE_WORKFLOW_TWO, + ), + ) + log.append( + f"Attached approval context to the running workflow: {REMOTE_WORKFLOW_TWO}" + ) + workflow.logger.info( + "attached approval context for %s, messaging the existing workflow", + REMOTE_WORKFLOW_TWO, + ) + # Query the remote workflows for supported languages. languages_output = await self.nexus_client.execute_operation( NexusRemoteGreetingService.get_languages, diff --git a/nexus_messaging/ondemandpattern/handler/service_handler.py b/nexus_messaging/ondemandpattern/handler/service_handler.py index 2865b94cf..6a82adbf4 100644 --- a/nexus_messaging/ondemandpattern/handler/service_handler.py +++ b/nexus_messaging/ondemandpattern/handler/service_handler.py @@ -10,11 +10,16 @@ import nexusrpc from temporalio import nexus from temporalio.client import Client, WorkflowHandle +from temporalio.common import WorkflowIDConflictPolicy -from nexus_messaging.ondemandpattern.handler.workflows import GreetingWorkflow +from nexus_messaging.ondemandpattern.handler.workflows import ( + ATTACH_APPROVAL_CONTEXT_SIGNAL, + GreetingWorkflow, +) from nexus_messaging.ondemandpattern.service import ( ApproveInput, ApproveOutput, + AttachApprovalContextInput, GetLanguageInput, GetLanguagesInput, GetLanguagesOutput, @@ -39,7 +44,8 @@ def _get_workflow_handle( GreetingWorkflow.run, self._get_workflow_id(user_id) ) - # Starts a new GreetingWorkflow with the caller-specified user ID. + # Starts a new GreetingWorkflow with the caller-specified user ID, + # or attached to one already running. # This is an async Nexus operation backed by temporal_operation. @nexus.temporal_operation async def run_from_remote( @@ -51,6 +57,10 @@ async def run_from_remote( return await client.start_workflow( GreetingWorkflow.run, id=self._get_workflow_id(input.user_id), + # Since attach_approval_context can create the GreetingWorkflow first, + # this operation needs to attach to the running execution rather than + # fail (default behavior). + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, ) @nexus.temporal_operation @@ -86,13 +96,11 @@ async def set_language( client: nexus.TemporalNexusClient, input: SetLanguageInput, ) -> nexus.TemporalOperationResult[Language]: - result = await self._get_workflow_handle( - client.client, input.user_id - ).execute_update( + return await client.start_workflow_update( + self._get_workflow_id(input.user_id), GreetingWorkflow.set_language_using_activity, input, ) - return nexus.TemporalOperationResult.sync(result) @nexus.temporal_operation async def approve( @@ -105,3 +113,20 @@ async def approve( GreetingWorkflow.approve, input ) return nexus.TemporalOperationResult.sync(ApproveOutput()) + + # Signals a Workflow, starting the Workflow first if it is not already running. + @nexus.temporal_operation + async def attach_approval_context( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: AttachApprovalContextInput, + ) -> nexus.TemporalOperationResult[None]: + await client.client.start_workflow( + GreetingWorkflow.run, + id=self._get_workflow_id(input.user_id), + task_queue=nexus.info().task_queue, + start_signal=ATTACH_APPROVAL_CONTEXT_SIGNAL, + start_signal_args=[input], + ) + return nexus.TemporalOperationResult.sync(None) diff --git a/nexus_messaging/ondemandpattern/handler/workflows.py b/nexus_messaging/ondemandpattern/handler/workflows.py index 99fc57289..0b2d614a5 100644 --- a/nexus_messaging/ondemandpattern/handler/workflows.py +++ b/nexus_messaging/ondemandpattern/handler/workflows.py @@ -7,6 +7,7 @@ import asyncio from datetime import timedelta +from typing import Optional from temporalio import workflow from temporalio.exceptions import ApplicationError @@ -14,12 +15,15 @@ from nexus_messaging.ondemandpattern.handler.activities import call_greeting_service from nexus_messaging.ondemandpattern.service import ( ApproveInput, + AttachApprovalContextInput, GetLanguagesInput, GetLanguagesOutput, Language, SetLanguageInput, ) +ATTACH_APPROVAL_CONTEXT_SIGNAL = "attach_approval_context" + @workflow.defn class GreetingWorkflow: @@ -30,6 +34,7 @@ def __init__(self) -> None: Language.ENGLISH: "Hello, world", } self.language = Language.ENGLISH + self.approval_context: Optional[str] = None self.lock = asyncio.Lock() @workflow.run @@ -54,9 +59,24 @@ def get_language(self) -> Language: @workflow.signal def approve(self, input: ApproveInput) -> None: - workflow.logger.info("Approval signal received for user %s", input.user_id) + workflow.logger.info( + "Approval signal received for user %s (context: %s)", + input.user_id, + self.approval_context, + ) self.approved_for_release = True + # Attaches supporting information for the eventual approval. Delivered with + # Signal-with-Start, so this may be the message that creates the workflow. + @workflow.signal(name=ATTACH_APPROVAL_CONTEXT_SIGNAL) + def attach_approval_context(self, input: AttachApprovalContextInput) -> None: + workflow.logger.info( + "attach_approval_context signal received for user %s: %s", + input.user_id, + input.note, + ) + self.approval_context = input.note + @workflow.update def set_language(self, input: SetLanguageInput) -> Language: workflow.logger.info("setLanguage update received for user %s", input.user_id) diff --git a/nexus_messaging/ondemandpattern/service.py b/nexus_messaging/ondemandpattern/service.py index 8f347d32b..0056785ac 100644 --- a/nexus_messaging/ondemandpattern/service.py +++ b/nexus_messaging/ondemandpattern/service.py @@ -58,6 +58,12 @@ class ApproveOutput: pass +@dataclass +class AttachApprovalContextInput: + note: str + user_id: str + + @nexusrpc.service class NexusRemoteGreetingService: # Starts a new GreetingWorkflow with the given workflow ID (asynchronous). @@ -70,3 +76,6 @@ class NexusRemoteGreetingService: set_language: nexusrpc.Operation[SetLanguageInput, Language] # Approves the specified workflow, allowing it to complete. approve: nexusrpc.Operation[ApproveInput, ApproveOutput] + # Attaches supporting information for the Workflow approval, either by + # messaging a running Workflow or creating a Workflow. + attach_approval_context: nexusrpc.Operation[AttachApprovalContextInput, None] diff --git a/tests/conftest.py b/tests/conftest.py index 7f7fe066f..da0f9d200 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -45,6 +45,12 @@ async def env(request) -> AsyncGenerator[WorkflowEnvironment, None]: dev_server_extra_args=[ "--dynamic-config-value", "activity.enableCallbacks=true", + "--dynamic-config-value", + "history.enableUpdateCallbacks=true", + "--dynamic-config-value", + "history.enableCHASMSignalBacklinks=true", + "--dynamic-config-value", + "history.enableSignalWithStartFromWorkflow=true", ], dev_server_download_version="v1.7.4-standalone-nexus-operations", ) diff --git a/tests/nexus_messaging/ondemandpattern_test.py b/tests/nexus_messaging/ondemandpattern_test.py index 75fd761c0..9bba9f624 100644 --- a/tests/nexus_messaging/ondemandpattern_test.py +++ b/tests/nexus_messaging/ondemandpattern_test.py @@ -12,6 +12,7 @@ from nexus_messaging.ondemandpattern.caller.workflows import CallerRemoteWorkflow from nexus_messaging.ondemandpattern.service import ( ApproveInput, + AttachApprovalContextInput, GetLanguageInput, GetLanguagesInput, Language, @@ -37,12 +38,27 @@ async def run(self) -> None: workflow_id = f"test-remote-{workflow.uuid4()}" - # Start a remote workflow. + # Signal-with-Start: no Workflow exists for this user yet, so this operation + # creates it and delivers the Signal. + await nexus_client.execute_operation( + NexusRemoteGreetingService.attach_approval_context, + AttachApprovalContextInput(note="created by signal", user_id=workflow_id), + ) + + # Start a remote Workflow. The Signal-with-Start above already created it, so + # this attaches to the running execution (USE_EXISTING conflict policy). handle = await nexus_client.start_operation( NexusRemoteGreetingService.run_from_remote, RunFromRemoteInput(user_id=workflow_id), ) + # Signal-with-Start again, this time against the already-running Workflow, so + # only the Signal is delivered. + await nexus_client.execute_operation( + NexusRemoteGreetingService.attach_approval_context, + AttachApprovalContextInput(note="signal only", user_id=workflow_id), + ) + # Query for supported languages. languages_output = await nexus_client.execute_operation( NexusRemoteGreetingService.get_languages,