본문 바로가기
C.W.K.
Lesson 06 of 06 · published

순간의 이벤트 스트림과 엔진의 모양

~14 min · server-sent-events, asyncio, launchd, deployment

Level 0첫 단어
0 XP0/47 lessons0/13 achievements
0/100 XP to next level100 XP to go0% complete

빠른 것부터 보여주려면 밀어 줘야 해

Spark 불변식 중 하나가 빠른 것부터야. 층마다 생기는 즉시 보여줘. 피파가 메뉴를 읽는 답은 한 줄씩 흘러 들어오고, 엔진은 줄이 파싱되는 즉시 result.add로 기록해. 폰이 몇 초마다 물어보는 방식이면 그 줄들은 결국 뭉텅이로 도착해. 그래서 엔진이 밀어 줘. 순간을 가리키는 op는 그 순간을 지켜보는 쪽으로 곧장 가.

하나뿐인 문에서 내보내기

시작점은 이미 봤어. Corpus.record의 마지막 단계야. moment_id를 단 op, 아니면 그 자체가 moment.*인 op는 그 순간 이름으로 {"type": op 이름, "record": 본문} 모양으로 나가. 이 규칙은 글자 그대로라서 빈틈이 하나 있어. mark.remove와 link.remove는 표시나 링크 자기 id만 들고 있어. 그래서 표현 남기기를 취소하거나 크럼 링크를 끊어도 지켜보는 쪽엔 아무것도 안 가. 클라이언트는 순간을 다시 읽을 때에야 그 변화를 봐. 턴 op는 특별해. 미러가 들고 있는 모양 그대로, 상태까지 담아서 나가. 어떤 턴은 엔진이 알아서 시작하거든. 버전 하나가 끝나면 따라붙는 콜드 리뷰 같은 거. 클라이언트는 그런 턴을 이 이벤트로만 알게 돼.

일부러 코퍼스에 안 넣는 이벤트도 몇 개 있어. Corpus.publish는 지켜보는 쪽에 일시적인 이벤트를 보내고 아무것도 기록하지 않아. 실용 번역 턴은 이걸로 turn.prose를 보내. 구조화된 결과 둘레에서 자라나는 말, 그러니까 피파가 아빠한테 하는 말이야. 쇼 카드 통역은 진행 상황을 exchange.* 이벤트로 보내고. 이벤트를 하나 놓쳐도 클라이언트가 잃는 건 진행 상황뿐이야. 실용 번역 턴은 끝날 때 다 쓴 말을 따로 기록해.

프로세스 하나, 루프 여럿

퍼뜨리는 쪽은 첫 번째 코드 블록의 작은 프로세스 내부 클래스 Broadcaster야. 순간 id마다 구독자 집합을 들고 있고, 구독자 하나는 짝 하나야. 그 구독자를 만든 이벤트 루프와 asyncio.Queue.

재밌는 건 이 짝이야. record는 작업 스레드에서 돌아. 코퍼스 쓰기는 asyncio.to_thread로 이벤트 루프 밖에 밀려나 있으니까. 그런데 asyncio 큐는 스레드 안전하지 않아. 작업 스레드가 거기 직접 넣으면 안 돼. 그래서 publish는 평범한 스레드 잠금 아래서 구독자 목록을 복사하고, 배달은 구독자 자기 루프한테 call_soon_threadsafe로 넘겨. 그 루프가 이미 닫혔으면 RuntimeError가 나고, 그 구독자는 조용히 빠져.

브로커도, Redis도, 메시지 버스도 없어. 그걸로 충분한 건 쓰는 쪽이 엔진 프로세스 하나뿐이라서야. 생길 수 있는 이벤트는 전부 그 안에서 태어나.

스트림 그 자체

두 번째 코드 블록이 라우트 GET /api/moments/{id}/events 전부야. Server-Sent Events로 나가.

  1. 모르는 순간이면 스트림을 열기 전에 404로 답해.
  2. 구독부터 해. 그러고 나서 순간의 스냅숏을 읽어서 첫 이벤트로 보내. 순서가 반대면, 읽기와 구독 사이에 기록된 op는 어디에도 안 실려. 이 순서면 양쪽에 다 실릴 수는 있는데, 클라이언트가 결과를 id로 맞추니까 상관없어.
  3. 그 뒤로는 오는 대로 이벤트를 넘겨. 15초 동안 조용하면 : keepalive를 보내. 클라이언트가 무시하는 SSE 주석이고, 연결이 놀고 있는 것처럼 보이지 않게 해 줘.
  4. 클라이언트가 떠나면 생성기가 닫히고, finally가 구독을 풀어.

응답 헤더 두 개가 중요해. Cache-Control: no-cache, 그리고 중간 프록시가 이벤트를 버퍼에 붙잡아 두지 않게 하는 X-Accel-Buffering: no. 테스트 하나가 순간을 구독하고, 작업 스레드에서 결과를 기록하고, turn.prose를 하나 보내고 나서, 둘이 순서대로 도착하는지 확인해. 구독을 푼 뒤엔 지켜보는 쪽이 아무도 없는지도 보고.

브라우저의 EventSource는 알아서 다시 붙어. 폰은 Swift에서 같은 라우트를 열어. 폰은 테일넷으로만 엔진에 닿으니까, 폰에서 순간 스트림이 끊기면 아빠한테 Tailscale을 확인하라고 알려줘. 웹도 실패한 요청에 같은 안내를 붙이고.

엔진의 모양

관심사Spark의 방식
프로세스launchd LaunchAgent 하나, com.cwk.spark-serve. python -m backend.serve를 RunAtLoad, KeepAlive, 10초 스로틀로 돌리고, 출력은 데이터 루트 아래 serve.log로 가
듣는 곳127.0.0.1에서 uvicorn. 포트는 가족 레지스트리가 Spark한테 준 번호(7900)
닿는 길Tailscale Serve가 그 루프백 포트 위에 HTTPS 쌍둥이(7901)를 열어. 레지스트리가 HTTP 쌍둥이를 선언했으면 그것도
형제 서비스cwkPippa, 가족 제어 창구, cwkBellows. 전부 레지스트리가 준 루프백 주소로 불러
웹 클라이언트빌드한 프런트엔드를 맨 마지막에 붙여. 그래야 /api 라우트가 페이지 대체 경로보다 먼저 이겨
예약 작업없어. 데이터 루트는 가족의 매일 백업이 맡아

루프백 규칙의 이유는 설정 모듈 독스트링에 있어. 그 모듈엔 LAN 호스트 이름을 하나도 적으면 안 돼. launchd 에이전트는 macOS 로컬 네트워크 관문을 못 넘거든. 엔진은 와일드카드 주소에서 듣는 일이 없고, 테일넷은 Serve를 거쳐서만 엔진에 닿아. 포트, 주소, 데이터 루트는 전부 가족 레지스트리에서 오고, 환경 변수는 명시적으로 덮어쓸 때만 써. 저장소에 든 plist조차 템플릿이야. @@HOME@@, @@DATA_ROOT@@ 같은 토큰이 있고, 설치 스크립트가 자기가 도는 Mac에 맞춰 채워.

어디 있는지부터 확인하는 설치

설치 스크립트는 가족 킷 템플릿에서 생성돼. 모든 엔진이 같은 모양이야(공유 패키지보다 복사본을 들이는 게 나은 이유는 Sibling Kit Quest가 다뤄). 세 번째 코드 블록이 그 첫 동작이야. Spark가 이 Mac에 속하는지 레지스트리한테 묻고, 아니면 더 가지 않아. 통과하면 conda 환경과 프런트엔드를 짓고, plist를 채우고, 예전 작업을 내리고 그 종료 제한 시간이 지날 때까지 기다려. 새 작업은 재시도를 곁들여 올리고, /api/health를 2초 간격으로 열다섯 번까지 찔러 봐. 마지막으로 두 번째 스크립트로 테일넷 쌍둥이를 여는데, 그 스크립트도 Tailscale을 건드리기 전에 레지스트리한테 같은 질문을 해.

빠진 한 줄에 가족이 비싸게 배운 교훈이 담겨 있어. 올린 뒤에 launchctl kickstart -k가 없다는 거야. RunAtLoad 작업은 올리자마자 이미 시작돼. 예전엔 그 뒤에 kickstart가 따라붙었는데, 그게 첫 부팅을 죽였어. 데이터베이스 마이그레이션 도중이었을 수도 있고.

원칙: 서비스가 어느 기계에서 도는지는 다른 누군가가 소유하는 사실로 둬. Spark는 포트, 데이터 루트, 이웃, 심지어 설치 허락까지 레지스트리에서 읽어. 그래서 옮기는 건 레지스트리의 결정이지 코드 수정이 아니야.

Code

cwkSpark, backend/events.py: 구독자마다 자기 루프에서 배달해 (원문 그대로, 45-52행)·python
    def publish(self, key: str, event: dict[str, Any]) -> None:
        with self._lock:
            watchers = list(self._subscribers.get(key, ()))
        for loop, queue in watchers:
            try:
                loop.call_soon_threadsafe(queue.put_nowait, event)
            except RuntimeError:
                self.unsubscribe(key, (loop, queue))
cwkSpark, backend/app.py: 지금의 순간, 이어서 모든 변화 (원문 그대로, 721-751행)·python
    @app.get("/api/moments/{moment_id}/events")
    async def moment_events(moment_id: str) -> StreamingResponse:
        """The moment as it is, then every change as it happens (SSE)."""
        if await asyncio.to_thread(corpus.moment, moment_id, include_deleted=True) is None:
            raise HTTPException(status_code=404, detail="moment not found")

        async def stream() -> AsyncIterator[bytes]:
            # Subscribe before the snapshot so nothing falls between them;
            # clients key results by id, so a repeat is harmless.
            subscriber = corpus.events.subscribe(moment_id)
            try:
                snapshot = await asyncio.to_thread(
                    corpus.moment, moment_id, include_deleted=True
                )
                yield _sse({"type": "snapshot", "moment": snapshot})
                queue = subscriber[1]
                while True:
                    try:
                        event = await asyncio.wait_for(queue.get(), timeout=15)
                    except asyncio.TimeoutError:
                        yield b": keepalive\n\n"
                        continue
                    yield _sse(event)
            finally:
                corpus.events.unsubscribe(moment_id, subscriber)

        return StreamingResponse(
            stream(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )
cwkSpark, the service install script in scripts/: 레지스트리가 Spark를 여기 두지 않았으면 거절해 (원문 그대로, 13-21행)·shell
# This script lives in the repository's scripts/; the repository is its member
# id in the family registry, which owns the engine's port and tailnet URLs.
REPO="${0:A:h:h}"
# Only on the Mac the family registry places this member on, including a
# machine's host profile override (cwkPippa todo #549, Pippa's Ark).
/usr/bin/python3 "$REPO/scripts/kit_family_endpoints.py" runs-here "${REPO:t}" || {
  echo "${0:t} runs only where the family registry places ${REPO:t}" >&2
  exit 1
}

External links

Exercise

직접 만드는 서비스에 실시간 화면을 붙여 봐. 키마다 구독자를 두는 방송기를 짜는데, publish는 어느 스레드에서 불러도 되고 call_soon_threadsafe로 구독자 루프에 배달하게 해. 이걸 Server-Sent Events 라우트로 내보내. 스냅숏을 읽기 전에 구독하고, 스냅숏을 첫 이벤트로 보내고, 15초 조용하면 주석 한 줄을 보내. 테스트는 둘이야. 하나는 작업 스레드에서 보내고 도착 순서를 확인해. 다른 하나는 구독과 스냅숏 사이에 쓰기를 끼워 넣고, id로 합친 뒤 클라이언트에 딱 한 벌만 남는 걸 보여줘. 마지막으로 서비스 설치 스크립트 첫 단계에, 설정상 이 서비스가 배정되지 않은 기계에선 돌기를 거부하는 검사를 넣어.
Hint
구독자는 구독할 때 잡은 (루프, 큐) 짝으로 저장해. 구독 해제는 finally 블록에서 해. 그래야 연결이 닫히면 늘 정리돼. 닫힌 루프에서 난 RuntimeError는 그 구독자를 빼는 신호로 써.

Progress

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

댓글 0

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

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