db/state: make COLLATE/MERGE/COMPRESS_WORKERS reach the aggregator

db/state/aggregator.go

@@ -158,7 +158,7 @@ func newAggregator(ctx context.Context, dirs datadir.Dirs, reorgBlockDepth uint6
 		leakDetector:       dbg.NewLeakDetector("agg", dbg.SlowTx()),
 		backgroundProgress: background.NewProgressSet(),
 		logger:             logger,
-		workers:            workersCfg{merge: 1, collateAndBuild: 1},
+		workers:            workersCfg{merge: dbg.MergeWorkers, collateAndBuild: dbg.CollateWorkers},
 
 		produce: true,
 	}
@@ -242,6 +242,7 @@ func (a *Aggregator) RegisterDomain(cfg statecfg.DomainCfg, salt *uint32, dirs d
 	}
 	a.d[cfg.Name].salt.Store(salt)
 	a.AddDependencyBtwnHistoryII(cfg.Name)
+	a.applyEnvCompressWorkers()
 	return nil
 }
 
@@ -259,6 +260,7 @@ func (a *Aggregator) RegisterII(cfg statecfg.InvIdxCfg, salt *uint32, dirs datad
 	}
 	a.iis[a.iisCount] = ii
 	a.iisCount++
+	a.applyEnvCompressWorkers()
 	return nil
 }
 
@@ -787,6 +789,16 @@ func (a *Aggregator) setBuildAccessorsWorkers(i int) {
 	})
 }
 
+// applyEnvCompressWorkers covers the window before the first preset lands: a restart with
+// files on disk merges with whatever the compressor config holds at registration.
+func (a *Aggregator) applyEnvCompressWorkers() {
+	if dbg.CompressWorkers <= 0 {
+		return
+	}
+	a.setCompressWorkers(dbg.CompressWorkers)
+	a.setBuildAccessorsWorkers(dbg.CompressWorkers)
+}
+
 func (a *Aggregator) setCompressWorkers(i int) {
 	a.workers.trySet(func() {
 		for _, d := range a.d {

db/state/aggregator2.go

@@ -144,9 +144,10 @@ func (opts AggOpts) WithErigonDBSettings(s *ErigonDBSettings) AggOpts { //nolint
 
 type workersCfg struct {
 	mu              sync.Mutex
-	editLocks       int // >0 while background build/merge pins config; Preset* writes are no-ops
+	editLocks       int // >0 while background build/merge pins config
 	merge           int // usually 1
 	collateAndBuild int
+	pending         []func() // requests that arrived while pinned
 }
 
 func (w *workersCfg) getMerge() int {
@@ -156,11 +157,7 @@ func (w *workersCfg) getMerge() int {
 }
 
 func (w *workersCfg) setMerge(n int) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	if w.editLocks == 0 {
-		w.merge = n
-	}
+	w.trySet(func() { w.merge = n })
 }
 
 func (w *workersCfg) getCollateAndBuild() int {
@@ -170,20 +167,19 @@ func (w *workersCfg) getCollateAndBuild() int {
 }
 
 func (w *workersCfg) setCollateAndBuild(n int) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	if w.editLocks == 0 {
-		w.collateAndBuild = n
-	}
+	w.trySet(func() { w.collateAndBuild = n })
 }
 
-// trySet runs fn under mu only while editing is unlocked (no background op holds it).
+// trySet runs fn under mu, or queues it for the last unlockEditing while a background op pins
+// the config: dropping the request loses it for the process — a restart merges before any preset.
 func (w *workersCfg) trySet(fn func()) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	if w.editLocks == 0 {
-		fn()
+	if w.editLocks > 0 {
+		w.pending = append(w.pending, fn)
+		return
 	}
+	fn()
 }
 
 // lockEditing is reentrant: overlapping build/merge ops each hold a lock, and
@@ -200,6 +196,13 @@ func (w *workersCfg) unlockEditing() {
 	if w.editLocks > 0 {
 		w.editLocks--
 	}
+	if w.editLocks != 0 {
+		return
+	}
+	for _, fn := range w.pending {
+		fn()
+	}
+	w.pending = nil
 }
 
 func CheckSnapshotsCompatibility(d datadir.Dirs) error {

db/state/aggregator2_test.go

@@ -49,3 +49,57 @@ func TestWorkersCfgEditingLockIsReentrant(t *testing.T) {
 	w.unlockEditing()
 	require.True(t, editable(), "extra unlock must not underflow and disable editing")
 }
+
+func TestWorkersCfgAppliesRequestDeferredWhilePinned(t *testing.T) {
+	t.Parallel()
+	w := &workersCfg{merge: 1, collateAndBuild: 1}
+	compress := 0
+
+	w.lockEditing()
+	w.setMerge(2)
+	w.setCollateAndBuild(4)
+	w.trySet(func() { compress = 8 })
+	require.Equal(t, 1, w.getMerge(), "must not change under a running build or merge")
+	require.Equal(t, 1, w.getCollateAndBuild())
+	require.Zero(t, compress)
+
+	w.unlockEditing()
+	require.Equal(t, 2, w.getMerge(), "held request applies on release")
+	require.Equal(t, 4, w.getCollateAndBuild())
+	require.Equal(t, 8, compress, "every queued request runs, not just the last")
+}
+
+func TestWorkersCfgHoldsRequestUntilLastPinReleases(t *testing.T) {
+	t.Parallel()
+	w := &workersCfg{merge: 1, collateAndBuild: 1}
+
+	w.lockEditing()
+	w.lockEditing()
+	w.setMerge(2)
+	w.setCollateAndBuild(4)
+
+	w.unlockEditing()
+	require.Equal(t, 1, w.getCollateAndBuild(), "one of two overlapping pins released is still pinned")
+
+	w.unlockEditing()
+	require.Equal(t, 4, w.getCollateAndBuild())
+}
+
+func TestWorkersCfgKeepsOnlyTheNewestRequest(t *testing.T) {
+	t.Parallel()
+	w := &workersCfg{merge: 1, collateAndBuild: 1}
+
+	w.lockEditing()
+	w.setMerge(2)
+	w.setCollateAndBuild(4)
+	w.setMerge(3)
+	w.setCollateAndBuild(6)
+	w.unlockEditing()
+	require.Equal(t, 3, w.getMerge())
+	require.Equal(t, 6, w.getCollateAndBuild())
+
+	// A pin/release cycle with no request in between must not resurrect a stale value.
+	w.lockEditing()
+	w.unlockEditing()
+	require.Equal(t, 6, w.getCollateAndBuild())
+}