NEXUS-484: Support UpdateWorkflow as a Nexus Operation#1631
Conversation
a913bdd to
79d24c1
Compare
| result = await wf_handle.result() | ||
| assert result == "Updated workflow status from Pending to Created" | ||
|
|
||
| # await assert_event_subsequence( |
There was a problem hiding this comment.
TODO:
- check why this is flaky on ubuntu-arm/windows
- verify links are present
- check resp is sync on the retried update below
There was a problem hiding this comment.
Probably obvious, but just to note we should try to match the test coverage we have in Go
ecc3c9e to
e38becf
Compare
| result_type: type | None = None, | ||
| rpc_metadata: Mapping[str, str | bytes] = {}, | ||
| rpc_timeout: timedelta | None = None, | ||
| # run_id: str | None = None, |
There was a problem hiding this comment.
Why are these commented out
There was a problem hiding this comment.
wasnt sure if they were required at the time, fixed it
|
|
||
|
|
||
| @dataclass(frozen=True) | ||
| class CancelUpdateWorkflowOptions: |
d8db6d8 to
351568c
Compare
| if run_id == None: | ||
| run_id = self._run_id | ||
| if first_execution_run_id == None: | ||
| first_execution_run_id = self.first_execution_run_id | ||
|
|
There was a problem hiding this comment.
If we decide to keep this, let's make sure to use is None
| if run_id == None: | |
| run_id = self._run_id | |
| if first_execution_run_id == None: | |
| first_execution_run_id = self.first_execution_run_id | |
| if run_id is None: | |
| run_id = self._run_id | |
| if first_execution_run_id is None: | |
| first_execution_run_id = self.first_execution_run_id | |
| arg: Any = temporalio.common._arg_unset, | ||
| *, | ||
| args: Sequence[Any] = [], | ||
| id: str | None = None, |
There was a problem hiding this comment.
IMO we should consider naming this update_id to more clearly differentiate between workflow_id and the other IDs necessary.
There was a problem hiding this comment.
The current code seems to be using "id" as the defacto operation-relevant unique ID - eg. start_workflow uses "id" for workflowID - this is the case for the primitive start_update/etc as well -
sdk-python/temporalio/client/_workflow.py
Line 923 in c6386e9
Was trying to follow that but explicit does make sense. It might look inconsistent if update calls out specific and start_workflow doesnt- because that part is already solidified/shipped iiuc
I agree with changing it update_id - because update has both workflow_id and update_id - but just not sure because of this
| # draft-review: check why run_id and first_execution_run_id are not used | ||
| # for update workflow in python sdk |
There was a problem hiding this comment.
These are sourced from the WorkflowHandle rather than passed as arguments.
| first_execution_run_id=first_execution_run_id, | ||
| ) | ||
| # If the update has already completed, return the result synchronously | ||
| # This is in-line with the Go implementation as well |
There was a problem hiding this comment.
| # This is in-line with the Go implementation as well |
| run_id: str | None = None, | ||
| first_execution_run_id: str | None = None, |
There was a problem hiding this comment.
As mentioned in other comments, I think we should remove these and always use self._run_id and self._first_execution_run_id
| nexus_handle: UpdateHandle[Any] = ( | ||
| UpdateHandle._unsafe_from_client_workflow_update_handle(update_handle) | ||
| ) |
There was a problem hiding this comment.
I know the workflow variant uses a WorkflowHandle here, but I'm not sure it's necessary. I think we should remove the UpdateHandle and just contruct the appropriate OperationToken here.
There was a problem hiding this comment.
Makes sense- will just re-generate from the handle there
| # draft-review: maybe just move it inline, no need for a function just to error out | ||
| # check after review in case theres some other way to override/supply custom cancels |
There was a problem hiding this comment.
The way to customize cancellation is to inherit from TemporalOperationHandler and override these methods. So we'll either need to rework how that's done or leave these methods in place.
82de56f to
407a188
Compare
83ed716 to
a750f5f
Compare
a750f5f to
1d14414
Compare
| first_execution_run_id=first_execution_run_id, | ||
| ) | ||
| # If the update has already completed, return the result synchronously | ||
| if update_handle._known_outcome is not None: |
There was a problem hiding this comment.
If this case happens should we clear the flag set in _reserve_async_start() that prevents 2 async operations from being triggered by the client? Thinking through it, it probably makes sense to just always leave the flag set otherwise it could feel a little non-deterministic, but thought I'd pose the question for others to also consider.
VegetarianOrc
left a comment
There was a problem hiding this comment.
LGTM! One question and one double checking for if the new UpdateHandle is used.
| @dataclass(frozen=True) | ||
| class UpdateHandle(Generic[OutputT]): | ||
| """A handle to a workflow update that is backing a Nexus operation. |
There was a problem hiding this comment.
I think this should be removed now? Is it used somewhere?
What was changed
Adds support for UpdateWorkflow as a Nexus operation
Why?
Part of effort to expose all Temporal primitives via Nexus operations
Checklist
Closes
NEXUS-484
How was this tested: