Skip to content

Commit f0b1df9

Browse files
authored
fix(realtime): share file editor presence across replicas (#9014)
* fix(realtime): share file editor presence across replicas * test(realtime): update shared document socket fixture * fix(realtime): refresh presence when reclaiming a session * fix(realtime): retry presence reads and add replica E2E proof
1 parent abd7bc4 commit f0b1df9

5 files changed

Lines changed: 298 additions & 25 deletions

File tree

‎apps/realtime/src/handlers/file-doc.redis-store.test.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,12 +147,21 @@ function createIo(sockets: FakeSocket[]) {
147147
const channel = (target: string, excluded: string[] = []) => ({
148148
except: (exclude: string) => channel(target, [...excluded, exclude]),
149149
emit: (event: string, payload: unknown) => emitTo(target, excluded, event, payload),
150+
fetchSockets: async () =>
151+
sockets.filter(
152+
(socket) =>
153+
socket.rooms.has(target) &&
154+
!excluded.some((room) => room === socket.id || socket.rooms.has(room))
155+
),
150156
})
151157
const to = vi.fn((target: string) => channel(target))
152158
return {
153159
to,
154160
in: vi.fn(() => ({ socketsLeave: () => {} })),
155161
local: { to },
162+
sockets: { sockets: { get: (id: string) => sockets.find((socket) => socket.id === id) } },
163+
on: vi.fn(),
164+
serverSideEmit: vi.fn(),
156165
} as unknown as IRoomManager['io']
157166
}
158167

@@ -161,6 +170,7 @@ function setup(id: string, sockets: FakeSocket[]) {
161170
const rooms = new Set<string>()
162171
const socket = {
163172
id,
173+
data: {},
164174
userId: 'user-1',
165175
userName: 'Test User',
166176
userImage: 'avatar.png',

‎apps/realtime/src/handlers/file-doc.test.ts‎

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -72,28 +72,40 @@ function createIo(deliver?: (message: SentMessage) => void) {
7272
const channel = (target: string, excluded: string[] = []) => ({
7373
except: (exclude: string) => channel(target, [...excluded, exclude]),
7474
emit: (event: string, payload: unknown) => emit({ target, except: excluded, event, payload }),
75+
fetchSockets: async () => [...testSockets.values()],
7576
})
7677
const to = vi.fn((target: string) => channel(target))
7778
const inFn = vi.fn((socketId: string) => ({
7879
socketsLeave: (room: string) => {
7980
left.push({ socketId, room })
8081
},
8182
}))
82-
// Doc-sync frames fan out via `io.local.to(...)` (cross-task delivery rides the Redis stream, not the
83-
// adapter). With the store disabled in tests, `local` is the whole room — mirror `to` so those emits
84-
// are recorded identically. Awareness/presence still use `io.to(...)`.
85-
return { io: { to, in: inFn, local: { to } } as unknown as IRoomManager['io'], sent, left }
83+
// With the shared store disabled, local and adapter broadcasts reach the same clients.
84+
return {
85+
io: {
86+
to,
87+
in: inFn,
88+
local: { to },
89+
sockets: { sockets: testSockets },
90+
on: vi.fn(),
91+
serverSideEmit: vi.fn(),
92+
} as unknown as IRoomManager['io'],
93+
sent,
94+
left,
95+
}
8696
}
8797

8898
/** Every socket id a test created, so `afterEach` can drop their rooms without a
8999
* hardcoded list drifting out of sync with the tests. */
90100
const createdSocketIds = new Set<string>()
101+
const testSockets = new Map<string, ReturnType<typeof createSocket>['socket']>()
91102

92103
function createSocket(id: string, overrides?: Record<string, unknown>) {
93104
createdSocketIds.add(id)
94105
const handlers: Record<string, Handler> = {}
95106
const socket = {
96107
id,
108+
data: {},
97109
userId: 'user-1',
98110
userName: 'Test User',
99111
// Set so the server's roster resolves the avatar from the socket (never the DB).
@@ -114,6 +126,7 @@ function createSocket(id: string, overrides?: Record<string, unknown>) {
114126
leave: vi.fn(),
115127
...overrides,
116128
}
129+
testSockets.set(id, socket)
117130
return { handlers, socket }
118131
}
119132

@@ -218,6 +231,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
218231
// map is cleared and never bleeds a counter into the next test.
219232
for (const id of createdSocketIds) cleanupFileDocForSocket(id, io, true)
220233
createdSocketIds.clear()
234+
testSockets.clear()
221235
await getFileDocStore().shutdown()
222236
})
223237

‎apps/realtime/src/handlers/file-doc.ts‎

Lines changed: 70 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -531,6 +531,59 @@ async function persistRoom(name: string, room: FileDocRoom, final: boolean): Pro
531531
}
532532
}
533533

534+
const FILE_DOC_PRESENCE_CHANGED = 'file-doc-presence-changed'
535+
const FILE_DOC_PRESENCE_RETRIES = 2
536+
const FILE_DOC_PRESENCE_RETRY_MS = 100
537+
const presenceRefreshes = new WeakMap<Server, Map<string, { revision: number }>>()
538+
539+
function setupFileDocPresence(io: Server) {
540+
if (presenceRefreshes.has(io)) return
541+
presenceRefreshes.set(io, new Map())
542+
io.on(FILE_DOC_PRESENCE_CHANGED, (name: string, fileId: string) => {
543+
if (fileDocRooms.has(name)) void refreshFileDocPresence(io, name, fileId)
544+
})
545+
}
546+
547+
async function refreshFileDocPresence(io: Server, name: string, fileId: string) {
548+
const refreshes = presenceRefreshes.get(io)
549+
if (!refreshes) return
550+
const pending = refreshes.get(name)
551+
if (pending) {
552+
pending.revision += 1
553+
return
554+
}
555+
const refresh = { revision: 0 }
556+
refreshes.set(name, refresh)
557+
try {
558+
let failures = 0
559+
while (true) {
560+
const revision = refresh.revision
561+
let sockets: Awaited<ReturnType<Server['fetchSockets']>>
562+
try {
563+
sockets = await io.to(name).except(`${name}:pending`).fetchSockets()
564+
} catch (error) {
565+
if (failures >= FILE_DOC_PRESENCE_RETRIES) throw error
566+
failures += 1
567+
await sleep(FILE_DOC_PRESENCE_RETRY_MS)
568+
continue
569+
}
570+
if (revision !== refresh.revision) continue
571+
const users: FileDocPresenceUser[] = []
572+
for (const socket of sockets) {
573+
const { fileDocPresence } = socket.data as AuthenticatedSocket['data']
574+
if (fileDocPresence?.roomName === name) users.push(fileDocPresence.user)
575+
}
576+
// Each replica refreshes its own clients, so a delayed remote snapshot cannot overwrite theirs.
577+
io.local.to(name).except(`${name}:pending`).emit(FILE_DOC_EVENTS.PRESENCE, { fileId, users })
578+
return
579+
}
580+
} catch (error) {
581+
logger.warn('Skipping file document presence refresh', { error: getErrorMessage(error) })
582+
} finally {
583+
refreshes.delete(name)
584+
}
585+
}
586+
534587
/**
535588
* Broadcast the room's collaborator roster to everyone in it, for the avatar stack. One entry
536589
* PER SESSION (socket) — the client excludes its own socket and dedupes the remainder per user
@@ -541,23 +594,8 @@ async function persistRoom(name: string, room: FileDocRoom, final: boolean): Pro
541594
* — so a peer cannot spoof or suppress an entry.
542595
*/
543596
function broadcastFileDocPresence(io: Server, name: string, room: FileDocRoom) {
544-
const users: FileDocPresenceUser[] = []
545-
// One entry PER SOCKET (session), not per clientID: a socket's several providers are the same
546-
// authenticated user, so any of its owners carries the identity; the client dedupes per user for the
547-
// avatar stack (see the roster comment above). An empty inner map is never stored, so `owner` exists.
548-
for (const [socketId, clientMap] of room.owners) {
549-
const owner = clientMap.values().next().value
550-
if (!owner) continue
551-
users.push({
552-
socketId,
553-
userId: owner.userId,
554-
userName: owner.userName,
555-
avatarUrl: owner.avatarUrl,
556-
})
557-
}
558-
io.to(name)
559-
.except(`${name}:pending`)
560-
.emit(FILE_DOC_EVENTS.PRESENCE, { fileId: room.fileId, users })
597+
if (getFileDocStore().enabled) io.serverSideEmit(FILE_DOC_PRESENCE_CHANGED, name, room.fileId)
598+
void refreshFileDocPresence(io, name, room.fileId)
561599
}
562600

563601
/** Whether the client has recorded that it seeded the document's initial content. */
@@ -1488,6 +1526,8 @@ export function cleanupFileDocForSocket(socketId: string, io: Server, endOfLife
14881526
// caret via its awareness removal.
14891527
const clientMap = room.owners.get(socketId)
14901528
room.owners.delete(socketId)
1529+
const socket = io.sockets.sockets.get(socketId) as AuthenticatedSocket | undefined
1530+
if (socket?.data.fileDocPresence?.roomName === name) socket.data.fileDocPresence = undefined
14911531
if (clientMap !== undefined && clientMap.size > 0) {
14921532
// Fires the awareness `update` handler with a non-socket origin → the removals
14931533
// are broadcast to every remaining client, so the departed carets vanish.
@@ -1505,16 +1545,15 @@ export function cleanupFileDocForSocket(socketId: string, io: Server, endOfLife
15051545
* workspace-files join shape (auth → readiness → validate → authorize → join),
15061546
* then runs the Yjs sync/awareness handshake.
15071547
*
1508-
* The avatar roster is derived from this room's own `owners` map and broadcast as
1509-
* `FILE_DOC_EVENTS.PRESENCE` — NOT the Redis-backed room-manager presence the workflow /
1510-
* table rooms use — because the file-doc room already owns an authoritative in-memory Y.Doc
1511-
* pinned to a single replica, so the session identity is right here with no extra store.
1548+
* The avatar roster is collected across the Socket.IO adapter from admitted sockets' authenticated
1549+
* identities. Ownership of document providers remains local to each socket's replica.
15121550
*/
15131551
export function setupWorkspaceFileDocHandlers(
15141552
socket: AuthenticatedSocket,
15151553
roomManager: IRoomManager
15161554
) {
15171555
const io = roomManager.io
1556+
setupFileDocPresence(io)
15181557
// Owner-qualified intent fences delayed admissions and leaves across document switches.
15191558
let currentFileRoom: string | null = null
15201559
/** Co-mounted providers share invalidation membership until their last admission settles. */
@@ -1815,7 +1854,12 @@ export function setupWorkspaceFileDocHandlers(
18151854
if (clientMap.size === 0) {
18161855
entry.owners.delete(otherSid)
18171856
socketToRoomName.delete(otherSid)
1857+
const previousSocket = io.sockets.sockets.get(otherSid) as
1858+
| AuthenticatedSocket
1859+
| undefined
1860+
if (previousSocket) previousSocket.data.fileDocPresence = undefined
18181861
io.in(otherSid).socketsLeave(name)
1862+
broadcastFileDocPresence(io, name, entry)
18191863
}
18201864
}
18211865

@@ -1852,6 +1896,11 @@ export function setupWorkspaceFileDocHandlers(
18521896
return
18531897
}
18541898

1899+
socket.data.fileDocPresence = {
1900+
roomName: name,
1901+
user: { socketId: socket.id, userId, userName, avatarUrl },
1902+
}
1903+
18551904
// Attribution for the server-side persist, refreshed to the actual editor on each edit in
18561905
// `handleMessage`.
18571906
if (!projectId) entry.lastEditorUserId = userId

‎apps/realtime/src/middleware/auth.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { ANONYMOUS_USER, ANONYMOUS_USER_ID } from '@sim/auth/principal'
22
import { createLogger } from '@sim/logger'
3+
import type { FileDocPresenceUser } from '@sim/realtime-protocol/file-doc'
34
import { toError } from '@sim/utils/errors'
45
import type { Socket } from 'socket.io'
56
import { auth } from '@/auth'
@@ -11,6 +12,7 @@ const logger = createLogger('SocketAuth')
1112
* Authenticated socket with user data attached.
1213
*/
1314
export interface AuthenticatedSocket extends Socket {
15+
data: { fileDocPresence?: { roomName: string; user: FileDocPresenceUser } }
1416
userId?: string
1517
userName?: string
1618
userEmail?: string

0 commit comments

Comments
 (0)