[wip] commitment: reuse pooled trie workers across Process calls

execution/commitment/parallel_mount.go

@@ -127,7 +127,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up
 		child := root.children[childIdx]
 		ni, ch := nib, child
 		g.Go(func() error {
-			w := p.workerPool.Get().(*HexPatriciaHashed)
+			w := NewHexPatriciaHashed(p.accountKeyLen, nil, p.cfg)
 			w.mountTo(base, ni)
 			if p.template != nil && p.template.traceW != nil {
 				w.traceW = tracePrefix(p.template.traceW, fmt.Sprintf("[mnt %x] ", ni))
@@ -154,8 +154,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up
 				return foldStorageRoot(gctx, foldSem, p.newStorageWorker, pu, n, pth, accountFresh)
 			})
 			if buildErr != nil {
-				w.resetForReuse()
-				p.workerPool.Put(w)
+				w.Release()
 				return fmt.Errorf("mount[%x] build: %w", ni, buildErr)
 			}
 			var tf time.Time
@@ -168,17 +167,15 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up
 				foldDur[ni] = time.Since(tf)
 			}
 			if err != nil {
-				w.resetForReuse()
-				p.workerPool.Put(w)
+				w.Release()
 				return fmt.Errorf("mount[%x] fold: %w", ni, err)
 			}
 			cells[ni] = c
 			present[ni] = true
 			if deferred := w.TakeDeferredUpdates(); len(deferred) > 0 {
 				pu.appendDeferred(deferred)
 			}
-			w.resetForReuse()
-			p.workerPool.Put(w)
+			w.Release()
 			return nil
 		})
 		childIdx++
@@ -252,7 +249,7 @@ func (p *ParallelPatriciaHashed) newStorageWorker(ctx context.Context) (*HexPatr
 	if p.template != nil {
 		traceW = p.template.traceW
 	}
-	return newDeferredStorageWorker(ctx, &p.workerPool, p.trieCtxFactory, traceW)
+	return newDeferredStorageWorker(ctx, p.accountKeyLen, p.cfg, p.trieCtxFactory, traceW)
 }
 
 // setAccountStorageRoot writes the folded storage-root cell sr onto the account leaf.

execution/commitment/parallel_patricia_hashed.go

@@ -31,10 +31,9 @@ import (
 type ParallelPatriciaHashed struct {
 	template       *HexPatriciaHashed
 	trieCtxFactory TrieContextFactory
-	workerPool     sync.Pool
-	cfg            TrieConfig
 
 	accountKeyLen int16
+	cfg           TrieConfig
 	numWorkers    int
 
 	rootHash atomic.Pointer[[]byte]
@@ -51,23 +50,12 @@ func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int1
 		template:       NewHexPatriciaHashed(accountKeyLen, nil, cfg),
 		trieCtxFactory: ctxFactory,
 		accountKeyLen:  accountKeyLen,
-		cfg:            cfg,
 		numWorkers:     runtime.NumCPU(),
+		cfg:            cfg,
 	}
-	p.resetPool()
 	return p
 }
 
-func (p *ParallelPatriciaHashed) resetPool() {
-	akl := p.accountKeyLen
-	cfg := p.cfg
-	p.workerPool = sync.Pool{
-		New: func() any {
-			return NewHexPatriciaHashed(akl, nil, cfg)
-		},
-	}
-}
-
 // SetNumWorkers overrides the worker count for the next Process call; n <= 0 falls back to runtime.NumCPU.
 func (p *ParallelPatriciaHashed) SetNumWorkers(n int) {
 	if n <= 0 {
@@ -110,13 +98,13 @@ func (p *ParallelPatriciaHashed) RootTrie() *HexPatriciaHashed {
 	return p.template
 }
 
-// Reset clears the published root hash, drops pooled workers, and resets the template so the instance can be reused.
+// Reset clears the published root hash and resets the template so the instance
+// can be reused; pooled workers stay cached for the next Process call.
 func (p *ParallelPatriciaHashed) Reset() {
 	if p.template != nil {
 		p.template.Reset()
 	}
 	p.rootHash.Store(nil)
-	p.resetPool()
 	if p.streaming != nil {
 		p.streaming.Reset()
 	}
@@ -129,7 +117,6 @@ func (p *ParallelPatriciaHashed) Release() {
 		p.template = nil
 	}
 	p.rootHash.Store(nil)
-	p.resetPool()
 	if p.streaming != nil {
 		p.streaming.Release()
 		p.streaming = nil

execution/commitment/parallel_patricia_hashed_test.go

@@ -126,6 +126,24 @@ func TestParallelPatriciaHashedSkeletonReset(t *testing.T) {
 	require.NotNil(t, p.template, "Reset preserves the template")
 }
 
+// Every checkout must be config-correct whether it hit the shared pool or
+// constructed fresh — that fungibility is what lets workers cross instances.
+func TestWorkerCheckoutAppliesConfig(t *testing.T) {
+	cfg := DefaultTrieConfig()
+	cfg.MemoizationOff = true
+
+	stale := NewHexPatriciaHashed(2*length.Addr, nil, DefaultTrieConfig())
+	stale.Release()
+
+	for range 2 {
+		w := NewHexPatriciaHashed(length.Addr, nil, cfg)
+		assert.Equal(t, int16(length.Addr), w.accountKeyLen)
+		assert.True(t, w.memoizationOff)
+		assert.Equal(t, cfg, w.cfg)
+		w.Release()
+	}
+}
+
 func TestParallelPatriciaHashedSkeletonRelease(t *testing.T) {
 	p := NewParallelPatriciaHashed(nil, length.Addr, DefaultTrieConfig())
 	require.NotNil(t, p.template)
@@ -872,6 +890,53 @@ func requireIncrementalEquiv(t *testing.T, k1 [][]byte, u1 []Update, k2 [][]byte
 	requireAllEnginesParity(t, k1, u1, k2, u2, workers)
 }
 
+// Folds two batches on ONE trie instance with the per-block Reset + SetState
+// lifecycle in between, so batch-2 runs on workers cached by batch-1.
+func reusedInstanceIncrementalRoot(t *testing.T, variant TrieVariant, workers int, k1 [][]byte, u1 []Update, k2 [][]byte, u2 []Update) []byte {
+	t.Helper()
+	ctx := context.Background()
+	ms := NewMockState(t)
+	ms.SetConcurrentCommitment(true)
+
+	cfg := DefaultTrieConfig()
+	cfg.Variant = variant
+	trie, ut := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
+	defer ut.Close()
+	defer trie.Release()
+	pt := trie.(*ParallelPatriciaHashed)
+	pt.SetNumWorkers(workers)
+	pt.SetTrieContextFactory(mockTrieCtxFactory(ms))
+	pt.ResetContext(ms)
+
+	process := func(keys [][]byte, upds []Update) ([]byte, []byte) {
+		require.NoError(t, ms.applyPlainUpdates(keys, upds))
+		WrapKeyUpdatesInto(t, ut, keys, upds)
+		root, err := pt.Process(ctx, ut, "", nil, WarmupConfig{})
+		require.NoError(t, err)
+		blob, err := pt.RootTrie().EncodeCurrentState(nil)
+		require.NoError(t, err)
+		return bytes.Clone(root), blob
+	}
+
+	_, blob := process(k1, u1)
+	pt.Reset()
+	require.NoError(t, pt.RootTrie().SetState(blob))
+	root2, _ := process(k2, u2)
+	return root2
+}
+
+func TestParallelReuseAcrossResetParity(t *testing.T) {
+	t.Parallel()
+	k1, u1 := genRandomAccountsStorage(256)
+	k2, u2 := sparseBatch2(k1, 3, false)
+	seqRoot, _ := incrementalRoot(t, modeSeq, 0, k1, u1, k2, u2)
+
+	for _, variant := range []TrieVariant{VariantParallelHexPatricia, VariantStreamingHexPatricia} {
+		root := reusedInstanceIncrementalRoot(t, variant, 8, k1, u1, k2, u2)
+		require.Equalf(t, seqRoot, root, "reused-instance %s incremental root vs sequential", variant)
+	}
+}
+
 func TestVerifyParallel_WideNested(t *testing.T) {
 	t.Parallel()
 	keys, upds := genWideNested(t)

execution/commitment/streaming_commitment.go

@@ -69,7 +69,6 @@ type StreamingCommitter struct {
 	accountKeyLen  int16
 	numWorkers     int
 
-	workerPool sync.Pool
 	trie       *prefixTrie
 	splits     map[byte]*splitState
 	eagerFloor uint64
@@ -123,18 +122,9 @@ func NewStreamingCommitter(ctxFactory TrieContextFactory, accountKeyLen int16, c
 		splits:         make(map[byte]*splitState),
 		eagerFloor:     defaultEagerFold,
 	}
-	sc.resetPool()
 	return sc
 }
 
-func (sc *StreamingCommitter) resetPool() {
-	akl := sc.accountKeyLen
-	cfg := sc.cfg
-	sc.workerPool = sync.Pool{
-		New: func() any { return NewHexPatriciaHashed(akl, nil, cfg) },
-	}
-}
-
 // SetNumWorkers overrides the worker count for the next Process call. Values
 // <= 0 fall back to runtime.NumCPU.
 func (sc *StreamingCommitter) SetNumWorkers(n int) {
@@ -566,7 +556,7 @@ func (sc *StreamingCommitter) markQueued(s *splitState, nib byte) {
 // foldKeys folds a snapshotted split's keys on a pooled worker whose overlay ctx
 // discards branch writes; the returned flushed flag reports a mid-fold self-flush.
 func (sc *StreamingCommitter) foldKeys(nib byte, keys []touchedKey) (cell, []*DeferredBranchUpdate, bool, error) {
-	w := sc.workerPool.Get().(*HexPatriciaHashed)
+	w := NewHexPatriciaHashed(sc.accountKeyLen, nil, sc.cfg)
 	w.mountTo(sc.base, int(nib))
 	if sc.traceW != nil {
 		w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[fold %x] ", nib)))
@@ -593,8 +583,7 @@ func (sc *StreamingCommitter) foldKeys(nib byte, keys []touchedKey) (cell, []*De
 		c, err = w.foldMounted(sc.bgCtx, int(nib))
 	}
 	deferred := w.TakeDeferredUpdates()
-	w.resetForReuse()
-	sc.workerPool.Put(w)
+	w.Release()
 	return c, deferred, ov.flushed, err
 }
 
@@ -746,7 +735,7 @@ func stitchSplitCells(base *HexPatriciaHashed, cells *[16]cell, present *[16]boo
 // deferred set.
 func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore.Weighted, base *HexPatriciaHashed, s *splitState, child *prefixNode) error {
 	ni := s.prefix[0]
-	w := sc.workerPool.Get().(*HexPatriciaHashed)
+	w := NewHexPatriciaHashed(sc.accountKeyLen, nil, sc.cfg)
 	w.mountTo(base, int(ni))
 	if sc.traceW != nil {
 		w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[split %x] ", ni)))
@@ -773,17 +762,15 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore.
 		return sr, err
 	}
 	if err := dfsSubtreeDeep(w, child, path, deepStorageRoot); err != nil {
-		w.resetForReuse()
-		sc.workerPool.Put(w)
+		w.Release()
 		for _, upd := range pu.deferredCombined {
 			putDeferredUpdate(upd)
 		}
 		return fmt.Errorf("split[%x] build: %w", ni, err)
 	}
 	c, err := w.foldMounted(ctx, int(ni))
 	if err != nil {
-		w.resetForReuse()
-		sc.workerPool.Put(w)
+		w.Release()
 		for _, upd := range pu.deferredCombined {
 			putDeferredUpdate(upd)
 		}
@@ -794,8 +781,7 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore.
 	if d := w.TakeDeferredUpdates(); len(d) > 0 {
 		newDeferred = append(newDeferred, d...)
 	}
-	w.resetForReuse()
-	sc.workerPool.Put(w)
+	w.Release()
 
 	s.mu.Lock()
 	for _, upd := range s.deferred {
@@ -814,7 +800,7 @@ func (sc *StreamingCommitter) DeepLocalFolds() uint64 { return sc.deepLocalFolds
 // newStorageWorker sources a concurrent-storage-fold worker; disjoint subtree
 // prefixes keep a mid-fold self-flush from racing another fold's writes.
 func (sc *StreamingCommitter) newStorageWorker(ctx context.Context) (*HexPatriciaHashed, func()) {
-	return newDeferredStorageWorker(ctx, &sc.workerPool, sc.trieCtxFactory, sc.traceW)
+	return newDeferredStorageWorker(ctx, sc.accountKeyLen, sc.cfg, sc.trieCtxFactory, sc.traceW)
 }
 
 // dropSplitDeferred returns every split's staged deferred branch updates to the pool.
@@ -942,7 +928,6 @@ func (sc *StreamingCommitter) Reset() {
 	}
 	sc.deferredForCaller = nil
 	sc.rootValid, sc.rootSeeded = false, false
-	sc.resetPool()
 }
 
 // releaseBase drops the scheduler's persistent base and its context.
@@ -967,5 +952,4 @@ func (sc *StreamingCommitter) Release() {
 	}
 	sc.deferredForCaller = nil
 	sc.rootValid, sc.rootSeeded = false, false
-	sc.resetPool()
 }

execution/commitment/streaming_deep_fold.go

@@ -23,7 +23,6 @@ import (
 	"io"
 	"math/bits"
 	"runtime"
-	"sync"
 
 	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/semaphore"
@@ -289,16 +288,15 @@ func storageRootFromSingleChild(base *HexPatriciaHashed) (cell, error) {
 
 // newDeferredStorageWorker yields a pooled trie worker for a deferring storage fold
 // and a release that returns it to the pool and frees its context.
-func newDeferredStorageWorker(ctx context.Context, pool *sync.Pool, factory TrieContextFactory, traceW io.Writer) (*HexPatriciaHashed, func()) {
-	w := pool.Get().(*HexPatriciaHashed)
+func newDeferredStorageWorker(ctx context.Context, accountKeyLen int16, cfg TrieConfig, factory TrieContextFactory, traceW io.Writer) (*HexPatriciaHashed, func()) {
+	w := NewHexPatriciaHashed(accountKeyLen, nil, cfg)
 	wctx, cleanup := factory(ctx)
 	w.ResetContext(wctx)
 	w.SetTraceWriter(traceW)
 	w.branchEncoder.setDeferUpdates(true)
 	w.SetLeaveDeferredForCaller(true)
 	return w, func() {
-		w.resetForReuse()
-		pool.Put(w)
+		w.Release()
 		if cleanup != nil {
 			cleanup()
 		}