feat: implement Backup functionality for datastore

- Added Backup method to Store interface and DefaultStore implementation to stream a Badger backup of the datastore.
- Introduced BackupRequest and BackupResponse messages in the state_rpc.proto file to handle backup requests and responses.
- Implemented backup streaming logic in StoreServer, including metadata handling for current and target heights.
- Created a backupStreamWriter to manage chunked writing of backup data.
- Updated client tests to validate the Backup functionality.
- Enhanced mock store to support Backup method for testing.
- Added unit tests for Backup functionality in the store package.

Author: randygrok

pkg/rpc/client/client.go

@@ -2,6 +2,8 @@ package client
 
 import (
 	"context"
+	"fmt"
+	"io"
 	"net/http"
 
 	"connectrpc.com/connect"
@@ -92,6 +94,54 @@ func (c *Client) GetMetadata(ctx context.Context, key string) ([]byte, error) {
 	return resp.Msg.Value, nil
 }
 
+// Backup streams a datastore backup into the provided writer and returns the final metadata emitted by the server.
+// The writer is not closed by this method.
+func (c *Client) Backup(ctx context.Context, params *pb.BackupRequest, dst io.Writer) (*pb.BackupMetadata, error) {
+	if dst == nil {
+		return nil, fmt.Errorf("backup destination writer cannot be nil")
+	}
+
+	if params == nil {
+		params = &pb.BackupRequest{}
+	}
+
+	stream, err := c.storeClient.Backup(ctx, connect.NewRequest(params))
+	if err != nil {
+		return nil, err
+	}
+	defer stream.Close() // Best effort; ignore close error to preserve primary result.
+
+	var lastMetadata *pb.BackupMetadata
+	for stream.Receive() {
+		msg := stream.Msg()
+		if metadata := msg.GetMetadata(); metadata != nil {
+			lastMetadata = metadata
+			continue
+		}
+
+		if chunk := msg.GetChunk(); chunk != nil {
+			if _, err := dst.Write(chunk); err != nil {
+				_ = stream.Close()
+				return lastMetadata, fmt.Errorf("failed to write backup chunk: %w", err)
+			}
+		}
+	}
+
+	if err := stream.Err(); err != nil {
+		return lastMetadata, err
+	}
+
+	if lastMetadata == nil {
+		return nil, fmt.Errorf("backup stream completed without metadata")
+	}
+
+	if !lastMetadata.GetCompleted() {
+		return lastMetadata, fmt.Errorf("backup stream ended without completion metadata")
+	}
+
+	return lastMetadata, nil
+}
+
 // GetPeerInfo returns information about the connected peers
 func (c *Client) GetPeerInfo(ctx context.Context) ([]*pb.PeerInfo, error) {
 	req := connect.NewRequest(&emptypb.Empty{})

pkg/rpc/client/client_test.go

@@ -1,7 +1,9 @@
 package client
 
 import (
+	"bytes"
 	"context"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -20,6 +22,7 @@ import (
 	"github.com/evstack/ev-node/pkg/rpc/server"
 	"github.com/evstack/ev-node/test/mocks"
 	"github.com/evstack/ev-node/types"
+	pb "github.com/evstack/ev-node/types/pb/evnode/v1"
 	rpc "github.com/evstack/ev-node/types/pb/evnode/v1/v1connect"
 )
 
@@ -173,6 +176,34 @@ func TestClientGetBlockByHash(t *testing.T) {
 	mockStore.AssertExpectations(t)
 }
 
+func TestClientBackup(t *testing.T) {
+	mockStore := mocks.NewMockStore(t)
+	mockP2P := mocks.NewMockP2PRPC(t)
+
+	mockStore.On("Height", mock.Anything).Return(uint64(15), nil)
+	mockStore.On("Backup", mock.Anything, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		writer := args.Get(1).(io.Writer)
+		_, _ = writer.Write([]byte("chunk-1"))
+		_, _ = writer.Write([]byte("chunk-2"))
+	}).Return(uint64(42), nil)
+
+	testServer, client := setupTestServer(t, mockStore, mockP2P)
+	defer testServer.Close()
+
+	var buf bytes.Buffer
+	metadata, err := client.Backup(context.Background(), &pb.BackupRequest{TargetHeight: 10}, &buf)
+	require.NoError(t, err)
+	require.NotNil(t, metadata)
+	require.Equal(t, "chunk-1chunk-2", buf.String())
+	require.True(t, metadata.GetCompleted())
+	require.Equal(t, uint64(15), metadata.GetCurrentHeight())
+	require.Equal(t, uint64(10), metadata.GetTargetHeight())
+	require.Equal(t, uint64(0), metadata.GetSinceVersion())
+	require.Equal(t, uint64(42), metadata.GetLastVersion())
+
+	mockStore.AssertExpectations(t)
+}
+
 func TestClientGetPeerInfo(t *testing.T) {
 	// Create mocks
 	mockStore := mocks.NewMockStore(t)

pkg/rpc/server/server.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"io"
 
 	"net/http"
 	"time"
@@ -188,6 +189,166 @@ func (s *StoreServer) GetMetadata(
 	}), nil
 }
 
+// Backup streams a Badger backup of the datastore so it can be persisted externally.
+func (s *StoreServer) Backup(
+	ctx context.Context,
+	req *connect.Request[pb.BackupRequest],
+	stream *connect.ServerStream[pb.BackupResponse],
+) error {
+	since := req.Msg.GetSinceVersion()
+	targetHeight := req.Msg.GetTargetHeight()
+
+	currentHeight, err := s.store.Height(ctx)
+	if err != nil {
+		return connect.NewError(connect.CodeInternal, fmt.Errorf("failed to get current height: %w", err))
+	}
+
+	if targetHeight != 0 && targetHeight > currentHeight {
+		return connect.NewError(
+			connect.CodeFailedPrecondition,
+			fmt.Errorf("requested target height %d exceeds current height %d", targetHeight, currentHeight),
+		)
+	}
+
+	initialMetadata := &pb.BackupMetadata{
+		CurrentHeight: currentHeight,
+		TargetHeight:  targetHeight,
+		SinceVersion:  since,
+		Completed:     false,
+		LastVersion:   0,
+	}
+
+	if err := stream.Send(&pb.BackupResponse{
+		Response: &pb.BackupResponse_Metadata{
+			Metadata: initialMetadata,
+		},
+	}); err != nil {
+		return err
+	}
+
+	writer := newBackupStreamWriter(stream, defaultBackupChunkSize)
+	version, err := s.store.Backup(ctx, writer, since)
+	if err != nil {
+		var connectErr *connect.Error
+		if errors.As(err, &connectErr) {
+			return connectErr
+		}
+		if errors.Is(err, context.Canceled) {
+			return connect.NewError(connect.CodeCanceled, err)
+		}
+		if errors.Is(err, context.DeadlineExceeded) {
+			return connect.NewError(connect.CodeDeadlineExceeded, err)
+		}
+		return connect.NewError(connect.CodeInternal, fmt.Errorf("failed to execute backup: %w", err))
+	}
+
+	if err := writer.Flush(); err != nil {
+		var connectErr *connect.Error
+		if errors.As(err, &connectErr) {
+			return connectErr
+		}
+		if errors.Is(err, context.Canceled) {
+			return connect.NewError(connect.CodeCanceled, err)
+		}
+		if errors.Is(err, context.DeadlineExceeded) {
+			return connect.NewError(connect.CodeDeadlineExceeded, err)
+		}
+		return connect.NewError(connect.CodeInternal, fmt.Errorf("failed to flush backup stream: %w", err))
+	}
+
+	completedMetadata := &pb.BackupMetadata{
+		CurrentHeight: currentHeight,
+		TargetHeight:  targetHeight,
+		SinceVersion:  since,
+		LastVersion:   version,
+		Completed:     true,
+	}
+
+	if err := stream.Send(&pb.BackupResponse{
+		Response: &pb.BackupResponse_Metadata{
+			Metadata: completedMetadata,
+		},
+	}); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+const defaultBackupChunkSize = 128 * 1024
+
+var _ io.Writer = (*backupStreamWriter)(nil)
+
+type backupStreamWriter struct {
+	stream    *connect.ServerStream[pb.BackupResponse]
+	buf       []byte
+	chunkSize int
+}
+
+func newBackupStreamWriter(stream *connect.ServerStream[pb.BackupResponse], chunkSize int) *backupStreamWriter {
+	if chunkSize <= 0 {
+		chunkSize = defaultBackupChunkSize
+	}
+
+	return &backupStreamWriter{
+		stream:    stream,
+		buf:       make([]byte, 0, chunkSize),
+		chunkSize: chunkSize,
+	}
+}
+
+func (w *backupStreamWriter) Write(p []byte) (int, error) {
+	written := 0
+	for len(p) > 0 {
+		space := w.chunkSize - len(w.buf)
+		if space == 0 {
+			if err := w.flush(); err != nil {
+				return written, err
+			}
+			space = w.chunkSize - len(w.buf)
+		}
+
+		if space > len(p) {
+			space = len(p)
+		}
+
+		w.buf = append(w.buf, p[:space]...)
+		p = p[space:]
+		written += space
+
+		if len(w.buf) == w.chunkSize {
+			if err := w.flush(); err != nil {
+				return written, err
+			}
+		}
+	}
+	return written, nil
+}
+
+func (w *backupStreamWriter) Flush() error {
+	return w.flush()
+}
+
+func (w *backupStreamWriter) flush() error {
+	if len(w.buf) == 0 {
+		return nil
+	}
+
+	chunk := make([]byte, len(w.buf))
+	copy(chunk, w.buf)
+
+	if err := w.stream.Send(&pb.BackupResponse{
+		Response: &pb.BackupResponse_Chunk{
+			Chunk: chunk,
+		},
+	}); err != nil {
+		return err
+	}
+
+	w.buf = w.buf[:0]
+	return nil
+}
+
 type ConfigServer struct {
 	config config.Config
 	signer []byte

pkg/store/backup.go

@@ -0,0 +1,43 @@
+package store
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	badger4 "github.com/ipfs/go-ds-badger4"
+)
+
+// Backup streams the underlying Badger datastore snapshot into the provided writer.
+// The returned uint64 corresponds to the last version contained in the backup stream,
+// which can be re-used to generate incremental backups via the since parameter.
+func (s *DefaultStore) Backup(ctx context.Context, writer io.Writer, since uint64) (uint64, error) {
+	if err := ctx.Err(); err != nil {
+		return 0, err
+	}
+
+	// Try to leverage a native backup implementation if the underlying datastore exposes one.
+	type backupable interface {
+		Backup(io.Writer, uint64) (uint64, error)
+	}
+	if dsBackup, ok := s.db.(backupable); ok {
+		version, err := dsBackup.Backup(writer, since)
+		if err != nil {
+			return 0, fmt.Errorf("datastore backup failed: %w", err)
+		}
+		return version, nil
+	}
+
+	// Default Badger datastore used across ev-node.
+	badgerDatastore, ok := s.db.(*badger4.Datastore)
+	if !ok {
+		return 0, fmt.Errorf("backup is not supported by the configured datastore")
+	}
+
+	// `badger.DB.Backup` internally orchestrates a consistent snapshot without pausing writes.
+	version, err := badgerDatastore.DB.Backup(writer, since)
+	if err != nil {
+		return 0, fmt.Errorf("badger backup failed: %w", err)
+	}
+	return version, nil
+}

pkg/store/store_test.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"bytes"
 	"context"
 	"encoding/binary"
 	"errors"
@@ -1127,3 +1128,24 @@ func TestRollbackDAIncludedHeightGetMetadataError(t *testing.T) {
 	require.Contains(err.Error(), "failed to get DA included height")
 	require.Contains(err.Error(), "metadata retrieval failed")
 }
+
+func TestDefaultStoreBackup(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	kv, err := NewDefaultInMemoryKVStore()
+	require.NoError(t, err)
+
+	s := New(kv)
+	t.Cleanup(func() {
+		require.NoError(t, s.Close())
+	})
+
+	require.NoError(t, s.SetMetadata(ctx, "backup-test", []byte("value")))
+
+	var buf bytes.Buffer
+	version, err := s.Backup(ctx, &buf, 0)
+	require.NoError(t, err)
+	require.NotZero(t, buf.Len())
+	require.NotZero(t, version)
+}

pkg/store/types.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"context"
+	"io"
 
 	"github.com/evstack/ev-node/types"
 )
@@ -50,6 +51,10 @@ type Store interface {
 	// Aggregator is used to determine if the rollback is performed on the aggregator node.
 	Rollback(ctx context.Context, height uint64, aggregator bool) error
 
+	// Backup writes a consistent backup stream to writer. The returned version can be used
+	// as the starting point for incremental backups.
+	Backup(ctx context.Context, writer io.Writer, since uint64) (uint64, error)
+
 	// Close safely closes underlying data storage, to ensure that data is actually saved.
 	Close() error
 }