던플 개발기 (8): 실시간 가격 게이트웨이 — WebSocket, dedupe, 그리고 poison 메시지

지난 편에서 Kafka 이벤트 스트림까지 만들었다면, 이번엔 그 이벤트를 브라우저까지 흘려보내는 이야기입니다. 던플 마켓은 사용자가 게임을 켜 둔 채 옆에 열어 놓고 시세 흐름을 지켜보는 화면을 지향합니다. 그러려면 새 시세가 저장되는 순간 화면이 스스로 갱신돼야 합니다. F5를 누르게 만들면 그건 이미 터미널이 아닙니다.

그런데 막상 만들어 보니, 실시간의 어려움은 "빠르게 보내기"가 아니었습니다. Kafka에서 WebSocket으로 이벤트를 밀어내는 건 몇 줄이면 됩니다. 진짜 문제는 그 뒤에 있었습니다. 같은 이벤트가 두 번 오면? 깨진 메시지 하나가 스트림 전체를 막으면? 소켓이 반쯤 죽은 채로 남아 있으면? 이 편은 그 "눈에 안 띄어야 하는" 부분들을 어떻게 처리했는지 정리한 기록입니다.

한눈에 보면

게이트웨이는 DB write도 provider 호출도 소유하지 않습니다. 저장된 fact를 fan-out하는 transport 배선만 담당합니다.
push는 "새로 호출한 데이터"가 아니라 **"저장된 새 snapshot이 생겼다는 알림"**입니다. 클라이언트는 compact REST snapshot을 refetch해 채웁니다.
중복은 두 겹으로 거릅니다. consumer는 eventId로 broker 재전달을 dedupe하고, 클라이언트는 version(관측 epoch ms)으로 오래된 이벤트가 최신 행을 덮어쓰지 못하게 합니다.
dedupe 저장소는 무한히 커지지 않습니다. bounded FIFO seen-store(기본 1만 키)로 메모리를 묶었습니다.
poison 메시지는 죽이지 않고 치웁니다. 파싱 불가능한 메시지는 offset을 전진시키며 skip해, partition이 무한 재시도로 막히지 않게 했습니다.
정작 발을 잡은 건 로직이 아니라 런타임이었습니다. type:module 앱에 CommonJS 번들이 로드돼 require is not defined로 죽고, 토픽이 없는 채로 먼저 뜬 consumer가 죽었습니다.

왜 게이트웨이를 따로 두나

시세 상세 페이지에 실시간을 붙이는 가장 쉬운 길은, 화면에 데이터를 내려 주는 BFF가 WebSocket도 같이 들고 있는 것입니다. 그렇게 하지 않았습니다.

이유는 경계였습니다. 던플의 규칙은 명확합니다. market-bff는 REST를 외부에 제공하고 내부는 gRPC로 조회하며, Kafka를 직접 알지 않습니다. Kafka를 소비하는 곳은 오직 게이트웨이 하나입니다. 만약 BFF가 Kafka consumer까지 겸하면, "요청-응답으로 데이터를 조회하는 책임"과 "스트림을 구독해 소켓으로 밀어내는 책임"이 한 프로세스에 섞입니다. 둘은 부하 특성도, 장애 모드도, 스케일 단위도 다릅니다. 소켓이 수천 개 열린 상태에서 재시작하는 것과 stateless REST를 재시작하는 것은 전혀 다른 일입니다.

그래서 게이트웨이의 경계를 좁게 못박았습니다.

DB write를 소유하지 않습니다. provider(네오플)를 직접 호출하지도 않습니다.
push는 새 provider 호출이 아니라 저장된 fact의 알림입니다. 클라이언트는 compact REST snapshot을 refetch해 전체 데이터를 채웁니다.
모든 페이지를 실시간으로 만들지 않습니다. /item/[id] 상세와 관심 목록 기반만 구독 대상입니다.
도메인·프로토콜 로직은 market-realtime 라이브러리에 두고, 이 앱은 transport 배선만 합니다.

이 마지막 항목이 중요합니다. 구독 레지스트리, 프로토콜 파싱, freshness 판정 같은 "무슨 뜻인가"는 RealtimeHub가 들고, 게이트웨이 앱은 "WebSocket 소켓과 Kafka consumer를 hub에 연결"하는 배선만 담당합니다. 덕분에 fan-out 로직은 브로커나 소켓 없이 loopback으로 테스트할 수 있습니다.

Kafka에서 브라우저까지

전체 흐름은 한 방향입니다. market-worker가 시세를 저장하고 outbox로 이벤트를 발행하면, 게이트웨이 consumer가 그걸 받아 RealtimeHub로 넘기고, hub가 해당 아이템을 구독 중인 세션들에게만 밀어냅니다.

flowchart LR
  W[market-worker] -->|outbox 발행| K[(Kafka<br/>market.price.events)]
  K -->|eachMessage| C[realtime-gateway<br/>Kafka consumer]
  C --> R[routeMarketEvent]
  R --> H[RealtimeHub<br/>구독 레지스트리]
  H -->|itemId 구독 세션| S1[WebSocket sink]
  H -->|itemId 구독 세션| S2[SSE sink]
  S1 --> B1[브라우저 A]
  S2 --> B2[브라우저 B]

여기서 두 가지를 의식적으로 정했습니다.

첫째, 구독하는 토픽을 리터럴로 박지 않았습니다. 토픽 이름이 바뀌었는데 게이트웨이만 옛 문자열을 붙들고 있으면, 아무 에러 없이 그냥 아무것도 안 옵니다. 그래서 이벤트 계약(market-events)에서 토픽을 파생합니다.

// 토픽 이름을 리터럴로 박으면 rename 때 조용히 stale 구독이 된다.
// 이벤트 계약에서 파생해 그런 표류를 막는다.
const PRICE_TOPIC = TOPIC_BY_EVENT_TYPE['market.price.updated'];
const GATEWAY_TOPICS = [PRICE_TOPIC /* 대시보드 채널 신호 ... */];

둘째, push는 얇게 보냅니다. 이벤트 하나로 전체 시세를 다시 그리지 않습니다. price-updated 메시지는 행(row)을 그 자리에서 patch할 만큼의 요약만 싣고, 창(window) 기준 집계는 클라이언트가 REST로 재조정합니다. 그리고 push에는 version을 실어 보냅니다.

return {
  type: 'market.price.updated',
  itemId,
  observedAt,
  freshness,
  version: new Date(observedAt).getTime(), // 관측 epoch ms
  listingPriceSummary: listingSide(price),
  soldPriceSummary: soldSide(price),
};

version이 관측 시각의 epoch ms라는 점이 뒤에서 dedupe의 두 번째 축이 됩니다. freshness(FRESH/DELAYED/STALE/MAINTENANCE)는 (2)편에서 다룬 그 라벨을 그대로 재사용해, 실시간 화면에서도 데이터의 신선도를 숨기지 않습니다.

중복은 여기서 걸러진다

Kafka consumer는 at-least-once입니다. 리밸런싱, 재시작, 커밋 타이밍에 따라 같은 메시지가 두 번 이상 배달될 수 있습니다. 이걸 그대로 두면 화면이 같은 갱신을 두 번 깜빡이거나, 인기 신호가 부풀려집니다. 그래서 dedupe를 두 지점에 두었습니다.

(1) 게이트웨이에서 broker 재전달을 거릅니다. 키는 이벤트의 eventId입니다. idempotent() 래퍼가 이미 본 키면 handler를 건너뜁니다.

await consumer.run(
  idempotent(handler, (e) => parseEventId(e.value)),
  (error, at) => log(`skipping poison message at ${at.topic}@${at.offset}: ${error.message}`)
);
 
/** 수신 이벤트에서 dedup 키(eventId)를 뽑는다. */
export function parseEventId(value: unknown): string | undefined {
  if (typeof value === 'object' && value !== null && 'eventId' in value) {
    const id = (value as { eventId: unknown }).eventId;
    return typeof id === 'string' ? id : undefined;
  }
  return undefined;
}

idempotent() 자체는 도메인을 모릅니다. "키를 어떻게 뽑을지"를 호출부가 함수로 넘기고, 래퍼는 그 키가 이미 있으면 skip, 없으면 처리하고 기록합니다.

flowchart TD
  E[수신 이벤트] --> P["parseEventId(value.eventId)"]
  P --> Q{seen.has key?}
  Q -->|예| SK["skip — 재전달로 판단"]
  Q -->|아니오| HN["handler.handle 실행"]
  HN --> AD["seen.add key"]
  AD --> EV{size가 상한 초과?}
  EV -->|예| FIFO["가장 오래된 key 축출"]
  EV -->|아니오| DONE[유지]

처음엔 seen-store가 그냥 Set이었습니다. 코드 리뷰에서 바로 걸렸습니다. 오래 도는 consumer에서 Set은 지금껏 본 모든 이벤트를 하나씩 쌓아 무한히 커집니다. dedupe가 실제로 노리는 건 "시간상 가까이 붙은 재전달"이지 "한 달 전에 본 이벤트"가 아닙니다. 그래서 크기를 묶은 FIFO 축출 store로 바꿨습니다.

export function createBoundedSeenStore(maxKeys = 10_000): SeenStore {
  const keys = new Set<string>();
  return {
    has: (key) => keys.has(key),
    add: (key) => {
      keys.add(key);
      if (keys.size > maxKeys) {
        const oldest = keys.values().next().value; // 가장 오래된 키
        if (oldest !== undefined) keys.delete(oldest);
      }
    },
  };
}

Set의 삽입 순서 보장을 이용해 가장 오래된 키부터 버립니다. 프로덕션에서는 재시작 너머로도 유지되도록 durable store(Redis 등)를 주입할 수 있게 인터페이스만 열어 뒀습니다. 지금은 in-memory로도 "가까운 재전달"이라는 실제 위협은 충분히 잡습니다.

(2) 클라이언트에서 순서 역전을 거릅니다. dedupe 키만으로는 못 막는 게 하나 있습니다. 파티션 재전달로 오래된 관측이 최신 관측보다 늦게 도착하는 경우입니다. 이건 "중복"이 아니라 "순서 역전"이라 eventId dedupe가 통과시킵니다. 그래서 위에서 실은 version(관측 epoch ms)을 클라이언트가 비교해, 지금 화면의 행보다 오래된 version이면 무시합니다. 서버 dedupe는 "같은 걸 두 번", 클라이언트 version은 "오래된 걸 나중에" — 두 축이 서로 다른 실패를 막습니다.

poison 메시지를 죽이지 않고 치우기

가장 신경 쓴 부분입니다. Kafka consumer의 무서운 특성 하나는, handler가 throw하면 그 메시지를 커밋하지 않고 다시 시도한다는 점입니다. 정상적일 땐 재시도가 안전망이지만, 애초에 처리가 불가능한 메시지라면 재앙이 됩니다. 깨진 메시지 하나가 영원히 실패하며 파티션을 그 offset에 붙들어, 뒤에 쌓인 정상 이벤트가 전부 막힙니다. 이걸 poison 메시지라고 부릅니다.

핵심 판단은 이것이었습니다. 스트림은 진실의 원천이 아닙니다. 진실은 DB/outbox에 있고, Kafka는 그걸 알리는 전송 계층일 뿐입니다. 그러니 스트림에서 처리 못 할 메시지를 만나면, 붙들고 재시도할 게 아니라 skip하고 넘어가는 게 맞습니다. 잃는 건 "알림 한 번"이고, 실제 데이터는 DB에 그대로 있으며 클라이언트는 어차피 REST로 재조정하기 때문입니다.

poison은 두 층에서 나타나서, 두 곳에서 걸러 냅니다.

flowchart TD
  M[Kafka 메시지 bytes] --> J{JSON.parse 성공?}
  J -->|아니오| PO1["onPoison 호출<br/>offset 전진 · skip"]
  J -->|예| SC{"parseMarketEvent<br/>스키마 통과?"}
  SC -->|아니오| PO2["로그 후 skip<br/>partition 안 막음"]
  SC -->|예| OK[hub fan-out]
  PO1 -. 만약 throw했다면 .-> BAD["무한 재시도 → partition wedge"]
  PO2 -. 만약 throw했다면 .-> BAD

층 1 — transport 파싱. 바이트가 유효한 JSON이 아니면 여기서 끝납니다. JSON.parse를 감싸 onPoison 콜백으로 넘기고 return합니다. offset은 전진하므로 파티션이 막히지 않습니다.

eachMessage: async ({ topic, partition, message }) => {
  const raw = message.value?.toString('utf-8');
  let value: unknown;
  try {
    value = raw === undefined ? undefined : JSON.parse(raw);
  } catch (error: unknown) {
    // 비-JSON payload는 복구 불가 — 파티션을 막지 말고 skip.
    // 진실의 원천은 스트림이 아니라 DB/outbox다.
    onPoison(error instanceof Error ? error : new Error(String(error)), {
      topic, partition, offset: message.offset,
    });
    return;
  }
  await handler.handle({ /* ... */ value });
},

층 2 — 도메인 스키마. JSON으로는 멀쩡한데 이벤트 계약과 안 맞는 경우입니다. parseMarketEvent가 throw하면 라우터가 잡아서 로그만 남기고 skip합니다.

let event;
try {
  event = parseMarketEvent(received.value);
} catch (error: unknown) {
  // poison 이벤트가 파티션을 막게 두지 않는다(kafkajs는 영원히 재시도한다).
  // DB/outbox가 진실의 원천이므로 skip하고 넘어간다.
  log(`skipping unparseable event at ${received.topic}@${received.offset}: ...`);
  return;
}

여기에 코드 리뷰에서 나온 세 번째 poison이 더해졌습니다. 반쯤 죽은 소켓에 쓰다 난 예외입니다. fan-out 루프가 여러 세션에 순서대로 push하는데, 중간의 죽은 소켓에 send하다 예외가 나면 그게 루프를 타고 올라가 Kafka handler까지 전파돼, 결국 정상 메시지를 poison처럼 실패시킵니다. 그래서 sink 쓰기를 감싸, 죽은 transport는 예외를 밖으로 던지는 대신 그 세션만 조용히 버리게 했습니다.

sinks.set(sessionId, (message) => {
  if (socket.readyState !== socket.OPEN) return;
  try {
    socket.send(JSON.stringify(message));
  } catch {
    // 반쯤 죽은 소켓에 쓰다 난 예외가 fan-out 루프(그리고 Kafka handler)로
    // 전파되면 안 된다. 세션을 버린다.
    dropSession(sessionId);
  }
});

세 경우 모두 원칙은 하나입니다. 처리 불가능한 하나가 처리 가능한 전부를 막게 두지 않는다. 그리고 이건 (2)편에서 세운 구분과 이어집니다. 거기서 "점검으로 실패한 메시지는 DLQ로 보내지 않는다"고 했는데, 반대로 여기서는 "정말 처리 불가능한 poison은 붙들지 말고 치운다"입니다. 잠깐 안 되는 것과 영영 안 되는 것을 다르게 다루는, 같은 위생 감각입니다.

한 가지는 솔직히 남겨 뒀습니다. 지금 poison은 skip하며 로그만 남깁니다. "왜 poison이었는지"를 나중에 파헤치려면 격리 저장이 필요할 텐데(payload hash, offset, 축약 표본), 이건 발생 빈도를 보고 붙일 여지로 뒀습니다. DB가 진실이라 급하지 않았습니다.

번들과 런타임에서 부딪힌 것

로직을 다 짜고 로컬 end-to-end 검증(Kafka 이벤트 → 게이트웨이 → WebSocket 클라이언트)을 돌리는데, 정작 발을 잡은 건 dedupe도 poison도 아니었습니다. 런타임 두 가지였습니다.

첫째, require is not defined. 게이트웨이 앱의 package.json은 "type": "module"입니다. NodeNext가 ESM-only인 @dunplay/* 소스 라이브러리를 typecheck 시점에 제대로 해석하게 하려면 필요합니다. 그런데 webpack이 만드는 번들은 CommonJS(externals를 require()로 부름)입니다. 그러니 Node가 dist/main.js를 앱의 ESM 타입으로 로드하면서 require가 없다며 죽었습니다. 소스는 ESM으로 두되, 빌드 산출물 디렉터리만 CommonJS로 표시해 풀었습니다.

// output.clean이 매 빌드마다 dist를 지우므로, emit 직후 marker를 다시 쓴다.
compiler.hooks.afterEmit.tap('EmitCommonJsDistMarker', () => {
  writeFileSync(join(compiler.outputPath, 'package.json'), '{ "type": "commonjs" }\n');
});

둘째, 없는 토픽을 구독하다 죽는 startup 순서 함정. 갓 띄운 로컬 브로커에서는 아직 아무 producer도 토픽을 만들지 않았을 수 있습니다. 그 상태로 게이트웨이가 먼저 떠서 구독하면 UNKNOWN_TOPIC_OR_PARTITION으로 죽습니다. "누가 먼저 뜨느냐"에 게이트웨이 생존이 걸리는 건 취약합니다. 그래서 구독 전에 토픽을 멱등하게 보장합니다.

const created = await admin.createTopics({
  topics: [{ topic, numPartitions: 1, replicationFactor: 1 }],
  waitForLeaders: true,
});
log(created ? `created topic ${topic}` : `topic ${topic} already exists`);

createTopics는 이미 있으면 no-op이라 멱등하고, waitForLeaders는 "방금 만들었지만 아직 리더가 없는" 경쟁 상태를 피해 줍니다. 게이트웨이가 producer보다 먼저 떠도 죽지 않습니다.

이 둘은 로직의 우아함과는 무관합니다. 하지만 실시간 시스템에서 "재시작할 때마다 순서 운에 따라 죽는 프로세스"는 그 자체로 장애입니다. dedupe와 poison을 아무리 정교하게 짜도, 프로세스가 뜨질 못하면 소용이 없습니다. 이런 것들이 로컬 검증을 실제로 통과시키는 마지막 관문이었습니다.

정리하면

이 편의 전제는 하나였습니다. 실시간의 어려움은 "빠르게 보내기"가 아니라 "중복과 실패를 눈에 안 띄게 처리하기"에 있다.

게이트웨이를 따로 뒀습니다. DB write도 provider 호출도 없이, 저장된 fact를 fan-out하는 transport 배선만 담당합니다. push는 조회가 아니라 알림이고, 클라이언트는 REST로 재조정합니다.
중복은 두 축으로 거릅니다. consumer의 eventId dedupe가 "같은 걸 두 번"을, 클라이언트의 version 비교가 "오래된 걸 나중에"를 막습니다. dedupe 저장소는 FIFO로 묶어 메모리를 무한히 키우지 않습니다.
poison은 죽이지 않고 치웁니다. transport 파싱·도메인 스키마·죽은 소켓 쓰기 세 층에서, 처리 불가능한 하나가 처리 가능한 전부를 막지 않게 skip합니다. 스트림이 아니라 DB가 진실이기 때문입니다.
정작 마지막 관문은 로직이 아니라 런타임(ESM/CJS 번들, 토픽 보장)이었습니다.

빠른 시스템은 잘 만든 티가 나지만, 잘 만든 실시간 시스템은 아무 티도 안 나는 게 정상입니다. 사용자는 화면이 조용히 갱신되는 것만 봅니다. 그 뒤에서 중복이 걸러지고 깨진 메시지가 치워지고 죽은 소켓이 정리되는 걸 눈치채지 못한다면, 그게 이 게이트웨이가 제 일을 한 것입니다.

다음 편에서는 이렇게 늘어난 앱들 — 수집 워커, API, BFF, 그리고 이 게이트웨이까지 — 을 Docker 전체 스택 하나로 묶어 윈도우·맥 어디서든 프로덕션과 같은 위상으로 띄운 이야기를 다루겠습니다.

카테고리

사이드 프로젝트

자주쓰는 태그