Merge perf/zero-copy-rows: a row on its way to the wire is a window, not a copy

The last term linear in conversation length. decodeRecord copied every message
out of the stored array on every send; the elements are windows onto a frame
that is already on the heap.

  ALIAS ON THE WAY TO THE WIRE. COPY ON THE WAY INTO THE CACHE.

  AssembleBody/streamed (COLD log)   B/op -16.2%  allocs -16.0%  sec -15.7%
  live heap while writing            slice 2.3-4.3MiB  streamed 0.94MiB flat

Two instruments were measuring nothing and were fixed first: the benchmark and
the cache test both read a log the same process had written, which the tree
cache serves resident without decoding a record.

Author: gluck

internal/provider/anthropic/assemble_bench_test.go

@@ -80,11 +80,6 @@ func TestStreamedAssemblyDoesNotRetainTheConversation(t *testing.T) {
 	})
 	a := &Anthropic{Model: "claude-opus-5", ReminderRenderer: "tag"}
 
-	// The store's own cache holds what this daemon appended, which is neither
-	// path's doing and is bounded by the translation budget. It is measured
-	// once and subtracted, so what remains is what THE ASSEMBLY retains.
-	base := liveHeap()
-
 	sliceLive := func() uint64 {
 		perMessage, lts := provider.CollectRows(provider.TranslationRows(rows, 0))
 		req, err := a.projectMessagesWithLTs(oneRowEach(perMessage), lts, snap, nil, 4096, false, "claude-opus-5")
@@ -104,15 +99,22 @@ func TestStreamedAssemblyDoesNotRetainTheConversation(t *testing.T) {
 		return w.live
 	}()
 
+	// NO BASELINE, AND THAT IS THE POINT. Subtracting a floor measured before
+	// the runs made this test flaky: the floor includes the fixture's own
+	// garbage, the heap falls below it once that is collected, and an
+	// unsigned subtraction then reports zero held for BOTH paths -- which
+	// passes or fails depending on the collector's mood.
+	//
+	// The two runs are measured the same way in the same process, so their
+	// DIFFERENCE is the conversation and nothing else. The slice assembler is
+	// holding a body's worth that the streamed one is not.
 	body := bodyBytes(t, a, rows, snap)
-	sliceHeld := sliceLive - base
-	streamedHeld := streamedLive - base
-	t.Logf("held while writing: slice=%d KiB streamed=%d KiB (body=%d KiB, base=%d KiB)",
-		sliceHeld>>10, streamedHeld>>10, body>>10, base>>10)
-	require.Less(t, streamedHeld, sliceHeld/2,
-		"the streamed assembler must not hold the conversation the slice one held")
-	require.Less(t, streamedHeld, body/2,
-		"a streamed body must not retain a body's worth of anything")
+	t.Logf("live heap while writing: slice=%d KiB streamed=%d KiB (body=%d KiB)",
+		sliceLive>>10, streamedLive>>10, body>>10)
+	require.Greater(t, sliceLive, streamedLive,
+		"the slice assembler must hold more than the streamed one")
+	require.GreaterOrEqual(t, sliceLive-streamedLive, body/2,
+		"the gap between them should be most of a body: the conversation the slice path holds")
 }
 
 // liveHeap is the live heap after a collection: the floor a measurement is
@@ -181,11 +183,18 @@ func setPatch(kv map[string]string) message.Patch {
 // benchRowLog is a translator channel in a real store, holding a plausible
 // turn shape: a user message, an assistant message with a tool call, and its
 // result.
+//
+// IT REOPENS THE BACKEND BEFORE HANDING THE LOG BACK, and that is the whole
+// difference between measuring something and measuring nothing. An append
+// SEEDS the tree cache, so a log read in the process that wrote it is served
+// resident and never decodes a single record -- which is not the case that
+// costs anything. The rows a real send pays for are the ones written before
+// the daemon started: cold, and decoded from the frame every time.
 func benchRowLog(tb testing.TB, n int) store.Log[[]json.RawMessage] {
 	tb.Helper()
-	be, err := store.NewXwalBackend(tb.TempDir(), 0)
+	root := tb.TempDir()
+	be, err := store.NewXwalBackend(root, 0)
 	require.NoError(tb, err)
-	tb.Cleanup(func() { be.Close() })
 	outfit, err := be.CreateOutfit("l", setPatch(map[string]string{"system.model": "m"}))
 	require.NoError(tb, err)
 	aria, _, err := be.ForkWith(outfit, 0, setPatch(map[string]string{"system.cwd": "/tmp"}))
@@ -211,7 +220,14 @@ func benchRowLog(tb testing.TB, n int) store.Log[[]json.RawMessage] {
 		})
 		require.NoError(tb, err)
 	}
-	return rows
+	require.NoError(tb, be.Close())
+
+	cold, err := store.NewXwalBackend(root, 0)
+	require.NoError(tb, err)
+	tb.Cleanup(func() { cold.Close() })
+	coldRows, err := cold.OpenTranslator(aria, "anthropic")
+	require.NoError(tb, err)
+	return coldRows
 }
 
 func padding(i int) string {

internal/store/rowsplit.go

@@ -0,0 +1,121 @@
+package store
+
+import (
+	"encoding/json"
+
+	"github.com/jack-work/figaro/internal/store/xwal"
+)
+
+// Rows without the copy.
+//
+// A translator record holds a JSON ARRAY of wire messages, and decoding it
+// with encoding/json costs one allocation and one memcpy PER MESSAGE:
+// json.RawMessage.UnmarshalJSON is `append(m[0:0], data...)`, so a 2KB row
+// becomes a 2KB copy that lives until the request ends. Multiplied by the
+// conversation, that was the last term still linear in history -- roughly
+// 2.6KB and 24 allocations per row per send, measured.
+//
+// The array is already on the heap, whole, in r.Payload. Its elements are
+// WINDOWS onto it. Splitting rather than decoding turns N copies into N slice
+// headers.
+//
+// # THE RULE, AND IT IS NOT NEGOTIABLE
+//
+//	ALIAS ON THE WAY TO THE WIRE. COPY ON THE WAY INTO THE CACHE.
+//
+// An aliased row is a view into the frame the segment handed back. That is
+// safe -- the codec allocates a fresh buffer per read (segment/codec.go:
+// `buf := make([]byte, length)`), nothing is mmap'd, nothing is reused, and no
+// caller writes through a row -- but it PINS the frame for as long as the row
+// lives, and it makes the row's apparent size (len) smaller than the memory it
+// holds.
+//
+// For a row on its way to the request body that is free: it lives for one
+// write and is dropped, so nothing is pinned that was not already resident.
+//
+// For a row on its way INTO THE TREE CACHE it is a lie. The cache sizes an
+// entry as the sum of its rows' lengths (transEntrySize), so an entry whose
+// rows are windows onto a larger frame reports less than it holds, and the
+// budget stops bounding what it believes it is bounding. THAT IS EXACTLY THE
+// FAILURE eeec2bc0 WAS WRITTEN ABOUT -- "it held SLICES RETURNED BY THE LOG,
+// pinning payloads the window believed it had evicted" -- and re-introducing
+// it here, one layer down, would be the same bug wearing a different hat.
+//
+// So the two decoders are separate functions with separate call sites:
+// ScanRange aliases, everything else copies. Enforced by
+// TestCacheSeededRowsDoNotAliasTheirFrame.
+
+// splitJSONArray splits a JSON array into its elements WITHOUT COPYING: every
+// returned row is a window onto b. ok is false for anything that is not an
+// array, which sends the caller back to encoding/json.
+func splitJSONArray(b []byte) ([]json.RawMessage, bool) {
+	i := skipJSONSpace(b, 0)
+	if i >= len(b) || b[i] != '[' {
+		return nil, false
+	}
+	i = skipJSONSpace(b, i+1)
+	if i < len(b) && b[i] == ']' {
+		if skipJSONSpace(b, i+1) != len(b) {
+			return nil, false
+		}
+		// An empty array is an empty slice, not a nil one: encoding/json
+		// makes that distinction and this function answers as it does.
+		return []json.RawMessage{}, true
+	}
+	var out []json.RawMessage
+	for {
+		start := i
+		end, ok := xwal.JSONValueEnd(b, i)
+		if !ok || end <= start {
+			return nil, false
+		}
+		// Three-index slice: a row may not grow into its neighbour by append.
+		out = append(out, json.RawMessage(b[start:end:end]))
+		i = skipJSONSpace(b, end)
+		if i >= len(b) {
+			return nil, false
+		}
+		switch b[i] {
+		case ',':
+			i = skipJSONSpace(b, i+1)
+			if i >= len(b) || b[i] == ']' {
+				return nil, false // a trailing comma is not our shape
+			}
+		case ']':
+			if skipJSONSpace(b, i+1) != len(b) {
+				return nil, false
+			}
+			return out, true
+		default:
+			return nil, false
+		}
+	}
+}
+
+func skipJSONSpace(b []byte, i int) int {
+	for i < len(b) {
+		switch b[i] {
+		case ' ', '\t', '\n', '\r':
+			i++
+		default:
+			return i
+		}
+	}
+	return i
+}
+
+// aliasRows reports the rows of a payload as windows onto it, for a T that is
+// []json.RawMessage and nothing else. ok is false for every other T and for a
+// payload the splitter does not recognise, and the caller decodes normally.
+func aliasRows[T any](payload []byte, v *T) bool {
+	rows, ok := any(v).(*[]json.RawMessage)
+	if !ok {
+		return false
+	}
+	parts, ok := splitJSONArray(payload)
+	if !ok {
+		return false
+	}
+	*rows = parts
+	return true
+}

internal/store/rowsplit_test.go

@@ -0,0 +1,168 @@
+package store
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+	"unsafe"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/jack-work/figaro/internal/store/xwal"
+)
+
+// aliases reports whether row is a window onto payload rather than a copy.
+// It compares the ADDRESSES of the backing arrays, which is the only honest
+// way to ask: equal bytes prove nothing.
+func aliases(payload []byte, row []byte) bool {
+	if len(payload) == 0 || len(row) == 0 {
+		return false
+	}
+	base := uintptr(unsafe.Pointer(&payload[0]))
+	at := uintptr(unsafe.Pointer(&row[0]))
+	return at >= base && at < base+uintptr(len(payload))
+}
+
+// The splitter must agree with encoding/json on every payload the store can
+// hold, and disagree with it about nothing but where the bytes live.
+func TestSplitJSONArrayMatchesEncodingJSON(t *testing.T) {
+	cases := []string{
+		`[]`,
+		`[{"role":"user"}]`,
+		`[{"role":"user"},{"role":"assistant"}]`,
+		`[{"a":[1,2,{"b":"]"}]},{"c":"}"}]`,
+		`[{"text":"a \" quote and a \\\\ backslash"}]`,
+		`[{"text":"bracket ] inside a string"},{"text":"brace } inside"}]`,
+		`[{"text":"\u00e9\u005d"}]`,
+		`  [ {"a":1} , {"b":2} ]  `,
+		`[1,2,3]`,
+		`[true,false,null]`,
+		`[{"nested":{"deep":{"deeper":[{"x":1}]}}}]`,
+		`[{"text":"newline\nand\ttab"}]`,
+		`[""]`,
+		`["]","[",","]`,
+	}
+	for _, in := range cases {
+		t.Run(in, func(t *testing.T) {
+			// ONE conversion, used for both. []byte(s) allocates a fresh
+			// array every time it is written, so converting twice compares
+			// two unrelated buffers -- which reported copies that were not
+			// copies, and could just as easily have reported an alias that
+			// was not one, if the two happened to land adjacent.
+			payload := []byte(in)
+			var want []json.RawMessage
+			require.NoError(t, json.Unmarshal(payload, &want), "fixture is not valid JSON")
+
+			got, ok := splitJSONArray(payload)
+			require.True(t, ok, "splitter refused a payload encoding/json accepted")
+			require.Equal(t, len(want), len(got))
+			for i := range want {
+				require.JSONEq(t, string(want[i]), string(got[i]), "element %d", i)
+			}
+			if len(got) > 0 && len(got[0]) > 0 {
+				require.True(t, aliases(payload, got[0]),
+					"the split element is a copy: the whole point is that it is not")
+			}
+		})
+	}
+}
+
+// Anything that is not an array of ours falls back to encoding/json rather
+// than being guessed at.
+func TestSplitJSONArrayRefusesWhatItCannotRead(t *testing.T) {
+	for _, in := range []string{
+		`null`, `{"not":"an array"}`, `[`, `[{"a":1}`, `[{"a":1},]`, `[,]`,
+		`[{"a":1}] trailing`, ``, `"a string"`, `[{"a":1}][{"b":2}]`,
+	} {
+		t.Run(in, func(t *testing.T) {
+			_, ok := splitJSONArray([]byte(in))
+			require.False(t, ok, "the splitter accepted something it should have refused")
+		})
+	}
+}
+
+// A row on its way to the wire is a window onto the frame.
+func TestScannedRowsAliasTheirFrame(t *testing.T) {
+	payload := []byte(`[{"role":"user","content":"hello"},{"role":"assistant","content":"hi"}]`)
+	rec := xwal.Record{ChannelLT: 1, MainLT: 1, Payload: payload}
+
+	aliased, ok := decodeRecordAliased[[]json.RawMessage](rec)
+	require.True(t, ok)
+	require.Len(t, aliased.Payload, 2)
+	for i, row := range aliased.Payload {
+		require.True(t, aliases(payload, row), "row %d was copied", i)
+	}
+
+	owned, ok := decodeRecord[[]json.RawMessage](rec)
+	require.True(t, ok)
+	require.Len(t, owned.Payload, 2)
+	for i, row := range owned.Payload {
+		require.False(t, aliases(payload, row), "row %d aliases the frame it must own", i)
+		require.JSONEq(t, string(aliased.Payload[i]), string(row))
+	}
+}
+
+// THE RULE, ENFORCED: alias on the way to the wire, copy on the way into the
+// cache.
+//
+// A cached entry is sized as the sum of its rows' lengths. If those rows were
+// windows onto a bigger frame, the entry would report less than it holds and
+// the budget would stop bounding what it believes it bounds -- the same
+// failure eeec2bc0 was written about, one layer down. So a row that reaches
+// the cache must own its bytes, and this test reaches into the cache to check.
+func TestCacheSeededRowsDoNotAliasTheirFrame(t *testing.T) {
+	root := t.TempDir()
+	b, err := NewXwalBackend(root, 0)
+	require.NoError(t, err)
+
+	outfit, err := b.CreateOutfit("l", setPatch(map[string]string{"system.model": "m"}))
+	require.NoError(t, err)
+	aria, _, err := b.ForkWith(outfit, 0, setPatch(map[string]string{"system.cwd": "/tmp"}))
+	require.NoError(t, err)
+	rows, err := b.OpenTranslator(aria, "anthropic")
+	require.NoError(t, err)
+
+	big := strings.Repeat("x", 4096)
+	for i := 0; i < 8; i++ {
+		_, err := rows.Append(Entry[[]json.RawMessage]{
+			FigaroLT: uint64(i + 1),
+			Payload: []json.RawMessage{
+				json.RawMessage(fmt.Sprintf(`{"role":"user","content":%q}`, big)),
+				json.RawMessage(fmt.Sprintf(`{"role":"assistant","content":%q}`, big)),
+			},
+		})
+		require.NoError(t, err)
+	}
+
+	// A COLD BACKEND, OR THIS TEST PROVES NOTHING. An append seeds the tree
+	// cache with the payload the WRITER passed -- never a decoded frame -- so
+	// a read on the same process is served from memory and the decode path
+	// under test never runs. It did not, and the test passed anyway against a
+	// deliberately broken decoder. Reopening is what makes the read cold.
+	require.NoError(t, b.Close())
+	cold, err := NewXwalBackend(root, 0)
+	require.NoError(t, err)
+	defer cold.Close()
+	coldRows, err := cold.OpenTranslator(aria, "anthropic")
+	require.NoError(t, err)
+
+	read := coldRows.Read()
+	require.Len(t, read, 8)
+
+	// TWO ROWS OF ONE RECORD ARE THE TEST. In the frame they are separated
+	// by exactly one comma, so a WINDOW onto that frame puts row 1 at
+	// &row0 + len(row0) + 1. A copy cannot land there except by an accident
+	// no allocator will hand you twice at this size.
+	//
+	// (cap(row) != len(row) is NOT the test: json.RawMessage decodes with
+	// append, which rounds up to a size class, so an owned 4124-byte row has
+	// capacity 4864 and the naive check calls it a window.)
+	for i, e := range read {
+		require.Len(t, e.Payload, 2)
+		first, second := e.Payload[0], e.Payload[1]
+		gap := uintptr(unsafe.Pointer(&second[0])) - uintptr(unsafe.Pointer(&first[0]))
+		require.NotEqual(t, uintptr(len(first)+1), gap,
+			"entry %d: the two rows are contiguous in one frame, so the cache holds windows", i)
+	}
+}

internal/store/xwal/xwal.go

@@ -2475,6 +2475,11 @@ func fastDecodeFrame(f []byte) (uint64, []byte, []byte, int64, bool) {
 	return mainLT, payload, f[i:end], ts, true
 }
 
+// JSONValueEnd is exported so the store's row splitter uses THIS scanner
+// rather than growing a second one. A JSON grammar implemented twice is a
+// JSON grammar that disagrees with itself on the row that matters.
+func JSONValueEnd(b []byte, start int) (int, bool) { return jsonValueEnd(b, start) }
+
 func jsonValueEnd(b []byte, start int) (int, bool) {
 	if start >= len(b) {
 		return 0, false
@@ -2517,8 +2522,12 @@ func jsonValueEnd(b []byte, start int) (int, bool) {
 		}
 		return 0, false
 	default:
+		// A bare scalar ends at the first structural byte that can follow it.
+		// ']' belongs in that set: it cannot occur INSIDE a number, true,
+		// false or null, and without it an array's last scalar element runs
+		// off the end of the array.
 		for i := start; i < len(b); i++ {
-			if b[i] == ',' || b[i] == '}' {
+			if b[i] == ',' || b[i] == '}' || b[i] == ']' {
 				return i, i > start
 			}
 		}