Skip to content

Commit 05fe34a

Browse files
randygrokrandygrok
committed
feat: add Backup command for datastore with streaming functionality
1 parent 9663554 commit 05fe34a

3 files changed

Lines changed: 285 additions & 0 deletions

File tree

‎apps/evm/single/main.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,14 @@ func main() {
2323

2424
// Add configuration flags to NetInfoCmd so it can read RPC address
2525
config.AddFlags(rollcmd.NetInfoCmd)
26+
backupCmd := rollcmd.NewBackupCmd()
27+
config.AddFlags(backupCmd)
2628

2729
rootCmd.AddCommand(
2830
cmd.InitCmd(),
2931
cmd.RunCmd,
3032
cmd.NewRollbackCmd(),
33+
backupCmd,
3134
rollcmd.VersionCmd,
3235
rollcmd.NetInfoCmd,
3336
rollcmd.StoreUnsafeCleanCmd,

‎pkg/cmd/backup.go‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
package cmd
2+
3+
import (
4+
"bufio"
5+
"context"
6+
"errors"
7+
"fmt"
8+
"io"
9+
"os"
10+
"path/filepath"
11+
"strings"
12+
"time"
13+
14+
"github.com/spf13/cobra"
15+
16+
clientrpc "github.com/evstack/ev-node/pkg/rpc/client"
17+
pb "github.com/evstack/ev-node/types/pb/evnode/v1"
18+
)
19+
20+
// NewBackupCmd creates a cobra command that streams a datastore backup via the RPC client.
21+
func NewBackupCmd() *cobra.Command {
22+
cmd := &cobra.Command{
23+
Use: "backup",
24+
Short: "Stream a datastore backup to a local file via RPC",
25+
SilenceUsage: true,
26+
RunE: func(cmd *cobra.Command, args []string) error {
27+
nodeConfig, err := ParseConfig(cmd)
28+
if err != nil {
29+
return fmt.Errorf("error parsing config: %w", err)
30+
}
31+
32+
rpcAddress := strings.TrimSpace(nodeConfig.RPC.Address)
33+
if rpcAddress == "" {
34+
return fmt.Errorf("RPC address not found in node configuration")
35+
}
36+
37+
baseURL := rpcAddress
38+
if !strings.HasPrefix(baseURL, "http://") && !strings.HasPrefix(baseURL, "https://") {
39+
baseURL = fmt.Sprintf("http://%s", baseURL)
40+
}
41+
42+
outputPath, err := cmd.Flags().GetString("output")
43+
if err != nil {
44+
return err
45+
}
46+
47+
if outputPath == "" {
48+
timestamp := time.Now().UTC().Format("20060102-150405")
49+
outputPath = fmt.Sprintf("evnode-backup-%s.badger", timestamp)
50+
}
51+
52+
outputPath, err = filepath.Abs(outputPath)
53+
if err != nil {
54+
return fmt.Errorf("failed to resolve output path: %w", err)
55+
}
56+
57+
if err := os.MkdirAll(filepath.Dir(outputPath), 0o755); err != nil {
58+
return fmt.Errorf("failed to create output directory: %w", err)
59+
}
60+
61+
force, err := cmd.Flags().GetBool("force")
62+
if err != nil {
63+
return err
64+
}
65+
66+
if !force {
67+
if _, statErr := os.Stat(outputPath); statErr == nil {
68+
return fmt.Errorf("output file %s already exists (use --force to overwrite)", outputPath)
69+
} else if !errors.Is(statErr, os.ErrNotExist) {
70+
return fmt.Errorf("failed to inspect output file: %w", statErr)
71+
}
72+
}
73+
74+
file, err := os.OpenFile(outputPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
75+
if err != nil {
76+
return fmt.Errorf("failed to open output file: %w", err)
77+
}
78+
defer file.Close()
79+
80+
writer := bufio.NewWriterSize(file, 1<<20) // 1 MiB buffer for fewer syscalls.
81+
bytesCount := &countingWriter{}
82+
streamWriter := io.MultiWriter(writer, bytesCount)
83+
84+
targetHeight, err := cmd.Flags().GetUint64("target-height")
85+
if err != nil {
86+
return err
87+
}
88+
89+
sinceVersion, err := cmd.Flags().GetUint64("since-version")
90+
if err != nil {
91+
return err
92+
}
93+
94+
ctx := cmd.Context()
95+
if ctx == nil {
96+
ctx = context.Background()
97+
}
98+
99+
client := clientrpc.NewClient(baseURL)
100+
101+
metadata, backupErr := client.Backup(ctx, &pb.BackupRequest{
102+
TargetHeight: targetHeight,
103+
SinceVersion: sinceVersion,
104+
}, streamWriter)
105+
if backupErr != nil {
106+
// Remove the partial file on failure to avoid keeping corrupt snapshots.
107+
_ = writer.Flush()
108+
_ = file.Close()
109+
_ = os.Remove(outputPath)
110+
return fmt.Errorf("backup failed: %w", backupErr)
111+
}
112+
113+
if err := writer.Flush(); err != nil {
114+
_ = file.Close()
115+
_ = os.Remove(outputPath)
116+
return fmt.Errorf("failed to flush backup data: %w", err)
117+
}
118+
119+
if !metadata.GetCompleted() {
120+
_ = file.Close()
121+
_ = os.Remove(outputPath)
122+
return fmt.Errorf("backup stream ended without completion metadata")
123+
}
124+
125+
cmd.Printf("Backup saved to %s (%d bytes)\n", outputPath, bytesCount.Bytes())
126+
cmd.Printf("Current height: %d\n", metadata.GetCurrentHeight())
127+
cmd.Printf("Target height: %d\n", metadata.GetTargetHeight())
128+
cmd.Printf("Since version: %d\n", metadata.GetSinceVersion())
129+
cmd.Printf("Last version: %d\n", metadata.GetLastVersion())
130+
131+
return nil
132+
},
133+
}
134+
135+
cmd.Flags().String("output", "", "Path to the backup file (defaults to ./evnode-backup-<timestamp>.badger)")
136+
cmd.Flags().Uint64("target-height", 0, "Target chain height to align the backup with (0 disables the check)")
137+
cmd.Flags().Uint64("since-version", 0, "Generate an incremental backup starting from the provided version")
138+
cmd.Flags().Bool("force", false, "Overwrite the output file if it already exists")
139+
140+
return cmd
141+
}
142+
143+
type countingWriter struct {
144+
total int64
145+
}
146+
147+
func (c *countingWriter) Write(p []byte) (int, error) {
148+
c.total += int64(len(p))
149+
return len(p), nil
150+
}
151+
152+
func (c *countingWriter) Bytes() int64 {
153+
return c.total
154+
}

‎pkg/cmd/backup_test.go‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
package cmd
2+
3+
import (
4+
"io"
5+
"net/http"
6+
"net/http/httptest"
7+
"os"
8+
"path/filepath"
9+
"strings"
10+
"testing"
11+
12+
"github.com/rs/zerolog"
13+
"github.com/spf13/cobra"
14+
"github.com/stretchr/testify/mock"
15+
"github.com/stretchr/testify/require"
16+
"golang.org/x/net/http2"
17+
"golang.org/x/net/http2/h2c"
18+
19+
"github.com/evstack/ev-node/pkg/config"
20+
"github.com/evstack/ev-node/pkg/rpc/server"
21+
"github.com/evstack/ev-node/test/mocks"
22+
"github.com/evstack/ev-node/types/pb/evnode/v1/v1connect"
23+
)
24+
25+
func TestBackupCmd_Success(t *testing.T) {
26+
t.Parallel()
27+
28+
mockStore := mocks.NewMockStore(t)
29+
30+
mockStore.On("Height", mock.Anything).Return(uint64(15), nil)
31+
mockStore.On("Backup", mock.Anything, mock.Anything, uint64(9)).Run(func(args mock.Arguments) {
32+
writer := args.Get(1).(io.Writer)
33+
_, _ = writer.Write([]byte("chunk-1"))
34+
_, _ = writer.Write([]byte("chunk-2"))
35+
}).Return(uint64(21), nil)
36+
37+
logger := zerolog.Nop()
38+
storeServer := server.NewStoreServer(mockStore, logger)
39+
mux := http.NewServeMux()
40+
storePath, storeHandler := v1connect.NewStoreServiceHandler(storeServer)
41+
mux.Handle(storePath, storeHandler)
42+
43+
httpServer := httptest.NewServer(h2c.NewHandler(mux, &http2.Server{}))
44+
defer httpServer.Close()
45+
46+
tempDir, err := os.MkdirTemp("", "evnode-backup-*")
47+
require.NoError(t, err)
48+
t.Cleanup(func() {
49+
_ = os.RemoveAll(tempDir)
50+
})
51+
52+
backupCmd := NewBackupCmd()
53+
config.AddFlags(backupCmd)
54+
55+
rootCmd := &cobra.Command{Use: "root"}
56+
config.AddGlobalFlags(rootCmd, "test")
57+
rootCmd.AddCommand(backupCmd)
58+
59+
outPath := filepath.Join(tempDir, "snapshot.badger")
60+
rpcAddr := strings.TrimPrefix(httpServer.URL, "http://")
61+
62+
output, err := executeCommandC(
63+
rootCmd,
64+
"backup",
65+
"--home="+tempDir,
66+
"--evnode.rpc.address="+rpcAddr,
67+
"--output", outPath,
68+
"--target-height", "12",
69+
"--since-version", "9",
70+
)
71+
72+
require.NoError(t, err, "command failed: %s", output)
73+
74+
data, readErr := os.ReadFile(outPath)
75+
require.NoError(t, readErr)
76+
require.Equal(t, "chunk-1chunk-2", string(data))
77+
78+
require.Contains(t, output, "Backup saved to")
79+
require.Contains(t, output, "Current height: 15")
80+
require.Contains(t, output, "Target height: 12")
81+
require.Contains(t, output, "Since version: 9")
82+
require.Contains(t, output, "Last version: 21")
83+
84+
mockStore.AssertExpectations(t)
85+
}
86+
87+
func TestBackupCmd_ExistingFileWithoutForce(t *testing.T) {
88+
t.Parallel()
89+
90+
mockStore := mocks.NewMockStore(t)
91+
logger := zerolog.Nop()
92+
storeServer := server.NewStoreServer(mockStore, logger)
93+
mux := http.NewServeMux()
94+
storePath, storeHandler := v1connect.NewStoreServiceHandler(storeServer)
95+
mux.Handle(storePath, storeHandler)
96+
97+
httpServer := httptest.NewServer(h2c.NewHandler(mux, &http2.Server{}))
98+
defer httpServer.Close()
99+
100+
tempDir, err := os.MkdirTemp("", "evnode-backup-existing-*")
101+
require.NoError(t, err)
102+
t.Cleanup(func() {
103+
_ = os.RemoveAll(tempDir)
104+
})
105+
106+
outPath := filepath.Join(tempDir, "snapshot.badger")
107+
require.NoError(t, os.WriteFile(outPath, []byte("existing"), 0o600))
108+
109+
backupCmd := NewBackupCmd()
110+
config.AddFlags(backupCmd)
111+
112+
rootCmd := &cobra.Command{Use: "root"}
113+
config.AddGlobalFlags(rootCmd, "test")
114+
rootCmd.AddCommand(backupCmd)
115+
116+
rpcAddr := strings.TrimPrefix(httpServer.URL, "http://")
117+
118+
output, err := executeCommandC(
119+
rootCmd,
120+
"backup",
121+
"--home="+tempDir,
122+
"--evnode.rpc.address="+rpcAddr,
123+
"--output", outPath,
124+
)
125+
126+
require.Error(t, err)
127+
require.Contains(t, output, "already exists (use --force to overwrite)")
128+
}

0 commit comments

Comments
 (0)