Merge pull request #21 from 0xPolygon/pullupstream/v0.0.3

Pullupstream/v0.3.0

ci/e2e-group9/forced_batches_vector_group1_test.go

@@ -24,7 +24,7 @@ import (
 )
 
 const (
-	forkID5 = 5
+	forkID5 uint64 = 5
 )
 
 func TestForcedBatchesVectorFiles(t *testing.T) {
@@ -64,6 +64,7 @@ func TestForcedBatchesVectorFiles(t *testing.T) {
 				log.Info("###################")
 				genesisActions := vectors.GenerateGenesisActions(testCase.Genesis)
 				require.NoError(t, opsman.SetGenesis(genesisActions))
+				require.NoError(t, opsman.SetForkID(forkID5))
 				require.NoError(t, opsman.Setup())
 
 				// Check initial root

cmd/readme.md

@@ -2,10 +2,10 @@
 
 ### Create snapshots
 ```
-go run ./cmd snapshot --cfg config/environments/local/local.node.config.toml
+go run ./cmd snapshot --cfg config/environments/local/local.node.config.toml --output ./folder/
 ```
 
 ### Restore snapshots
 ```
-go run ./cmd restore --cfg config/environments/local/local.node.config.toml -is zkevmpubliccorestatedb_1685614455_v0.1.0_undefined.sql.tar.gz -ih zkevmpublicstatedb_1685615051_v0.1.0_undefined.sql.tar.gz
+go run ./cmd restore --cfg config/environments/local/local.node.config.toml -is ./folder/zkevmpubliccorestatedb_1685614455_v0.1.0_undefined.sql.tar.gz -ih ./folder/zkevmpublicstatedb_1685615051_v0.1.0_undefined.sql.tar.gz
 ```

cmd/restore.go

@@ -1,7 +1,12 @@
 package main
 
 import (
+	"bufio"
 	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
 	"strconv"
 	"strings"
 
@@ -45,9 +50,16 @@ func restore(ctx *cli.Context) error {
 		return errors.New("stateDB input file must end in .sql.tar.gz")
 	}
 
-	// Run migrations to create schemas and tables
-	runStateMigrations(c.StateDB)
-
+	d, err := db.NewSQLDB(c.StateDB)
+	if err != nil {
+		log.Error("error conecting to stateDB. Error: ", err)
+		return err
+	}
+	_, err = d.Exec(ctx.Context, "DROP SCHEMA IF EXISTS state CASCADE; DROP TABLE IF EXISTS gorp_migrations;")
+	if err != nil {
+		log.Error("error dropping state schema or migration table. Error: ", err)
+		return err
+	}
 	port, err := strconv.Atoi(c.StateDB.Port)
 	if err != nil {
 		log.Error("error converting port to int. Error: ", err)
@@ -64,12 +76,11 @@ func restore(ctx *cli.Context) error {
 		log.Error("error: ", err)
 		return err
 	}
-	restore.Role = c.StateDB.User
-	restore.Schemas = append(restore.Schemas, "state")
+	params := []string{"--no-owner", "--no-acl", "--format=c"}
 	log.Info("Restore stateDB snapshot started, please wait...")
-	restoreExec := restore.Exec(inputFileStateDB, pg.ExecOptions{StreamPrint: false})
+	restoreExec := execCommand(restore, inputFileStateDB, pg.ExecOptions{StreamPrint: false}, params)
 	if restoreExec.Error != nil {
-		log.Error("error restoring snapshot. Error: ", restoreExec.Error.Err)
+		log.Error("error restoring stateDB snapshot. Error: ", restoreExec.Error.Err)
 		log.Debug("restoreExec.Output: ", restoreExec.Output)
 		return err
 	}
@@ -84,12 +95,12 @@ func restore(ctx *cli.Context) error {
 		log.Error("error converting port to int. Error: ", err)
 		return err
 	}
-	d, err := db.NewSQLDB(c.HashDB)
+	d, err = db.NewSQLDB(c.HashDB)
 	if err != nil {
 		log.Error("error conecting to hashdb. Error: ", err)
 		return err
 	}
-	_, err = d.Exec(ctx.Context, "DROP SCHEMA IF EXISTS state CASCADE; CREATE SCHEMA IF NOT EXISTS state;")
+	_, err = d.Exec(ctx.Context, "DROP SCHEMA IF EXISTS state CASCADE;")
 	if err != nil {
 		log.Error("error dropping and creating state schema. Error: ", err)
 		return err
@@ -105,17 +116,62 @@ func restore(ctx *cli.Context) error {
 		log.Error("error: ", err)
 		return err
 	}
-	restore.Role = c.HashDB.User
-	restore.Schemas = append(restore.Schemas, "state")
-	restore.Options = []string{"--no-owner", "--no-acl"}
+
 	log.Info("Restore HashDB snapshot started, please wait...")
-	restoreExec = restore.Exec(inputFileHashDB, pg.ExecOptions{StreamPrint: false})
+	restoreExec = execCommand(restore, inputFileHashDB, pg.ExecOptions{StreamPrint: false}, params)
 	if restoreExec.Error != nil {
-		log.Error("error restoring snapshot. Error: ", restoreExec.Error.Err)
+		log.Error("error restoring hashDB snapshot. Error: ", restoreExec.Error.Err)
 		log.Debug("restoreExec.Output: ", restoreExec.Output)
 		return err
 	}
 
 	log.Info("Restore HashDB snapshot success")
 	return nil
 }
+
+func execCommand(x *pg.Restore, filename string, opts pg.ExecOptions, params []string) pg.Result {
+	result := pg.Result{}
+	options := append(params, x.Postgres.Parse()...)
+	options = append(options, fmt.Sprintf("%s%s", x.Path, filename))
+	log.Debug("Options: ", options)
+
+	result.FullCommand = strings.Join(options, " ")
+	cmd := exec.Command(pg.PGRestoreCmd, options...) //nolint:gosec
+	cmd.Env = append(os.Environ(), x.EnvPassword)
+	stderrIn, _ := cmd.StderrPipe()
+	go func(stderrIn io.ReadCloser, opts pg.ExecOptions, result *pg.Result) {
+		output := ""
+		reader := bufio.NewReader(stderrIn)
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				if errors.Is(err, io.EOF) {
+					result.Output = output
+					break
+				}
+				result.Error = &pg.ResultError{Err: fmt.Errorf("error reading output: %w", err), CmdOutput: output}
+				break
+			}
+
+			if opts.StreamPrint {
+				_, err = fmt.Fprint(opts.StreamDestination, line)
+				if err != nil {
+					result.Error = &pg.ResultError{Err: fmt.Errorf("error writing output: %w", err), CmdOutput: output}
+					break
+				}
+			}
+
+			output += line
+		}
+	}(stderrIn, opts, &result)
+	err := cmd.Start()
+	if err != nil {
+		result.Error = &pg.ResultError{Err: err, CmdOutput: result.Output}
+	}
+	err = cmd.Wait()
+	if exitError, ok := err.(*exec.ExitError); ok {
+		result.Error = &pg.ResultError{Err: err, ExitCode: exitError.ExitCode(), CmdOutput: result.Output}
+	}
+
+	return result
+}