Python Decorator로 async 재시도·백오프 구현

비동기 I/O가 많은 Python 서비스에서 네트워크 호출은 실패를 전제로 설계하는 편이 안전합니다. DNS 일시 장애, 연결 리셋, 원격 서버의 429/503 응답처럼 일시적(transient) 실패는 재시도로 회복될 가능성이 큽니다. 문제는 재시도 로직이 코드 곳곳에 흩어지면 정책(최대 횟수, 백오프, 지터, 예외 분류)이 제각각이 되어 장애 시 행동이 예측 불가능해진다는 점입니다.

이 글에서는 Python Decorator로 async 함수에 재시도·백오프를 일관되게 적용하는 방법을 구현 중심으로 정리합니다. 또한 asyncio 환경에서 특히 중요한 취소(cancellation) 전파, 타임아웃, 관측성(로그/메트릭), 동시성 폭주 방지까지 함께 다룹니다.

관련해서 API 레이트리밋 상황의 재시도 설계는 OpenAI 429/RateLimitError 재시도·백오프·큐 설계도 함께 참고하면 정책을 더 정교하게 잡는 데 도움이 됩니다.

왜 데코레이터로 재시도를 묶어야 하나

async 코드에서 재시도는 보통 아래 이유로 필요합니다.

원격 API의 일시 장애: 503, 게이트웨이 타임아웃, 연결 리셋
레이트리밋: 429 또는 SDK의 전용 예외
간헐적 네트워크 흔들림: TimeoutError, ConnectionError

하지만 무작정 재시도하면 더 큰 문제를 만들 수 있습니다.

동시 요청이 몰린 상태에서 재시도 폭풍(retry storm) 발생
모든 예외를 재시도하다가 치명적 버그를 숨김
asyncio.CancelledError까지 잡아버려 취소가 전파되지 않음

따라서 재시도는 “공통 정책”으로 묶고, 호출부에서는 “업무 로직”만 남기는 편이 유지보수에 유리합니다.

재시도 설계 체크리스트

구현에 앞서 정책을 먼저 정리합니다.

1) 어떤 실패를 재시도할 것인가

재시도 대상: 네트워크 일시 오류, 타임아웃, 레이트리밋, 5xx
재시도 금지: 인증 실패(401), 권한(403), 잘못된 요청(400), 스키마 오류 등

HTTP 클라이언트(httpx, aiohttp)를 쓴다면 상태코드 기반 + 예외 타입 기반을 함께 쓰는 게 일반적입니다.

2) 백오프는 어떻게 할 것인가

지수 백오프(exponential backoff): base * 2^attempt
최대 대기 상한(cap): 너무 길어지지 않도록 제한
지터(jitter): 여러 인스턴스가 동시에 재시도하지 않도록 랜덤 분산

3) 취소와 타임아웃

asyncio.CancelledError는 반드시 다시 raise 해서 취소를 전파
“각 시도당 타임아웃”과 “전체 작업 타임아웃”을 구분

4) 관측성

재시도 횟수, 마지막 예외, 총 대기시간
성공/실패 메트릭(예: Prometheus counter)

기본 형태: async 재시도·백오프 데코레이터

아래 코드는 표준 라이브러리만으로 동작하는 범용 데코레이터입니다.

retry_on으로 재시도할 예외 타입 지정
지수 백오프 + 지터
CancelledError 전파
on_retry 콜백으로 로그/메트릭 연결

import asyncio
import random
import time
from dataclasses import dataclass
from typing import Awaitable, Callable, Optional, Tuple, Type


@dataclass(frozen=True)
class RetryState:
    attempt: int
    max_attempts: int
    delay_seconds: float
    elapsed_seconds: float
    exc: BaseException


def async_retry(
    *,
    max_attempts: int = 5,
    base_delay: float = 0.2,
    max_delay: float = 5.0,
    jitter: float = 0.2,
    retry_on: Tuple[Type[BaseException], ...] = (OSError, TimeoutError),
    on_retry: Optional[Callable[[RetryState], None]] = None,
) -> Callable[[Callable[..., Awaitable]], Callable[..., Awaitable]]:
    """async 함수에 재시도 + 백오프를 적용하는 데코레이터.

    - jitter: 0.2면 delay의 +/-20% 범위로 랜덤 변동
    """

    def decorator(func: Callable[..., Awaitable]):
        async def wrapper(*args, **kwargs):
            start = time.monotonic()

            for attempt in range(1, max_attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except asyncio.CancelledError:
                    # 취소는 재시도하지 않고 즉시 전파
                    raise
                except retry_on as exc:
                    if attempt >= max_attempts:
                        raise

                    raw_delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
                    # 지터 적용: delay * (1 - jitter) ~ delay * (1 + jitter)
                    factor = 1.0 + random.uniform(-jitter, jitter)
                    delay = max(0.0, raw_delay * factor)

                    if on_retry is not None:
                        state = RetryState(
                            attempt=attempt,
                            max_attempts=max_attempts,
                            delay_seconds=delay,
                            elapsed_seconds=time.monotonic() - start,
                            exc=exc,
                        )
                        on_retry(state)

                    await asyncio.sleep(delay)

        return wrapper

    return decorator

사용 예시

import asyncio


def log_retry(state):
    print(
        f"retry attempt={state.attempt}/{state.max_attempts} "
        f"delay={state.delay_seconds:.2f}s elapsed={state.elapsed_seconds:.2f}s "
        f"exc={type(state.exc).__name__}: {state.exc}"
    )


@async_retry(max_attempts=4, base_delay=0.3, max_delay=3.0, on_retry=log_retry)
async def fetch_something():
    # 예시: 간헐적 실패를 흉내
    raise TimeoutError("upstream timeout")


async def main():
    try:
        await fetch_something()
    except Exception as e:
        print("final failure:", repr(e))


asyncio.run(main())

이 패턴의 장점은 호출부가 깔끔해지고, 재시도 정책을 한 곳에서 통제할 수 있다는 점입니다.

HTTP 상태코드 기반 재시도: httpx 예시

네트워크 예외만으로는 부족합니다. HTTP 호출은 정상적으로 응답이 와도 429/503처럼 “재시도 가치가 있는 상태코드”가 있습니다.

아래는 httpx를 사용하는 예시입니다.

import httpx


class RetryableHTTPStatus(Exception):
    def __init__(self, status_code: int, body: str):
        super().__init__(f"retryable status={status_code} body={body[:200]}")
        self.status_code = status_code
        self.body = body


RETRYABLE_STATUS = {429, 500, 502, 503, 504}


@async_retry(
    max_attempts=6,
    base_delay=0.5,
    max_delay=10.0,
    retry_on=(httpx.TimeoutException, httpx.TransportError, RetryableHTTPStatus),
)
async def get_json(url: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as client:
        resp = await client.get(url)

    if resp.status_code in RETRYABLE_STATUS:
        raise RetryableHTTPStatus(resp.status_code, resp.text)

    resp.raise_for_status()
    return resp.json()

핵심은 상태코드를 예외로 변환해서 데코레이터의 재시도 흐름에 합류시키는 것입니다. 이렇게 하면 “예외 타입 기반” 정책으로 일원화할 수 있습니다.

`Retry-After` 헤더 존중하기

429나 일부 503 응답은 Retry-After 헤더로 서버가 권장 대기시간을 알려줍니다. 이 값을 무시하면 레이트리밋이 더 길어질 수 있습니다.

데코레이터에서 “다음 딜레이를 동적으로 계산”할 수 있도록 compute_delay 훅을 추가하는 방식이 유용합니다.

from typing import Any


def async_retry2(
    *,
    max_attempts: int,
    retry_on: Tuple[Type[BaseException], ...],
    compute_delay: Callable[[int, BaseException], float],
):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except asyncio.CancelledError:
                    raise
                except retry_on as exc:
                    if attempt >= max_attempts:
                        raise
                    delay = max(0.0, float(compute_delay(attempt, exc)))
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

compute_delay에서 Retry-After를 해석하는 예시는 아래처럼 구성할 수 있습니다.

import email.utils
import datetime


def parse_retry_after(value: str) -> float:
    # 숫자면 초 단위
    if value.isdigit():
        return float(value)

    # 날짜 형식이면 현재 시각과의 차이
    dt = email.utils.parsedate_to_datetime(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=datetime.timezone.utc)
    now = datetime.datetime.now(datetime.timezone.utc)
    return max(0.0, (dt - now).total_seconds())

이런 방식은 레이트리밋 대응을 더 “서버 친화적”으로 만듭니다. 레이트리밋/백오프/큐까지 포함한 큰 그림은 앞서 언급한 OpenAI 429/RateLimitError 재시도·백오프·큐 설계에서 더 깊게 다룹니다.

타임아웃: 시도 단위 vs 전체 단위

재시도에서 자주 생기는 함정은 “타임아웃을 어디에 걸었는지”가 불명확해지는 것입니다.

시도 단위 타임아웃: 각 호출이 너무 오래 걸리지 않게 제한
전체 단위 타임아웃: 재시도를 포함한 전체 작업의 상한

asyncio.timeout()(Python 3.11+)을 쓰면 전체 타임아웃을 간단히 걸 수 있습니다. 부등호가 들어간 설명은 MDX에서 문제가 될 수 있으니, 코드는 그대로 블록으로 제시합니다.

import asyncio


@async_retry(max_attempts=5, base_delay=0.2)
async def flaky_call():
    await asyncio.sleep(2.0)
    return "ok"


async def main():
    try:
        async with asyncio.timeout(3.0):
            return await flaky_call()
    except TimeoutError:
        return "overall timeout"

주의할 점은, 전체 타임아웃이 걸리면 내부에서 CancelledError가 발생할 수 있다는 것입니다. 앞서 데코레이터에서 CancelledError를 재시도하지 않고 전파하도록 한 이유가 여기에 있습니다.

재시도 폭풍 방지: 동시성 제한(세마포어)

재시도는 실패 시 트래픽을 더 늘릴 수 있습니다. 특히 같은 업스트림을 여러 코루틴이 동시에 호출하면, 장애 시점에 재시도가 겹치며 더 큰 부하를 만들 수 있습니다.

간단한 완화책은 세마포어로 동시 호출 수를 제한하는 것입니다.

import asyncio


class ConcurrencyLimit:
    def __init__(self, limit: int):
        self._sem = asyncio.Semaphore(limit)

    async def __aenter__(self):
        await self._sem.acquire()

    async def __aexit__(self, exc_type, exc, tb):
        self._sem.release()


limit = ConcurrencyLimit(limit=20)


@async_retry(max_attempts=4, base_delay=0.3)
async def call_upstream(payload: dict):
    async with limit:
        # 실제 업스트림 호출 위치
        ...

재시도 정책과 동시성 제한을 함께 적용하면 “장애 시 더 때리는” 패턴을 어느 정도 막을 수 있습니다. 운영 환경에서 재시도 폭풍이 서비스 재시작 루프와 결합되면 장애가 확대될 수 있는데, 비슷한 관점의 장애 추적 방법으로 systemd 서비스가 계속 재시작될 때 원인 추적도 참고할 만합니다.

예외 분류 전략: 재시도 가능한 실패만 골라내기

실전에서는 retry_on=(Exception,) 같은 광범위한 설정을 피해야 합니다. 대신 아래처럼 “재시도 가능”을 명시적으로 좁히는 것이 안전합니다.

네트워크 계층 예외: TimeoutError, OSError, httpx.TransportError
원격 일시 장애를 의미하는 커스텀 예외: RetryableHTTPStatus
SDK가 제공하는 레이트리밋 예외

또한 “같은 예외 타입이라도 어떤 경우는 재시도하면 안 되는” 상황이 있습니다. 예를 들어 TimeoutError가 실제로는 로컬 이벤트 루프가 막혀서 발생한다면 재시도는 해결책이 아닙니다. 이런 경우에는 재시도 횟수를 줄이고, 별도의 성능/리소스 진단이 필요합니다. (컨테이너 환경이라면 OOM이나 GC 이슈로 지연이 생길 수도 있으니, 상황에 따라 K8s OOMKilled 반복? 메모리 리밋·GC 진단법 같은 관점도 같이 점검합니다.)

관측성: 재시도 로그를 “한 줄”로 남기기

재시도는 장애 분석에서 중요한 단서입니다. 하지만 로그를 너무 많이 남기면 비용이 커지고 노이즈가 됩니다. 추천 방식은 아래 중 하나입니다.

on_retry에서 경량 로그(시도, 딜레이, 예외 타입)
최종 실패 시에만 에러 로그 + 마지막 예외 체인

예시로 on_retry에서 구조화 로그를 남기려면:

import json


def on_retry_json(state: RetryState):
    print(
        json.dumps(
            {
                "event": "retry",
                "attempt": state.attempt,
                "max_attempts": state.max_attempts,
                "delay_seconds": round(state.delay_seconds, 3),
                "elapsed_seconds": round(state.elapsed_seconds, 3),
                "exc_type": type(state.exc).__name__,
                "exc": str(state.exc),
            },
            ensure_ascii=False,
        )
    )

마무리: 재시도는 “정책”이고 데코레이터는 “배포 수단”

async 재시도·백오프를 데코레이터로 구현하면 다음을 얻습니다.

재시도 정책을 한 곳에서 통제(일관성)
호출부는 업무 로직에 집중(가독성)
CancelledError 전파, 타임아웃, 지터 등 asyncio 특성을 안전하게 반영

다만 재시도는 만능이 아닙니다. 재시도가 자주 발생한다면 업스트림의 레이트리밋 정책, 네트워크 품질, 동시성 폭, 큐잉 전략까지 함께 봐야 합니다. 특히 429 대응은 단순 재시도만으로는 부족한 경우가 많으니, 설계 확장 관점에서 OpenAI 429/RateLimitError 재시도·백오프·큐 설계를 함께 읽고 “백오프 + 큐 + 동시성 제한”을 한 세트로 가져가는 것을 권장합니다.