Repository navigation
Propagate Nexus serialization context to workflows and standalone activities - #3171
Conversation
Newer servers convert a Nexus operation error sent in the old failure format to the new format, wrapping the handler's failure in an OperationError. Unwrap it whenever it is present, so the old-format assertions hold against both the pinned and older servers.
1fcabb1 to
512ff57
Compare
Drop tests whose code path another test already covers: the async result and failure decodes that share their converter with the sync paths, the start-before-operation edge case, and the end-to-end workflow failure case covered by the replay and client unit tests.
VegetarianOrc
left a comment
There was a problem hiding this comment.
Overall looking correct to me!
One thing I think we're missing here is threading the correct context to manual activity completion. Right now looks like manual activity completion will encode the result w/ the activity context and then be forwarded to the Nexus Operation result.
| /** | ||
| * The data converter for the result or failure in a poll response. An activity started from a | ||
| * Nexus operation handler uses the handler's Nexus context, which the server returns with the | ||
| * response; other activities use an activity context. | ||
| */ | ||
| private DataConverter resultDataConverter(PollActivityExecutionResponse response) { | ||
| SerializationContext context = | ||
| response.hasPropagatedNexusSerializationContext() | ||
| ? PropagatedNexusSerializationContext.fromProto( | ||
| response.getPropagatedNexusSerializationContext()) | ||
| // Currently, result doesn't have access to activity type and serialization context | ||
| // doesn't hold activity ID. | ||
| : new ActivitySerializationContext(namespace, null, null, null, null, false); | ||
| return dataConverter.withContext(context); | ||
| } |
There was a problem hiding this comment.
Codex is pointing out an edge case in how we're handling heartbeat details in the event of a timeout that I think is real and worth considering.
- Activity heartbeats and details encoded w/ activity context.
- Activity stops heartbeating
- Server times out the activity and embeds the activity-encoded last heartbeat details into the failure proto
- Client selects the nexus context to decode the failure. A custom codec may fail to decode the failure proto.
There was a problem hiding this comment.
Ahh yes that needs a fix, working on it now thanks.
There was a problem hiding this comment.
Updated this. For activity heartbeats, they now use NexusSerializationContext to encode details when started by a nexus caller.
Also included a fix for async completion of an SAA with NexusSerializationContext.
An activity completed through local manual completion reports its result, failure and heartbeats through a client the worker creates. For a standalone activity started from a Nexus operation handler, give that client the handler's Nexus context, matching the worker's own completion and heartbeat paths. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
VegetarianOrc
left a comment
There was a problem hiding this comment.
This looks good to me!
| * <p>When a handler starts a workflow or a standalone activity, this context is propagated to it | ||
| * through the server. Both use it only for their result and failures, which are returned to the | ||
| * caller, and a standalone activity also uses it for its heartbeat details, which the server puts | ||
| * into its timeout failure. A workflow's input, memo, static summary and details, signals, queries | ||
| * and updates keep the {@link WorkflowSerializationContext}; a standalone activity's input and | ||
| * static summary and details keep the {@link ActivitySerializationContext}. Clients reading their | ||
| * results, including handles obtained by ID, use the context the server returns. This requires a | ||
| * server that propagates the context; older servers ignore it. | ||
| * | ||
| * <p>A standalone activity completed asynchronously through {@link | ||
| * io.temporal.client.ActivityCompletionClient} gets this context only if it is passed in: read it | ||
| * from {@link io.temporal.activity.ActivityInfo#getNexusSerializationContext()}, store it with the | ||
| * task token, and pass it to {@link | ||
| * io.temporal.client.ActivityCompletionClient#withContext(NexusSerializationContext)}. |
There was a problem hiding this comment.
This is pretty detailed and seems accurate but I think overwhelming for users. Maybe let's take a shot at providing a more high level description about how the nexus context is used appropriately, and leave out the details of what remains the same in each scenario.
There was a problem hiding this comment.
Agreed, updated
What was changed
Nexus operation handlers now propagate their
NexusSerializationContextto the workflows andstandalone activities (SAA) they start.
(
propagated_nexus_serialization_context). The worker encodes the workflow's result and closingfailure with it. Clients decode the result with the context the server returns in the history,
including detached handles and runs after continue-as-new.
failure and last-attempt failure with it. Get-result and describe decode those with the context
the server returns.
and fall back to the start request's context on older servers.
heartbeat details, and static summary and details.
1.9.3-nexus-serialization-context.Why?
A Nexus handler returns the result and failure of the execution it starts to the Nexus caller. The
caller decodes them with the operation's
NexusSerializationContext. Until now the backingexecution encoded them with its own workflow or activity context, so a codec keyed on the context
couldn't decode them. Per-endpoint encryption is one example.
Checklist
Closes
How was this tested:
execution, and for workflow replay with an injected propagated context.
NexusSerializationContextPropagationTeston the dev server: a caller workflow goesthrough a Nexus endpoint to a handler that starts a workflow, a continue-as-new workflow or an
SAA. It checks history payload contexts, describe, and handles obtained by ID.
StandaloneNexusSerializationContextTestagainst an external server, for standalone Nexusoperations.