{"record":{"id":"cb9b3f4778b52b50","repo":"can1357/oh-my-pi","slug":"rpc-chunk-sequence-was-interrupted","errorCode":null,"errorMessage":"RPC chunk sequence was interrupted","messagePattern":"RPC chunk sequence was interrupted","errorType":"exception","errorClass":"RpcError","httpStatus":null,"severity":"error","filePath":"python/omp-rpc/src/omp_rpc/client.py","lineNumber":144,"sourceCode":"\n@dataclass(slots=True)\nclass _PendingRpcChunks:\n    chunk_id: str\n    count: int\n    byte_length: int\n    next_index: int = 0\n    chunks: list[bytes] = field(default_factory=list)\n    received_bytes: int = 0\n\n\nclass _RpcFrameDecoder:\n    def __init__(self) -> None:\n        self._pending: _PendingRpcChunks | None = None\n\n    def push(self, value: object) -> JsonObject | None:\n        if not isinstance(value, dict) or value.get(\"type\") != \"rpc_chunk\":\n            if self._pending is not None:\n                raise RpcError(\"RPC chunk sequence was interrupted\")\n            if not isinstance(value, dict):\n                raise RpcError(\"RPC frame must be a JSON object\")\n            return cast(JsonObject, value)\n\n        chunk_id = value.get(\"chunkId\")\n        index = value.get(\"index\")\n        count = value.get(\"count\")\n        byte_length = value.get(\"byteLength\")\n        data = value.get(\"data\")\n        max_chunk_count = (\n            _MAX_RPC_REASSEMBLED_BYTES + _RPC_CHUNK_PAYLOAD_BYTES - 1\n        ) // _RPC_CHUNK_PAYLOAD_BYTES\n        if (\n            not isinstance(chunk_id, str)\n            or not chunk_id\n            or len(chunk_id) > 128\n            or not isinstance(index, int)\n            or isinstance(index, bool)","sourceCodeStart":126,"sourceCodeEnd":162,"githubUrl":"https://github.com/can1357/oh-my-pi/blob/969062200754ea02cfac922e5ebb8c608c079e15/python/omp-rpc/src/omp_rpc/client.py#L126-L162","documentation":"_RpcFrameDecoder.push() reassembles multi-frame `rpc_chunk` sequences. Once a chunk sequence is pending, the very next frame must be another `rpc_chunk` of the same chunkId; if any non-chunk frame (or non-dict value) arrives first, it raises RpcError(\"RPC chunk sequence was interrupted\"). This preserves protocol integrity — a large RPC response whose chunks were interleaved with or truncated by other frames cannot be reassembled safely.","triggerScenarios":"A server streaming a chunked RPC response sends a regular frame (event, result, error) between chunks; the chunked message is truncated mid-stream and another message follows; a malformed client injects frames into the stream; multiple writers share the transport without chunk-aware serialization.","commonSituations":"Custom/older RPC peers that don't implement chunked framing; proxies or loggers rewriting the frame stream; a >256KB RPC payload (forcing chunking) combined with concurrent events on the same connection.","solutions":["Ensure the peer emits all chunks of a chunkId contiguously (same chunkId, indices 0..count-1) with no intervening frames.","Update the client/server to a matching protocol version that both support chunked responses.","Catch RpcError and resynchronize: reset/reconnect the RPC stream, since decoder state (_pending) must be discarded.","Check for concurrent writers on the transport — serialize sends with a lock."],"exampleFix":"# before: interleaving a status event mid-transfer\nsend({\"type\": \"event\", ...})   # interrupts pending chunks\n# after: buffer events until the chunk sequence completes\nsend_all_chunks_first(chunk_id, chunks)\nsend({\"type\": \"event\", ...})","handlingStrategy":"try-catch","validationCode":"# Validate frames before feeding the decoder: once a rpc_chunk sequence starts,\n# no other frame type may be sent until all `count` chunks arrive.\ndef validate_chunk_stream(frames):\n    pending = False\n    for f in frames:\n        if pending and not (isinstance(f, dict) and f.get(\"type\") == \"rpc_chunk\"):\n            raise ValueError(\"non-chunk frame inside a chunked RPC sequence\")\n        pending = isinstance(f, dict) and f.get(\"type\") == \"rpc_chunk\"","typeGuard":"def is_rpc_chunk(value: object) -> bool:\n    return isinstance(value, dict) and value.get(\"type\") == \"rpc_chunk\"","tryCatchPattern":"decoder = _RpcFrameDecoder()\ntry:\n    frame = decoder.push(raw_value)\nexcept RpcError as e:\n    if 'chunk sequence was interrupted' in str(e):\n        decoder = _RpcFrameDecoder()  # reset decoder state, then reconnect/resync\n        raise ConnectionError('RPC chunked response corrupted; stream resynchronized') from e\n    raise","preventionTips":["Emit all chunks of one chunkId back-to-back; never interleave events or other RPC frames.","Keep client and server protocol versions in sync regarding chunked responses.","Serialize transport writes (lock) so concurrent sends cannot splice a chunk stream.","Guard large RPC payloads (>256KB trigger chunking) against mid-stream sender failures and reconnect cleanly."],"tags":["rpc","protocol","streaming"],"backgroundTag":"rpc-chunk-sequence-interrupted","analyzedSha":"969062200754ea02cfac922e5ebb8c608c079e15","analyzedAt":"2026-08-31T10:29:35.737Z","schemaVersion":2},"datasetVersion":"2026-08-31T14:17:45.589Z"}