Uh oh!
There was an error while loading. Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork 3.8k
refactor(realtime): generalize presence server to multi-room [2/N]#5930
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Uh oh!
There was an error while loading. Please reload this page.
Merged
Changes from all commits
Commits
Show all changes
5 commits
Select commit
Hold shift + click to select a range
ed4eab3
refactor(realtime): generalize presence server to multi-room (RoomRef)
waleedlatif1 1fee9a7
fix(realtime): harden multi-room disconnect + id-guard room removal
waleedlatif1 7f70ab7
fix(realtime): only rebroadcast disconnect-fallback rooms whose remov…
waleedlatif1 7f9610c
fix(realtime): exclude the disconnecting socket from its farewell bro…
waleedlatif1 2129832
fix(realtime): make presence broadcasts liveness-aware (root-cause gh…
waleedlatif1 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,5 @@ | ||
| import { createLogger } from '@sim/logger' | ||
| import { parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' | ||
| import { cleanupPendingSubblocksForSocket } from '@/handlers/subblocks' | ||
| import { cleanupPendingVariablesForSocket } from '@/handlers/variables' | ||
| import type { AuthenticatedSocket } from '@/middleware/auth' | ||
| @@ -15,20 +16,50 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager | ||
| logger.error(`Socket ${socket.id} connection error:`, error) | ||
| }) | ||
| socket.on('disconnect', async (reason) => { | ||
| // `disconnecting` (not `disconnect`): here `socket.rooms` is still populated and | ||
| // authoritative, so presence is cleaned up even if the Redis room-set key was | ||
| // evicted or TTL-expired (which would leave the manager's stored rooms empty). | ||
| socket.on('disconnecting', async (reason) => { | ||
| try { | ||
| // Clean up pending debounce entries for this socket to prevent memory leaks | ||
| cleanupPendingSubblocksForSocket(socket.id) | ||
| cleanupPendingVariablesForSocket(socket.id) | ||
| const workflowIdHint = [...socket.rooms].find((roomId) => roomId !== socket.id) | ||
| const workflowId = await roomManager.removeUserFromRoom(socket.id, workflowIdHint) | ||
| // A socket may occupy multiple rooms (one per type). Remove it from every | ||
| // room the manager knows about. | ||
| const removedRooms = await roomManager.removeSocketFromAllRooms(socket.id) | ||
| if (workflowId) { | ||
| await roomManager.broadcastPresenceUpdate(workflowId) | ||
| logger.info( | ||
| `Socket ${socket.id} disconnected from workflow ${workflowId} (reason: ${reason})` | ||
| ) | ||
| // Union with the live Socket.IO membership (authoritative here, and it | ||
| // survives a Redis eviction/TTL lapse that would leave the manager's tracked | ||
| // rooms empty). Attempt removal for any room the manager didn't already | ||
| // remove — best-effort, since a transient Redis error can't be recovered here. | ||
| const wasInRooms = new Map<string, RoomRef>() | ||
| for (const room of removedRooms) wasInRooms.set(roomName(room), room) | ||
| for (const name of socket.rooms) { | ||
| // `wasInRooms.has(name)` already excludes every room the manager removed | ||
| // (same room-name key via the roomName/parseRoomName bijection), so any | ||
| // room reaching here was NOT in `removedRooms` and needs a removal attempt. | ||
| if (name === socket.id || wasInRooms.has(name)) continue | ||
| const ref = parseRoomName(name) | ||
| if (!ref) continue | ||
| wasInRooms.set(name, ref) | ||
| await roomManager.removeUserFromRoom(ref, socket.id) | ||
| } | ||
| // Broadcast a correction to every room this socket was in, EXCLUDING this | ||
| // socket — so it is never shown as a ghost collaborator even if its presence | ||
| // entry outlived a failed removal (transient Redis error; the hashes have no | ||
| // TTL). Any orphaned entry is additionally reclaimed by the next join's | ||
| // stale-presence sweep. | ||
| for (const room of wasInRooms.values()) { | ||
| await roomManager.broadcastPresenceUpdate(room, socket.id) | ||
| } | ||
greptile-apps[bot] marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| if (wasInRooms.size > 0) { | ||
| const rooms = Array.from(wasInRooms.values()) | ||
| .map((room) => `${room.type}:${room.id}`) | ||
| .join(', ') | ||
| logger.info(`Socket ${socket.id} disconnected from [${rooms}] (reason: ${reason})`) | ||
| } | ||
| } catch (error) { | ||
| logger.error(`Error handling disconnect for socket ${socket.id}:`, error) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.