WIP: data stream refactoring to testable code; add data stream catch up test; pending asserts

Author: tclemos

cmd/rpcdaemon/cli/config_zkevm.go

@@ -2,11 +2,12 @@ package cli
 
 import (
 	"fmt"
-	"github.com/0xPolygonHermez/zkevm-data-streamer/datastreamer"
+
+	"github.com/ledgerwatch/erigon/zk/datastream/server"
 	"github.com/ledgerwatch/log/v3"
 )
 
-func StartDataStream(server *datastreamer.StreamServer) error {
+func StartDataStream(server server.StreamServer) error {
 	if server == nil {
 		// no stream server to start, we might not have the right flags set to create one
 		return nil

eth/backend.go

@@ -129,8 +129,10 @@ import (
 	"github.com/ledgerwatch/erigon/turbo/stages/headerdownload"
 	"github.com/ledgerwatch/erigon/zk/contracts"
 	"github.com/ledgerwatch/erigon/zk/datastream/client"
+	"github.com/ledgerwatch/erigon/zk/datastream/server"
 	"github.com/ledgerwatch/erigon/zk/hermez_db"
 	"github.com/ledgerwatch/erigon/zk/l1_cache"
+	"github.com/ledgerwatch/erigon/zk/l1infotree"
 	"github.com/ledgerwatch/erigon/zk/legacy_executor_verifier"
 	zkStages "github.com/ledgerwatch/erigon/zk/stages"
 	"github.com/ledgerwatch/erigon/zk/syncer"
@@ -139,9 +141,10 @@ import (
 	"github.com/ledgerwatch/erigon/zk/utils"
 	"github.com/ledgerwatch/erigon/zk/witness"
 	"github.com/ledgerwatch/erigon/zkevm/etherman"
-	"github.com/ledgerwatch/erigon/zk/l1infotree"
 )
 
+var dataStreamServerFactory = server.NewZkEVMDataStreamServerFactory()
+
 // Config contains the configuration options of the ETH protocol.
 // Deprecated: use ethconfig.Config instead.
 type Config = ethconfig.Config
@@ -219,7 +222,7 @@ type Ethereum struct {
 	logger         log.Logger
 
 	// zk
-	dataStream      *datastreamer.StreamServer
+	streamServer    server.StreamServer
 	l1Syncer        *syncer.L1Syncer
 	etherManClients []*etherman.Client
 	l1Cache         *l1_cache.L1Cache
@@ -977,16 +980,17 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 				Level:       "warn",
 				Outputs:     nil,
 			}
+
 			// todo [zkevm] read the stream version from config and figure out what system id is used for
-			backend.dataStream, err = datastreamer.NewServer(uint16(httpCfg.DataStreamPort), uint8(backend.config.DatastreamVersion), 1, datastreamer.StreamType(1), file, httpCfg.DataStreamWriteTimeout, httpCfg.DataStreamInactivityTimeout, httpCfg.DataStreamInactivityCheckInterval, logConfig)
+			backend.streamServer, err = dataStreamServerFactory.CreateStreamServer(uint16(httpCfg.DataStreamPort), uint8(backend.config.DatastreamVersion), 1, datastreamer.StreamType(1), file, httpCfg.DataStreamWriteTimeout, httpCfg.DataStreamInactivityTimeout, httpCfg.DataStreamInactivityCheckInterval, logConfig)
 			if err != nil {
 				return nil, err
 			}
 
 			// recovery here now, if the stream got into a bad state we want to be able to delete the file and have
 			// the stream re-populated from scratch.  So we check the stream for the latest header and if it is
 			// 0 we can just set the datastream progress to 0 also which will force a re-population of the stream
-			latestHeader := backend.dataStream.GetHeader()
+			latestHeader := backend.streamServer.GetHeader()
 			if latestHeader.TotalEntries == 0 {
 				log.Info("[dataStream] setting the stream progress to 0")
 				backend.preStartTasks.WarmUpDataStream = true
@@ -1100,6 +1104,11 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 
 		l1InfoTreeUpdater := l1infotree.NewUpdater(cfg.Zk, l1InfoTreeSyncer)
 
+		var dataStreamServer server.DataStreamServer
+		if backend.streamServer != nil {
+			dataStreamServer = dataStreamServerFactory.CreateDataStreamServer(backend.streamServer, backend.chainConfig.ChainID.Uint64())
+		}
+
 		if isSequencer {
 			// if we are sequencing transactions, we do the sequencing loop...
 			witnessGenerator := witness.NewGenerator(
@@ -1129,10 +1138,9 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 			verifier := legacy_executor_verifier.NewLegacyExecutorVerifier(
 				*cfg.Zk,
 				legacyExecutors,
-				backend.chainConfig,
 				backend.chainDB,
 				witnessGenerator,
-				backend.dataStream,
+				dataStreamServer,
 			)
 
 			if cfg.Zk.Limbo {
@@ -1167,7 +1175,7 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 				backend.agg,
 				backend.forkValidator,
 				backend.engine,
-				backend.dataStream,
+				dataStreamServer,
 				backend.l1Syncer,
 				seqVerSyncer,
 				l1BlockSyncer,
@@ -1209,7 +1217,7 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 				backend.engine,
 				backend.l1Syncer,
 				streamClient,
-				backend.dataStream,
+				dataStreamServer,
 				l1InfoTreeUpdater,
 			)
 
@@ -1330,7 +1338,11 @@ func (s *Ethereum) Init(stack *node.Node, config *ethconfig.Config, chainConfig
 	// apiList := jsonrpc.APIList(chainKv, borDb, ethRpcClient, txPoolRpcClient, miningRpcClient, ff, stateCache, blockReader, backend.agg, httpRpcCfg, backend.engine, config, backend.l1Syncer)
 	// authApiList := jsonrpc.AuthAPIList(chainKv, ethRpcClient, txPoolRpcClient, miningRpcClient, ff, stateCache, blockReader, backend.agg, httpRpcCfg, backend.engine, config)
 
-	s.apiList = jsonrpc.APIList(chainKv, ethRpcClient, txPoolRpcClient, s.txPool2, miningRpcClient, ff, stateCache, blockReader, s.agg, &httpRpcCfg, s.engine, config, s.l1Syncer, s.logger, s.dataStream)
+	var dataStreamServer server.DataStreamServer
+	if s.streamServer != nil {
+		dataStreamServer = dataStreamServerFactory.CreateDataStreamServer(s.streamServer, config.Zk.L2ChainId)
+	}
+	s.apiList = jsonrpc.APIList(chainKv, ethRpcClient, txPoolRpcClient, s.txPool2, miningRpcClient, ff, stateCache, blockReader, s.agg, &httpRpcCfg, s.engine, config, s.l1Syncer, s.logger, dataStreamServer)
 
 	if config.SilkwormRpcDaemon && httpRpcCfg.Enabled {
 		interface_log_settings := silkworm.RpcInterfaceLogSettings{
@@ -1368,7 +1380,7 @@ func (s *Ethereum) Init(stack *node.Node, config *ethconfig.Config, chainConfig
 	}
 
 	go func() {
-		if err := cli.StartDataStream(s.dataStream); err != nil {
+		if err := cli.StartDataStream(s.streamServer); err != nil {
 			log.Error(err.Error())
 			return
 		}
@@ -1391,8 +1403,9 @@ func (s *Ethereum) PreStart() error {
 		// we don't know when the server has actually started as it doesn't expose a signal that is has spun up
 		// so here we loop and take a brief pause waiting for it to be ready
 		attempts := 0
+		dataStreamServer := dataStreamServerFactory.CreateDataStreamServer(s.streamServer, s.chainConfig.ChainID.Uint64())
 		for {
-			_, err = zkStages.CatchupDatastream(s.sentryCtx, "stream-catchup", tx, s.dataStream, s.chainConfig.ChainID.Uint64())
+			_, err = zkStages.CatchupDatastream(s.sentryCtx, "stream-catchup", tx, dataStreamServer)
 			if err != nil {
 				if errors.Is(err, datastreamer.ErrAtomicOpNotAllowed) {
 					attempts++

turbo/jsonrpc/daemon.go

@@ -17,18 +17,17 @@ import (
 	"github.com/ledgerwatch/erigon/rpc"
 	"github.com/ledgerwatch/erigon/turbo/rpchelper"
 	"github.com/ledgerwatch/erigon/turbo/services"
+	"github.com/ledgerwatch/erigon/zk/datastream/server"
 	"github.com/ledgerwatch/erigon/zk/sequencer"
 	"github.com/ledgerwatch/erigon/zk/syncer"
-
 	txpool2 "github.com/ledgerwatch/erigon/zk/txpool"
-	"github.com/0xPolygonHermez/zkevm-data-streamer/datastreamer"
 )
 
 // APIList describes the list of available RPC apis
 func APIList(db kv.RoDB, eth rpchelper.ApiBackend, txPool txpool.TxpoolClient, rawPool *txpool2.TxPool, mining txpool.MiningClient,
 	filters *rpchelper.Filters, stateCache kvcache.Cache,
 	blockReader services.FullBlockReader, agg *libstate.Aggregator, cfg *httpcfg.HttpCfg, engine consensus.EngineReader,
-	ethCfg *ethconfig.Config, l1Syncer *syncer.L1Syncer, logger log.Logger, datastreamServer *datastreamer.StreamServer,
+	ethCfg *ethconfig.Config, l1Syncer *syncer.L1Syncer, logger log.Logger, dataStreamServer server.DataStreamServer,
 ) (list []rpc.API) {
 	// non-sequencer nodes should forward on requests to the sequencer
 	rpcUrl := ""
@@ -69,7 +68,7 @@ func APIList(db kv.RoDB, eth rpchelper.ApiBackend, txPool txpool.TxpoolClient, r
 	otsImpl := NewOtterscanAPI(base, db, cfg.OtsMaxPageSize)
 	gqlImpl := NewGraphQLAPI(base, db)
 	overlayImpl := NewOverlayAPI(base, db, cfg.Gascap, cfg.OverlayGetLogsTimeout, cfg.OverlayReplayBlockTimeout, otsImpl)
-	zkEvmImpl := NewZkEvmAPI(ethImpl, db, cfg.ReturnDataLimit, ethCfg, l1Syncer, rpcUrl, datastreamServer)
+	zkEvmImpl := NewZkEvmAPI(ethImpl, db, cfg.ReturnDataLimit, ethCfg, l1Syncer, rpcUrl, dataStreamServer)
 
 	if cfg.GraphQLEnabled {
 		list = append(list, rpc.API{

turbo/jsonrpc/zkevm_api.go

@@ -16,7 +16,6 @@ import (
 
 	zktypes "github.com/ledgerwatch/erigon/zk/types"
 
-	"github.com/0xPolygonHermez/zkevm-data-streamer/datastreamer"
 	"github.com/holiman/uint256"
 	"github.com/ledgerwatch/erigon-lib/common/hexutil"
 	"github.com/ledgerwatch/erigon-lib/kv/membatchwithdb"
@@ -92,7 +91,7 @@ type ZkEvmAPIImpl struct {
 	l1Syncer         *syncer.L1Syncer
 	l2SequencerUrl   string
 	semaphores       map[string]chan struct{}
-	datastreamServer *server.DataStreamServer
+	datastreamServer server.DataStreamServer
 }
 
 func (api *ZkEvmAPIImpl) initializeSemaphores(functionLimits map[string]int) {
@@ -113,22 +112,17 @@ func NewZkEvmAPI(
 	zkConfig *ethconfig.Config,
 	l1Syncer *syncer.L1Syncer,
 	l2SequencerUrl string,
-	datastreamServer *datastreamer.StreamServer,
+	dataStreamServer server.DataStreamServer,
 ) *ZkEvmAPIImpl {
 
-	var streamServer *server.DataStreamServer
-	if datastreamServer != nil {
-		streamServer = server.NewDataStreamServer(datastreamServer, zkConfig.Zk.L2ChainId)
-	}
-
 	a := &ZkEvmAPIImpl{
 		ethApi:           base,
 		db:               db,
 		ReturnDataLimit:  returnDataLimit,
 		config:           zkConfig,
 		l1Syncer:         l1Syncer,
 		l2SequencerUrl:   l2SequencerUrl,
-		datastreamServer: streamServer,
+		datastreamServer: dataStreamServer,
 	}
 
 	a.initializeSemaphores(map[string]int{

turbo/stages/zk_stages.go

@@ -3,7 +3,6 @@ package stages
 import (
 	"context"
 
-	"github.com/0xPolygonHermez/zkevm-data-streamer/datastreamer"
 	proto_downloader "github.com/ledgerwatch/erigon-lib/gointerfaces/downloader"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon-lib/state"
@@ -16,11 +15,12 @@ import (
 	"github.com/ledgerwatch/erigon/turbo/engineapi/engine_helpers"
 	"github.com/ledgerwatch/erigon/turbo/shards"
 	"github.com/ledgerwatch/erigon/turbo/snapshotsync/freezeblocks"
+	"github.com/ledgerwatch/erigon/zk/datastream/server"
+	"github.com/ledgerwatch/erigon/zk/l1infotree"
 	"github.com/ledgerwatch/erigon/zk/legacy_executor_verifier"
 	zkStages "github.com/ledgerwatch/erigon/zk/stages"
 	"github.com/ledgerwatch/erigon/zk/syncer"
 	"github.com/ledgerwatch/erigon/zk/txpool"
-	"github.com/ledgerwatch/erigon/zk/l1infotree"
 )
 
 // NewDefaultZkStages creates stages for zk syncer (RPC mode)
@@ -36,7 +36,7 @@ func NewDefaultZkStages(ctx context.Context,
 	engine consensus.Engine,
 	l1Syncer *syncer.L1Syncer,
 	datastreamClient zkStages.DatastreamClient,
-	datastreamServer *datastreamer.StreamServer,
+	dataStreamServer server.DataStreamServer,
 	infoTreeUpdater *l1infotree.Updater,
 ) []*stagedsync.Stage {
 	dirs := cfg.Dirs
@@ -54,7 +54,7 @@ func NewDefaultZkStages(ctx context.Context,
 		zkStages.StageL1SyncerCfg(db, l1Syncer, cfg.Zk),
 		zkStages.StageL1InfoTreeCfg(db, cfg.Zk, infoTreeUpdater),
 		zkStages.StageBatchesCfg(db, datastreamClient, cfg.Zk, controlServer.ChainConfig, &cfg.Miner),
-		zkStages.StageDataStreamCatchupCfg(datastreamServer, db, cfg.Genesis.Config.ChainID.Uint64(), cfg.DatastreamVersion, cfg.HasExecutors()),
+		zkStages.StageDataStreamCatchupCfg(dataStreamServer, db, cfg.Genesis.Config.ChainID.Uint64(), cfg.DatastreamVersion, cfg.HasExecutors()),
 		stagedsync.StageBlockHashesCfg(db, dirs.Tmp, controlServer.ChainConfig, blockWriter),
 		stagedsync.StageSendersCfg(db, controlServer.ChainConfig, false, dirs.Tmp, cfg.Prune, blockReader, controlServer.Hd, nil),
 		stagedsync.StageExecuteBlocksCfg(
@@ -99,7 +99,7 @@ func NewSequencerZkStages(ctx context.Context,
 	agg *state.Aggregator,
 	forkValidator *engine_helpers.ForkValidator,
 	engine consensus.Engine,
-	datastreamServer *datastreamer.StreamServer,
+	dataStreamServer server.DataStreamServer,
 	sequencerStageSyncer *syncer.L1Syncer,
 	l1Syncer *syncer.L1Syncer,
 	l1BlockSyncer *syncer.L1Syncer,
@@ -120,7 +120,7 @@ func NewSequencerZkStages(ctx context.Context,
 		zkStages.StageL1SequencerSyncCfg(db, cfg.Zk, sequencerStageSyncer),
 		zkStages.StageL1InfoTreeCfg(db, cfg.Zk, infoTreeUpdater),
 		zkStages.StageSequencerL1BlockSyncCfg(db, cfg.Zk, l1BlockSyncer),
-		zkStages.StageDataStreamCatchupCfg(datastreamServer, db, cfg.Genesis.Config.ChainID.Uint64(), cfg.DatastreamVersion, cfg.HasExecutors()),
+		zkStages.StageDataStreamCatchupCfg(dataStreamServer, db, cfg.Genesis.Config.ChainID.Uint64(), cfg.DatastreamVersion, cfg.HasExecutors()),
 		zkStages.StageSequenceBlocksCfg(
 			db,
 			cfg.Prune,
@@ -138,7 +138,7 @@ func NewSequencerZkStages(ctx context.Context,
 			cfg.Genesis,
 			cfg.Sync,
 			agg,
-			datastreamServer,
+			dataStreamServer,
 			cfg.Zk,
 			&cfg.Miner,
 			txPool,