feat: implement RDB saver (#12)

* feat: replication: send PSYNC command to the master server

* feat: add GetReplicationID()

* feat: implement rdb saver

* feat: improve AuxiliaryField writing in RDBSaver

* fix: lint err

Author: mhughdo

.DS_Store

@@ -38,6 +38,7 @@ type Server struct {
 	masterAddr     string
 	messageChan    chan client.Message
 	masterClient   *client.Client
+	replicationID  string
 	disconnectChan chan *client.Client
 }
 
@@ -50,6 +51,10 @@ func NewServer(cfg *config.Config) *Server {
 		parts := strings.Split(replicaOf, " ")
 		masterAddr = fmt.Sprintf("%s:%s", parts[0], parts[1])
 	}
+	var replicationID string
+	if isMaster {
+		replicationID = utils.GenerateRandomAlphanumeric(40)
+	}
 	s := &Server{
 		mu:             sync.Mutex{},
 		cfg:            cfg,
@@ -61,6 +66,7 @@ func NewServer(cfg *config.Config) *Server {
 		masterAddr:     masterAddr,
 		messageChan:    make(chan client.Message),
 		masterClient:   nil,
+		replicationID:  replicationID,
 		disconnectChan: make(chan *client.Client),
 	}
 	s.cFactory = command.NewCommandFactory(store, cfg, s)
@@ -160,6 +166,10 @@ func (s *Server) startReplication(ctx context.Context) (err error) {
 		return fmt.Errorf("failed to send REPLCONF to master: %v", err)
 	}
 
+	if err := s.sendPsyncToMaster(ctx); err != nil {
+		return fmt.Errorf("failed to send PSYNC to master: %v", err)
+	}
+
 	return nil
 }
 
@@ -222,6 +232,21 @@ func (s *Server) sendReplconfToMaster(ctx context.Context) error {
 	return nil
 }
 
+func (s *Server) sendPsyncToMaster(ctx context.Context) error {
+	psyncCmd := resp.CreatePsyncCommand("?", "-1")
+	response, err := s.sendAndReceive(psyncCmd)
+	if err != nil {
+		return fmt.Errorf("failed to send PSYNC to master: %v", err)
+	}
+
+	if response.Type != resp.SimpleString || !strings.HasPrefix(response.String(), "FULLRESYNC") {
+		return fmt.Errorf("unexpected response from master for PSYNC: %v", response)
+	}
+
+	logger.Info(ctx, "Successfully sent PSYNC to master and received FULLRESYNC")
+	return nil
+}
+
 func (s *Server) loop(ctx context.Context) {
 	for {
 		select {
@@ -342,3 +367,9 @@ func (s *Server) Close(_ context.Context) error {
 	}
 	return nil
 }
+
+func (s *Server) GetReplicationID() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.replicationID
+}

doc/RDB File Format.html

@@ -16,6 +16,7 @@ var (
 
 type ServerInfoProvider interface {
 	GetReplicaInfo() []map[string]string
+	GetReplicationID() string
 }
 
 type Command interface {
@@ -74,6 +75,10 @@ func NewCommandFactory(kv keyval.KV, cfg *config.Config, serverInfo ServerInfoPr
 			"xrange":   &XRange{kv: kv},
 			"xread":    &XRead{kv: kv},
 			"replconf": &ReplConf{},
+			"psync": &Psync{
+				serverInfo: serverInfo,
+			},
+			"save": &Save{kv: kv},
 		},
 	}
 }

dump2.rdb

@@ -7,7 +7,6 @@ import (
 	"github.com/codecrafters-io/redis-starter-go/internal/app/server/config"
 	"github.com/codecrafters-io/redis-starter-go/internal/client"
 	"github.com/codecrafters-io/redis-starter-go/pkg/resp"
-	"github.com/codecrafters-io/redis-starter-go/pkg/utils"
 )
 
 const (
@@ -36,7 +35,7 @@ var sections = map[string]SectionInfo{
 	REPLICATION: {
 		DynamicFields: map[string]DynamicFieldHandler{
 			"role":               determineRole,
-			"master_replid":      generateMasterReplID,
+			"master_replid":      getMasterReplID,
 			"master_repl_offset": getMasterReplOffset,
 			"connected_slaves": func(cfg *config.Config, serverInfo ServerInfoProvider) string {
 				return fmt.Sprintf("%d", len(serverInfo.GetReplicaInfo()))
@@ -107,10 +106,6 @@ func determineRole(cfg *config.Config, _ ServerInfoProvider) string {
 	return "master"
 }
 
-func generateMasterReplID(_ *config.Config, _ ServerInfoProvider) string {
-	return utils.GenerateRandomAlphanumeric(40)
-}
-
 func getMasterReplOffset(_ *config.Config, _ ServerInfoProvider) string {
 	return "0"
 }
@@ -125,3 +120,7 @@ func buildReplicaInfo(cfg *config.Config, serverInfo ServerInfoProvider) string
 	}
 	return sb.String()
 }
+
+func getMasterReplID(_ *config.Config, serverInfo ServerInfoProvider) string {
+	return serverInfo.GetReplicationID()
+}

internal/app/server/server.go

@@ -0,0 +1,34 @@
+package command
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/codecrafters-io/redis-starter-go/internal/client"
+	"github.com/codecrafters-io/redis-starter-go/pkg/resp"
+)
+
+type Psync struct {
+	serverInfo ServerInfoProvider
+}
+
+func (p *Psync) Execute(c *client.Client, wr *resp.Writer, args []*resp.Resp) error {
+	if len(args) != 2 {
+		return wr.WriteError(errors.New("wrong number of arguments for 'psync' command"))
+	}
+
+	replicationID := args[0].String()
+	offset := args[1].String()
+
+	if replicationID != "?" || offset != "-1" {
+		return wr.WriteError(errors.New("invalid arguments for 'psync' command"))
+	}
+
+	response := fmt.Sprintf("FULLRESYNC %s 0", p.serverInfo.GetReplicationID())
+
+	return wr.WriteSimpleValue(resp.SimpleString, []byte(response))
+}
+
+func (p *Psync) IsBlocking(_ []*resp.Resp) bool {
+	return false
+}

pkg/command/command.go

@@ -12,7 +12,7 @@ import (
 type ReplConf struct{}
 
 func (rc *ReplConf) Execute(c *client.Client, wr *resp.Writer, args []*resp.Resp) error {
-	if len(args) < 2 {
+	if len(args) < 1 {
 		return wr.WriteError(errors.New("wrong number of arguments for 'replconf' command"))
 	}
 
@@ -25,11 +25,10 @@ func (rc *ReplConf) Execute(c *client.Client, wr *resp.Writer, args []*resp.Resp
 		port := args[1].String()
 		c.ListeningPort = port
 	case "capa":
-		if len(args) != 2 {
+		if len(args) < 2 {
 			return wr.WriteError(errors.New("wrong number of arguments for 'replconf capa' command"))
 		}
-		// capability := args[1].String()
-		// Ignore capability for now
+		// We don't need to handle/save the capa arguments
 	default:
 		return wr.WriteError(fmt.Errorf("unknown replconf subcommand: %s", subCommand))
 	}

pkg/command/info.go

@@ -0,0 +1,35 @@
+package command
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/codecrafters-io/redis-starter-go/internal/client"
+	"github.com/codecrafters-io/redis-starter-go/pkg/keyval"
+	"github.com/codecrafters-io/redis-starter-go/pkg/rdb"
+	"github.com/codecrafters-io/redis-starter-go/pkg/resp"
+)
+
+// SaveCommand implements the SAVE command.
+type Save struct {
+	kv keyval.KV
+}
+
+func (c *Save) Execute(cl *client.Client, wr *resp.Writer, args []*resp.Resp) error {
+	file, err := os.Create("dump2.rdb")
+	if err != nil {
+		return wr.WriteError(fmt.Errorf("failed to create dump.rdb: %w", err))
+	}
+	defer file.Close()
+
+	saver := rdb.NewRDBSaver(c.kv.Export())
+	if err := saver.SaveRDB(file); err != nil {
+		return wr.WriteError(fmt.Errorf("failed to save RDB: %w", err))
+	}
+
+	return wr.WriteSimpleValue(resp.SimpleString, []byte("OK"))
+}
+
+func (c *Save) IsBlocking(_ []*resp.Resp) bool {
+	return true
+}

pkg/command/psync.go

@@ -46,6 +46,8 @@ type Value struct {
 
 type KV interface {
 	RestoreRDB(data map[string]Value)
+	GetExpiry(key string) uint64
+	IsExpired(key string) bool
 	Get(key string) []byte
 	Set(key string, value []byte) error
 	Expire(key string, duration time.Duration)
@@ -57,6 +59,7 @@ type KV interface {
 	Keys() []string
 	Type(key string) string
 	GetStream(key string, createIfNotExists bool) (*Stream, error)
+	Export() map[string]Value
 }
 
 type kv struct {
@@ -188,3 +191,31 @@ func (kv *kv) GetStream(key string, createIfNotExist bool) (*Stream, error) {
 	}
 	return s.Data.(*Stream), nil
 }
+
+func (kv *kv) GetExpiry(key string) uint64 {
+	kv.mu.RLock()
+	defer kv.mu.RUnlock()
+	if v, ok := kv.store[key]; ok {
+		return v.Expiry
+	}
+	return 0
+}
+
+func (kv *kv) IsExpired(key string) bool {
+	kv.mu.RLock()
+	defer kv.mu.RUnlock()
+	if v, ok := kv.store[key]; ok {
+		return v.Expiry > 0 && v.Expiry < uint64(time.Now().UnixMilli())
+	}
+	return false
+}
+
+func (kv *kv) Export() map[string]Value {
+	kv.mu.RLock()
+	defer kv.mu.RUnlock()
+	exportedData := make(map[string]Value, len(kv.store))
+	for k, v := range kv.store {
+		exportedData[k] = v
+	}
+	return exportedData
+}

pkg/command/replconf.go

@@ -201,6 +201,14 @@ func (p *RDBParser) handlerAuxiliaryField(ctx context.Context) error {
 			return fmt.Errorf("failed to parse used-mem: %w", err)
 		}
 		logger.Info(ctx, "RDB memory usage when created %.2f MB", float64(usedMem)/1024/1024)
+	case "redis-bits":
+		redisBits, err := strconv.ParseInt(value, 10, 64)
+		if err != nil {
+			return fmt.Errorf("failed to parse redis-bits: %w", err)
+		}
+		logger.Info(ctx, "RDB is %d bit", redisBits)
+	case "aof-base":
+		logger.Info(ctx, "AOF enabled: %s", value)
 	}
 
 	return nil