Python 데코레이터로 async 재시도·백오프 구현

서론

운영 환경의 비동기 코드는 항상 불안정한 외부 세계와 맞닿아 있습니다. HTTP API, gRPC, 메시지 큐, 데이터베이스 등은 순간적인 네트워크 글리치나 일시적 장애를 피할 수 없습니다. 이때 무작정 실패를 사용자에게 전파하기보다는, 재시도(retry)와 백오프(backoff) 로 복구 가능성을 최대한 활용하는 것이 중요합니다.

특히 asyncio 기반 코드에서는 다음과 같은 제약이 있습니다.

time.sleep() 을 쓰면 이벤트 루프를 블로킹해 전체 성능을 망가뜨립니다.
재시도 로직을 매번 복붙하면 코드가 지저분해지고, 버그가 생기기 쉽습니다.

이 글에서는 Python 데코레이터 로 async 재시도·백오프를 깔끔하게 추상화하는 방법을 설명합니다. 실무에서 자주 쓰는 지수 백오프(Exponential Backoff), Jitter, 예외 필터링, 동기/비동기 겸용 데코레이터 까지 단계적으로 구현해 보겠습니다.

중간중간 운영 이슈와 연결해서, 예를 들어 gRPC 503·데드라인 초과와 같은 상황에서 재시도 전략이 왜 중요한지, 관련 글인 gRPC 마이크로서비스 503·데드라인 초과 디버깅 과도 함께 생각해볼 수 있습니다.

왜 재시도·백오프인가?

재시도가 필요한 전형적인 상황

외부 HTTP API 5xx 응답
gRPC UNAVAILABLE, DEADLINE_EXCEEDED
데이터베이스 데드락 (예: MySQL 1213, PostgreSQL 일시 잠금)
메시지 브로커 일시적인 연결 끊김

예를 들어 MySQL에서는 Deadlock found when trying to get lock; try restarting transaction 같은 에러가 나면 트랜잭션을 재시도 하라고 명시합니다. 이를 수동으로 처리하기보다는, MySQL Deadlock 1213 로그로 원인 찾고 재현하기 글에서 다루듯, 재현·분석과 더불어 재시도 전략을 함께 설계하는 것이 일반적입니다.

백오프가 필요한 이유

재시도를 할 때, 간격 없이 빠르게 연속 호출 하면

서버에 부하를 더 키우고
네트워크/서비스 장애를 오히려 악화시키며
같은 타임슬롯에 모든 인스턴스가 동시에 재시도하는 스톰(storm) 현상을 유발할 수 있습니다.

그래서 보통 다음과 같은 전략을 사용합니다.

지수 백오프: delay = base * 2 ** attempt
최대 지연 한도(max delay)
Jitter(랜덤성): 모든 클라이언트가 같은 시점에 재시도하지 않도록 약간 랜덤을 섞음

기본: 동기 재시도 데코레이터부터

먼저 동기 함수용 재시도 데코레이터를 구현해 봅니다. 이후 이를 async 버전으로 확장할 것입니다.

import functools
import time
from typing import Callable, Type, Tuple


def retry(
    *,
    retries: int = 3,
    delay: float = 0.5,
    backoff: float = 2.0,
    exceptions: Tuple[Type[BaseException], ...] = (Exception,),
) -> Callable:
    """동기 함수용 재시도 데코레이터.

    Args:
        retries: 최대 재시도 횟수 (실제 시도는 retries + 1)
        delay: 최초 대기 시간(초)
        backoff: 시도마다 delay에 곱해지는 계수
        exceptions: 재시도 대상 예외 타입 튜플
    """

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            _delay = delay
            for attempt in range(retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == retries:
                        # 마지막 시도도 실패하면 예외 전파
                        raise
                    time.sleep(_delay)
                    _delay *= backoff

        return wrapper

    return decorator

이 패턴의 핵심은 다음과 같습니다.

decorator 클로저에 설정값을 캡처
wrapper 에서 for attempt in range(retries + 1) 로 재시도 루프
마지막 시도까지 실패하면 그대로 예외를 다시 던짐

이제 이 아이디어를 async 함수에도 적용해야 합니다.

async 함수에 적용할 때의 차이점

`time.sleep()` 대신 `asyncio.sleep()`

async 함수 안에서 time.sleep() 을 쓰면 이벤트 루프 전체를 블로킹합니다. 따라서 반드시 await asyncio.sleep() 을 사용해야 합니다.

동기 함수: time.sleep(1)
비동기 함수: await asyncio.sleep(1)

데코레이터에서 async 함수 감지하기

Python 표준 라이브러리의 inspect.iscoroutinefunction() 으로 함수가 코루틴인지 판별할 수 있습니다.

import inspect

inspect.iscoroutinefunction(async_func)  # True / False

이를 활용하면 하나의 데코레이터로 sync/async 둘 다 지원 하는 것도 가능합니다. 하지만 처음에는 이해를 위해 async 전용 버전부터 구현해 보겠습니다.

async 전용 재시도·백오프 데코레이터

import asyncio
import functools
from typing import Callable, Type, Tuple, Awaitable, Any


def async_retry(
    *,
    retries: int = 3,
    delay: float = 0.5,
    backoff: float = 2.0,
    max_delay: float | None = None,
    exceptions: Tuple[Type[BaseException], ...] = (Exception,),
) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
    """async 함수 전용 재시도·백오프 데코레이터.

    Args:
        retries: 최대 재시도 횟수
        delay: 최초 대기 시간(초)
        backoff: 지수 백오프 계수
        max_delay: 최대 대기 시간(초), None이면 무제한
        exceptions: 재시도 대상 예외 타입 튜플
    """

    def decorator(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
        if not asyncio.iscoroutinefunction(func):
            raise TypeError("async_retry can only be used with async functions")

        @functools.wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            _delay = delay
            for attempt in range(retries + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    if attempt == retries:
                        raise

                    # 백오프 대기
                    await asyncio.sleep(_delay)

                    # 다음 시도를 위한 딜레이 계산
                    _delay *= backoff
                    if max_delay is not None:
                        _delay = min(_delay, max_delay)

        return wrapper

    return decorator

사용 예시는 다음과 같습니다.

import aiohttp


@async_retry(retries=5, delay=0.2, backoff=2.0, max_delay=5.0,
             exceptions=(aiohttp.ClientError, asyncio.TimeoutError))
async def fetch_json(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url, timeout=3) as resp:
        resp.raise_for_status()
        return await resp.json()

여기서

aiohttp.ClientError, asyncio.TimeoutError 에 대해서만 재시도
최대 5회 재시도, 최초 0.2초에서 시작해 2배씩 증가, 최대 5초

같은 패턴은 gRPC 클라이언트, Redis, S3 SDK 등 어디에나 그대로 적용할 수 있습니다. 예를 들어 gRPC의 UNAVAILABLE 나 DEADLINE_EXCEEDED 에 대해 재시도하는 전략은 gRPC 마이크로서비스 503·데드라인 초과 디버깅 에서 언급한 장애 유형을 완화하는 데 직접적으로 도움됩니다.

Jitter(랜덤 지연) 추가하기

지수 백오프만 쓰면, 여러 인스턴스가 동시에 장애를 만나 모두 같은 타이밍에 재시도하게 됩니다. 이를 피하려면 Jitter 를 섞어야 합니다.

대표적인 전략은 AWS에서 소개한 것처럼:

full jitter: delay = random(0, base * 2 ** attempt)
equal jitter: delay = base * 2 ** attempt / 2 + random(0, base * 2 ** attempt / 2)

여기서는 full jitter 를 간단히 구현해 봅니다.

import asyncio
import functools
import random
from typing import Callable, Type, Tuple, Awaitable, Any


def async_retry_with_jitter(
    *,
    retries: int = 3,
    base_delay: float = 0.5,
    backoff: float = 2.0,
    max_delay: float | None = None,
    exceptions: Tuple[Type[BaseException], ...] = (Exception,),
) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
    """Jitter가 포함된 async 재시도·백오프 데코레이터."""

    def decorator(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
        if not asyncio.iscoroutinefunction(func):
            raise TypeError("async_retry_with_jitter can only be used with async functions")

        @functools.wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            for attempt in range(retries + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions:
                    if attempt == retries:
                        raise

                    # 지수 백오프 기반 최대 딜레이 계산
                    exp_delay = base_delay * (backoff ** attempt)
                    if max_delay is not None:
                        exp_delay = min(exp_delay, max_delay)

                    # full jitter: 0 ~ exp_delay 사이 랜덤
                    jittered = random.uniform(0, exp_delay)
                    await asyncio.sleep(jittered)

        return wrapper

    return decorator

이렇게 하면 모든 인스턴스가 동일한 타임슬롯에 재시도하는 것을 크게 줄일 수 있습니다.

sync/async 겸용 데코레이터 만들기

실제 서비스 코드에서는 동기 함수와 비동기 함수가 섞여 있는 경우가 많습니다. 이때 하나의 데코레이터로 둘 다 처리 하면 사용성이 좋아집니다.

핵심 아이디어는:

asyncio.iscoroutinefunction(func) 으로 async 여부 판별
async면 await func(...) 와 asyncio.sleep() 사용
sync면 func(...) 와 time.sleep() 사용

import asyncio
import functools
import time
from typing import Callable, Type, Tuple, Any


def retry_any(
    *,
    retries: int = 3,
    delay: float = 0.5,
    backoff: float = 2.0,
    max_delay: float | None = None,
    exceptions: Tuple[Type[BaseException], ...] = (Exception,),
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """sync/async 함수 모두 지원하는 재시도·백오프 데코레이터."""

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        is_async = asyncio.iscoroutinefunction(func)

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs) -> Any:
            _delay = delay
            for attempt in range(retries + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions:
                    if attempt == retries:
                        raise
                    await asyncio.sleep(_delay)
                    _delay *= backoff
                    if max_delay is not None:
                        _delay = min(_delay, max_delay)

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs) -> Any:
            _delay = delay
            for attempt in range(retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions:
                    if attempt == retries:
                        raise
                    time.sleep(_delay)
                    _delay *= backoff
                    if max_delay is not None:
                        _delay = min(_delay, max_delay)

        # async 함수에는 async_wrapper, sync 함수에는 sync_wrapper 반환
        return async_wrapper if is_async else sync_wrapper

    return decorator

사용 예시:

@retry_any(retries=3, delay=0.1)
def read_file(path: str) -> str:
    with open(path) as f:
        return f.read()


@retry_any(retries=5, delay=0.2)
async def call_api(url: str) -> str:
    # aiohttp, httpx 등 사용
    ...

주의할 점은, retry_any 자체는 항상 sync 함수로 정의되어야 하고, async 함수에 적용되면 반환된 wrapper가 async 라는 점입니다. 이 패턴은 FastAPI, aiohttp, aiokafka 등 다양한 라이브러리와 함께 사용할 때 유용합니다.

예외 필터링과 조건부 재시도

실무에서는 단순히 예외 타입만으로 재시도 여부를 결정하기 어려운 경우가 많습니다.

HTTP 4xx 는 재시도해도 소용없지만, 5xx 는 재시도 가치가 있음
gRPC 상태 코드 중 일부만 재시도 대상
DB 에러 코드 중 데드락·타임아웃만 재시도

이를 위해 커스텀 판별 함수(predicate) 를 받도록 확장해 보겠습니다.

from collections.abc import Callable as CCallable
from typing import Any


def async_retry_conditional(
    *,
    retries: int = 3,
    delay: float = 0.5,
    backoff: float = 2.0,
    max_delay: float | None = None,
    should_retry: CCallable[[BaseException], bool] | None = None,
) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
    """조건부 재시도 로직을 지원하는 async 데코레이터."""

    def decorator(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
        if not asyncio.iscoroutinefunction(func):
            raise TypeError("async_retry_conditional can only be used with async functions")

        @functools.wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            _delay = delay
            for attempt in range(retries + 1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:  # 타입은 should_retry에서 필터링
                    # 재시도 여부 결정
                    if should_retry is not None and not should_retry(e):
                        raise

                    if attempt == retries:
                        raise

                    await asyncio.sleep(_delay)
                    _delay *= backoff
                    if max_delay is not None:
                        _delay = min(_delay, max_delay)

        return wrapper

    return decorator

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

import httpx


def http_should_retry(exc: BaseException) -> bool:
    if isinstance(exc, httpx.HTTPStatusError):
        status = exc.response.status_code
        # 5xx 만 재시도
        return 500 <= status < 600
    if isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout)): 
        return True
    return False


@async_retry_conditional(retries=4, delay=0.3, backoff=2.0,
                         should_retry=http_should_retry)
async def get_user(user_id: int) -> dict:
    async with httpx.AsyncClient(timeout=3) as client:
        resp = await client.get(f"https://api.example.com/users/{user_id}")
        resp.raise_for_status()
        return resp.json()

이렇게 하면 예외 타입뿐 아니라 예외의 속성(HTTP 상태 코드, gRPC 코드, DB 에러 코드 등) 에 따라 정교하게 재시도 정책을 제어할 수 있습니다.

재시도 데코레이터 설계 시 주의할 점

1. idempotent 여부

재시도는 항상 멱등(idempotent) 호출에만 안전합니다.

GET, HEAD 처럼 읽기 전용 요청은 보통 멱등
POST, PATCH 는 멱등이 아닐 수 있음 (중복 결제, 중복 주문 등)

결제나 주문과 같은 도메인에서는 MSA 사가(Saga) 패턴 구현으로 중복결제 방지하기 에서 다루듯, 사가 패턴, 아웃박스 패턴, 중복 요청 방지 토큰 등과 조합해서 설계해야 합니다.

2. 타임아웃과의 조합

재시도는 전체 타임아웃 과 함께 설계해야 합니다.

개별 호출 타임아웃: timeout=3
전체 작업 타임아웃: asyncio.wait_for(..., timeout=10) 등

재시도 횟수와 딜레이를 더했을 때 전체 타임아웃을 초과하지 않도록 계산해야 합니다.

3. 로깅과 관측 가능성

재시도는 장애를 숨기는 것 이 아니라, 완화하는 것 입니다. 따라서 다음을 적절히 로깅/메트릭으로 남기는 것이 좋습니다.

재시도 횟수, 실패 비율
마지막으로 실패한 예외 타입/메시지
백오프 딜레이

간단한 예로, 데코레이터에 로거를 주입하여 각 시도마다 로그를 남길 수 있습니다.

import logging

logger = logging.getLogger(__name__)


def async_retry_logged(..., logger: logging.Logger | None = None):
    ...
    async def wrapper(*args, **kwargs):
        ...
        try:
            return await func(*args, **kwargs)
        except exceptions as e:
            if logger:
                logger.warning(
                    "Retrying %s attempt=%d error=%r", func.__name__, attempt, e
                )
            ...

4. 라이브러리 재사용 vs 직접 구현

이미 잘 만들어진 라이브러리도 많습니다.

tenacity
backoff

이 글에서처럼 직접 구현하는 것은 개념 이해와 커스터마이징 에 좋지만, 실제 프로덕션에서는 검증된 라이브러리를 쓰는 편이 안전한 경우도 많습니다. 다만 async 지원 여부, 타입 힌트, 의존성 정책 등을 고려해야 합니다.

마무리

이 글에서는 Python 데코레이터를 활용해 다음을 구현했습니다.

기본 동기 재시도·백오프 데코레이터
asyncio 기반 async 전용 재시도·백오프
지수 백오프 + Jitter 를 활용한 스톰 완화
sync/async 겸용 retry_any 데코레이터
예외 내용에 따른 조건부 재시도

비동기 환경에서 재시도는 성능과 안정성 사이의 균형을 잡는 핵심 도구입니다. 적절한 백오프와 Jitter, 멱등성 보장, 타임아웃·로깅 전략을 함께 설계하면, 일시적인 장애에 훨씬 탄탄한 시스템을 만들 수 있습니다.

실제 운영 환경에서 재시도 데코레이터를 적용해 보면서, 서비스 특성에 맞는 재시도 정책(횟수, 딜레이, 대상 예외)을 계속 튜닝해 나가는 것이 중요합니다.

서론