Worker는 Thread가 아니다에서는 worker를 concurrency primitive가 아니라 실행 시도의 수명주기를 소유하는 역할로 정의했다. 바로 앞의 Conversation은 Run이 아니다에서는 session 안의 각 실행을 별도 run_id로 식별했다.
이제 가장 중요한 실패를 넣어보자. Worker가 job을 가져간 뒤 Pod가 종료된다. Queue에는 이미 전달된 것으로 보이지만 PostgreSQL에는 최종 결과가 없다. 이때 시스템은 작업을 잃어도, 성공한 것으로 오해해도, 무한히 재실행해도 안 된다.
새 job을 Worker A에 전달
아직 XACK하지 못한 전달
“완료”가 아니라 “미확인 전달”
유휴 시간이 충분히 긴 entry를 재개
durable 결과를 남긴 뒤에만 PEL에서 제거
핵심은 한 문장이다.
전달(delivery)은 완료(completion)가 아니다.
Consumer group은 일을 나누고 미완료 전달을 기억한다
API는 durable run row를 만든 뒤 Redis Stream에 작은 실행 지시를 추가한다. Queue payload에 전체 대화와 비밀 정보를 복제하기보다 worker가 기준이 되는 영속 저장소에서 읽을 run_id를 중심으로 전달하는 편이 낫다.
message_id = await redis.xadd(
"agent:jobs",
{
"run_id": str(run_id),
"session_id": str(session_id),
},
)Consumer group을 한 번 생성한다. 0은 기존 entry부터, $는 생성 시점 이후 entry부터 읽겠다는 의미이므로 배포 정책에 맞게 선택해야 한다.
try:
await redis.xgroup_create(
"agent:jobs",
"agent-workers",
id="0",
mkstream=True,
)
except ResponseError as error:
if "BUSYGROUP" not in str(error):
raise여러 worker가 같은 group에서 XREADGROUP과 >를 사용하면 아직 그 group에 전달되지 않은 새 entry를 나눠 받는다.
messages = await redis.xreadgroup(
groupname="agent-workers",
consumername=worker_id,
streams={"agent:jobs": ">"},
count=4,
block=5_000,
)Redis는 전달된 entry의 ID와 consumer를 group의 Pending Entries List, 즉 PEL에 기록한다. PEL은 “실행이 끝났다”는 기록이 아니라 누군가에게 전달됐지만 아직 확인되지 않았다는 기록이다.
| state | Stream/consumer group에서 보이는 의미 | 애플리케이션이 확인할 것 |
|---|---|---|
| 아직 전달되지 않음 | group cursor 뒤의 새 entry | queue lag |
| 전달됐지만 ack 전 | PEL의 pending entry | run/attempt state, idle time |
XACK 완료 | PEL에서 제거 | durable terminal result |
| stream에서 삭제/trim | 원본 payload가 없을 수 있음 | 보존 정책과 PEL 참조 |
Ack는 결과 저장 뒤에만 한다
안전한 기본 순서는 다음과 같다.
이제 PEL의 전달 기록을 제거
async def handle_entry(entry_id: str, fields: dict[bytes, bytes]) -> None:
run_id = fields[b"run_id"].decode()
attempt = await run_store.start_attempt(run_id, worker_id)
try:
result = await execute_agent(run_id, attempt.id)
except Exception as error:
policy = classify(error)
await run_store.record_failure(attempt.id, error, policy=policy)
if policy == "retryable":
# retry_at을 durable store에 기록한다.
# 별도 scheduler가 준비된 시점에 새 job을 발행한다.
await run_store.schedule_retry(run_id, attempt.id, retry_at=next_retry_at())
await redis.xack("agent:jobs", "agent-workers", entry_id)
return
if policy == "terminal":
await run_store.fail_run(run_id, attempt.id, error)
else:
await run_store.quarantine(run_id, attempt.id, error)
await redis.xack("agent:jobs", "agent-workers", entry_id)
return
await run_store.complete_attempt(
run_id=run_id,
attempt_id=attempt.id,
result=result,
)
await redis.xack("agent:jobs", "agent-workers", entry_id)먼저 ack하고 결과 저장에 실패하면 job은 PEL에서 사라졌는데 사용자는 결과를 찾을 수 없다. 반대로 결과 저장에 성공한 직후 process가 종료되면 ack되지 않은 entry가 재전달될 수 있다. 후자는 중복을 안전하게 처리할 수 있다면 복구 가능하다. 반면 의도적인 재시도는 PEL에 그냥 남겨 두는 방식이 아니라, 다음 시도 시점과 attempt 수를 durable store에 기록한 뒤 별도의 scheduler가 다시 발행하도록 설계한다. PEL 회수는 worker 장애로 확인되지 않은 전달을 복구하는 장치이고, backoff는 재시도 정책의 책임이다.
또한 XACK은 PEL의 참조를 제거하지, Stream entry 자체를 삭제하지 않는다. 메모리 보존과 삭제는 XTRIM, XDEL 또는 별도 retention 정책의 책임이다.
Worker가 죽으면 pending entry가 남는다
Worker A가 entry를 받은 뒤 종료되면 그 entry는 A의 pending 항목으로 남는다. 다른 worker가 단순히 >로 새 메시지만 읽어서는 가져오지 못한다. 별도의 recovery loop가 오래 유휴 상태인 pending 항목을 회수해야 한다.
async def reclaim_stale_jobs() -> None:
cursor = "0-0"
while True:
cursor, entries, deleted = await redis.xautoclaim(
"agent:jobs",
"agent-workers",
worker_id,
min_idle_time=120_000,
start_id=cursor,
count=20,
)
for entry_id, fields in entries:
await handle_entry(entry_id, fields)
if cursor == "0-0":
breakXAUTOCLAIM의 min_idle_time은 “이 시간이 지나면 원래 worker가 반드시 죽었다”는 증명이 아니다. 일시적인 네트워크 지연이나 오래 걸리는 tool call 때문에 살아 있는 worker가 여전히 실행 중일 수 있다. 너무 짧으면 같은 run을 두 worker가 동시에 실행하고, 너무 길면 복구가 늦어진다.
따라서 timeout은 관측한 실행 시간 분포를 바탕으로 정하고, 긴 작업에는 heartbeat나 lease 갱신을 추가한다. Queue의 idle time만으로 실행 소유권을 완전히 판단하지 말고 durable attempt에도 ownership을 둔다.
Attempt와 fencing으로 늦게 돌아온 worker를 막는다
Run은 사용자 관점의 실행이고 attempt는 worker가 맡은 한 번의 시도다.
응답이 늦어져 실행 소유권이 불확실해짐
더 새로운 fencing token을 받아 실행
DB는 active_fence와 일치하는 attempt의 결과만 기록한다.
Worker A가 멈춘 것으로 판단해 B가 회수했는데 A가 늦게 다시 응답할 수 있다. “마지막 write가 이긴다”에 맡기면 오래된 A가 B의 결과를 덮어쓴다. 새 attempt마다 증가하는 fencing token을 발급하고, 현재 token과 일치하는 write만 허용할 수 있다.
UPDATE runs
SET status = 'succeeded',
result_json = $1,
finished_at = now()
WHERE run_id = $2
AND active_fence = $3
AND status = 'running';영향받은 row가 0개면 해당 worker는 실행 권한을 잃은 오래된 attempt다. 결과를 버리고 queue 처리 state를 다시 확인해야 한다.
At-least-once는 부수 효과를 중복시킬 수 있다
재전달 가능한 queue에서 “코드가 한 번만 실행된다”고 가정하면 안 된다.
| Crash 시점 | 재전달 뒤 가능한 결과 | 필요한 방어 |
|---|---|---|
| job 수신 전 | 다른 worker가 정상 수신 | 없음 |
| tool 호출 전 | Agent 계산 재실행 | 비용/시도 제한 |
| 외부 API 성공 후 기록 전 | 외부 부수 효과 중복 | idempotency key, execution ledger |
| 결과 DB commit 후 ack 전 | 완료 run 재실행 | terminal state 확인, 멱등 완료 |
| ack 후 | queue 기준 완료 | durable 결과가 이미 있어야 함 |
Worker는 실행을 시작하기 전에 terminal state를 확인해야 한다.
run = await run_store.get(run_id)
if run.status in {"succeeded", "failed", "cancelled"}:
await redis.xack("agent:jobs", "agent-workers", entry_id)
return하지만 status 확인만으로 외부 부수 효과의 중복을 막을 수는 없다. 이메일 전송, 결제, 티켓 생성처럼 process 밖에서 일어나는 tool은 (run_id, tool_call_id) 같은 안정적인 멱등성 키를 사용해야 한다. 이 경계는 Agent Tool은 정확히 한 번 실행될 수 있을까에서 다룬 원칙과 같다.
재시도는 실패를 반복하는 기능이 아니라 정책이다
재시도는 실패를 반복하는 기능이 아니라, 어디까지·언제·어떻게 다시 시도할지 정하는 정책이다. 모든 오류를 같은 방식으로 재시도하면 장애를 증폭시킨다.
def classify(error: Exception) -> str:
if isinstance(error, (RateLimitError, TemporaryNetworkError)):
return "retryable"
if isinstance(error, (InvalidInputError, PermissionDenied)):
return "terminal"
return "unknown"정책에는 최소한 다음 항목이 필요하다.
- 최대 attempt 수와 전체 elapsed-time budget
- exponential backoff와 jitter
- 오류 분류와 provider별 제한
- replay 가능한 checkpoint
- 부수 효과의 멱등성
- 최대 횟수 이후 dead-letter 또는 quarantine 경로
계속 실패하는 작업을 PEL에 영원히 남겨두면 recovery loop가 같은 실패를 반복한다. 최대 횟수를 넘으면 durable run을 최종 실패로 기록하고, 원본 entry ID·오류 분류·attempt history를 조사 가능한 저장소에 남긴 뒤 ack한다.
if attempt.number >= MAX_ATTEMPTS:
await run_store.fail_permanently(run_id, attempt.id, error)
await redis.xadd(
"agent:jobs:dead",
{"run_id": run_id, "source_id": entry_id, "reason": error.code},
)
await redis.xack("agent:jobs", "agent-workers", entry_id)Dead-letter stream은 자동 재실행 목록이 아니라 격리와 조사 경계다. 재투입은 원인을 고친 뒤 명시적으로 수행해야 한다.
Queue 자체의 durability도 설정해야 한다
Redis Stream을 사용한다고 모든 서버 장애에서 job이 자동 보존되는 것은 아니다. Redis persistence, replication, failover와 백업 정책이 실제 요구 수준에 맞아야 한다. RDB snapshot과 AOF는 서로 다른 durability·성능 trade-off를 갖는다.
또한 Stream retention을 PEL과 무관하게 공격적으로 줄이면 pending ID는 남았지만 원본 entry가 이미 trim된 state가 생길 수 있다. 버전별 명령 semantics를 확인하고, 최소한 최대 복구 시간보다 긴 보존 기간을 둔다. Queue를 유일한 업무 기록으로 삼지 않고 PostgreSQL의 run과 attempt를 durable source of truth로 두는 이유이기도 하다.
운영에서는 새 job과 pending job을 함께 본다
Queue length 하나만 보면 장애를 놓친다. 새 entry가 모두 전달되어 queue lag는 0인데, 죽은 consumer 아래 pending job 수백 개가 남을 수 있다.
아직 consumer에 전달되지 않은 entry
전달됐지만 XACK되지 않은 entry
가장 오래 멈춰 있는 pending entry
run·attempt 지표에서 보는 실행 상태
XINFO GROUPS, XINFO CONSUMERS, XPENDING을 애플리케이션의 run 지표와 함께 관찰한다. Consumer 이름도 Pod의 임의 hostname만 쓰기보다 배포와 instance를 추적할 수 있게 정한다.
장애 테스트는 process를 실제로 죽여야 한다
정상 처리 unit test만으로 recovery path를 검증할 수 없다.
| Test | Expected result |
|---|---|
| job 수신 직후 worker 종료 | entry가 pending으로 남고 다른 worker가 회수 |
| 결과 commit 직후, ack 전 종료 | 재전달되지만 종료 상태가 확정된 run은 재실행하지 않고 ack |
| tool 부수 효과 직후 종료 | 같은 idempotency key로 중복 효과 방지 |
| 살아 있는 긴 작업을 너무 일찍 reclaim | fencing이 오래된 attempt의 write를 거부 |
| retryable 오류 반복 | backoff 후 최대 횟수에서 격리 |
| terminal 입력 오류 | 불필요한 retry 없이 실패 기록 후 ack |
| stream 원본이 trim된 pending ID | 탐지·경보하고 durable run으로 복구 판단 |
async def test_crash_after_commit_does_not_repeat_tool(worker, queue, store):
job = await queue.publish(run_id="R1")
await worker.run_until_result_committed(job)
await worker.kill_before_ack()
await recovery_worker.reclaim(job)
assert await store.run_status("R1") == "succeeded"
assert await store.tool_execution_count("R1") == 1
assert await queue.is_pending(job) is False이 테스트는 mock exception 하나보다 시스템의 실제 보장을 더 잘 드러낸다. 별도 process나 container로 worker를 실행하고 강제 종료해 PEL과 DB state를 함께 확인하는 통합 테스트가 필요하다.
구현 체크리스트
XREADGROUP의 새 entry 읽기와 pending recovery loop가 모두 있는가?- PEL을 완료 목록이 아니라 미확인 전달 목록으로 이해하는가?
- 최종 state를 durable하게 저장한 뒤에만
XACK하는가? XACK과 stream entry 삭제를 구분하는가?- Run과 attempt를 분리하고 stale worker write를 거부하는가?
- reclaim timeout이 실제 작업 시간과 heartbeat 정책에 근거하는가?
- 종료 상태가 확정된 run이 재전달되면 안전하게 ack하는가?
- 모든 외부 부수 효과에 안정적인 idempotency key가 있는가?
- retry budget과 poison-job 격리 경로가 있는가?
- Redis persistence와 Stream retention이 복구 요구를 만족하는가?
- queue lag뿐 아니라 pending count와 oldest idle time을 경보하는가?
- ack 전후에 worker를 실제로 죽이는 통합 테스트가 있는가?
다음은 취소 요청을 실행 경계까지 전달하는 일이다
Consumer group은 worker process가 사라져도 실행 의도를 남기고 다른 worker가 가져갈 수 있게 한다. 하지만 사용자가 실행 중인 작업을 멈춰 달라고 요청하면, 새 worker가 그것을 다시 회수해서는 안 된다. 취소 의도는 durable하게 남고, worker·원격 tool·queue가 같은 종료 상태를 보도록 연결되어야 한다.
다음 글에서는 Agent를 멈춘다는 것: Cancellation은 버튼이 아니라 Protocol이다를 다룬다. 취소 요청부터 cooperative stop, 부수 효과, terminal race와 ACK 순서를 연결해, 복구 가능한 queue가 이미 취소된 작업을 되살리지 않게 만드는 방법을 살펴본다.
Discussion
토론
질문, 정정, 반례, 다른 해석이 있다면 남겨 주세요.