연결이 끊긴 뒤에도 남는 running: 스트리밍 실행의 종료를 설계하는 법

스트리밍 API는 정상 경로가 간단해 보인다. 외부 연산을 시작하고, 결과 조각을 전송하고, 마지막에 실행 상태를 succeeded로 바꾸면 된다.

하지만 사용자가 응답 중간에 탭을 닫으면 이야기가 달라진다. HTTP 연결은 종료됐는데 모델 호출이 계속되거나, 생성기는 닫혔는데 실행 기록은 running으로 남을 수 있다. 완료 이벤트를 받은 화면과 데이터베이스의 최종 상태가 서로 다를 수도 있다.

이 문제를 해결하려면 연결, 실행, 영속 기록의 수명을 각각 정의하고 그 사이의 종료 책임을 연결해야 한다. 이 글은 Python 비동기 생성기의 최소 재현에서 출발해, 취소 처리와 트랜잭션 경계, 다중 워커의 실행 소유권까지 확장한다.

예제는 Python 3.12.13에서 검증했다. AnyIO와 SQLAlchemy 관련 설명은 각각 4.x 및 2.0 문서를 기준으로 한다. 뒤쪽의 작업 리스와 이벤트 저장 구조는 설계 예시이며, 앞부분의 재현 실험과 구분한다.

1. 먼저 네 가지 수명을 분리한다

네 수명을 항상 일치시킬 필요는 없다. 대신 제품 동작에 따라 연결 규칙을 정해야 한다.

• 요청에 종속된 실행: 연결 단절을 취소 신호로 사용한다. 정리 책임은 요청을 처리하는 실행자에게 있다.
• 계속 진행해야 하는 작업: 연결 단절은 구독 해제로 처리한다. 실행 책임은 요청과 독립된 실행자가 가진다.

여기서 판단이 흐리면 같은 프로그램에 상반된 코드가 생긴다. 한 계층은 연결 종료 시 태스크를 취소하고, 다른 계층은 완료 알림을 기다린다. 둘 다 개별적으로는 자연스러운 코드지만, 합쳐진 시스템의 계약은 성립하지 않는다.

2. async for를 빠져나왔다고 상류 생성기가 즉시 닫히지는 않는다

중첩된 스트리밍은 대개 다음과 같이 구성된다.

외부 응답 생성기 → 실행 이벤트 생성기 → HTTP 응답 생성기

이때 각 계층이 단순히 async for로 읽고 yield만 하면, 바깥 생성기의 종료가 안쪽 생성기의 명시적인 종료로 연결되지 않을 수 있다. 특히 바깥쪽이 값을 전달한 직후 멈춰 있는 경우가 중요하다.

다음 예제는 상류 자원의 정리 시점을 검증한다.

import asyncio
from contextlib import aclosing


async def upstream(closed):
    try:
        yield "first chunk"
        await asyncio.Event().wait()
    finally:
        closed.append("upstream closed")


async def relay(source):
    async with aclosing(source) as chunks:
        async for chunk in chunks:
            yield chunk


async def check():
    closed = []
    response = relay(upstream(closed))

    assert await anext(response) == "first chunk"
    await response.aclose()

    # 이벤트 루프 종료나 GC를 기다리기 전에 정리가 끝나야 한다.
    assert closed == ["upstream closed"]


asyncio.run(check())

aclosing()을 제거한 별도 재현에서는 상류 생성기의 참조를 유지한 채 바깥 생성기만 닫았을 때 정리 목록이 비어 있었다. aclosing()을 적용하면 response.aclose()가 돌아오기 전에 상류의 정리가 끝났다. 소비 태스크를 cancel()한 경우에도 정리와 취소 전파를 함께 확인했다.

contextlib.aclosing()은 블록을 벗어날 때 비동기 생성기의 aclose()를 호출한다. 정리 시점을 반복문과 같은 실행 문맥에 묶어 준다는 점이 핵심이다. Python contextlib 공식 문서

이 규칙은 스트리밍 어댑터마다 적용해야 한다. 직접 만든 생성기뿐 아니라 SDK가 반환한 스트림도 누가 닫는지 정해야 한다. HTTP 프레임워크가 가장 바깥 응답을 닫아 준다는 사실만으로 모든 상류 자원의 수명이 보장되지는 않는다.

3. 오류와 취소는 서로 다른 제어 흐름이다

실행 기록을 다음처럼 처리하는 코드를 생각해 보자.

try:
    result = await execute()
except Exception:
    record_failure()
else:
    record_success(result)

Python 3.12에서 asyncio.CancelledError는 BaseException의 하위 타입이다. 따라서 위의 except Exception은 취소를 처리하지 않는다. 비동기 생성기를 aclose()하면 정지한 지점에는 GeneratorExit이 들어간다. 취소와 명시적인 생성기 닫기를 별도로 검증해야 하는 이유다. asyncio 태스크 취소 문서, 비동기 생성기 메서드 문서

실행 상태 기록에는 적어도 다음 불변식이 필요하다.

1. 정상적으로 정리된 실행은 종결 상태를 가진다.
2. 성공으로 확정한 실행을 뒤늦은 연결 종료가 stopped로 바꾸지 않는다.
3. 부분 응답이 있다면 중지 기록과 함께 보존한다.
4. 취소 신호는 정리를 마친 뒤 상위 계층으로 전파한다.

단일 실행자 안에서는 finalized 같은 지역 변수가 중복 마감을 막는 데 도움이 된다. 성공 또는 실패를 커밋한 뒤 이를 참으로 바꾸고, 마지막 정리 경로에서 아직 마감하지 않은 실행을 처리하는 방식이다.

다만 finally에 도달했다는 이유만으로 모든 경우를 취소로 분류하면 안 된다. 종료 기록 자체의 DB 오류, 프로세스 중단, 외부 작업의 결과 불명확 상태는 별도로 다뤄야 한다. 지역 변수 하나로 해결되는 범위는 같은 실행자 내부의 제어 흐름이다.

4. finally 안의 await도 취소될 수 있다

정리 코드를 finally로 옮기면 다음 문제가 보인다.

finally:
    await save_terminal_state()

이 await는 어떤 취소 정책 아래에서 실행되는가?

AnyIO의 취소 스코프 안에서는 취소 상태가 유지되는 동안 대기 지점마다 취소 예외가 다시 발생할 수 있다. 따라서 비동기 종료 기록에는 제한된 정리 구간이 필요할 수 있다. move_on_after(..., shield=True)처럼 외부 스코프의 취소를 잠시 막되 자체 제한 시간을 두는 방식이다. 이 보호는 AnyIO 스코프의 의미에 따른 것이며, 임의의 asyncio.Task.cancel()까지 무조건 차단한다고 가정해서는 안 된다. AnyIO 취소·정리 문서

정리 구간에서는 다음을 구분하는 편이 좋다.

• 자원 정리: 소켓·응답 스트림·임시 파일을 닫는다.
• 상태 마감: 실행 결과를 영속 저장한다.
• 복구 대상 표시: 제한 시간이나 DB 장애로 마감하지 못했다면 복구 경로가 이를 발견할 수 있게 한다.

작업 전체를 취소로부터 보호하면 사용자의 중지 요청이 무력해진다. 반대로 보호 구간을 무제한으로 두면 종료 처리가 끝나지 않는다. 보호 범위는 짧게 두고, 마감 실패는 관측 가능한 오류로 남겨야 한다.

동기 DB 호출을 쓰는 경우에는 Python 수준의 await 취소가 끼어들지 않더라도 쿼리가 이벤트 루프를 오래 막을 수 있다. 이 경우에도 DB 연결·쿼리 제한 시간과 프로세스 종료 정책이 필요하다.

5. 완료를 보냈다는 사실과 완료를 저장했다는 사실

다음 순서는 실패 구간을 만든다.

완료 이벤트 전송 → DB 커밋

이벤트 전송 직후 프로세스가 종료되면 사용자는 완료를 봤지만, 새로고침한 화면에서는 실행 중으로 보일 수 있다.

종결 결과를 다시 조회할 수 있어야 한다면 다음 순서가 더 다루기 쉽다.

결과·종결 상태 저장 → 커밋 → 완료 이벤트 전송

이 순서에서도 커밋 직후 연결이 끊길 수 있다. 하지만 이미 저장한 결과를 재조회할 수 있다. 전달 실패를 복구할 수 있는 방향으로 실패 구간을 옮긴 것이다.

SQLAlchemy를 사용할 때는 flush()와 commit()도 구분해야 한다. flush()는 변경 SQL을 현재 트랜잭션에 반영하는 단계다. 해당 변경이 최종 커밋되었다는 뜻은 아니다. 같은 세션에서 갱신된 객체가 보이는 것만으로 저장 검증을 끝내면 안 된다. SQLAlchemy 세션·트랜잭션 문서

검증은 다른 연결의 새 트랜잭션에서 수행한다. 생산자 세션이 아직 열려 있어도 종결 상태, 종료 시각, 부분 응답이 조회돼야 한다. 오래 열린 읽기 트랜잭션은 격리 수준에 따라 이전 스냅샷을 볼 수 있으므로 테스트의 읽기 경계도 명확해야 한다.

장시간 외부 호출을 DB 트랜잭션 하나로 감싸는 설계도 주의해야 한다. 실행 시작, 진행 체크포인트, 종료를 짧은 트랜잭션으로 기록하면 연결·락 점유 시간을 줄이고 프로세스 중단 뒤의 복구 기준을 만들기 쉽다.

6. 다중 워커에서는 종료 권한까지 저장해야 한다

실행자를 하나 더 늘리면 지역 finalized 변수만으로는 부족하다.

워커 A: 작업 인수 → 긴 지연
복구기: A의 리스 만료 판단
워커 B: 같은 작업 재인수 → 완료 저장
워커 A: 뒤늦게 복귀 → 오래된 결과 저장 시도

이 문제는 “누가 마지막으로 썼는가”로 해결하면 안 된다. 현재 실행 권한을 가진 시도만 저장할 수 있어야 한다.

한 가지 설계는 실행 기록에 다음 값을 두는 것이다.

작업 재인수는 원자적으로 수행하며 세대 번호를 증가시킨다. 실행자는 자신이 받은 번호를 가지고 저장한다. 예를 들어 종결 갱신은 다음 조건을 갖는다. SQL은 바인딩 변수 표기를 사용하는 개념 예시다.

UPDATE execution_run
SET state = :terminal_state,
    finished_at = CURRENT_TIMESTAMP,
    result_ref = :result_ref
WHERE run_id = :run_id
  AND execution_epoch = :my_epoch
  AND state = 'running'
  AND lease_until > CURRENT_TIMESTAMP;

갱신 건수가 0이면 종결 권한을 잃었거나 이미 상태가 바뀐 것이다. 그 실행자는 종결 알림이나 후속 결과 포인터를 새로 기록해서는 안 된다. 진행 이벤트와 체크포인트에도 같은 권한 규칙이 필요하다. 파일을 먼저 저장했다면 시도별 경로를 사용하고, 최종 결과 포인터는 권한 검증에 성공한 트랜잭션에서 연결하는 편이 안전하다.

리스 갱신과 만료 판단에는 가능한 한 DB 시각을 일관되게 사용한다. 프로세스별 시계 차이로 실행 소유권이 흔들리는 일을 줄일 수 있다. 세대 번호는 오래된 실행자의 쓰기를 막는 펜싱 토큰 역할을 한다.

그러나 이 SQL이 외부 부수 효과까지 한 번만 실행되게 하지는 않는다. A가 외부 요청을 성공시킨 뒤 결과 저장 전에 멈추었다면 B의 재시도가 같은 효과를 일으킬 수 있다. 외부 API의 멱등성 키, 효과 실행 전후의 단계 기록, 결과 조회 또는 보상 작업을 별도로 설계해야 한다. 멱등성 키도 재시도마다 달라지는 세대 번호와 분리하고, 같은 논리적 효과에는 같은 키를 사용해야 한다.

7. 계속 실행할 작업은 구독 연결에서 독립시킨다

탭을 닫아도 계속 실행해야 한다면 다음 세 경로를 분리할 수 있다.

명령: 실행 요청 → 요청 식별자 중복 검사 → 작업 저장
실행: 작업 인수 → 단계 실행 → 상태·결과·이벤트 저장
관측: 저장된 진행 내역 조회 + 실시간 이벤트 구독

여기서 create_task() 호출만 추가하면 프로세스 내부의 동시 실행은 만들 수 있다. 하지만 프로세스 재시작 뒤의 작업 복구, 실행 인수, 재시도 기준까지 자동으로 생기지는 않는다.

구독을 다시 열었을 때도 마지막으로 받은 이벤트 이후를 복원할 수 있어야 한다. SSE에는 id 필드와 재연결 시 Last-Event-ID를 보내는 표준 동작이 있다. 하지만 서버의 이벤트 보관과 재생은 애플리케이션이 구현해야 한다. 또한 브라우저의 기본 EventSource가 아닌 fetch 기반 스트림 클라이언트에서는 재연결과 커서 전달을 직접 구현해야 한다. WHATWG Server-sent events 표준

영속 이벤트를 설계한다면 (run_id, sequence)를 기준으로 중복을 제거하고, 같은 실행 안의 순서를 복원할 수 있다. 시도 번호도 함께 저장하면 재시도 전후의 기록을 구별하기 쉽다. 단, 증가하는 숫자를 할당했다고 커밋 순서까지 보장되지는 않는다. 여러 작성자가 있으면 실행별 직렬화나 커밋 순서를 고려한 커서 규칙이 필요하다.

모든 토큰을 영속 이벤트로 남기는 것도 필수는 아니다. 단계 시작·종료·결과는 저장하고, 토큰은 일정 크기의 청크 또는 주기적 스냅샷으로 합치는 선택지가 있다. 필요한 재생 정확도와 쓰기 비용을 함께 결정하면 된다.

8. 완료 알림은 저장된 상태에서 파생시킨다

완료 상태 저장과 알림 전송 사이에도 실패 구간이 있다.

• 상태 저장 후 알림 전송 전 장애: 완료했는데 알림이 없다.
• 알림 전송 후 전달 확인 저장 전 장애: 재시도하면서 같은 알림이 반복된다.

한 가지 방법은 종결 상태를 바꾸는 트랜잭션 안에서 알림 발송용 레코드도 함께 저장하는 것이다. 별도 전달자가 이 레코드를 읽어 알림을 보내고 전달 결과를 기록한다. 종결 갱신이 앞의 소유권 조건을 통과했을 때만 알림 레코드를 생성해야 한다.

전송 자체는 반복될 수 있으므로 알림에는 안정적인 event_id를 두고 수신 측에서 중복을 제거한다. 같은 실행의 종결 알림을 유일하게 만드는 제약도 도움이 된다. 이 방식으로 상태와 알림 의도를 함께 보존할 수 있지만, 외부 알림 채널에 대한 정확히 한 번 전달을 보장하는 것은 아니다.

화면의 알림 문구도 이 계약을 반영해야 한다. “완료” 알림은 저장된 결과를 열 수 있어야 하고, “중지”는 외부 시스템의 모든 변경이 취소됐다는 뜻으로 사용하면 안 된다. 작업 진행 중 이미 발생한 효과와 응답 생성 중단은 다른 사실이다.

9. 정상 응답보다 종료 경계를 테스트한다

검증 단위를 HTTP 200으로 잡으면 이 문제들을 놓치기 쉽다. 아래와 같이 실패를 주입하는 위치와 이후에 성립해야 하는 상태를 한 쌍으로 적는 편이 효과적이다.

이 글의 최소 재현에서는 첫 청크 직후 닫기, aclosing()에 의한 즉시 정리, 대기 중 태스크 취소 전파를 직접 확인했다. DB를 사용하는 별도의 통합 검증에서는 스트림을 닫은 직후 독립 연결로 중지 상태와 부분 응답, 종료 시각이 커밋됐는지 확인했다. 리스 재인수와 알림 재전송 항목은 제안한 설계를 실제 구현할 때 추가해야 할 실패 주입 시나리오다.

운영 지표도 평균 응답 시간만으로는 부족하다. 리스가 만료된 실행 수, 종결 기록 실패 수, 종결 상태와 알림 전달의 시간 차이, 오래된 실행 세대의 쓰기 거부 횟수 같은 값이 복구 경로의 건강 상태를 보여 준다.

마무리

스트리밍 시스템에서 종료는 여러 단계로 이루어진다. 소비를 멈추고, 상류 자원을 정리하고, 실행 상태를 확정하고, 그 사실을 사용자에게 전달한다.

설계 검토에서는 다음 세 질문을 먼저 확인할 만하다.

1. 이 작업을 끝낼 책임은 현재 누구에게 있는가?
2. 그 실행자가 사라져도 다음 실행자가 판단할 기록이 남는가?
3. 사용자에게 보낸 최종 상태를 다시 조회해도 같은 결과를 얻는가?

이 세 가지가 명확하면 취소는 예외 처리 한 줄에 묻히지 않는다. 정상 완료와 같은 수준으로 검증할 수 있는 실행 프로토콜이 된다.