fix(rpc): accept exact-boundary v2 chunks
This commit is contained in:
@@ -147,7 +147,7 @@ export class RpcFrameDecoder {
|
||||
count < 2 ||
|
||||
count > Math.ceil(MAX_RPC_REASSEMBLED_BYTES / RPC_CHUNK_PAYLOAD_BYTES) ||
|
||||
index >= count ||
|
||||
byteLength <= MAX_RPC_FRAME_BYTES ||
|
||||
byteLength < MAX_RPC_FRAME_BYTES ||
|
||||
byteLength > MAX_RPC_REASSEMBLED_BYTES
|
||||
)
|
||||
throw new Error("invalid rpc chunk metadata");
|
||||
|
||||
@@ -208,6 +208,27 @@ describe("RPC frame encoding", () => {
|
||||
expect(decoded).toEqual(frame);
|
||||
});
|
||||
|
||||
it("accepts a chunked logical frame at the exact physical-frame boundary", () => {
|
||||
const frame = {
|
||||
id: "request-boundary",
|
||||
type: "response",
|
||||
command: "get_state",
|
||||
success: true,
|
||||
data: { payload: "" },
|
||||
};
|
||||
const emptyBytes = Buffer.byteLength(JSON.stringify(frame), "utf8");
|
||||
frame.data.payload = "x".repeat(MAX_RPC_FRAME_BYTES - emptyBytes);
|
||||
expect(Buffer.byteLength(JSON.stringify(frame), "utf8")).toBe(MAX_RPC_FRAME_BYTES);
|
||||
|
||||
const encoder = new RpcFrameEncoder();
|
||||
encoder.setProtocolVersion(2);
|
||||
const decoder = new RpcFrameDecoder();
|
||||
let decoded: object | undefined;
|
||||
for (const line of encoder.encode(frame).trimEnd().split("\n")) decoded = decoder.push(JSON.parse(line));
|
||||
|
||||
expect(decoded).toEqual(frame);
|
||||
});
|
||||
|
||||
it("preserves terminal message counts above the protocol v2 ceiling", () => {
|
||||
const encoder = new RpcFrameEncoder();
|
||||
encoder.setProtocolVersion(2);
|
||||
|
||||
@@ -164,7 +164,7 @@ class _RpcFrameDecoder:
|
||||
or count < 2
|
||||
or count > max_chunk_count
|
||||
or index >= count
|
||||
or byte_length <= _MAX_RPC_FRAME_BYTES
|
||||
or byte_length < _MAX_RPC_FRAME_BYTES
|
||||
or byte_length > _MAX_RPC_REASSEMBLED_BYTES
|
||||
or not isinstance(data, str)
|
||||
or not data
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import signal
|
||||
@@ -11,6 +13,7 @@ import time
|
||||
import unittest
|
||||
|
||||
from omp_rpc import RpcClient, RpcCommandError, RpcConcurrencyError, RpcError, host_tool
|
||||
from omp_rpc.client import _RpcFrameDecoder
|
||||
|
||||
|
||||
FAKE_SERVER = textwrap.dedent(
|
||||
@@ -735,6 +738,38 @@ class RpcClientTests(unittest.TestCase):
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def test_protocol_v2_decoder_accepts_exact_logical_boundary(self) -> None:
|
||||
frame = {
|
||||
"id": "request-boundary",
|
||||
"type": "response",
|
||||
"command": "get_state",
|
||||
"success": True,
|
||||
"data": {"payload": ""},
|
||||
}
|
||||
encoded_empty = json.dumps(frame, separators=(",", ":")).encode("utf-8")
|
||||
frame["data"]["payload"] = "x" * (1024 * 1024 - len(encoded_empty))
|
||||
encoded = json.dumps(frame, separators=(",", ":")).encode("utf-8")
|
||||
self.assertEqual(len(encoded), 1024 * 1024)
|
||||
|
||||
decoder = _RpcFrameDecoder()
|
||||
chunk_size = 256 * 1024
|
||||
count = (len(encoded) + chunk_size - 1) // chunk_size
|
||||
decoded = None
|
||||
for index in range(count):
|
||||
chunk = encoded[index * chunk_size : (index + 1) * chunk_size]
|
||||
decoded = decoder.push(
|
||||
{
|
||||
"type": "rpc_chunk",
|
||||
"chunkId": "exact-boundary",
|
||||
"index": index,
|
||||
"count": count,
|
||||
"byteLength": len(encoded),
|
||||
"data": base64.b64encode(chunk).decode("ascii"),
|
||||
}
|
||||
)
|
||||
|
||||
self.assertEqual(decoded, frame)
|
||||
|
||||
def test_command_builder_supports_common_rpc_options(self) -> None:
|
||||
client = RpcClient(
|
||||
executable="omp",
|
||||
|
||||
Reference in New Issue
Block a user