feat: use a different RequestID for each ws request within a single connection (#4373)

Signed-off-by: nikolay <[email protected]>

Author: natanasow

packages/ws-server/src/webSocketServer.ts

@@ -71,10 +71,8 @@ const app = websockify(new Koa());
 
 app.ws.use((ctx: Koa.Context, next: Koa.Next) => {
   const connectionId = subscriptionService.generateId();
-  const requestId = uuid();
   ctx.websocket.id = connectionId;
-  ctx.websocket.requestId = requestId;
-  return context.run({ requestId, connectionId }, next);
+  next();
 });
 
 app.ws.use(async (ctx: Koa.Context) => {
@@ -86,12 +84,6 @@ app.ws.use(async (ctx: Koa.Context) => {
   ctx.websocket.limiter = limiter;
   ctx.websocket.wsMetricRegistry = wsMetricRegistry;
 
-  const requestDetails = new RequestDetails({
-    requestId: ctx.websocket.requestId,
-    ipAddress: ctx.request.ip,
-    connectionId: ctx.websocket.id,
-  });
-
   logger.info(
     // @ts-ignore
     `New connection established. Current active connections: ${ctx.app.server._connections}`,
@@ -115,9 +107,17 @@ app.ws.use(async (ctx: Koa.Context) => {
   limiter.applyLimits(ctx);
 
   // listen on message event
-  ctx.websocket.on(
-    'message',
-    AsyncResource.bind(async (msg) => {
+  ctx.websocket.on('message', async (msg) => {
+    const requestId = uuid();
+    ctx.websocket.requestId = requestId;
+
+    const requestDetails = new RequestDetails({
+      requestId,
+      ipAddress: ctx.request.ip,
+      connectionId: ctx.websocket.id,
+    });
+
+    context.run({ requestId, connectionId: requestDetails.connectionId! }, async () => {
       // Increment the total messages counter for each message received
       wsMetricRegistry.getCounter('totalMessageCounter').inc();
 
@@ -225,8 +225,8 @@ app.ws.use(async (ctx: Koa.Context) => {
       // Update the connection duration histogram with the calculated duration
       const methodLabel = Array.isArray(request) ? WS_CONSTANTS.BATCH_REQUEST_METHOD_NAME : request.method;
       wsMetricRegistry.getHistogram('messageDuration').labels(methodLabel).observe(msgDurationInMiliSeconds);
-    }),
-  );
+    });
+  });
 
   if (pingInterval > 0) {
     setInterval(async () => {