Scaling Gateways

The default room map lives inside one process, so broadcasts stop at the replica boundary until a Redis adapter carries them.

Syntaxapp.useWebSocketAdapter(adapter) // before app.listen()

socket.io keeps its room map in an adapter: the component it asks to deliver a broadcast. The default adapter is in-memory. With one process that is invisible. With two replicas behind a load balancer, each process knows only the sockets it is holding, so server.to('room:42').emit() reaches the members on that replica and silently misses the rest. Nothing throws. Half your users just never see the message.

Swap the adapter

@socket.io/redis-adapter (8.3.0) replaces the default with one that publishes every broadcast over Redis pub/sub, so each replica re-emits to its own local members. In Nest you subclass IoAdapter from @nestjs/platform-socket.io, override createIOServer(), and install it with app.useWebSocketAdapter() before app.listen().

Redis alone is not enough

A socket.io connection starts with HTTP long-polling by default, and those follow-up requests must land on the replica that began the handshake. Without that you get intermittent 400 responses. Either enable sticky sessions on the load balancer or set transports: ['websocket'] in every client so there is one request and no session to stick.

What stays local

The adapter forwards packets; it stores nothing. socket.rooms, and any counter or Map you keep in a provider, describe one replica only. Use fetchSockets(), which returns remote sockets too, read your own fields from socket.data, and use serverSideEmit() for replica-to-replica messages.

Example

Example · typescript
// src/adapters/redis-io.adapter.ts
import { INestApplication, Logger } from '@nestjs/common';
import { IoAdapter } from '@nestjs/platform-socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';
import { Server, ServerOptions } from 'socket.io';

export class RedisIoAdapter extends IoAdapter {
  private readonly logger = new Logger(RedisIoAdapter.name);
  private adapterConstructor: ReturnType<typeof createAdapter>;
  private clients: ReturnType<typeof createClient>[] = [];

  constructor(
    app: INestApplication,
    private readonly url: string,
  ) {
    super(app); // IoAdapter takes the app so it can reuse the HTTP server
  }

  async connectToRedis(): Promise<void> {
    // Two connections: a subscriber cannot issue other commands, so the
    // publisher is duplicated rather than shared.
    const pubClient = createClient({ url: this.url });
    const subClient = pubClient.duplicate();

    // Without these listeners a dropped Redis connection is an
    // unhandled 'error' event and the process exits.
    pubClient.on('error', (e) => this.logger.error(e));
    subClient.on('error', (e) => this.logger.error(e));

    await Promise.all([pubClient.connect(), subClient.connect()]);
    this.clients = [pubClient, subClient];
    this.adapterConstructor = createAdapter(pubClient, subClient);
  }

  // Called once per gateway server, during app.listen().
  createIOServer(port: number, options?: ServerOptions): Server {
    const server: Server = super.createIOServer(port, options);
    server.adapter(this.adapterConstructor); // every broadcast now goes via Redis
    return server;
  }

  async closeRedis(): Promise<void> {
    await Promise.all(this.clients.map((client) => client.quit()));
  }
}

When to use it

  • Keep a chat room working after the deployment is scaled from one pod to three.
  • Report a true online-user count instead of one count per replica.
  • Push a job-completed event from the worker replica to the browser on another replica.
  • Invalidate an in-memory cache on every replica without adding a second message bus.

More examples

Installing it in main.ts

The ordering is the whole lesson: connect, install, then listen. A call to useWebSocketAdapter() placed after listen() compiles, runs, logs nothing unusual, and leaves you with exactly the per-replica behaviour you were trying to remove. Keep the Redis URL in configuration rather than hard-coded so a local single-replica run can point at a local Redis.

Example · typescript
// src/main.ts
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { RedisIoAdapter } from './adapters/redis-io.adapter';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  const adapter = new RedisIoAdapter(
    app,
    process.env.REDIS_URL ?? 'redis://localhost:6379',
  );

  // Connect first: createIOServer() runs while listen() boots the gateways,
  // and it needs the adapter constructor to already exist.
  await adapter.connectToRedis();

  // Must happen before listen(). After that the gateways are built and
  // still hold the default in-memory adapter.
  app.useWebSocketAdapter(adapter);

  app.enableShutdownHooks();
  await app.listen(3000);
}

bootstrap();

Two spellings of the same adapter wiring

Worth recognising both, because tutorials use the constructor option while every Nest example uses the method call. The .local pair matters during a rolling deploy: a replica shutting down can clean up only its own sockets, and that is exactly what .local expresses.

Example · typescript
// src/adapters/equivalent-forms.ts
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';
import { Server } from 'socket.io';

const pubClient = createClient({ url: 'redis://localhost:6379' });
const subClient = pubClient.duplicate();
await Promise.all([pubClient.connect(), subClient.connect()]);

// Shorthand, as the socket.io docs show it: the adapter as a server option.
const io = new Server(3000, { adapter: createAdapter(pubClient, subClient) });

// Longhand: construct the server, then set the adapter. Identical result,
// and the only form available inside a Nest IoAdapter subclass, because
// super.createIOServer() has already built the server by then.
const io2 = new Server(3000);
io2.adapter(createAdapter(pubClient, subClient));

// Once an adapter is installed, these two differ:
io.to('room:42').emit('tick', 1); // every replica's members
io.local.to('room:42').emit('tick', 1); // only sockets on this process

// The same .local modifier exists on the cluster-wide helpers.
await io.in('room:42').socketsLeave('room:42'); // everywhere
await io.local.in('room:42').socketsLeave('room:42'); // here only

Cluster-wide counts and replica-to-replica messages

Two things that quietly break at two replicas and are fixed here. Any presence or seat count must come from fetchSockets() rather than a local variable, and any in-process cache that gateways read needs an invalidation signal, which serverSideEmit() provides without inventing a second pub/sub channel. Note that serverSideEmit does not call the sender's own handler, so invalidate locally as well.

Example · typescript
// src/presence/presence.gateway.ts
import {
  MessageBody,
  OnGatewayInit,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from '@nestjs/websockets';
import { Server } from 'socket.io';
import { SettingsCache } from './settings.cache';

@WebSocketGateway()
export class PresenceGateway implements OnGatewayInit {
  @WebSocketServer()
  private readonly server: Server;

  constructor(private readonly settings: SettingsCache) {}

  afterInit(server: Server): void {
    // serverSideEmit (socket.io 4.1.0+) travels between replicas and is
    // never delivered to clients. Receive it with server.on().
    server.on('settings:changed', (version: number) => {
      this.settings.invalidate(version);
    });
  }

  async publishSettings(version: number): Promise<void> {
    this.settings.invalidate(version); // serverSideEmit skips the sender
    this.server.serverSideEmit('settings:changed', version);
  }

  @SubscribeMessage('presence:count')
  async count(@MessageBody('room') room: string) {
    // fetchSockets() (socket.io 4.0.0+) returns RemoteSocket objects for
    // sockets held by other replicas, with their socket.data included.
    // A counter kept in this provider would only ever see one replica.
    const sockets = await this.server.in(room).fetchSockets();
    return {
      room,
      sockets: sockets.length,
      users: new Set(sockets.map((s) => s.data.userId as string)).size,
    };
  }
}

Related

nest ws roomsnest gatewaysnest cache redisnest k8snest shutdown

Discussion

  • Be the first to comment on this lesson.