commitment: build the ModeParallel prefix trie off the execution thread

execution/commitment/commitment.go

@@ -1380,6 +1380,9 @@
 }
 
 func (t *Updates) SetMode(m Mode) {
+	if t.parallel != nil {
+		t.parallel.Reset()
+	}
 	t.mode = m
 	switch t.mode {
 	case ModeDirect:
@@ -1498,7 +1501,7 @@
 		hashedKey := t.hashKey(keyBytes)
 		ik := t.parallel.internKey(keyBytes)
 		t.keys[key] = struct{}{}
-		t.parallel.Insert(hashedKey, ik, nil)
+		t.parallel.Collect(hashedKey, ik, nil)
 	default:
 	}
 }
@@ -1559,7 +1562,7 @@
 			ik = t.parallel.internKey(keyBytes)
 			t.keys[key] = struct{}{}
 		}
-		t.parallel.Insert(hashedKey, ik, u)
+		t.parallel.Collect(hashedKey, ik, u)
 	default:
 	}
 }
@@ -1581,7 +1584,7 @@
 		}
 		dedupKey := string(hashedKey)
 		if _, ok := t.keys[dedupKey]; !ok {
-			t.parallel.Insert(hashedKey, nil, nil)
+			t.parallel.Collect(common.ToBytesZeroCopy(dedupKey), nil, nil)
 			t.keys[dedupKey] = struct{}{}
 		}
 	case ModeUpdate:
@@ -1700,7 +1703,7 @@
 }
 
 // fn must not retain hk or pk slices after returning: they're backed by reusable arena memory.
 func (t *Updates) HashSort(ctx context.Context, warmuper *Warmuper, fn func(hk, pk []byte, update *Update) error) error {
 	// [:cap] because the loops below reslice batchSlab many times.
 	defer func() { clear(t.batchSlab[:cap(t.batchSlab)]) }()
 

execution/commitment/commitment_test.go

@@ -864,6 +864,13 @@ func TestUpdatesModeParallel_NewAllocates(t *testing.T) {
 	require.Equal(t, uint64(0), ut.Size())
 }
 
+func requireParallelCount(t *testing.T, ut *Updates, n int, msg string) {
+	t.Helper()
+	require.Equal(t, uint64(n), ut.Size())
+	ut.parallel.Build()
+	require.EqualValues(t, n, ut.parallel.trie.root.subtreeCount, msg)
+}
+
 func TestUpdatesModeParallel_TouchPlainKeyRoutes(t *testing.T) {
 	t.Parallel()
 
@@ -880,15 +887,10 @@ func TestUpdatesModeParallel_TouchPlainKeyRoutes(t *testing.T) {
 		ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage)
 	}
 
-	require.Equal(t, uint64(len(keys)), ut.Size())
-	require.NotNil(t, ut.parallel.trie.root)
-	require.EqualValues(t, len(keys), ut.parallel.trie.root.subtreeCount,
-		"every touched key must show up in the prefix trie")
+	requireParallelCount(t, ut, len(keys), "every touched key must show up in the prefix trie")
 
 	ut.TouchPlainKey(string(keys[0]), []byte("v2"), ut.TouchStorage)
-	require.Equal(t, uint64(len(keys)), ut.Size())
-	require.EqualValues(t, len(keys), ut.parallel.trie.root.subtreeCount,
-		"duplicate TouchPlainKey must not double-count in the trie")
+	requireParallelCount(t, ut, len(keys), "duplicate TouchPlainKey must not double-count in the trie")
 }
 
 func TestUpdatesModeParallel_RepeatTouchDoesNotRehash(t *testing.T) {
@@ -910,9 +912,7 @@ func TestUpdatesModeParallel_RepeatTouchDoesNotRehash(t *testing.T) {
 	}
 
 	require.EqualValues(t, 1, hashCalls.Load(), "a repeat touch must not rehash the key")
-	require.Equal(t, uint64(1), ut.Size())
-	require.NotNil(t, ut.parallel.trie.root)
-	require.EqualValues(t, 1, ut.parallel.trie.root.subtreeCount)
+	requireParallelCount(t, ut, 1, "the single touched key must reach the trie")
 }
 
 func TestUpdatesModeParallel_TouchHashedKey(t *testing.T) {
@@ -928,8 +928,7 @@ func TestUpdatesModeParallel_TouchHashedKey(t *testing.T) {
 	ut.TouchHashedKey(hk2)
 	ut.TouchHashedKey(hk1)
 
-	require.Equal(t, uint64(2), ut.Size())
-	require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount)
+	requireParallelCount(t, ut, 2, "a repeated hashed key must not double-count in the trie")
 }
 
 func TestUpdatesModeParallel_Reset(t *testing.T) {
@@ -945,8 +944,7 @@ func TestUpdatesModeParallel_Reset(t *testing.T) {
 	for _, k := range keys {
 		ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage)
 	}
-	require.Equal(t, uint64(2), ut.Size())
-	require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount)
+	requireParallelCount(t, ut, 2, "both touched keys must reach the trie")
 
 	ut.Reset()
 
@@ -960,8 +958,7 @@ func TestUpdatesModeParallel_Reset(t *testing.T) {
 	for _, k := range keys {
 		ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage)
 	}
-	require.Equal(t, uint64(2), ut.Size())
-	require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount)
+	requireParallelCount(t, ut, 2, "a collection reused after Reset must refill the trie")
 }
 
 func TestUpdatesModeParallel_Close(t *testing.T) {

execution/commitment/fork_walk_test.go

@@ -400,6 +400,7 @@ func TestForkWalk_ForkStartsHelpersOnlyOnIdleLeases(t *testing.T) {
 		}
 		bw := &walker{trie: base}
 		require.NoError(t, fw.attach(ctx, bw))
+		ut.parallel.Build()
 		root := ut.parallel.trie.root
 		err := fw.walk(ctx, bw, root, append(make([]byte, 0, forkPathCap), root.ext...))
 		fw.detach(bw)
@@ -493,6 +494,7 @@ func TestForkWalk_HelperErrorIsNotMaskedByCancellation(t *testing.T) {
 	bw := &walker{trie: base}
 	require.NoError(t, fw.attach(ctx, bw))
 	require.Same(t, first, bw.lease, "the forking walker must hold the lease whose reads wait for the helper to fail")
+	ut.parallel.Build()
 	root := ut.parallel.trie.root
 	err = fw.walk(ctx, bw, root, append(make([]byte, 0, forkPathCap), root.ext...))
 	fw.detach(bw)
@@ -573,9 +575,10 @@ func absentDeleteCorpus(freshRegion bool) (k1 [][]byte, u1 []Update, k2 [][]byte
 
 func TestForkWalk_SplitsExcludesTheNodesOwnKey(t *testing.T) {
 	pu := newParallelUpdate()
-	pu.Insert(nibs(0x01, 0x02), []byte("pk-D"), nil)
-	pu.Insert(nibs(0x01, 0x02, 0x03), []byte("pk-A"), nil)
-	pu.Insert(nibs(0x01, 0x02, 0x04), []byte("pk-B"), nil)
+	pu.Collect(nibs(0x01, 0x02), []byte("pk-D"), nil)
+	pu.Collect(nibs(0x01, 0x02, 0x03), []byte("pk-A"), nil)
+	pu.Collect(nibs(0x01, 0x02, 0x04), []byte("pk-B"), nil)
+	pu.Build()
 	node := pu.trie.root.children[0]
 	require.NotNil(t, node.plainKey)
 	require.Equal(t, uint32(3), node.subtreeCount)

execution/commitment/parallel_collect_test.go

@@ -0,0 +1,192 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package commitment
+
+import (
+	"bytes"
+	"math/rand"
+	"slices"
+	"testing"
+	"testing/synctest"
+	"unsafe"
+
+	"github.com/holiman/uint256"
+	"github.com/stretchr/testify/require"
+)
+
+func randomTouchCases(seed int64, n, dupEvery int) []touchEntry {
+	rnd := rand.New(rand.NewSource(seed))
+	cases := make([]touchEntry, 0, n)
+	for i := range n {
+		hk := make([]byte, 64)
+		for j := range hk {
+			hk[j] = byte(rnd.Intn(16))
+		}
+		var pk []byte
+		if i%7 != 0 {
+			pk = []byte{byte(i), byte(i >> 8)}
+		}
+		var upd *Update
+		if i%3 != 0 {
+			upd = &Update{Flags: BalanceUpdate, Balance: *uint256.NewInt(uint64(i) + 1)}
+		}
+		cases = append(cases, touchEntry{hashedKey: hk, plainKey: pk, update: upd})
+		if dupEvery > 0 && i%dupEvery == 0 {
+			cases = append(cases, touchEntry{
+				hashedKey: slices.Clone(hk),
+				plainKey:  pk,
+				update:    &Update{Flags: NonceUpdate, Nonce: uint64(i) + 1},
+			})
+		}
+	}
+	return cases
+}
+
+func buildReferenceTrie(cases []touchEntry) *prefixTrie {
+	tr := newPrefixTrie()
+	for _, c := range cases {
+		tr.Insert(c.hashedKey, c.plainKey, c.update)
+	}
+	return tr
+}
+
+func collectAll(pu *parallelUpdate, cases []touchEntry) {
+	for _, c := range cases {
+		pu.Collect(c.hashedKey, c.plainKey, c.update)
+	}
+}
+
+func buildChunked(cases []touchEntry) (*parallelUpdate, bool) {
+	pu := newParallelUpdate()
+	collectAll(pu, cases)
+	backgrounded := pu.spare != nil
+	pu.Build()
+	return pu, backgrounded
+}
+
+func TestChunkBuild_WalkOrderIsSortedWhateverTheArrivalOrder(t *testing.T) {
+	t.Parallel()
+
+	cases := randomTouchCases(31337, 2*touchChunkKeys+3, 0)
+	shuffled := slices.Clone(cases)
+	rnd := rand.New(rand.NewSource(9))
+	rnd.Shuffle(len(shuffled), func(i, j int) { shuffled[i], shuffled[j] = shuffled[j], shuffled[i] })
+
+	pu, backgrounded := buildChunked(shuffled)
+	require.True(t, backgrounded, "the batch must cross a chunk boundary")
+	walk := collectWalk(pu.trie)
+	require.True(t, slices.IsSortedFunc(walk, func(a, b walkEntry) int { return bytes.Compare(a.prefix, b.prefix) }),
+		"the walk must emit nodes in hashed-key order")
+
+	var terminals, want [][]byte
+	for _, e := range walk {
+		if e.plainKey != nil {
+			terminals = append(terminals, e.prefix)
+		}
+	}
+	for _, c := range cases {
+		if c.plainKey != nil {
+			want = append(want, c.hashedKey)
+		}
+	}
+	slices.SortFunc(want, bytes.Compare)
+	require.Equal(t, want, terminals, "every collected key must appear once, in sorted order")
+}
+
+func TestChunkBuild_MatchesInsertionOrderTrie(t *testing.T) {
+	t.Parallel()
+
+	for _, n := range []int{1, 2, 17, 5000, touchChunkKeys - 1, touchChunkKeys, touchChunkKeys + 1, 3*touchChunkKeys + 7} {
+		cases := randomTouchCases(int64(n)*7919+3, n, 4)[:n]
+		want := collectWalk(buildReferenceTrie(cases))
+		pu, backgrounded := buildChunked(cases)
+		require.Equal(t, n >= touchChunkKeys, backgrounded,
+			"n=%d: hand-off must trigger exactly when the batch fills a chunk", n)
+		require.Equal(t, want, collectWalk(pu.trie),
+			"chunked build must match insertion-order build for n=%d", n)
+	}
+}
+
+func TestChunkBuild_BuffersSurviveBatches(t *testing.T) {
+	t.Parallel()
+
+	cases := randomTouchCases(777, touchChunkKeys+8, 0)
+	pu := newParallelUpdate()
+	buffers := func() []uintptr {
+		return []uintptr{uintptr(unsafe.Pointer(unsafe.SliceData(pu.pending))), uintptr(unsafe.Pointer(unsafe.SliceData(pu.spare)))}
+	}
+	var first []uintptr
+	for batch := range 3 {
+		pu.Reset()
+		collectAll(pu, cases)
+		pu.Build()
+		require.EqualValues(t, len(cases), pu.trie.root.subtreeCount)
+		if batch == 1 {
+			first = buffers()
+		}
+	}
+	require.GreaterOrEqual(t, cap(pu.pending), touchChunkKeys, "a recycled buffer must keep its capacity")
+	require.GreaterOrEqual(t, cap(pu.spare), touchChunkKeys, "a recycled buffer must keep its capacity")
+	require.ElementsMatch(t, first, buffers(), "later batches must reuse the same two buffers")
+}
+
+func TestChunkBuild_KeyTouchedInSeveralChunksKeepsArrivalOrder(t *testing.T) {
+	t.Parallel()
+
+	cases := randomTouchCases(41, 2*touchChunkKeys+5, 0)
+	hk := bytes.Repeat([]byte{0x0a}, 64)
+	for i, at := range []int{touchChunkKeys - 1, touchChunkKeys, 2*touchChunkKeys + 1} {
+		cases[at] = touchEntry{hashedKey: hk, plainKey: []byte("pk"), update: &Update{Flags: BalanceUpdate, Balance: *uint256.NewInt(uint64(i) + 1)}}
+	}
+	want := collectWalk(buildReferenceTrie(cases))
+	reordered := slices.Concat(cases[touchChunkKeys:], cases[:touchChunkKeys])
+	require.NotEqual(t, want, collectWalk(buildReferenceTrie(reordered)), "the fixture must be sensitive to chunk order")
+
+	pu, backgrounded := buildChunked(cases)
+	require.True(t, backgrounded)
+	require.Equal(t, want, collectWalk(pu.trie), "a key touched in several chunks must keep its last update")
+}
+
+func TestChunkBuild_TouchHashedKeyCopiesCallerBuffer(t *testing.T) {
+	t.Parallel()
+
+	ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash)
+	defer ut.Close()
+
+	buf := make([]byte, 64)
+	for v := byte(1); v <= 2; v++ {
+		for i := range buf {
+			buf[i] = v
+		}
+		ut.TouchHashedKey(buf)
+	}
+
+	require.Equal(t, uint64(2), ut.Size())
+	ut.parallel.Build()
+	require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount,
+		"a reused caller buffer must not collapse two keys into one")
+	require.Len(t, ut.parallel.trie.root.children, 2)
+}
+
+func TestChunkBuild_AbandonedCollectionLeavesNoGoroutine(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		collectAll(newParallelUpdate(), randomTouchCases(5, touchChunkKeys, 0))
+		synctest.Wait()
+	})
+}

execution/commitment/parallel_patricia_hashed.go

@@ -253,6 +253,7 @@ func (p *ParallelPatriciaHashed) Process(
 	defer func() { observeRound(p.metrics, roundStart) }()
 
 	pu := updates.parallel
+	pu.Build()
 	if pu.trie == nil || pu.trie.root == nil || pu.trie.root.subtreeCount == 0 {
 		// A consumed (or never-touched) collection must return the carried root; folding
 		// an empty base would publish the empty-trie root instead.