jan/openreplay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: improve socket session finding reliability

Browse source 
Add retry mechanism when finding session sockets to handle race
conditions. The function now makes up to 3 attempts with 500ms delays
before failing, which should improve reliability when sessions are
connecting or reconnecting.

Also includes code style reformatting for consistency.

Signed-off-by: rjshrjndrn <rjshrjndrn@gmail.com>
This commit is contained in:
rjshrjndrn 2025-03-22 16:20:45 +01:00
parent f4bf1b8960
commit 508c87725f
2 changed files with 195 additions and 180 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
assist/Dockerfile
View file

@ -7,6 +7,8 @@ ENV ENTERPRISE_BUILD=${envarg} \
MAXMINDDB_FILE=/home/openreplay/geoip.mmdb \
PRIVATE_ENDPOINTS=false \
LISTEN_PORT=9001 \
MAX_RETRIES=3 \
RETRY_DELAY=500 \
NODE_ENV=production
WORKDIR /work
COPY package.json .

33
assist/utils/socketHandlers.js
View file

@ -23,8 +23,8 @@ const {
IncreaseOnlineRooms,
DecreaseOnlineRooms,
} = require('../utils/metrics');
const {logger} = require('./logger');
const deepMerge = require('@fastify/deepmerge')({all: true});
const { logger } = require('./logger');
const deepMerge = require('@fastify/deepmerge')({ all: true });
const findSessionSocketId = async (io, roomId, tabId) => {
let pickFirstSession = tabId === undefined;
@ -58,12 +58,12 @@ async function getRoomData(io, roomID) {
tabsCount = -1;
agentsCount = -1;
}
return {tabsCount, agentsCount, tabIDs, agentIDs};
return { tabsCount, agentsCount, tabIDs, agentIDs };
}
function processNewSocket(socket) {
socket._connectedAt = new Date();
let {projectKey: connProjectKey, sessionId: connSessionId, tabId: connTabId} = extractPeerId(socket.handshake.query.peerId);
let { projectKey: connProjectKey, sessionId: connSessionId, tabId: connTabId } = extractPeerId(socket.handshake.query.peerId);
socket.handshake.query.roomId = `${connProjectKey}-${connSessionId}`;
socket.handshake.query.projectKey = connProjectKey;
socket.handshake.query.sessId = connSessionId;
@ -78,7 +78,7 @@ async function onConnect(socket) {
IncreaseOnlineConnections(socket.handshake.query.identity);
const io = getServer();
const {tabsCount, agentsCount, tabIDs, agentIDs} = await getRoomData(io, socket.handshake.query.roomId);
const { tabsCount, agentsCount, tabIDs, agentIDs } = await getRoomData(io, socket.handshake.query.roomId);
if (socket.handshake.query.identity === IDENTITIES.session) {
// Check if session with the same tabID already connected, if so, refuse new connexion
@ -150,7 +150,7 @@ async function onDisconnect(socket) {
}
logger.debug("checking for number of connected agents and sessions");
const io = getServer();
let {tabsCount, agentsCount, tabIDs, agentIDs} = await getRoomData(io, socket.handshake.query.roomId);
let { tabsCount, agentsCount, tabIDs, agentIDs } = await getRoomData(io, socket.handshake.query.roomId);
if (tabsCount === -1 && agentsCount === -1) {
DecreaseOnlineRooms();
@ -175,14 +175,14 @@ async function onUpdateEvent(socket, ...args) {
}
args[0] = updateSessionData(socket, args[0])
socket.handshake.query.sessionInfo = deepMerge(socket.handshake.query.sessionInfo, args[0]?.data, {tabId: args[0]?.meta?.tabId});
socket.handshake.query.sessionInfo = deepMerge(socket.handshake.query.sessionInfo, args[0]?.data, { tabId: args[0]?.meta?.tabId });
// Update sessionInfo for all agents in the room
const io = getServer();
const connected_sockets = await io.in(socket.handshake.query.roomId).fetchSockets();
for (let item of connected_sockets) {
if (item.handshake.query.identity === IDENTITIES.session && item.handshake.query.sessionInfo) {
item.handshake.query.sessionInfo = deepMerge(item.handshake.query.sessionInfo, args[0]?.data, {tabId: args[0]?.meta?.tabId});
item.handshake.query.sessionInfo = deepMerge(item.handshake.query.sessionInfo, args[0]?.data, { tabId: args[0]?.meta?.tabId });
} else if (item.handshake.query.identity === IDENTITIES.agent) {
socket.to(item.id).emit(EVENTS_DEFINITION.listen.UPDATE_EVENT, args[0]);
}
@ -213,7 +213,20 @@ async function onAny(socket, eventName, ...args) {
handleEvent(eventName, socket, args[0]);
logger.debug(`received event:${eventName}, from:${socket.handshake.query.identity}, sending message to session of room:${socket.handshake.query.roomId}`);
const io = getServer();
let socketId = await findSessionSocketId(io, socket.handshake.query.roomId, args[0]?.meta?.tabId);
const MAX_RETRIES = parseInt(process.env.MAX_RETRIES || 3);
const RETRY_DELAY = parseInt(process.env.RETRY_DELAY || 500); // ms
async function findSessionWithRetry(io, roomId, tabId, retries = 0) {
let socketId = await findSessionSocketId(io, socket.handshake.query.roomId, tabId);
if (socketId === null && retries < MAX_RETRIES) {
logger.debug(`Session not found, retry ${retries + 1}/${MAX_RETRIES}`);
await new Promise(resolve => setTimeout(resolve, RETRY_DELAY));
return findSessionWithRetry(io, roomId, tabId, retries + 1);
}
return socketId;
}
let socketId = await findSessionWithRetry(io, socket.handshake.query.roomId, args[0]?.meta?.tabId);
if (socketId === null) {
logger.debug(`session not found for:${socket.handshake.query.roomId}`);
io.to(socket.id).emit(EVENTS_DEFINITION.emit.NO_SESSIONS);
@ -227,7 +240,7 @@ async function onAny(socket, eventName, ...args) {
// Back compatibility (add top layer with meta information)
function updateSessionData(socket, sessionData) {
if (sessionData?.meta === undefined && socket.handshake.query.identity === IDENTITIES.session) {
sessionData = {meta: {tabId: socket.handshake.query.tabId, version: 1}, data: sessionData};
sessionData = { meta: { tabId: socket.handshake.query.tabId, version: 1 }, data: sessionData };
}
return sessionData
}