Nexus worker and workflow-backed operations #813
Conversation
dandavison
force-pushed
the
nexus
branch
2 times, most recently
from
a845a6f to
7a121bb
Compare
dandavison
force-pushed
the
nexus
branch
10 times, most recently
from
862e4f9 to
8ac0192
Compare
dandavison
force-pushed
the
nexus
branch
4 times, most recently
from
8940d51 to
520aecf
Compare
dandavison
force-pushed
the
nexus
branch
2 times, most recently
from
2fc971f to
8bd6011
Compare
dandavison
force-pushed
the
nexus
branch
2 times, most recently
from
5d54f48 to
8ddfb94
Compare
dandavison
force-pushed
the
nexus
branch
7 times, most recently
from
fa3e8ec to
f7bf47b
Compare
dandavison
force-pushed
the
nexus
branch
2 times, most recently
from
2762c9c to
3607c37
Compare
|
|
||
|
|
||
| @dataclass(frozen=True) | ||
| class WorkflowEventLink: |
| @@ -195,6 +195,37 @@ def __setstate__(self, state: object) -> None: | |||
| ) | |||
|
|
|||
|
|
|||
| @dataclass(frozen=True) | |||
| class NexusCompletionCallback: | |||
There was a problem hiding this comment.
I wonder if this is best in the client if it's only for clients
| @@ -908,6 +908,12 @@ def _error_to_failure( | |||
| failure.child_workflow_execution_failure_info.retry_state = ( | |||
| temporalio.api.enums.v1.RetryState.ValueType(error.retry_state or 0) | |||
| ) | |||
| # TODO(nexus-prerelease): test coverage for this | |||
| elif isinstance(error, temporalio.exceptions.NexusOperationError): | |||
There was a problem hiding this comment.
For symmetry reasons, I suspect we also need to convert NexusHandlerError (in theory a failure conversion from a failure should be able to convert back to a failure)
| self._type = type | ||
| self._retryable = retryable |
There was a problem hiding this comment.
Users of this exception should be allowed to see these properties/attributes IMO, even if we don't think they ever will
| temporalio.nexus.logger.exception( | ||
| "Failed to execute Nexus operation cancel method", err | ||
| ) |
There was a problem hiding this comment.
We need to complete the nexus task with a failure if the call failed
| # TODO(nexus-prerelease): expose this in a principled way | ||
| env._http_port = http_port # type: ignore |
There was a problem hiding this comment.
Mentioned in other comment, but not sure Python needs to do any Nexus HTTP testing
There was a problem hiding this comment.
HTTP is the primary way I'm testing Nexus. See tests/nexus/test_handler.py. Since the Nexus SDK currently lacks a Nexus server the tests have to be in the Temporal repo, but we can move them later.
| await _test_start_operation(test_case, True, env) | ||
|
|
||
|
|
||
| async def _test_start_operation( |
There was a problem hiding this comment.
Do we need to test that Nexus operation can start outside of a workflow? I don't think that is supported in Python SDK, can we wait until we do have out-of-workflow Nexus operation start to add tests confirming it? In the meantime, can you just use a workflow anytime you need to start a Nexus operation in your tests (the only officially supported way to start a Nexus operation in Python)?
| @@ -0,0 +1,37 @@ | |||
| import temporalio.api | |||
| import temporalio.api.common | |||
There was a problem hiding this comment.
Some of these imports not needed I don't think
| ) | ||
|
|
||
|
|
||
| @pytest.mark.parametrize( |
There was a problem hiding this comment.
IMO these large parameterized test cases are hard to read and reason about (have to jump around quite a bit to see the whole picture). If it's not too much extra code, can we have separate tests that maybe use limited helpers where needed? We've had flakes and maintenance needs on tests like test_unfinished_handler_on_workflow_termination and such and they are a bit rough on developers coming to maintain.
Draft PR available for initial review.
This is the worker side only; I've separated the workflow caller into a separate PR.