The decoder goroutine in loadWAL inlines a ~95-line switch that
allocates a per-record-type pooled slice, dispatches to the right
record.Decoder method, and wraps any decoder error in a
*wlog.CorruptionErr with the current segment/offset. Pull that body
out into a new method:

  (*Head).decodeWALRecord(rec []byte, typ record.Type,
                          dec *record.Decoder,
                          segment int, offset int64) (any, error)

It returns:
  - (decoded slice, nil) for handled record types,
  - (nil, nil)           for types loadWAL does not act on,
  - (nil, *wlog.CorruptionErr) on decode failure (segment/offset are
                          embedded in the error so callers do not
                          need a back-channel to the reader).

The function reads only from its arguments and writes only to the
goroutine-safe sync.Pools on Head, so it is safe to call from
multiple goroutines concurrently. That property is not exercised
here — the existing single-goroutine pipeline is preserved verbatim
— but it is the shape required by the follow-up commits that
introduce a decoder pool.

No behaviour change.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>

BenchmarkLoadWLs currently runs every shape against a WAL written
with compression.None. Real Prometheus deployments default to snappy,
where r.Next() does the decompression on the producer-side hot path
inside the WAL decoder goroutine. Replay benchmarks that skip
compression therefore understate the producer-side CPU cost.

Add a `compress` field to the case struct (zero value treated as
compression.None so the existing cases are unchanged) and one new
case sized for the regime parallel-WAL-decode work targets:
batches=200, seriesPerBatch=2500, samplesPerSeries=240 — 500k series
x 240 samples = ~120M samples, written with compression.Snappy.

Subtest names now include `compress=<type>` so benchstat can
distinguish the new shape from the existing baseline.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>

Add two new fields to HeadOptions:

  - ParallelWALDecode (bool, default false) — represents the
    'parallel-wal-decode' feature flag.
  - ParallelWALDecodeConcurrency (int, default
    min(max(GOMAXPROCS/4, 2), 4) when ParallelWALDecode is enabled)
    — the size of the decoder pool.

Normalise the concurrency default in NewHead alongside the existing
WALReplayConcurrency handling, so a positive explicit value is
preserved and zero/negative gets clamped.

Wire a flag-gated entry point at the top of loadWAL:

  if h.opts.ParallelWALDecode {
      return h.loadWALParallel(...)
  }

For now loadWALParallel is a stub that returns an error: the byte
reader, decoder pool, and reorder buffer described in the design doc
land in the follow-up commit. Returning an explicit error keeps the
behaviour-when-flag-off path bit-identical and prevents the flag from
silently no-op'ing WAL replay if it is enabled prematurely (e.g. by a
test that gets ahead of the implementation).

The CLI wiring for --enable-feature=parallel-wal-decode arrives in a
later commit; until then the flag is only settable directly on
HeadOptions, which is sufficient for tests and benchmarks.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>

The decoder goroutine in loadWAL inlines the per-record read/decode/
push loop directly. Pull that body out into a separate method:

  (*Head).serialDecodeWALRecords(r *wlog.Reader,
                                 syms *labels.SymbolTable,
                                 decoded chan<- any) error

The function reads from r sequentially, calls the existing
decodeWALRecord for each record, and pushes typed values onto decoded
in WAL byte order. It returns when r drains or when decodeWALRecord
returns a *wlog.CorruptionErr.

It deliberately does NOT close decoded; loadWAL keeps the deferred
close on the launching goroutine. That contract keeps the close-once
responsibility local to the call site rather than spreading it across
the producer family — which matters once the parallel path adds a
producer with a different goroutine topology (a byte reader plus a
decoder pool plus a reorder buffer, where the reorder buffer is the
last writer but is not the function the caller invokes directly).

No behaviour change. loadWAL is byte-equivalent under the flag-off
path; tests covering WAL replay, corruption handling, and OOO replay
pass unchanged.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>

loadWAL has ~340 lines of worker-pool setup, exemplar processor, and
router/finalisation that the parallel decode path needs to share verbatim
— only the decoder goroutine differs between the serial and the
upcoming parallel pipeline. Pull the shared body out into a new method:

  (*Head).runWALReplay(r, syms, multiRef, mmappedChunks, oooMmappedChunks,
                       produce func(*wlog.Reader, *labels.SymbolTable,
                                    chan<- any) error) (err error)

The producer is launched inside runWALReplay's own goroutine, which
captures any returned error into decodeErr and closes the decoded
channel via defer. The producer therefore must not close decoded
itself — the close-once contract is centralised in runWALReplay so the
parallel path (whose internal goroutine topology differs) can plug in
without re-implementing channel lifecycle.

loadWAL is now a small dispatcher:

  if h.opts.ParallelWALDecode {
      return h.loadWALParallel(...)
  }
  return h.runWALReplay(..., h.serialDecodeWALRecords)

The 340-line body did not move — it stayed in place under the new
function name and signature — so the diff is the wrapper plus the
producer parameter plus the one-line change in the decoder goroutine.
This keeps the refactor easy to verify by inspection.

No behaviour change. loadWAL with the flag off is bit-identical to
the previous commit; WAL replay, corruption handling, and OOO replay
tests pass unchanged.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>

Replace the loadWALParallel stub with the byte-reader → decoder-pool →
reorder-buffer pipeline described in the design doc. All new code
lives in a new file, tsdb/head_wal_parallel.go, so head_wal.go does
not grow further.

Pipeline shape:

  1. Byte reader (1 goroutine) reads from wlog.Reader, captures
     type / segment / offset, copies the record bytes into a pooled
     []byte (rawRecordBufPool), and dispatches a rawRecord{seq, ...}
     on a buffered channel sized K*4. Unknown record types are
     filtered out before they consume a seq.
  2. Decoder pool (K = ParallelWALDecodeConcurrency goroutines): each
     constructs its own record.Decoder with the shared
     *labels.SymbolTable (so dedupelabels symbol state is shared
     across the pool — that is what the earlier FIXME-fix commit on
     record.NewDecoder enables), consumes rawRecords, calls
     decodeWALRecord, returns the pooled bytes, and emits a
     decodedRecord{seq, val} on decodedOOO. On decode failure the
     goroutine reports the error via walDecodeFirstErr and emits a
     nil-val sentinel so the reorder buffer can advance.
  3. Reorder buffer (runs inline in the parallelDecodeWALRecords
     function): maintains a min-heap by seq, skips sentinels, and
     emits onto the existing `decoded` channel in WAL byte order so
     the router's contract is unchanged. Once the expected-seq
     pointer reaches the seq of the first reported error, the
     buffer stops emitting and only drains — preserving the serial
     path's behaviour where records past the first corruption are
     not applied to the head.

parallelDecodeWALRecords is the producer passed to runWALReplay; the
worker pool, exemplar processor, and router are shared verbatim with
the serial path. The reorder buffer does NOT close `decoded` —
runWALReplay's launching goroutine does that on return via defer.

walDecodeFirstErr is a tiny mutex-guarded min-seq tracker. The serial
path's contract is "fail on the first offending record in WAL byte
order"; multiple decoders can hit corruption concurrently, so we
reduce by min seq rather than wall-clock order.

Add an end-to-end equivalence smoke test: write a WAL containing
series, samples, a tombstone, and an exemplar, replay it twice (flag
off, then K=4), and assert NumSeries / MinTime / MaxTime match
between the two runs. Race-clean under -race. More granular unit
tests for the reorder buffer, first-error scenarios, and back-
pressure interactions land in follow-up commits.

The CLI flag (--enable-feature=parallel-wal-decode) wires up in a
later commit; for now the feature is exercisable only via
HeadOptions, which is sufficient for tests and the benchmark suite.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>

The end-to-end smoke test in the previous commit verifies the
parallel pipeline produces equivalent head state to the serial path
on a small mixed WAL. This commit fills out the test coverage on the
load-bearing internals:

Reorder buffer:
  - Emits records in seq order regardless of arrival order.
  - Skips nil-val sentinels without breaking ordering of surrounding
    records.
  - Buffers out-of-order arrivals in the heap and emits contiguously
    once the low-seq lands.
  - Stops emitting once the expected-seq pointer reaches the seq of
    the first reported error (preserving the serial path's contract
    that records past the first corruption are not applied).

walDecodeFirstErr:
  - Retains the lowest-seq error across multiple concurrent reports.
  - Reports (0, nil) when no error has been recorded.

Extended end-to-end equivalence:
  - Replays a WAL containing histograms, float histograms, and
    metadata in addition to series and samples; asserts NumSeries /
    MinTime / MaxTime match between serial and parallel runs.

All tests pass under -race.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>

…rency

To measure the parallel WAL decode delta with benchstat, both
BenchmarkLoadWLs and BenchmarkLoadRealWLs need to iterate over the K
values we care about. Add that, with two compatibility concerns
addressed up front:

  - BenchmarkLoadWLs: the K dimension is opt-in per shape via a new
    parallelDecodeKs []int field on the case struct. Shapes that
    don't set it (every existing case) keep their old subtest names
    and only run the serial path, so older baseline.txt files
    captured before this commit remain comparable with benchstat.
    The new snappy stress shape opts in with K ∈ {0, 2, 4, 8}.
  - BenchmarkLoadRealWLs: subtest names now include
    parallel_decode=K. This does break compatibility with any
    pre-existing real-WAL baseline files, but real-WAL benches
    require BENCHMARK_LOAD_REAL_WLS_DIR so the population of
    affected baselines is small, and getting the K comparison is
    exactly the point.

The K iteration runs each K as its own b.Run subtest with its own
b.Loop, so b.Loop's timing model is preserved per K. The WAL is
written once per outer subtest and re-read for every K.

```release-notes
NONE
```

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Signed-off-by: Nicolas Takashi <[email protected]>