Python async contextmanager로 누수·락 해결

비동기 Python을 운영 환경에 올리면, 동기 코드에서는 잘 안 보이던 문제가 빠르게 터집니다. 대표적으로 커넥션/세션 누수, 락 미해제로 인한 교착(Deadlock) 또는 장기 블로킹, 취소(CancelledError) 시 정리 누락이 있습니다.

이 글은 contextlib.asynccontextmanager를 중심으로 “자원 수명 주기”를 코드 구조로 강제해 누수와 락 문제를 줄이는 실전 패턴을 다룹니다. 특히 예외·취소·타임아웃이 섞인 상황에서 try/finally를 매번 손으로 쓰지 않고도 안정적으로 정리되는 구조를 만드는 것이 목표입니다.

관련해서 await 누락은 비동기 버그의 또 다른 주요 원인입니다. 비슷한 결로, 비동기 수명 주기에서 await가 빠지면 정리가 안 되거나 락이 풀리지 않는 문제가 생깁니다. 필요하면 Python async 데코레이터에서 await 누락 버그 잡기도 함께 참고하세요.

왜 async에서 누수·락이 더 자주 터질까

비동기 환경에서 누수와 락이 더 위험한 이유는 다음과 같습니다.

취소가 정상 흐름처럼 자주 발생: 타임아웃, 상위 태스크 취소 전파 등으로 CancelledError가 빈번합니다.
예외 경로가 다양: 네트워크 오류, 재시도 로직, 서킷 브레이커 등으로 예외가 여러 층에서 발생합니다.
동시성으로 증폭: 누수 하나가 “한 번”이 아니라, 동시 요청 수만큼 곱해집니다.

이때 가장 확실한 해결은 “정리 코드가 실행되도록” 기도하는 게 아니라, 언어 문법(async with)으로 정리를 강제하는 것입니다.

`asynccontextmanager` 기본: `try/finally`를 구조화하기

contextlib.asynccontextmanager는 비동기 제너레이터 기반으로 컨텍스트 매니저를 만들 수 있게 해줍니다.

yield 이전: 자원 획득
yield 이후: 자원 해제(반드시 실행되어야 하는 정리)

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def managed_resource(name: str):
    print(f"acquire: {name}")
    try:
        yield {"name": name}
    finally:
        # 예외/취소가 나도 여기로 들어오는 것이 핵심
        print(f"release: {name}")

async def main():
    async with managed_resource("demo") as r:
        await asyncio.sleep(0.1)
        print(r)

asyncio.run(main())

이 패턴이 중요한 이유는 정리를 호출하는 위치가 호출자에게 흩어지지 않기 때문입니다. 호출자는 async with만 지키면 되고, 정리 규칙은 한 군데에 모입니다.

커넥션/세션 누수 방지: DB 트랜잭션을 `async with`로 강제

운영에서 자주 보는 실수는 다음입니다.

커넥션을 열고 예외가 나서 close()가 실행되지 않음
트랜잭션을 시작하고 예외가 나서 rollback()이 실행되지 않음
커밋/롤백 분기가 복잡해져 일부 경로에서 누락

아래는 asyncpg 스타일을 가정한 예시지만, 핵심은 동일합니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def transaction(conn):
    tx = conn.transaction()
    await tx.start()
    try:
        yield conn
    except Exception:
        # 예외 시 롤백을 확실히
        await tx.rollback()
        raise
    else:
        # 정상 종료 시 커밋
        await tx.commit()

# 사용처
async def create_user(conn, user_id: str):
    async with transaction(conn) as c:
        await c.execute("INSERT INTO users(id) VALUES($1)", user_id)

풀(pool)과 함께 쓰기: 커넥션 반환을 강제

풀에서 커넥션을 빌리고 반환하지 않으면, 결국 풀 고갈로 장애가 납니다. 풀에서 빌리는 부분도 컨텍스트로 감싸면 안전합니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def acquire_conn(pool):
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

async def handler(pool):
    async with acquire_conn(pool) as conn:
        async with transaction(conn):
            await conn.execute("SELECT 1")

이렇게 하면 “반환을 잊어버리는” 실수를 구조적으로 제거합니다.

락(뮤텍스/세마포어) 미해제 방지: `asyncio.Lock`를 감싸기

asyncio.Lock는 원래도 async with lock:을 지원합니다. 하지만 실무에서는 다음 요구가 생깁니다.

락 획득에 타임아웃을 걸고 싶다
락 경쟁이 심할 때 로그/메트릭을 남기고 싶다
락을 잡은 채로 오래 실행되면 경고를 띄우고 싶다

이런 부가 기능을 넣으려면 asynccontextmanager가 깔끔합니다.

import asyncio
from contextlib import asynccontextmanager

class LockTimeoutError(RuntimeError):
    pass

@asynccontextmanager
async def timed_lock(lock: asyncio.Lock, timeout_s: float, name: str = "lock"):
    try:
        await asyncio.wait_for(lock.acquire(), timeout=timeout_s)
    except asyncio.TimeoutError as e:
        raise LockTimeoutError(f"timeout acquiring {name} in {timeout_s}s") from e

    try:
        yield
    finally:
        lock.release()

# 사용처
async def update_shared_state(lock, state: dict):
    async with timed_lock(lock, timeout_s=0.5, name="state"):
        state["count"] = state.get("count", 0) + 1
        await asyncio.sleep(0.1)

핵심은 finally에서 release()가 항상 실행된다는 점입니다. 예외는 물론, 상위에서 태스크가 취소되더라도 컨텍스트 종료 과정이 진행되며 정리 코드가 실행될 가능성이 높습니다(단, 이벤트 루프가 강제 종료되는 등 비정상 종료는 별도).

“취소”를 제대로 다루기: `CancelledError`에서 정리 보장하기

비동기에서 가장 까다로운 케이스는 취소 전파입니다. 예를 들어 락을 잡은 뒤 작업 중 취소되면, 락을 풀지 못해 다른 요청이 영원히 대기할 수 있습니다.

asynccontextmanager의 finally는 이 상황에서 큰 안전망이지만, 정리 단계에서 또 await가 필요할 수 있습니다(예: 네트워크로 반납, 세션 종료). 이때 취소가 정리 단계까지 끼어들면 정리가 중단될 수 있습니다.

Python 3.11 이상이면 asyncio.shield() 또는 TaskGroup/timeout과 함께, 정리 구간을 신중히 보호할 수 있습니다. 예시는 다음처럼 “정리 자체는 최대한 짧게” 유지하고, 꼭 필요한 await만 수행하는 식이 좋습니다.

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def safe_close(resource):
    try:
        yield resource
    finally:
        # 정리에서 시간이 걸리면 취소로 끊길 수 있으니 최소화
        # 필요 시 shield로 보호(상황에 따라 남용은 금물)
        await asyncio.shield(resource.aclose())

정리 단계에서 shield를 쓰면 “취소를 무시한다”는 의미가 아니라, 정리 작업만큼은 완료시키고 그 다음 취소를 다시 전파받는 형태로 동작합니다. 다만 정리가 무한정 걸리면 종료가 지연되므로, 정리 작업도 타임아웃을 두는 편이 안전합니다.

실전 패턴 1: Redis 분산락을 누수 없이

분산락은 더 위험합니다. 로컬 락은 프로세스가 죽으면 풀리지만, Redis 락은 TTL/해제 로직이 엉키면 “잠금이 남아” 장애로 이어질 수 있습니다.

아래는 개념 예시입니다(라이브러리별 API는 다를 수 있음).

from contextlib import asynccontextmanager
import uuid

@asynccontextmanager
async def redis_lock(redis, key: str, ttl_ms: int = 30000):
    token = str(uuid.uuid4())

    # SET key value NX PX ttl
    ok = await redis.set(key, token, nx=True, px=ttl_ms)
    if not ok:
        raise RuntimeError(f"lock busy: {key}")

    try:
        yield
    finally:
        # 안전한 해제: 토큰이 내 것일 때만 삭제(Lua 스크립트 권장)
        lua = (
            "if redis.call('get', KEYS[1]) == ARGV[1] then "
            "  return redis.call('del', KEYS[1]) "
            "else return 0 end"
        )
        await redis.eval(lua, 1, key, token)

이 구조를 쓰면,

락 획득/해제 로직이 한 곳에 모이고
호출자는 async with만 지키면 되며
예외/취소가 나도 finally에서 해제를 시도합니다

TTL은 “최후의 안전장치”이지, 해제 로직을 대체하지 않습니다. 해제 누락이 잦다면 TTL을 늘리는 게 아니라, 이런 식으로 구조적으로 해제를 강제해야 합니다.

실전 패턴 2: 파일·임시 디렉터리·스트림 정리

비동기 작업에서 임시 파일을 만들고 업로드/처리하는 경우, 예외가 나면 임시 파일이 쌓입니다. 동기에서는 with open()이 익숙하지만, 비동기 스트림/클라이언트에서도 동일한 수명 주기 강제가 필요합니다.

from contextlib import asynccontextmanager
from pathlib import Path
import aiofiles
import tempfile
import shutil

@asynccontextmanager
async def temp_dir(prefix: str = "job-"):
    d = Path(tempfile.mkdtemp(prefix=prefix))
    try:
        yield d
    finally:
        shutil.rmtree(d, ignore_errors=True)

async def write_temp_file():
    async with temp_dir() as d:
        p = d / "data.txt"
        async with aiofiles.open(p, "w") as f:
            await f.write("hello")
        # 여기서 예외가 나도 디렉터리는 정리됨

흔한 함정: `async with` 안에서 또 다른 `await` 누락

컨텍스트 매니저를 잘 만들어도, 사용처에서 await를 빼먹으면 정리 보장이 깨지거나(혹은 아예 실행이 안 되거나) 타이밍이 꼬일 수 있습니다. 특히 “정리 함수가 코루틴인데 호출만 하고 await를 안 함” 같은 케이스가 대표적입니다.

정리 코드는 반드시 finally에서 await까지 포함해 완료시키기
호출자는 async with로 감싸고, 컨텍스트 생성/해제 과정에서 비동기 호출이 있다면 빠짐없이 await

이 주제는 비동기 데코레이터에서도 동일하게 나타납니다. 자세한 사례는 Python async 데코레이터에서 await 누락 버그 잡기를 참고하면 디버깅 감이 빨리 옵니다.

디버깅/관측 포인트: 누수·락 문제를 빨리 찾는 법

컨텍스트 매니저 패턴을 도입했다면, 다음 관측 포인트를 함께 넣으면 효과가 큽니다.

락 대기 시간: 락 획득 전후 시간을 재서 일정 임계치 초과 시 경고
풀 사용량: 커넥션 풀의 in_use/available 지표를 수집
취소율: 타임아웃으로 인한 취소가 급증하면 정리 코드가 압박받음

분산 환경이라면 OpenTelemetry로 트레이싱을 붙여 “어떤 요청이 락을 오래 잡는지”를 찾는 것도 좋습니다. 서비스 전반의 병목을 추적하는 방법은 OpenTelemetry로 MSA 분산 트랜잭션 추적 실전 글이 도움이 됩니다.

체크리스트: 운영 코드에 적용할 때

자원 획득/해제는 가능한 한 async with로 감싼다
asynccontextmanager 내부는 try/finally를 기본으로 하고, 예외 시 롤백/해제를 명시한다
정리 단계의 await는 짧고 확실하게(필요 시 shield + 타임아웃)
락은 “획득 성공 후”에만 해제하도록 구조화한다(획득 실패 경로에서 release() 호출 금지)
분산락은 토큰 검증을 포함한 안전 해제(Lua 등)로 구현한다

마무리

contextlib.asynccontextmanager는 단순히 문법 설탕이 아니라, 비동기 시스템에서 가장 흔한 장애 원인인 자원 누수와 락 미해제를 “코드 구조”로 제거하는 도구입니다. async with를 경계로 자원의 생애를 명확히 자르고, 정리 로직을 한 곳에 모으면 예외·취소·타임아웃이 뒤섞여도 안정성이 크게 올라갑니다.

다음 단계로는, 락/풀/취소율 같은 지표를 계측하고(가능하면 트레이싱까지) “누가 오래 잡고 있는지”를 관측 가능하게 만드는 것을 추천합니다.

왜 async에서 누수·락이 더 자주 터질까

asynccontextmanager 기본: try/finally를 구조화하기

커넥션/세션 누수 방지: DB 트랜잭션을 async with로 강제