fix(socket.io): keep the message stream alive when a handler throws (#17666)

* fix(socket.io): keep the message stream alive when a handler throws

a handler that throws synchronously escapes the mergeMap in bindMessageHandlers and
errors the stream, and through the framework wiring the callback is always async, so a
handler whose error filter rethrows rejects the same way: the stream dies, that event
goes silent for the client and rxjs reports an unhandled error

the ws adapter got a sync-only guard for this in 39fbdda, the socket.io adapter had
none, and neither guard covers the async case, which is the only one a real gateway
can reach

both adapters now defer the handler call and catchError the stream, so a sync throw and
a rejected handler only drop that one message and the stream survives. a filter that
rethrows is an app bug, so the failure gets logged through the adapter logger instead
of being swallowed silently

* style(socket.io,ws): format adapter guard changes

Run prettier over the files touched by the stream guard. lint:ci only
runs oxlint, so the formatting drift would otherwise land and be
reformatted by the next contributor's lint-staged run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Kamil Myśliwiec <mail@kamilmysliwiec.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

packages/platform-socket.io/adapters/io-adapter.ts

@@ -2,9 +2,18 @@ import {
   AbstractWsAdapter,
   type MessageMappingProperties,
 } from '@nestjs/websockets';
-import { fromEvent, Observable } from 'rxjs';
-import { filter, first, map, mergeMap, share, takeUntil } from 'rxjs/operators';
+import { defer, EMPTY, fromEvent, Observable } from 'rxjs';
+import {
+  catchError,
+  filter,
+  first,
+  map,
+  mergeMap,
+  share,
+  takeUntil,
+} from 'rxjs/operators';
 import { Namespace, Server, ServerOptions, Socket } from 'socket.io';
+import { Logger } from '@nestjs/common';
 import { isFunction, isNil } from '@nestjs/common/internal';
 import { DISCONNECT_EVENT } from '@nestjs/websockets/internal';
 
@@ -12,6 +21,7 @@ import { DISCONNECT_EVENT } from '@nestjs/websockets/internal';
  * @publicApi
  */
 export class IoAdapter extends AbstractWsAdapter {
+  protected readonly logger = new Logger(IoAdapter.name);
   private readonly disconnectMap = new WeakMap<Socket, Observable<any>>();
 
   public create(
@@ -51,9 +61,17 @@ export class IoAdapter extends AbstractWsAdapter {
       const source$ = fromEvent(socket, message).pipe(
         mergeMap((payload: any) => {
           const { data, ack } = this.mapPayload(payload);
-          return transform(callback(data, ack)).pipe(
+          // defer turns a sync throw into an error notification and catchError
+          // stops it from tearing down this event's stream, which would silence
+          // that event for the client. A handler whose error filter rethrows is
+          // an app bug, so the failure gets logged instead of dropped silently.
+          return defer(() => transform(callback(data, ack))).pipe(
             filter((response: any) => !isNil(response)),
             map((response: any) => [response, ack, isAckHandledManually]),
+            catchError(err => {
+              this.logger.error(err);
+              return EMPTY;
+            }),
           );
         }),
         takeUntil(disconnect$),

packages/platform-socket.io/test/io-adapter.spec.ts

@@ -1,5 +1,9 @@
-import { of } from 'rxjs';
+import { EventEmitter } from 'events';
+import { config, from, mergeAll, of } from 'rxjs';
 import { IoAdapter } from '../adapters/io-adapter.js';
+import { WsProxy } from '../../websockets/context/ws-proxy.js';
+import { WsExceptionsHandler } from '../../websockets/exceptions/ws-exceptions-handler.js';
+import { WebSocketsController } from '../../websockets/web-sockets-controller.js';
 
 describe('IoAdapter', () => {
   let adapter: IoAdapter;
@@ -44,5 +48,127 @@ describe('IoAdapter', () => {
       );
       expect(messageCalls).toHaveLength(2);
     });
+
+    it('should not let a throwing handler tear down the message stream', async () => {
+      const socket = new EventEmitter() as any;
+      let calls = 0;
+      const handler = {
+        message: 'test-event',
+        methodName: 'handleTestEvent',
+        callback: () => {
+          calls++;
+          if (calls === 1) {
+            throw new Error('handler blew up');
+          }
+          return { event: 'test-event-reply', data: 'recovered' };
+        },
+        isAckHandledManually: false,
+      };
+
+      const replies: any[] = [];
+      socket.on('test-event-reply', (payload: any) => replies.push(payload));
+
+      adapter.bindMessageHandlers(socket, [handler], (data: any) => of(data));
+
+      socket.emit('test-event', { data: {} });
+      socket.emit('test-event', { data: {} });
+
+      expect(calls).toBe(2);
+      expect(replies).toEqual(['recovered']);
+    });
+
+    it('should not ack a throwing message but keep acking the next one', () => {
+      const socket = new EventEmitter() as any;
+      let calls = 0;
+      const handler = {
+        message: 'test-event',
+        methodName: 'handleTestEvent',
+        callback: () => {
+          calls++;
+          if (calls === 1) {
+            throw new Error('handler blew up');
+          }
+          return { data: 'ok' };
+        },
+        isAckHandledManually: false,
+      };
+
+      const ack = vi.fn();
+      adapter.bindMessageHandlers(socket, [handler], (data: any) => of(data));
+
+      socket.emit('test-event', [{ a: 1 }, ack]);
+      expect(calls).toBe(1);
+      expect(ack).not.toHaveBeenCalled();
+
+      socket.emit('test-event', [{ a: 1 }, ack]);
+      expect(calls).toBe(2);
+      expect(ack).toHaveBeenCalledWith({ data: 'ok' });
+    });
+
+    it('should not let a rejected handler tear down the message stream', async () => {
+      const unhandled: unknown[] = [];
+      config.onUnhandledError = err => unhandled.push(err);
+      const logError = vi
+        .spyOn(adapter['logger'], 'error')
+        .mockImplementation(() => undefined);
+
+      // an app-level @Catch() filter that rethrows
+      const exceptionsHandler = new WsExceptionsHandler();
+      exceptionsHandler.setCustomFilters([
+        {
+          exceptionMetatypes: [],
+          func: (exception: any) => {
+            throw exception;
+          },
+        },
+      ] as any);
+
+      const socket = new EventEmitter() as any;
+      const replies: any[] = [];
+      socket.on('reply', (payload: any) => replies.push(payload));
+
+      let calls = 0;
+      const callback = new WsProxy()
+        .create(
+          async () => {
+            if (++calls === 1) throw new Error('handler blew up');
+            return { event: 'reply', data: 'ok' };
+          },
+          exceptionsHandler,
+          'test-event',
+        )
+        .bind(undefined, socket);
+
+      // same transform WebSocketsController.subscribeMessages passes in
+      const transform = (data: any) =>
+        from(
+          WebSocketsController.prototype.pickResult.call(undefined, data),
+        ).pipe(mergeAll());
+
+      adapter.bindMessageHandlers(
+        socket,
+        [
+          {
+            message: 'test-event',
+            methodName: 'm',
+            callback,
+            isAckHandledManually: false,
+          },
+        ],
+        transform,
+      );
+
+      const flush = () => new Promise(resolve => setTimeout(resolve, 5));
+      socket.emit('test-event', {});
+      await flush();
+      socket.emit('test-event', {});
+      await flush();
+      config.onUnhandledError = null;
+
+      expect(calls).toBe(2);
+      expect(replies).toEqual(['ok']);
+      expect(unhandled).toEqual([]);
+      expect(logError).toHaveBeenCalledTimes(1);
+    });
   });
 });

packages/platform-ws/adapters/ws-adapter.ts

@@ -4,7 +4,14 @@ import * as http from 'http';
 import { createRequire } from 'module';
 import type { Duplex } from 'stream';
 import { EMPTY, fromEvent, Observable } from 'rxjs';
-import { filter, first, mergeMap, share, takeUntil } from 'rxjs/operators';
+import {
+  catchError,
+  filter,
+  first,
+  mergeMap,
+  share,
+  takeUntil,
+} from 'rxjs/operators';
 import { loadPackageSync, isNil, normalizePath } from '@nestjs/common/internal';
 import {
   CLOSE_EVENT,
@@ -159,6 +166,13 @@ export class WsAdapter extends AbstractWsAdapter {
       mergeMap(data =>
         this.bindMessageHandler(data, handlersMap, transform).pipe(
           filter(result => !isNil(result)),
+          // a handler that rejects through an error filter that rethrows
+          // would error the client's stream and silence every later message;
+          // that rethrow is an app bug, so it gets logged, not dropped silently
+          catchError(err => {
+            this.logger.error(err);
+            return EMPTY;
+          }),
         ),
       ),
       takeUntil(close$),

packages/platform-ws/test/ws-adapter.spec.ts

@@ -1,6 +1,18 @@
+import { EventEmitter } from 'events';
 import { createServer } from 'http';
-import { lastValueFrom, of, toArray, type Observable } from 'rxjs';
+import {
+  config,
+  from,
+  lastValueFrom,
+  mergeAll,
+  of,
+  toArray,
+  type Observable,
+} from 'rxjs';
 import { WsAdapter } from '../adapters/ws-adapter.js';
+import { WsProxy } from '../../websockets/context/ws-proxy.js';
+import { WsExceptionsHandler } from '../../websockets/exceptions/ws-exceptions-handler.js';
+import { WebSocketsController } from '../../websockets/web-sockets-controller.js';
 
 describe('WsAdapter', () => {
   describe('bindMessageHandler', () => {
@@ -136,6 +148,78 @@ describe('WsAdapter', () => {
     });
   });
 
+  describe('bindMessageHandlers', () => {
+    const frame = (payload: unknown) => ({ data: JSON.stringify(payload) });
+
+    it('should not let a rejecting handler tear down the message stream', async () => {
+      const unhandled: unknown[] = [];
+      config.onUnhandledError = err => unhandled.push(err);
+
+      // an app-level @Catch() filter that rethrows
+      const exceptionsHandler = new WsExceptionsHandler();
+      exceptionsHandler.setCustomFilters([
+        {
+          exceptionMetatypes: [],
+          func: (exception: any) => {
+            throw exception;
+          },
+        },
+      ] as any);
+
+      const adapter = new WsAdapter();
+      const logError = vi
+        .spyOn(adapter['logger'], 'error')
+        .mockImplementation(() => undefined);
+
+      const client = new EventEmitter() as any;
+      client.readyState = 1; // OPEN_STATE
+      const replies: any[] = [];
+      client.send = (payload: string) => replies.push(JSON.parse(payload));
+
+      let calls = 0;
+      const callback = new WsProxy()
+        .create(
+          async () => {
+            if (++calls === 1) throw new Error('handler blew up');
+            return 'ok';
+          },
+          exceptionsHandler,
+          'known',
+        )
+        .bind(undefined, client);
+
+      // same transform WebSocketsController.subscribeMessages passes in
+      const realTransform = (data: any) =>
+        from(
+          WebSocketsController.prototype.pickResult.call(undefined, data),
+        ).pipe(mergeAll());
+
+      adapter.bindMessageHandlers(
+        client,
+        [
+          {
+            message: 'known',
+            methodName: 'm',
+            callback,
+            isAckHandledManually: false,
+          },
+        ],
+        realTransform,
+      );
+
+      client.emit('message', frame({ event: 'known', data: {} }));
+      await new Promise(resolve => setTimeout(resolve, 5));
+      client.emit('message', frame({ event: 'known', data: {} }));
+      await new Promise(resolve => setTimeout(resolve, 5));
+      config.onUnhandledError = null;
+
+      expect(calls).toBe(2);
+      expect(replies).toEqual(['ok']);
+      expect(unhandled).toEqual([]);
+      expect(logError).toHaveBeenCalledTimes(1);
+    });
+  });
+
   describe('dispose', () => {
     it('should remove the upgrade listener it added to a caller-supplied server', async () => {
       const httpServer = createServer();