Better streamlining of sender recovery (erigontech#18390)

## Summary

- Refactor sender recovery stage to process individual transactions
instead of blocks for better parallelism

## Sender Recovery Stage (`execution/stagedsync/stage_senders.go`)

- Change job granularity from block-level to transaction-level, this
means that we have parallelism on the chain-tip.
- Each worker now processes individual transactions instead of entire
blocks
- Transactions from the same block can be processed concurrently across
workers
- Added `pendingBlock` tracking to aggregate results before writing to
collector

Before there was no parallelism in sender recovery.

next step: Query senders from txpool when possible. this is marginal but
it could make sense in the future.

Author: Giulio2002

execution/stagedsync/stage_senders.go

@@ -43,6 +43,7 @@ import (
 	"github.com/erigontech/erigon/execution/stagedsync/headerdownload"
 	"github.com/erigontech/erigon/execution/stagedsync/stages"
 	"github.com/erigontech/erigon/execution/types"
+	"github.com/erigontech/erigon/execution/types/accounts"
 	"github.com/erigontech/erigon/node/ethconfig"
 )
 
@@ -115,6 +116,7 @@ func SpawnRecoverSendersStage(cfg SendersCfg, s *StageState, u Unwinder, tx kv.R
 	defer logEvery.Stop()
 
 	startFrom := s.BlockNumber + 1
+	recoveryStart := time.Now()
 
 	jobs := make(chan *senderRecoveryJob, cfg.batchSize)
 	out := make(chan *senderRecoveryJob, cfg.batchSize)
@@ -136,6 +138,18 @@ func SpawnRecoverSendersStage(cfg SendersCfg, s *StageState, u Unwinder, tx kv.R
 	collectorSenders.SortAndFlushInBackground(true)
 	collectorSenders.LogLvl(log.LvlDebug)
 
+	// pendingBlocks tracks blocks that are being processed
+	// key: blockIndex, value: {senders slice, txCount, received count}
+	type pendingBlock struct {
+		senders  []byte
+		txCount  int
+		received int
+		hash     common.Hash
+	}
+	pendingBlocks := make(map[int]*pendingBlock)
+	var pendingMu sync.Mutex
+	var lastBlockIndex int
+
 	errCh := make(chan senderRecoveryError)
 	go func() {
 		defer dbg.LogPanic()
@@ -147,12 +161,6 @@ func SpawnRecoverSendersStage(cfg SendersCfg, s *StageState, u Unwinder, tx kv.R
 			select {
 			case <-quitCh:
 				return
-			case <-logEvery.C:
-				n := s.BlockNumber
-				if j != nil {
-					n += uint64(j.index)
-				}
-				logger.Info(fmt.Sprintf("[%s] Recovery", logPrefix), "block_number", n, "ch", fmt.Sprintf("%d/%d", len(jobs), cap(jobs)))
 			case j, ok = <-out:
 				if !ok {
 					return
@@ -162,13 +170,31 @@ func SpawnRecoverSendersStage(cfg SendersCfg, s *StageState, u Unwinder, tx kv.R
 					return
 				}
 
-				k := make([]byte, 4)
-				binary.BigEndian.PutUint32(k, uint32(j.index))
-				index := int(binary.BigEndian.Uint32(k))
-				if err := collectorSenders.Collect(dbutils.BlockBodyKey(s.BlockNumber+uint64(index)+1, j.blockHash), j.senders); err != nil {
-					errCh <- senderRecoveryError{err: j.err}
+				pendingMu.Lock()
+				pb := pendingBlocks[j.blockIndex]
+				if pb == nil {
+					pendingMu.Unlock()
+					errCh <- senderRecoveryError{err: fmt.Errorf("unexpected block index %d", j.blockIndex)}
 					return
 				}
+				// Copy sender address to the correct position
+				fromValue := j.from.Value()
+				copy(pb.senders[j.txIndex*length.Addr:], fromValue[:])
+				pb.received++
+
+				// Check if block is complete
+				if pb.received == pb.txCount {
+					if err := collectorSenders.Collect(dbutils.BlockBodyKey(s.BlockNumber+uint64(j.blockIndex)+1, pb.hash), pb.senders); err != nil {
+						pendingMu.Unlock()
+						errCh <- senderRecoveryError{err: err}
+						return
+					}
+					delete(pendingBlocks, j.blockIndex)
+					if j.blockIndex > lastBlockIndex {
+						lastBlockIndex = j.blockIndex
+					}
+				}
+				pendingMu.Unlock()
 			}
 		}
 	}()
@@ -237,26 +263,50 @@ Loop:
 			continue
 		}
 
-		j := &senderRecoveryJob{
-			body:        body,
-			blockNumber: blockNumber,
-			blockTime:   header.Time,
-			blockHash:   blockHash,
-			index:       int(blockNumber) - int(s.BlockNumber) - 1,
+		blockIndex := int(blockNumber) - int(s.BlockNumber) - 1
+		if blockIndex < 0 {
+			panic(blockIndex) //uint-underflow
 		}
-		if j.index < 0 {
-			panic(j.index) //uint-underflow
+
+		// Skip blocks with no transactions
+		if len(body.Transactions) == 0 {
+			// Write empty senders for blocks with no transactions
+			if err := collectorSenders.Collect(dbutils.BlockBodyKey(s.BlockNumber+uint64(blockIndex)+1, blockHash), nil); err != nil {
+				return err
+			}
+			continue
 		}
-		select {
-		case recoveryErr := <-errCh:
-			if recoveryErr.err != nil {
-				cancelWorkers()
-				if err := handleRecoverErr(recoveryErr); err != nil {
-					return err
+
+		// Register pending block
+		pendingMu.Lock()
+		pendingBlocks[blockIndex] = &pendingBlock{
+			senders: make([]byte, len(body.Transactions)*length.Addr),
+			txCount: len(body.Transactions),
+			hash:    blockHash,
+		}
+		pendingMu.Unlock()
+
+		// Send individual transaction jobs
+		for txIdx, txn := range body.Transactions {
+			j := &senderRecoveryJob{
+				txn:         txn,
+				blockNumber: blockNumber,
+				blockTime:   header.Time,
+				blockHash:   blockHash,
+				blockIndex:  blockIndex,
+				txIndex:     txIdx,
+			}
+			select {
+			case recoveryErr := <-errCh:
+				if recoveryErr.err != nil {
+					cancelWorkers()
+					if err := handleRecoverErr(recoveryErr); err != nil {
+						return err
+					}
+					break Loop
 				}
-				break Loop
+			case jobs <- j:
 			}
-		case jobs <- j:
 		}
 	}
 
@@ -304,6 +354,7 @@ Loop:
 		if err = s.Update(tx, to); err != nil {
 			return err
 		}
+		log.Debug(fmt.Sprintf("[%s] Recovery done", logPrefix), "from", startFrom, "to", to, "blocks", to-startFrom+1, "took", time.Since(recoveryStart))
 	}
 
 	if !useExternalTx {
@@ -321,12 +372,13 @@ type senderRecoveryError struct {
 }
 
 type senderRecoveryJob struct {
-	body        *types.Body
-	senders     []byte
+	txn         types.Transaction
+	from        accounts.Address
 	blockHash   common.Hash
 	blockNumber uint64
 	blockTime   uint64
-	index       int
+	blockIndex  int // index of the block relative to s.BlockNumber
+	txIndex     int // index of the tx within the block
 	err         error
 }
 
@@ -348,20 +400,15 @@ func recoverSenders(ctx context.Context, logPrefix string, cryptoContext *secp25
 			return
 		}
 
-		body := job.body
-		job.body = nil // reduce ram usage and help GC
 		signer := types.MakeSigner(config, job.blockNumber, job.blockTime)
-		job.senders = make([]byte, len(body.Transactions)*length.Addr)
-		for i, txn := range body.Transactions {
-			from, err := signer.SenderWithContext(cryptoContext, txn)
-			if err != nil {
-				job.err = fmt.Errorf("%w: error recovering sender for tx=%x, %v",
-					rules.ErrInvalidBlock, txn.Hash(), err)
-				break
-			}
-			fromValue := from.Value()
-			copy(job.senders[i*length.Addr:], fromValue[:])
+		from, err := signer.SenderWithContext(cryptoContext, job.txn)
+		if err != nil {
+			job.err = fmt.Errorf("%w: error recovering sender for tx=%x, %v",
+				rules.ErrInvalidBlock, job.txn.Hash(), err)
+		} else {
+			job.from = from
 		}
+		job.txn = nil // reduce ram usage and help GC
 
 		// prevent sending to close channel
 		if err := common.Stopped(quit); err != nil {