Bisect: original PR commit + gemini fix + mdbx fix only

Contains only the original changes (Pool QueueWithRetry, cache
filterDirtyFiles regexp, Close→Release lifecycle) plus the mdbx
buffered channel fix. No diagnostic commits, no clearInProgress fix,
no force-schedule fallback, no test harness changes.

Purpose: isolate whether the "Wrong trie root" devnet failures are
caused by the original lifecycle change or by the fix commits.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: 2 people

db/kv/mdbx/kv_mdbx.go

@@ -819,7 +819,7 @@ func NewAsyncTx(tx kv.Tx, queueSize int) *asyncTx {
 }
 
 func (a *asyncTx) Apply(ctx context.Context, f func(kv.Tx) error) error {
-	rc := make(chan error)
+	rc := make(chan error, 1) // buffered: if caller abandons via ctx.Done(), the mdbx thread must not block
 	a.requests <- &applyTx{rc, a.Tx, f}
 	select {
 	case err := <-rc:
@@ -843,7 +843,7 @@ func NewAsyncRwTx(tx kv.RwTx, queueSize int) *asyncRwTx {
 }
 
 func (a *asyncRwTx) Apply(ctx context.Context, f func(kv.Tx) error) error {
-	rc := make(chan error)
+	rc := make(chan error, 1) // buffered: if caller abandons via ctx.Done(), the mdbx thread must not block
 	a.requests <- &applyTx{rc, a.RwTx, f}
 	select {
 	case err := <-rc:
@@ -854,7 +854,7 @@ func (a *asyncRwTx) Apply(ctx context.Context, f func(kv.Tx) error) error {
 }
 
 func (a *asyncRwTx) ApplyRw(ctx context.Context, f func(kv.RwTx) error) error {
-	rc := make(chan error)
+	rc := make(chan error, 1) // buffered: if caller abandons via ctx.Done(), the mdbx thread must not block
 	a.requests <- &applyRwTx{rc, a.RwTx, f}
 	select {
 	case err := <-rc:

db/state/dirty_files.go

@@ -258,8 +258,16 @@ func (i *FilesItem) closeFilesAndRemove() {
 	}
 }
 
+var filterDirtyFilesReCache sync.Map // pattern string → *regexp.Regexp
+
 func filterDirtyFiles(fileNames []string, stepSize, stepsInFrozenFile uint64, filenameBase, ext string, logger log.Logger) (res []*FilesItem) {
-	re := regexp.MustCompile(`^v(\d+(?:\.\d+)?)-` + filenameBase + `\.(\d+)-(\d+)\.` + ext + `$`)
+	pattern := `^v(\d+(?:\.\d+)?)-` + filenameBase + `\.(\d+)-(\d+)\.` + ext + `$`
+	reVal, ok := filterDirtyFilesReCache.Load(pattern)
+	if !ok {
+		re := regexp.MustCompile(pattern)
+		reVal, _ = filterDirtyFilesReCache.LoadOrStore(pattern, re)
+	}
+	re := reVal.(*regexp.Regexp)
 	var err error
 
 	for _, name := range fileNames {

execution/exec/txtask.go

@@ -699,10 +699,41 @@ type QueueWithRetry struct {
 	capacity int
 }
 
+var queuePool sync.Pool
+
 func NewQueueWithRetry(capacity int) *QueueWithRetry {
+	if v := queuePool.Get(); v != nil {
+		q := v.(*QueueWithRetry)
+		if q.capacity == capacity && q.newTasks != nil {
+			return q
+		}
+		// If capacity is wrong or channel is nil, we don't put it back here;
+		// another Get() might return a valid one, or we'll allocate a new one.
+	}
 	return &QueueWithRetry{newTasks: make(chan Task, capacity), capacity: capacity}
 }
 
+// Release drains the queue and returns it to the pool for reuse.
+// The channel is preserved (not closed), avoiding reallocation of the
+// 100K-element buffer on the next NewQueueWithRetry call.
+// Must be called only after all producers and consumers have stopped.
+func (q *QueueWithRetry) Release() {
+	q.lock.Lock()
+	if q.newTasks == nil {
+		q.lock.Unlock()
+		return
+	}
+	// Drain channel.
+	for len(q.newTasks) > 0 {
+		<-q.newTasks
+	}
+	// Clear retry heap, keep backing array.
+	q.retires = q.retires[:0]
+	q.closed = false
+	q.lock.Unlock()
+	queuePool.Put(q)
+}
+
 func (q *QueueWithRetry) NewTasksLen() int {
 	q.lock.Lock()
 	defer q.lock.Unlock()
@@ -733,7 +764,7 @@ func (q *QueueWithRetry) Add(ctx context.Context, t Task) {
 	newTasks := q.newTasks
 	q.lock.Unlock()
 
-	if !closed {
+	if !closed && newTasks != nil {
 		select {
 		case <-ctx.Done():
 			return
@@ -747,12 +778,12 @@ func (q *QueueWithRetry) Add(ctx context.Context, t Task) {
 // No limit on amount of txs added by this method.
 func (q *QueueWithRetry) ReTry(t Task) {
 	q.lock.Lock()
-	if q.closed {
+	newTasks := q.newTasks
+	if q.closed || newTasks == nil {
 		q.lock.Unlock()
 		return
 	}
 	heap.Push(&q.retires, t)
-	newTasks := q.newTasks
 	q.lock.Unlock()
 	select {
 	case newTasks <- nil:

execution/stagedsync/exec3_parallel.go

@@ -905,15 +905,17 @@ func (pe *parallelExecutor) run(ctx context.Context) (context.Context, context.C
 
 	pe.execLoopGroup.Go(func() error {
 		defer pe.rws.Close()
-		defer pe.in.Close()
+		defer execLoopCtxCancel() // cancel workers' context when exec loop exits
 		pe.resetWorkers(execLoopCtx, pe.rs, nil)
 		return pe.execLoop(execLoopCtx)
 	})
 
 	return execLoopCtx, func() {
 		execLoopCtxCancel()
-		defer pe.stopWorkers()
-		defer pe.in.Close()
+		defer func() {
+			pe.stopWorkers()
+			pe.in.Release()
+		}()
 
 		if err := pe.wait(ctx); err != nil {
 			pe.logger.Debug("exec loop cancel failed", "err", err)