Implementing GC Using Version Vector

api/converter/from_pb.go

@@ -102,12 +102,25 @@ func FromChangePack(pbPack *api.ChangePack) (*change.Pack, error) {
 		return nil, err
 	}
 
+	versionVector, err := FromVersionVector(pbPack.VersionVector)
+	if err != nil {
+		return nil, err
+	}
+
 	pack := &change.Pack{
 		DocumentKey:     key.Key(pbPack.DocumentKey),
 		Checkpoint:      fromCheckpoint(pbPack.Checkpoint),
 		Changes:         changes,
 		MinSyncedTicket: minSyncedTicket,
 		IsRemoved:       pbPack.IsRemoved,
+		VersionVector:   versionVector,
+	}
+
+	if pbPack.MinSyncedVersionVector != nil {
+		pack.MinSyncedVersionVector, err = FromVersionVector(pbPack.MinSyncedVersionVector)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	if pbPack.Snapshot != nil {

api/converter/to_pb.go

@@ -136,12 +136,18 @@ func ToChangePack(pack *change.Pack) (*api.ChangePack, error) {
 		return nil, err
 	}
 
+	pbVersionVector, err := ToVersionVector(pack.VersionVector)
+	if err != nil {
+		return nil, err
+	}
+
 	return &api.ChangePack{
 		DocumentKey:     pack.DocumentKey.String(),
 		Checkpoint:      ToCheckpoint(pack.Checkpoint),
 		Changes:         pbChanges,
 		Snapshot:        pack.Snapshot,
 		MinSyncedTicket: ToTimeTicket(pack.MinSyncedTicket),
+		VersionVector:   pbVersionVector,
 		IsRemoved:       pack.IsRemoved,
 	}, nil
 }

api/docs/yorkie/v1/resources.openapi.yaml

@@ -259,7 +259,7 @@ components:
         minSyncedTicket:
           $ref: '#/components/schemas/yorkie.v1.TimeTicket'
           additionalProperties: false
-          description: ""
+          description: Deprecated
           title: min_synced_ticket
           type: object
         minSyncedVersionVector:
@@ -280,6 +280,12 @@ components:
           description: ""
           title: snapshot_version_vector
           type: object
+        versionVector:
+          $ref: '#/components/schemas/yorkie.v1.VersionVector'
+          additionalProperties: false
+          description: ""
+          title: version_vector
+          type: object
       title: ChangePack
       type: object
     yorkie.v1.Checkpoint:

api/docs/yorkie/v1/yorkie.openapi.yaml

@@ -467,7 +467,7 @@ components:
         minSyncedTicket:
           $ref: '#/components/schemas/yorkie.v1.TimeTicket'
           additionalProperties: false
-          description: ""
+          description: Deprecated
           title: min_synced_ticket
           type: object
         minSyncedVersionVector:
@@ -488,6 +488,12 @@ components:
           description: ""
           title: snapshot_version_vector
           type: object
+        versionVector:
+          $ref: '#/components/schemas/yorkie.v1.VersionVector'
+          additionalProperties: false
+          description: ""
+          title: version_vector
+          type: object
       title: ChangePack
       type: object
     yorkie.v1.Checkpoint:

api/yorkie/v1/resources.proto

@@ -48,6 +48,7 @@ message ChangePack {
   TimeTicket min_synced_ticket = 5; // Deprecated
   VersionVector min_synced_version_vector = 8;
   bool is_removed = 6;
+  VersionVector version_vector = 9;
 }
 
 message Change {

pkg/document/change/id.go

@@ -101,11 +101,11 @@ func (id ID) NewTimeTicket(delimiter uint32) *time.Ticket {
 func (id ID) SyncClocks(other ID) ID {
 	lamport := id.lamport + 1
 	if id.lamport < other.lamport {
-		lamport = other.lamport
+		lamport = other.lamport + 1
 	}
 
-	newID := NewID(id.clientSeq, InitialServerSeq, lamport, id.actorID, id.versionVector)
-	newID.versionVector.Set(other.actorID, other.lamport)
+	newID := NewID(id.clientSeq, InitialServerSeq, lamport, id.actorID, id.versionVector.Max(other.versionVector))
+	newID.versionVector.Set(id.actorID, lamport)
 	return newID
 }
 
@@ -114,16 +114,18 @@ func (id ID) SyncClocks(other ID) ID {
 func (id ID) SetClocks(otherLamport int64, vector time.VersionVector) ID {
 	lamport := id.lamport + 1
 	if id.lamport < otherLamport {
-		lamport = otherLamport
+		lamport = otherLamport + 1
 	}
 
-	return NewID(
-		id.clientSeq,
-		id.serverSeq,
-		lamport,
-		id.actorID,
-		vector,
-	)
+	newID := NewID(id.clientSeq, id.serverSeq, lamport, id.actorID, id.versionVector.Max(vector))
+	newID.versionVector.Set(id.actorID, lamport)
+
+	return newID
+}
+
+// SetVersionVector sets version vector
+func (id ID) SetVersionVector(vector time.VersionVector) ID {
+	return NewID(id.clientSeq, id.serverSeq, id.lamport, id.actorID, vector)
 }
 
 // SetActor sets actorID.

pkg/document/change/pack.go

@@ -35,6 +35,9 @@ type Pack struct {
 	// Snapshot is a byte array that encode the document.
 	Snapshot []byte
 
+	// VersionVector is the version vector of the document.
+	VersionVector time.VersionVector
+
 	// SnapshotVersionVector is the version vector of the snapshot if it exists.
 	SnapshotVersionVector time.VersionVector
 
@@ -55,13 +58,15 @@ func NewPack(
 	key key.Key,
 	cp Checkpoint,
 	changes []*Change,
+	versionVector time.VersionVector,
 	snapshot []byte,
 ) *Pack {
 	return &Pack{
-		DocumentKey: key,
-		Checkpoint:  cp,
-		Changes:     changes,
-		Snapshot:    snapshot,
+		DocumentKey:   key,
+		Checkpoint:    cp,
+		Changes:       changes,
+		VersionVector: versionVector,
+		Snapshot:      snapshot,
 	}
 }
 

pkg/document/crdt/root.go

@@ -147,7 +147,7 @@ func (r *Root) GarbageCollect(vector time.VersionVector) (int, error) {
 	count := 0
 
 	for _, pair := range r.gcElementPairMap {
-		if vector.After(pair.elem.RemovedAt()) {
+		if vector.EqualToOrAfter(pair.elem.RemovedAt()) {
 			if err := pair.parent.Purge(pair.elem); err != nil {
 				return 0, err
 			}
@@ -157,7 +157,7 @@ func (r *Root) GarbageCollect(vector time.VersionVector) (int, error) {
 	}
 
 	for _, pair := range r.gcNodePairMap {
-		if vector.After(pair.Child.RemovedAt()) {
+		if vector.EqualToOrAfter(pair.Child.RemovedAt()) {
 			if err := pair.Parent.Purge(pair.Child); err != nil {
 				return 0, err
 			}

pkg/document/document.go

@@ -219,7 +219,12 @@ func (d *Document) ApplyChangePack(pack *change.Pack) error {
 		d.GarbageCollect(pack.MinSyncedVersionVector)
 	}
 
-	// 05. Update the status.
+	// 05. Remove detached client's lamport from version vector if it exists
+	if pack.MinSyncedVersionVector != nil {
+		d.doc.changeID = d.doc.changeID.SetVersionVector(d.doc.changeID.VersionVector().Filter(pack.MinSyncedVersionVector.Keys()))
+	}
+
+	// 06. Update the status.
 	if pack.IsRemoved {
 		d.SetStatus(StatusRemoved)
 	}

pkg/document/document_test.go

@@ -558,14 +558,14 @@ func TestDocument(t *testing.T) {
 		docB.SetActor(actorB)
 		assert.Equal(t, "{}", docB.VersionVector().Marshal())
 		assert.NoError(t, docB.ApplyChangePack(packA))
-		assert.Equal(t, "{000000000000000000000001:2}", docB.VersionVector().Marshal())
+		assert.Equal(t, "{000000000000000000000001:2,000000000000000000000002:3}", docB.VersionVector().Marshal())
 
 		assert.NoError(t, docB.Update(func(r *json.Object, p *presence.Presence) error {
 			r.SetString("k2", "3")
 			return nil
 		}))
 		assert.Equal(t, int64(2), docB.VersionVector().VersionOf(actorA))
-		assert.Equal(t, int64(3), docB.VersionVector().VersionOf(actorB))
+		assert.Equal(t, int64(4), docB.VersionVector().VersionOf(actorB))
 		packB := docB.CreateChangePack()
 		packB.MinSyncedTicket = time.InitialTicket
 		assert.True(t, packB.Changes[0].AfterOrEqual(packA.Changes[1]))

pkg/document/internal_document.go

@@ -173,6 +173,11 @@ func (d *InternalDocument) ApplyChangePack(pack *change.Pack, disableGC bool) er
 		}
 	}
 
+	// 04. Remove detached client's lamport from version vector if it exists
+	if pack.MinSyncedVersionVector != nil {
+		d.changeID = d.changeID.SetVersionVector(d.changeID.VersionVector().Filter(pack.MinSyncedVersionVector.Keys()))
+	}
+
 	return nil
 }
 
@@ -196,7 +201,7 @@ func (d *InternalDocument) CreateChangePack() *change.Pack {
 	changes := d.localChanges
 
 	cp := d.checkpoint.IncreaseClientSeq(uint32(len(changes)))
-	return change.NewPack(d.key, cp, changes, nil)
+	return change.NewPack(d.key, cp, changes, d.VersionVector(), nil)
 }
 
 // SetActor sets actor into this document. This is also applied in the local

pkg/document/time/version_vector.go

@@ -17,6 +17,8 @@
 package time
 
 import (
+	"bytes"
+	"sort"
 	"strconv"
 	"strings"
 )
@@ -38,6 +40,11 @@ func (v VersionVector) Set(id *ActorID, i int64) {
 	v[id.bytes] = i
 }
 
+// Unset unsets the given actor's version to the given value.
+func (v VersionVector) Unset(id *ActorID) {
+	delete(v, id.bytes)
+}
+
 // VersionOf returns the version of the given actor.
 func (v VersionVector) VersionOf(id *ActorID) int64 {
 	return v[id.bytes]
@@ -58,8 +65,16 @@ func (v VersionVector) Marshal() string {
 
 	builder.WriteRune('{')
 
+	keys := make([]actorID, 0, len(v))
+	for k := range v {
+		keys = append(keys, k)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		return bytes.Compare(keys[i][:], keys[j][:]) < 0
+	})
+
 	isFirst := true
-	for k, val := range v {
+	for _, k := range keys {
 		if !isFirst {
 			builder.WriteRune(',')
 		}
@@ -71,7 +86,7 @@ func (v VersionVector) Marshal() string {
 
 		builder.WriteString(id.String())
 		builder.WriteRune(':')
-		builder.WriteString(strconv.FormatInt(val, 10))
+		builder.WriteString(strconv.FormatInt(v[k], 10))
 		isFirst = false
 	}
 	builder.WriteRune('}')
@@ -97,9 +112,91 @@ func (v VersionVector) AfterOrEqual(other VersionVector) bool {
 	return true
 }
 
-// After returns whether this VersionVector is causally after the given ticket.
-func (v VersionVector) After(other *Ticket) bool {
-	actorID := other.ActorID()
-	ticket := NewTicket(v.VersionOf(actorID), MaxDelimiter, actorID)
-	return ticket.Compare(other) >= 0
+// EqualToOrAfter returns whether this VersionVector's every field is equal or after than given ticket.
+func (v VersionVector) EqualToOrAfter(other *Ticket) bool {
+	return v[other.actorID.bytes] >= other.lamport
+}
+
+// Min returns new vv consists of every min value in each column.
+func (v VersionVector) Min(other VersionVector) VersionVector {
+	minVV := NewVersionVector()
+
+	for key, value := range v {
+		if otherValue, exists := other[key]; exists {
+			if value < otherValue {
+				minVV[key] = value
+			} else {
+				minVV[key] = otherValue
+			}
+		} else {
+			minVV[key] = 0
+		}
+	}
+
+	for key, _ := range other {
+		if _, exists := v[key]; !exists {
+			minVV[key] = 0
+		}
+	}
+
+	return minVV
+}
+
+// Max returns new vv consists of every max value in each column.
+func (v VersionVector) Max(other VersionVector) VersionVector {
+	maxVV := NewVersionVector()
+
+	for key, value := range v {
+		if otherValue, exists := other[key]; exists {
+			if value > otherValue {
+				maxVV[key] = value
+			} else {
+				maxVV[key] = otherValue
+			}
+		} else {
+			maxVV[key] = value
+		}
+	}
+
+	for key, value := range other {
+		if _, exists := v[key]; !exists {
+			maxVV[key] = value
+		}
+	}
+
+	return maxVV
+}
+
+// MaxLamport returns new vv consists of every max value in each column.
+func (v VersionVector) MaxLamport() int64 {
+	var maxLamport int64 = -1
+
+	for _, value := range v {
+		if value > maxLamport {
+			maxLamport = value
+		}
+	}
+
+	return maxLamport
+}
+
+// Filter returns filtered version vector which keys are only from filter
+func (v VersionVector) Filter(filter []actorID) VersionVector {
+	filteredVV := NewVersionVector()
+
+	for _, value := range filter {
+		filteredVV[value] = v[value]
+	}
+
+	return filteredVV
+}
+
+// Keys returns filtered version vector which keys are only from filter
+func (v VersionVector) Keys() []actorID {
+	var actors []actorID
+	for id := range v {
+		actors = append(actors, id)
+	}
+
+	return actors
 }