feat(extstore): Add support for Nexus task handling - #1676
Conversation
|
|
||
| def _decode_reference(self, payload: Payload) -> ExternalStorageReference | None: | ||
| """Decode an external storage reference from a payload.""" | ||
| if len(payload.external_payloads) == 0: |
There was a problem hiding this comment.
The Nexus task transport drops the external_payloads field. Regardless, it shouldn't be checking for this field anyway and should only rely on the metadata.
| ): | ||
| await asyncio.shield(self._bridge_worker().complete_nexus_task(completion)) | ||
|
|
||
| async def _encode_completion( |
There was a problem hiding this comment.
I'm confused by the lack of decoding. Is that because it was already happening and just didn't do anything since it saw no claims?
There was a problem hiding this comment.
Decoding happens in _DummyPayloadSerializer.deserialize.
| self._data_converter.payload_converter, | ||
| completion.failure, | ||
| ) | ||
| await self._encode_completion(completion) |
There was a problem hiding this comment.
Would it be possible to fold this step into self._complete_task? The intent of adding that method was to ensure that all locations that complete a task have the same behavior and this seems like it's applied throughout.
There was a problem hiding this comment.
If I did that, then extra fault handling would have to be written. For example, if external storage faulted on the success completion, I'd have to catch that exception, create a failure, attempt to encode and store that failure, and then submit that completion. It would be duplicated exception handling code and might be convoluted to discriminate.
3 participants
What was changed
external_payloadscondition from reference check.Why?
Round out the support for external storage for all task handlers.
Checklist
How was this tested: Integration tests
Any docs updates needed? No