NexusSerializationContext for data/failure converters - #1828
NexusSerializationContext for data/failure converters#1828JoshuaFrenchwood wants to merge 3 commits into
Conversation
f36df7a to
0774bf3
Compare
0774bf3 to
21ee938
Compare
| if self._input.summary: | ||
| command.user_metadata.summary.CopyFrom( | ||
| self._payload_converter.to_payload(self._input.summary) | ||
| self._summary_payload_converter.to_payload(self._input.summary) |
There was a problem hiding this comment.
This appears to be the only case where we ensure a contextless serialization for summary, and I'm not sure that would remain true for codec application. Why do this?
There was a problem hiding this comment.
Removed this, and just used the context serialization for summary
| self._workflow_context_failure_converter, | ||
| ) | ||
| summary_payload_converter = payload_converter | ||
| failure_converter = self._context_free_failure_converter |
There was a problem hiding this comment.
We should have a todo here. @VegetarianOrc already needs to extend this for worker callbacks.
There was a problem hiding this comment.
Added a TODO for this
| service_handler is None | ||
| or operation not in service_handler.service.operation_definitions | ||
| ): | ||
| return self._data_converter |
There was a problem hiding this comment.
Should this really be a silent fallback to not having context? Why did we even need to find the service handler?
There was a problem hiding this comment.
I removed the fallback from here, and just returned the nexus serialization context directly
| rpc_metadata: Mapping[str, str | bytes] | ||
| rpc_timeout: timedelta | None | ||
| result_type: type[Any] | None | ||
| _nexus_serialization_context: NexusSerializationContext | None = None |
There was a problem hiding this comment.
It doesn't seem like this should be on the intercepted object.
There was a problem hiding this comment.
Was able to remove this because I moved decoding to the nexus operation handle
| ), | ||
| ) | ||
|
|
||
| async def get_nexus_operation_result( |
There was a problem hiding this comment.
This may need some thought, it appears to work differently from any other interceptor or handle in the SDK, which is why you needed to pass the context through the interceptor. None of the others actually perform decoding, and this is the only result getter which allows for interception.
There was a problem hiding this comment.
Agreed, I moved the decoding logic into the nexus operation handle.
070740b to
c91cfa3
Compare
2 participants
What was changed
Added NexusSerializationContext for nexus callers and sync handlers. This allows data and failure converters to use nexus endpoint, service and operation to be used for encoding and decoding.
Why?
This enables codecs to select serialization behavior or encryption keys by Nexus endpoint, service, or operation.
For example, workflows calling two Nexus endpoints can encrypt each endpoint’s payloads with a different key while ensuring that inputs, results, and failures are decoded with the converter selected for the corresponding operation.
Checklist
Closes
How was this tested:
Added unit/functional tests to verify that NexusSerializationContext works as expected.