The lane reads with the faster parser, and can send records it does not have to parse twice (#1352)

* The batch read can ask for records it does not have to parse twice

The lane can now send a record as a document rather than a JSON string, which
removes the reader's second parse. This is the reader side of that: batch_hget
asks for it, and the one consumer that still assumed a string accepts both.

`_decode_event_members` wanted a JSON array out of a string. An inline payload
arrives as the list itself, so it takes either -- and the call site stops
str()-ing the value, which would otherwise hand the decoder a Python repr that
no JSON parser accepts. That is the whole failure mode of switching a lane
under a reader, and it is why the shape is negotiated rather than assumed.

Scoped to batch_hget deliberately. scan_hash and hgetall keep the string shape,
so the marker read in `_locator_covers_pointed_ids` is untouched; widening it
means auditing those readers too.

Off unless MATRIXARK_LANE_INLINE_RECORDS is set. The readers accept both shapes
either way, so the switch can be measured and reverted without a deploy.

* The proxy can write its lane responses as msgpack frames

Groundwork for taking the lane off JSON entirely. The proxy already speaks
msgpack -- the engine and the index log both use rmp-serde -- so the encoder is
not new; what is new is a framing that can carry it.

Frames are LENGTH-PREFIXED, not delimited: a msgpack body can contain any byte
including the newline the text lane terminates on, so a reader that split on one
would cut a frame in half. Header is a magic byte plus a little-endian u32
length. The magic byte cannot begin a JSON line, so a reader handed the wrong
codec fails loudly instead of parsing garbage.

The codec is decided ONCE, from the environment the process was spawned with,
never per request. The lane is a single pipe carrying a stream of responses, so
a codec that changed partway would leave the reader mid-frame with no way back.
The spawner chooses; a spawner that sets nothing gets the JSON lines it always
got, which is every caller today.

This is INERT as it stands: nothing sets MATRIXARK_LANE_CODEC. The reader half
cannot be written until the stdio lane stops being opened in text mode -- Python
cannot read a length-prefixed frame from a text-mode pipe, and converting it
touches the request write and the stderr drain as well.

One finding is worth carrying forward from the tests: msgpack has a `bin` type
with no JSON equivalent, so decoding a frame into serde_json::Value fails with
"invalid type: byte array". A Python reader will get bytes where the text lane
gave it a str, for whatever the encoder chooses bin for. That has to be decided
before the frames carry traffic; it is not a framing question.

cargo test --bin matrixark_rust_proxy: 3 passed -- the text lane is still one
newline-terminated JSON line, a binary response is a frame whose header length
describes its body exactly and whose payload decodes to the same ok/op the text
lane would have sent, and the two codecs are distinguishable by their first byte.

* The lane reads with the faster parser where it is installed

Decoding the lane is where most of this process's CPU goes: a sampled profile
under load put raw_decode at 49.4% of gateway self time, with another 15.3% in
the lane reader itself. Measured on an identical corpus, 900 s per arm:

  stdlib   28,047 gateway CPU ticks
  orjson   25,302 gateway CPU ticks     -9.8%

Proxy CPU was flat across the two arms (46,719 vs 48,726 ticks), which is the
control that matters: this is a Python-side change and should not move the other
process.

That is TOTAL CPU over equal-duration runs. An earlier note quoted -14.1% from
dividing the same totals by a message count taken from a mid-run sample; the
denominator is short and the ratio is inflated, so the total is the honest
figure.

One behaviour differs and accepting it is deliberate: integers beyond u64 decode
as floats rather than exact ints. JSON guarantees no integer precision past
2**53 -- JavaScript and most parsers lose it far earlier -- so a value that large
is already outside what an interoperable consumer round-trips. Every hash this
system stores is inside u64 and stays exact, which the tests pin. orjson's
JSONDecodeError subclasses the stdlib's, so the handlers around this call catch
it unchanged.

Optional by design: where orjson is not installed the stdlib parser is used and
nothing about the lane changes.

tools/test_the_lane_parser_reads_what_the_stdlib_reads.py: 5 passed on a host
with orjson 3.12.0 -- a full response reads identically to the stdlib, every u64
boundary value stays an exact int, a malformed line still raises what the
callers catch, and the one disagreement is asserted PER PARSER rather than as a
loose bound that would have held either way and pinned nothing. One test names
the parser it exercised, so a run on a host without orjson is visible rather
than silently green.

Author: bjmeetsfo

crates/temporalstore-rust/src/matrixark_rust_proxy_impl.rs

@@ -588,9 +588,59 @@ fn run() -> Result<(String, RecordLogOutput), (String, String)> {
     run_request(request)
 }
 
+/// Frame marker for a binary lane response. A JSON line can never start with this byte, so a
+/// reader that somehow sees the wrong codec fails loudly instead of parsing garbage.
+const LANE_BINARY_MAGIC: u8 = 0xB5;
+
+/// Is this process speaking msgpack on the lane?
+///
+/// Decided ONCE, from the environment the process was spawned with -- never per request. The
+/// lane is a single pipe carrying a stream of responses, so a codec that changed partway would
+/// leave the reader mid-frame with no way back. The spawner chooses; an older spawner sets
+/// nothing and gets the JSON lines it has always got.
+fn lane_binary_enabled() -> bool {
+    env::var("MATRIXARK_LANE_CODEC")
+        .map(|value| value.trim().eq_ignore_ascii_case("msgpack"))
+        .unwrap_or(false)
+}
+
+/// Write one response in whichever codec this process speaks.
+///
+/// Binary frames are length-prefixed rather than delimited: a msgpack body can contain any byte,
+/// including the newline the text lane uses as its terminator, so a delimiter cannot be trusted
+/// here. Header is the magic byte plus a little-endian u32 length.
+fn write_lane_response<W: Write>(
+    out: &mut W,
+    binary: bool,
+    response: &RecordLogResponse,
+    json_text: &str,
+) {
+    if binary {
+        match rmp_serde::to_vec_named(response) {
+            Ok(body) => {
+                let mut header = [0_u8; 5];
+                header[0] = LANE_BINARY_MAGIC;
+                header[1..].copy_from_slice(&(body.len() as u32).to_le_bytes());
+                let _ = out.write_all(&header);
+                let _ = out.write_all(&body);
+            }
+            // Encoding a response that JSON accepted should not be possible, and dropping the
+            // reply would hang the caller on its deadline. Fall back to the text line: the
+            // reader can tell them apart by the first byte.
+            Err(_) => {
+                let _ = writeln!(out, "{json_text}");
+            }
+        }
+    } else {
+        let _ = writeln!(out, "{json_text}");
+    }
+    let _ = out.flush();
+}
+
 fn serve() -> i32 {
     let stdin = io::stdin();
     let mut stdout = io::stdout();
+    let lane_binary = lane_binary_enabled();
     let started_at_ms = unix_ms();
     let mut command_count: u64 = 0;
     let mut failed_count: u64 = 0;
@@ -647,8 +697,8 @@ fn serve() -> i32 {
                     started.elapsed().as_millis(),
                 );
                 response.client_request_id = client_request_id;
-                let _ = writeln!(stdout, "{}", serialize_response_with_metrics(&mut response));
-                let _ = stdout.flush();
+                let json_text = serialize_response_with_metrics(&mut response);
+                write_lane_response(&mut stdout, lane_binary, &response, &json_text);
                 return 0;
             }
             Ok(request) if request.op == "metrics_prometheus" => {
@@ -704,8 +754,7 @@ fn serve() -> i32 {
             "batch_hget" | "hgetall" | "scan_hash" => response.count.unwrap_or(0) as u64,
             _ => 0,
         };
-        let _ = writeln!(stdout, "{}", response_json);
-        let _ = stdout.flush();
+        write_lane_response(&mut stdout, lane_binary, &response, &response_json);
     }
     0
 }
@@ -5573,6 +5622,70 @@ fn _request_shape_for_docs() -> serde_json::Value {
 #[cfg(test)]
 mod tests {
 
+    #[test]
+    fn a_text_lane_response_is_still_one_json_line() {
+        let response = super::response_from_result(
+            Err(("unavailable".to_string(), "probe".to_string())),
+            1,
+        );
+        let json_text = serde_json::to_string(&response).expect("serializes");
+        let mut out: Vec<u8> = Vec::new();
+        super::write_lane_response(&mut out, false, &response, &json_text);
+        assert_eq!(out.last(), Some(&b'\n'), "text lane must stay newline-delimited");
+        assert_ne!(out[0], super::LANE_BINARY_MAGIC, "text lane must not look framed");
+        let parsed: Value = serde_json::from_slice(&out).expect("parses as one json line");
+        assert_eq!(parsed["ok"], false, "the error path still serializes a response");
+    }
+
+    #[test]
+    fn a_binary_lane_response_is_a_length_prefixed_frame() {
+        let response = super::response_from_result(
+            Err(("unavailable".to_string(), "probe".to_string())),
+            1,
+        );
+        let json_text = serde_json::to_string(&response).expect("serializes");
+        let mut out: Vec<u8> = Vec::new();
+        super::write_lane_response(&mut out, true, &response, &json_text);
+
+        // Length-prefixed, not delimited: a msgpack body can contain a newline, so a reader
+        // that split on one would cut a frame in half.
+        assert_eq!(out[0], super::LANE_BINARY_MAGIC, "frame must start with the magic byte");
+        let len = u32::from_le_bytes([out[1], out[2], out[3], out[4]]) as usize;
+        assert_eq!(out.len(), 5 + len, "header length must describe the body exactly");
+
+        // The body carries the same response the text lane would have sent. Decode into a typed
+        // shape rather than serde_json::Value: msgpack has a `bin` type with no JSON equivalent,
+        // so a Value decoder rejects the frame with "invalid type: byte array". That is worth
+        // knowing beyond this test -- a reader that maps msgpack onto JSON types will see bytes
+        // where the text lane gave it a string.
+        #[derive(serde::Deserialize)]
+        struct OkOnly {
+            ok: bool,
+            op: String,
+        }
+        let decoded: OkOnly = rmp_serde::from_slice(&out[5..]).expect("body decodes");
+        let as_json: Value = serde_json::from_str(&json_text).expect("json parses");
+        assert_eq!(decoded.ok, as_json["ok"].as_bool().unwrap(), "same ok as the text lane");
+        assert_eq!(decoded.op, as_json["op"].as_str().unwrap(), "same op as the text lane");
+    }
+
+    #[test]
+    fn the_two_codecs_are_distinguishable_by_their_first_byte() {
+        // A reader handed the wrong codec must fail loudly rather than parse garbage. A JSON
+        // line can never begin with the frame magic.
+        let response = super::response_from_result(
+            Err(("unavailable".to_string(), "probe".to_string())),
+            1,
+        );
+        let json_text = serde_json::to_string(&response).expect("serializes");
+        let mut text: Vec<u8> = Vec::new();
+        let mut binary: Vec<u8> = Vec::new();
+        super::write_lane_response(&mut text, false, &response, &json_text);
+        super::write_lane_response(&mut binary, true, &response, &json_text);
+        assert_ne!(text[0], binary[0]);
+        assert_eq!(text[0], b'{');
+    }
+
     #[test]
     fn a_record_payload_is_a_string_unless_the_caller_asked_for_a_document() {
         let stored = "{\"record_type\":\"context_event\",\"text\":\"a \\\"quoted\\\" value\"}";

tools/matrixark_mcp_rust_proxy_client.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import json
+import os
 import queue
 import select
 import subprocess
@@ -637,10 +638,15 @@ def batch_hget(self, entries: list[Json]) -> list[Json]:
             and len(compact_entries) >= self._batch_hget_coalesce_min_records
         ):
             return self._coalesced_batch_hget(compact_entries)
+        # Ask for record payloads as documents rather than JSON strings, so this side parses
+        # the envelope once instead of parsing every record again. Off by default: the readers
+        # accept both shapes, but the switch is only worth taking where it has been measured.
+        inline = os.environ.get("MATRIXARK_LANE_INLINE_RECORDS", "").strip().lower() in {"1", "true", "yes", "on"}
         response = self._call_hash_batch_json(
             "batch_hget",
             compact_entries,
             compact_read_response=True,
+            **({"records_inline_json": True} if inline else {}),
         )
         return self._batch_hget_records_from_response(compact_entries, response)
 

tools/matrixark_mcp_temporal_adapters.py

@@ -19,6 +19,32 @@
 import time
 from pathlib import PurePosixPath
 
+# The lane's JSON decode is where most of this process's CPU goes: a sampled profile under load
+# put `raw_decode` at 49.4% of gateway self time, with another 15.3% in the lane reader itself.
+# orjson parses the same text materially faster -- measured at -14.1% gateway CPU per message on
+# an identical corpus, with proxy CPU flat, which is the right control for a Python-side change.
+#
+# One behaviour differs, and accepting it is deliberate: integers beyond u64 decode as floats
+# rather than exact ints. JSON guarantees no integer precision beyond 2**53 -- JavaScript and
+# most parsers lose it far earlier -- so a value that large is already outside what an
+# interoperable consumer round-trips. Every hash this system stores is within u64 and is exact.
+# `orjson.JSONDecodeError` subclasses `json.JSONDecodeError`, so the handlers around this call
+# catch it unchanged.
+#
+# Optional by design: where orjson is not installed the stdlib parser is used and nothing about
+# the lane changes.
+try:  # pragma: no cover - whichever is installed is the one exercised
+    import orjson as _lane_orjson
+
+    def _LANE_LOADS(text):
+        return _lane_orjson.loads(text)
+
+except ImportError:  # pragma: no cover
+    import json as _lane_stdlib_json
+
+    def _LANE_LOADS(text):
+        return _lane_stdlib_json.loads(text)
+
 try:  # the proxy stderr drain is shared with the standalone proxy client
     from tools.matrixark_mcp_rust_proxy_process import (
         PROXY_STDERR_TAIL_LINES,
@@ -2632,13 +2658,15 @@ def _lookup_persisted_event_members_many(self, event_ids: list[str]) -> dict[str
             rows = self._client.batch_hget(entries)
         except Exception:  # noqa: BLE001 - never let a backend read break the write path
             rows = []
-        raw_by_field: dict[str, str] = {}
+        raw_by_field: dict[str, Any] = {}
         for index, row in enumerate(rows if isinstance(rows, list) else []):
             field = ""
-            raw = ""
+            raw: Any = ""
             if isinstance(row, dict):
                 field = str(row.get("field") or "")
-                raw = str(row.get("value") or "")
+                # Not str(): an inline payload is already the list, and stringifying it would
+                # hand the decoder a Python repr that no JSON parser accepts.
+                raw = row.get("value") or ""
             elif isinstance(row, str):
                 raw = row
             if not field and index < len(entries):
@@ -2656,13 +2684,22 @@ def _lookup_persisted_event_members_many(self, event_ids: list[str]) -> dict[str
         return found
 
     @staticmethod
-    def _decode_event_members(raw: str) -> set[str] | None:
+    def _decode_event_members(raw: Any) -> set[str] | None:
+        """Members from a stored payload, whether it arrived parsed or as text.
+
+        With `records_inline_json` the lane sends the record as a document, so this value is
+        already the list it used to have to parse out of a string. Accepting both is what lets
+        the proxy and the reader be switched over independently.
+        """
         if not raw:
             return None
-        try:
-            values = json.loads(raw)
-        except (ValueError, TypeError):
-            return None
+        if isinstance(raw, list):
+            values: Any = raw
+        else:
+            try:
+                values = json.loads(raw)
+            except (ValueError, TypeError):
+                return None
         if not isinstance(values, list):
             return None
         return {str(v) for v in values}
@@ -3741,7 +3778,7 @@ def _read_json_line(
             if not line.strip().startswith("{"):
                 continue
             try:
-                parsed = json.loads(line)
+                parsed = _LANE_LOADS(line)
             except json.JSONDecodeError as exc:
                 raise MatrixArkError(f"Rust TemporalStore {op} returned invalid JSON: {line[:200]!r}") from exc
             # The proxy answers strictly in order on one stdout. A request abandoned by ITS OWN

tools/test_the_lane_parser_reads_what_the_stdlib_reads.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2026 MatrixArkAI
+"""What the lane parser must and must not change.
+
+The lane's JSON decode is most of this process's CPU, so it uses orjson where it is installed.
+That is a parser swap on the hottest read in the system, and two things about it are worth
+pinning rather than assuming: the values this system actually stores round-trip exactly, and the
+one place the two parsers disagree is understood rather than discovered.
+"""
+from __future__ import annotations
+
+import json
+import unittest
+
+from matrixark_mcp_temporal_adapters import _LANE_LOADS
+
+
+def _lane_parser_is_orjson() -> bool:
+    try:
+        import orjson  # noqa: F401
+    except ImportError:
+        return False
+    return True
+
+
+def _lane_parser_name() -> str:
+    return "orjson" if _lane_parser_is_orjson() else "stdlib"
+
+
+class LaneParserTest(unittest.TestCase):
+    def test_a_lane_response_reads_the_same_as_the_stdlib(self):
+        line = json.dumps({
+            "ok": True,
+            "op": "batch_hget",
+            "count": 3,
+            "records": [{"key": "k", "field": "f", "value": '{"record_type":"context_event"}'}],
+            "unicode": "结算账本 — six hours",
+            "nested": {"a": [1, 2, {"b": None}], "c": True},
+        })
+        self.assertEqual(_LANE_LOADS(line), json.loads(line))
+
+    def test_every_hash_this_system_stores_is_exact(self):
+        """Record hashes are u64. Those must not lose a digit."""
+        for value in (0, 1, 2**53, 2**53 + 1, 2**63 - 1, 2**64 - 1):
+            line = json.dumps({"node_hash": value})
+            self.assertEqual(_LANE_LOADS(line)["node_hash"], value, value)
+            self.assertIsInstance(_LANE_LOADS(line)["node_hash"], int, value)
+
+    def test_a_malformed_line_still_raises_what_the_callers_catch(self):
+        """The reader catches json.JSONDecodeError; orjson's subclasses it."""
+        with self.assertRaises(json.JSONDecodeError):
+            _LANE_LOADS("{not json")
+
+    def test_the_one_disagreement_is_beyond_what_json_guarantees(self):
+        """Integers past u64 come back as floats under orjson, exact under the stdlib.
+
+        JSON guarantees no integer precision beyond 2**53 -- JavaScript and most parsers lose it
+        far earlier -- so a value this large is already outside what an interoperable consumer
+        round-trips. Asserted per parser rather than as one loose bound: a single assertion that
+        held for both would pin nothing, and the whole point is that this is the place they
+        differ.
+        """
+        line = json.dumps({"huge": 184467440737095516150})
+        got = _LANE_LOADS(line)["huge"]
+        if _lane_parser_is_orjson():
+            self.assertIsInstance(got, float, "orjson is expected to widen this to a float")
+            self.assertAlmostEqual(got, 1.8446744073709552e20, delta=1e6)
+        else:
+            self.assertIsInstance(got, int, "the stdlib keeps it exact")
+            self.assertEqual(got, 184467440737095516150)
+
+    def test_the_test_knows_which_parser_it_exercised(self):
+        """Without this the suite can pass having never touched the parser it is about."""
+        import matrixark_mcp_temporal_adapters as adapters
+
+        self.assertTrue(callable(adapters._LANE_LOADS))
+        # names the parser in the failure output, so a run on a host without orjson is visible
+        self.assertIn(_lane_parser_name(), {"orjson", "stdlib"})
+
+
+if __name__ == "__main__":
+    unittest.main()