Flush state db (#1831)

* flush state db * flush state db * update prover image * improve log message
0xPolygonHermez · Mar 16, 2023 · 470216d · 470216d
1 parent 61f4a23
commit 470216d
Show file tree

Hide file tree

Showing 12 changed files with 839 additions and 245 deletions.
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -104,7 +104,7 @@ services:
   zkevm-prover:
     container_name: zkevm-prover
     restart: unless-stopped
-    image: hermeznetwork/zkevm-prover:45df44e
+    image: hermeznetwork/zkevm-prover:40338c3
     depends_on:
       zkevm-state-db:
         condition: service_healthy

diff --git a/merkletree/pb/statedb.pb.go b/merkletree/pb/statedb.pb.go
diff --git a/merkletree/pb/statedb_grpc.pb.go b/merkletree/pb/statedb_grpc.pb.go
diff --git a/merkletree/tree.go b/merkletree/tree.go
@@ -9,6 +9,7 @@ import (
 	"github.com/0xPolygonHermez/zkevm-node/hex"
 	"github.com/0xPolygonHermez/zkevm-node/merkletree/pb"
 	"github.com/ethereum/go-ethereum/common"
+	"google.golang.org/protobuf/types/known/emptypb"
 )
 
 // StateTree provides methods to access and modify state in merkletree
@@ -312,3 +313,9 @@ func (tree *StateTree) setProgram(ctx context.Context, key []uint64, data []byte
 	})
 	return err
 }
+
+// Flush flushes all changes to the persistent storage.
+func (tree *StateTree) Flush(ctx context.Context) error {
+	_, err := tree.grpcClient.Flush(ctx, &emptypb.Empty{})
+	return err
+}
diff --git a/proto/src/proto/statedb/v1/statedb.proto b/proto/src/proto/statedb/v1/statedb.proto
@@ -16,16 +16,18 @@ message Version {
  * Define all methods implementes by the gRPC
  * Get: get the value for a specific key
  * Set: set the value for a specific key
- * SetProgram: set the byte data for a specific hash
- * GetProgram: get the byte data for a specific hash
+ * SetProgram: set the byte data for a specific key
+ * GetProgram: get the byte data for a specific key
  * Flush: wait for all the pendings writes to the DB are done
  */
 service StateDBService {
     rpc Set(SetRequest) returns (SetResponse) {}
     rpc Get(GetRequest) returns (GetResponse) {}
     rpc SetProgram(SetProgramRequest) returns (SetProgramResponse) {}
     rpc GetProgram(GetProgramRequest) returns (GetProgramResponse) {}
-    rpc Flush (google.protobuf.Empty) returns (google.protobuf.Empty) {}
+    rpc LoadDB(LoadDBRequest) returns (google.protobuf.Empty) {}
+    rpc LoadProgramDB(LoadProgramDBRequest) returns (google.protobuf.Empty) {}
+    rpc Flush (google.protobuf.Empty) returns (FlushResponse) {}
 }
 
 ///////////////////
@@ -37,34 +39,38 @@ service StateDBService {
  * @param {old_root} - merkle-tree root
  * @param {key} - key to set
  * @param {value} - scalar value to set (HEX string format)
- * @param {persistent} - indicates if it should be stored in the database (true) or only in the memory cache (false)
+ * @param {persistent} - indicates if it should be stored in the SQL database (true) or only in the memory cache (false)
  * @param {details} - indicates if it should return all response parameters (true) or just the new root (false)
+ * @param {get_db_read_log} - indicates if it should return the DB reads generated during the execution of the request
  */
 message SetRequest {
     Fea old_root = 1;
     Fea key = 2;
     string value = 3;
     bool persistent = 4;
     bool details = 5;
+    bool get_db_read_log = 6;
 }
 
 /**
  * @dev GetRequest
  * @param {root} - merkle-tree root
  * @param {key} - key to look for
  * @param {details} - indicates if it should return all response parameters (true) or just the new root (false)
+ * @param {get_db_read_log} - indicates if it should return the DB reads generated during the execution of the request
  */
 message GetRequest {
     Fea root = 1;
     Fea key = 2;
     bool details = 3;
+    bool get_db_read_log = 4;
 }
 
 /**
  * @dev SetProgramRequest
- * @param {key} - hash to set
+ * @param {key} - key to set
  * @param {data} - Program data to store
- * @param {persistent} - indicates if it should be stored in the database (true) or only in the memory cache (false)
+ * @param {persistent} - indicates if it should be stored in the SQL database (true) or only in the memory cache (false)
  */
 message SetProgramRequest {
     Fea key = 1;
@@ -74,12 +80,32 @@ message SetProgramRequest {
 
 /**
  * @dev GetProgramRequest
- * @param {key} - hash to get program data
+ * @param {key} - key to get program data
  */
 message GetProgramRequest {
     Fea key = 1;
 }
 
+/**
+ * @dev LoadDBRequest
+ * @param {input_db} - list of db records (MT) to load in the database
+ * @param {persistent} - indicates if it should be stored in the SQL database (true) or only in the memory cache (false)
+ */
+message LoadDBRequest {
+    map<string, FeList> input_db = 1;
+    bool persistent = 2;
+}
+
+/**
+ * @dev LoadProgramDBRequest
+ * @param {input_program_db} - list of db records (program) to load in the database
+ * @param {persistent} - indicates if it should be stored in the SQL database (true) or only in the memory cache (false)
+ */
+message LoadProgramDBRequest {
+    map<string, bytes> input_program_db = 1;
+    bool persistent = 2;
+}
+
 /////////////////////
 // Responses messages
 /////////////////////
@@ -96,6 +122,8 @@ message GetProgramRequest {
  * @param {old_value} - old value (HEX string format)
  * @param {new_value} - new value (HEX string format)
  * @param {mode}
+ * @param {proof_hash_counter}
+ * @param {db_read_log} - list of db records read during the execution of the request
  * @param {result} - result code
  */
 message SetResponse {
@@ -109,7 +137,9 @@ message SetResponse {
     string old_value = 8;
     string new_value = 9;
     string mode = 10;
-    ResultCode result = 11;
+    uint64 proof_hash_counter = 11;
+    map<string, FeList> db_read_log = 12;
+    ResultCode result = 13;
 }
 
 /**
@@ -121,6 +151,8 @@ message SetResponse {
  * @param {ins_value} - value found (HEX string format)
  * @param {is_old0} - is new insert or delete
  * @param {value} - value retrieved (HEX string format)
+ * @param {proof_hash_counter}
+ * @param {db_read_log} - list of db records read during the execution of the request
  * @param {result} - result code
  */
 message GetResponse {
@@ -131,7 +163,9 @@ message GetResponse {
     string ins_value = 5;
     bool is_old0 = 6;
     string value = 7;
-    ResultCode result = 8;
+    uint64 proof_hash_counter = 8;
+    map<string, FeList> db_read_log = 9;
+    ResultCode result = 10;
 }
 
 /**
@@ -152,6 +186,14 @@ message GetProgramResponse {
     ResultCode result = 2;
 }
 
+/**
+ * @dev FlushResponse
+ * @param {result} - result code
+ */
+message FlushResponse {
+    ResultCode result = 1;
+}
+
 /**
  * @dev Array of 4 FE
  * @param {fe0} - Field Element value for pos 0
@@ -166,9 +208,17 @@ message Fea {
     uint64 fe3 = 4;
 }
 
+/**
+ * @dev FE (Field Element) List
+ * @param {fe} - list of Fe
+*/
+message FeList {
+    repeated uint64 fe = 1;
+}
+
 /**
  * @dev Siblings List
- * @param {sibling} - sibling
+ * @param {sibling} - list of siblings
 */
 message SiblingList {
     repeated uint64 sibling = 1;
@@ -182,9 +232,10 @@ message ResultCode {
     enum Code {
         CODE_UNSPECIFIED = 0;
         CODE_SUCCESS = 1;
-        CODE_KEY_NOT_FOUND = 2;
-        CODE_DB_ERROR = 3;
+        CODE_DB_KEY_NOT_FOUND = 2; // Requested key was not found in database
+        CODE_DB_ERROR = 3; // Error connecting to database, or processing request
         CODE_INTERNAL_ERROR = 4;
+        CODE_SMT_INVALID_DATA_SIZE = 14; // Invalid size for the data of MT node
     }
     Code code = 1;
 }
diff --git a/sequencer/dbmanager.go b/sequencer/dbmanager.go
@@ -181,6 +181,13 @@ func (d *dbManager) storeProcessedTxAndDeleteFromPool() {
 	for {
 		txToStore := <-d.txsStore.Ch
 		d.checkIfReorg()
+
+		// Flush the state db
+		err := d.state.FlushMerkleTree(d.ctx)
+		if err != nil {
+			log.Fatalf("StoreProcessedTxAndDeleteFromPool. Error flushing state db: %v", err)
+		}
+
 		log.Debugf("Storing tx %v", txToStore.txResponse.TxHash)
 		dbTx, err := d.BeginStateTransaction(d.ctx)
 		if err != nil {
@@ -539,3 +546,8 @@ func (d *dbManager) GetLatestVirtualBatchTimestamp(ctx context.Context, dbTx pgx
 func (d *dbManager) CountReorgs(ctx context.Context, dbTx pgx.Tx) (uint64, error) {
 	return d.state.CountReorgs(ctx, dbTx)
 }
+
+// FlushMerkleTree persists updates in the Merkle tree
+func (d *dbManager) FlushMerkleTree(ctx context.Context) error {
+	return d.state.FlushMerkleTree(ctx)
+}
diff --git a/sequencer/finalizer.go b/sequencer/finalizer.go
@@ -344,7 +344,7 @@ func (f *finalizer) processTransaction(ctx context.Context, tx *TxTracker) error
 		hash = tx.HashStr
 	}
 	log.Infof("processTransaction: single tx. Batch.BatchNumber: %d, BatchNumber: %d, OldStateRoot: %s, txHash: %s, GER: %s", f.batch.batchNumber, f.processRequest.BatchNumber, f.processRequest.OldStateRoot, hash, f.processRequest.GlobalExitRoot.String())
-	result, err := f.executor.ProcessBatch(ctx, f.processRequest, false)
+	result, err := f.executor.ProcessBatch(ctx, f.processRequest, true)
 	if err != nil {
 		log.Errorf("failed to process transaction, isClaim: %v, err: %s", tx.IsClaim, err)
 		return err
@@ -703,7 +703,7 @@ func (f *finalizer) reprocessFullBatch(ctx context.Context, batchNum uint64, exp
 		log.Infof("reprocessFullBatch: Tx position %d. TxHash: %s", i, tx.Hash())
 	}
 
-	result, err := f.executor.ProcessBatch(ctx, processRequest, true)
+	result, err := f.executor.ProcessBatch(ctx, processRequest, false)
 	if err != nil {
 		log.Errorf("failed to process batch, err: %s", err)
 		return nil, err

diff --git a/sequencer/interfaces.go b/sequencer/interfaces.go
@@ -73,6 +73,7 @@ type stateInterface interface {
 	GetLatestVirtualBatchTimestamp(ctx context.Context, dbTx pgx.Tx) (time.Time, error)
 	CountReorgs(ctx context.Context, dbTx pgx.Tx) (uint64, error)
 	GetLatestGer(ctx context.Context, gerFinalityNumberOfBlocks uint64) (state.GlobalExitRoot, time.Time, error)
+	FlushMerkleTree(ctx context.Context) error
 }
 
 type workerInterface interface {
@@ -113,6 +114,7 @@ type dbManagerInterface interface {
 	UpdateTxStatus(ctx context.Context, hash common.Hash, newStatus pool.TxStatus, isWIP bool) error
 	GetLatestVirtualBatchTimestamp(ctx context.Context, dbTx pgx.Tx) (time.Time, error)
 	CountReorgs(ctx context.Context, dbTx pgx.Tx) (uint64, error)
+	FlushMerkleTree(ctx context.Context) error
 }
 
 type dbManagerStateInterface interface {
@@ -142,6 +144,7 @@ type dbManagerStateInterface interface {
 	GetLatestVirtualBatchTimestamp(ctx context.Context, dbTx pgx.Tx) (time.Time, error)
 	CountReorgs(ctx context.Context, dbTx pgx.Tx) (uint64, error)
 	GetLatestGer(ctx context.Context, gerFinalityNumberOfBlocks uint64) (state.GlobalExitRoot, time.Time, error)
+	FlushMerkleTree(ctx context.Context) error
 }
 
 type ethTxManager interface {