웹소켓 서버

지난 포스트에서는 SvelteKit과 node-adapter를 사용해 custom server를 구축하여 웹소켓 서버를 초기화하는 로직들을 간단히 작성했다. 이번 포스트에서는 웹소켓 서버에서 메시지 수신 시 처리를 어떻게 할지 고민하고 구현해보겠다.

그냥 수신한 메시지를 처리하고 사용자한테 전달만 하면 되는 거 아닌가요?

Node.js가 싱글 스레드 기반 런타임이기 때문에, CPU 코어를 모두 활용하지 못한다는 단점이 있다. 이를 극복하기 위해 pm2의 클러스터 모드(멀티 프로세싱)를 활용하려고 한다. 첫 회사를 그만둔 후 구직 중 면접에서 pm2를 왜 여지껏 사용하지 않았는지 집요하게 여쭤봐주신 면접관님 덕분에 프리랜스를 진행할 때에도, 개인 프로젝트를 진행할 때에도 항상 pm2 사용을 고려해보게 되었다(그 당시에는 pm2라는게 있구나~ 정도만 알고 있었고 사용해본 경험은 없었다). 당시에는 면접을 망쳤다는 생각 뿐이었지만 면접을 망칠 때마다 얻어가는 것이 있어서 좋다는 생각이 든다.
멀티 프로세스 환경을 구축할 때에는 '그냥' 구현할 때보다 훨씬 많은 고민을 필요로 한다. 예를 들어, 싱글 프로세스 환경에서 웹소켓 서버를 구축한다고 가정해보자.

이 메시지 전송이라는 행위를 어떻게 수행해야 하는가? wss.clients에는 현재 웹소켓 서버에 연결된 클라이언트들을 저장하기 때문에, 해당 필드의 각 요소들(socket)에 roomId를 추가해 저장하겠다. 라이브러리를 까보면 이런 타입으로 저장되어 있다.

// WebSocket Server
class Server<
  T extends typeof WebSocket.WebSocket = typeof WebSocket.WebSocket,
  U extends typeof IncomingMessage = typeof IncomingMessage,
> extends EventEmitter {
  options: ServerOptions<T, U>;
  path: string;
  clients: Set<InstanceType<T>>;
  ...
}

// WebSocket socket.
declare class WebSocket extends EventEmitter {
  ...

나는 이 WebSocket 타입에 userId, roomId 필드를 추가했다.

import { WebSocket as WebSocket_ } from 'ws';

export interface WebSocket extends WebSocket_ {
  roomId?: string;
  userId?: string;
}

(ConnectionManager를 만들면서 roomId로 접속 중인 클라이언트를 모두 가져올 clientPool을 추후에 추가했다. 읽다 보면 해당 내용이 나올 것이다)
이런 상황에서, 메시지 전송은 이렇게 수행하면 될 것이다.

wss.clients.filter(socket => socket.roomId === roomId).forEach(socket => socket.send(message));

(메서드 프로토타입을 대강 구현한 후 게임 진행을 맞춰보다가 roomId, userId 외에도 playerIdx까지 이에 포함시키면 어떨까 하는 생각이 들었다. 클라이언트들이 자신이 몇 번 인덱스인지 매번 보내도록 하는 것보다 서버 측에서 관리하는 것이 더 낫다고 판단했다.)
클라이언트에 포함된 필드들이 partial하다보니, 매번 사용할 때마다 유효성 검사를 진행하거나 assert하거나... 해야 하는데 이보다는 처음 연결할 땐 빈 필드, 연결 초기화 후엔 채워진 필드로 다루도록 아래와 같이 구현했다.

export interface WebSocket extends WebSocket_ {
  roomId?: string;
  userId?: string;
  playerIdx?: PlayerIdx;
}

interface ConnectionInfo {
  roomId: string;
  userId: string;
  playerIdx: PlayerIdx;
}

export type PendingWebSocket = WebSocket_ & Partial<ConnectionInfo>;

export type ActiveWebSocket = WebSocket_ & ConnectionInfo;

이후 메서드들은 전부 ActiveWebSocket으로부터 playerIdx와 userId, roomId를 가져다 쓰게 되었다.

Connection Manager

우선, 연결된 클라이언트들을 관리하기 위해 wss.clients를 모두 뒤져보는 것은 비효율적이다. 100개의 connection이 수립된 상태면 한 번의 웹소켓 메시지 전달에 100번의 탐색이, 1000개면 1000번의 탐색이 이뤄져야 한다. 이를 개선하기 위해 JS의 Map을 사용하겠다. key는 roomId, value는 connection들을 담은 배열로 구성해보겠다.

export class WebSocketConnectionManager {
  private clientPool: Map<string, WebSocket[]> = new Map();
  
  
  addClient({ roomId, client }: { roomId: string, client: ActiveWebSocket }) {
    const connections = this.clientPool.get(roomId);
    if (connections === undefined) {
      this.clientPool.set(roomId, [client]);
      return;
    }
    connections.push(client);
  }

  removeClient({ roomId, userId }: { roomId: string, userId: string }) {
    const connections = this.clientPool.get(roomId);
    if (!connections || connections.length === 0) return;
    connections.splice(connections.findIndex((e) => e.userId === userId), 1);
  }

  private getClientsByRoomId(roomId: string) {
    return this.clientPool.get(roomId);
  }
}

getter를 통해 가져온 property가 mutable하기 때문에 클라이언트(socket)를 반환받아 수정할 수 있는 문제가 있으므로, immutable한 형태로 반환하는 것을 추후에 고려해야 한다. TODO 주석을 남겨두고 미래의 나에게 맡기겠다(콜백을 통해 readonly array로 처리하든...). 메시지를 전송하는 로직은 다른 클래스에 일임하여 분리하겠다. 그 클래스는 바로...

Response Dispatcher

dispatch하는 주체는 클라이언트, 서버 양측에 존재해야 하므로 구분을 위해 response를 이름에 포함했다. ConnectionManager를 주입해서 방에 소속된 클라이언트들을 가져와 메시지를 보내는 형식으로 작성했다.

export class WebSocketResponseDispatcher {
  constructor(
    connectionManager: WebSocketConnectionManager,
  ) {
    this.connectionManager = connectionManager;
  }

  private connectionManager: WebSocketConnectionManager;

  dispatch({
    roomId, payload,
  }: WebSocketBrokerMessage) {
    const clients = this.connectionManager.getClientsByRoomId(roomId);
    if (clients === undefined) {
      return;
    }
    clients.forEach(client => {
      client.send(JSON.stringify(payload));
    });
  }
}

Message Broker

pm2를 사용해서 프로세스를 실행하면 wss는 몇 개 생성될까?

initWebSocketServer 내부에 pid를 출력하도록 해보니 cpu core 개수만큼 생성된다. 이런 경우를 가정해보자.

• 클라이언트 1은 room:1을 생성하고 접속했다.
• 클라이언트 1의 요청을 처음 수신한 서버는 7번 서버이고, 다른 클라이언트 2, 3, 4는 각각 6, 5, 4번 서버를 통해 room:1에 접속했다.
• 클라이언트 1이 게임을 시작하는 요청인 웹소켓 메시지 StartMessage를 전송했고, 3번 서버 인스턴스가 이를 수신했다.
• 이제 3번 서버 인스턴스는 방의 상태를 업데이트하여 db 및 redis에 적용하고, room:1에 접속한 클라이언트들에게 게임 시작을 알린다.

과연 클라이언트 1, 2, 3, 4는 게임을 제대로 시작이나 할 수 있을까? 각 턴에는 타임아웃이 있기 때문에(60s) 클라이언트 1은 첫 착수를 한 뒤 쓸쓸히 180초 뒤에 혼자 두 번째 착수를 맞이한다...

pm2를 통해 초기화되는 각 서버 인스턴스들은 어떻게 동작해야 클라이언트 1의 메시지를 2, 3, 4에게 잘 전달할 수 있을까? 우선 기존 예시를 코드로 작성해보겠다.

export class WebSocketMessageHandler {
  private async handleStart(client: ActiveWebSocket): Promise<MessageProcessResult> {
    // 1. DB에 저장하고..
    // 2. redis 상태 업데이트하고..
    // 3. 사용자한테 메시지 전달하고..
  }
  
  handleMessage(client: WebSocket, rawMessage: string) {
    try {
      const message = JSON.parse(rawMessage) as InboundWebSocketMessage;
      switch(message.type) {
        case 'START':
          this.handleStart(message);
          break;
        ...
}

3번은 이렇게 작성할 수 있겠다.

const startMessage: OutboundStartMessage = { type: 'START', /* 추가 정보들 */ };
responseDispatcher.dispatch(startMessage);

dispatcher 내의 dispatch가 호출하는 connectionManager의 clientPool이 서버 인스턴스마다 달라지니, 단순히 해당 인스턴스의 클라이언트들에게만 send하는 대신 Broker를 배치해보자. 이 Broker는 메시지를 보내야할 때마다 다른 서버 인스턴스들에게 이벤트를 전달해주고, 이를 수신하는 역할을 수행해야 한다. 프로세스 간 통신(node-ipc)을 통해 구현할지 다른 것을 통해 구현할지 고민하던 차에, redis의 pub/sub을 (이참에) 이용해보자는 생각이 들어 이를 택했다.

export class WebSocketMessageBroker {
  constructor(
    redis: RedisClientType,
    responseDispatcher: WebSocketResponseDispatcher,
  ) {
    this.publisher = redis;
    this.subscriber = redis.duplicate();
    this.subscriber.connect();
    this.responseDispatcher = responseDispatcher;
  }

  private subscriber: RedisClientType;
  private publisher: RedisClientType;
  private responseDispatcher: WebSocketResponseDispatcher;

  publishMessage({ message, roomId }: { roomId: string, message: OutboundWebSocketMessage }) {
    this.publisher.publish('message', JSON.stringify({ payload: message, roomId }));
  }

  subscribeMessage() {
    this.subscriber.subscribe('message', (rawMessage) => {
      const message = parseJson<WebSocketBrokerMessage>(rawMessage);
      if (typeof message === 'string') return;
      const { roomId, payload } = message;
      this.responseDispatcher.dispatch({ roomId, payload });
    });
  }
}

message 채널로 무언가를 publish하면 subscriber가 이를 받는 구조이다. subscriber는 쿼리 등 다른 작업을 수행하지 못 하고 subscribe만 해야 하기에 redis 객체를 duplicate하였다. 결과는 다음과 같다.

Message Handler

메시지를 받아서 처리하는 녀석도 필요하다. 많이 기니 각 메서드들의 세부 구현은 생략하겠다.

export class WebSocketMessageHandler {
  constructor(redis: RedisClientType) {
    this.redis = redis;
  }

  private redis: RedisClientType;
  
  async processMessage(client: ActiveWebSocket, message: InboundWebSocketMessage): Promise<MessageProcessResult> {
    switch (message.type) {
      case 'START':
        return this.handleStart(client);
      case "CONNECTED":
        return this.handleUserConnected(client, message);
      case "LEAVE":
        return this.handleUserLeave(client);
      case "READY":
        return this.handleReady(client);
      case "CANCEL_READY":
        return this.handleCancelReady(client);
      case "MOVE":
        return this.handleMove(client, message);
      case "REPORT":
        return this.handleReport(client, message);
      default:
        return {
          success: false,
          payload: { type: 'ERROR', }
        };
    }
  }
}

processMessage 메서드가 return하는 MessageProcessResult 타입은 이렇게 구성했다.

interface MessageProcessResult {
  success: boolean;
  shouldBroadcast: boolean;
  payload: OutboundWebSocketMessage;
}

success 필드가 false이면 서버에서 처리에 실패, true이면 서버에서 처리에 실패하지 않았다고, shouldBroadcast 필드가 false이면 요청자에게만 메시지를 전달하는 것으로 간주하겠다. 예를 들면...

private async handleStart(client: ActiveWebSocket): Promise<MessageProcessResult> {
  if (client.playerIdx !== 0) {
    return {
      success: true,
      shouldBroadcast: false,
      payload: { type: 'BAD_REQ', message: 'unauthorized' },
    };
  }
  ...

요청자에게만 메시지를 전달하는 경우에 대해 간단히 이야기하자면, pm2의 cluster 모드로 일반적인 http request를 받을 땐 인스턴스들이 '적절히' 돌아가며 요청을 처리하지만(round-robin load balancing으로 보이네요) websocket request의 경우 sticky session으로 취급된다. 즉, 요청자는 지정된 서버 인스턴스와만 상호작용하게 된다(로그를 통해 확인도 해봤다).

Orchestrator

한편, 위와 같이 핵심 기능들을 수행해줄 클래스들을 작성했는데 각 메서드들을 언제 어디서 호출할 것인가? 이벤트 기반으로 만들지, 조율자(orchestrator)가 이를 수행하게 할지 고민해봤는데 조율자를 두는 것이 서버 사이드에는 더 나을 것 같다는 판단을 했다. 조금 풀어서 설명해보자면...

조율자의 장단점

장점은 우선,

• 의존하는 각 클래스(핸들러, 브로커(와 디스패쳐), 커넥션 매니저)가 특정한 역할을 각자 담당하기 때문에 구조적인 책임 분리가 명확함
• 조율자가 각 클래스 간 상호작용을 중앙 제어하기 때문에 명확한 워크플로우를 가짐
• 독립적인 각 컴포넌트(클래스)의 유닛 테스트가 쉽다

정도가 되겠다. 단점은,

• 단일 실패 지점 - 전체 시스템에 영향을 미침
• "병렬" 처리가 제한적이다(이벤트 기반 아키텍처에선 동시에 다른 지점에서 이벤트를 구독하면 병렬 처리도 가능하다)
• 컴포넌트 간 통신이 제한적이다(조율자를 거치거나 끔찍한 행동의존성 추가하기)

이벤트 기반의 장단점

장점은

• 느슨한 결합
• 본질적인 비동기 처리(+병렬 처리)

정도가 되겠으며 단점은

• 연쇄적인 이벤트 등으로 인한 flow 추적 난이도 상승(코드 베이스를 꿰고 있어야 쉽다)
• 상태 일관성 유지 난이도 상승(여러 이벤트에 걸친 원자적 작업 구현이 어려움. redis에 의존한다던지 큐에 담아뒀다 여러 이벤트가 모두 resolve 될 때 다음 이벤트 핸들러를 진행한다던지...)
• 오류 처리(복구)가 복잡
• 테스트 복잡성

등이 있겠다.

Trade-off: 확장성

가장 고민한 지점은 위의 장단점이 아닌 "확장성"이었다. Pre-define되지 않은 어떤 특수한 기능이 추가되어야 할 때를 가정해보자.

이벤트 기반	새 이벤트 추가 → (기존 코드에서) 새 이벤트 발행 → 새로운 메서드(필요시 클래스도) 추가 및 새 이벤트 구독 → 기능 구현
조율자 기반	새 메서드 추가(필요시 클래스도 + 의존성) → 기존 코드(메서드) 수정(새로 추가한 메서드를 호출하도록) → 기능 구현

이런 과정들이 지속적으로 발생, 즉 새로운 기능이 지속적으로 추가되거나 확장 방향이 다양하면 orchestrator의 한계가 기술 부채가 될 것이다. 이 경우에는 이벤트 기반으로 접근하는 것이 유연성을 제공할 것이다. 반대로, 새로운 기능 추가가 드물고 확장 방향도 정해져 있으면 이벤트 기반 접근이 오히려 조율자보다 더 많은 공수를 필요로 할 것이다. 도메인, 즉 Blokus라는 보드게임을 실시간 웹앱으로 구현하는 과정에선 규칙도 이미 정해져 있고 워크플로우도 정의되어 있기 때문에 웹소켓 핸들러가 수행할 역할의 범위도 이미 정해져 있다. 추후 ELK 도입 예정이 있긴 한데, 이외의 다른 기능이 생길 것 같진 않기 때문에 중앙 제어 형식으로 구현하는 것이 낫다고 판단했다.

정리

서버 사이드의 웹소켓 핸들러는 이렇게 구현되었다. 클라이언트 사이드에서는 events.EventEmitter를 통해 이벤트 기반 아키텍처로 컴포넌트를 구성해볼 것이다(지금은 그냥 코드를 쓰다보니 갓 클래스를 만들어버리고 말았다).

마무리

쿠키-세션 기반 인증 같은 서버 측 로직들을 위주로 포스트 시리즈를 구성해보려 했는데 클라이언트 측 구현이 너무 커져서 클라이언트 측 로직들을 설명하지 않을 수 없을 것 같다. 구현 속도가 포스트를 작성하는 속도보다 지나치게 빠른 것도 문제다. 일단 아직 미구현된 피쳐들 구현을 일단락하고 포스트를 작성하는 것이 좋을 것 같다(크게 게임 마무리, retire, 통계 작업 등이 있겠다).