fix(wal): chunk oversized payloads so wide-row ingest stays durable (#677)

[atirna]

Description

Wide ingest requests (a single client body with tens of KB per row) can exceed
the WAL's 100MB single-entry payload cap while staying well under the server's
1GB request limit. Before this change every such write was rejected wholesale
by the WAL: ingest kept reporting healthy while each writer's WAL directory
held a 7-byte header-only file for hours, and the only trace of the rejection
was a log line. There was no cap value that fixed it, any byte cap large
enough for a full row-bounded buffer is absurd, and any row bound small enough
to fit the cap throttles narrow-row ingest.

The WAL now splits an oversized payload into multiple entries at msgpack
element boundaries, so the same request's data reaches the WAL as N
size-valid entries:

• row-format payloads (a top-level msgpack array of records) split between
  records, each chunk a shorter array of the same records
• columnar payloads (the zero-copy path's top-level map with a columns
  sub-map) split by row range; m carries over and each column array is
  sliced to the same [lo, hi) rows

Chunks are self-contained msgpack values, so the reader replays them with no
format change: concatenating the chunks' decoded rows reproduces the original
request. A payload that still cannot be split (a single record or single row
larger than the cap) keeps the loud ErrPayloadTooLarge rejection, and that
rejection is now counted in the arc_wal_oversized_payloads_total metric
instead of living only in a log line.

Fixes #677

Why

The WAL is the durability window between ingest accepting a write and the
Parquet flush landing. An operator watching a bulk load of wide rows believed
that window was protected; it was not, and nothing on the metrics endpoint
said so. Chunking removes the mismatch at the boundary that owns it (the WAL
entry size), which resolves the issue for both proposed fix shapes without a
config change and without throttling narrow-row ingest.

Verification

• before, on current main: a 117.2 MiB columnar msgpack request fed through
  the real decode and buffer write path with a real WAL writer attached is
  accepted by ingest while the WAL directory holds a single 7-byte
  header-only file, and the rejection appears only as a WAL write failed - data may be lost on crash log line
• after: the same request produces multiple WAL entries each under the cap;
  reading the file back with the WAL reader returns entries whose decoded row
  union equals the request, and replaying them through the recovery path
  (WriteColumnarDirectNoWAL) applies every row
• new regression tests fail on the pre-fix tree and pass after:
  go test ./internal/wal/ -run 'ChunksOversized|SingleOversized|PayloadAtLimitSingleEntry'
  and go test ./internal/ingest/ -run 'TestWideRequest'
• full focused suites green: go test ./internal/wal/ ./internal/metrics/ ./internal/ingest/ ./internal/cluster/... -count=1, plus the new tests
  under -race; gofmt -l empty and go vet clean on the touched packages

Notes

• chunk target is 32MB, sized so every chunk clears the 100MB single-entry
  cap after container headers, the WAL envelope, and the JSON + base64
  envelope replication wraps around a payload
• replication streams each chunk as its own ReplicateEntry under the
  protocol's own 100MB message cap

[xe-nvdk]

Thorough work on a delicate path, and the analysis in the description matches what I measured. Verified locally on the branch:

• the #677 repro behaves exactly as described: on current main, the wide columnar request leaves a 7-byte header-only WAL file; on this branch it lands as chunked entries whose replayed row union equals the request
• both new regression tests fail on the pre-fix tree and pass here; go test ./internal/wal/ ./internal/metrics/ ./internal/ingest/ -count=1 green, the new tests green under -race, go vet and gofmt -l clean
• type fidelity through the columnar re-encode holds in an adversarial probe (int64, max uint64, float32, NaN, nil, bool, bin, msgpack ext timestamps), row alignment across columns preserved, every chunk under the cap
• envelope-marked payloads (0x01, the replication receiver path) fall through the splitter untouched and keep the loud rejection, and RawMessage passthrough in the msgpack fork is verbatim, so the interleaved encoder writes in encodeColumnarRange are sound

Three changes requested:

1. encodeColumnarRange panics on ragged columns. A columnar payload whose arrays differ in length crashes with slice bounds out of range at arr[lo:hi]: the row count is taken from one map entry and every other column is sliced to it. Today's ingest decoders validate lengths before the WAL sees the payload, so this is unreachable from client input, and that is exactly why it needs the guard here: the splitter runs on the request path inside the WAL layer, and its safety currently depends on a validation invariant that lives two packages away. Validate the decoded column lengths in splitOversizedPayload and turn a mismatch into the loud rejection path.

2. The columnar split's memory cost. Decoding every column to []interface{} and re-encoding boxes each cell: I measured a 154MB narrow-row columnar payload (two columns, 9M rows) costing 4.0GB of total allocations with ~900MB held during the split, a ~26x amplification on the ingest request path. Bulk loads are the workload that hits this code, and several concurrent oversized requests would make this a real OOM risk. The array path already avoids this by carrying raw element bytes; the columnar path can do the same: walk each column's elements with DecodeRaw to record boundaries, then emit each chunk's column arrays by writing a new array header plus the raw byte range. That also makes chunks byte-faithful instead of re-encoded.

3. The release notes credit line landed at the end of the file, under the Iceberg database-drop entry, which is not yours. Move it into the #677 entry. The branch also predates today's merges of #673 and #675, which added entries at the same spot in the bug fixes section, so this file will need a rebase either way.

One smaller ask, fold into whichever change touches it: a splitOversizedPayload failure (msgpack parse error now, the ragged rejection after item 1) returns an error that bumps neither arc_wal_oversized_payloads_total nor arc_wal_failed_writes_total; only the per-buffer total_wal_errors counter sees it. Since the point of the metric is that a WAL recording nothing must not look healthy, every oversized-payload rejection path should count in arc_wal_oversized_payloads_total.

Two observations, no action needed in this PR: a chunk can legally reach ~100MB when a single element lands on a 32MB boundary, and anything above roughly 75MB then fails replication's WriteMessage after the JSON plus base64 envelope; that exposure predates this PR (any unsplit 75-100MB entry has it) so I would rather track it as a follow-up issue than grow this one. And a crash mid-sequence replays a prefix of the request's chunks, which is fine since rows are independent, but worth a sentence in the splitOversizedPayload doc comment.

[atirna]

thx for the review.

pushed the rework:

• ragged column arrays reject before slicing, and every oversized rejection increments arc_wal_oversized_payloads_total; arc_wal_failed_writes_total stays tied to physical WAL write failures.
• column chunks now retain raw msgpack element bytes rather than decoding values to []interface{}.
• moved the fix(wal): chunk oversized payloads so wide-row ingest stays durable (#677) #696 credit into the WAL payload byte cap is irreconcilable with the row-bounded ingest buffer on wide rows: WAL silently records nothing #677 release-note entry and merged current main.

Focused WAL, ingest, metrics, and cluster checks, the WAL race test, vet, and gofmt are green.

[xe-nvdk]

Re-reviewed the rework. Three of the four asks are confirmed fixed, verified locally on the branch:

• ragged columns now reject deterministically with the column named in the error (column "b" has 10 values, want 3000 in my probe), no panic, and the rejection bumps arc_wal_oversized_payloads_total; your regression test covers it
• every oversized rejection path, parse errors included, now routes through oversizedPayloadError, so the metric holds the #677 "never silently healthy" property
• credit line sits in the #677 entry and the rebase over today's merges is clean; the wal, metrics, and full ingest suites pass, the chunking and wide-request tests pass under -race, vet and gofmt clean; my type-fidelity probe still passes and chunks are now byte-faithful as a bonus

The memory ask is the one that did not land. Re-measuring the same 154MB narrow-row columnar payload (two columns, 9M rows): total allocations went from 4.0GB to 3.8GB, and the memory held during the split went up, from ~0.9GB to ~1.5GB. The cause is that DecodeRaw copies each element into its own allocation (d.rec = make(...) per call), so the per-element boxing was traded for 18M small copies plus the []msgpack.RawMessage slice headers.

The shape that actually removes the cost: record offsets instead of bytes. The decoder reads a bytes.Reader unbuffered (it satisfies the fork's bufReader interface), so after any Skip() the absolute position is exactly len(raw) - reader.Len(). Per column: DecodeArrayLen, note the position as the first boundary, then Skip() through rowsPerChunk elements at a time, noting the position at each chunk boundary — that is O(chunks) bookkeeping, not O(rows). Emitting a chunk is then an EncodeArrayLen(hi-lo) followed by one Write of the original payload's byte range per column. Held memory drops to roughly the produced chunks, and the per-element allocations disappear entirely. The ragged check falls out for free by comparing each column's element count during the same walk.

Everything else is done; this last item is worth one more pass because the splitter runs on the ingest request path and the bulk loads that trigger it are exactly the workloads that run several oversized requests at once.

[atirna]

pushed the offset-based column split. It now records only per-chunk offsets with DecodeArrayLen and Skip, then writes the original element spans, so the splitter no longer allocates one RawMessage per cell.

[xe-nvdk]

Approved. The offset-based split closes the last item, verified locally on c714a85:

• re-measuring the same 154MB narrow-row columnar payload (two columns, 9M rows): total allocations fell from 3.8GB to 386MB and held memory from ~1.5GB to ~322MB, which is the output chunks plus buffer growth and nothing else; the split also runs about twice as fast
• the position bookkeeping is sound: bytes.Reader satisfies the msgpack fork's unbuffered reader interface, so len(payload) - reader.Len() after Skip() is an exact absolute offset, and the O(chunks) boundary arrays are the only per-column state
• the ragged check moved into the same walk and still rejects cleanly with the column named, bumping arc_wal_oversized_payloads_total; the nil-column guard on DecodeArrayLen returning -1 is a nice touch
• chunks are byte-verbatim element spans now, and my fidelity probe (int64, max uint64, float32, NaN, nil, bool, bin, ext timestamps) plus row-alignment and cap checks all pass
• full wal, metrics, and ingest suites green; the chunking, ragged, and wide-request tests green under -race; vet and gofmt clean; CI green

Thanks for the careful iteration on this one. The WAL is better for it.