recover stale transactions when awaiting receipt

ethpandaops · Jan 17, 2024 · cd1cfe4 · cd1cfe4
1 parent 7d564ca
commit cd1cfe4
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 0 deletions.
diff --git a/pkg/coordinator/wallet/manager.go b/pkg/coordinator/wallet/manager.go
@@ -110,4 +110,8 @@ func (manager *Manager) processBlockTransactions(block *execution.Block) {
 			}
 		}
 	}
+
+	for _, wallet := range wallets {
+		wallet.processStaleConfirmations(block)
+	}
 }
diff --git a/pkg/coordinator/wallet/wallet.go b/pkg/coordinator/wallet/wallet.go
@@ -32,6 +32,7 @@ type Wallet struct {
 
 	confirmedNonce   uint64
 	confirmedBalance *big.Int
+	lastConfirmation uint64
 
 	txNonceChans map[uint64]*nonceStatus
 	txNonceMutex sync.Mutex
@@ -246,6 +247,8 @@ func (wallet *Wallet) processTransactionInclusion(block *execution.Block, tx *ty
 		return
 	}
 
+	wallet.lastConfirmation = block.Number
+
 	if receipt != nil {
 		wallet.confirmedBalance = wallet.confirmedBalance.Sub(wallet.confirmedBalance, tx.Value())
 		txFee := new(big.Int).Mul(receipt.EffectiveGasPrice, big.NewInt(int64(receipt.GasUsed)))
@@ -271,6 +274,38 @@ func (wallet *Wallet) processTransactionInclusion(block *execution.Block, tx *ty
 	}
 }
 
+func (wallet *Wallet) processStaleConfirmations(block *execution.Block) {
+	if !wallet.isReady {
+		return
+	}
+
+	if len(wallet.txNonceChans) > 0 && block.Number > wallet.lastConfirmation+10 {
+		wallet.lastConfirmation = block.Number
+		clients := block.GetSeenBy()
+		client := clients[0]
+
+		lastNonce, err := client.GetRPCClient().GetNonceAt(context.Background(), wallet.address, big.NewInt(int64(block.Number)))
+		if err != nil {
+			return
+		}
+
+		wallet.txNonceMutex.Lock()
+		defer wallet.txNonceMutex.Unlock()
+
+		if wallet.confirmedNonce >= lastNonce {
+			return
+		}
+
+		for n := range wallet.txNonceChans {
+			if n < lastNonce {
+				logrus.WithError(err).Warnf("recovering stale confirmed transactions for %v (nonce %v)", wallet.address.String(), n)
+				close(wallet.txNonceChans[n].channel)
+				delete(wallet.txNonceChans, n)
+			}
+		}
+	}
+}
+
 func (wallet *Wallet) processTransactionReceival(_ *execution.Block, tx *types.Transaction) {
 	if !wallet.isReady {
 		return