NATS JetStream은 처리 중인 메시지를 왜 또 보낼까?

이번 시간에는 NATS JetStream 워커가 오래 처리하는 동안 같은 메시지가 언제 다시 전달되는지 시간을 재 보고, in progress로 이를 막는 방법과 처리 뒤 ack 전에 죽었을 때 중복을 무엇으로 거를지까지 배워가는 시간입니다.

예를 들어서 주문 하나를 처리하는 데 12초가 걸리는 워커 두 개(a, b)가 JetStream 컨슈머 하나에서 메시지를 나눠 받는다고 해볼게요. 여기에 주문 1건을 넣으면, 워커는 하나도 죽지 않았고 오류도 없는데 주문 하나가 세 번 처리돼요. 왜 그럴까요?

컨슈머의 ack_wait이 5초이기 때문이에요. JetStream은 ack가 ack_wait 안에 오지 않으면 처리가 끝났는지와 상관없이 같은 메시지를 다시 보냅니다.

워커가 메시지에 보낼 수 있는 응답은 ack 말고도 넷이 더 있어요.

nak는 처리에 실패했으니 다시 보내 달라는 응답, nak 지연은 정한 시간 뒤에 다시 보내 달라는 응답, in progress는 아직 처리 중이니 기다려 달라는 응답, term은 다시 보내지 말라는 응답이에요. JetStream의 스트림과 컨슈머 기본 동작이 더 궁금하면 JetStream을 다룬 글을 참고하세요.

실습 준비

이 글의 명령은 모두 실습 코드의 nats-jetstream-ack-redelivery/ 디렉터리에서 실행합니다.

• compose.yaml: NATS 서버, nats-box, Python 워커 컨테이너 정의
• Dockerfile: 워커 이미지
• common.sh: 실습 블록이 불러 쓰는 셸 함수(n, cinfo, worker, ready)
• w.py: pull 컨슈머에서 받아 옵션대로 ack, nak, term, in progress로 응답하는 워커
• show.py: 실습 2의 처리 기록 DB에서 두 테이블을 출력하는 보조 스크립트

구성은 이렇게 되어 있어요.

• NATS 서버: nats:2.15.0. -js로 JetStream을 켜고, -sd /data에 스트림을 저장하고, -m 8222로 모니터링 포트를 엽니다.
• 명령줄 도구: natsio/nats-box:0.20.0에 들어 있는 nats CLI 0.5.0
• 워커: Python 3.12와 nats-py 2.16.0. 워커 하나가 컨테이너 하나이고 NATS 연결도 하나씩 따로 맺습니다.

compose.yaml은 이 세 서비스(nats, nats-box box, 워커 py)를 정의해요.

아래는 py 서비스의 Dockerfile이에요. pydata 볼륨은 실습 2에서 처리 기록 DB 파일을 두는 곳입니다.

FROM python:3.12-slim
RUN pip install --no-cache-dir --root-user-action=ignore nats-py==2.16.0

실습에서 쓸 셸 함수는 common.sh에 들어 있어요. 실습마다 명령 블록 맨 앞의 source ./common.sh로 불러옵니다. common.sh는 compose 프로젝트 이름을 natsak로 정하고 아래 함수들을 정의해요. 그 밖의 함수 adv_start, adv_stop은 이 글에서는 쓰지 않아요.

n은 미리 띄워 둔 nats-box 컨테이너 natsak-cli에서 nats 명령을 실행해요. 그래서 명령마다 컨테이너를 새로 만들지 않습니다.

worker는 natsak-<이름> 컨테이너로 워커 w.py를 띄우고, ready는 그 워커가 첫 pull 요청을 보낼 때까지 기다려요.

cinfo는 컨슈머 상태에서 전달 위치, ack floor(이 번호까지는 모두 처리 완료로 친 stream sequence), ack를 기다리는 메시지 수, 대기 중인 pull 요청 수만 꺼냅니다.

COMPOSE_IGNORE_ORPHANS는 docker compose run으로 이름을 붙여 띄운 컨테이너 때문에 나오는 경고를 꺼 줘요. 이제 서버를 띄우고 스트림 ORDERS를 만들어 주세요.

source ./common.sh
mkdir -p out
docker compose up -d
docker compose build py
docker compose run -d --name natsak-cli box tail -f /dev/null > /dev/null
n --version
n stream add ORDERS --subjects 'orders.>' --storage file --retention limits --defaults | grep -E "Subjects|Retention|Host Version"

0.5.0
                     Subjects: orders.>
                    Retention: Limits
                 Host Version: 2.15.0

워커 w.py는 pull 컨슈머에서 메시지를 받아, 받은 시간과 str_seq(stream sequence), cons_seq(consumer sequence), delivered(이 메시지를 이 컨슈머가 보낸 횟수)를 한 줄로 찍고 옵션에 따라 응답해요.

옵션의 뜻은 w.py의 주석에 적어 두었어요. --db, --idempotent, --crash-before-ack는 실습 2에서 씁니다.

시간은 time.time()을 UTC로 찍어요. 컨테이너들은 같은 Docker VM 시계를 쓰니까 워커끼리 시간을 비교할 수 있습니다.

글에 실린 출력은 같은 스트림에 다른 메시지를 먼저 넣은 뒤 잰 것이라, str_seq 번호는 따라 하실 때와 다를 수 있어요.

실습 1: 오래 걸리는 처리와 in progress

JetStream 컨슈머는 ack가 없으면 다른 워커에게 다시 보내는데, 그 기준이 처리 완료가 아니라 시간이라서 문제가 생겨요.

ack_wait 5초 컨슈머 s2에 워커 a, b를 붙이고 주문 1건을 넣을게요. 두 워커 모두 처리에 12초가 걸리고, 처리를 마치면 ack합니다. 컨슈머는 --ack explicit(메시지마다 ack를 받아야 처리 완료로 치는 ack 정책)과 --max-deliver=-1(같은 메시지를 보낼 최대 횟수 제한 없음)로 만들어요.

none은 처리 중에 아무것도 보내지 않고, progress는 2초마다 in_progress()를 보내요.

pull 컨슈머는 대기 중인 pull 요청이 있어야 메시지를 다시 보낼 수 있어요. 그래서 두 워커를 먼저 띄워 요청을 걸어 둔 뒤 발행합니다. 각각 세 번씩 해 볼게요.

source ./common.sh

for mode in ${MODES:-none progress}; do
  for r in ${ROUNDS:-1 2 3}; do
    echo "=== $mode 회차 $r"

    n consumer add ORDERS s2 --pull --ack explicit --wait 5s --max-deliver=-1 --filter orders.s2.$mode.r$r --defaults > /dev/null
    extra=""; [ $mode = progress ] && extra="--progress-every 2"

    worker a s2 --work 12 $extra --fetch-timeout 60 --run-for 60
    ready a
    worker b s2 --work 12 $extra --fetch-timeout 60 --run-for 60
    ready b

    n pub orders.s2.$mode.r$r "order-$r" > /dev/null
    docker wait natsak-a natsak-b > /dev/null

    for w in a b; do docker logs natsak-$w > out/s2-$mode-r$r-$w.log 2>&1; echo "--- $w"; grep -v "in progress" out/s2-$mode-r$r-$w.log; echo "(in progress $(grep -c 'in progress' out/s2-$mode-r$r-$w.log)번)"; done
    cinfo s2

    docker rm -f natsak-a natsak-b > /dev/null
    n consumer rm ORDERS s2 -f > /dev/null
  done
done

none의 1회차 출력(발췌)이에요.

--- a
01:06:15.213 받음 str_seq=10 cons_seq=1 delivered=1 본문=order-1
01:06:27.216 처리 완료
01:06:27.217 -> ack 반환
01:06:27.220 받음 str_seq=10 cons_seq=3 delivered=3 본문=order-1
01:06:39.233 처리 완료
01:06:39.233 -> ack 반환
--- b
01:06:20.212 받음 str_seq=10 cons_seq=2 delivered=2 본문=order-1
01:06:32.226 처리 완료
01:06:32.226 -> ack 반환

a가 주문(스트림 안 번호 str_seq=10)을 처리하는 동안, 5초 만에 b가 같은 주문을 받았어요. 그리고 a는 처리를 마치고 ack를 보낸 직후 같은 주문을 세 번째로 받았습니다.

세 번 모두 같은 모양이었습니다.

• b가 같은 str_seq를 delivered=2로 받은 시간: a의 첫 수신 뒤 4.999~5.005초
• a가 ack한 직후 같은 메시지를 delivered=3으로 다시 받은 시간: ack 뒤 3ms
• 처리 완료: 메시지 1건에 3번

5초에 b가 받은 건 a의 ack_wait이 지났기 때문이에요.

세 번째 전달은 b의 ack_wait 때문이에요. b도 받은 지 5초가 지나도록 ack하지 않았으니 서버는 이 메시지를 다시 보낼 대상에 올렸고, 그때 두 워커 모두 처리 중이라 보낼 곳이 없었어요.

출력에 보이는 것은 a의 ack 반환 3ms 뒤 delivered=3 수신이에요. 서버 안에서 a의 ack가 반영되기 전에, a가 이어서 보낸 pull 요청으로 메시지가 나간 것으로 보여요. 출력에서 끌어낸 추론입니다.

같은 실험을 미리 한 번 돌렸을 때는 ack가 먼저 반영되어 2번 처리로 끝났어요. 그러니 따라 해 보시면 2번이 나올 수도 있습니다.

이번엔 progress의 결과예요. 1회차만 실었어요.

=== progress 회차 1
01:10:04 Published 7 bytes to "orders.s2.progress.r1"
--- a
01:09:56.029 대기 시작
01:10:04.802 받음 str_seq=13 cons_seq=1 delivered=1 본문=order-1
01:10:16.819 처리 완료
01:10:16.819 -> ack 반환
01:10:56.034 끝: 응답으로 마친 메시지 1건
(in progress 5번)
--- b
01:10:03.817 대기 시작
01:11:03.819 끝: 응답으로 마친 메시지 0건
(in progress 0번)
  Last Delivered Message: Consumer sequence: 1 Stream sequence: 13 Last delivery: 49.55s ago
...

a가 2초마다 in progress를 다섯 번 보내는 동안 b는 60초 내내 아무것도 받지 못했어요. 컨슈머의 Consumer sequence: 1대로 전달은 한 번뿐이었고, 세 번 모두 같았습니다.

NATS 문서(JetStream Consumers)는 in progress를 받으면 ack_wait 타이머를 다시 센다고 설명해요. 서버 소스에서도 in progress는 그 메시지의 전달 시간을 지금으로 바꿉니다.

ack_wait은 서버가 ack를 기다리는 시간이고, 서버는 그동안 처리가 진행 중인지 알지 못해요.

그러니 처리가 그보다 길어질 수 있으면 ack_wait을 늘리거나, ack_wait보다 짧은 간격으로 in progress를 보내 주세요.

실습 2: 처리는 끝났는데 ack 전에 죽으면

포인트 적립 서비스를 예로 들어 볼게요.

워커가 메시지 user-1 100을 받아 SQLite 파일 /state/points.db의 points 테이블에 100포인트를 더해 커밋하고, ack를 보내기 직전에 죽어요(--crash-before-ack는 os._exit(1)로 프로세스를 바로 끝냅니다). 그다음 새 워커를 띄웁니다.

두 워커가 동시에 돌지 않도록 docker wait로 앞 워커가 끝난 것을 확인한 뒤 띄워요. ack_wait은 3초입니다.

• s7a: 멱등 처리 없이 받을 때마다 더합니다.
• s7b: processed(stream_seq PRIMARY KEY, ...) 테이블에 처리 기록을 넣는 것과 포인트를 더하는 것을 한 트랜잭션으로 커밋합니다. 이미 있는 stream_seq면 기본 키 충돌로 트랜잭션이 취소되므로 적립을 건너뛰고 ack만 합니다(w.py의 add_points).

두 테이블의 내용은 show.py로 출력해요.

source ./common.sh

show() { docker compose run --rm -T py python show.py; }

for c in s7a s7b; do
  echo "=== $c"
  docker compose run --rm -T py rm -f /state/points.db
  opt=""; [ $c = s7b ] && opt="--idempotent"

  n consumer add ORDERS $c --pull --ack explicit --wait 3s --max-deliver=-1 --filter orders.$c --defaults > /dev/null
  n pub orders.$c "user-1 100" > /dev/null

  worker p1 $c --db /state/points.db $opt --crash-before-ack
  echo "p1 종료 코드: $(docker wait natsak-p1)"
  docker logs natsak-p1

  worker p2 $c --db /state/points.db $opt --max 1
  docker wait natsak-p2 > /dev/null
  docker logs natsak-p2

  show
  docker rm -f natsak-p1 natsak-p2 > /dev/null
done

=== s7a
01:20:31 Published 10 bytes to "orders.s7a"
p1 종료 코드: 1
01:20:40.005 대기 시작
01:20:40.006 받음 str_seq=67 cons_seq=1 delivered=1 본문=user-1 100
01:20:40.008 적립 커밋 user-1 total=100
01:20:40.008 ack 전에 종료
01:20:47.413 대기 시작
01:20:47.414 받음 str_seq=67 cons_seq=2 delivered=2 본문=user-1 100
01:20:47.418 적립 커밋 user-1 total=200
01:20:47.418 -> ack 반환
01:20:47.418 끝: 응답으로 마친 메시지 1건
points: [('user-1', 200)]
processed: []
=== s7b
01:21:03 Published 10 bytes to "orders.s7b"
p1 종료 코드: 1
01:21:11.371 대기 시작
01:21:11.373 받음 str_seq=68 cons_seq=1 delivered=1 본문=user-1 100
01:21:11.375 적립 커밋 user-1 total=100
01:21:11.375 ack 전에 종료
01:21:20.409 대기 시작
01:21:20.411 받음 str_seq=68 cons_seq=2 delivered=2 본문=user-1 100
01:21:20.411 이미 처리한 stream_seq=68, 적립 건너뜀
01:21:20.411 -> ack 반환
01:21:20.411 끝: 응답으로 마친 메시지 1건
points: [('user-1', 100)]
processed: [(68, 1, 1)]

멱등 처리가 없는 s7a는 주문 1건에 200포인트가 쌓였어요.

s7b는 두 번째 워커가 같은 str_seq=68을 받았지만 처리 기록에 이미 있어서 건너뛰었고, 100포인트로 끝났습니다.

여기서 중복을 거르는 키는 stream sequence여야 해요. 왜 consumer sequence는 안 될까요? consumer sequence는 전달할 때마다 새 번호가 붙기 때문이에요.

이 실습에서도 같은 메시지가 cons_seq=1과 cons_seq=2로 왔어요. cons_seq를 키로 쓰면 재전달을 새 메시지로 보고 다시 처리하게 됩니다.

그리고 stream sequence는 스트림 하나 안에서만 고유하니까, 여러 스트림을 처리한다면 스트림 이름과 함께 키로 써 주세요.

처리 기록과 비즈니스 쓰기를 한 트랜잭션으로 묶는 방법과 다른 멱등 방법이 더 궁금하면 at-least-once와 멱등 처리를 다룬 글을 참고하세요.

다만 이 키로 거를 수 없는 중복도 있어요. 발행자가 응답을 받지 못해 같은 주문을 다시 보내면, 스트림에 새 stream sequence로 저장되거든요.

그 경우는 발행할 때 Nats-Msg-Id 헤더를 붙여 서버가 같은 ID를 한 번만 저장하게 하는 중복 제거로 막거나, 주문 번호처럼 본문에 있는 업무 키로 걸러 주세요.

정리

범위: NATS 2.15.0(nats:2.15.0), nats CLI 0.5.0, Python nats-py 2.16.0, 파일 저장 limits 스트림 ORDERS, 실험마다 3회.

• 처리 12초, ack_wait 5초에서 다른 워커가 4.999~5.005초에 같은 메시지를 받아 1건이 세 번 처리됐고(미리 돌린 한 번은 두 번), 2초마다 in_progress()를 보내자 60초 동안 전달은 한 번뿐이었어요. 처리가 ack_wait보다 길어질 수 있으면 ack_wait보다 짧은 간격으로 in progress를 보내거나 ack_wait을 늘려 주세요.
• 처리를 커밋한 뒤 ack 전에 죽자 멱등 처리 없는 s7a는 주문 1건에 200포인트, stream_seq 처리 기록을 같은 트랜잭션에 넣은 s7b는 100포인트였어요. 중복은 consumer sequence가 아니라 stream sequence로 거르세요.
• 이 글에 싣지 않은 추가 측정에서는 지연 없는 nak가 3초 동안 같은 메시지를 5071~5485번 돌려받았고 재전달 간격을 차례로 늘리는 backoff [1, 3, 6] 컨슈머에서도 5537번이었으며, nak(delay=3)은 3.001~3.007초 간격으로 지켜졌어요. term한 깨진 주문은 ack_wait 2초가 다섯 번 지나도 다시 오지 않았어요. 잠시 뒤 다시 하면 될 실패에는 지연을 붙인 nak를, 다시 받아도 실패할 메시지에는 term을 보내세요.
• 이 글에 싣지 않은 추가 측정에서는 ack를 기다리는 메시지 수의 한도인 max_ack_pending 3에서 응답 없이 멈춘 워커 하나가 한도를 채우자 다른 워커는 16.9~22.7초 동안 아무것도 받지 못하다가 ack_wait 30초가 지나서야 받았어요. 컨슈머 상태의 Outstanding Acks가 한도에 붙어 있는지 지켜봐 주세요.
• 이 글에 싣지 않은 추가 측정에서는 네트워크를 끊은 워커의 ack()가 예외 없이 반환됐는데 약 7초 뒤 다른 워커가 같은 메시지를 받았고, 서버의 확인 응답까지 기다리는 ack_sync()는 1초 뒤 TimeoutError를 냈을 뿐 재전달은 같았어요. AckSync를 써도 처리 쪽 중복 제거는 빼지 마세요.