Implement checking wether snapshot is in progress before continuing

src/raft/fms_snapshot.go

@@ -4,11 +4,14 @@ import (
 	"encoding/json"
 	"github.com/echovault/echovault/src/utils"
 	"github.com/hashicorp/raft"
+	"time"
 )
 
 type SnapshotOpts struct {
-	config utils.Config
+	config         utils.Config
-	data   map[string]interface{}
+	data           map[string]interface{}
+	startSnapshot  func()
+	finishSnapshot func()
 }
 
 type Snapshot struct {
@@ -23,7 +26,8 @@ func NewFSMSnapshot(opts SnapshotOpts) *Snapshot {
 
 // Persist implements FSMSnapshot interface
 func (s *Snapshot) Persist(sink raft.SnapshotSink) error {
-	// TODO: Turn on snapshot in-progress flag
+	s.options.startSnapshot()
+
 	o, err := json.Marshal(s.options.data)
 
 	if err != nil {
@@ -36,10 +40,12 @@ func (s *Snapshot) Persist(sink raft.SnapshotSink) error {
 		return err
 	}
 
+	<-time.After(5 * time.Second)
+
 	return nil
 }
 
 // Release implements FSMSnapshot interface
 func (s *Snapshot) Release() {
-	// TODO: Turn off snapshot in-progress flag
+	s.options.finishSnapshot()
 }

src/raft/fsm.go

@@ -76,8 +76,10 @@ func (fsm *FSM) Apply(log *raft.Log) interface{} {
 // Snapshot implements raft.FSM interface
 func (fsm *FSM) Snapshot() (raft.FSMSnapshot, error) {
 	return NewFSMSnapshot(SnapshotOpts{
-		config: fsm.options.Config,
+		config:         fsm.options.Config,
-		data:   fsm.options.Server.GetState(),
+		data:           fsm.options.Server.GetState(),
+		startSnapshot:  fsm.options.Server.StartSnapshot,
+		finishSnapshot: fsm.options.Server.FinishSnapshot,
 	}), nil
 }
 

src/raft/raft.go

@@ -144,10 +144,10 @@ func (r *Raft) HasJoinedCluster() bool {
 }
 
 func (r *Raft) AddVoter(
-	id raft.ServerID,
+		id raft.ServerID,
-	address raft.ServerAddress,
+		address raft.ServerAddress,
-	prevIndex uint64,
+		prevIndex uint64,
-	timeout time.Duration,
+		timeout time.Duration,
 ) error {
 	if r.IsRaftLeader() {
 		raftConfig := r.raft.GetConfiguration()

src/server/cluster.go

@@ -47,3 +47,11 @@ func (server *Server) raftApply(ctx context.Context, cmd []string) ([]byte, erro
 
 	return r.Response, nil
 }
+
+func (server *Server) StartSnapshot() {
+	server.SnapshotInProgress.Store(true)
+}
+
+func (server *Server) FinishSnapshot() {
+	server.SnapshotInProgress.Store(false)
+}

src/server/server.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"github.com/echovault/echovault/src/memberlist"
 	"github.com/echovault/echovault/src/modules/acl"
@@ -36,6 +37,8 @@ type Server struct {
 
 	ACL    *acl.ACL
 	PubSub *pubsub.PubSub
+
+	SnapshotInProgress atomic.Bool
 }
 
 func (server *Server) StartTCP(ctx context.Context) {
@@ -241,11 +244,17 @@ func (server *Server) Start(ctx context.Context) {
 }
 
 func (server *Server) TakeSnapshot() error {
-	// TODO: Check if there's a snapshot currently in progress
+	if server.SnapshotInProgress.Load() {
-	go func() {
+		return errors.New("snapshot already in progress")
-		err := server.raft.TakeSnapshot()
+	}
-		log.Println(err)
+	if server.IsInCluster() {
-	}()
+		// Handle snapshot in cluster mode
+		go func() {
+			err := server.raft.TakeSnapshot()
+			log.Println(err)
+		}()
+	}
+	// Handle snapshot in standalone mode
 	return nil
 }
 

src/utils/types.go

@@ -19,6 +19,8 @@ type Server interface {
 	GetACL() interface{}
 	GetPubSub() interface{}
 	TakeSnapshot() error
+	StartSnapshot()
+	FinishSnapshot()
 }
 
 type ContextServerID string