Clear presence when housekeeping process

client/client.go

@@ -382,6 +382,8 @@ func (c *Client) Detach(ctx context.Context, doc *document.Document, options ...
 		return err
 	}
 
+	// TODO(raararaara): We need to revert the presence clearing from the local
+	// changes, if the server fails to detach the document.
 	res, err := c.client.DetachDocument(
 		ctx,
 		withShardKey(connect.NewRequest(&api.DetachDocumentRequest{

pkg/document/document.go

@@ -454,6 +454,11 @@ func (d *Document) BroadcastEventHandlers() map[string]func(
 	return d.broadcastEventHandlers
 }
 
+// setInternalDoc sets the given internal document to this document.
+func (d *Document) setInternalDoc(internalDoc *InternalDocument) {
+	d.doc = internalDoc
+}
+
 func messageFromMsgAndArgs(msgAndArgs ...interface{}) string {
 	if len(msgAndArgs) == 0 {
 		return ""

pkg/document/internal_document.go

@@ -136,6 +136,18 @@ func (d *InternalDocument) Checkpoint() change.Checkpoint {
 	return d.checkpoint
 }
 
+// SyncCheckpoint syncs the checkpoint and the changeID with the given serverSeq
+// and clientSeq.
+func (d *InternalDocument) SyncCheckpoint(serverSeq int64, clientSeq uint32) {
+	d.changeID = change.NewID(
+		clientSeq,
+		serverSeq,
+		d.changeID.Lamport(),
+		d.changeID.ActorID(),
+	)
+	d.checkpoint = d.checkpoint.SyncClientSeq(clientSeq)
+}
+
 // HasLocalChanges returns whether this document has local changes or not.
 func (d *InternalDocument) HasLocalChanges() bool {
 	return len(d.localChanges) > 0
@@ -372,3 +384,10 @@ func (d *InternalDocument) AddOnlineClient(clientID string) {
 func (d *InternalDocument) RemoveOnlineClient(clientID string) {
 	d.onlineClients.Delete(clientID)
 }
+
+// ToDocument converts this document to Document.
+func (d *InternalDocument) ToDocument() *Document {
+	doc := New(d.key)
+	doc.setInternalDoc(d)
+	return doc
+}

server/clients/clients.go

@@ -22,7 +22,12 @@ import (
 	"errors"
 
 	"github.com/yorkie-team/yorkie/api/types"
+	"github.com/yorkie-team/yorkie/pkg/document"
+	"github.com/yorkie-team/yorkie/pkg/document/json"
+	"github.com/yorkie-team/yorkie/pkg/document/presence"
+	"github.com/yorkie-team/yorkie/server/backend"
 	"github.com/yorkie-team/yorkie/server/backend/database"
+	"github.com/yorkie-team/yorkie/server/packs"
 )
 
 var (
@@ -46,42 +51,78 @@ func Activate(
 // Deactivate deactivates the given client.
 func Deactivate(
 	ctx context.Context,
-	db database.Database,
+	be *backend.Backend,
 	refKey types.ClientRefKey,
 ) (*database.ClientInfo, error) {
-	// TODO(hackerwins): We need to remove the presence of the client from the document.
-	// Be careful that housekeeping is executed by the leader. And documents are sharded
-	// by the servers in the cluster. So, we need to consider the case where the leader is
-	// not the same as the server that handles the document.
-
-	// TODO(raararaara): When deactivating a client, we need to update three DB properties
-	// (ClientInfo.Status, ClientInfo.Documents, SyncedSeq) in DB.
-	// Updating the sub-properties of ClientInfo guarantees atomicity as it involves a single MongoDB document.
-	// However, SyncedSeqs are stored in separate documents, so we can't ensure atomic updates for both.
-	// Currently, if SyncedSeqs update fails, it mainly impacts GC efficiency without causing major issues.
-	// We need to consider implementing a correction logic to remove SyncedSeqs in the future.
-	clientInfo, err := db.DeactivateClient(ctx, refKey)
+	// NOTE(hackerwins): Before deactivating the client, we need to detach all
+	// attached documents from the client.
+	// Because detachments and deactivation are separate steps, failure of steps
+	// must be considered. If each step of detachments is failed, some documents
+	// are still attached and the client is not deactivated. In this case,
+	// the client or housekeeping process should retry the deactivation.
+	clientInfo, err := be.DB.FindClientInfoByRefKey(ctx, refKey)
+	if err != nil {
+		return nil, err
+	}
+
+	// 01. Detach attached documents from the client.
+	actorID, err := clientInfo.ID.ToActorID()
+	if err != nil {
+		return nil, err
+	}
+
+	projectInfo, err := be.DB.FindProjectInfoByID(ctx, clientInfo.ProjectID)
 	if err != nil {
 		return nil, err
 	}
+	project := projectInfo.ToProject()
+
+	for docID, info := range clientInfo.Documents {
+		if info.Status != database.DocumentAttached {
+			continue
+		}
+
+		docInfo, err := be.DB.FindDocInfoByRefKey(ctx, types.DocRefKey{
+			ProjectID: clientInfo.ProjectID,
+			DocID:     docID,
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		doc, err := packs.BuildDocForCheckpoint(ctx, be, docInfo, info.ServerSeq, info.ClientSeq, actorID)
+		if err != nil {
+			return nil, err
+		}
+
+		if err := doc.Update(func(root *json.Object, p *presence.Presence) error {
+			p.Clear()
+			return nil
+		}); err != nil {
+			return nil, err
+		}
 
-	// TODO(raararaara): We're currently updating SyncedSeq one by one. This approach is similar
-	// to n+1 query problem. We need to investigate if we can optimize this process by using a single query in the future.
-	for docID, clientDocInfo := range clientInfo.Documents {
-		if err := db.UpdateSyncedSeq(
-			ctx,
-			clientInfo,
-			types.DocRefKey{
-				ProjectID: refKey.ProjectID,
-				DocID:     docID,
-			},
-			clientDocInfo.ServerSeq,
-		); err != nil {
+		// TODO(hackerwins): This is a temporary solution to detach the document
+		// from the client. Documents are shared between multiple servers in the
+		// cluster to simplify the implementation including the distributed lock.
+		// In the future, we need to request the detachments to the load balancer
+		// and the load balancer will forward the request to the server that has
+		// the document.
+		if _, err = packs.PushPull(ctx, be, project, clientInfo, docInfo, doc.CreateChangePack(), packs.PushPullOptions{
+			Mode:   types.SyncModePushOnly,
+			Status: document.StatusDetached,
+		}); err != nil {
 			return nil, err
 		}
 	}
 
-	return clientInfo, err
+	// 02. Deactivate the client.
+	clientInfo, err = be.DB.DeactivateClient(ctx, refKey)
+	if err != nil {
+		return nil, err
+	}
+
+	return clientInfo, nil
 }
 
 // FindActiveClientInfo find the active client info by the given ref key.

server/clients/housekeeping.go

@@ -69,7 +69,7 @@ func DeactivateInactives(
 
 	deactivatedCount := 0
 	for _, clientInfo := range candidates {
-		if _, err := Deactivate(ctx, be.DB, clientInfo.RefKey()); err != nil {
+		if _, err := Deactivate(ctx, be, clientInfo.RefKey()); err != nil {
 			return database.DefaultProjectID, err
 		}
 

server/documents/documents.go

@@ -71,7 +71,7 @@ func ListDocumentSummaries(
 		}
 
 		if includeSnapshot {
-			doc, err := packs.BuildDocumentForServerSeq(ctx, be, docInfo, docInfo.ServerSeq)
+			doc, err := packs.BuildInternalDocForServerSeq(ctx, be, docInfo, docInfo.ServerSeq)
 			if err != nil {
 				return nil, err
 			}
@@ -102,7 +102,7 @@ func GetDocumentSummary(
 		return nil, err
 	}
 
-	doc, err := packs.BuildDocumentForServerSeq(ctx, be, docInfo, docInfo.ServerSeq)
+	doc, err := packs.BuildInternalDocForServerSeq(ctx, be, docInfo, docInfo.ServerSeq)
 	if err != nil {
 		return nil, err
 	}
@@ -135,7 +135,7 @@ func GetDocumentSummaries(
 		snapshot := ""
 		if includeSnapshot {
 			// TODO(hackerwins, kokodak): Resolve the N+1 problem.
-			doc, err := packs.BuildDocumentForServerSeq(ctx, be, docInfo, docInfo.ServerSeq)
+			doc, err := packs.BuildInternalDocForServerSeq(ctx, be, docInfo, docInfo.ServerSeq)
 			if err != nil {
 				return nil, err
 			}
@@ -179,7 +179,7 @@ func GetDocumentByServerSeq(
 		return nil, err
 	}
 
-	doc, err := packs.BuildDocumentForServerSeq(ctx, be, docInfo, serverSeq)
+	doc, err := packs.BuildInternalDocForServerSeq(ctx, be, docInfo, serverSeq)
 	if err != nil {
 		return nil, err
 	}

server/packs/packs.go

@@ -30,6 +30,7 @@ import (
 	"github.com/yorkie-team/yorkie/pkg/document"
 	"github.com/yorkie-team/yorkie/pkg/document/change"
 	"github.com/yorkie-team/yorkie/pkg/document/key"
+	"github.com/yorkie-team/yorkie/pkg/document/time"
 	"github.com/yorkie-team/yorkie/pkg/units"
 	"github.com/yorkie-team/yorkie/server/backend"
 	"github.com/yorkie-team/yorkie/server/backend/database"
@@ -210,8 +211,27 @@ func PushPull(
 	return respPack, nil
 }
 
-// BuildDocumentForServerSeq returns a new document for the given serverSeq.
-func BuildDocumentForServerSeq(
+// BuildDocForCheckpoint returns a new document for the given checkpoint.
+func BuildDocForCheckpoint(
+	ctx context.Context,
+	be *backend.Backend,
+	docInfo *database.DocInfo,
+	serverSeq int64,
+	clientSeq uint32,
+	actorID *time.ActorID,
+) (*document.Document, error) {
+	internalDoc, err := BuildInternalDocForServerSeq(ctx, be, docInfo, serverSeq)
+	if err != nil {
+		return nil, err
+	}
+
+	internalDoc.SetActor(actorID)
+	internalDoc.SyncCheckpoint(serverSeq, clientSeq)
+	return internalDoc.ToDocument(), nil
+}
+
+// BuildInternalDocForServerSeq returns a new document for the given serverSeq.
+func BuildInternalDocForServerSeq(
 	ctx context.Context,
 	be *backend.Backend,
 	docInfo *database.DocInfo,

server/packs/pushpull.go

@@ -139,7 +139,7 @@ func pullSnapshot(
 	initialServerSeq int64,
 ) (*ServerPack, error) {
 	// Build document from DB if the size of changes for the response is greater than the snapshot threshold.
-	doc, err := BuildDocumentForServerSeq(ctx, be, docInfo, initialServerSeq)
+	doc, err := BuildInternalDocForServerSeq(ctx, be, docInfo, initialServerSeq)
 	if err != nil {
 		return nil, err
 	}

server/rpc/yorkie_server.go

@@ -94,7 +94,7 @@ func (s *yorkieServer) DeactivateClient(
 	}
 
 	project := projects.From(ctx)
-	_, err = clients.Deactivate(ctx, s.backend.DB, types.ClientRefKey{
+	_, err = clients.Deactivate(ctx, s.backend, types.ClientRefKey{
 		ProjectID: project.ID,
 		ClientID:  types.IDFromActorID(actorID),
 	})

server/server.go

@@ -150,7 +150,7 @@ func (r *Yorkie) DeactivateClient(ctx context.Context, c1 *client.Client) error
 		return err
 	}
 
-	_, err = clients.Deactivate(ctx, r.backend.DB, types.ClientRefKey{
+	_, err = clients.Deactivate(ctx, r.backend, types.ClientRefKey{
 		ProjectID: project.ID,
 		ClientID:  types.IDFromActorID(c1.ID()),
 	})

test/integration/presence_test.go

@@ -466,4 +466,27 @@ func TestPresence(t *testing.T) {
 
 		assert.Equal(t, expected, responsePairs)
 	})
+
+	t.Run("presence clear by server test", func(t *testing.T) {
+		ctx := context.Background()
+
+		d1 := document.New(helper.TestDocKey(t))
+		d2 := document.New(helper.TestDocKey(t))
+		assert.NoError(t, c1.Attach(ctx, d1))
+		assert.NoError(t, c2.Attach(ctx, d2))
+		defer func() {
+			// No need to detach c1
+			assert.NoError(t, c2.Detach(ctx, d2))
+		}()
+
+		assert.NoError(t, c1.Sync(ctx))
+		assert.NoError(t, c2.Sync(ctx))
+		assert.Equal(t, 2, len(d1.AllPresences()))
+		assert.Equal(t, 2, len(d2.AllPresences()))
+
+		assert.NoError(t, defaultServer.DeactivateClient(ctx, c1))
+		assert.NoError(t, c2.Sync(ctx))
+
+		assert.Equal(t, 1, len(d2.AllPresences()))
+	})
 }