Add bookmarks to streamer (#2642)

* update tools

* fix error

* update data streamer version

* refactor encoded

* refactor encoded

* gitignore

.gitignore

@@ -11,7 +11,9 @@
 /test/contracts/bin/**/*.abi
 
 /tools/datastreamer/*.bin
-/test/datastream.bin
+/test/datastreamer/*.db/*
+/test/*.bin
+/test/*.db/*
 
 **/.DS_Store
 .vscode

go.mod

@@ -3,7 +3,7 @@ module github.com/0xPolygonHermez/zkevm-node
 go 1.19
 
 require (
-	github.com/0xPolygonHermez/zkevm-data-streamer v0.0.8
+	github.com/0xPolygonHermez/zkevm-data-streamer v0.0.10
 	github.com/didip/tollbooth/v6 v6.1.2
 	github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127
 	github.com/ethereum/go-ethereum v1.13.2
@@ -58,7 +58,7 @@ require (
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/crate-crypto/go-kzg-4844 v0.3.0 // indirect
 	github.com/cyphar/filepath-securejoin v0.2.4 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/deckarep/golang-set/v2 v2.1.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 // indirect
 	github.com/dlclark/regexp2 v1.7.0 // indirect
@@ -112,7 +112,7 @@ require (
 	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
 	github.com/pjbgf/sha1cd v0.3.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/procfs v0.10.1 // indirect
 	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect

go.sum

@@ -42,8 +42,8 @@ cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3f
 dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
 dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
-github.com/0xPolygonHermez/zkevm-data-streamer v0.0.8 h1:hOByFEvUC8hJnfbINMFzXxBru07AQLEhN50afow6Eu8=
-github.com/0xPolygonHermez/zkevm-data-streamer v0.0.8/go.mod h1:UqLxA+/R20fm63Mp+J7wYMfh6WoE+6vBj6rOmFGuRm4=
+github.com/0xPolygonHermez/zkevm-data-streamer v0.0.10 h1:BiWU1twGWg8j9KXOCGitUZ264h2XLKRFPVuQ3gWnmEA=
+github.com/0xPolygonHermez/zkevm-data-streamer v0.0.10/go.mod h1:gAHakobPs4k6Ku0VeAhbGo4T0OaNswXBkPp/2ff4rXo=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
@@ -145,8 +145,9 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/cyphar/filepath-securejoin v0.2.4 h1:Ugdm7cg7i6ZK6x3xDF1oEu1nfkyfH53EtKeQYTC3kyg=
 github.com/cyphar/filepath-securejoin v0.2.4/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/deckarep/golang-set/v2 v2.1.0 h1:g47V4Or+DUdzbs8FxCCmgb6VYd+ptPAngjM6dtGktsI=
 github.com/deckarep/golang-set/v2 v2.1.0/go.mod h1:VAky9rY/yGXJOLEDv3OMci+7wtDpOF4IN+y82NBOac4=
 github.com/decred/dcrd/crypto/blake256 v1.0.0 h1:/8DMNYp9SGi5f0w7uCm6d6M4OU2rGFK09Y2A4Xv7EE0=
@@ -599,8 +600,9 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
 github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
 github.com/poy/onpar v1.1.2 h1:QaNrNiZx0+Nar5dLgTVp5mXkyoVFIbepjyEoGSnhbAY=
 github.com/prometheus/client_golang v1.16.0 h1:yk/hx9hDbrGHovbci4BY+pRMfSuuat626eFsHb7tmT8=

sequencer/dbmanager.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/0xPolygonHermez/zkevm-data-streamer/datastreamer"
-	"github.com/0xPolygonHermez/zkevm-node/hex"
 	"github.com/0xPolygonHermez/zkevm-node/log"
 	"github.com/0xPolygonHermez/zkevm-node/pool"
 	"github.com/0xPolygonHermez/zkevm-node/state"
@@ -175,6 +174,17 @@ func (d *dbManager) sendDataToStreamer() {
 				continue
 			}
 
+			bookMark := state.DSBookMark{
+				Type:          state.BookMarkTypeL2Block,
+				L2BlockNumber: l2Block.L2BlockNumber,
+			}
+
+			_, err = d.streamServer.AddStreamBookmark(bookMark.Encode())
+			if err != nil {
+				log.Errorf("failed to add stream bookmark for l2block %v: %v", l2Block.L2BlockNumber, err)
+				continue
+			}
+
 			blockStart := state.DSL2BlockStart{
 				BatchNumber:    l2Block.BatchNumber,
 				L2BlockNumber:  l2Block.L2BlockNumber,
@@ -324,13 +334,12 @@ func (d *dbManager) StoreProcessedTxAndDeleteFromPool(ctx context.Context, tx tr
 		if err != nil {
 			return err
 		}
-		encoded := hex.EncodeToHex(binaryTxData)
 
 		l2Transaction := state.DSL2Transaction{
 			EffectiveGasPricePercentage: uint8(tx.response.EffectivePercentage),
 			IsValid:                     1,
 			EncodedLength:               uint32(len(binaryTxData)),
-			Encoded:                     []byte(encoded),
+			Encoded:                     binaryTxData,
 		}
 
 		d.dataToStream <- state.DSL2FullBlock{

sequencer/sequencer.go

@@ -202,6 +202,16 @@ func (s *Sequencer) updateDataStreamerFile(ctx context.Context, streamServer *da
 			log.Fatal(err)
 		}
 
+		bookMark := state.DSBookMark{
+			Type:          state.BookMarkTypeL2Block,
+			L2BlockNumber: genesisL2Block.L2BlockNumber,
+		}
+
+		_, err = streamServer.AddStreamBookmark(bookMark.Encode())
+		if err != nil {
+			log.Fatal(err)
+		}
+
 		genesisBlock := state.DSL2BlockStart{
 			BatchNumber:    genesisL2Block.BatchNumber,
 			L2BlockNumber:  genesisL2Block.L2BlockNumber,
@@ -241,20 +251,20 @@ func (s *Sequencer) updateDataStreamerFile(ctx context.Context, streamServer *da
 
 		log.Infof("Latest entry: %+v", latestEntry)
 
-		switch latestEntry.EntryType {
+		switch latestEntry.Type {
 		case state.EntryTypeL2BlockStart:
 			log.Info("Latest entry type is L2BlockStart")
 			currentL2Block = binary.LittleEndian.Uint64(latestEntry.Data[8:16])
 		case state.EntryTypeL2Tx:
 			log.Info("Latest entry type is L2Tx")
-			for latestEntry.EntryType == state.EntryTypeL2Tx {
+			for latestEntry.Type == state.EntryTypeL2Tx {
 				currentTxIndex++
 				latestEntry, err = streamServer.GetEntry(header.TotalEntries - currentTxIndex)
 				if err != nil {
 					log.Fatal(err)
 				}
 			}
-			if latestEntry.EntryType != state.EntryTypeL2BlockStart {
+			if latestEntry.Type != state.EntryTypeL2BlockStart {
 				log.Fatal("Latest entry is not a L2BlockStart")
 			}
 			currentL2Block = binary.LittleEndian.Uint64(latestEntry.Data[8:16])
@@ -301,6 +311,16 @@ func (s *Sequencer) updateDataStreamerFile(ctx context.Context, streamServer *da
 				currentTxIndex = 0
 			}
 
+			bookMark := state.DSBookMark{
+				Type:          state.BookMarkTypeL2Block,
+				L2BlockNumber: l2block.L2BlockNumber,
+			}
+
+			_, err = streamServer.AddStreamBookmark(bookMark.Encode())
+			if err != nil {
+				log.Fatal(err)
+			}
+
 			blockStart := state.DSL2BlockStart{
 				BatchNumber:    l2block.BatchNumber,
 				L2BlockNumber:  l2block.L2BlockNumber,

state/datastream.go

@@ -10,12 +10,16 @@ import (
 const (
 	// StreamTypeSequencer represents a Sequencer stream
 	StreamTypeSequencer datastreamer.StreamType = 1
+	// EntryTypeBookMark represents a bookmark entry
+	EntryTypeBookMark datastreamer.EntryType = datastreamer.EtBookmark
 	// EntryTypeL2BlockStart represents a L2 block start
 	EntryTypeL2BlockStart datastreamer.EntryType = 1
 	// EntryTypeL2Tx represents a L2 transaction
 	EntryTypeL2Tx datastreamer.EntryType = 2
 	// EntryTypeL2BlockEnd represents a L2 block end
 	EntryTypeL2BlockEnd datastreamer.EntryType = 3
+	// BookMarkTypeL2Block represents a L2 block bookmark
+	BookMarkTypeL2Block byte = 0
 )
 
 // DSL2FullBlock represents a data stream L2 full block and its transactions
@@ -87,7 +91,21 @@ type DSL2BlockEnd struct {
 func (b DSL2BlockEnd) Encode() []byte {
 	bytes := make([]byte, 0)
 	bytes = binary.LittleEndian.AppendUint64(bytes, b.L2BlockNumber)
-	bytes = append(bytes, b.BlockHash.Bytes()...)
-	bytes = append(bytes, b.StateRoot.Bytes()...)
+	bytes = append(bytes, b.BlockHash[:]...)
+	bytes = append(bytes, b.StateRoot[:]...)
+	return bytes
+}
+
+// DSBookMark represents a data stream bookmark
+type DSBookMark struct {
+	Type          byte
+	L2BlockNumber uint64
+}
+
+// Encode returns the encoded DSBookMark as a byte slice
+func (b DSBookMark) Encode() []byte {
+	bytes := make([]byte, 0)
+	bytes = append(bytes, b.Type)
+	bytes = binary.LittleEndian.AppendUint64(bytes, b.L2BlockNumber)
 	return bytes
 }

tools/datastreamer/Makefile

@@ -6,18 +6,28 @@ check-go:
 
 # Targets that require the checks
 generate-file: check-go
+rebuild: check-go
+decode-entry: check-go
+adecode-l2block: check-go
 
 arguments := $(wordlist 2,$(words $(MAKECMDGOALS)),$(MAKECMDGOALS))
 
 .PHONY: generate-file
 generate-file: ## Runs the tool to populate the binary file
 	go run main.go generate -cfg config/tool.config.toml
 
-rebuild: ## Runs the tool to rebuild state roots from a given block
-	go run main.go rebuild -cfg config/tool.config.toml
+.PHONY: decode-l2block
+decode-l2block: ## Runs the tool to decode a given l2 block number
+	go run main.go decode-l2block -cfg config/tool.config.toml -l2block $(arguments)
 
+.PHONY: decode-entry
 decode-entry: ## Runs the tool to decode a given entry number
-	go run main.go decode -cfg config/tool.config.toml -entry $(arguments)
+	go run main.go decode-entry -cfg config/tool.config.toml -entry $(arguments)
+
+.PHONY: rebuild
+rebuild: ## Runs the tool to rebuild state roots from a given block
+	go run main.go rebuild -cfg config/tool.config.toml -l2block $(arguments)
+
 
 ## Help display.
 ## Pulls comments from beside commands and prints a nicely formatted

tools/datastreamer/config/tool.config.toml

@@ -4,7 +4,7 @@ Port = 6901
 Filename = "datastream.bin"
     [StreamServer.Log]
     Environment = "development"
-    Level = "info"
+    Level = "debug"
     Outputs = ["stdout"]
 
 [StateDB]
@@ -13,7 +13,7 @@ Password = "state_password"
 Name = "state_db"
 Host = "localhost"
 Port = "5432"
-EnableLog = false	
+EnableLog = false
 MaxConns = 200
 
 [Executor]

tools/datastreamer/db/db.go

@@ -116,7 +116,7 @@ func scanL2Block(row pgx.Row) (*state.DSL2Block, error) {
 
 // GetL2Transactions returns the L2 transactions
 func (db *StateDB) GetL2Transactions(ctx context.Context, minL2Block, maxL2Block uint64) ([]*state.DSL2Transaction, error) {
-	const l2TxSQL = `SELECT t.effective_percentage, LENGTH(t.encoded), t.encoded
+	const l2TxSQL = `SELECT t.effective_percentage, t.encoded
 					 FROM state.transaction t
 					 WHERE l2_block_num BETWEEN $1 AND $2
 					 ORDER BY t.l2_block_num ASC`
@@ -142,13 +142,15 @@ func (db *StateDB) GetL2Transactions(ctx context.Context, minL2Block, maxL2Block
 
 func scanL2Transaction(row pgx.Row) (*state.DSL2Transaction, error) {
 	l2Transaction := state.DSL2Transaction{}
+	encoded := []byte{}
 	if err := row.Scan(
 		&l2Transaction.EffectiveGasPricePercentage,
-		&l2Transaction.EncodedLength,
-		&l2Transaction.Encoded,
+		&encoded,
 	); err != nil {
-		return &l2Transaction, err
+		return nil, err
 	}
+	l2Transaction.Encoded = common.Hex2Bytes(string(encoded))
+	l2Transaction.EncodedLength = uint32(len(l2Transaction.Encoded))
 	l2Transaction.IsValid = 1
 	return &l2Transaction, nil
 }