Queue 배치 수집 미적용과 부분 실패 처리 정책 개선 및 테스트 결과

기존 구조에서 확인한 문제점

시나리오 설정

기존 슈도 코드

async function consumeQueueBatch(batch) {
  const preparation = await prepareAnalyticsBatch(
    batch.messages.map(message => message.body)
  );

  try {
    await storage.insertCanonicalEvents(preparation.rowsToInsert);
  } catch (error) {
    for (const item of preparation.accepted) {
      item.message.retry();
    }

    return;
  }

  await materialize(preparation.affectedSiteIds);

  for (const item of preparation.accepted) {
    item.message.ack();
  }
}

문제점

100개 중 1개만 INSERT 오류를 일으켜도 ClickHouse 요청 전체가 실패하기 때문에, 기존 구현은 유효한 100개 메시지를 모두 retry했습니다.

즉, 실제 결과는 다음과 같았습니다.

ClickHouse bulk INSERT: 실패
ack: 0개
retry: 100개

이 구조에는 다음 문제가 있었습니다.

Cloudflare Queue는 최대 batch 크기가 100이며, 메시지별 ack()와 retry()를 사용할 수 있습니다. 메시지가 설정된 최대 retry 횟수에 도달하면 DLQ로 이동합니다. Cloudflare Queue batching and retries

바뀐 구조

시나리오 설정

동일한 100개 batch를 사용합니다.

변경된 슈도 코드

async function consumeQueueBatch(batch) {
  const preparation = await prepareEachMessage(batch.messages);

  const existingRows = preparation.accepted.filter(
    item => item.alreadyPersisted
  );

  const insertItems = preparation.accepted.filter(
    item => !item.alreadyPersisted
  );

  try {
    // 정상 경로: bulk INSERT 1회
    await storage.insertCanonicalEvents(
      insertItems.map(item => item.row)
    );

    addPending(existingRows);
    addPending(insertItems);
  } catch (error) {
    // 오류 경로: 실패 이벤트 격리
    const result = await isolateInsertFailure(
      insertItems,
      error
    );

    addPending(existingRows);
    addPending(result.persisted);

    for (const failed of result.failed) {
      failed.message.retry({
        delaySeconds: retryDelay(failed.message.attempts)
      });
    }
  }

  try {
    // 저장 성공 이벤트가 속한 site만 materialize
    await materialize(pendingSiteIds());

    for (const message of pendingMessages()) {
      message.ack();
    }
  } catch (error) {
    // materializer 실패 site의 메시지만 retry
    retryFailedSites(error);
    ackSuccessfulSites();
  }
}

INSERT 오류 격리 슈도 코드

async function isolateInsertFailure(items, initialError) {
  if (items.length === 1) {
    return {
      persisted: [],
      failed: [{ item: items[0], error: initialError }]
    };
  }

  if (!isRetryableClickHouseError(initialError)) {
    return {
      persisted: [],
      failed: items.map(item => ({
        item,
        error: initialError
      }))
    };
  }

  const [left, right] = splitInHalf(items);

  const leftResult = await tryInsert(left);
  const rightResult = await tryInsert(right);

  const persisted = [
    ...leftResult.successful,
    ...rightResult.successful
  ];

  const failed = [
    ...isolateFailed(leftResult),
    ...isolateFailed(rightResult)
  ];

  return { persisted, failed };
}

기존 문제를 해결한 방식

동일한 시나리오에서 개선된 결과

로컬 실제 ClickHouseClient HTTP 경계 테스트

초기 bulk INSERT: 100행, HTTP 503
정상 sub-batch INSERT: 99행, HTTP 200
sentinel singleton INSERT: HTTP 503
sentinel 데이터: 서버에 이미 저장됨

결과:

ack: 99개
retry: sentinel 1개
추가 INSERT: 없음
재전달 sentinel: SELECT dedup 후 ack

실제 Cloudflare Queue/DLQ 플랫폼 테스트

disposable Cloudflare Queue와 Worker를 사용해 실제 플랫폼에서 확인했습니다.

첫 batch:
  100개 처리
  99개 ack
  sentinel 1개 retry

재전달:
  sentinel 단일 batch, attempts=2
  sentinel 단일 batch, attempts=3

최대 retry 도달:
  sentinel 1개 DLQ 이동

DLQ payload에는 원본 sentinel body와 함께 CF-sourceQueueName, CF-sourceMessageId, CF-Content-Type 메타데이터가 포함됐습니다. Cloudflare Queue DLQ

결과적으로 정상 이벤트 99개는 재처리하지 않고 완료되며, 실제로 실패한 이벤트만 bounded retry 후 DLQ로 이동하는 구조로 개선되었습니다.