commitment: fix warmuper arena data race in HashSort (erigontech#21432)

## Problem

`HashSort` streams each batch's hashed/plain keys into a reused
`byteArena` bump buffer and hands every key to the async warmuper as a
sub-slice for MDBX prefetch. At each 10k batch boundary it reset that
single arena while warmup workers were still reading earlier keys — the
next batch's `arenaAlloc` overwrote bytes a worker was mid-read on. Data
race.

Latent on main: `HexToCompact` tolerates the garbage (at worst a wasted
prefetch). On nibblesv2 (erigontech#21146) `EncodeKeyV2` validates nibbles and
panics on the corrupted byte: `panic: nibbles v2: nibble at index 68 is
0xff`, mainnet ~blk 24.83M, mid commitment.

## Fix

Replace the single arena with a 2-slot ring (`arenaRingSize`) keyed by a
generation counter. Each warmed key is tagged with the current `gen`;
the warmuper keeps a per-slot in-flight count (`outstanding[gen %
ringSize]`). Before a batch boundary reuses a slot, the producer calls
`WaitBufferFree(slot)`, which blocks until that slot's
previous-generation warm items have drained — so no worker still
references the bytes about to be overwritten. Workers decrement on
completion and broadcast on drain-to-zero; a waker goroutine releases
any waiter on ctx cancellation.

Zero-copy, no per-key allocation: the arena is pre-sized once per batch
(`arenaEnsureCap`), and an over-capacity key falls back to an
independent allocation rather than reallocating the buffer (which would
invalidate live sub-slices). Wired at both `HashSort` batch boundaries
for `ModeDirect` and `ModeUpdate`; the `nil`-warmuper path is unchanged.

## Tests

- `TestHashSort_WarmupArenaNoRace` — `-race` repro; DATA RACE in
`HexToCompact` on the old single-arena wiring, green after. Covers
`ModeDirect` and `ModeUpdate`.
- `WaitBufferFree` behaviour: blocks until a straggler drains, fast-path
when the slot is already empty, unblocks on ctx cancel.
- Slot-reuse invariant `curArena == gen % arenaRingSize` survives a
cancel landing inside a boundary wait; `arenaAlloc` returns
non-overlapping sub-slices and falls back cleanly on over-capacity.

`make lint` clean, `make erigon integration` builds, commitment package
green under `-race`.

---------

Co-authored-by: Alex Sharov <AskAlexSharov@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: 3 people

execution/commitment/commitment.go

@@ -1457,36 +1457,46 @@ type Updates struct {
 	nibbles       [16]*etl.Collector
 
 	batchSlab []KeyUpdate // grow-only slab for HashSort batch (avoids per-key heap allocs)
-	byteArena []byte      // grow-only byte arena for HashSort key copies
+
+	// Ring of byte arenas for HashSort key copies; a slot is reused only after its prior generation's warm items drain.
+	arenas   [arenaRingSize][]byte
+	curArena int
+	gen      uint64
 }
 
+// arenaRingSize is how many byte arenas HashSort cycles; raising it only adds memory headroom, never affects correctness.
+const arenaRingSize = 2
+
 // arenaAlloc appends b to the byte arena and returns the sub-slice.
 // The returned slice is valid until the arena is reset.
 // The arena must have sufficient capacity (via arenaEnsureCap) before
 // accumulating a batch; if capacity is exceeded, arenaAlloc falls back
 // to an independent heap allocation to keep previously returned
 // sub-slices valid.
 func (t *Updates) arenaAlloc(b []byte) []byte {
-	off := len(t.byteArena)
+	arena := t.arenas[t.curArena]
+	off := len(arena)
 	needed := off + len(b)
-	if needed > cap(t.byteArena) {
+	if needed > cap(arena) {
 		// Arena capacity exceeded — fall back to an independent allocation.
 		// This keeps previously returned sub-slices valid while avoiding a
 		// panic that would crash a production node.
 		result := make([]byte, len(b))
 		copy(result, b)
 		return result
 	}
-	t.byteArena = t.byteArena[:needed]
-	copy(t.byteArena[off:], b)
-	return t.byteArena[off:needed]
+	arena = arena[:needed]
+	copy(arena[off:], b)
+	t.arenas[t.curArena] = arena
+	return arena[off:needed]
 }
 
-// arenaEnsureCap ensures the byte arena has at least cap bytes of capacity.
-// Must be called before each batch to prevent mid-batch reallocation.
+// arenaEnsureCap reserves at least c bytes in every ring buffer; call before a batch so a mid-batch grow can't reallocate and invalidate returned sub-slices.
 func (t *Updates) arenaEnsureCap(c int) {
-	if cap(t.byteArena) < c {
-		t.byteArena = make([]byte, 0, c)
+	for i := range t.arenas {
+		if cap(t.arenas[i]) < c {
+			t.arenas[i] = make([]byte, 0, c)
+		}
 	}
 }
 
@@ -1811,12 +1821,14 @@ func (t *Updates) HashSort(ctx context.Context, warmuper *Warmuper, fn func(hk,
 		clear(t.keys)
 
 		t.batchSlab = t.batchSlab[:0]
-		// Pre-allocate arena to avoid mid-batch reallocation that would
-		// invalidate previously returned sub-slices (hk/pk in batchSlab).
-		// Worst case: storage keys produce 128-byte nibblized hashed keys +
-		// 52-byte plain keys = 180 bytes/key. Use 192 with headroom.
+		if warmuper != nil {
+			if err := warmuper.WaitBufferFree(t.curArena); err != nil {
+				return err
+			}
+		}
+		// Pre-size the arena so a mid-batch grow can't reallocate and invalidate live sub-slices (≤180 B/key, 192 with headroom).
 		t.arenaEnsureCap(hashSortBatchSize * 192)
-		t.byteArena = t.byteArena[:0]
+		t.arenas[t.curArena] = t.arenas[t.curArena][:0]
 		var prevKey []byte
 
 		err := t.etl.Load(nil, "", func(k, v []byte, table etl.CurrentTableReader, next etl.LoadNextFunc) error {
@@ -1838,7 +1850,7 @@ func (t *Updates) HashSort(ctx context.Context, warmuper *Warmuper, fn func(hk,
 						startDepth++
 					}
 				}
-				warmuper.WarmKey(hk, startDepth)
+				warmuper.WarmKey(hk, startDepth, t.gen)
 				prevKey = append(prevKey[:0], hk...)
 			}
 
@@ -1854,11 +1866,17 @@ func (t *Updates) HashSort(ctx context.Context, warmuper *Warmuper, fn func(hk,
 						return err
 					}
 				}
+				t.batchSlab = t.batchSlab[:0]
+				nextGen := t.gen + 1
+				slot := int(nextGen % arenaRingSize)
 				if warmuper != nil {
-					warmuper.DrainPending()
+					if err := warmuper.WaitBufferFree(slot); err != nil {
+						return err
+					}
 				}
-				t.batchSlab = t.batchSlab[:0]
-				t.byteArena = t.byteArena[:0]
+				t.gen = nextGen
+				t.arenas[slot] = t.arenas[slot][:0]
+				t.curArena = slot
 			}
 			return nil
 		}, etl.TransformArgs{Quit: ctx.Done()})
@@ -1882,8 +1900,13 @@ func (t *Updates) HashSort(ctx context.Context, warmuper *Warmuper, fn func(hk,
 
 	case ModeUpdate:
 		t.batchSlab = t.batchSlab[:0]
+		if warmuper != nil {
+			if err := warmuper.WaitBufferFree(t.curArena); err != nil {
+				return err
+			}
+		}
 		t.arenaEnsureCap(hashSortBatchSize * 144)
-		t.byteArena = t.byteArena[:0]
+		t.arenas[t.curArena] = t.arenas[t.curArena][:0]
 		var prevKey []byte
 		var processErr error
 
@@ -1906,7 +1929,7 @@ func (t *Updates) HashSort(ctx context.Context, warmuper *Warmuper, fn func(hk,
 						startDepth++
 					}
 				}
-				warmuper.WarmKey(hk, startDepth)
+				warmuper.WarmKey(hk, startDepth, t.gen)
 				prevKey = append(prevKey[:0], hk...)
 			}
 
@@ -1923,11 +1946,18 @@ func (t *Updates) HashSort(ctx context.Context, warmuper *Warmuper, fn func(hk,
 						return false
 					}
 				}
+				t.batchSlab = t.batchSlab[:0]
+				nextGen := t.gen + 1
+				slot := int(nextGen % arenaRingSize)
 				if warmuper != nil {
-					warmuper.DrainPending()
+					if err := warmuper.WaitBufferFree(slot); err != nil {
+						processErr = err
+						return false
+					}
 				}
-				t.batchSlab = t.batchSlab[:0]
-				t.byteArena = t.byteArena[:0]
+				t.gen = nextGen
+				t.arenas[slot] = t.arenas[slot][:0]
+				t.curArena = slot
 			}
 			return true
 		})
@@ -1971,7 +2001,11 @@ func (t *Updates) Reset() {
 	default:
 	}
 	t.batchSlab = t.batchSlab[:0]
-	t.byteArena = t.byteArena[:0]
+	for i := range t.arenas {
+		t.arenas[i] = t.arenas[i][:0]
+	}
+	t.curArena = 0
+	t.gen = 0
 }
 
 type KeyUpdate struct {