본문 바로가기
C.W.K.
Stream
More →
Lesson 05 of 05 · published

폰이 끊겨도 턴은 안 죽는다

~18 min · master-chief, server-sent-events, idempotency, reliability

Level 0구경꾼
0 XP0/33 lessons0/11 achievements
0/100 XP to next level100 XP to go0% complete

폰은 끊기게 돼 있어

아빠의 턴은 주머니 속 폰에서, 손목의 워치에서, 이동통신망 위의 Telegram에서, 잠드는 노트북에서 와. 진짜 Chief 턴은 몇 분씩 돌기도 해. 도구 쓰고, 위임하고, 길게 답하고. 서버가 턴을 그걸 실어 온 요청 안에서 돌리면, 연결이 끊길 때마다 턴이 중간에 취소돼. 답은 날아가고, 상태는 반쯤 쓰이고, 모델이 한 일은 돈만 내고 버려져. 같이 쓰는 방이라 더 나빠. 다음에 Chief를 여는 기기는 답 없는 질문을 보게 되니까.

턴과 연결을 떼어낸다

Chief 라우트는 순진한 스트리밍 핸들러가 한데 붙여놓는 일 두 개를 갈라. produce() 코루틴은 턴을 쥐어. 턴 락을 기다리고, 브레인의 원래 채팅 핸들러를 돌리고, 그 스트림을 asyncio.Queue로 다 받아내. 답은 그 과정에서 저장돼. 시작은 spawn_detached로 해. create_task에, 태스크가 끝날 때까지 강한 참조를 쥐는 걸 더한 거야 (참조 없는 태스크는 도는 중에 가비지 컬렉션될 수 있다고 파이썬 문서가 경고하거든). 그리고 앱의 background_streams 집합에 등록돼서, 정상 종료가 이걸 기다려줘. consume() 제너레이터는 연결을 쥐어. 큐를 읽고 조각마다 클라이언트로 내보내. 10초 동안 조용하면 : chief waiting을 내보내. Server-Sent Events의 주석 줄이라 클라이언트는 무시하지만, 턴이 줄 서 있거나 모델이 생각하는 동안 프록시랑 소켓이 스트림을 죽은 걸로 판정하지 않게 해줘.

폰이 끊기면 죽는 건 consume()뿐이야. produce()는 끝까지 받아내고, 답은 대화에 내려앉아. 엔진 테스트가 정확히 이걸 해. 턴 도중에 클라이언트 요청을 취소하고, 모델이 마저 끝내게 두고, 마지막 행이 "finished offline"인지 확인해. 2026-10-08 콜드 리뷰도 이 모양을 점검했고 문제없다고 했어.

나중에 묻되, 두 번 시키지 않는다

스트림을 놓친 클라이언트는 턴을 다시 시작하지 않고 무슨 일이 있었는지 물어야 해. 그래서 클라이언트마다 보내기 전에 요청 id를 찍어. Telegram은 tg-<chat>-<message>, WebUI는 webui-<uuid>, 피파 앱은 pm-turn-<uuid>, 워치는 pw-msg-<uuid>. 엔진은 영문자·숫자·대시·밑줄 8자에서 80자까지 받고, 메시지 테이블의 유니크 인덱스 덕에 id 하나는 턴 하나만 가질 수 있어. 피파 앱의 아웃박스 드레인이 기준 클라이언트야. 보내기 전에 엔진한테 이 턴 이미 있냐고 물어. completed면 저장된 답을 다시 틀고, pending이면 계속 확인하고, 모르는 id일 때만 모델 호출을 시작해.

이걸 정직하게 만든 상태 두 개가 있어. 둘 다 콜드 리뷰에서 나왔어. 락을 기다리는 턴은 아직 행이 없었어. 그래서 턴 상태가 404로 답했고, 클라이언트는 다들 그걸 "도착 안 했네, 다시 보내"로 읽어. 첫 사본이 아직 기다리는 중에 다시 보낸 클라이언트는 턴을 두 번 돌렸지 (2번 지적). 이제 기다리는 턴의 id는 대기 맵에 들어 있고, pending, queued: true로 읽혀. 그리고 첫 전송이랑 경주하듯 겹친 재전송은, 받아들여지고 나서 행이 이미 있는 걸 보고 status 이벤트로 already_accepted를 답해. 에러가 아니라 상태야. 네이티브 아웃박스는 error 이벤트를 영구 실패로 보고 그 턴을 격리하거든. 이미 답이 달린 턴들한테 정확히 그 짓을 했었어 (8번 지적).

철회는 중단이 아냐

기다리던 턴이 안 돌고 끝나는 길은 두 가지고, 클라이언트는 어느 쪽인지 알아야 해. 아빠가 Stop을 누르면 cancel_waiting이 그 대화의 stop 리비전을 올리고, 아직 기다리는 턴을 전부 취소해 (워커 보고는 대기 목록에 아예 안 올라가서, Stop이 보고를 철회하는 일은 없어). 프로듀서는 취소를 받고, 리비전이 움직인 걸 보고, error를 withdrawn 상태로 보내. 문구는 Stopped before it began: this turn was withdrawn and will not run.이고, 뜻은 '다시 보내지 마'야. 반대로 서버가 내려가는 중이면, 종료 절차가 먼저 withdraw_waiting_for_shutdown을 불러. 리비전은 그대로고, 프로듀서는 status를 interrupted 상태로 보내. 문구는 Interrupted before it began; this turn was not run. Send it again.이야. 아웃박스 클라이언트는 서버가 돌아오면 딱 그렇게 해.

그 종료 경로는 2번 지적 때문에 생겼어. 재시작이 돌던 턴을, 백엔드가 내건 30초 유예 중 1초쯤에서 잘라 버렸어. 런처한테 따로 고정 3초 마감이 있었거든. 줄 서 있던 턴은 그냥 사라졌고. 고친 커밋(f3bd5968)은 런처가 서비스의 실제 종료 시간 제한을 따르게 했고, 기다리는 턴을 먼저 철회해서 유예가 잘라버릴 모델 호출을 아무도 시작하지 않게 했어. Telegram은 끊긴 뒤 턴 상태부터 묻고 엔진이 안 받았을 때만 같은 tg- id로 다시 보내게 바뀌었고, WebUI는 보낼 말을 큐로 되돌려 보류하게 바뀌었어. 13번 지적 밑으로 작은 항목 둘이 뒤따랐어. stop 리비전은 이제 시작할 때 읽어 들이는 chief_stop_revisions 테이블에 살아. 재시작으로 0이 된 카운터가 모든 클라이언트에서 원격 Stop으로 읽혀서 모든 큐를 붙잡았거든. 그리고 steer는 이제 돌고 있는 턴만 끝내. 예전엔 아빠의 다른 기기에서 줄 서 있던 턴까지 철회했어.

요청이 끝나는 모든 길에 이름을 붙이고, 그 이름이 클라이언트한테 다음 할 일을 알려주게 해. completed, pending, already accepted, withdrawn, interrupted. 답이 다섯이고, 저마다 맞는 클라이언트 반응이 딱 하나씩이야. "다시 해"라는 에러와 "절대 하지 마"라는 에러는 같은 에러가 아냐.

에러는 뭐라도 말해야 해

마지막 항목은 제일 작고 제일 민망해. Chief 라우트 자체의 에러 이벤트는 message 키를 썼는데, 클라이언트는 전부 text를 읽어. WebUI엔 "Error: undefined"가 떴어. 이제 라우트가 직접 내보내는 에러와 상태 이벤트엔 전부 text가 실리고, 와이어 문구는 모듈 맨 위에 상수로 모여 있어. 그 위엔 문구를 거기 두게 만든 지적 번호가 주석으로 달려 있고.

Code

엔진 발췌 (Chief 라우트에서 잘라냄): 프로듀서가 끝나는 두 길, 그리고 컨슈머의 하트비트·python
except asyncio.CancelledError:
    if master_chief.stop_revision(chief.id) != stop_seen:
        # Dad's Stop withdrew it while it waited its turn.
        await queue.put(_event({"type": "error", "state": "withdrawn", "text": WITHDRAWN_TEXT}))
    else:
        # Not a Stop: the server is going down. A notice, not
        # an error, so a client that keeps an outbox sends it
        # again once the server is back.
        await queue.put(_event({"type": "status", "state": "interrupted", "text": INTERRUPTED_TEXT}))
    raise

# ...

producer = spawn_detached(produce())

async def consume():
    while True:
        try:
            chunk = await asyncio.wait_for(queue.get(), timeout=10)
        except asyncio.TimeoutError:
            yield ": chief waiting\n\n"
            continue
        if chunk is None:
            break
        yield chunk
직접 돌려볼 수 있는 최소 스케치: 클라이언트 스트림이 사라진 뒤에도 떼어낸 프로듀서가 답을 저장한다·python
from __future__ import annotations

import asyncio

saved: dict[str, str] = {}               # request id -> persisted answer
_keep: set[asyncio.Task] = set()         # strong references to detached work


def spawn_detached(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    _keep.add(task)
    task.add_done_callback(_keep.discard)
    return task


async def handle(request_id: str, text: str):
    """Answer one turn. Returns the generator the client reads."""
    queue: asyncio.Queue = asyncio.Queue()

    async def produce():                  # owns the turn: runs to the end
        try:
            answer = ""
            for word in ("this", "reply", "outlives", "the", "phone"):
                await asyncio.sleep(0.1)  # the model, slowly
                answer += word + " "
                await queue.put(word)
            saved[request_id] = answer.strip()  # persisted whoever listens
        finally:
            await queue.put(None)

    spawn_detached(produce())

    async def consume():                  # owns only the connection
        while True:
            try:
                item = await asyncio.wait_for(queue.get(), timeout=10)
            except asyncio.TimeoutError:
                yield ": waiting"         # a heartbeat keeps proxies patient
                continue
            if item is None:
                return
            yield item

    return consume()


async def phone() -> None:
    stream = await handle("pm-turn-0001", "a long question")
    async for word in stream:
        print("phone got:", word)
        if word == "reply":
            print("-- the phone drops --")
            return                        # the consumer dies here


async def main() -> None:
    await phone()
    await asyncio.sleep(0.5)              # later, the phone comes back
    print("turn status:", saved.get("pm-turn-0001", "unknown id"))


asyncio.run(main())

External links

Exercise

스케치를 돌려서 답이 끊김을 버티는지 확인해. 그다음 기능을 늘려 봐. handle()에 요청 id가 이미 saved에 있거나 아직 만들어지는 중이면 "already accepted"를 돌려주는 확인을 넣어. 그리고 completed·pending·unknown 중 하나로 답하는 turn_status(request_id)를 추가해. 그리고 끊긴 뒤 상태부터 묻고, unknown일 때만 같은 id로 다시 보내는 클라이언트를 짜. 끊기고 다시 시도하는 동안 모델 호출 횟수를 세봐. 맞는 답은 한 번이야.
Hint
받아들였지만 안 끝난 턴을 담는 맵이 하나 더 필요해. produce()가 시작하기 전에 채우고, 그 finally에서 비워. 그게 없으면 턴 도중 상태 확인이 unknown으로 답하고, 클라이언트의 재전송이 턴을 두 번 돌려. 콜드 리뷰 2번 지적의 축소판이야.

Progress

Progress is local-only — sign in to sync across devices.
← Previous한 번에 한 턴Next →퀴즈 · 4 questions
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인 — 댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.