e3: one more reconst deadlock fix (erigontech#7207)

eth/stagedsync/exec3.go

@@ -1094,8 +1094,12 @@ func reconstituteStep(last bool,
 
 					select {
 					case workCh <- txTask:
-					case <-ctx.Done():
-						return ctx.Err()
+					case <-reconstWorkersCtx.Done():
+						// if ctx canceled, then maybe it's because of error in errgroup
+						//
+						// errgroup doesn't play with pattern where some 1 goroutine-producer is outside of errgroup
+						// but RwTx doesn't allow move between goroutines
+						return g.Wait()
 					}
 				}
 				inputTxNum++

turbo/snapshotsync/block_snapshots.go

@@ -1059,7 +1059,6 @@ func NewBlockRetire(workers int, tmpDir string, snapshots *RoSnapshots, db kv.Ro
 	return &BlockRetire{workers: workers, tmpDir: tmpDir, snapshots: snapshots, db: db, downloader: downloader, notifier: notifier}
 }
 func (br *BlockRetire) Snapshots() *RoSnapshots { return br.snapshots }
-func (br *BlockRetire) Working() bool           { return br.working.Load() }
 func (br *BlockRetire) NeedSaveFilesListInDB() bool {
 	return br.needSaveFilesListInDB.CompareAndSwap(true, false)
 }
@@ -1955,15 +1954,15 @@ func ForEachHeader(ctx context.Context, s *RoSnapshots, walker func(header *type
 }
 
 type Merger struct {
-	lvl      log.Lvl
-	workers  int
-	tmpDir   string
-	chainID  uint256.Int
-	notifier DBEventNotifier
+	lvl             log.Lvl
+	compressWorkers int
+	tmpDir          string
+	chainID         uint256.Int
+	notifier        DBEventNotifier
 }
 
-func NewMerger(tmpDir string, workers int, lvl log.Lvl, chainID uint256.Int, notifier DBEventNotifier) *Merger {
-	return &Merger{tmpDir: tmpDir, workers: workers, lvl: lvl, chainID: chainID, notifier: notifier}
+func NewMerger(tmpDir string, compressWorkers int, lvl log.Lvl, chainID uint256.Int, notifier DBEventNotifier) *Merger {
+	return &Merger{tmpDir: tmpDir, compressWorkers: compressWorkers, lvl: lvl, chainID: chainID, notifier: notifier}
 }
 
 type Range struct {
@@ -2078,7 +2077,7 @@ func (m *Merger) merge(ctx context.Context, toMerge []string, targetFile string,
 		expectedTotal += d.Count()
 	}
 
-	f, err := compress.NewCompressor(ctx, "Snapshots merge", targetFile, m.tmpDir, compress.MinPatternScore, m.workers, log.LvlTrace)
+	f, err := compress.NewCompressor(ctx, "Snapshots merge", targetFile, m.tmpDir, compress.MinPatternScore, m.compressWorkers, log.LvlTrace)
 	if err != nil {
 		return err
 	}