Add forced batches tx to addrQueue

sequencer/addrqueue.go

@@ -18,6 +18,7 @@ type addrQueue struct {
 	currentBalance    *big.Int
 	readyTx           *TxTracker
 	notReadyTxs       map[uint64]*TxTracker
+	forcedTxs         map[common.Hash]struct{}
 	pendingTxsToStore map[common.Hash]struct{}
 }
 
@@ -30,6 +31,7 @@ func newAddrQueue(addr common.Address, nonce uint64, balance *big.Int) *addrQueu
 		currentBalance:    balance,
 		readyTx:           nil,
 		notReadyTxs:       make(map[uint64]*TxTracker),
+		forcedTxs:         make(map[common.Hash]struct{}),
 		pendingTxsToStore: make(map[common.Hash]struct{}),
 	}
 }
@@ -78,6 +80,11 @@ func (a *addrQueue) addTx(tx *TxTracker) (newReadyTx, prevReadyTx, replacedTx *T
 	}
 }
 
+// addForcedTx adds a forced tx to the list of forced txs
+func (a *addrQueue) addForcedTx(txHash common.Hash) {
+	a.forcedTxs[txHash] = struct{}{}
+}
+
 // addPendingTxToStore adds a tx to the list of pending txs to store in the DB (trusted state)
 func (a *addrQueue) addPendingTxToStore(txHash common.Hash) {
 	a.pendingTxsToStore[txHash] = struct{}{}
@@ -110,7 +117,7 @@ func (a *addrQueue) ExpireTransactions(maxTime time.Duration) ([]*TxTracker, *Tx
 
 // IsEmpty returns true if the addrQueue is empty
 func (a *addrQueue) IsEmpty() bool {
-	return a.readyTx == nil && len(a.notReadyTxs) == 0 && len(a.pendingTxsToStore) == 0
+	return a.readyTx == nil && len(a.notReadyTxs) == 0 && len(a.forcedTxs) == 0 && len(a.pendingTxsToStore) == 0
 }
 
 // deleteTx deletes the tx from the addrQueue
@@ -133,6 +140,15 @@ func (a *addrQueue) deleteTx(txHash common.Hash) (deletedReadyTx *TxTracker) {
 	}
 }
 
+// deleteForcedTx deletes the tx from the addrQueue
+func (a *addrQueue) deleteForcedTx(txHash common.Hash) {
+	if _, found := a.forcedTxs[txHash]; found {
+		delete(a.forcedTxs, txHash)
+	} else {
+		log.Warnf("tx (%s) not found in forcedTxs list", txHash.String())
+	}
+}
+
 // deletePendingTxToStore delete a tx from the list of pending txs to store in the DB (trusted state)
 func (a *addrQueue) deletePendingTxToStore(txHash common.Hash) {
 	if _, found := a.pendingTxsToStore[txHash]; found {

sequencer/finalizer.go

@@ -70,7 +70,8 @@ type finalizer struct {
 }
 
 type transactionToStore struct {
-	txTracker     *TxTracker
+	hash          common.Hash
+	from          common.Address
 	response      *state.ProcessTransactionResponse
 	batchResponse *state.ProcessBatchResponse
 	batchNumber   uint64
@@ -207,10 +208,8 @@ func (f *finalizer) storePendingTransactions(ctx context.Context) {
 			// Now f.storedFlushID >= tx.flushId, we can store tx
 			f.storeProcessedTx(ctx, tx)
 
-			if tx.txTracker != nil {
-				// Delete the txTracker from the pending list in the worker (addrQueue)
-				f.worker.DeletePendingTxToStore(tx.txTracker.Hash, tx.txTracker.From)
-			}
+			// Delete the tx from the pending list in the worker (addrQueue)
+			f.worker.DeletePendingTxToStore(tx.hash, tx.from)
 
 			f.pendingTransactionsToStoreWG.Done()
 		case <-ctx.Done():
@@ -264,6 +263,32 @@ func (f *finalizer) listenForClosingSignals(ctx context.Context) {
 		// ForcedBatch ch
 		case fb := <-f.closingSignalCh.ForcedBatchCh:
 			log.Debugf("finalizer received forced batch at block number: %v", fb.BlockNumber)
+
+			// Add the forced batch's transactions to the worker
+			batchNumber, err := f.dbManager.GetLastBatchNumber(ctx)
+			if err != nil {
+				// An error accessing the database is fatal
+				f.halt(ctx, err)
+			}
+
+			// Decode the transactions inside the forced batch
+			forkID := f.dbManager.GetForkIDByBatchNumber(batchNumber)
+			txs, _, _, err := state.DecodeTxs(fb.RawTxsData, forkID)
+			if err != nil {
+				// A forced batch can contain anything so this may happen
+				log.Warnf("failed to decode transactions from forced batch, Err: %v", err)
+			} else {
+				// If txs could be extracted from the forced batch, add them to the worker
+				for _, tx := range txs {
+					sender, err := state.GetSender(tx)
+					if err != nil {
+						log.Warnf("failed to get sender from tx, Err: %v", err)
+						continue
+					}
+					f.worker.AddForcedTx(tx.Hash(), sender)
+				}
+			}
+
 			f.nextForcedBatchesMux.Lock()
 			f.nextForcedBatches = f.sortForcedBatches(append(f.nextForcedBatches, fb))
 			if f.nextForcedBatchDeadline == 0 {
@@ -301,18 +326,16 @@ func (f *finalizer) updateLastPendingFlushID(newFlushID uint64) {
 // addPendingTxToStore adds a pending tx that is ready to be stored in the state DB once its flushid has been stored by the executor
 func (f *finalizer) addPendingTxToStore(ctx context.Context, txToStore transactionToStore) {
 	f.pendingTransactionsToStoreWG.Add(1)
-	if txToStore.txTracker != nil {
-		f.worker.AddPendingTxToStore(txToStore.txTracker.Hash, txToStore.txTracker.From)
-	}
+
+	f.worker.AddPendingTxToStore(txToStore.hash, txToStore.from)
+
 	select {
 	case f.pendingTransactionsToStore <- txToStore:
 	case <-ctx.Done():
 		// If context is cancelled before we can send to the channel, we must decrement the WaitGroup count and
 		// delete the pending TxToStore added in the worker
 		f.pendingTransactionsToStoreWG.Done()
-		if txToStore.txTracker != nil {
-			f.worker.DeletePendingTxToStore(txToStore.txTracker.Hash, txToStore.txTracker.From)
-		}
+		f.worker.DeletePendingTxToStore(txToStore.hash, txToStore.from)
 	}
 }
 
@@ -692,7 +715,8 @@ func (f *finalizer) handleProcessTransactionResponse(ctx context.Context, tx *Tx
 	}
 
 	txToStore := transactionToStore{
-		txTracker:     tx,
+		hash:          tx.Hash,
+		from:          tx.From,
 		response:      result.Responses[0],
 		batchResponse: result,
 		batchNumber:   f.batch.batchNumber,
@@ -709,9 +733,7 @@ func (f *finalizer) handleProcessTransactionResponse(ctx context.Context, tx *Tx
 
 	f.batch.countOfTxs++
 
-	if tx != nil {
-		f.updateWorkerAfterSuccessfulProcessing(ctx, tx, result)
-	}
+	f.updateWorkerAfterSuccessfulProcessing(ctx, tx.Hash, tx.From, false, result)
 
 	return nil, nil
 }
@@ -730,8 +752,14 @@ func (f *finalizer) handleForcedTxsProcessResp(ctx context.Context, request stat
 			}
 		}
 
+		from, err := state.GetSender(txResp.Tx)
+		if err != nil {
+			log.Errorf("handleForcedTxsProcessResp: failed to get sender: %s", err)
+		}
+
 		txToStore := transactionToStore{
-			txTracker:     nil,
+			hash:          txResp.TxHash,
+			from:          from,
 			response:      txResp,
 			batchResponse: result,
 			batchNumber:   request.BatchNumber,
@@ -747,6 +775,10 @@ func (f *finalizer) handleForcedTxsProcessResp(ctx context.Context, request stat
 		f.updateLastPendingFlushID(result.FlushID)
 
 		f.addPendingTxToStore(ctx, txToStore)
+
+		if err == nil {
+			f.updateWorkerAfterSuccessfulProcessing(ctx, txResp.TxHash, from, true, result)
+		}
 	}
 }
 
@@ -765,20 +797,26 @@ func (f *finalizer) storeProcessedTx(ctx context.Context, txToStore transactionT
 	metrics.TxProcessed(metrics.TxProcessedLabelSuccessful, 1)
 }
 
-func (f *finalizer) updateWorkerAfterSuccessfulProcessing(ctx context.Context, tx *TxTracker, result *state.ProcessBatchResponse) {
-	// Delete the transaction from the txSorted list
-	f.worker.DeleteTx(tx.Hash, tx.From)
-	log.Debug("tx deleted from txSorted list", "txHash", tx.Hash.String(), "from", tx.From.Hex())
+func (f *finalizer) updateWorkerAfterSuccessfulProcessing(ctx context.Context, txHash common.Hash, txFrom common.Address, isForced bool, result *state.ProcessBatchResponse) {
+	// Delete the transaction from the worker
+	if isForced {
+		f.worker.DeleteForcedTx(txHash, txFrom)
+		log.Debug("forced tx deleted from worker", "txHash", txHash.String(), "from", txFrom.Hex())
+		return
+	} else {
+		f.worker.DeleteTx(txHash, txFrom)
+		log.Debug("tx deleted from worker", "txHash", txHash.String(), "from", txFrom.Hex())
+	}
 
 	start := time.Now()
-	txsToDelete := f.worker.UpdateAfterSingleSuccessfulTxExecution(tx.From, result.ReadWriteAddresses)
+	txsToDelete := f.worker.UpdateAfterSingleSuccessfulTxExecution(txFrom, result.ReadWriteAddresses)
 	for _, txToDelete := range txsToDelete {
 		err := f.dbManager.UpdateTxStatus(ctx, txToDelete.Hash, pool.TxStatusFailed, false, txToDelete.FailedReason)
 		if err != nil {
 			log.Errorf("failed to update status to failed in the pool for tx: %s, err: %s", txToDelete.Hash.String(), err)
-		} else {
-			metrics.TxProcessed(metrics.TxProcessedLabelFailed, 1)
+			continue
 		}
+		metrics.TxProcessed(metrics.TxProcessedLabelFailed, 1)
 	}
 	metrics.WorkerProcessingTime(time.Since(start))
 }