Fix worker panic bug and add some info level logs (#107)

* Fix worker panic bug and add some info level logs

* Changed to use ClosePreconfResultFn to avoid recover

* Revert some unnecessary changes

* add preconf tests

* revert txpool log

* Add cost in preconf api

* fix close timing

* add PreconfResponse buffer

* Optimize the pre-confirmation queue performance from a maximum of 200ms+ to 2ms+

* Lower some preconf log levels

Author: asdv23

core/events.go

@@ -37,8 +37,9 @@ type NewPreconfTxEvent struct {
 
 // NewPreconfTxRequestEvent is posted when a preconf transaction request enters the transaction pool.
 type NewPreconfTxRequest struct {
-	Tx            *types.Transaction
-	PreconfResult chan<- *PreconfResponse
+	Tx                   *types.Transaction
+	PreconfResult        chan<- *PreconfResponse
+	ClosePreconfResultFn func()
 }
 
 type PreconfResponse struct {

core/txpool/txpool.go

@@ -627,6 +627,7 @@ func (pool *TxPool) Locals() []common.Address {
 
 // PendingPreconfTxs retrieves the preconf transactions and the pending transactions.
 func (pool *TxPool) PendingPreconfTxs(enforceTips bool) ([]*types.Transaction, map[common.Address]types.Transactions) {
+	defer preconf.MetricsPreconfTxPoolFilterCost(time.Now())
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 
@@ -676,36 +677,36 @@ func (pool *TxPool) extractPreconfTxsFromPending(pending map[common.Address]type
 
 	// removes the preconf transaction from the pending map, maintaining the order.
 	preconfTxs := make([]*types.Transaction, 0)
-	for _, preconfTx := range pool.preconfTxs.Transactions() {
-		from, _ := types.Sender(pool.signer, preconfTx)
+	for _, preconfTx := range pool.preconfTxs.TxEntries() {
+		preconfTxHash := preconfTx.Tx.Hash()
 
 		// Get the slice of transactions for the target address
-		txs, exists := pending[from]
+		txs, exists := pending[preconfTx.From]
 
 		// If the transaction isn't in pending map but it's expected to be there,
 		// show the error log.
 		if !exists || len(txs) == 0 {
-			log.Error("Missing transaction in pending map, please report the issue", "hash", preconfTx.Hash())
-			pool.preconfTxs.Remove(preconfTx.Hash()) // remove it prevent log always print
+			log.Error("Missing transaction in pending map, please report the issue", "hash", preconfTxHash)
+			pool.preconfTxs.Remove(preconfTxHash) // remove it prevent log always print
 			continue
 		}
 
 		// Create a new slice to hold the transactions that are not deleted
 		var newTxs []*types.Transaction
 		for _, tx := range txs {
-			if tx.Hash() != preconfTx.Hash() {
+			if tx.Hash() != preconfTxHash {
 				newTxs = append(newTxs, tx) // Only keep the transactions that are not to be deleted
 			}
 		}
 
 		// Update the slice in the map
 		if len(newTxs) == 0 {
-			delete(pending, from) // If the slice is empty, delete the entry for that address
+			delete(pending, preconfTx.From) // If the slice is empty, delete the entry for that address
 		} else {
-			pending[from] = newTxs // Replace with the new slice
+			pending[preconfTx.From] = newTxs // Replace with the new slice
 		}
 
-		preconfTxs = append(preconfTxs, preconfTx)
+		preconfTxs = append(preconfTxs, preconfTx.Tx)
 	}
 	return preconfTxs
 }
@@ -1279,15 +1280,13 @@ func (pool *TxPool) addPreconfTx(tx *types.Transaction) {
 		return
 	}
 
-	pool.preconfTxs.Add(tx)
+	pool.preconfTxs.Add(from, tx)
 
 	// handle preconf txs
 	pool.handlePreconfTxs([]*types.Transaction{tx})
 }
 
 func (pool *TxPool) handlePreconfTxs(news []*types.Transaction) {
-	defer preconf.MetricsPreconfTxPoolHandleCost(time.Now())
-
 	for _, tx := range news {
 		txHash := tx.Hash()
 		if !pool.preconfTxs.Contains(txHash) {
@@ -1296,25 +1295,28 @@ func (pool *TxPool) handlePreconfTxs(news []*types.Transaction) {
 		}
 
 		// send preconf request event
-		result := make(chan *core.PreconfResponse)
+		result := make(chan *core.PreconfResponse, 1) // buffer 1 to avoid worker blocking
 		pool.preconfTxRequestFeed.Send(core.NewPreconfTxRequest{
 			Tx:            tx,
 			PreconfResult: result,
+			ClosePreconfResultFn: func() {
+				close(result)
+			},
 		})
-		log.Trace("txpool sent preconf tx request", "tx", txHash)
+		log.Debug("txpool sent preconf tx request", "tx", txHash)
 
 		// avoid race condition
 		tx := tx
 		// goroutine to avoid blocking
 		go func() {
-			defer close(result)
+			defer preconf.MetricsPreconfTxPoolHandleCost(time.Now())
 
-			// wait for preconf ready, no need to wait again after it's ready, as it only needs to wait once when the service restarts
-			// wait for 10s, if not ready, skip
+			// If preconfReadyCh is not closed, it means this is a preconf tx restored from journal after system restart.
+			// In this case, we don't need to execute preconfirmation again to avoid resource contention with worker.
 			select {
 			case <-pool.preconfReadyCh:
-			case <-time.After(time.Minute):
-				log.Error("preconf txs not ready, skip handle", "tx", txHash)
+			default:
+				log.Info("preconf txs not ready, skip handle", "tx", txHash)
 				return
 			}
 
@@ -1326,10 +1328,11 @@ func (pool *TxPool) handlePreconfTxs(news []*types.Transaction) {
 			// timeout
 			timeout := time.NewTimer(pool.config.Preconf.PreconfTimeout)
 			defer timeout.Stop()
+			now := time.Now()
 			// wait for miner.worker preconf response
 			select {
 			case response := <-result:
-				log.Trace("txpool received preconf tx response", "tx", txHash)
+				log.Trace("txpool received preconf tx response", "tx", txHash, "duration", time.Since(now))
 				if response.Err == nil {
 					event.Status = core.PreconfStatusSuccess
 				} else {
@@ -1347,7 +1350,7 @@ func (pool *TxPool) handlePreconfTxs(news []*types.Transaction) {
 				}
 			case <-timeout.C:
 				event.Status = core.PreconfStatusFailed
-				event.Reason = "preconf timeout"
+				event.Reason = fmt.Sprintf("preconf timeout, over %s timeout", time.Since(now))
 			}
 
 			// send preconf event
@@ -1359,7 +1362,7 @@ func (pool *TxPool) handlePreconfTxs(news []*types.Transaction) {
 				log.Trace("preconf success", "tx", txHash)
 				if pool.journal != nil {
 					if err := pool.journal.insert(tx); err != nil {
-						log.Warn("Failed to journal preconf success transaction", "tx", txHash, "err", err)
+						log.Error("Failed to journal preconf success transaction", "tx", txHash, "err", err)
 					} else {
 						log.Trace("preconf success transaction journaled", "tx", txHash)
 					}

core/txpool/txpool_test.go

@@ -2557,6 +2557,10 @@ func TestExtractPreconfTxsFromPending(t *testing.T) {
 	tx4 := types.NewTransaction(0, addr3, big.NewInt(400), 21000, big.NewInt(1), nil)
 	tx4, _ = types.SignTx(tx4, signer, key3)
 
+	from1, _ := types.Sender(signer, tx1)
+	from2, _ := types.Sender(signer, tx2)
+	from3, _ := types.Sender(signer, tx3)
+	from4, _ := types.Sender(signer, tx4)
 	// Create a test TxPool
 	createTxPool := func() *TxPool {
 		pool := &TxPool{
@@ -2570,7 +2574,7 @@ func TestExtractPreconfTxsFromPending(t *testing.T) {
 			preconfTxs: preconf.NewFIFOTxSet(),
 		}
 		// Add pre-confirmed transaction
-		pool.preconfTxs.Add(tx1)
+		pool.preconfTxs.Add(from1, tx1)
 		return pool
 	}
 
@@ -2636,9 +2640,9 @@ func TestExtractPreconfTxsFromPending(t *testing.T) {
 		pool.config.Preconf.AllPreconfs = true
 
 		// Add all transactions to the pre-confirmed set
-		pool.preconfTxs.Add(tx2)
-		pool.preconfTxs.Add(tx3)
-		pool.preconfTxs.Add(tx4)
+		pool.preconfTxs.Add(from2, tx2)
+		pool.preconfTxs.Add(from3, tx3)
+		pool.preconfTxs.Add(from4, tx4)
 
 		pending := make(map[common.Address]types.Transactions)
 		pending[addr1] = types.Transactions{tx1, tx2}

eth/api_backend.go

@@ -353,7 +353,7 @@ func (b *EthAPIBackend) sendTxWithPreconf(ctx context.Context, tx *types.Transac
 			}
 		case <-ctx.Done():
 			log.Trace("preconf tx event not received", "tx", txHash, "err", ctx.Err())
-			return nil, errors.New("preconf tx event not received")
+			return nil, fmt.Errorf("preconf tx event not received, txHash: %s, err: %w", txHash, ctx.Err())
 		}
 	}
 }

internal/ethapi/api.go

@@ -48,6 +48,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/params"
+	"github.com/ethereum/go-ethereum/preconf"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/rpc"
 )
@@ -2161,6 +2162,8 @@ func (s *TransactionAPI) SendRawTransaction(ctx context.Context, input hexutil.B
 // SendRawTransactionWithPreconf will add the signed preconf transaction to the transaction pool and return the preconf result.
 // The sender is responsible for signing the transaction and using the correct nonce.
 func (s *TransactionAPI) SendRawTransactionWithPreconf(ctx context.Context, input hexutil.Bytes) (*core.NewPreconfTxEvent, error) {
+	defer preconf.MetricsPreconfAPIHandleCost(time.Now())
+
 	tx := new(types.Transaction)
 	if err := tx.UnmarshalBinary(input); err != nil {
 		return nil, err
@@ -2177,6 +2180,9 @@ func (s *TransactionAPI) SendRawTransactionWithPreconf(ctx context.Context, inpu
 		return nil, errors.New("only replay-protected (EIP-155) transactions allowed over RPC")
 	}
 
+	now := time.Now()
+	log.Trace("ethapi sendRawTransactionWithPreconf", "tx", tx.Hash())
+
 	// Send the transaction with preconf
 	result, err := s.b.SendTxWithPreconf(ctx, tx)
 	if err != nil {
@@ -2192,9 +2198,9 @@ func (s *TransactionAPI) SendRawTransactionWithPreconf(ctx context.Context, inpu
 
 	if tx.To() == nil {
 		addr := crypto.CreateAddress(from, tx.Nonce())
-		log.Info("Submitted contract creation", "hash", tx.Hash().Hex(), "from", from, "nonce", tx.Nonce(), "contract", addr.Hex(), "value", tx.Value())
+		log.Info("Submitted preconf contract creation", "hash", tx.Hash().Hex(), "from", from, "nonce", tx.Nonce(), "contract", addr.Hex(), "value", tx.Value())
 	} else {
-		log.Info("Submitted transaction", "hash", tx.Hash().Hex(), "from", from, "nonce", tx.Nonce(), "recipient", tx.To(), "value", tx.Value())
+		log.Info("Submitted preconf transaction", "hash", tx.Hash().Hex(), "from", from, "nonce", tx.Nonce(), "recipient", tx.To(), "value", tx.Value(), "duration", time.Since(now))
 	}
 
 	return result, nil

miner/preconf_checker.go

@@ -75,7 +75,7 @@ func NewPreconfChecker(chainConfig *params.ChainConfig, chain *core.BlockChain,
 
 func (c *preconfChecker) loop() {
 	if !c.minerConfig.EnablePreconfChecker {
-		log.Debug("preconf checker is disabled, skip loop")
+		log.Info("preconf checker is disabled, skip loop")
 		return
 	}
 	for {
@@ -298,7 +298,7 @@ func (c *preconfChecker) precheck() error {
 	// The distance between envblock.number and (engine_sync_target.number or unsafe_l2.number) should not exceed 6.
 	// Here's an explanation for why it's 6: a deposit transaction from L1 includes at least 1(l1.header.number-1 rather than l1.header.number-2) L1 block,
 	// which corresponds to 6 L2 blocks. Therefore, we can have up to 6 blocks in advance to ensure that preconfirmations are not affected by deposit transactions.
-	if envBlockNumber-engineSyncTargetBlockNumber > c.minerConfig.EthToleranceBlock() || envBlockNumber-unsafeL2BlockNumber > 6 {
+	if envBlockNumber-engineSyncTargetBlockNumber > c.minerConfig.PreconfBufferBlock || envBlockNumber-unsafeL2BlockNumber > c.minerConfig.PreconfBufferBlock {
 		// When there are a large number of preconfirmation transactions in the queue, it may cause the future 6 blocks to be
 		// filled with preconfirmation transactions. At this point, stop new preconfirmation transactions from entering,
 		// because there may be unblocked deposit transactions in future blocks, which cannot be predicted at this time.
@@ -356,15 +356,15 @@ func (c *preconfChecker) PausePreconf() chan<- []*types.Transaction {
 	}
 	c.unSealedPreconfTxsCh = make(chan []*types.Transaction, 1) // buffer 1 to avoid worker block
 
-	log.Trace("pause preconf")
+	log.Debug("pause preconf")
 	return c.unSealedPreconfTxsCh
 }
 
 func (c *preconfChecker) UnpausePreconf(env *environment, preconfReady func()) {
 	defer c.mu.Unlock()
 	c.env = env
 	c.envUpdatedAt = time.Now()
-	log.Trace("unpause preconf", "env.header.Number", env.header.Number.Int64(), "env.gasPool", env.gasPool.Gas(), "envUpdatedAt", c.envUpdatedAt)
+	log.Debug("unpause preconf", "env.header.Number", env.header.Number.Int64(), "env.gasPool", env.gasPool.Gas(), "envUpdatedAt", c.envUpdatedAt)
 	// reset env
 	c.env.header.Number = new(big.Int).Add(c.env.header.Number, common.Big1)
 	c.env.gasPool.SetGas(c.env.header.GasLimit)
@@ -402,4 +402,6 @@ func (c *preconfChecker) UnpausePreconf(env *environment, preconfReady func()) {
 
 	// notify txpool that preconf is ready
 	preconfReady()
+
+	log.Info("ready to preconf", "env.header.Number", env.header.Number.Int64(), "env.gasPool", env.gasPool.Gas(), "envUpdatedAt", c.envUpdatedAt, "deposit_txs", len(c.depositTxs), "unsealed_preconf_txs", len(unsealedPreconfTxs))
 }

miner/worker.go

@@ -653,19 +653,24 @@ func (w *worker) mainLoop() {
 				}
 			}
 		case ev := <-w.preconfTxRequestCh:
-			log.Trace("worker received preconf tx request", "tx", ev.Tx.Hash())
+			now := time.Now()
+			log.Debug("worker received preconf tx request", "tx", ev.Tx.Hash())
+
 			receipt, err := w.preconfChecker.Preconf(ev.Tx)
 			if err != nil {
 				// Not fatal, just warn to the log
 				log.Warn("preconf failed", "tx", ev.Tx.Hash(), "err", err)
 			}
-			// Prevent panic caused by writing after ev.PreconfResult is closed
+			log.Trace("worker preconf tx executed", "tx", ev.Tx.Hash(), "duration", time.Since(now))
+
 			select {
 			case ev.PreconfResult <- &core.PreconfResponse{Receipt: receipt, Err: err}:
-				log.Trace("worker sent preconf tx response", "tx", ev.Tx.Hash())
+				log.Debug("worker sent preconf tx response", "tx", ev.Tx.Hash(), "duration", time.Since(now))
 			case <-time.After(time.Second):
 				log.Warn("preconf tx response timeout, preconf result is closed?", "tx", ev.Tx.Hash())
 			}
+			ev.ClosePreconfResultFn()
+
 		case ev := <-w.txsCh:
 			if w.chainConfig.Optimism != nil && !w.config.RollupComputePendingBlock {
 				continue // don't update the pending-block snapshot if we are not computing the pending block
@@ -1255,6 +1260,7 @@ func (w *worker) fillTransactions(interrupt *int32, env *environment) error {
 	// Split the pending transactions into locals and remotes
 	// Fill the block with all available pending transactions.
 	preconfTxs, pending := w.eth.TxPool().PendingPreconfTxs(true)
+	log.Debug("find preconf txs to fill into block", "count", len(preconfTxs))
 
 	var unsealedPreconfTxs []*types.Transaction
 	if len(preconfTxs) > 0 {