Executor block by block

cmd/utils/flags.go

@@ -471,11 +471,6 @@ var (
 		Usage: "Batch seal time. Defaults to 12s",
 		Value: "12s",
 	}
-	SequencerNonEmptyBatchSealTime = cli.StringFlag{
-		Name:  "zkevm.sequencer-non-empty-batch-seal-time",
-		Usage: "Batch seal time. Defaults to 3s",
-		Value: "3s",
-	}
 	SequencerHaltOnBatchNumber = cli.Uint64Flag{
 		Name:  "zkevm.sequencer-halt-on-batch-number",
 		Usage: "Halt the sequencer on this batch number",

core/blockchain_zkevm.go

@@ -296,37 +296,6 @@ func PrepareBlockTxExecution(
 	return &blockContextImpl, excessDataGas, &blockGer, &blockL1BlockHash, nil
 }
 
-func FinalizeBlockExecutionWithHistoryWrite(
-	engine consensus.Engine, stateReader state.StateReader,
-	header *types.Header, txs types.Transactions, uncles []*types.Header,
-	stateWriter state.WriterWithChangeSets, cc *chain.Config,
-	ibs *state.IntraBlockState, receipts types.Receipts,
-	withdrawals []*types.Withdrawal, headerReader consensus.ChainHeaderReader,
-	isMining bool, excessDataGas *big.Int,
-) (newBlock *types.Block, newTxs types.Transactions, newReceipt types.Receipts, err error) {
-	newBlock, newTxs, newReceipt, err = FinalizeBlockExecution(
-		engine,
-		stateReader,
-		header,
-		txs,
-		uncles,
-		stateWriter,
-		cc,
-		ibs,
-		receipts,
-		withdrawals,
-		headerReader,
-		isMining,
-		excessDataGas,
-	)
-
-	if err := stateWriter.WriteHistory(); err != nil {
-		return nil, nil, nil, fmt.Errorf("writing history for block %d failed: %w", header.Number.Uint64(), err)
-	}
-
-	return newBlock, newTxs, newReceipt, nil
-}
-
 func CreateReceiptForBlockInfoTree(receipt *types.Receipt, chainConfig *chain.Config, blockNum uint64, execResult *ExecutionResult) *types.Receipt {
 	// [hack]TODO: remove this after bug is fixed
 	localReceipt := receipt.Clone()

core/vm/zk_batch_counters.go

@@ -234,7 +234,7 @@ func (bcc *BatchCounterCollector) CombineCollectors(verifyMerkleProof bool) (Cou
 // rlp level counters and execution level counters
 // this one returns the counters as they are so far, without adding processBatchLevelData, processChangeL2Block and decodeChangeL2BlockTx
 // used to save batch counter progress without adding the said counters twice
-func (bcc *BatchCounterCollector) CombineCollectorsNoChanges(verifyMerkleProof bool) Counters {
+func (bcc *BatchCounterCollector) CombineCollectorsNoChanges() Counters {
 	// combine all the counters we have so far
 
 	// if we have external coutners use them, otherwise create new

eth/backend.go

@@ -635,7 +635,19 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 
 	}()
 
+	tx, err := backend.chainDB.BeginRw(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
 	if !config.DeprecatedTxPool.Disable {
+		// we need to start the pool before stage loop itself
+		// the pool holds the info about how execution stage should work - as regular or as limbo recovery
+		if err := backend.txPool2.StartIfNotStarted(ctx, backend.txPool2DB, tx); err != nil {
+			return nil, err
+		}
+
 		backend.txPool2Fetch.ConnectCore()
 		backend.txPool2Fetch.ConnectSentries()
 		var newTxsBroadcaster *txpool2.NewSlotsStreams
@@ -696,12 +708,6 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 
 	backend.ethBackendRPC, backend.miningRPC, backend.stateChangesClient = ethBackendRPC, miningRPC, stateDiffClient
 
-	tx, err := backend.chainDB.BeginRw(ctx)
-	if err != nil {
-		return nil, err
-	}
-	defer tx.Rollback()
-
 	// create buckets
 	if err := createBuckets(tx); err != nil {
 		return nil, err
@@ -833,7 +839,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 				backend.engine,
 			)
 
-			var legacyExecutors []legacy_executor_verifier.ILegacyExecutor
+			var legacyExecutors []*legacy_executor_verifier.Executor = make([]*legacy_executor_verifier.Executor, 0, len(cfg.ExecutorUrls))
 			if len(cfg.ExecutorUrls) > 0 && cfg.ExecutorUrls[0] != "" {
 				levCfg := legacy_executor_verifier.Config{
 					GrpcUrls:              cfg.ExecutorUrls,
@@ -853,7 +859,6 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 				backend.chainConfig,
 				backend.chainDB,
 				witnessGenerator,
-				backend.l1Syncer,
 				backend.dataStream,
 			)
 
@@ -866,12 +871,6 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 			// we switch context from being an RPC node to a sequencer
 			backend.txPool2.ForceUpdateLatestBlock(executionProgress)
 
-			// we need to start the pool before stage loop itself
-			// the pool holds the info about how execution stage should work - as regular or as limbo recovery
-			if err := backend.txPool2.StartIfNotStarted(ctx, backend.txPool2DB, tx); err != nil {
-				return nil, err
-			}
-
 			l1BlockSyncer := syncer.NewL1Syncer(
 				ctx,
 				ethermanClients,
@@ -967,10 +966,6 @@ func createBuckets(tx kv.RwTx) error {
 		return err
 	}
 
-	if err := txpool.CreateTxPoolBuckets(tx); err != nil {
-		return err
-	}
-
 	return nil
 }
 

eth/ethconfig/config_zkevm.go

@@ -31,7 +31,6 @@ type Zk struct {
 	DatastreamVersion                      int
 	SequencerBlockSealTime                 time.Duration
 	SequencerBatchSealTime                 time.Duration
-	SequencerNonEmptyBatchSealTime         time.Duration
 	SequencerHaltOnBatchNumber             uint64
 	ExecutorUrls                           []string
 	ExecutorStrictMode                     bool

eth/stagedsync/stage_indexes_zkevm.go

@@ -0,0 +1,9 @@
+package stagedsync
+
+import (
+	"github.com/gateway-fm/cdk-erigon-lib/kv"
+)
+
+func PromoteHistory(logPrefix string, tx kv.RwTx, changesetBucket string, start, stop uint64, cfg HistoryCfg, quit <-chan struct{}) error {
+	return promoteHistory(logPrefix, tx, changesetBucket, start, stop, cfg, quit)
+}

eth/stagedsync/stages/stages_zk.go

@@ -28,7 +28,7 @@ var (
 	ForkId                      SyncStage = "ForkId"
 	L1SequencerSync             SyncStage = "L1SequencerSync"
 	L1InfoTree                  SyncStage = "L1InfoTree"
-	HighestUsedL1InfoIndex      SyncStage = "HighestUsedL1InfoTree"
-	SequenceExecutorVerify      SyncStage = "SequenceExecutorVerify"
-	L1BlockSync                 SyncStage = "L1BlockSync"
+	// HighestUsedL1InfoIndex      SyncStage = "HighestUsedL1InfoTree"
+	SequenceExecutorVerify SyncStage = "SequenceExecutorVerify"
+	L1BlockSync            SyncStage = "L1BlockSync"
 )

smt/pkg/db/mdbx.go

@@ -30,6 +30,8 @@ const TableAccountValues = "HermezSmtAccountValues"
 const TableMetadata = "HermezSmtMetadata"
 const TableHashKey = "HermezSmtHashKey"
 
+var HermezSmtTables = []string{TableSmt, TableStats, TableAccountValues, TableMetadata, TableHashKey}
+
 type EriDb struct {
 	kvTx kv.RwTx
 	tx   SmtDbTx

turbo/cli/default_flags.go

@@ -191,7 +191,6 @@ var DefaultFlags = []cli.Flag{
 	&utils.SmtRegenerateInMemory,
 	&utils.SequencerBlockSealTime,
 	&utils.SequencerBatchSealTime,
-	&utils.SequencerNonEmptyBatchSealTime,
 	&utils.SequencerHaltOnBatchNumber,
 	&utils.ExecutorUrls,
 	&utils.ExecutorStrictMode,

turbo/cli/flags_zkevm.go

@@ -76,12 +76,6 @@ func ApplyFlagsForZkConfig(ctx *cli.Context, cfg *ethconfig.Config) {
 		panic(fmt.Sprintf("could not parse sequencer batch seal time timeout value %s", sequencerBatchSealTimeVal))
 	}
 
-	sequencerNonEmptyBatchSealTimeVal := ctx.String(utils.SequencerNonEmptyBatchSealTime.Name)
-	sequencerNonEmptyBatchSealTime, err := time.ParseDuration(sequencerNonEmptyBatchSealTimeVal)
-	if err != nil {
-		panic(fmt.Sprintf("could not parse sequencer batch seal time timeout value %s", sequencerNonEmptyBatchSealTimeVal))
-	}
-
 	effectiveGasPriceForEthTransferVal := ctx.Float64(utils.EffectiveGasPriceForEthTransfer.Name)
 	effectiveGasPriceForErc20TransferVal := ctx.Float64(utils.EffectiveGasPriceForErc20Transfer.Name)
 	effectiveGasPriceForContractInvocationVal := ctx.Float64(utils.EffectiveGasPriceForContractInvocation.Name)
@@ -128,7 +122,6 @@ func ApplyFlagsForZkConfig(ctx *cli.Context, cfg *ethconfig.Config) {
 		SmtRegenerateInMemory:                  ctx.Bool(utils.SmtRegenerateInMemory.Name),
 		SequencerBlockSealTime:                 sequencerBlockSealTime,
 		SequencerBatchSealTime:                 sequencerBatchSealTime,
-		SequencerNonEmptyBatchSealTime:         sequencerNonEmptyBatchSealTime,
 		SequencerHaltOnBatchNumber:             ctx.Uint64(utils.SequencerHaltOnBatchNumber.Name),
 		ExecutorUrls:                           strings.Split(strings.ReplaceAll(ctx.String(utils.ExecutorUrls.Name), " ", ""), ","),
 		ExecutorStrictMode:                     ctx.Bool(utils.ExecutorStrictMode.Name),

turbo/stages/zk_stages.go

@@ -117,7 +117,6 @@ func NewSequencerZkStages(ctx context.Context,
 		zkStages.StageL1InfoTreeCfg(db, cfg.Zk, l1InfoTreeSyncer),
 		zkStages.StageSequencerL1BlockSyncCfg(db, cfg.Zk, l1BlockSyncer),
 		zkStages.StageDataStreamCatchupCfg(datastreamServer, db, cfg.Genesis.Config.ChainID.Uint64(), cfg.DatastreamVersion, cfg.HasExecutors()),
-		zkStages.StageSequencerInterhashesCfg(db, notifications.Accumulator),
 		zkStages.StageSequenceBlocksCfg(
 			db,
 			cfg.Prune,
@@ -139,11 +138,11 @@ func NewSequencerZkStages(ctx context.Context,
 			cfg.Zk,
 			txPool,
 			txPoolDb,
+			verifier,
 			uint16(cfg.YieldSize),
 		),
 		stagedsync.StageHashStateCfg(db, dirs, cfg.HistoryV3, agg),
 		zkStages.StageZkInterHashesCfg(db, true, true, false, dirs.Tmp, blockReader, controlServer.Hd, cfg.HistoryV3, agg, cfg.Zk),
-		zkStages.StageSequencerExecutorVerifyCfg(db, verifier, txPool, controlServer.ChainConfig, cfg.Zk),
 		stagedsync.StageHistoryCfg(db, cfg.Prune, dirs.Tmp),
 		stagedsync.StageLogIndexCfg(db, cfg.Prune, dirs.Tmp),
 		stagedsync.StageCallTracesCfg(db, cfg.Prune, 0, dirs.Tmp),

zk/datastream/server/data_stream_server.go

@@ -43,6 +43,7 @@ type DataStreamServer struct {
 	stream  *datastreamer.StreamServer
 	chainId uint64
 	highestBlockWritten,
+	highestClosedBatchWritten,
 	highestBatchWritten *uint64
 }
 
@@ -118,7 +119,33 @@ func NewDataStreamEntries(size int) *DataStreamEntries {
 	}
 }
 
-func (srv *DataStreamServer) CommitEntriesToStreamProto(entries []DataStreamEntryProto, latestBlockNum, latestBatchNum *uint64) error {
+func (srv *DataStreamServer) commitAtomicOp(latestBlockNum, latestBatchNum, latestClosedBatch *uint64) error {
+	if err := srv.stream.CommitAtomicOp(); err != nil {
+		return err
+	}
+
+	// copy the values in case they are changed outside the function
+	// pointers are used for easier check if we should set check them from the DS or not
+	// since 0 is a valid number, we can't use it
+	if latestBlockNum != nil {
+		a := *latestBlockNum
+		srv.highestBlockWritten = &a
+	}
+
+	if latestBatchNum != nil {
+		a := *latestBatchNum
+		srv.highestBatchWritten = &a
+	}
+
+	if latestClosedBatch != nil {
+		a := *latestClosedBatch
+		srv.highestClosedBatchWritten = &a
+	}
+
+	return nil
+}
+
+func (srv *DataStreamServer) commitEntriesToStreamProto(entries []DataStreamEntryProto) error {
 	for _, entry := range entries {
 		entryType := entry.Type()
 
@@ -137,16 +164,6 @@ func (srv *DataStreamServer) CommitEntriesToStreamProto(entries []DataStreamEntr
 			}
 		}
 	}
-
-	if latestBlockNum != nil {
-		a := *latestBlockNum
-		srv.highestBlockWritten = &a
-	}
-
-	if latestBatchNum != nil {
-		a := *latestBatchNum
-		srv.highestBatchWritten = &a
-	}
 	return nil
 }
 
@@ -454,24 +471,12 @@ func (srv *DataStreamServer) GetHighestBatchNumber() (uint64, error) {
 		return *srv.highestBatchWritten, nil
 	}
 
-	header := srv.stream.GetHeader()
-
-	if header.TotalEntries == 0 {
-		return 0, nil
+	entry, found, err := srv.getLastEntryOfType(datastreamer.EntryType(types.EntryTypeBatchStart))
+	if err != nil {
+		return 0, err
 	}
-
-	entryNum := header.TotalEntries - 1
-	var err error
-	var entry datastreamer.FileEntry
-	for {
-		entry, err = srv.stream.GetEntry(entryNum)
-		if err != nil {
-			return 0, err
-		}
-		if entry.Type == datastreamer.EntryType(1) {
-			break
-		}
-		entryNum -= 1
+	if !found {
+		return 0, nil
 	}
 
 	batch, err := types.UnmarshalBatchStart(entry.Data)
@@ -484,6 +489,28 @@ func (srv *DataStreamServer) GetHighestBatchNumber() (uint64, error) {
 	return batch.Number, nil
 }
 
+func (srv *DataStreamServer) GetHighestClosedBatch() (uint64, error) {
+	if srv.highestClosedBatchWritten != nil {
+		return *srv.highestClosedBatchWritten, nil
+	}
+	entry, found, err := srv.getLastEntryOfType(datastreamer.EntryType(types.EntryTypeBatchEnd))
+	if err != nil {
+		return 0, err
+	}
+	if !found {
+		return 0, nil
+	}
+
+	batch, err := types.UnmarshalBatchEnd(entry.Data)
+	if err != nil {
+		return 0, err
+	}
+
+	srv.highestClosedBatchWritten = &batch.Number
+
+	return batch.Number, nil
+}
+
 // must be done on offline server
 // finds the position of the block bookmark entry and deletes from it onward
 // blockNumber 10 would return the stream to before block 10 bookmark
@@ -523,3 +550,21 @@ func (srv *DataStreamServer) UnwindToBatchStart(batchNumber uint64) error {
 
 	return srv.stream.TruncateFile(entryNum)
 }
+
+func (srv *DataStreamServer) getLastEntryOfType(entryType datastreamer.EntryType) (datastreamer.FileEntry, bool, error) {
+	header := srv.stream.GetHeader()
+	emtryEntry := datastreamer.FileEntry{}
+
+	// loop will become infinite if using unsigned type
+	for entryNum := int64(header.TotalEntries - 1); entryNum >= 0; entryNum-- {
+		entry, err := srv.stream.GetEntry(uint64(entryNum))
+		if err != nil {
+			return emtryEntry, false, err
+		}
+		if entry.Type == entryType {
+			return entry, true, nil
+		}
+	}
+
+	return emtryEntry, false, nil
+}