Correlation ID 전파

📋 개요

이 문서는 HTTP 요청부터 비즈니스 로직을 거쳐 외부 시스템(GCP Pub/Sub)까지 x-correlation-id를 일관되게 전파하는 구현 내용을 설명합니다.

목표

HTTP 요청에서 시작된 correlation-id를 애플리케이션 전체 생명주기 동안 추적
명시적인 파라미터 전달 없이 비동기 호출 체인 전체에 컨텍스트 유지
분산 시스템 간 요청 추적 및 디버깅 지원
로그 집계 및 성능 모니터링 향상

범위

HTTP 요청
  → HttpRequestLoggingInterceptor (correlation-id 추출)
  → 비즈니스 로직 (Service/Handler)
  → CrossEventPublisher.publishXXX()
  → EventEmitter2.emit('event', standardEvent)
  → EventBridgeService @OnEvent('event')
  → GCP Pub/Sub.publish()
  → 다른 도메인 수신

🎯 현재 문제점

Before (구현 전)

1. HTTP Request
   └─ x-correlation-id: abc-123 ✅

2. HttpRequestLoggingInterceptor
   └─ correlationId 추출 및 로깅 ✅
   └─ 응답 헤더 설정 ✅
   └─ 하지만 컨텍스트 저장 없음 ❌

3. Controller → Service → EventPublisher
   └─ correlationId가 전달되지 않음 ❌

4. StandardEvent 발행
   └─ correlationId 누락 ❌

5. GCP Pub/Sub
   └─ 추적 정보 없음 ❌
   └─ 다음 도메인에서 원본 요청 추적 불가 ❌

영향

분산 시스템에서 요청 추적 불가
크로스 도메인 이벤트의 원인 파악 어려움
디버깅 시 로그 연관성 파악 어려움
성능 병목 지점 분석 어려움

🏗️ 설계 결정사항

1. AsyncLocalStorage 기반 CLS (Continuation Local Storage) 패턴 채택

선택 이유:

✅ 명시적 파라미터 전달 없이 비동기 호출 체인 전체에 컨텍스트 유지
✅ NestJS DI와 자연스럽게 통합
✅ 테스트하기 쉬움
✅ Node.js 14+ 공식 지원 (async_hooks 모듈)
✅ 검증된 패턴 (Express, NestJS REQUEST scope 내부에서 사용)

대안 비교:

방법	장점	단점	채택 여부
AsyncLocalStorage	자동 컨텍스트 전파, 깔끔한 코드	약간의 성능 오버헤드 (~0.1ms)	✅ 채택
명시적 파라미터 전달	명확한 의존성	모든 함수 시그니처 수정 필요	❌
ThreadLocal (Java 스타일)	익숙한 패턴	Node.js는 싱글 스레드, 적합하지 않음	❌
cls-hooked	레거시 솔루션	유지보수 중단, 버그 많음	❌

2. StandardEvent 구조 확장

export interface StandardEvent<T = any> {
  eventId: string;
  type: string;
  timestamp: string;
  source: string;
  payload: T;
  version: string;
  correlationId?: string; // 🆕 추가
}

설계 원칙:

correlationId는 optional 필드 → 하위 호환성 유지
최상위 레벨 필드 → 표준화된 위치
타입 안전성 유지

3. BaseCrossEventPublisher 도입

모든 CrossEventPublisher의 공통 로직 추출
중복 코드 제거
일관된 correlation-id 처리

🔬 AsyncLocalStorage 동작 원리

핵심 개념

AsyncLocalStorage는 "전역 저장소"가 아니라 **"비동기 실행 컨텍스트별 저장소"**입니다.

오해 vs 실제

❌ 잘못된 이해: "전역 저장소에 여러 요청의 데이터가 섞인다"

// 이렇게 동작하는 것이 아닙니다!
globalStorage = {
  'request-1': { correlationId: 'abc-123' },
  'request-2': { correlationId: 'def-456' },
  'request-3': { correlationId: 'ghi-789' },
};
// → 어떤 걸 꺼내야 하지? (혼선 발생) ❌

✅ 실제 동작: "각 비동기 실행 컨텍스트가 독립적인 저장소를 가짐"

// 요청 1의 실행 컨텍스트
[Request 1 Context]
  └─ AsyncLocalStorage Store: { correlationId: 'abc-123' }
     └─ Controller
        └─ Service
           └─ EventPublisher
              └─ getCorrelationId() → 'abc-123' ✅

// 요청 2의 실행 컨텍스트 (완전히 분리됨)
[Request 2 Context]
  └─ AsyncLocalStorage Store: { correlationId: 'def-456' }
     └─ Controller
        └─ Service
           └─ EventPublisher
              └─ getCorrelationId() → 'def-456' ✅

동작 증명 코드

import { AsyncLocalStorage } from 'async_hooks';

const asyncLocalStorage = new AsyncLocalStorage<{ id: string }>();

// 3개의 요청을 동시에 처리
async function simulateRequest(requestId: string, delay: number) {
  return asyncLocalStorage.run({ id: requestId }, async () => {
    console.log(`[${requestId}] Request started`);

    await new Promise((resolve) => setTimeout(resolve, delay));
    await nestedFunction(requestId);

    console.log(`[${requestId}] Request completed`);
  });
}

async function nestedFunction(expectedId: string) {
  const context = asyncLocalStorage.getStore();
  console.log(`[${expectedId}] context.id = ${context?.id}`);

  // 항상 올바른 컨텍스트를 가져옴!
  if (context?.id !== expectedId) {
    throw new Error('Context mismatch!'); // 절대 발생하지 않음
  }
}

// 동시 실행
Promise.all([simulateRequest('REQUEST-1', 200), simulateRequest('REQUEST-2', 100), simulateRequest('REQUEST-3', 150)]);

/* 출력:
[REQUEST-1] Request started
[REQUEST-2] Request started
[REQUEST-3] Request started
[REQUEST-2] context.id = REQUEST-2  ✅
[REQUEST-3] context.id = REQUEST-3  ✅
[REQUEST-1] context.id = REQUEST-1  ✅
→ 모든 요청이 올바른 컨텍스트 유지!
*/

내부 동작 원리

// AsyncLocalStorage 내부 (단순화된 버전)
class AsyncLocalStorage<T> {
  private stores = new Map<number, T>();

  run<R>(store: T, callback: () => R): R {
    // 1. 새로운 실행 컨텍스트 ID 생성 (async_hooks가 자동 관리)
    const executionId = createNewExecutionContext();

    // 2. 이 실행 컨텍스트에 store 저장
    this.stores.set(executionId, store);

    // 3. 콜백 실행 (이 안의 모든 비동기 작업도 같은 executionId 유지)
    try {
      return callback();
    } finally {
      this.stores.delete(executionId);
    }
  }

  getStore(): T | undefined {
    // 현재 실행 컨텍스트의 ID를 자동으로 파악하여 해당 store 반환
    const currentExecutionId = getCurrentExecutionId(); // Node.js가 자동 추적
    return this.stores.get(currentExecutionId);
  }
}

핵심:

Node.js의 async_hooks가 각 비동기 작업에 고유한 실행 컨텍스트 ID를 자동 할당
await, Promise, setTimeout 등을 거쳐도 ID 유지
getStore() 호출 시 현재 실행 중인 컨텍스트를 자동 파악

⚠️ 컨텍스트 유실 원인과 해결

왜 `run()`이 RxJS/Nest 파이프라인에서 유실될 수 있는가

AsyncLocalStorage.run(store, callback)은 “callback이 시작되면서 생성된 비동기 흐름”에만 store를 전파합니다.
NestJS는 인터셉터 → 컨트롤러 → 파이프/가드 → 응답으로 이어지는 과정에서 내부적으로 새로운 Promise 체인과 마이크로태스크를 생성할 수 있습니다.
defer(() => run(() => next.handle()))처럼 구독 시점에 run으로 감싼 패턴은, 그 이전/외부에서 이미 만들어진 비동기 체인까지 포섭하지 못해 컨텍스트가 enabled: false로 보이거나 getStore()가 undefined를 반환할 수 있습니다.

문제가 되는 패턴:

return defer(() =>
  requestContextService.run(ctx, () => {
    return next.handle().pipe(/* ... */);
  })
);

유실 방지 패턴:

인터셉터 초반에 enterWith(ctx)로 “현재 실행 컨텍스트”에 저장소를 주입합니다. 그러면 이후 생성되는 모든 비동기 작업(컨트롤러 async/await, 내부 Promise, RxJS 체인 등)으로 전파됩니다.

requestContextService.enterWith(ctx);
return next.handle().pipe(/* ... */);

Request Body/DTO로 `correlationId` 전달 비권장

클라이언트 조작 위험, 모든 DTO/문서 확산, 내부 호출/이벤트 경로 누락 등으로 일관성이 깨집니다.
표준은 헤더(x-correlation-id) + 서버 측 컨텍스트(ALS)입니다.
필요 시 접근 편의를 위해 req.correlationId 같은 읽기용 속성만 인터셉터에서 복사해 두고, 비즈니스 로직은 RequestContextService.getCorrelationId()로 조회하세요.

📐 구현 아키텍처

After (구현 후) 데이터 흐름

1. HTTP Request
   └─ x-correlation-id: abc-123

2. HttpRequestLoggingInterceptor
   └─ RequestContext { correlationId: 'abc-123' }
   └─ AsyncLocalStorage.enterWith(context) 🆕

3. Controller → Service → EventPublisher
   └─ AsyncLocalStorage에서 자동으로 컨텍스트 유지 🆕

4. CrossEventPublisher.publishXXX()
   └─ requestContextService.getCorrelationId() → 'abc-123' 🆕
   └─ StandardEvent { correlationId: 'abc-123', ... } 🆕

5. EventBridgeService
   └─ Pub/Sub.publish({
        data: event,
        attributes: { correlationId: 'abc-123' } 🆕
      })

6. /api/events (Pub/Sub Push)
   └─ x-correlation-id: abc-123 (헤더) 🆕
   └─ 다음 도메인으로 전파 🆕

컴포넌트 다이어그램

┌─────────────────────────────────────────────────────────────┐
│                      HTTP Request Layer                      │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ HttpRequestLoggingInterceptor                        │  │
│  │  - Extract correlationId from headers               │  │
│  │  - Create RequestContext                            │  │
│  │  - Inject with AsyncLocalStorage.enterWith()       │  │
│  └────────────────┬─────────────────────────────────────┘  │
└───────────────────┼─────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────┐
│                   Request Context Layer                      │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ RequestContextService                                │  │
│  │  - AsyncLocalStorage<RequestContext>                │  │
│  │  - getCorrelationId(): string                       │  │
│  │  - getRequestId(): string                           │  │
│  │  - run<T>(context, callback): T                     │  │
│  └────────────────┬─────────────────────────────────────┘  │
└───────────────────┼─────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────┐
│                   Business Logic Layer                       │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ Controllers → Services → Handlers                    │  │
│  │  (자동으로 컨텍스트 유지, 수정 불필요)                │  │
│  └────────────────┬─────────────────────────────────────┘  │
└───────────────────┼─────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────┐
│                    Event Publishing Layer                    │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ BaseCrossEventPublisher (Abstract)                  │  │
│  │  - Inject RequestContextService                     │  │
│  │  - createStandardEvent() → includes correlationId   │  │
│  └────────────────┬─────────────────────────────────────┘  │
│                   │                                          │
│  ┌────────────────┴─────────────────────────────────────┐  │
│  │ Concrete Publishers (extends Base)                   │  │
│  │  - UserCrossEventPublisher                          │  │
│  │  - SleepCrossEventPublisher                         │  │
│  │  - MobileCrossEventPublisher                        │  │
│  │  - ... (10+ publishers)                             │  │
│  └────────────────┬─────────────────────────────────────┘  │
└───────────────────┼─────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────┐
│                   Event Bridge Layer                         │
│  ┌──────────────────────────────────────────────────────┐  │
│  │ EventBridgeService                                   │  │
│  │  - Listen to 'event' channel                        │  │
│  │  - Publish to GCP Pub/Sub with correlationId       │  │
│  └────────────────┬─────────────────────────────────────┘  │
└───────────────────┼─────────────────────────────────────────┘
                    │
                    ▼
           GCP Pub/Sub (events topic)
                    │
                    ▼
         Other Domains (continue tracing)

🔧 구현 요약

전파 메커니즘

AsyncLocalStorage 기반 자동 전파:

HTTP Request
  → Interceptor에서 AsyncLocalStorage.enterWith(context) 주입
  → 비즈니스 로직 전체에서 자동으로 컨텍스트 유지
  → Publisher에서 getCorrelationId()로 조회 후 이벤트에 포함
  → Pub/Sub attributes로 전파
  → 다음 도메인에서 수신 및 계속 추적

구현 단계

Phase 1: RequestContextService 생성

@Injectable()
export class RequestContextService {
  private readonly asyncLocalStorage = new AsyncLocalStorage<RequestContext>();

  enterWith(context: RequestContext): void { ... }
  getCorrelationId(): string | undefined { ... }
}

Phase 2: HTTP Interceptor 통합

// Interceptor에서 컨텍스트 주입
this.requestContextService.enterWith({
  requestId: 'req-123',
  correlationId: 'cor-abc', // 헤더에서 추출
  timestamp: new Date(),
});

Phase 3: StandardEvent 확장

export interface StandardEvent<T = any> {
  // ... 기존 필드들
  correlationId?: string; // 🆕 추가
}

Phase 4: BaseCrossEventPublisher

protected createStandardEvent<T>(type: string, payload: T): StandardEvent<T> {
  return {
    eventId: uuidv4(),
    type,
    payload,
    correlationId: this.requestContext.getCorrelationId(), // 🆕 자동 조회
  };
}

Phase 5: Publisher 마이그레이션

@Injectable()
export class UserCrossEventPublisher extends BaseCrossEventPublisher {
  async publishUserCreated(userId: string) {
    // createStandardEvent가 자동으로 correlationId 포함
    const event = this.createStandardEvent('user.created', { userId });
    await this.publishCrossEvent(event);
  }
}

Phase 6: Pub/Sub 통합

// 발행 측
await topic.publishMessage({
  data: Buffer.from(JSON.stringify(event)),
  attributes: { correlationId: event.correlationId }, // 🆕 attributes에 포함
});

// 수신 측
const correlationId = pubSubMessage.message.attributes?.correlationId;

📋 개요​

목표​

범위​

🎯 현재 문제점​

Before (구현 전)​

영향​

🏗️ 설계 결정사항​

1. AsyncLocalStorage 기반 CLS (Continuation Local Storage) 패턴 채택​

2. StandardEvent 구조 확장​

3. BaseCrossEventPublisher 도입​

🔬 AsyncLocalStorage 동작 원리​

핵심 개념​

오해 vs 실제​

❌ 잘못된 이해: "전역 저장소에 여러 요청의 데이터가 섞인다"​

✅ 실제 동작: "각 비동기 실행 컨텍스트가 독립적인 저장소를 가짐"​

동작 증명 코드​

내부 동작 원리​

⚠️ 컨텍스트 유실 원인과 해결​

왜 run()이 RxJS/Nest 파이프라인에서 유실될 수 있는가​

Request Body/DTO로 correlationId 전달 비권장​

📐 구현 아키텍처​

After (구현 후) 데이터 흐름​

컴포넌트 다이어그램​

🔧 구현 요약​

전파 메커니즘​

구현 단계​

📚 참고자료​