실시간 데이터를 수집하는 Node.js Worker의 안정성과 구조 개선하기

Node.jsInfluxDBWebSocket
백엔드2026년 8월 19일

들어가며

약 1년 전에 Node.js로 만든 Worker Application으로 시계열 데이터 저장 & 실시간 스트리밍하기라는 글을 작성한 적이 있다. 가상화폐 거래소의 WebSocket에서 티커 데이터를 받아 Redis Pub/Sub으로 실시간 스트리밍하고, 확정된 봉을 InfluxDB에 적재하는 Worker를 구현한 과정을 정리한 글이었다.

그 글은 다음과 같은 개선 사항을 남기면서 끝났다.

  • Worker Health Check 기능 (via Redis Hash)
  • WebSocket 연결 해제 시 retry 로직 (Exponential Backoff)
  • Sentry 연동
  • Containerize & Auto-scaling

그 사이 프로젝트가 폴리레포에서 모노레포로 옮겨가면서 전체 코드베이스를 다시 작성했고, Worker도 그 과정에서 통째로 새로 쓰였다. 결과적으로 위 네 항목 중 세 개는 구현되었고, 하나는 의도적으로 하지 않았다. 그리고 당시 글에서 이유까지 붙여 설명했던 설계 하나는, 다시 보니 애초에 필요하지 않았다는 결론이 나서 걷어내게 되었다.

이 글에서는 그 항목들을 하나씩 어떻게 처리했는지, 그리고 걷어낸 것은 왜 걷어냈는지를 정리해보고자 한다.

재연결을 소켓 자체의 책임으로 넘기기

이전 구현에서 재연결은 추상 Worker 클래스가 직접 들고 있었다. onClose에서 invalidate()로 상태를 정리하고 reconnect()를 호출하는 구조였는데, 문제는 이 클래스가 이미 너무 많은 일을 하고 있었다는 점이다. WebSocket lifecycle, 거래소마다 다른 구독·파싱 처리, Redis 발행, InfluxDB 적재, keepalive가 한 클래스에 다 들어 있었다.

그래서 이번에는 WebSocket lifecycle만 담당하는 StreamSocket을 별도 패키지로 분리하고, 거래소마다 달라지는 부분은 훅으로 넘겨받도록 했다.

export class StreamSocket {
    private ws: WebSocket | null = null;
    private attempts = 0;

    private reconnect(): void {
        if (this.attempts >= this.maxAttempts) {
            console.error(`[${this.label}] reconnect attempts exhausted`);
            this.options.reconnect?.onExhausted?.();
            return;
        }
        const delay = Math.min(this.baseMs * 2 ** this.attempts, this.maxMs);
        this.attempts += 1;
        console.warn(`[${this.label}] reconnecting in ${delay}ms (attempt ${this.attempts})`);
        this.reconnectTimer = setTimeout(() => this.connect(), delay);
    }
}

남겨둔 항목 중 하나였던 Exponential Backoff는 위와 같이 해결했다. 기본값은 1초에서 시작해 매 시도마다 두 배씩 늘리고, 상한은 60초, 최대 10회다. 재시도 횟수를 전부 소진하면 onExhausted를 호출하는데, 해당 콜백이 호출되면 process.exit(1)을 통해 프로세스를 종료한다. 거래소에 아예 닿지 않는 상황이라면 프로세스가 조용히 죽은 채로 남아 있는 것보다 컨테이너 재시작에 맡기는 편이 낫다고 판단했다.

다만 구현하면서 예상하지 못했던 함정이 두 개 있었다.

첫 번째는 이미 지나간 소켓이 늦게 이벤트를 뱉는 경우이다. 재연결로 새 소켓을 만든 뒤에 이전 소켓의 close 이벤트가 도착하면, 그것이 또 재연결을 트리거해서 연결이 두 배로 늘어난다. 그래서 모든 핸들러가 자기 자신이 현재 소켓인지 먼저 확인하도록 했다.

ws.on("close", () => {
    if (this.ws !== ws) return;
    this.clearPing();
    this.options.onClose?.();
    if (!this.closed) this.reconnect();
});

두 번째 문제는 종료 처리에서 나왔다. close()는 재연결을 멈추고 소켓을 버리는 동작인데, 이때 this.ws를 먼저 null로 만든다. 그러면 뒤이어 도착하는 close·error 이벤트가 전부 위 가드(this.ws !== ws)에 걸려 조용히 무시된다. 재연결도, onClose 콜백도 돌지 않는다.

문제는 여기서 리스너까지 떼면 안 된다는 것이다. 연결 수립 중(CONNECTING)인 소켓을 닫으면 error 이벤트가 비동기로 발생하는데, 이 이벤트를 받을 리스너가 하나도 없으면 ws는 그것을 예외로 던져 프로세스를 죽인다. 그러므로 참조만 버리고 리스너는 그대로 둔 채 커넥션을 닫도록 구성했다.

close(): void {
    this.closed = true;
    this.clearTimers();

    // 참조를 먼저 끊어 reconnect 액션은 무시하되, 리스너는 그대로 부착해둔다
    const ws = this.ws;
    this.ws = null;
    ws?.close();
}

실제로 연결에 실패하도록 만들어 확인해보면, 다음과 같이 재시도 간격이 두 배씩 벌어지는 것을 볼 수 있다.

worker-exp-backoff

Half-open 연결 감지하기

위 재연결 로직을 붙이고 나서도 피드가 멈추는 일이 있었다. 원인은 half-open 연결이었다.

소켓의 readyState는 OPEN이고, keepalive ping에 대한 pong도 정상적으로 돌아온다. 그런데 캔들 데이터만 오지 않는다. 이 상태에서는 close도 error도 발생하지 않으므로 앞서 만든 재연결은 아예 발동하지 않는다. 연결은 살아 있고 데이터만 죽어 있는 것이다.

결국 연결 상태가 아니라 데이터 도착 시각을 봐야 한다는 결론에 도달했다. 실제 캔들을 받을 때마다 그 시각을 기록해두고, 주기적으로 마지막 수신 이후 얼마나 지났는지 검사하도록 처리했다.

// 실제 캔들이 도착한 마지막 시각, 연결된 소켓이 임계값보다 조용하면
// half-open으로 보고 강제 재연결시키며, 상태도 down으로 보고한다
private lastMessageAt = Date.now();

// 서브클래스에서 실제 캔들 tick마다 호출 - pong 같은 control frame은 세지 않는다
protected markFeedAlive(): void {
    this.lastMessageAt = Date.now();
}

private tick(): void {
    const fresh = Date.now() - this.lastMessageAt < WORKER_FEED_STALE_MS;
    if (this.connected && !fresh) {
        console.warn(`[worker] feed silent for ${WORKER_FEED_STALE_MS / 1000}s — forcing reconnect`);
        this.socket?.refresh();
        this.lastMessageAt = Date.now();
    }
    void this.reportStatus(this.connected && fresh);
}

여기서 중요한 점은 markFeedAlive()를 control frame에서 호출하지 않는 것이다. pong을 받았다고 피드가 살아있다고 판단하면 이 검사가 무력화되는데, half-open 상태에서도 pong은 정상적으로 오기 때문이다.

강제 재연결은 close()가 아니라 terminate()로 한다. half-open 소켓에 정상 종료 핸드셰이크를 시도하면 응답을 기다리며 그대로 멈춰 있을 수 있기 때문이다.

refresh(): void {
    if (this.closed || !this.ws) return;
    this.ws.terminate();
}

피드가 멈춘 상황을 만들어 확인해보면, 아래와 같이 워커가 이를 감지하고 스스로 재연결하는 것을 알 수 있다.

worker-half-open

Health Check에 무엇을 담을 것인가

기존에 남겨둔 개선 항목에 "Worker Health Check (via Redis Hash)"라고 적어두었으니 Redis Hash에 상태를 쓰는 것 자체는 어렵지 않았다. 실제로 고민이 된 부분은 거기에 무엇을 쓸 것인가였다.

처음에는 소켓 연결 여부를 쓰려고 했지만, 앞서 본 half-open 때문에 그것만으로는 의미가 없었다. 그래서 status를 connected && fresh, 즉 연결되어 있으면서 WORKER_FEED_STALE_MS 내에 캔들을 받았는지로 정의했다. 소켓만 살아 있고 데이터가 끊긴 워커는 대시보드에서 Down으로 드러나게 된다.

private async reportStatus(up: boolean): Promise<void> {
    try {
        await this.redis.hSet(this.workerKey, {
            exchangeId: this.config.exchangeId,
            status: String(up),
            startedAt: this.startedAt,
            updatedAt: Date.now(),
        });
        await this.redis.expire(this.workerKey, WORKER_KEY_TTL_SECONDS);
    } catch (err) {
        console.error(`[worker] status heartbeat error: ${err}`);
    }
}

매번 TTL을 다시 밀어주는 이유는 워커를 죽였을 때 아무도 갱신하지 않는 키가 대시보드에 영구히 남는 것을 막기 위함이다.

이제 관리자 화면을 확인해보면 아래와 같이 해당 워커가 Down으로 표시되는 것을 볼 수 있다.

worker-status-dashboard

Redis를 경유하던 저장 경로를 걷어내기

이전 글에서는 WebSocket으로 받은 데이터를 Redis Pub/Sub으로 발행하고, 별도의 프로세서가 그 채널을 구독해서 InfluxDB에 적재했다. 그리고 굳이 Redis를 경유한 이유를 두 가지로 설명했다.

  • 거래소마다 데이터 형식이 다르기 때문에 이미 전처리된 데이터를 DB에 바로 삽입하기 위함
  • 각 거래소 Worker마다 코드를 작성하면 불필요한 코드 중복이 생겨 공통 로직으로 처리하기 위함

지금 돌이켜 생각해보면 두 이유 모두 Redis가 필요한 이유가 아니었다. 형식 정규화와 코드 중복 제거는 이미 거래소별 구현체가 자기 거래소의 형식을 공통 형태로 변환해 넘기기 때문에 자체적으로 해결하고 있었다. 그렇기에 여기서 Pub/Sub을 한 번 왕복하는 것은 정규화에 아무것도 더하지 않고, 대신 프로세스 하나와 구독 상태 하나를 추가로 관리하게 만들었을 뿐이었다.

따라서 지금은 확정된 캔들을 받은 자리에서 곧바로 적재하도록 수정했다.

const channel = REDIS_CHANNELS.TICKER(this.config.exchangeId, symbol);
this.redis.publish(channel, JSON.stringify(candle)).catch((err) => {
    console.error(`[worker] Redis publish failed: ${err}`);
});
this.publishLastPrice(symbol, candle.close);

if (kline.confirm) {
    const point = new Point("candles")
        .tag("exchangeId", this.config.exchangeId)
        .tag("symbol", symbol)
        .floatField("open", candle.open)
        // ...
        .timestamp(kline.start);
    this.writeApi.writePoint(point);
}

집계를 InfluxDB에서 워커로 가져오기

캔들 저장은 계속 1분봉으로만 하고, 조회 시점에 요청한 타임프레임으로 다운샘플하는 전략은 그대로 유지했다. 다만 그 집계를 어디서 수행하는지를 변경했다.

처음에는 Flux의 aggregateWindow로 OHLCV 다섯 필드를 각각 집계한 뒤 union으로 합치는 쿼리를 사용했다. 짧은 범위에서는 잘 동작했지만, 긴 범위를 요청하면 InfluxDB가 시리즈 전체를 메모리에 올려둔 채로 있다가 OOM으로 죽는 상황이 발생했다.

따라서 집계를 InfluxDB에서 수행하는 대신에 raw 1분봉을 스트리밍으로 읽어 프로세스 안에서 직접 타임프레임 단위로 묶도록 수정했다. 그렇게 하면 InfluxDB는 저장된 행을 순서대로 흘려보내기만 하면 되므로, 쿼리 한 건이 붙잡는 메모리가 요청 범위와 무관하게 일정해진다. 범위가 길어질수록 늘어나는 것은 오가는 행의 수일 뿐이고, 워커는 받은 행을 그때그때 해당 버킷에 누적한 뒤 버리면 된다. 버킷을 나누는 기준은 기존 aggregateWindow가 쓰던 것과 같게 맞췄다. 집계 위치만 옮기고 결과는 그대로 두기 위해서다.

Sentry와 컨테이너화

Worker는 요청을 받아 응답하는 서버가 아니라 계속 떠 있는 프로세스라, 잡히지 않은 예외가 곧 장애로 이어진다. 그런데 Sentry 기본 통합이 uncaughtException과 unhandledRejection을 자동으로 캡처하고 flush까지 처리해주기 때문에, 대부분은 SDK를 연동하는 것만으로 구현할 수 있었다. 직접 캡처를 넣은 곳은 예외가 전역 핸들러까지 올라오지 않는 지점과 startup 실패뿐이다. 다만 tracesSampleRate는 0으로 두었다. 당장 성능 추적은 필요하지 않다고 판단했기 때문이다.

컨테이너화는 모노레포로 옮기면서 자연히 해결됐다. CI가 앱별로 이미지를 빌드해 GHCR에 올리고, 각 호스트가 그 이미지를 받아 구동하도록 처리했다.

마무리

1년 전에 남긴 네 항목 중 세 개는 이렇게 정리되었다. 마지막 Auto-scaling의 경우 결론부터 말하자면 진행하지 않았다.

물론 구조적으로는 가능하다. 어느 거래소를 맡을지는 환경변수로 정하고, 워커 식별자도 컨테이너마다 배정하기에 같은 거래소에 여러 프로세스를 띄울 수 있다. 백필과 워밍업 요청은 여러 워커가 같은 큐를 blPop으로 나눠 가져가므로, 몇 개가 붙어도 작업이 중복되지 않는다. 그런데 실제로 스케일을 조정해야 할 만큼 부하가 커진 적이 없었다. 따라서 이 부분은 필요해질 때 다시 고려해보기로 했다.

댓글