More transactions into batches (#1214)

* do not seal a batch immediately on overflow

# Conflicts:
#	zk/stages/stage_sequence_execute.go
#	zk/stages/stage_sequence_execute_state.go

* only add tx to block state if no overflow was detected

* skip to the next transaction when finding initial OOC

core/vm/zk_batch_counters.go

@@ -88,6 +88,12 @@ func (bcc *BatchCounterCollector) AddNewTransactionCounters(txCounters *Transact
 	return bcc.CheckForOverflow(false) //no need to calculate the merkle proof here
 }
 
+func (bcc *BatchCounterCollector) RemovePreviousTransactionCounters() {
+	lastTx := bcc.transactions[len(bcc.transactions)-1]
+	bcc.UndoTransactionCountersCache(lastTx)
+	bcc.transactions = bcc.transactions[:len(bcc.transactions)-1]
+}
+
 func (bcc *BatchCounterCollector) ClearTransactionCounters() {
 	bcc.transactions = bcc.transactions[:0]
 }
@@ -154,7 +160,7 @@ func (bcc *BatchCounterCollector) CheckForOverflow(verifyMerkleProof bool) (bool
 		for _, v := range combined {
 			logText += fmt.Sprintf(" %s: initial: %v used: %v (remaining: %v)", v.name, v.initialAmount, v.used, v.remaining)
 		}
-		log.Info(logText)
+		log.Debug(logText)
 	}
 
 	return overflow, nil
@@ -275,3 +281,15 @@ func (bcc *BatchCounterCollector) UpdateExecutionAndProcessingCountersCache(txCo
 		bcc.processingCombinedCounters[k].used += v.used
 	}
 }
+
+func (bcc *BatchCounterCollector) UndoTransactionCountersCache(txCounters *TransactionCounter) {
+	for k, v := range txCounters.rlpCounters.counters {
+		bcc.rlpCombinedCounters[k].used -= v.used
+	}
+	for k, v := range txCounters.executionCounters.counters {
+		bcc.executionCombinedCounters[k].used -= v.used
+	}
+	for k, v := range txCounters.processingCounters.counters {
+		bcc.processingCombinedCounters[k].used -= v.used
+	}
+}

zk/stages/stage_sequence_execute.go

@@ -395,36 +395,46 @@ func sequencingStageStep(
 
 					switch anyOverflow {
 					case overflowCounters:
+						// remove the last attempted counters as we may want to continue processing this batch with other transactions
+						batchCounters.RemovePreviousTransactionCounters()
+
 						if batchState.isLimboRecovery() {
 							panic("limbo transaction has already been executed once so they must not overflow counters while re-executing")
 						}
 
 						if !batchState.isL1Recovery() {
-							log.Info(fmt.Sprintf("[%s] overflowed adding transaction to batch", logPrefix), "batch", batchState.batchNumber, "tx-hash", txHash, "has-any-transactions-in-this-batch", batchState.hasAnyTransactionsInThisBatch)
 							/*
 								There are two cases when overflow could occur.
-								1. The block DOES not contains any transactions.
+								1. The block DOES not contain any transactions.
 									In this case it means that a single tx overflow entire zk-counters.
 									In this case we mark it so. Once marked it will be discarded from the tx-pool async (once the tx-pool process the creation of a new batch)
+									Block production then continues as normal looking for more suitable transactions
 									NB: The tx SHOULD not be removed from yielded set, because if removed, it will be picked again on next block. That's why there is i++. It ensures that removing from yielded will start after the problematic tx
 								2. The block contains transactions.
-									In this case, we just have to remove the transaction that overflowed the zk-counters and all transactions after it, from the yielded set.
-									This removal will ensure that these transaction could be added in the next block(s)
+									In this case we make note that we have had a transaction that overflowed and continue attempting to process transactions
+									Once we reach the cap for these attempts we will stop producing blocks and consider the batch done
 							*/
 							if !batchState.hasAnyTransactionsInThisBatch {
+								// mark the transaction to be removed from the pool
 								cfg.txPool.MarkForDiscardFromPendingBest(txHash)
-								log.Info(fmt.Sprintf("single transaction %s overflow counters", txHash))
+								log.Info(fmt.Sprintf("[%s] single transaction %s cannot fit into batch", logPrefix, txHash))
+							} else {
+								batchState.newOverflowTransaction()
+								log.Info(fmt.Sprintf("[%s] transaction %s overflow counters", logPrefix, txHash), "count", batchState.overflowTransactions)
+								if batchState.reachedOverflowTransactionLimit() {
+									log.Info(fmt.Sprintf("[%s] closing batch due to counters", logPrefix), "count", batchState.overflowTransactions)
+									runLoopBlocks = false
+									break LOOP_TRANSACTIONS
+								}
 							}
 
-							runLoopBlocks = false
-							break LOOP_TRANSACTIONS
+							// continue on processing other transactions and skip this one
+							continue
 						}
 
 						if batchState.isResequence() && cfg.zk.SequencerResequenceStrict {
 							return fmt.Errorf("strict mode enabled, but resequenced batch %d overflowed counters on block %d", batchState.batchNumber, blockNumber)
 						}
-
-						break LOOP_TRANSACTIONS
 					case overflowGas:
 						if batchState.isAnyRecovery() {
 							panic(fmt.Sprintf("block gas limit overflow in recovery block: %d", blockNumber))

zk/stages/stage_sequence_execute_state.go

@@ -15,6 +15,8 @@ import (
 	"github.com/ledgerwatch/erigon/zk/txpool"
 )
 
+const maximumOverflowTransactionAttempts = 5
+
 type BatchContext struct {
 	ctx        context.Context
 	cfg        *SequenceBlockCfg
@@ -45,6 +47,7 @@ type BatchState struct {
 	batchL1RecoveryData           *BatchL1RecoveryData
 	limboRecoveryData             *LimboRecoveryData
 	resequenceBatchJob            *ResequenceBatchJob
+	overflowTransactions          int
 }
 
 func newBatchState(forkId, batchNumber, blockNumber uint64, hasExecutorForThisBatch, l1Recovery bool, txPool *txpool.TxPool, resequenceBatchJob *ResequenceBatchJob) *BatchState {
@@ -149,6 +152,14 @@ func (bs *BatchState) onBuiltBlock(blockNumber uint64) {
 	bs.builtBlocks = append(bs.builtBlocks, blockNumber)
 }
 
+func (bs *BatchState) newOverflowTransaction() {
+	bs.overflowTransactions++
+}
+
+func (bs *BatchState) reachedOverflowTransactionLimit() bool {
+	return bs.overflowTransactions >= maximumOverflowTransactionAttempts
+}
+
 // TYPE BATCH L1 RECOVERY DATA
 type BatchL1RecoveryData struct {
 	recoveredBatchDataSize int