Revert "Change data stream format (#3597)" (#3603)

This reverts commit 5a76fb5.

Author: ToniRamirezM

Makefile

@@ -164,7 +164,6 @@ generate-code-from-proto: ## Generates code from proto files
 	cd proto/src/proto/hashdb/v1 && protoc --proto_path=. --proto_path=../../../../include --go_out=../../../../../merkletree/hashdb --go-grpc_out=../../../../../merkletree/hashdb --go_opt=paths=source_relative --go-grpc_opt=paths=source_relative hashdb.proto
 	cd proto/src/proto/executor/v1 && protoc --proto_path=. --go_out=../../../../../state/runtime/executor --go-grpc_out=../../../../../state/runtime/executor --go-grpc_opt=paths=source_relative --go_opt=paths=source_relative executor.proto
 	cd proto/src/proto/aggregator/v1 && protoc --proto_path=. --proto_path=../../../../include --go_out=../../../../../aggregator/prover --go-grpc_out=../../../../../aggregator/prover --go-grpc_opt=paths=source_relative --go_opt=paths=source_relative aggregator.proto
-	cd proto/src/proto/datastream/v1 && protoc --proto_path=. --proto_path=../../../../include --go_out=../../../../../state/datastream --go-grpc_out=../../../../../state/datastream --go-grpc_opt=paths=source_relative --go_opt=paths=source_relative datastream.proto
 
 ## Help display.
 ## Pulls comments from beside commands and prints a nicely formatted

proto/src/proto/datastream/v1/datastream.proto

@@ -26,7 +26,6 @@ type Batch struct {
 	imRemainingResources    state.BatchResources // remaining batch resources when processing tx-by-tx
 	finalRemainingResources state.BatchResources // remaining batch resources when a L2 block is processed
 	closingReason           state.ClosingReason
-	finalLocalExitRoot      common.Hash
 }
 
 func (w *Batch) isEmpty() bool {
@@ -95,7 +94,6 @@ func (f *finalizer) setWIPBatch(ctx context.Context, wipStateBatch *state.Batch)
 		countOfTxs:              wipStateBatchCountOfTxs,
 		imRemainingResources:    remainingResources,
 		finalRemainingResources: remainingResources,
-		finalLocalExitRoot:      wipStateBatch.LocalExitRoot,
 	}
 
 	return wipBatch, nil
@@ -295,7 +293,6 @@ func (f *finalizer) openNewWIPBatch(ctx context.Context, batchNumber uint64, sta
 		imRemainingResources:    maxRemainingResources,
 		finalRemainingResources: maxRemainingResources,
 		closingReason:           state.EmptyClosingReason,
-		finalLocalExitRoot:      newStateBatch.LocalExitRoot,
 	}, err
 }
 
@@ -331,9 +328,6 @@ func (f *finalizer) closeWIPBatch(ctx context.Context) error {
 			log.Errorf("error committing close wip batch, error: %v", err)
 			return err
 		}
-
-		// Sent batch to DS
-		f.DSSendBatch(f.wipBatch.batchNumber, f.wipBatch.finalStateRoot, f.wipBatch.finalLocalExitRoot)
 	}
 
 	return nil

sequencer/batch.go

@@ -3,25 +3,22 @@ package sequencer
 import (
 	"github.com/0xPolygonHermez/zkevm-node/log"
 	"github.com/0xPolygonHermez/zkevm-node/state"
-	"github.com/0xPolygonHermez/zkevm-node/state/datastream"
-	"github.com/ethereum/go-ethereum/common"
 )
 
-func (f *finalizer) DSSendL2Block(batchNumber uint64, blockResponse *state.ProcessBlockResponse, l1InfoTreeIndex uint32, minTimestamp uint64) error {
+func (f *finalizer) DSSendL2Block(batchNumber uint64, blockResponse *state.ProcessBlockResponse, l1InfoTreeIndex uint32) error {
 	forkID := f.stateIntf.GetForkIDByBatchNumber(batchNumber)
 
 	// Send data to streamer
 	if f.streamServer != nil {
 		l2Block := state.DSL2Block{
 			BatchNumber:     batchNumber,
 			L2BlockNumber:   blockResponse.BlockNumber,
-			Timestamp:       blockResponse.Timestamp,
-			Min_timestamp:   minTimestamp,
+			Timestamp:       int64(blockResponse.Timestamp),
 			L1InfoTreeIndex: l1InfoTreeIndex,
 			L1BlockHash:     blockResponse.BlockHashL1,
 			GlobalExitRoot:  blockResponse.GlobalExitRoot,
 			Coinbase:        f.sequencerAddress,
-			ForkID:          forkID,
+			ForkID:          uint16(forkID),
 			BlockHash:       blockResponse.BlockHash,
 			StateRoot:       blockResponse.BlockHash, //From etrog, the blockhash is the block root
 		}
@@ -60,23 +57,9 @@ func (f *finalizer) DSSendBatchBookmark(batchNumber uint64) {
 	// Check if stream server enabled
 	if f.streamServer != nil {
 		// Send batch bookmark to the streamer
-		f.dataToStream <- datastream.BookMark{
-			Type:  datastream.BookmarkType_BOOKMARK_TYPE_BATCH,
+		f.dataToStream <- state.DSBookMark{
+			Type:  state.BookMarkTypeBatch,
 			Value: batchNumber,
 		}
 	}
 }
-
-func (f *finalizer) DSSendBatch(batchNumber uint64, stateRoot common.Hash, localExitRoot common.Hash) {
-	forkID := f.stateIntf.GetForkIDByBatchNumber(batchNumber)
-
-	if f.streamServer != nil {
-		// Send batch to the streamer
-		f.dataToStream <- datastream.Batch{
-			Number:        batchNumber,
-			ForkId:        forkID,
-			StateRoot:     stateRoot.Bytes(),
-			LocalExitRoot: localExitRoot.Bytes(),
-		}
-	}
-}

sequencer/datastreamer.go

@@ -981,7 +981,6 @@ func TestFinalizer_closeWIPBatch(t *testing.T) {
 			// arrange
 			stateMock.Mock.On("CloseWIPBatch", ctx, receipt, mock.Anything).Return(tc.managerErr).Once()
 			stateMock.On("BeginStateTransaction", ctx).Return(dbTxMock, nilErr).Once()
-			stateMock.On("GetForkIDByBatchNumber", mock.Anything).Return(uint64(state.FORKID_BLUEBERRY))
 			if tc.managerErr == nil {
 				dbTxMock.On("Commit", ctx).Return(nilErr).Once()
 			} else {

sequencer/finalizer_test.go

@@ -197,7 +197,7 @@ func (f *finalizer) handleProcessForcedBatchResponse(ctx context.Context, newBat
 		}
 
 		// Send L2 block to data streamer
-		err = f.DSSendL2Block(newBatchNumber, forcedL2BlockResponse, 0, forcedL2BlockResponse.Timestamp)
+		err = f.DSSendL2Block(newBatchNumber, forcedL2BlockResponse, 0)
 		if err != nil {
 			//TODO: we need to halt/rollback the L2 block if we had an error sending to the data streamer?
 			log.Errorf("error sending L2 block %d to data streamer, error: %v", forcedL2BlockResponse.BlockNumber, err)

sequencer/forcedbatch.go

@@ -427,7 +427,7 @@ func (f *finalizer) storeL2Block(ctx context.Context, l2Block *L2Block) error {
 	log.Infof("l2 block %d [%d] transactions updated as selected in the pooldb", blockResponse.BlockNumber, l2Block.trackingNum)
 
 	// Send L2 block to data streamer
-	err = f.DSSendL2Block(f.wipBatch.batchNumber, blockResponse, l2Block.getL1InfoTreeIndex(), l2Block.timestamp)
+	err = f.DSSendL2Block(f.wipBatch.batchNumber, blockResponse, l2Block.getL1InfoTreeIndex())
 	if err != nil {
 		//TODO: we need to halt/rollback the L2 block if we had an error sending to the data streamer?
 		log.Errorf("error sending L2 block %d [%d] to data streamer, error: %v", blockResponse.BlockNumber, l2Block.trackingNum, err)