Module: Mbeditor::CollaborationDocStore

Defined in:
app/services/mbeditor/collaboration_doc_store.rb

Overview

Thread-safe, in-memory cache of opaque Yjs bytes per active file. It never interprets the bytes (no Ruby CRDT): it only buffers the latest snapshot plus recent deltas so a late-joiner can sync instantly. Modeled on the existing TTL caches (GitInfoService, AvailabilityProbe): Mutex + monotonic clock.

Constant Summary collapse

GRACE_TTL =

Grace window (seconds): a room with no activity for longer than this is evicted by sweep!. Because every op refreshes last_activity, a room that just went empty survives this long, so a quick reopen recovers its buffer.

300
ROOM_CAP =

Hard cap on cached rooms. Exceeding it evicts the least-recently-active room, so process memory stays bounded even if rooms are never swept.

200
SWEEP_INTERVAL =

Idle GC rides on traffic: every write/read attempts a sweep, but the scan runs at most once per this interval so a busy room doesn't pay for it on every op. With no explicit scheduler, memory stays bounded over a long session as long as some activity continues. Kept well under GRACE_TTL so an idle room is reclaimed soon after its grace window elapses.

60
MAX_DELTAS =

Hard cap on buffered deltas per room; a long editing session on one file would otherwise grow it without limit. Over the cap the oldest go, which costs a late joiner an incomplete replay — it then waits for the next snapshot instead of syncing from the buffer.

500

Class Method Summary collapse

Class Method Details

.claim_seed(path, now: monotonic) ⇒ Object

Grant the right to seed a room's shared document from disk, to exactly one client. Without this every client that finds the room empty inserts the whole file at offset 0, and Yjs merges those inserts into two concatenated copies — the file appended to itself. That happens whenever two editor tabs attach to a room the server has no state for: a fresh room, or one that was evicted while both were idle. The claim is per room, so the loser waits for the winner's content instead of inventing its own.



43
44
45
46
47
48
49
50
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 43

def claim_seed(path, now: monotonic)
  MUTEX.synchronize do
    room = touch(path, now)
    next false if room[:seed_claimed] || room[:snapshot] || !room[:deltas].empty?

    room[:seed_claimed] = true
  end
end

.join(path, now: monotonic) ⇒ Object

Subscriber accounting. A room with a live subscriber must never be evicted: eviction empties the server's copy while clients still hold content, and the next client to attach would be granted a seed it must not perform.



55
56
57
58
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 55

def join(path, now: monotonic)
  MUTEX.synchronize { touch(path, now)[:subscribers] += 1 }
  nil
end

.leave(path, now: monotonic) ⇒ Object 



60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 60

def leave(path, now: monotonic)
  MUTEX.synchronize do
    room = rooms[path]
    next unless room

    room[:subscribers] -= 1 if room[:subscribers] > 0
    # The claimer can leave before it ever seeds (a tab closed during the
    # handshake). Releasing the claim on an empty, empty-handed room keeps the
    # next opener from deferring to content that will never arrive.
    room[:seed_claimed] = false if room[:subscribers].zero? && room[:snapshot].nil? && room[:deltas].empty?
  end
  nil
end

.record_update(path, bytes, now: monotonic) ⇒ Object 



74
75
76
77
78
79
80
81
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 74

def record_update(path, bytes, now: monotonic)
  MUTEX.synchronize do
    room = touch(path, now)
    room[:deltas] << bytes
    room[:deltas].shift while room[:deltas].size > MAX_DELTAS
  end
  nil
end

.replace_snapshot(path, bytes, now: monotonic) ⇒ Object 



83
84
85
86
87
88
89
90
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 83

def replace_snapshot(path, bytes, now: monotonic)
  MUTEX.synchronize do
    room = touch(path, now)
    room[:snapshot] = bytes
    room[:deltas] = []
  end
  nil
end

.reset! ⇒ Object 



111
112
113
114
115
116
117
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 111

def reset!
  MUTEX.synchronize do
    @rooms = {}
    @last_sweep = nil
  end
  nil
end

.state_for(path, now: monotonic) ⇒ Object 



92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 92

def state_for(path, now: monotonic)
  MUTEX.synchronize do
    room = rooms[path]
    if room
      room[:last_activity] = now
      result = { snapshot: room[:snapshot], deltas: room[:deltas].dup }
    else
      result = { snapshot: nil, deltas: [] }
    end
    maybe_sweep(now)
    result
  end
end

.sweep!(now: monotonic, grace: GRACE_TTL) ⇒ Object 



106
107
108
109
 
# File 'app/services/mbeditor/collaboration_doc_store.rb', line 106

def sweep!(now: monotonic, grace: GRACE_TTL)
  MUTEX.synchronize { evict_idle(now, grace) }
  nil
end