Fix snapshot just return the latestVectorClock not SVM

1. Now snapshot save latest vector clock not SVM. 2. Clients build their own SVM using the latest VectorClock in snapshots. - So client's SVM is not latest state, it doesn't know peer's vector clock. It need more check - This is due to the overhead of storing and transferring SVMs. #789
yorkie-team · Feb 12, 2024 · f6d9a58 · f6d9a58 · github-actions · Feb 12, 2024
1 parent 679cd90
commit f6d9a58
Show file tree

Hide file tree

Showing 18 changed files with 868 additions and 820 deletions.
diff --git a/admin/client.go b/admin/client.go
@@ -332,7 +332,7 @@ func (c *Client) ListChangeSummaries(
 		seq,
 		snapshotMeta.Msg.Lamport,
 		snapshotMeta.Msg.Snapshot,
-		snapshotMeta.Msg.SyncedVectorMap,
+		snapshotMeta.Msg.LatestVectorClock,
 	)
 
 	if err != nil {

diff --git a/api/converter/from_pb.go b/api/converter/from_pb.go
@@ -102,19 +102,14 @@ func FromChangePack(pbPack *api.ChangePack) (*change.Pack, error) {
 		return nil, err
 	}
 
-	syncedVectorMap, err := time.NewSyncedVectorMapFromJSON(pbPack.SyncedVectorMap)
-	if err != nil {
-		return nil, err
-	}
-
 	return &change.Pack{
-		DocumentKey:     key.Key(pbPack.DocumentKey),
-		Checkpoint:      fromCheckpoint(pbPack.Checkpoint),
-		Changes:         changes,
-		Snapshot:        pbPack.Snapshot,
-		MinSyncedTicket: minSyncedTicket,
-		IsRemoved:       pbPack.IsRemoved,
-		SyncedVectorMap: syncedVectorMap,
+		DocumentKey:       key.Key(pbPack.DocumentKey),
+		Checkpoint:        fromCheckpoint(pbPack.Checkpoint),
+		Changes:           changes,
+		Snapshot:          pbPack.Snapshot,
+		MinSyncedTicket:   minSyncedTicket,
+		IsRemoved:         pbPack.IsRemoved,
+		LatestVectorClock: pbPack.LatestVectorClock,
 	}, nil
 }
 

diff --git a/api/yorkie/v1/admin.pb.go b/api/yorkie/v1/admin.pb.go
diff --git a/api/yorkie/v1/admin.proto b/api/yorkie/v1/admin.proto
@@ -130,7 +130,7 @@ message GetSnapshotMetaRequest {
 message GetSnapshotMetaResponse {
   bytes snapshot = 1;
   int64 lamport = 2  [jstype = JS_STRING];
-  string synced_vector_map = 3;
+  string latest_vector_clock = 3;
 }
 
 message SearchDocumentsRequest {

diff --git a/api/yorkie/v1/resources.pb.go b/api/yorkie/v1/resources.pb.go
diff --git a/api/yorkie/v1/resources.proto b/api/yorkie/v1/resources.proto
@@ -46,7 +46,7 @@ message ChangePack {
   repeated Change changes = 4;
   TimeTicket min_synced_ticket = 5;
   bool is_removed = 6;
-  string synced_vector_map = 7;
+  string latest_vector_clock = 7;
 }
 
 message Change {

diff --git a/pkg/document/change/pack.go b/pkg/document/change/pack.go
@@ -42,8 +42,8 @@ type Pack struct {
 	// IsRemoved is a flag that indicates whether the document is removed.
 	IsRemoved bool
 
-	// SyncedVectorMap is SnapShot's synced vector map information.
-	SyncedVectorMap time.SyncedVectorMap
+	// LatestVectorClock is the latest vector clock of the document.
+	LatestVectorClock string
 }
 
 // NewPack creates a new instance of Pack.

diff --git a/pkg/document/document.go b/pkg/document/document.go
@@ -202,7 +202,7 @@ func (d *Document) ApplyChangePack(pack *change.Pack) error {
 		d.cloneRoot = nil
 		d.clonePresences = nil
 		if err := d.doc.applySnapshot(pack.Snapshot, pack.Checkpoint.ServerSeq,
-			pack.SyncedVectorMap); err != nil {
+			pack.LatestVectorClock); err != nil {
 			return err
 		}
 	} else {

diff --git a/pkg/document/internal_document.go b/pkg/document/internal_document.go
@@ -114,20 +114,21 @@ func NewInternalDocumentFromSnapshot(
 	serverSeq int64,
 	lamport int64,
 	snapshot []byte,
-	syncedVectorMap string,
+	LatestVectorClock string,
 ) (*InternalDocument, error) {
 	obj, presences, err := converter.BytesToSnapshot(snapshot)
 	if err != nil {
 		return nil, err
 	}
 
-	svm, err := time.NewSyncedVectorMapFromJSON(syncedVectorMap)
+	id := change.InitialID.SyncLamport(lamport)
+
+	// Create a new instance of SyncedVectorMap from the given LatestVectorClock.
+	svm, err := time.NewSVMFromLatestVectorClock(LatestVectorClock)
 	if err != nil {
 		return nil, err
 	}
 
-	id := change.InitialID.SyncLamport(lamport)
-
 	if svm == nil {
 		svm = time.InitialSyncedVectorMap(id.Lamport())
 	}
@@ -163,7 +164,7 @@ func (d *InternalDocument) HasLocalChanges() bool {
 func (d *InternalDocument) ApplyChangePack(pack *change.Pack) error {
 	// 01. Apply remote changes to both the cloneRoot and the document.
 	if len(pack.Snapshot) > 0 {
-		if err := d.applySnapshot(pack.Snapshot, pack.Checkpoint.ServerSeq, pack.SyncedVectorMap); err != nil {
+		if err := d.applySnapshot(pack.Snapshot, pack.Checkpoint.ServerSeq, pack.LatestVectorClock); err != nil {
 			return err
 		}
 	} else {
@@ -256,17 +257,30 @@ func (d *InternalDocument) RootObject() *crdt.Object {
 func (d *InternalDocument) applySnapshot(
 	snapshot []byte,
 	serverSeq int64,
-	syncedVectorMap time.SyncedVectorMap,
+	syncedVectorMap string,
 ) error {
 	rootObj, presences, err := converter.BytesToSnapshot(snapshot)
 	if err != nil {
 		return err
 	}
 
+	svm, err := time.NewSVMFromLatestVectorClock(syncedVectorMap)
+	if err != nil {
+		return err
+	}
+
+	// Remove the initial actor's vector clock and replace it with the client's vector clock.
+	// This is because the initial actor's vector clock is used for the snapshot.
+	initID := time.InitialActorID.String()
+	delete(svm, d.ActorID().String())
+	delete(svm[initID], initID)
+	svm[d.ActorID().String()] = svm[time.InitialActorID.String()]
+	delete(svm, initID)
+
 	d.root = crdt.NewRoot(rootObj)
 	d.presences = presences
 	d.changeID = d.changeID.SyncLamport(serverSeq)
-	d.syncedVectorMap = syncedVectorMap
+	d.syncedVectorMap = svm
 
 	return nil
 }

diff --git a/pkg/document/time/vector_clock.go b/pkg/document/time/vector_clock.go
@@ -35,34 +35,6 @@ func (vc VectorClock) EncodeToString() (string, error) {
 	return string(bytes), nil
 }
 
-// NewSyncedVectorMapFromJSON creates a new instance of SyncedVectorMap.
-func NewSyncedVectorMapFromJSON(encodedChange string) (SyncedVectorMap, error) {
-	if encodedChange == "" {
-		return nil, nil
-	}
-
-	vc := SyncedVectorMap{}
-	if err := json.Unmarshal([]byte(encodedChange), &vc); err != nil {
-		return nil, fmt.Errorf("unmarshal vector clock: %w", err)
-	}
-
-	return vc, nil
-}
-
-// EncodeToString encodes the given syncedVectorMap to string.
-func (svm SyncedVectorMap) EncodeToString() (string, error) {
-	if svm == nil {
-		return "", nil
-	}
-
-	bytes, err := json.Marshal(svm)
-	if err != nil {
-		return "", fmt.Errorf("marshal syncedVectorMap to bytes: %w", err)
-	}
-
-	return string(bytes), nil
-}
-
 // InitialSyncedVectorMap creates an initial synced vector map.
 func InitialSyncedVectorMap(lamport int64) SyncedVectorMap {
 	actorID := InitialActorID.String()
@@ -134,3 +106,24 @@ func (svm SyncedVectorMap) ChangeActorID(id *ActorID) {
 	delete(svm[newID], initID)
 	delete(svm, initID)
 }
+
+// NewSVMFromLatestVectorClock creates a new instance of SyncedVectorMap from the given latestVectorClock.
+func NewSVMFromLatestVectorClock(latestVectorClock string) (SyncedVectorMap, error) {
+	if latestVectorClock == "" {
+		return nil, nil
+	}
+
+	vc := VectorClock{}
+	if err := json.Unmarshal([]byte(latestVectorClock), &vc); err != nil {
+		return nil, fmt.Errorf("unmarshal vector clock: %w", err)
+	}
+
+	svm := SyncedVectorMap{}
+
+	for k := range vc {
+		svm[k] = VectorClock{}
+	}
+	svm[InitialActorID.String()] = vc
+
+	return svm, nil
+}
diff --git a/server/backend/database/mongo/client.go b/server/backend/database/mongo/client.go
@@ -975,19 +975,19 @@ func (c *Client) CreateSnapshotInfo(
 		return err
 	}
 
-	syncedVectorMap, err := doc.SyncedVectorMap().EncodeToString()
+	latestVectorClock, err := doc.SyncedVectorMap()[doc.ActorID().String()].EncodeToString()
 	if err != nil {
 		return err
 	}
 
 	if _, err := c.collection(ColSnapshots).InsertOne(ctx, bson.M{
-		"project_id":        docRefKey.ProjectID,
-		"doc_id":            docRefKey.DocID,
-		"server_seq":        doc.Checkpoint().ServerSeq,
-		"lamport":           doc.Lamport(),
-		"snapshot":          snapshot,
-		"created_at":        gotime.Now(),
-		"synced_vector_map": syncedVectorMap,
+		"project_id":          docRefKey.ProjectID,
+		"doc_id":              docRefKey.DocID,
+		"server_seq":          doc.Checkpoint().ServerSeq,
+		"lamport":             doc.Lamport(),
+		"snapshot":            snapshot,
+		"created_at":          gotime.Now(),
+		"latest_vector_clock": latestVectorClock,
 	}); err != nil {
 		return fmt.Errorf("insert snapshot: %w", err)
 	}

diff --git a/server/backend/database/snapshot_info.go b/server/backend/database/snapshot_info.go
@@ -45,8 +45,8 @@ type SnapshotInfo struct {
 	// CreatedAt is the time when the snapshot is created.
 	CreatedAt time.Time `bson:"created_at"`
 
-	// SyncedVectorMap is the synced vector map of the snapshot.
-	SyncedVectorMap string `bson:"synced_vector_map"`
+	// LatestVectorClock is the latest vector clock of the snapshot.
+	LatestVectorClock string `bson:"synced_vector_map"`
 }
 
 // DeepCopy returns a deep copy of the SnapshotInfo.
@@ -56,14 +56,14 @@ func (i *SnapshotInfo) DeepCopy() *SnapshotInfo {
 	}
 
 	return &SnapshotInfo{
-		ID:              i.ID,
-		ProjectID:       i.ProjectID,
-		DocID:           i.DocID,
-		ServerSeq:       i.ServerSeq,
-		Lamport:         i.Lamport,
-		Snapshot:        i.Snapshot,
-		CreatedAt:       i.CreatedAt,
-		SyncedVectorMap: i.SyncedVectorMap,
+		ID:                i.ID,
+		ProjectID:         i.ProjectID,
+		DocID:             i.DocID,
+		ServerSeq:         i.ServerSeq,
+		Lamport:           i.Lamport,
+		Snapshot:          i.Snapshot,
+		CreatedAt:         i.CreatedAt,
+		LatestVectorClock: i.LatestVectorClock,
 	}
 }
 

diff --git a/server/packs/packs.go b/server/packs/packs.go
@@ -213,7 +213,7 @@ func BuildDocumentForServerSeq(
 		snapshotInfo.ServerSeq,
 		snapshotInfo.Lamport,
 		snapshotInfo.Snapshot,
-		snapshotInfo.SyncedVectorMap,
+		snapshotInfo.LatestVectorClock,
 	)
 	if err != nil {
 		return nil, err

diff --git a/server/packs/pushpull.go b/server/packs/pushpull.go
@@ -24,7 +24,6 @@ import (
 	"github.com/yorkie-team/yorkie/api/converter"
 	"github.com/yorkie-team/yorkie/api/types"
 	"github.com/yorkie-team/yorkie/pkg/document/change"
-	"github.com/yorkie-team/yorkie/pkg/document/time"
 	"github.com/yorkie-team/yorkie/server/backend"
 	"github.com/yorkie-team/yorkie/server/backend/database"
 	"github.com/yorkie-team/yorkie/server/logging"
@@ -157,22 +156,14 @@ func pullSnapshot(
 		}
 	}
 
-	// Remove the initial actor's vector clock and replace it with the client's vector clock.
-	// This is because the initial actor's vector clock is used for the snapshot.
-	initID := time.InitialActorID.String()
-	delete(doc.SyncedVectorMap(), string(clientInfo.ID))
-	delete(doc.SyncedVectorMap()[initID], initID)
-	doc.SyncedVectorMap()[clientInfo.ID.String()] = doc.SyncedVectorMap()[time.InitialActorID.String()]
-	delete(doc.SyncedVectorMap(), initID)
-
 	cpAfterPull := cpAfterPush.NextServerSeq(docInfo.ServerSeq)
 
 	snapshot, err := converter.SnapshotToBytes(doc.RootObject(), doc.AllPresences())
 	if err != nil {
 		return nil, err
 	}
 
-	syncedVectorMap, err := doc.SyncedVectorMap().EncodeToString()
+	latestVectorClock, err := doc.SyncedVectorMap()[doc.ActorID().String()].EncodeToString()
 	if err != nil {
 		return nil, err
 	}
@@ -186,7 +177,7 @@ func pullSnapshot(
 		cpAfterPull,
 	)
 
-	return NewServerPack(docInfo.Key, cpAfterPull, nil, snapshot, syncedVectorMap), err
+	return NewServerPack(docInfo.Key, cpAfterPull, nil, snapshot, latestVectorClock), err
 }
 
 func pullChangeInfos(

diff --git a/server/packs/serverpacks.go b/server/packs/serverpacks.go
@@ -50,8 +50,8 @@ type ServerPack struct {
 	// IsRemoved is a flag that indicates whether the document is removed.
 	IsRemoved bool
 
-	// SyncedVectorMap is SnapShot's synced information.
-	SyncedVectorMap string
+	// LatestVectorClock is the latest vector clock of the document.
+	LatestVectorClock string
 }
 
 // NewServerPack creates a new instance of ServerPack.
@@ -60,14 +60,14 @@ func NewServerPack(
 	cp change.Checkpoint,
 	changeInfos []*database.ChangeInfo,
 	snapshot []byte,
-	syncedVectorMap string,
+	latestVectorClock string,
 ) *ServerPack {
 	return &ServerPack{
-		DocumentKey:     key,
-		Checkpoint:      cp,
-		ChangeInfos:     changeInfos,
-		Snapshot:        snapshot,
-		SyncedVectorMap: syncedVectorMap,
+		DocumentKey:       key,
+		Checkpoint:        cp,
+		ChangeInfos:       changeInfos,
+		Snapshot:          snapshot,
+		LatestVectorClock: latestVectorClock,
 	}
 }
 
@@ -130,13 +130,13 @@ func (p *ServerPack) ToPBChangePack() (*api.ChangePack, error) {
 	}
 
 	return &api.ChangePack{
-		DocumentKey:     p.DocumentKey.String(),
-		Checkpoint:      converter.ToCheckpoint(p.Checkpoint),
-		Changes:         pbChanges,
-		Snapshot:        p.Snapshot,
-		MinSyncedTicket: converter.ToTimeTicket(p.MinSyncedTicket),
-		IsRemoved:       p.IsRemoved,
-		SyncedVectorMap: p.SyncedVectorMap,
+		DocumentKey:       p.DocumentKey.String(),
+		Checkpoint:        converter.ToCheckpoint(p.Checkpoint),
+		Changes:           pbChanges,
+		Snapshot:          p.Snapshot,
+		MinSyncedTicket:   converter.ToTimeTicket(p.MinSyncedTicket),
+		IsRemoved:         p.IsRemoved,
+		LatestVectorClock: p.LatestVectorClock,
 	}, nil
 }
 

diff --git a/server/packs/snapshots.go b/server/packs/snapshots.go
@@ -78,7 +78,7 @@ func storeSnapshot(
 		snapshotInfo.ServerSeq,
 		snapshotInfo.Lamport,
 		snapshotInfo.Snapshot,
-		snapshotInfo.SyncedVectorMap,
+		snapshotInfo.LatestVectorClock,
 	)
 	if err != nil {
 		return err

diff --git a/server/rpc/admin_server.go b/server/rpc/admin_server.go
@@ -231,15 +231,15 @@ func (s *adminServer) GetSnapshotMeta(
 		return nil, err
 	}
 
-	syncedVectorMap, err := doc.SyncedVectorMap().EncodeToString()
+	latestVectorClock, err := doc.SyncedVectorMap()[doc.ActorID().String()].EncodeToString()
 	if err != nil {
 		return nil, err
 	}
 
 	return connect.NewResponse(&api.GetSnapshotMetaResponse{
-		Lamport:         doc.Lamport(),
-		Snapshot:        snapshot,
-		SyncedVectorMap: syncedVectorMap,
+		Lamport:           doc.Lamport(),
+		Snapshot:          snapshot,
+		LatestVectorClock: latestVectorClock,
 	}), nil
 }