mirror of
https://github.com/simstudioai/sim.git
synced 2026-02-03 11:14:58 -05:00
improvement(rooms): redis client closed should fail fast
This commit is contained in:
@@ -14,6 +14,7 @@ import { createLogger } from '@sim/logger'
|
||||
import { useParams } from 'next/navigation'
|
||||
import { io, type Socket } from 'socket.io-client'
|
||||
import { getEnv } from '@/lib/core/config/env'
|
||||
import { useOperationQueueStore } from '@/stores/operation-queue/store'
|
||||
|
||||
const logger = createLogger('SocketContext')
|
||||
|
||||
@@ -138,6 +139,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
|
||||
const [authFailed, setAuthFailed] = useState(false)
|
||||
const initializedRef = useRef(false)
|
||||
const socketRef = useRef<Socket | null>(null)
|
||||
const triggerOfflineMode = useOperationQueueStore((state) => state.triggerOfflineMode)
|
||||
|
||||
const params = useParams()
|
||||
const urlWorkflowId = params?.workflowId as string | undefined
|
||||
@@ -341,9 +343,12 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
|
||||
})
|
||||
})
|
||||
|
||||
socketInstance.on('join-workflow-error', ({ error }) => {
|
||||
socketInstance.on('join-workflow-error', ({ error, code }) => {
|
||||
isRejoiningRef.current = false
|
||||
logger.error('Failed to join workflow:', error)
|
||||
logger.error('Failed to join workflow:', { error, code })
|
||||
if (code === 'ROOM_MANAGER_UNAVAILABLE') {
|
||||
triggerOfflineMode()
|
||||
}
|
||||
})
|
||||
|
||||
socketInstance.on('workflow-operation', (data) => {
|
||||
|
||||
@@ -12,15 +12,49 @@ import {
|
||||
import { persistWorkflowOperation } from '@/socket/database/operations'
|
||||
import type { AuthenticatedSocket } from '@/socket/middleware/auth'
|
||||
import { checkRolePermission } from '@/socket/middleware/permissions'
|
||||
import type { IRoomManager } from '@/socket/rooms'
|
||||
import type { IRoomManager, UserSession } from '@/socket/rooms'
|
||||
import { WorkflowOperationSchema } from '@/socket/validation/schemas'
|
||||
|
||||
const logger = createLogger('OperationsHandlers')
|
||||
|
||||
export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
|
||||
socket.on('workflow-operation', async (data) => {
|
||||
const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
|
||||
const session = await roomManager.getUserSession(socket.id)
|
||||
if (!roomManager.isReady()) {
|
||||
socket.emit('operation-forbidden', {
|
||||
type: 'ROOM_MANAGER_UNAVAILABLE',
|
||||
message: 'Realtime unavailable',
|
||||
})
|
||||
if (data?.operationId) {
|
||||
socket.emit('operation-failed', {
|
||||
operationId: data.operationId,
|
||||
error: 'Realtime unavailable',
|
||||
retryable: true,
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
let workflowId: string | null = null
|
||||
let session: UserSession | null = null
|
||||
|
||||
try {
|
||||
workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
|
||||
session = await roomManager.getUserSession(socket.id)
|
||||
} catch (error) {
|
||||
logger.error('Error loading session for workflow operation:', error)
|
||||
socket.emit('operation-forbidden', {
|
||||
type: 'ROOM_MANAGER_UNAVAILABLE',
|
||||
message: 'Realtime unavailable',
|
||||
})
|
||||
if (data?.operationId) {
|
||||
socket.emit('operation-failed', {
|
||||
operationId: data.operationId,
|
||||
error: 'Realtime unavailable',
|
||||
retryable: true,
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if (!workflowId || !session) {
|
||||
socket.emit('operation-forbidden', {
|
||||
|
||||
@@ -48,6 +48,21 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
|
||||
operationId,
|
||||
} = data
|
||||
|
||||
if (!roomManager.isReady()) {
|
||||
socket.emit('operation-forbidden', {
|
||||
type: 'ROOM_MANAGER_UNAVAILABLE',
|
||||
message: 'Realtime unavailable',
|
||||
})
|
||||
if (operationId) {
|
||||
socket.emit('operation-failed', {
|
||||
operationId,
|
||||
error: 'Realtime unavailable',
|
||||
retryable: true,
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
const sessionWorkflowId = await roomManager.getWorkflowIdForSocket(socket.id)
|
||||
const session = await roomManager.getUserSession(socket.id)
|
||||
|
||||
@@ -37,6 +37,21 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager:
|
||||
socket.on('variable-update', async (data) => {
|
||||
const { workflowId: payloadWorkflowId, variableId, field, value, timestamp, operationId } = data
|
||||
|
||||
if (!roomManager.isReady()) {
|
||||
socket.emit('operation-forbidden', {
|
||||
type: 'ROOM_MANAGER_UNAVAILABLE',
|
||||
message: 'Realtime unavailable',
|
||||
})
|
||||
if (operationId) {
|
||||
socket.emit('operation-failed', {
|
||||
operationId,
|
||||
error: 'Realtime unavailable',
|
||||
retryable: true,
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
const sessionWorkflowId = await roomManager.getWorkflowIdForSocket(socket.id)
|
||||
const session = await roomManager.getUserSession(socket.id)
|
||||
|
||||
@@ -20,6 +20,15 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
|
||||
return
|
||||
}
|
||||
|
||||
if (!roomManager.isReady()) {
|
||||
logger.warn(`Join workflow rejected: Room manager unavailable`)
|
||||
socket.emit('join-workflow-error', {
|
||||
error: 'Realtime unavailable',
|
||||
code: 'ROOM_MANAGER_UNAVAILABLE',
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
logger.info(`Join workflow request from ${userId} (${userName}) for workflow ${workflowId}`)
|
||||
|
||||
// Verify workflow access
|
||||
@@ -128,12 +137,19 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
|
||||
// Undo socket.join and room manager entry if any operation failed
|
||||
socket.leave(workflowId)
|
||||
await roomManager.removeUserFromRoom(socket.id)
|
||||
socket.emit('join-workflow-error', { error: 'Failed to join workflow' })
|
||||
socket.emit('join-workflow-error', {
|
||||
error: roomManager.isReady() ? 'Failed to join workflow' : 'Realtime unavailable',
|
||||
code: roomManager.isReady() ? undefined : 'ROOM_MANAGER_UNAVAILABLE',
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
socket.on('leave-workflow', async () => {
|
||||
try {
|
||||
if (!roomManager.isReady()) {
|
||||
return
|
||||
}
|
||||
|
||||
const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
|
||||
const session = await roomManager.getUserSession(socket.id)
|
||||
|
||||
|
||||
@@ -26,6 +26,10 @@ export class MemoryRoomManager implements IRoomManager {
|
||||
logger.info('MemoryRoomManager initialized (single-pod mode)')
|
||||
}
|
||||
|
||||
isReady(): boolean {
|
||||
return true
|
||||
}
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
this.workflowRooms.clear()
|
||||
this.socketToWorkflow.clear()
|
||||
|
||||
@@ -96,17 +96,6 @@ export class RedisRoomManager implements IRoomManager {
|
||||
this._io = io
|
||||
this.redis = createClient({
|
||||
url: redisUrl,
|
||||
socket: {
|
||||
reconnectStrategy: (retries) => {
|
||||
if (retries > 10) {
|
||||
logger.error('Redis reconnection failed after 10 attempts')
|
||||
return new Error('Redis reconnection failed')
|
||||
}
|
||||
const delay = Math.min(retries * 100, 3000)
|
||||
logger.warn(`Redis reconnecting in ${delay}ms (attempt ${retries})`)
|
||||
return delay
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
this.redis.on('error', (err) => {
|
||||
@@ -122,12 +111,21 @@ export class RedisRoomManager implements IRoomManager {
|
||||
logger.info('Redis client ready')
|
||||
this.isConnected = true
|
||||
})
|
||||
|
||||
this.redis.on('end', () => {
|
||||
logger.warn('Redis client connection closed')
|
||||
this.isConnected = false
|
||||
})
|
||||
}
|
||||
|
||||
get io(): Server {
|
||||
return this._io
|
||||
}
|
||||
|
||||
isReady(): boolean {
|
||||
return this.isConnected
|
||||
}
|
||||
|
||||
async initialize(): Promise<void> {
|
||||
if (this.isConnected) return
|
||||
|
||||
|
||||
@@ -48,6 +48,11 @@ export interface IRoomManager {
|
||||
*/
|
||||
initialize(): Promise<void>
|
||||
|
||||
/**
|
||||
* Whether the room manager is ready to serve requests
|
||||
*/
|
||||
isReady(): boolean
|
||||
|
||||
/**
|
||||
* Clean shutdown
|
||||
*/
|
||||
|
||||
@@ -85,6 +85,11 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
|
||||
res.end(JSON.stringify({ error: authResult.error }))
|
||||
return
|
||||
}
|
||||
|
||||
if (!roomManager.isReady()) {
|
||||
sendError(res, 'Room manager unavailable', 503)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Handle workflow deletion notifications from the main API
|
||||
|
||||
Reference in New Issue
Block a user