From 816d0eca7a97b76af0388bb31c3b86b3a6648f05 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 12:46:48 -0700 Subject: [PATCH 1/5] fix(realtime): address post-merge review-comment findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A re-audit of every inline review comment on the merged stack surfaced real issues that the thread-resolutions and prior audits missed. Fixes: Presence server (#5930 comments): - connection.ts: snapshot `socket.rooms` SYNCHRONOUSLY before the first await. Socket.IO clears the room set once the synchronous part of a `disconnecting` handler returns, so reading it after `await removeSocketFromAllRooms` saw an empty set — the eviction fallback was dead. (Cursor: "Disconnect fallback misses live rooms".) - workflow-room-service: restore the original managers' final unconditional room-state wipe via a new `deleteRoom(room)` manager method, so a deleted workflow leaves no lingering presence/meta even if a per-socket removal failed or a socket joined mid-teardown. (Cursor: "Deletion skips final room wipe".) Files (#5932 comments): - workspace-file-manager.uploadWorkspaceFile now fans out the live-tree signal (all direct-upload paths: multipart fallback, copilot create, /api/files/upload, v1 files — the presigned path already notified). (Cursor: "Creates miss live tree fan-out".) - use-workspace-files-room: clear the pending retry timer on join success; and a module-scoped intended-room guard defers the unmount `leave` so a rapid remount re-claims the room and skips a stale leave — fixing presence flap + a leave-after-join race. (Cursor: "Retry timer survives join success" + "Remount churns files presence".) - workspace-files handler: roll back a partial join (leave room + remove presence) in the catch, mirroring the workflow join. (Cursor: "Join failure skips membership rollback".) +2 tests (deleteRoom). 127 realtime tests pass, both apps tsc clean, api-validation + boundaries green. --- apps/realtime/src/handlers/connection.ts | 11 ++++++-- apps/realtime/src/handlers/workflow.test.ts | 1 + .../src/handlers/workspace-files.test.ts | 1 + apps/realtime/src/handlers/workspace-files.ts | 8 ++++++ .../realtime/src/rooms/memory-manager.test.ts | 11 ++++++++ apps/realtime/src/rooms/memory-manager.ts | 4 +++ apps/realtime/src/rooms/redis-manager.ts | 8 ++++++ apps/realtime/src/rooms/types.ts | 7 +++++ .../src/rooms/workflow-room-service.ts | 23 +++++++++------- .../files/hooks/use-workspace-files-room.ts | 27 ++++++++++++++++++- .../workspace/workspace-file-manager.ts | 6 +++++ 11 files changed, 94 insertions(+), 13 deletions(-) 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..50a4b321ef9 100644 --- a/apps/realtime/src/handlers/workspace-files.ts +++ b/apps/realtime/src/handlers/workspace-files.ts @@ -159,6 +159,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', diff --git a/apps/realtime/src/rooms/memory-manager.test.ts b/apps/realtime/src/rooms/memory-manager.test.ts index d5d2512a2d8..2153c347136 100644 --- a/apps/realtime/src/rooms/memory-manager.test.ts +++ b/apps/realtime/src/rooms/memory-manager.test.ts @@ -120,6 +120,17 @@ describe('MemoryRoomManager multi-room', () => { expect(await manager.getUserSession('socket-2')).not.toBeNull() }) + 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/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index 2de65bc14c5..6c9b4fe30de 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -311,6 +311,14 @@ export class RedisRoomManager implements IRoomManager { return exists > 0 } + async deleteRoom(room: RoomRef): Promise { + try { + await this.redis.del([KEYS.roomUsers(room), KEYS.roomMeta(room)]) + } catch (error) { + logger.error(`Failed to delete room ${room.type}:${room.id}:`, 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..e31eb2e5167 100644 --- a/apps/realtime/src/rooms/workflow-room-service.ts +++ b/apps/realtime/src/rooms/workflow-room-service.ts @@ -23,26 +23,29 @@ export class WorkflowRoomService { 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 - } - this.manager.emitToRoom(room, 'workflow-deleted', { - workflowId, - message: 'This workflow has been deleted', - timestamp: Date.now(), - }) + if (users.length > 0) { + this.manager.emitToRoom(room, 'workflow-deleted', { + workflowId, + message: 'This workflow has been deleted', + timestamp: Date.now(), + }) + } // 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. + // Drop presence state for each socket. for (const user of users) { await this.manager.removeUserFromRoom(room, user.socketId) } + // Final unconditional wipe — the workflow is gone, so no state may linger even + // if a per-socket removal failed or a socket joined mid-teardown (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)` ) 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..5079a4bb1a5 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 @@ -14,6 +14,16 @@ const logger = createLogger('WorkspaceFilesRoom') const MAX_JOIN_RETRIES = 3 const JOIN_RETRY_BASE_MS = 1000 +/** + * The workspace whose files room this page currently wants to be in (module-scoped + * because only one Files page mounts at a time). A remount (React strict-mode, a + * fast route re-mount) runs the old cleanup then the new effect synchronously; the + * new effect re-claims this before the deferred leave fires, so the stale leave is + * skipped — preventing presence flapping and a leave-after-join race that could + * drop the fresh membership. + */ +let intendedFilesWorkspaceId: string | null = null + interface PresenceUpdatePayload extends PresenceAvatarUser { folderId?: string | null } @@ -77,6 +87,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) => { @@ -94,6 +110,7 @@ export function useWorkspaceFilesRoom( } // Join now if the socket is already connected; `connect` covers (re)connects. + intendedFilesWorkspaceId = workspaceId if (socket.connected) join() socket.on('connect', join) socket.on('join-workspace-files-success', handleJoinSuccess) @@ -103,13 +120,21 @@ 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([]) + + // Defer the leave: if the page re-mounts for the same workspace this tick, the + // new effect re-claims `intendedFilesWorkspaceId` and this leave is skipped, + // avoiding a flap and a leave-after-join race. A real navigation away leaves + // the value cleared/changed, so the leave fires. + if (intendedFilesWorkspaceId === workspaceId) intendedFilesWorkspaceId = null + queueMicrotask(() => { + if (intendedFilesWorkspaceId !== workspaceId) socket.emit('leave-workspace-files') + }) } }, [socket, workspaceId, queryClient]) 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, From 954ae21bb1621b5d17a8b919fb64980912693306 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 12:58:56 -0700 Subject: [PATCH 2/5] fix(files): scope workspace-files leave to a workspace (deferred-leave safety) Self-review of the deferred-leave guard found a real bug: leave-workspace-files was not workspace-scoped, so after a workspace switch (A->B) the deferred leave from A would evict the socket from its new room B. The leave now carries the workspaceId and the server no-ops if the socket's current files room differs. Also excludes the leaving socket from the leave broadcast (consistent with disconnect). --- apps/realtime/src/handlers/workspace-files.ts | 8 ++++++-- .../[workspaceId]/files/hooks/use-workspace-files-room.ts | 7 +++++-- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/apps/realtime/src/handlers/workspace-files.ts b/apps/realtime/src/handlers/workspace-files.ts index 50a4b321ef9..13f6b2f2898 100644 --- a/apps/realtime/src/handlers/workspace-files.ts +++ b/apps/realtime/src/handlers/workspace-files.ts @@ -177,14 +177,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/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 5079a4bb1a5..92dbc94e5ce 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 @@ -130,10 +130,13 @@ export function useWorkspaceFilesRoom( // Defer the leave: if the page re-mounts for the same workspace this tick, the // new effect re-claims `intendedFilesWorkspaceId` and this leave is skipped, // avoiding a flap and a leave-after-join race. A real navigation away leaves - // the value cleared/changed, so the leave fires. + // the value cleared/changed, so the leave fires. The leave is scoped to this + // workspace so, after a workspace switch, it can't evict the new room. if (intendedFilesWorkspaceId === workspaceId) intendedFilesWorkspaceId = null queueMicrotask(() => { - if (intendedFilesWorkspaceId !== workspaceId) socket.emit('leave-workspace-files') + if (intendedFilesWorkspaceId !== workspaceId) { + socket.emit('leave-workspace-files', { workspaceId }) + } }) } }, [socket, workspaceId, queryClient]) From 895a792623038bc0b21df7ab2cc9c38aea16abbd Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 13:08:21 -0700 Subject: [PATCH 3/5] fix(realtime): close files-room presence leak + validate join payload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Architecture-audit findings: - S1 (real Redis leak): the files room inherited the shared manager but not the workflow join's liveness sweep, so an UNGRACEFUL disconnect (pod crash — no `disconnecting` event) left its presence entry in the no-TTL room hash forever. Added a shared `sweepStalePresence(manager, room)` (fetchSockets liveness + remove not-live-AND-stale entries, matching the workflow 75min threshold) and run it on files join; also filter the join ack through `filterVisiblePresence` so a joiner never briefly sees an un-swept ghost. - S2: validate the client-supplied `workspaceId` on files join before it reaches the DB query (matches the /api/workspace-files-changed guard; fails closed). - N2: corrected the notify doc — it is awaited (guaranteed dispatch before a Node route returns) and hard-bounded to NOTIFY_TIMEOUT_MS, not "never block". +1 test (sweepStalePresence keeps live/fresh, reclaims not-live-stale). 128 realtime tests pass, both apps tsc clean, biome clean. --- apps/realtime/src/handlers/workspace-files.ts | 26 ++++++++++++- .../realtime/src/rooms/memory-manager.test.ts | 23 +++++++++++ .../realtime/src/rooms/presence-visibility.ts | 38 +++++++++++++++++++ apps/sim/lib/realtime/notify.ts | 11 ++++-- 4 files changed, 94 insertions(+), 4 deletions(-) diff --git a/apps/realtime/src/handlers/workspace-files.ts b/apps/realtime/src/handlers/workspace-files.ts index 13f6b2f2898..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, diff --git a/apps/realtime/src/rooms/memory-manager.test.ts b/apps/realtime/src/rooms/memory-manager.test.ts index 2153c347136..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,28 @@ 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')) 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/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 { From a76e12e88349498af2a8228e748cc411bf0dfe59 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 13:13:15 -0700 Subject: [PATCH 4/5] fix(realtime): workflow-deletion always notifies + cleans by socket.io membership MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review-round findings on #5937: - Always emit `workflow-deleted` (was guarded by users.length>0), so a socket still in the Socket.IO room after a Redis presence eviction is told the workflow is gone before socketsLeave kicks it — the editor no longer keeps showing a deleted workflow. (Cursor: "Silent kick skips deletion event".) - Clean per-socket state for the UNION of live Socket.IO members and presence-tracked sockets, so an evicted/late-joined socket's room mapping + session are dropped too — not just presence-snapshot sockets. (Greptile: "Room deletion leaves reverse state".) - deleteRoom now logs AND rethrows on Redis failure (like addUserToRoom) so a failed wipe isn't reported as a clean deletion; the request surfaces it. (Greptile: "Room deletion failures are suppressed".) The two "deferred leave drops new membership" P1s were already fixed by the workspace-scoped leave in a prior commit (leave carries { workspaceId }; server no-ops on mismatch). 128 tests pass, tsc + biome clean. --- apps/realtime/src/rooms/redis-manager.ts | 3 ++ .../src/rooms/workflow-room-service.ts | 44 ++++++++++++------- 2 files changed, 31 insertions(+), 16 deletions(-) diff --git a/apps/realtime/src/rooms/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index 6c9b4fe30de..945d03d539c 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -312,10 +312,13 @@ export class RedisRoomManager implements IRoomManager { } 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 } } diff --git a/apps/realtime/src/rooms/workflow-room-service.ts b/apps/realtime/src/rooms/workflow-room-service.ts index e31eb2e5167..f12a6c337d4 100644 --- a/apps/realtime/src/rooms/workflow-room-service.ts +++ b/apps/realtime/src/rooms/workflow-room-service.ts @@ -22,32 +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) { - this.manager.emitToRoom(room, 'workflow-deleted', { - workflowId, - message: 'This workflow has been deleted', - timestamp: Date.now(), - }) + 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. - 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 state may linger even - // if a per-socket removal failed or a socket joined mid-teardown (matches the - // pre-refactor managers, which ended deletion with an unconditional room drop). + // 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)` ) } From ffbf40889617518387078e7f7de1a864eff86c1f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 13:26:30 -0700 Subject: [PATCH 5/5] refactor(files): drop module-scoped deferred-leave; rely on workspace-scoped leave MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes the one non-idiomatic construct (a module-level mutable `intendedFilesWorkspaceId` + queueMicrotask). It only guarded a same-workspace CONCURRENT remount, which doesn't occur in production (folder nav is shallow/no remount; list<->detail is sequential) — a dev-StrictMode-only case. The real cross-workspace race is already handled by the workspace-scoped leave: if B's join runs first (auto-leaving A), A's leave no-ops because the socket's current files room is B. Simpler, idiomatic, prod-correct. --- .../files/hooks/use-workspace-files-room.ts | 27 ++++--------------- 1 file changed, 5 insertions(+), 22 deletions(-) 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 92dbc94e5ce..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 @@ -14,16 +14,6 @@ const logger = createLogger('WorkspaceFilesRoom') const MAX_JOIN_RETRIES = 3 const JOIN_RETRY_BASE_MS = 1000 -/** - * The workspace whose files room this page currently wants to be in (module-scoped - * because only one Files page mounts at a time). A remount (React strict-mode, a - * fast route re-mount) runs the old cleanup then the new effect synchronously; the - * new effect re-claims this before the deferred leave fires, so the stale leave is - * skipped — preventing presence flapping and a leave-after-join race that could - * drop the fresh membership. - */ -let intendedFilesWorkspaceId: string | null = null - interface PresenceUpdatePayload extends PresenceAvatarUser { folderId?: string | null } @@ -110,7 +100,6 @@ export function useWorkspaceFilesRoom( } // Join now if the socket is already connected; `connect` covers (re)connects. - intendedFilesWorkspaceId = workspaceId if (socket.connected) join() socket.on('connect', join) socket.on('join-workspace-files-success', handleJoinSuccess) @@ -127,17 +116,11 @@ export function useWorkspaceFilesRoom( socket.off('workspace-files-changed', handleChanged) setPresenceUsers([]) - // Defer the leave: if the page re-mounts for the same workspace this tick, the - // new effect re-claims `intendedFilesWorkspaceId` and this leave is skipped, - // avoiding a flap and a leave-after-join race. A real navigation away leaves - // the value cleared/changed, so the leave fires. The leave is scoped to this - // workspace so, after a workspace switch, it can't evict the new room. - if (intendedFilesWorkspaceId === workspaceId) intendedFilesWorkspaceId = null - queueMicrotask(() => { - if (intendedFilesWorkspaceId !== workspaceId) { - socket.emit('leave-workspace-files', { workspaceId }) - } - }) + // 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])