diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index 2f180e7fa53..8b0d646ee73 100644 --- a/apps/realtime/src/handlers/connection.ts +++ b/apps/realtime/src/handlers/connection.ts @@ -21,6 +21,13 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager // evicted or TTL-expired (which would leave the manager's stored rooms empty). socket.on('disconnecting', async (reason) => { try { + // Snapshot the live Socket.IO room membership SYNCHRONOUSLY, before any + // await: Socket.IO clears `socket.rooms` via leaveAll() as soon as the + // synchronous portion of this `disconnecting` handler returns (i.e. at the + // first await below), so reading it afterwards would see an empty set and + // the eviction fallback would be dead. + const liveRoomNames = [...socket.rooms] + // Clean up pending debounce entries for this socket to prevent memory leaks cleanupPendingSubblocksForSocket(socket.id) cleanupPendingVariablesForSocket(socket.id) @@ -29,13 +36,13 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager // room the manager knows about. const removedRooms = await roomManager.removeSocketFromAllRooms(socket.id) - // Union with the live Socket.IO membership (authoritative here, and it + // Union with the snapshotted Socket.IO membership (authoritative, 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() for (const room of removedRooms) wasInRooms.set(roomName(room), room) - for (const name of socket.rooms) { + for (const name of liveRoomNames) { // `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. diff --git a/apps/realtime/src/handlers/workflow.test.ts b/apps/realtime/src/handlers/workflow.test.ts index 89ae5521e2f..417960c6e5d 100644 --- a/apps/realtime/src/handlers/workflow.test.ts +++ b/apps/realtime/src/handlers/workflow.test.ts @@ -61,6 +61,7 @@ function createRoomManager(overrides?: Partial): IRoomManager { broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined), getRoomUsers: vi.fn().mockResolvedValue([]), hasRoom: vi.fn().mockResolvedValue(false), + deleteRoom: vi.fn().mockResolvedValue(undefined), addUserToRoom: vi.fn().mockResolvedValue(undefined), getUserSession: vi.fn().mockResolvedValue(null), updateUserActivity: vi.fn().mockResolvedValue(undefined), diff --git a/apps/realtime/src/handlers/workspace-files.test.ts b/apps/realtime/src/handlers/workspace-files.test.ts index 9ecc943c8bb..76f20d9a7e5 100644 --- a/apps/realtime/src/handlers/workspace-files.test.ts +++ b/apps/realtime/src/handlers/workspace-files.test.ts @@ -55,6 +55,7 @@ function createRoomManager(overrides?: Partial): IRoomManager { broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined), getRoomUsers: vi.fn().mockResolvedValue([]), hasRoom: vi.fn().mockResolvedValue(false), + deleteRoom: vi.fn().mockResolvedValue(undefined), addUserToRoom: vi.fn().mockResolvedValue(undefined), getUserSession: vi.fn().mockResolvedValue(null), updateUserActivity: vi.fn().mockResolvedValue(undefined), diff --git a/apps/realtime/src/handlers/workspace-files.ts b/apps/realtime/src/handlers/workspace-files.ts index c88d403e44e..964857e0633 100644 --- a/apps/realtime/src/handlers/workspace-files.ts +++ b/apps/realtime/src/handlers/workspace-files.ts @@ -5,6 +5,7 @@ import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms import { eq } from 'drizzle-orm' import type { AuthenticatedSocket } from '@/middleware/auth' import type { IRoomManager, UserPresence } from '@/rooms' +import { filterVisiblePresence, sweepStalePresence } from '@/rooms/presence-visibility' const logger = createLogger('WorkspaceFilesHandlers') @@ -77,6 +78,19 @@ export function setupWorkspaceFilesHandlers( return } + // Validate the client-supplied id before it reaches the DB query (matches + // the /api/workspace-files-changed guard; join payloads are otherwise raw + // client input). + if (typeof workspaceId !== 'string' || workspaceId.length === 0) { + socket.emit('join-workspace-files-error', { + workspaceId: typeof workspaceId === 'string' ? workspaceId : '', + error: 'Invalid workspace id', + code: 'INVALID_PAYLOAD', + retryable: false, + }) + return + } + const room = filesRoom(workspaceId) let authorized: Awaited> @@ -130,6 +144,10 @@ export function setupWorkspaceFilesHandlers( } } + // Reclaim any presence orphaned by an ungraceful disconnect (pod crash + // fires no `disconnecting` event; the room hashes have no TTL). + await sweepStalePresence(roomManager, room) + socket.join(roomName(room)) const presence: UserPresence = { @@ -147,7 +165,13 @@ export function setupWorkspaceFilesHandlers( await roomManager.addUserToRoom(room, socket.id, presence) - const presenceUsers = await roomManager.getRoomUsers(room) + // Filter the join ack to live members so a new joiner never briefly sees a + // ghost from an entry the sweep hasn't reclaimed yet. + const presenceUsers = await filterVisiblePresence( + roomManager.io, + room, + await roomManager.getRoomUsers(room) + ) socket.emit('join-workspace-files-success', { workspaceId, socketId: socket.id, @@ -159,6 +183,14 @@ export function setupWorkspaceFilesHandlers( logger.info(`User ${userId} (${userName}) joined files room for workspace ${workspaceId}`) } catch (error) { logger.error('Error joining workspace files room:', error) + // Roll back any partial join so a failed attempt can't leave the socket in + // the Socket.IO room or a stale presence entry behind (mirrors the workflow + // join's rollback), before signalling a retryable failure. + try { + const room = filesRoom(workspaceId) + socket.leave(roomName(room)) + await roomManager.removeUserFromRoom(room, socket.id) + } catch {} socket.emit('join-workspace-files-error', { workspaceId, error: 'Failed to join workspace files', @@ -169,14 +201,18 @@ export function setupWorkspaceFilesHandlers( } ) - socket.on('leave-workspace-files', async () => { + socket.on('leave-workspace-files', async (payload?: { workspaceId?: string }) => { try { if (!roomManager.isReady()) return const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKSPACE_FILES) if (!room) return + // Scope the leave to a specific workspace when the client provides one: a + // deferred leave from a prior page must not evict the socket from a room it + // has since switched into (workspace A→B leaves A's leave targeting B). + if (payload?.workspaceId && payload.workspaceId !== room.id) return socket.leave(roomName(room)) await roomManager.removeUserFromRoom(room, socket.id) - await roomManager.broadcastPresenceUpdate(room) + await roomManager.broadcastPresenceUpdate(room, socket.id) } catch (error) { logger.error('Error leaving workspace files room:', error) } diff --git a/apps/realtime/src/rooms/memory-manager.test.ts b/apps/realtime/src/rooms/memory-manager.test.ts index d5d2512a2d8..af98414f540 100644 --- a/apps/realtime/src/rooms/memory-manager.test.ts +++ b/apps/realtime/src/rooms/memory-manager.test.ts @@ -8,6 +8,7 @@ import { ROOM_TYPES, type RoomRef } from '@sim/realtime-protocol/rooms' import { beforeEach, describe, expect, it, vi } from 'vitest' import { MemoryRoomManager } from '@/rooms/memory-manager' +import { sweepStalePresence } from '@/rooms/presence-visibility' import type { UserPresence } from '@/rooms/types' function fakeIo(liveSocketIds: string[] = []) { @@ -120,6 +121,39 @@ describe('MemoryRoomManager multi-room', () => { expect(await manager.getUserSession('socket-2')).not.toBeNull() }) + it('sweepStalePresence reclaims not-live stale entries but keeps live and fresh ones', async () => { + const { io } = fakeIo(['socket-live']) + const m = new MemoryRoomManager(io) + await m.initialize() + + const staleMs = 76 * 60 * 1000 + await m.addUserToRoom(FILES, 'socket-live', presence(FILES, 'socket-live', 'u1')) + await m.addUserToRoom(FILES, 'socket-dead', { + ...presence(FILES, 'socket-dead', 'u2'), + joinedAt: Date.now() - staleMs, + lastActivity: Date.now() - staleMs, + }) + await m.addUserToRoom(FILES, 'socket-recent', presence(FILES, 'socket-recent', 'u3')) + + await sweepStalePresence(m, FILES) + + const remaining = (await m.getRoomUsers(FILES)).map((u) => u.socketId).sort() + // socket-dead: not live + stale → removed. socket-live: live → kept. + // socket-recent: not live but fresh (transient) → kept. + expect(remaining).toEqual(['socket-live', 'socket-recent']) + }) + + it('deleteRoom unconditionally drops all room state', async () => { + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-2', presence(FILES, 'socket-2', 'user-2')) + expect(await manager.hasRoom(FILES)).toBe(true) + + await manager.deleteRoom(FILES) + + expect(await manager.hasRoom(FILES)).toBe(false) + expect(await manager.getRoomUsers(FILES)).toHaveLength(0) + }) + it('ignores removal of a room the socket is not in (id-guarded)', async () => { await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) diff --git a/apps/realtime/src/rooms/memory-manager.ts b/apps/realtime/src/rooms/memory-manager.ts index 0e25921a47e..90001f53b53 100644 --- a/apps/realtime/src/rooms/memory-manager.ts +++ b/apps/realtime/src/rooms/memory-manager.ts @@ -149,6 +149,10 @@ export class MemoryRoomManager implements IRoomManager { return this.rooms.has(roomKey(room)) } + async deleteRoom(room: RoomRef): Promise { + this.rooms.delete(roomKey(room)) + } + async updateUserActivity( room: RoomRef, socketId: string, diff --git a/apps/realtime/src/rooms/presence-visibility.ts b/apps/realtime/src/rooms/presence-visibility.ts index 4680a581964..7e43b9fde29 100644 --- a/apps/realtime/src/rooms/presence-visibility.ts +++ b/apps/realtime/src/rooms/presence-visibility.ts @@ -1,5 +1,14 @@ import { type RoomRef, roomName } from '@sim/realtime-protocol/rooms' import type { Server } from 'socket.io' +import type { IRoomManager } from '@/rooms/types' + +/** + * How stale a not-live presence entry must be before a join-time sweep reclaims + * it. Kept above the 1h socket-key TTL so a normally-idle collaborator is never + * evicted; only genuinely-orphaned entries (e.g. a crashed pod that never fired + * `disconnecting`) are cleared. Matches the workflow join sweep. + */ +const STALE_PRESENCE_THRESHOLD_MS = 75 * 60 * 1000 /** * Filters a room's stored presence down to what should actually be broadcast: @@ -34,3 +43,32 @@ export async function filterVisiblePresence( return candidates } } + +/** + * Reclaims orphaned presence entries in a room: any stored socket that is no + * longer a live Socket.IO member AND has been idle past + * {@link STALE_PRESENCE_THRESHOLD_MS} is removed. This is how a room-users hash + * (which has no TTL) is bounded against ungraceful disconnects — a pod crash + * fires no `disconnecting` event, so its entries would otherwise persist forever. + * Run on join, like the workflow room does. No-op when the liveness lookup fails + * (so a transient adapter blip can't evict live collaborators). + */ +export async function sweepStalePresence(manager: IRoomManager, room: RoomRef): Promise { + let liveIds: Set + try { + const liveSockets = await manager.io.in(roomName(room)).fetchSockets() + liveIds = new Set(liveSockets.map((socket) => socket.id)) + } catch { + return + } + + const now = Date.now() + const users = await manager.getRoomUsers(room) + for (const user of users) { + if (liveIds.has(user.socketId)) continue + const lastSeen = user.lastActivity || user.joinedAt || 0 + if (now - lastSeen > STALE_PRESENCE_THRESHOLD_MS) { + await manager.removeUserFromRoom(room, user.socketId) + } + } +} diff --git a/apps/realtime/src/rooms/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index 2de65bc14c5..945d03d539c 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -311,6 +311,17 @@ export class RedisRoomManager implements IRoomManager { return exists > 0 } + async deleteRoom(room: RoomRef): Promise { + // Log AND rethrow (like addUserToRoom): a failed wipe must not be reported as a + // clean deletion by the caller — the request surfaces it (and can be retried). + try { + await this.redis.del([KEYS.roomUsers(room), KEYS.roomMeta(room)]) + } catch (error) { + logger.error(`Failed to delete room ${room.type}:${room.id}:`, error) + throw error + } + } + async updateUserActivity( room: RoomRef, socketId: string, diff --git a/apps/realtime/src/rooms/types.ts b/apps/realtime/src/rooms/types.ts index 80525195ef8..c9c8ee18704 100644 --- a/apps/realtime/src/rooms/types.ts +++ b/apps/realtime/src/rooms/types.ts @@ -101,6 +101,13 @@ export interface IRoomManager { /** Whether a room currently has any presence. */ hasRoom(room: RoomRef): Promise + /** + * Unconditionally drop all state for a room (presence + metadata). Used when a + * room's underlying resource is destroyed (e.g. a deleted workflow) to guarantee + * no state lingers even if per-socket removals failed or a socket joined mid-teardown. + */ + deleteRoom(room: RoomRef): Promise + /** Update a socket's activity (cursor, selection, lastActivity) within a room. */ updateUserActivity( room: RoomRef, diff --git a/apps/realtime/src/rooms/workflow-room-service.ts b/apps/realtime/src/rooms/workflow-room-service.ts index 4623647a7da..f12a6c337d4 100644 --- a/apps/realtime/src/rooms/workflow-room-service.ts +++ b/apps/realtime/src/rooms/workflow-room-service.ts @@ -22,29 +22,44 @@ export class WorkflowRoomService { logger.info(`Handling workflow deletion notification for ${workflowId}`) const room = workflowRoom(workflowId) - const users = await this.manager.getRoomUsers(room) - if (users.length === 0) { - logger.debug(`No active users found for deleted workflow ${workflowId}`) - return - } + const name = roomName(room) + // Always notify — reach every socket still in the Socket.IO room so the client + // clears the deleted workflow, even if that socket's Redis presence was evicted + // (in which case it would be missing from getRoomUsers). Emitting to an empty + // room is a harmless no-op. this.manager.emitToRoom(room, 'workflow-deleted', { workflowId, message: 'This workflow has been deleted', timestamp: Date.now(), }) + // Clean per-socket state for every socket that is either a live Socket.IO member + // OR still has presence — so an evicted/late-joined socket's room mapping and + // session are dropped too, not just the presence-tracked ones. + const socketIds = new Set() + try { + const liveSockets = await this.manager.io.in(name).fetchSockets() + for (const s of liveSockets) socketIds.add(s.id) + } catch (error) { + logger.warn(`Could not enumerate sockets for deleted workflow ${workflowId}`, error) + } + for (const user of await this.manager.getRoomUsers(room)) socketIds.add(user.socketId) + // Remove every socket from the Socket.IO room (cross-pod via the Redis adapter). - const name = roomName(room) await this.manager.io.in(name).socketsLeave(name) - // Drop presence state for each socket; empty-room cleanup is handled by the manager. - for (const user of users) { - await this.manager.removeUserFromRoom(room, user.socketId) + for (const socketId of socketIds) { + await this.manager.removeUserFromRoom(room, socketId) } + // Final unconditional wipe — the workflow is gone, so no room state may linger + // even if a per-socket removal failed (matches the pre-refactor managers, which + // ended deletion with an unconditional room drop). + await this.manager.deleteRoom(room) + logger.info( - `Cleaned up workflow room ${workflowId} after deletion (${users.length} users disconnected)` + `Cleaned up workflow room ${workflowId} after deletion (${socketIds.size} sockets removed)` ) } diff --git a/apps/sim/app/workspace/[workspaceId]/files/hooks/use-workspace-files-room.ts b/apps/sim/app/workspace/[workspaceId]/files/hooks/use-workspace-files-room.ts index e2c4a61befc..795d233f4e0 100644 --- a/apps/sim/app/workspace/[workspaceId]/files/hooks/use-workspace-files-room.ts +++ b/apps/sim/app/workspace/[workspaceId]/files/hooks/use-workspace-files-room.ts @@ -77,6 +77,12 @@ export function useWorkspaceFilesRoom( }) => { if (data.workspaceId !== workspaceId) return retries = 0 + // Cancel any retry scheduled by a prior retryable error so it can't fire an + // extra join after we're already in. + if (retryTimer) { + clearTimeout(retryTimer) + retryTimer = null + } setPresenceUsers(data.presenceUsers ?? []) } const handleJoinError = (data: JoinErrorPayload) => { @@ -103,13 +109,18 @@ export function useWorkspaceFilesRoom( return () => { if (retryTimer) clearTimeout(retryTimer) - socket.emit('leave-workspace-files') socket.off('connect', join) socket.off('join-workspace-files-success', handleJoinSuccess) socket.off('join-workspace-files-error', handleJoinError) socket.off('workspace-files:presence-update', handlePresence) socket.off('workspace-files-changed', handleChanged) setPresenceUsers([]) + + // Leave the room, scoped to THIS workspace: the server no-ops if the socket + // has already switched to another workspace's files room (so a workspace + // A→B switch, where B's join runs first and auto-leaves A, can't have A's + // leave evict the fresh B membership). + socket.emit('leave-workspace-files', { workspaceId }) } }, [socket, workspaceId, queryClient]) diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index e738130b6de..71ef65c1ced 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -11,9 +11,14 @@ const NOTIFY_TIMEOUT_MS = 2000 /** * Best-effort fan-out to the realtime server that a workspace's file tree changed, * so every browser currently viewing that workspace's files refetches. File - * mutations happen over the HTTP API (not the socket); this is the lossy liveness - * signal — a dropped notification only degrades to stale-until-refetch, so it must - * never throw or block the originating mutation. + * mutations happen over the HTTP API (not the socket); this is a lossy liveness + * signal — a dropped notification only degrades to stale-until-refetch. + * + * Never throws. Callers `await` it (rather than fire-and-forget) so the fetch is + * guaranteed to dispatch before a Node route handler returns — a floating promise + * can be dropped after the response is sent. It is a normally-sub-millisecond + * local call and is hard-bounded to {@link NOTIFY_TIMEOUT_MS}, so it adds that + * latency only when the socket pod is unreachable. */ export async function notifyWorkspaceFilesChanged(workspaceId: string): Promise { try { diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts index 6967a7d313f..662885c2ec5 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts @@ -23,6 +23,7 @@ import { resolveWorkflowAliasForWorkspace } from '@/lib/copilot/vfs/workflow-ali import { isReservedWorkflowAliasBackingDisplayPath } from '@/lib/copilot/vfs/workflow-aliases' import { generateRestoreName } from '@/lib/core/utils/restore-name' import type { DbOrTx } from '@/lib/db/types' +import { notifyWorkspaceFilesChanged } from '@/lib/realtime/notify' import { getServePathPrefix } from '@/lib/uploads' import { deleteFile, @@ -380,6 +381,11 @@ export async function uploadWorkspaceFile( const pathPrefix = getServePathPrefix() const serveUrl = `${pathPrefix}${encodeURIComponent(uploadResult.key)}?context=workspace` + // Fan out the live-tree signal for the direct-upload paths (multipart + // fallback, copilot create, /api/files/upload, v1 files) — the presigned + // path already notifies from its register route. + await notifyWorkspaceFilesChanged(workspaceId) + return { id: fileId, name: uniqueName,