Support Workflow Update as a Nexus Operation#2194
Conversation
fc51ac9 to
c219b68
Compare
52d2045 to
121ff88
Compare
| // requires server-side support for delivering Workflow Update completion callbacks to Nexus, which | ||
| // the bundled ephemeral server (Temporal 1.31.2) does not yet provide (workflow-run completion | ||
| // callbacks do work there). Enable once the test server delivers Update completion callbacks. | ||
| test.skip('UpdateWorkflow Nexus operation - pending update completes asynchronously via callback', async (t) => { |
There was a problem hiding this comment.
Same comment as dotnet, this should be enabled
| case OperationTokenType.UPDATE_WORKFLOW: { | ||
| let updateToken; | ||
| try { | ||
| updateToken = loadUpdateWorkflowOperationToken(token); |
There was a problem hiding this comment.
Why are we reparsing the token, we already have opToken?
There was a problem hiding this comment.
Good catch, now using opToken
| /** | ||
| * The Run ID extracted from the operation token. May be empty if the run was not pinned. | ||
| */ | ||
| readonly runId: string; |
There was a problem hiding this comment.
The run ID should be know here since the server returns the run ID on the update response
There was a problem hiding this comment.
Now using handle.workflowRunId
| export interface NexusUpdateWorkflowOptions<Ret, Args extends any[]> { | ||
| /** | ||
| * ID of the Workflow to send the Update to. | ||
| */ | ||
| readonly workflowId: string; | ||
|
|
||
| /** | ||
| * Run ID of the Workflow to send the Update to. If unset, the Update is sent to the most recent | ||
| * run of the Workflow. | ||
| */ | ||
| readonly runId?: string; | ||
|
|
||
| /** | ||
| * The Update definition (as returned by `defineUpdate`) or the Update name. | ||
| */ | ||
| readonly update: UpdateDefinition<Ret, Args> | string; |
There was a problem hiding this comment.
I think we should consider threading the Name type through to the UpdateDefinition like the WorkflowHandle
executeUpdate<Ret, Args extends [], Name extends string = string>(
def: UpdateDefinition<Ret, Args, Name> | string,
options?: WorkflowUpdateOptions & { args?: Args }
): Promise<Ret>;| /** | ||
| * The Update lifecycle stage to wait for. Only {@link WorkflowUpdateStage.ACCEPTED} is supported | ||
| * for Nexus operations; any other value is rejected. Defaults to | ||
| * {@link WorkflowUpdateStage.ACCEPTED}. | ||
| */ | ||
| readonly waitForStage?: WorkflowUpdateStage; |
There was a problem hiding this comment.
Should we expose this at all if there's only one valid value and we default to it?
There was a problem hiding this comment.
Good point, removed it
| const url = new URL( | ||
| `temporal:///namespaces/${encodeURIComponent(wl.namespace)}/workflows/${encodeURIComponent( | ||
| wl.workflowId | ||
| )}/${encodeURIComponent(wl.runId ?? '')}/history` |
There was a problem hiding this comment.
If the wl.runId property is coalesced to '' does the URL still resolve to a valid page on the UI?
There was a problem hiding this comment.
Point. RunId is now required
| if (sleepMs > 0) { | ||
| await workflow.sleep(sleepMs); | ||
| } |
There was a problem hiding this comment.
Is this sleep necessary? I'm wondering if there's a potential flake being hidden by this.
There was a problem hiding this comment.
workflow.sleep suspends the update handler after it accepts, that is what forces the async completion callback path. So when it sleeps we get the async test, when it doesn't sleep it's a sync test.
|
|
||
| /** | ||
| * Sends an Update to a Workflow as the backing operation for the current Nexus Operation. | ||
| * | ||
| * The Update request is augmented with the Nexus Operation's request ID (for deduplication), its | ||
| * request links, and a completion callback carrying the operation token, so the Update's | ||
| * completion is delivered back to the Nexus caller. | ||
| * | ||
| * Only asynchronous, `ACCEPTED`-stage Updates are supported: this method returns once the Update | ||
| * is accepted, yielding an asynchronous {@link TemporalOperationResult} whose completion is | ||
| * delivered via the callback. If the Update has already completed by the time it is accepted (for | ||
| * example a retried request with the same Update ID, or an Update that completes immediately), a | ||
| * synchronous result is returned instead; if that completed Update failed (e.g. validation | ||
| * rejection, which is non-retryable), it surfaces as a failed Nexus Operation. | ||
| * | ||
| * @experimental Nexus support in Temporal SDK is experimental. | ||
| */ | ||
| updateWorkflow<Ret, Args extends any[] = []>( | ||
| options: NexusUpdateWorkflowOptions<Ret, Args> | ||
| ): Promise<TemporalOperationResult<Ret>>; |
There was a problem hiding this comment.
I think we should put this function onto the Nexus specific workflow handle to match the signal method.
| 'failed', | ||
| 'Nexus operation Workflow Updates only support the ACCEPTED wait stage for async updates' | ||
| ); | ||
| } |
There was a problem hiding this comment.
Similar to the above comment, I think we should consider removing this check and just always using ACCEPTED.
Co-authored-by: Alex Mazzeo <alex.mazzeo@temporal.io>
Co-authored-by: Alex Mazzeo <alex.mazzeo@temporal.io>
Co-authored-by: Alex Mazzeo <alex.mazzeo@temporal.io>
Co-authored-by: Alex Mazzeo <alex.mazzeo@temporal.io>
3 participants
What was changed
Why?
Checklist
Closes
How was this tested: