Cherrypick/2642 2647 (0xPolygonHermez#2648)

* Add bookmarks to streamer (0xPolygonHermez#2642)

* update tools

* fix error

* update data streamer version

* refactor encoded

* refactor encoded

* gitignore

* fix tool (0xPolygonHermez#2647)

.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

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=

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

@@ -177,6 +177,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,
@@ -216,20 +226,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])
@@ -276,6 +286,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

@@ -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,25 @@ 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
 	}
+	tx, err := state.DecodeTx(string(encoded))
+	if err != nil {
+		return nil, err
+	}
+
+	binaryTxData, err := tx.MarshalBinary()
+	if err != nil {
+		return nil, err
+	}
+
+	l2Transaction.Encoded = binaryTxData
+	l2Transaction.EncodedLength = uint32(len(l2Transaction.Encoded))
 	l2Transaction.IsValid = 1
 	return &l2Transaction, nil
 }