Introduce presence memory cache (#1515)

Presence is now stored in-memory(volatile) and removed when the client
is detached. This design avoids modifying ServerSeq while still enabling
efficient presence handling.

Changes:
- Store presences in presenceCache, operations in operationCache + DB
- Merge presences and operations when serving queries
- Clear presences when client detaches during checkpoint update

Author: hackerwins

pkg/document/innerpresence/change.go

@@ -46,3 +46,8 @@ func (c *Change) Execute(actorID time.ActorID, presences *Map) {
 		presences.Store(actorID.String(), c.Presence)
 	}
 }
+
+// IsClear returns true if the change is of type Clear.
+func (c *Change) IsClear() bool {
+	return c.ChangeType == Clear
+}

server/backend/database/change_info.go

@@ -139,6 +139,11 @@ func (i *ChangeInfo) DeepCopy() *ChangeInfo {
 	return clone
 }
 
+// PresenceOnly returns true if this change is a presence-only change.
+func (i *ChangeInfo) PresenceOnly() bool {
+	return len(i.Operations) == 0 && i.PresenceChange != nil
+}
+
 // EncodePresenceChange encodes the given PresenceChange into bytes array.
 func EncodePresenceChange(p *innerpresence.Change) ([]byte, error) {
 	if p == nil {

server/backend/database/mongo/changestore.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/google/btree"
 
+	"github.com/yorkie-team/yorkie/api/types"
 	"github.com/yorkie-team/yorkie/pkg/errors"
 	"github.com/yorkie-team/yorkie/server/backend/database"
 )
@@ -38,10 +39,12 @@ type ChangeRange struct {
 	To   int64 // Ending server sequence
 }
 
-// ChangeStore manages individual document changes using B-tree
+// ChangeStore manages individual document changes using B-tree. It allows
+// efficient storage and retrieval of changes based on their server sequence numbers.
 type ChangeStore struct {
-	mu   sync.RWMutex
-	tree *btree.BTreeG[*database.ChangeInfo]
+	mu     sync.RWMutex
+	ranges []ChangeRange
+	tree   *btree.BTreeG[*database.ChangeInfo]
 }
 
 // NewChangeStore creates a new instance of ChangeStore.
@@ -83,6 +86,11 @@ func (s *ChangeStore) EnsureChanges(
 		for _, change := range changes {
 			s.tree.ReplaceOrInsert(change)
 		}
+
+		// 4. Mark this requested range as fetched so we don't re-request
+		// sequences inside it again even if some sequence numbers weren't
+		// present in the returned changes (i.e. "missing server seq").
+		s.ranges = mergeAdjacentRanges(append(s.ranges, r))
 	}
 
 	return nil
@@ -125,10 +133,45 @@ func (s *ChangeStore) ChangesInRange(from, to int64) []*database.ChangeInfo {
 	return result
 }
 
+// RemoveChangesByActor removes presence changes for the given actor,
+// keeping Clear changes so the final state remains cleared.
+func (s *ChangeStore) RemoveChangesByActor(actorID types.ID) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var toDelete []*database.ChangeInfo
+	s.tree.Ascend(func(item *database.ChangeInfo) bool {
+		if item != nil && item.ActorID == actorID && !item.PresenceChange.IsClear() {
+			toDelete = append(toDelete, item)
+		}
+		return true
+	})
+
+	for _, item := range toDelete {
+		s.tree.Delete(item)
+	}
+}
+
+// ReplaceOrInsert replaces or inserts the given changes into the store.
+func (s *ChangeStore) ReplaceOrInsert(changes []*database.ChangeInfo) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for _, pch := range changes {
+		s.tree.ReplaceOrInsert(pch)
+	}
+}
+
 // calcMissingRanges calculates which ranges need to be fetched
 func (s *ChangeStore) calcMissingRanges(from, to int64) []ChangeRange {
 	if s.tree.Len() == 0 {
-		return []ChangeRange{{From: from, To: to}}
+		// If tree is empty but we've already fetched some ranges, mark those
+		// fetched ranges as present to avoid refetching them. Otherwise,
+		// request full range.
+		if len(s.ranges) == 0 {
+			return []ChangeRange{{From: from, To: to}}
+		}
+		// fallthrough to create seqMap and mark fetched ranges
 	}
 
 	var missingRanges []ChangeRange
@@ -156,6 +199,23 @@ func (s *ChangeStore) calcMissingRanges(from, to int64) []ChangeRange {
 		return true
 	})
 
+	// Also mark sequences within fetchedRanges as found so we don't request
+	// them again. This covers the case where the fetcher returned fewer
+	// changes than requested (holes) and we want to avoid re-fetching the
+	// same range repeatedly.
+	for _, fr := range s.ranges {
+		// ignore fetched ranges that don't overlap our target
+		if fr.To < from || fr.From > to {
+			continue
+		}
+
+		start := max(fr.From, from)
+		end := min(fr.To, to)
+		for seq := start; seq <= end; seq++ {
+			seqMap[seq] = true
+		}
+	}
+
 	// Now find contiguous missing ranges
 	var inRange bool = false
 	var startMissing int64

server/backend/database/mongo/client.go

@@ -51,11 +51,12 @@ type Client struct {
 	config *Config
 	client *mongo.Client
 
-	cacheManager *cache.Manager
-	clientCache  *cache.LRUWithStats[types.ClientRefKey, *database.ClientInfo]
-	docCache     *cache.LRUWithStats[types.DocRefKey, *database.DocInfo]
-	changeCache  *cache.LRUWithStats[types.DocRefKey, *ChangeStore]
-	vectorCache  *cache.LRUWithStats[types.DocRefKey, *cmap.Map[types.ID, time.VersionVector]]
+	cacheManager  *cache.Manager
+	clientCache   *cache.LRUWithStats[types.ClientRefKey, *database.ClientInfo]
+	docCache      *cache.LRUWithStats[types.DocRefKey, *database.DocInfo]
+	changeCache   *cache.LRUWithStats[types.DocRefKey, *ChangeStore]
+	presenceCache *cache.LRUWithStats[types.DocRefKey, *ChangeStore]
+	vectorCache   *cache.LRUWithStats[types.DocRefKey, *cmap.Map[types.ID, time.VersionVector]]
 }
 
 // Dial creates an instance of Client and dials the given MongoDB.
@@ -119,6 +120,12 @@ func Dial(conf *Config) (*Client, error) {
 	}
 	cacheManager.RegisterCache(changeCache)
 
+	presenceCache, err := cache.NewLRUWithStats[types.DocRefKey, *ChangeStore](conf.ChangeCacheSize, "presences")
+	if err != nil {
+		return nil, fmt.Errorf("initialize presence cache: %w", err)
+	}
+	cacheManager.RegisterCache(presenceCache)
+
 	vectorCache, err := cache.NewLRUWithStats[types.DocRefKey, *cmap.Map[types.ID, time.VersionVector]](
 		conf.VectorCacheSize, "vectors",
 	)
@@ -135,10 +142,11 @@ func Dial(conf *Config) (*Client, error) {
 
 		cacheManager: cacheManager,
 
-		clientCache: clientCache,
-		docCache:    docCache,
-		changeCache: changeCache,
-		vectorCache: vectorCache,
+		clientCache:   clientCache,
+		docCache:      docCache,
+		changeCache:   changeCache,
+		presenceCache: presenceCache,
+		vectorCache:   vectorCache,
 	}
 
 	if conf.CacheStatsEnabled {
@@ -159,6 +167,7 @@ func (c *Client) Close() error {
 	c.clientCache.Purge()
 	c.docCache.Purge()
 	c.changeCache.Purge()
+	c.presenceCache.Purge()
 	c.vectorCache.Purge()
 
 	return nil
@@ -1020,6 +1029,13 @@ func (c *Client) UpdateClientInfoAfterPushPull(
 		return err
 	}
 
+	// NOTE(hackerwins): Clear presence changes of the given client on the
+	// document if the client is no longer attached to the document.
+	docKey := types.DocRefKey{ProjectID: info.ProjectID, DocID: docInfo.ID}
+	if pCache, ok := c.presenceCache.Get(docKey); ok && !attached {
+		pCache.RemoveChangesByActor(info.ID)
+	}
+
 	var updater bson.M
 	if attached {
 		updater = bson.M{
@@ -1467,25 +1483,31 @@ func (c *Client) CreateChangeInfos(
 		return docInfo, checkpoint, nil
 	}
 
-	// 02. Optimized batch processing
 	initialServerSeq := docInfo.ServerSeq
 	now := gotime.Now()
 
-	// Pre-allocate models slice to avoid dynamic allocations
-	models := make([]mongo.WriteModel, 0, len(changes))
+	// 02. Store the changes.
+	//    - presence-only changes are stored in presenceCache.
+	//    - regular changes are stored in DB.
 	hasOperations := false
-
+	var prChanges []*database.ChangeInfo
+	var opChanges []mongo.WriteModel
 	for _, cn := range changes {
 		serverSeq := docInfo.IncreaseServerSeq()
 		checkpoint = checkpoint.NextServerSeq(serverSeq)
 		cn.ServerSeq = serverSeq
 		checkpoint = checkpoint.SyncClientSeq(cn.ClientSeq)
 
+		if cn.PresenceOnly() {
+			prChanges = append(prChanges, cn)
+			continue
+		}
+
 		if len(cn.Operations) > 0 {
 			hasOperations = true
 		}
 
-		models = append(models, mongo.NewUpdateOneModel().SetFilter(bson.M{
+		opChanges = append(opChanges, mongo.NewUpdateOneModel().SetFilter(bson.M{
 			"project_id": refKey.ProjectID,
 			"doc_id":     refKey.DocID,
 			"server_seq": cn.ServerSeq,
@@ -1500,10 +1522,22 @@ func (c *Client) CreateChangeInfos(
 		}}).SetUpsert(true))
 	}
 
-	if len(changes) > 0 {
+	if len(prChanges) > 0 {
+		var store *ChangeStore
+		if cached, ok := c.presenceCache.Get(refKey); ok {
+			store = cached
+		} else {
+			store = NewChangeStore()
+			c.presenceCache.Add(refKey, store)
+		}
+
+		store.ReplaceOrInsert(prChanges)
+	}
+
+	if len(opChanges) > 0 {
 		if _, err := c.collection(ColChanges).BulkWrite(
 			ctx,
-			models,
+			opChanges,
 			options.BulkWrite().SetOrdered(false),
 		); err != nil {
 			return nil, change.InitialCheckpoint, fmt.Errorf("create changes of %s: %w", refKey, err)
@@ -1679,17 +1713,25 @@ func (c *Client) FindChangeInfosBetweenServerSeqs(
 		return nil, nil
 	}
 
-	// Get or create a change range store for this document
-	var store *ChangeStore
+	// 01. Create a temporary change store to hold the changes.
+	store := NewChangeStore()
+
+	// 02. Fill the store with presence only changes.
+	if prStore, ok := c.presenceCache.Get(docRefKey); ok {
+		store.ReplaceOrInsert(prStore.ChangesInRange(from, to))
+	}
+
+	// 03. Fill the store with regular changes.
+	var opStore *ChangeStore
 	if cached, ok := c.changeCache.Get(docRefKey); ok {
-		store = cached
+		opStore = cached
 	} else {
-		store = NewChangeStore()
-		c.changeCache.Add(docRefKey, store)
+		opStore = NewChangeStore()
+		c.changeCache.Add(docRefKey, opStore)
 	}
 
 	// Calculate missing ranges and fetch them in a single operation
-	if err := store.EnsureChanges(from, to, func(from, to int64) ([]*database.ChangeInfo, error) {
+	if err := opStore.EnsureChanges(from, to, func(from, to int64) ([]*database.ChangeInfo, error) {
 		// NOTE(hackerwins): Paginate fetching to avoid loading too many ChangeInfos at once.
 		const chunkSize int64 = 1000
 		var infos []*database.ChangeInfo
@@ -1724,6 +1766,8 @@ func (c *Client) FindChangeInfosBetweenServerSeqs(
 		return nil, err
 	}
 
+	store.ReplaceOrInsert(opStore.ChangesInRange(from, to))
+
 	return store.ChangesInRange(from, to), nil
 }
 
@@ -2211,6 +2255,9 @@ func (c *Client) purgeDocumentInternals(
 	counts := make(map[string]int64)
 
 	c.changeCache.Remove(types.DocRefKey{ProjectID: projectID, DocID: docID})
+	c.presenceCache.Remove(types.DocRefKey{ProjectID: projectID, DocID: docID})
+	c.vectorCache.Remove(types.DocRefKey{ProjectID: projectID, DocID: docID})
+
 	res, err := c.collection(ColChanges).DeleteMany(ctx, bson.M{
 		"project_id": projectID,
 		"doc_id":     docID,