Pool QueueWithRetry and cache filterDirtyFiles regexp

QueueWithRetry (79.5GB, 18.8% of allocs): pool via sync.Pool with
Release() that drains the 100K-element channel without closing it,
preserving the 1.6MB buffer across reuses. parallelExecutor.run uses
Release() instead of Close(); workers exit via context cancellation
when the exec loop goroutine defers execLoopCtxCancel(). Cleanup
ordering ensures stopWorkers() completes before Release().

filterDirtyFiles regexp (14.6GB, 3.3% of allocs): cache compiled
regexps in sync.Map keyed by pattern string. Each unique
(filenameBase, ext) pair compiles once instead of per-call.

Combined expected savings: ~94GB (22% of total allocations).

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

Author: 2 people

db/state/dirty_files.go

@@ -258,8 +258,15 @@ 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 {
+		reVal, _ = filterDirtyFilesReCache.LoadOrStore(pattern, regexp.MustCompile(pattern))
+	}
+	re := reVal.(*regexp.Regexp)
 	var err error
 
 	for _, name := range fileNames {

execution/exec/txtask.go

@@ -699,10 +699,36 @@ 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 {
+			return q
+		}
+		queuePool.Put(q) // wrong capacity, return to pool for another caller
+	}
 	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()
+	// 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()

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)