comgu

들어가며

JWT 로그인은 서버가 "이 사용자는 인증되었다"는 내용을 담은 토큰에 서명해서 발급하고, 클라이언트가 이 토큰을 요청마다 보내면 서버가 서명만 확인해서 사용자를 식별하는 방식입니다. 세션 방식과 달리 서버가 로그인 상태를 저장하지 않아도 된다는 점이 핵심입니다.

실무에서는 토큰을 하나만 쓰지 않고 Access Token과 Refresh Token 두 개를 함께 씁니다. 이 글에서는 로그인 요청 하나가 어떤 토큰이 되는지, 서버가 그 토큰을 어떻게 검증하는지, 만료되면 어떻게 재발급하는지, 로그아웃하면 무엇이 지워지고 무엇이 남는지를 실제 HTTP 요청·응답과 FastAPI 코드로 차례로 따라갑니다.

예제 기준 버전: Python 3.12, FastAPI 0.162, PyJWT 2.7, bcrypt 5.0, redis-py 8.1. 글에 나오는 요청과 응답은 예제 서버를 실제로 실행해서 확인한 결과입니다.

두 토큰의 역할

토큰을 둘로 나누는 이유는 서로 충돌하는 두 요구를 동시에 만족시키기 위해서입니다. API 호출은 빈번하므로 매번 저장소를 조회하지 않는 것이 좋고, 그러려면 서버가 토큰을 기억하지 않아야 합니다. 그런데 서버가 기억하지 않는 토큰은 탈취되어도 만료 전까지 막을 방법이 없습니다. 그래서 자주 쓰는 Access Token은 저장하지 않되 수명을 짧게 하고, 드물게 쓰는 Refresh Token은 수명을 길게 하되 서버가 저장해서 통제합니다.

JWT의 구조

로그인에 성공하면 서버는 다음과 같은 문자열을 돌려줍니다. 점(.)으로 구분된 세 부분으로 되어 있습니다.

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI0MiIsInR5cGUiOiJhY2Nlc3MiLCJpYXQiOjE3OTQyOTgwMzAsImV4cCI6MTc5NDI5ODkzMH0.p7HmnCKr-86VfJnE7bFIE9nfKvzcGycVocbCciQwQ1U
└────────── Header ──────────┘ └──────────────────────────── Payload ────────────────────────────┘ └──────────── Signature ────────────┘

앞의 두 부분은 JSON을 Base64URL로 인코딩한 것이라 누구나 디코딩해서 읽을 수 있습니다.

// Header: 서명 알고리즘
{"alg": "HS256", "typ": "JWT"}

// Payload: 실제 정보
{"sub": "42", "type": "access", "iat": 1794298030, "exp": 1794298930}

세 번째 부분인 Signature는 앞의 두 부분을 서버만 아는 비밀키로 서명한 값입니다.

Signature = HMAC-SHA256( base64url(Header) + "." + base64url(Payload), 비밀키 )

읽을 수는 있지만 고칠 수는 없다

Payload는 암호화된 것이 아니라 인코딩만 된 것이므로 비밀번호나 개인정보를 넣으면 안 됩니다. 대신 내용을 수정할 수는 없습니다. 실제로 위 토큰의 Payload에서 sub를 "42"에서 "1"로 바꾸고 Signature는 그대로 둔 채 서버에 보내 보면 다음과 같이 거절됩니다.

jwt.exceptions.InvalidSignatureError: Signature verification failed

서버가 Header와 바뀐 Payload로 서명을 다시 계산하면 토큰에 붙어 온 Signature와 값이 달라지기 때문입니다. 공격자가 바뀐 Payload에 맞는 서명을 새로 만들려면 비밀키가 필요한데, 비밀키는 서버에만 있습니다.

시나리오로 따라가기: 로그인부터 로그아웃까지

예제 서버는 엔드포인트 네 개로 구성됩니다. 전체 코드는 글 마지막에 있습니다.

장면 0: 회원가입 시점의 준비

서버는 비밀번호를 그대로 저장하지 않고, bcrypt나 argon2 같은 해시 함수로 변환한 값만 저장합니다.

password_hash = bcrypt.hashpw(b"mypassword123", bcrypt.gensalt())
# b'$2b$12$Kx9...' 형태. 원래 비밀번호로 되돌릴 수 없음

해시는 되돌릴 수 없으므로, 로그인할 때는 입력받은 비밀번호를 같은 방식으로 해시해서 저장된 값과 일치하는지만 비교합니다.

장면 1: 로그인 — POST /auth/login

클라이언트가 이메일과 비밀번호를 보냅니다. 비밀번호가 평문으로 전송되므로 HTTPS는 필수입니다.

POST /auth/login
Content-Type: application/json

{"email": "user@example.com", "password": "mypassword123"}

서버는 사용자를 찾고 비밀번호를 확인합니다.

@app.post("/auth/login")
def login(body: LoginRequest, response: Response):
    user = USERS.get(body.email)
    if user is None or not bcrypt.checkpw(body.password.encode(), user["password_hash"]):
        raise HTTPException(401, "invalid email or password")
    return issue_tokens(response, user["id"])

이메일이 없는 경우와 비밀번호가 틀린 경우에 같은 에러 메시지를 돌려주는 점에 주목할 만합니다. 실제로 두 경우 모두 401 {"detail": "invalid email or password"}가 나옵니다. 메시지를 구분하면 공격자가 어떤 이메일이 가입되어 있는지 알아낼 수 있습니다.

확인에 성공하면 issue_tokens()가 두 토큰을 만듭니다.

def issue_tokens(response: Response, user_id: str) -> dict:
    jti = uuid.uuid4().hex
    r.set(f"refresh:{jti}", user_id, ex=REFRESH_TTL)     # 이 Refresh Token이 유효하다는 기록
    r.sadd(f"user_refresh:{user_id}", jti)               # 사용자별 Refresh Token 목록

    access = create_token(user_id, "access", ACCESS_TTL)
    refresh = create_token(user_id, "refresh", REFRESH_TTL, jti)
    response.set_cookie(
        "refresh_token", refresh,
        httponly=True, secure=True, samesite="strict",
        path="/auth", max_age=int(REFRESH_TTL.total_seconds()),
    )
    return {"access_token": access, "token_type": "bearer"}

실제 응답은 다음과 같습니다. 토큰 문자열은 줄였습니다.

HTTP/1.1 200 OK
set-cookie: refresh_token=eyJ...; HttpOnly; Max-Age=1209600; Path=/auth; SameSite=strict; Secure
content-type: application/json

{"access_token": "eyJ...", "token_type": "bearer"}

Access Token은 응답 본문으로, Refresh Token은 쿠키로 전달됩니다. 이 시점에 Redis에는 키 두 개가 생깁니다.

refresh:d477afb9b6c747d6a90319e03f0cf560   → "42"  (TTL 14일)
user_refresh:42                            → {"d477afb9b6c747d6a90319e03f0cf560"}

Access Token에 대한 기록은 어디에도 남지 않습니다.

장면 2: API 호출 — GET /api/me

클라이언트는 이후 모든 API 요청에 Access Token을 헤더로 붙입니다.

GET /api/me
Authorization: Bearer eyJ...

FastAPI에서는 의존성 하나로 검증을 처리합니다.

bearer = HTTPBearer()

def decode_token(token: str, expected_type: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "invalid token")
    if payload.get("type") != expected_type:
        raise HTTPException(401, "invalid token type")
    return payload

def current_user(cred: HTTPAuthorizationCredentials = Depends(bearer)) -> str:
    return decode_token(cred.credentials, "access")["sub"]

@app.get("/api/me")
def me(user_id: str = Depends(current_user)):
    return {"user_id": user_id}

jwt.decode() 한 줄 안에서 다음 일이 순서대로 일어납니다.

1. 토큰을 점으로 나눠 Header, Payload, Signature를 분리합니다.
2. Header의 alg가 algorithms에 지정한 값인지 확인합니다.
3. Header와 Payload로 서명을 다시 계산해서 Signature와 비교합니다. 다르면 InvalidSignatureError입니다.
4. Payload의 exp를 현재 시각과 비교합니다. 지났으면 ExpiredSignatureError입니다.

그다음 type이 "access"인지 확인하고, 통과하면 sub에서 사용자 ID를 꺼냅니다.

이 과정 어디에서도 DB나 Redis를 조회하지 않습니다. 서명 계산은 CPU만으로 끝나므로 빠르고, 서버가 여러 대여도 같은 비밀키만 공유하면 어느 서버에서든 검증할 수 있습니다.

여러 종류의 토큰을 /api/me에 보냈을 때 실제 결과는 다음과 같습니다.

Refresh Token도 같은 비밀키로 서명되어 있으므로 서명 검증만으로는 Access Token과 구분되지 않습니다. 그래서 type 클레임으로 용도를 확인해야 Refresh Token을 Access Token 자리에 쓰는 것을 막을 수 있습니다.

장면 3: Access Token 만료

발급 후 15분이 지나면 exp 확인에서 걸립니다.

HTTP/1.1 401 Unauthorized

{"detail": "token expired"}

클라이언트는 이 응답을 받아도 사용자를 바로 로그인 화면으로 보내지 않습니다. 대신 Refresh Token으로 새 Access Token을 받아 오는 다음 단계로 넘어갑니다.

장면 4: 재발급 — POST /auth/refresh

POST /auth/refresh
Cookie: refresh_token=eyJ...

요청 본문은 비어 있습니다. 쿠키의 Path=/auth가 요청 경로와 맞으므로 브라우저가 Refresh Token 쿠키를 자동으로 붙여 보냅니다.

@app.post("/auth/refresh")
def refresh(response: Response, refresh_token: str | None = Cookie(default=None)):
    if refresh_token is None:
        raise HTTPException(401, "no refresh token")
    payload = decode_token(refresh_token, "refresh")
    user_id, jti = payload["sub"], payload["jti"]

    if r.delete(f"refresh:{jti}") == 0:
        # 서명은 유효한데 저장소에 없다 → 이미 한 번 쓰인 토큰이 다시 들어왔다
        revoke_all(user_id)
        raise HTTPException(401, "refresh token reused")
    r.srem(f"user_refresh:{user_id}", jti)
    return issue_tokens(response, user_id)

순서는 다음과 같습니다.

1. Access Token과 같은 방식으로 서명, 만료, type을 확인합니다. 여기까지는 저장소를 조회하지 않습니다.
2. Redis에서 refresh:{jti} 키를 삭제합니다. 삭제된 개수가 1이면 유효한 토큰이었다는 뜻입니다.
3. 새 Access Token과 새 Refresh Token을 발급하고, 새 jti를 Redis에 저장합니다.

2번에서 "조회 후 삭제"를 두 번에 나눠 하지 않고 DELETE 한 번의 결과로 판단하는 이유는 동시성 때문입니다. 같은 Refresh Token으로 요청 두 개가 동시에 들어와도 Redis의 DELETE는 원자적이므로 둘 중 하나만 1을 받습니다.

실행해 보면 응답의 Set-Cookie로 새 Refresh Token이 내려오고, Redis의 jti도 바뀝니다.

재발급 전: refresh:d477afb9b6c747d6a90319e03f0cf560
재발급 후: refresh:543da375613f4e78b6ca1c6496550ecc

이처럼 재발급할 때마다 Refresh Token도 새것으로 바꾸고 이전 것은 폐기하는 방식을 Refresh Token Rotation이라고 합니다. 각 Refresh Token은 한 번만 쓸 수 있게 됩니다.

클라이언트는 새로 받은 Access Token으로 장면 3에서 실패했던 요청을 다시 보냅니다. 사용자 입장에서는 아무 일도 없었던 것처럼 보입니다.

장면 5: 이미 쓴 Refresh Token이 다시 들어오면

Rotation의 효과는 Refresh Token이 탈취되었을 때 드러납니다. 장면 4에서 폐기된 첫 번째 Refresh Token을 누군가 복사해 두었다가 보낸다고 가정합니다.

POST /auth/refresh
Cookie: refresh_token=eyJ...(폐기된 첫 번째 토큰)

HTTP/1.1 401 Unauthorized

{"detail": "refresh token reused"}

이 토큰은 서명도 유효하고 만료되지도 않았지만, Redis에 jti가 없습니다. 서버가 직접 발급했는데 이미 사용된 토큰이 다시 들어왔다는 뜻이므로, 서버는 탈취를 의심하고 그 사용자의 Refresh Token을 전부 폐기합니다.

def revoke_all(user_id: str) -> None:
    key = f"user_refresh:{user_id}"
    for jti in r.smembers(key):
        r.delete(f"refresh:{jti}")
    r.delete(key)

실제로 이 요청 직후 Redis의 키는 모두 사라지고, 정상 사용자가 갖고 있던 최신 Refresh Token으로 재발급을 요청해도 401 refresh token reused가 나옵니다. 정상 사용자도 다시 로그인해야 하는데, 이것이 의도된 동작입니다. 서버 입장에서는 둘 중 누가 공격자인지 구분할 수 없으므로, 둘 다 끊고 비밀번호를 아는 쪽만 다시 들어오게 하는 것입니다.

한 가지 주의할 점이 있습니다. 브라우저 탭 여러 개가 동시에 Access Token 만료를 감지하고 같은 Refresh Token으로 재발급을 요청하면, 두 번째 요청이 재사용으로 판정되어 정상 사용자가 로그아웃됩니다. 클라이언트에서 재발급 요청을 하나로 묶거나, 서버에서 직전에 폐기된 토큰을 몇 초간 허용하는 식으로 보완합니다.

장면 6: 로그아웃 — POST /auth/logout

@app.post("/auth/logout", status_code=204)
def logout(response: Response, refresh_token: str | None = Cookie(default=None)):
    if refresh_token:
        try:
            payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
            r.delete(f"refresh:{payload['jti']}")
            r.srem(f"user_refresh:{payload['sub']}", payload["jti"])
        except jwt.InvalidTokenError:
            pass
    response.delete_cookie("refresh_token", path="/auth")

서버는 Redis에서 해당 jti를 지우고, 같은 이름과 경로로 Max-Age=0인 쿠키를 내려 브라우저의 쿠키를 지웁니다.

HTTP/1.1 204 No Content
set-cookie: refresh_token=""; Max-Age=0; Path=/auth; ...

이제 Refresh Token은 서버에서도 클라이언트에서도 사라졌습니다. 그런데 로그아웃 전에 받은 Access Token으로 /api/me를 호출해 보면 결과는 이렇습니다.

HTTP/1.1 200 OK

{"user_id": "42"}

로그아웃해도 Access Token은 만료될 때까지 유효합니다. 서버가 Access Token을 저장하지 않으므로 지울 대상 자체가 없기 때문입니다. 클라이언트가 메모리에서 토큰을 버리면 정상 사용 흐름에서는 문제가 없지만, 누군가 토큰을 복사해 두었다면 남은 시간(최대 15분) 동안은 쓸 수 있습니다. Access Token의 수명을 짧게 잡는 이유가 바로 이것입니다.

Refresh Token은 어디에 저장하나

Refresh Token을 다루는 방식은 크게 두 가지입니다.

클라이언트가 직접 저장하는 방식은 응답 본문으로 Refresh Token을 받아 localStorage나 앱 저장소에 두고, 재발급할 때 요청 본문이나 헤더에 직접 넣어 보냅니다. 모바일·네이티브 앱에서는 지금도 이 방식이 표준이며, 저장소로는 iOS Keychain이나 Android Keystore 같은 보안 저장소를 씁니다.

쿠키 방식은 이 글의 예제처럼 서버가 Set-Cookie로 내려주고, 브라우저가 저장했다가 경로가 맞는 요청에 자동으로 붙여 보냅니다. 클라이언트 코드는 토큰 값을 알 필요가 없고, HttpOnly 속성 때문에 알 수도 없습니다.

웹에서 쿠키 방식을 선호하는 이유는 XSS입니다. localStorage에 있는 값은 페이지에 주입된 악성 스크립트가 읽어서 외부로 보낼 수 있지만, HttpOnly 쿠키는 JavaScript로 읽을 수 없습니다.

예제에서 쓴 쿠키 속성을 정리하면 다음과 같습니다.

Path를 /auth/refresh가 아니라 /auth로 잡은 것은 로그아웃 엔드포인트에서도 Refresh Token을 받아 Redis에서 지워야 하기 때문입니다. 동시에 /api/... 요청에는 쿠키가 붙지 않습니다. 실제로 확인해 보면 /api/me 요청에는 Cookie 헤더가 아예 없고, /auth/refresh 요청에만 붙습니다. 수명이 긴 토큰이 네트워크에 노출되는 횟수를 최소한으로 줄이는 설정입니다.

Access Token은 응답 본문으로 받아 JavaScript 메모리 변수에 둡니다. 새로고침하면 사라지지만, 그때는 쿠키에 남아 있는 Refresh Token으로 재발급을 받으면 됩니다.

쿠키 방식에서 따로 챙길 것

브라우저가 쿠키를 자동으로 보내기 때문에 생기는 문제가 두 가지 있습니다.

첫째는 CSRF입니다. 다른 사이트가 사용자 브라우저를 통해 우리 서버로 요청을 보내게 만들면 쿠키가 함께 전송될 수 있습니다. SameSite=Strict 또는 Lax로 대부분 막을 수 있고, /auth/refresh가 공격자에게 돌려주는 것은 응답 본문의 Access Token인데 다른 출처에서는 CORS 때문에 그 응답을 읽을 수 없다는 점도 방어에 도움이 됩니다.

둘째는 프론트엔드와 API의 도메인이 다를 때의 설정입니다. 다른 출처로 보내는 fetch는 기본적으로 쿠키를 보내지 않으므로 다음과 같이 지정해야 합니다.

await fetch("https://api.example.com/auth/refresh", {
  method: "POST",
  credentials: "include",
});

서버는 CORS 응답에 Access-Control-Allow-Credentials: true와 정확한 출처를 지정해야 합니다. 이때 Access-Control-Allow-Origin: *는 쓸 수 없습니다.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

클라이언트는 서명을 검증하나

보통은 검증하지 않습니다. 클라이언트는 토큰을 그대로 보관했다가 서버에 다시 보내기만 하고, 진위 확인은 서버의 일입니다.

HS256처럼 대칭키 방식이라면 검증에 비밀키가 필요하므로 클라이언트는 애초에 검증할 수 없습니다. 비밀키를 클라이언트에 주면 누구나 토큰을 위조할 수 있게 됩니다. 검증하지 않아도 되는 이유는 클라이언트가 토큰을 HTTPS로 서버에게서 직접 받았기 때문입니다. 토큰이 정말 그 서버에서 왔다는 것은 TLS가 이미 보장합니다.

클라이언트가 할 수 있는 일은 Payload를 디코딩해서 exp를 읽고, 만료 직전에 미리 재발급을 요청하는 정도입니다. 이것은 검증이 아니라 단순히 내용을 읽는 것입니다.

const payload = JSON.parse(atob(accessToken.split(".")[1].replace(/-/g, "+").replace(/_/g, "/")));
const msLeft = payload.exp * 1000 - Date.now();   // 남은 시간

예외는 RS256이나 ES256 같은 공개키 방식입니다. 이 경우 서버는 개인키로 서명하고 공개키를 JWKS로 공개하므로, 공개키를 받은 쪽은 누구든 서명을 검증할 수 있습니다. 대표적인 예가 OAuth 2.0 / OpenID Connect에서 id_token을 받은 애플리케이션이 IdP의 공개키로 검증하는 경우입니다.

Access Token을 즉시 무효화해야 한다면

장면 6에서 본 것처럼 Access Token은 만료 전까지 막을 수 없습니다. 비밀번호 변경이나 계정 정지처럼 즉시 끊어야 하는 상황이 있다면 서버에 상태를 다시 둘 수밖에 없습니다. 흔히 쓰는 방법은 두 가지입니다.

• 블랙리스트: Access Token에도 jti를 넣고, 폐기할 토큰의 jti를 Redis에 저장합니다. TTL은 그 토큰의 남은 수명만큼만 주면 되므로 저장소가 무한히 커지지 않습니다.
• 토큰 버전: 사용자마다 버전 번호를 저장해 두고 토큰에 발급 당시 버전을 넣습니다. 버전을 올리면 그 이전에 발급된 토큰이 한꺼번에 무효가 됩니다.

어느 쪽이든 매 요청마다 Redis를 조회하게 되므로, 저장소 조회 없이 검증한다는 JWT의 장점은 그만큼 줄어듭니다. 대부분의 서비스는 Access Token 수명을 5~15분으로 짧게 잡는 것으로 충분하다고 보고 이 방법을 쓰지 않습니다.

서명 알고리즘과 algorithms 고정

어떤 알고리즘을 쓰든 jwt.decode()에 algorithms를 반드시 명시해야 합니다. 토큰의 Header에 적힌 alg를 그대로 믿으면, 공격자가 "alg": "none"으로 서명이 아예 없는 토큰을 만들어 보낼 수 있습니다. 예제처럼 algorithms=["HS256"]으로 고정해 두면 이런 토큰은 401 invalid token으로 거절됩니다. 실제로 서명 없이 만든 다음 토큰을 보내 확인했습니다.

eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0.eyJzdWIiOiIxIiwidHlwZSI6ImFjY2VzcyIsImV4cCI6MTc5NDI5ODA5MH0.
                                                                                               ↑ 서명 부분이 비어 있음

비밀키는 코드에 쓰지 않고 환경 변수 등으로 주입하며, HS256이라면 충분히 긴 랜덤 값(예: openssl rand -hex 32)을 씁니다.

정리

JWT 로그인은 "자주 쓰는 토큰은 저장하지 않고 서명만 검증하고, 드물게 쓰는 토큰은 저장해서 통제한다"는 구조입니다.

• 로그인하면 서버는 수명이 짧은 Access Token과 수명이 긴 Refresh Token을 발급합니다. Refresh Token의 jti만 Redis에 저장합니다.
• API 요청은 Access Token의 서명, 만료 시각, 용도(type)만 확인하고 저장소는 조회하지 않습니다. Payload는 누구나 읽을 수 있지만, 비밀키 없이는 수정할 수 없습니다.
• Access Token이 만료되면 Refresh Token으로 두 토큰을 모두 새로 받습니다. 이전 Refresh Token은 폐기되며(Rotation), 폐기된 토큰이 다시 들어오면 탈취로 보고 그 사용자의 토큰을 전부 폐기합니다.
• 로그아웃은 Refresh Token을 지우는 것입니다. Access Token은 만료될 때까지 유효하므로 수명을 짧게 잡습니다.
• 웹에서는 Refresh Token을 HttpOnly, Secure, SameSite, 좁은 Path를 지정한 쿠키에, Access Token은 메모리에 둡니다. 모바일 앱은 보안 저장소에 직접 저장합니다.
• 클라이언트는 서명을 검증하지 않습니다. 서버는 algorithms를 고정해서 검증합니다.

부록: 예제 전체 코드

import uuid
from datetime import datetime, timedelta, timezone

import bcrypt
import jwt
import redis
from fastapi import Cookie, Depends, FastAPI, HTTPException, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel

SECRET_KEY = "change-me-to-a-long-random-string"   # 실제로는 환경 변수로 주입
ALGORITHM = "HS256"
ACCESS_TTL = timedelta(minutes=15)
REFRESH_TTL = timedelta(days=14)

app = FastAPI()
r = redis.Redis(decode_responses=True)
bearer = HTTPBearer()

# 예제용 사용자 저장소 (실제로는 DB)
USERS = {
    "user@example.com": {
        "id": "42",
        "password_hash": bcrypt.hashpw(b"mypassword123", bcrypt.gensalt()),
    }
}


class LoginRequest(BaseModel):
    email: str
    password: str


def create_token(user_id: str, token_type: str, ttl: timedelta, jti: str | None = None) -> str:
    now = datetime.now(timezone.utc)
    payload = {"sub": user_id, "type": token_type, "iat": now, "exp": now + ttl}
    if jti:
        payload["jti"] = jti
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def decode_token(token: str, expected_type: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "invalid token")
    if payload.get("type") != expected_type:
        raise HTTPException(401, "invalid token type")
    return payload


def issue_tokens(response: Response, user_id: str) -> dict:
    jti = uuid.uuid4().hex
    r.set(f"refresh:{jti}", user_id, ex=REFRESH_TTL)
    r.sadd(f"user_refresh:{user_id}", jti)

    access = create_token(user_id, "access", ACCESS_TTL)
    refresh = create_token(user_id, "refresh", REFRESH_TTL, jti)
    response.set_cookie(
        "refresh_token", refresh,
        httponly=True, secure=True, samesite="strict",
        path="/auth", max_age=int(REFRESH_TTL.total_seconds()),
    )
    return {"access_token": access, "token_type": "bearer"}


def revoke_all(user_id: str) -> None:
    key = f"user_refresh:{user_id}"
    for jti in r.smembers(key):
        r.delete(f"refresh:{jti}")
    r.delete(key)


def current_user(cred: HTTPAuthorizationCredentials = Depends(bearer)) -> str:
    return decode_token(cred.credentials, "access")["sub"]


@app.post("/auth/login")
def login(body: LoginRequest, response: Response):
    user = USERS.get(body.email)
    if user is None or not bcrypt.checkpw(body.password.encode(), user["password_hash"]):
        raise HTTPException(401, "invalid email or password")
    return issue_tokens(response, user["id"])


@app.get("/api/me")
def me(user_id: str = Depends(current_user)):
    return {"user_id": user_id}


@app.post("/auth/refresh")
def refresh(response: Response, refresh_token: str | None = Cookie(default=None)):
    if refresh_token is None:
        raise HTTPException(401, "no refresh token")
    payload = decode_token(refresh_token, "refresh")
    user_id, jti = payload["sub"], payload["jti"]

    if r.delete(f"refresh:{jti}") == 0:
        # 서명은 유효한데 저장소에 없다 → 이미 한 번 쓰인 토큰이 다시 들어왔다
        revoke_all(user_id)
        raise HTTPException(401, "refresh token reused")
    r.srem(f"user_refresh:{user_id}", jti)
    return issue_tokens(response, user_id)


@app.post("/auth/logout", status_code=204)
def logout(response: Response, refresh_token: str | None = Cookie(default=None)):
    if refresh_token:
        try:
            payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
            r.delete(f"refresh:{payload['jti']}")
            r.srem(f"user_refresh:{payload['sub']}", payload["jti"])
        except jwt.InvalidTokenError:
            pass
    response.delete_cookie("refresh_token", path="/auth")

검증 환경: Python 3.12.3, FastAPI 0.162.2, PyJWT 2.7.0, bcrypt 5.0.0, redis-py 8.1.0. Redis는 fakeredis로 대체하고 FastAPI TestClient(https://testserver)로 각 장면의 요청을 실행했으며, 글의 응답은 그 결과에서 토큰 문자열만 줄인 것입니다.

들어가며

Django의 페이지네이션은 ORM의 기능이 아니라 django.core.paginator.Paginator라는 유틸리티 클래스입니다. 하는 일은 단순합니다. QuerySet을 일정한 크기로 잘라, 요청받은 "몇 번째 페이지"만 꺼내 줍니다.

이 자르기가 가능한 이유는 QuerySet의 슬라이싱이 SQL의 LIMIT / OFFSET으로 번역되기 때문입니다. 이 글에서는 게시판 요청 하나가 Paginator를 거쳐 어떤 SQL이 되는지, 잘못된 페이지 번호는 어떻게 처리되는지, COUNT(*)는 언제 몇 번 실행되는지를 차례로 따라갑니다.

예제 기준 버전: Django 6.1. 여기서 다루는 Paginator 동작은 Django 4.x·5.x에서도 같습니다. 글의 동작과 SQL은 실제 Django 소스를 확인하고 SQLite에서 실행해 검증했습니다.

기본 사용법과 핵심 객체

Paginator에 QuerySet과 페이지 크기를 넘기고, get_page()로 원하는 페이지를 꺼내면 끝입니다.

from django.core.paginator import Paginator
from .models import Post

qs = Post.objects.order_by("-created_at")   # 정렬 필수 (이유는 아래에서)
paginator = Paginator(qs, 10)               # 한 페이지에 10개

page = paginator.get_page(2)                # 2페이지 Page 객체

for post in page:                           # page.object_list 를 순회
    print(post.title)

여기에는 두 종류의 객체가 등장합니다. Paginator는 전체 목록에 대한 정보를, Page는 꺼낸 한 페이지에 대한 정보를 들고 있습니다.

ListView를 쓴다면 paginate_by = 10만 지정해도 같은 일을 해 주고, 템플릿에 page_obj, paginator, is_paginated가 전달됩니다.

정렬이 필요한 이유

정렬되지 않은 QuerySet을 넘기면 Paginator 생성 시점에 UnorderedObjectListWarning 경고가 발생합니다. ORDER BY 없이 OFFSET을 쓰면 DB가 행을 돌려주는 순서가 보장되지 않아, 페이지 사이에 항목이 겹치거나 빠질 수 있기 때문입니다. 모델 Meta.ordering이 지정돼 있으면 정렬된 것으로 봅니다.

시나리오로 따라가기: 요청 하나가 SQL이 되기까지

게시판에 글이 95개 있고 한 페이지에 10개씩 보여 주는 상황입니다. 뷰는 다음과 같습니다.

def post_list(request):
    qs = Post.objects.order_by("-created_at")
    page = Paginator(qs, 10).get_page(request.GET.get("page"))
    return render(request, "posts.html", {"page": page})

{% for post in page %}{{ post.title }}{% endfor %}

{% if page.has_previous %}<a href="?page={{ page.previous_page_number }}">이전</a>{% endif %}
{{ page.number }} / {{ page.paginator.num_pages }}
{% if page.has_next %}<a href="?page={{ page.next_page_number }}">다음</a>{% endif %}

장면 1: 처음 진입 — GET /posts/

URL에 page가 없으니 request.GET.get("page")는 None이고, get_page(None)은 1페이지를 돌려줍니다. 실행되는 SQL은 두 개입니다.

SELECT COUNT(*) FROM post;                                     -- 95 → 총 10페이지
SELECT ... FROM post ORDER BY created_at DESC LIMIT 10;        -- 최신 글 1~10번째

1페이지라서 has_previous()가 False이므로 "이전" 링크는 보이지 않습니다.

장면 2: 3페이지 클릭 — GET /posts/?page=3

같은 뷰가 처음부터 다시 실행됩니다. Paginator가 하는 계산은 아래가 전부입니다.

OFFSET = (3 - 1) × 10 = 20
LIMIT  = 10

SELECT COUNT(*) FROM post;
SELECT ... FROM post ORDER BY created_at DESC LIMIT 10 OFFSET 20;   -- 21~30번째

서버는 사용자가 어디까지 봤는지 기억하지 않습니다. 매 요청마다 URL의 page=3만 보고 다시 계산하며, COUNT(*)도 매번 다시 실행됩니다.

장면 3: 마지막 페이지 — GET /posts/?page=10

남은 글이 5개뿐이므로, Paginator는 슬라이스 끝을 전체 개수(95)로 잘라 qs[90:95]를 만듭니다. 그래서 SQL은 LIMIT 10이 아니라 LIMIT 5입니다.

SELECT ... FROM post ORDER BY created_at DESC LIMIT 5 OFFSET 90;    -- 91~95번째

has_next()가 False가 되어 "다음" 링크가 사라집니다.

데이터 쿼리는 언제 실행되나

get_page()가 돌려주는 Page의 object_list는 슬라이스된 QuerySet이라 아직 실행되지 않은 상태입니다. COUNT(*)는 get_page() 안에서 범위를 검증할 때 바로 실행되지만, 데이터 조회 SQL은 템플릿에서 {% for post in page %}로 순회하는 시점에 실행됩니다.

모바일 앱이라면: DRF PageNumberPagination

앱은 HTML 대신 JSON을 받습니다. DRF의 PageNumberPagination을 쓰면 GET /api/posts/?page=3 요청에 다음과 같이 응답합니다.

{
  "count": 95,
  "next": "https://example.com/api/posts/?page=4",
  "previous": "https://example.com/api/posts/?page=2",
  "results": [ ... 10개 ... ]
}

앱은 results를 그리고, 다음 페이지가 필요하면 next URL을 그대로 호출합니다. next가 null이면 마지막 페이지입니다. 내부적으로는 DRF도 Django의 Paginator를 사용하므로 SQL은 장면 2와 같습니다.

잘못된 페이지 번호는 어떻게 처리되나

같은 Paginator라도 어떤 메서드로, 어떤 계층에서 부르느냐에 따라 결과가 다릅니다. page()는 예외를 던지고, get_page()는 유효한 페이지로 바꿔 주며, ListView와 DRF는 404를 돌려줍니다.

표에서 눈여겨볼 점이 두 가지 있습니다.

첫째, get_page(0)은 1페이지가 아니라 마지막 페이지를 돌려줍니다. 1보다 작은 값도 범위 초과와 똑같이 EmptyPage로 처리되고, get_page()는 EmptyPage를 받으면 무조건 num_pages로 대체하기 때문입니다.

def get_page(self, number):
    try:
        number = self.validate_number(number)
    except PageNotAnInteger:
        number = 1
    except EmptyPage:          # 999든 0이든 -1이든
        number = self.num_pages
    return self.page(number)

둘째, ListView는 내부에서 get_page()가 아니라 page()를 쓰고, InvalidPage를 잡아 Http404로 바꿉니다. 대신 page=last라는 특수 값을 마지막 페이지로 해석해 줍니다. DRF도 범위 밖이나 숫자가 아닌 값에는 404(Invalid page.)를 돌려주고, last는 마지막 페이지로 처리합니다.

정리하면 사람이 보는 HTML 화면에서는 조용히 보정해 주는 get_page()가 편하고, API처럼 잘못된 요청을 명확히 거절해야 하는 곳에서는 page()와 예외 처리를 쓰는 편이 맞습니다.

내부 동작: COUNT로 범위를 검증하고, 인스턴스에 캐시한다

?page=999가 범위를 넘었다는 것을 Django가 아는 이유는 COUNT(*) 결과로 전체 페이지 수를 먼저 계산하기 때문입니다. 그리고 이 값은 Paginator 인스턴스에 캐시되어, 한 인스턴스에서는 COUNT(*)가 한 번만 실행됩니다.

get_page("999")의 실행 순서

1. validate_number("999")가 값을 정수로 바꿉니다. 실패하면 PageNotAnInteger, 1보다 작으면 EmptyPage입니다.
2. 범위를 비교하려고 num_pages에 접근하고, num_pages가 count를 읽는 순간 SELECT COUNT(*)가 실행됩니다. 95를 10으로 나눠 올림하면 10페이지입니다.
3. 999 > 10이므로 EmptyPage가 발생합니다.
4. get_page()가 예외를 잡아 page(10)을 다시 호출합니다. 이때 count는 이미 캐시되어 있어 COUNT(*)는 다시 실행되지 않습니다.
5. 순회 시점에 LIMIT 5 OFFSET 90으로 마지막 페이지 데이터만 조회합니다.

결과적으로 OFFSET 9980 같은 쿼리는 실행되지 않습니다. DB에 가는 것은 COUNT(*) 한 번과 마지막 페이지 조회 한 번뿐입니다.

관련 소스 (Django 6.1 기준, 요약)

class Paginator(BasePaginator):
    @cached_property
    def count(self):
        c = getattr(self.object_list, "count", None)
        if callable(c) and not inspect.isbuiltin(c) and method_has_no_args(c):
            return c()                      # QuerySet → SELECT COUNT(*)
        return len(self.object_list)        # 일반 리스트 → len()

    @cached_property
    def num_pages(self):
        if self.count == 0 and not self.allow_empty_first_page:
            return 0
        hits = max(1, self.count - self.orphans)
        return ceil(hits / self.per_page)

    def page(self, number):
        number = self.validate_number(number)   # 여기서 count 사용
        bottom = (number - 1) * self.per_page
        top = bottom + self.per_page
        if top + self.orphans >= self.count:
            top = self.count                    # 마지막 페이지는 끝을 count로 자름
        return self._get_page(self.object_list[bottom:top], number, self)

이 코드에서 세 가지를 읽을 수 있습니다.

• QuerySet을 넘기면 .count()로 DB에서 개수만 셉니다. 반면 Paginator(list(qs), 10)처럼 리스트를 넘기면 len()을 쓰므로, 리스트로 바꾸는 순간 전체 행을 메모리에 올리게 됩니다.
• 글이 0개일 때는 max(1, ...) 덕분에 num_pages가 1이 되어, allow_empty_first_page=True(기본값)이면 빈 1페이지를 에러 없이 보여 줄 수 있습니다.
• page()도 매번 self.count를 참조하므로, 1페이지를 요청하든 abc를 넘기든 COUNT(*)는 항상 실행됩니다.

cached_property 캐시의 수명

django.utils.functional.cached_property는 처음 접근할 때 함수를 실행하고, 그 결과를 인스턴스의 __dict__에 같은 이름으로 저장합니다. 그 뒤로는 일반 속성처럼 __dict__에서 바로 꺼내므로 함수가 다시 실행되지 않습니다.

paginator = Paginator(qs, 10)
paginator.count       # SELECT COUNT(*) 실행 → __dict__["count"] = 95
paginator.count       # 쿼리 없음
paginator.num_pages   # count 캐시 사용 → 쿼리 없음 (num_pages도 캐시됨)

템플릿의 {{ page.paginator.num_pages }}도 뷰에서 만든 바로 그 인스턴스를 가리키므로 추가 쿼리가 없습니다. 실제로 한 인스턴스에서 count, num_pages, get_page(999), page(2)를 연달아 호출해도 COUNT(*)는 한 번만 실행됩니다.

다만 캐시의 수명은 "요청"이 아니라 "인스턴스"입니다. 보통 뷰 함수 안에서 요청마다 Paginator를 새로 만들기 때문에 결과적으로 요청 단위처럼 동작하는 것입니다.

요청 A (?page=2) → Paginator#1 생성 → COUNT 1회 → 응답 → 인스턴스 소멸
요청 B (?page=3) → Paginator#2 생성 → COUNT 1회 → 응답 → 인스턴스 소멸

Paginator를 모듈 전역에 하나 만들어 재사용하는 것은 피해야 합니다. count가 처음 값으로 고정되어 새 글이 생겨도 페이지 수가 바뀌지 않습니다.

요청을 넘어 COUNT를 아끼고 싶다면

count를 오버라이드해 Django 캐시 프레임워크(Redis, Memcached 등)에 잠깐 저장하면, 여러 요청이 하나의 COUNT(*) 결과를 공유할 수 있습니다.

from django.core.cache import cache
from django.core.paginator import Paginator
from django.utils.functional import cached_property


class CachedCountPaginator(Paginator):
    def __init__(self, *args, cache_key, timeout=60, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache_key = cache_key
        self.timeout = timeout

    @cached_property
    def count(self):
        value = cache.get(self.cache_key)
        if value is None:
            value = super().count          # 캐시에 없을 때만 실제 COUNT(*)
            cache.set(self.cache_key, value, self.timeout)
        return value

paginator = CachedCountPaginator(qs, 10, cache_key="post_list_count", timeout=60)

이렇게 하면 60초 동안 1, 2, 3페이지를 각각 다른 요청(다른 인스턴스)으로 조회해도 COUNT(*)는 처음 한 번만 실행됩니다. 인스턴스 안의 cached_property와 프로세스 밖의 캐시가 두 단계로 동작하는 구조입니다.

대신 두 가지 대가가 있습니다.

• 새 글이 생겨도 페이지 수는 최대 timeout만큼 늦게 반영됩니다. 그 사이 마지막 페이지 근처에서는 실제 개수와 맞지 않는 결과가 나올 수 있습니다.
• 검색어나 필터가 붙는 목록이라면 조건마다 캐시 키를 달리해야 합니다. 키가 같으면 다른 조건의 개수를 잘못 가져다 씁니다.

검색·필터·정렬이 붙은 QuerySet

Paginator는 넘겨받은 QuerySet을 그대로 .count()하고 슬라이싱할 뿐이므로, 검색 조건과 정렬은 COUNT와 데이터 쿼리 양쪽에 그대로 들어갑니다. count, num_pages, 범위 검증 모두 "검색 결과 기준"으로 계산됩니다.

qs = Post.objects.filter(title__icontains="django").order_by("-created_at")
page = Paginator(qs, 10).get_page(3)

SELECT COUNT(*) FROM post WHERE title LIKE '%django%';
SELECT ... FROM post WHERE title LIKE '%django%'
ORDER BY created_at DESC LIMIT 10 OFFSET 20;

Paginator 자체는 문제가 없지만, 검색이나 정렬이 붙을 때 실수하기 쉬운 지점이 다섯 가지 있습니다.

1. 페이지 링크에서 검색 조건이 빠지는 문제

가장 흔한 버그입니다. 서버는 상태를 기억하지 않으므로, ?q=django&page=1에서 다음 링크를 ?page=2로만 만들면 2페이지에서는 검색 조건이 사라진 전체 목록이 나옵니다. Django 5.1부터 제공되는 querystring 템플릿 태그를 쓰면 기존 파라미터를 유지한 채 page만 바꿀 수 있습니다.

<a href="{% querystring page=page.next_page_number %}">다음</a>
{# 현재 URL이 ?q=django&sort=-likes&page=2 이면 → ?q=django&sort=-likes&page=3 #}

2. 정렬 기준에 동점이 있는 경우

order_by("-likes")처럼 값이 겹칠 수 있는 필드로만 정렬하면, 좋아요 수가 같은 글들 사이의 순서는 DB가 보장하지 않습니다. 그러면 같은 글이 1페이지와 2페이지에 모두 나오거나 빠질 수 있습니다. 이때는 정렬이 있으므로 UnorderedObjectListWarning도 뜨지 않습니다. 마지막에 고유한 필드를 붙여 순서를 확정하면 해결됩니다.

qs = qs.order_by("-likes", "-id")

3. 사용자가 정렬 기준을 고르는 경우

?sort= 값을 order_by()에 그대로 넣으면 없는 필드일 때 FieldError로 500 에러가 나고, 노출하고 싶지 않은 필드로도 정렬할 수 있습니다. 허용 목록으로 거르는 게 안전합니다.

ALLOWED_SORTS = {
    "latest": ("-created_at", "-id"),
    "popular": ("-likes", "-id"),
}
ordering = ALLOWED_SORTS.get(request.GET.get("sort"), ALLOWED_SORTS["latest"])
qs = qs.order_by(*ordering)

4. 다대다 조건으로 행이 중복되는 경우

filter(tags__name__in=["python", "django"])처럼 다대다 관계로 필터링하면 JOIN 때문에 두 태그를 모두 가진 글이 두 번 나옵니다. COUNT도 부풀려져, 실제로는 5개인 글이 10개로 계산됩니다. .distinct()를 붙이면 Django는 SELECT COUNT(*) FROM (SELECT DISTINCT ...) subquery 형태로 중복을 제거한 개수를 셉니다.

5. 검색 조건이 무거울수록 COUNT도 무거워짐

icontains는 LIKE '%...%'가 되어 일반 B-tree 인덱스를 쓰지 못하는데, 이 비용을 COUNT와 데이터 쿼리가 각각 치릅니다. 대용량에서 검색이 느리다면 페이지네이션보다 검색 방식(PostgreSQL 전문 검색, trigram 인덱스 등)을 먼저 검토하는 게 맞습니다. 앞 절의 CachedCountPaginator를 검색 목록에 쓴다면 캐시 키에 검색어와 필터 조건을 반드시 포함해야 합니다.

OFFSET 방식의 한계와 커서 페이지네이션

Paginator의 LIMIT/OFFSET 방식은 단순하고 "5페이지로 바로 가기"가 가능하지만, 데이터가 커지거나 자주 바뀌면 세 가지 문제가 드러납니다.

1. 매 요청마다 COUNT(*): 범위 검증과 num_pages 계산에 필수라서 생략할 수 없습니다. 행이 수백만 건이면 이 쿼리 자체가 병목이 됩니다.
2. 뒤 페이지일수록 느려짐: OFFSET 1000000이면 DB는 앞의 100만 행을 읽고 버린 뒤에야 결과를 돌려줍니다.
3. 목록이 바뀌면 중복·누락: 1페이지를 본 뒤 새 글이 하나 올라오면 모든 글이 한 칸씩 밀립니다. 그 상태로 2페이지를 요청하면 방금 1페이지 마지막에 본 글이 2페이지 첫 글로 또 나옵니다. 삭제가 일어나면 반대로 한 글을 건너뜁니다.

커서(keyset) 페이지네이션

커서 방식은 "몇 개를 건너뛸지" 대신 "마지막으로 본 값 다음부터"를 요청합니다. DRF의 CursorPagination이 이 방식을 제공합니다.

from rest_framework.pagination import CursorPagination

class PostCursorPagination(CursorPagination):
    page_size = 10
    ordering = "-created"     # 기본값도 "-created"

첫 요청과 next 커서를 따라간 두 번째 요청에서 실제로 생성된 SQL은 다음과 같습니다.

-- GET /api/posts/
SELECT ... FROM post ORDER BY created DESC LIMIT 11;

-- GET /api/posts/?cursor=cD04Ng%3D%3D   (p=86: 마지막으로 본 값)
SELECT ... FROM post WHERE created < 86 ORDER BY created DESC LIMIT 11;

COUNT(*)도 OFFSET도 없습니다. LIMIT 11은 10개를 보여 주고 1개를 더 읽어 다음 페이지가 있는지 확인하는 용도입니다. WHERE created < 86은 정렬 필드에 인덱스가 있으면 바로 그 위치로 찾아가므로, 몇 번째 페이지든 속도가 거의 같습니다. 앞에 새 글이 생겨도 기준 값이 고정돼 있어 중복이 생기지 않습니다.

대신 응답에는 next, previous, results만 있고 count가 없습니다. 전체 페이지 수를 알 수 없고 "5페이지로 바로 가기"도 불가능합니다. 또 DRF 문서는 정렬 기준으로 고유하고 값이 바뀌지 않는 필드(생성 시각, 자동 증가 ID 등)를 쓰도록 권장합니다.

정리

Django의 페이지네이션은 "페이지 번호를 LIMIT/OFFSET으로 바꾸고, COUNT(*)로 범위를 검증하는 도우미"입니다.

• 클라이언트는 페이지 번호만 보내고, 서버는 매 요청마다 (번호 - 1) × 크기만큼 건너뛰어 크기만큼 잘라 돌려줍니다. 마지막 페이지는 남은 개수만큼만 조회합니다.
• 범위를 검증하려면 전체 개수가 필요하므로, 요청마다 COUNT(*)가 실행됩니다. 이 값은 cached_property로 Paginator 인스턴스 하나에 한 번만 계산됩니다.
• 데이터 조회 SQL은 Page를 순회하는 시점에 실행되고, 범위를 벗어난 번호로는 데이터 쿼리가 나가지 않습니다.
• 잘못된 번호는 get_page()가 보정하고(1 미만·범위 초과 → 마지막 페이지, 숫자 아님 → 1페이지), page()는 예외를, ListView와 DRF는 404를 돌려줍니다.
• 검색·필터·정렬은 COUNT와 데이터 쿼리에 그대로 반영됩니다. 페이지 링크에 검색 조건을 유지하고, 고유한 필드를 덧붙여 정렬을 확정하는 것이 핵심입니다.
• 대용량이거나 자주 바뀌는 목록, 무한 스크롤에는 COUNT와 OFFSET이 없는 커서 페이지네이션이 더 잘 맞습니다.

검증 환경: Django 6.1.3, Django REST framework 3.18.3, SQLite. SQL은 CaptureQueriesContext로 캡처한 실제 쿼리에서 컬럼 목록만 ...로 줄였습니다.

서버를 운영하다 보면 ss -tan을 쳐봤을 때 ESTABLISHED, TIME_WAIT, CLOSE_WAIT 같은 상태값이 줄줄이 나오는 걸 보게 된다. 이 상태값들은 사실 TCP 연결이 어떻게 맺어지고 어떻게 끊어지는가를 그대로 반영한 것이다.

이 글에서는 TCP 연결의 시작인 3-way handshake와 종료인 4-way handshake를 정리하고, 그 과정에서 등장하는 상태 전이를 백엔드 개발자 관점에서 짚어본다.

들어가기 전에: TCP는 왜 핸드셰이크가 필요한가

TCP는 연결 지향(connection-oriented) 프로토콜이다. 데이터를 보내기 전에 양쪽이 "우리 이제 통신할 준비 됐어"라는 합의를 먼저 하고, 통신이 끝나면 "이제 그만 보낼게"라는 작별 인사를 한다.

이게 왜 필요할까. TCP가 보장하려는 것들 때문이다.

• 신뢰성: 보낸 데이터가 누락 없이, 순서대로 도착해야 한다.
• 순서 보장: 패킷마다 시퀀스 번호(sequence number)를 매겨 순서를 추적한다.
• 흐름 제어: 상대가 받을 수 있는 만큼만 보낸다.

이 모든 것의 출발점이 시퀀스 번호의 초기값을 서로 교환하고 동기화하는 것이다. 핸드셰이크는 본질적으로 "데이터 보내기 전에 시작 번호를 맞추는 절차"라고 이해하면 깔끔하다.

여기서 자주 등장할 TCP 헤더의 플래그(flag)를 먼저 정리하고 간다.

플래그 의미

3-way handshake: 연결의 시작

클라이언트가 서버에 연결을 맺는 과정이다. 이름 그대로 패킷 3개가 오간다.

   Client                                  Server
     |                                       |
     |  ---------- 1. SYN ---------------->   |
     |          seq = x                      |
     |                                       |
     |  <------- 2. SYN + ACK -------------   |
     |          seq = y, ack = x + 1          |
     |                                       |
     |  ---------- 3. ACK ---------------->   |
     |          ack = y + 1                  |
     |                                       |
   [ESTABLISHED]                       [ESTABLISHED]

1단계 — SYN (클라이언트 → 서버)

클라이언트가 "연결하고 싶다"고 요청한다. 이때 자신의 초기 시퀀스 번호(ISN, Initial Sequence Number) x를 담아 보낸다. ISN은 보안상 예측 불가능하도록 랜덤하게 생성된다.

2단계 — SYN + ACK (서버 → 클라이언트)

서버는 두 가지 일을 한 번에 한다.

• 클라이언트의 SYN을 잘 받았다는 확인(ACK). ack = x + 1을 보내 "다음엔 x+1번부터 받을게"라고 알린다.
• 동시에 서버도 자신의 초기 시퀀스 번호 y를 담은 SYN을 보낸다.

서버 입장에서 SYN과 ACK를 굳이 따로 보낼 이유가 없으니 하나의 패킷으로 합친다. 3-way가 4-way가 아닌 이유가 바로 여기에 있다.

3단계 — ACK (클라이언트 → 서버)

클라이언트가 서버의 SYN에 대한 확인(ack = y + 1)을 보낸다. 이 패킷이 도착하면 양쪽 모두 상대의 ISN을 알게 되었고, 연결이 완성된다.

상태 전이로 본 3-way handshake

   Client                                  Server
   CLOSED                                  LISTEN        ← 서버는 미리 대기 중
     |                                       |
   SYN_SENT  ---------- SYN ------------>     |
     |                                  SYN_RECEIVED
     |  <------- SYN + ACK ----------------   |
   ESTABLISHED                                |
     |  ---------- ACK ----------------->     |
     |                                  ESTABLISHED

• 서버는 socket → bind → listen을 마치고 LISTEN 상태로 대기한다.
• 클라이언트가 connect를 호출하면 SYN_SENT로 진입한다.
• 서버는 SYN을 받으면 SYN_RECEIVED가 되고, 마지막 ACK를 받으면 양쪽 모두 ESTABLISHED가 된다.

백엔드 관점 한 줄: 서버의 SYN_RECEIVED 상태를 악용하는 것이 바로 SYN Flood 공격이다. 공격자가 SYN만 잔뜩 보내고 마지막 ACK를 보내지 않으면 서버의 백로그 큐가 가득 차 정상 연결을 못 받게 된다. 리눅스의 tcp_syncookies가 이를 완화하는 대표적인 방어책이다.

4-way handshake: 연결의 종료

연결을 끊을 때는 패킷이 4개 오간다. 시작보다 한 단계가 더 많은데, 그 이유가 이 글의 핵심 포인트다.

   Client                                  Server
 (active close)                        (passive close)
     |                                       |
     |  ---------- 1. FIN ---------------->   |
   FIN_WAIT_1                            CLOSE_WAIT
     |                                       |
     |  <--------- 2. ACK -----------------   |
   FIN_WAIT_2                                 |
     |                              (남은 데이터 전송...)
     |  <--------- 3. FIN -----------------   |
     |                                  LAST_ACK
   TIME_WAIT                                  |
     |  ---------- 4. ACK ---------------->   |
     |                                    CLOSED
   (2MSL 대기 후)
   CLOSED

왜 3-way가 아니라 4-way인가

3-way handshake에서는 서버가 SYN과 ACK를 하나로 합칠 수 있었다. 하지만 종료할 때는 그게 안 된다.

연결을 먼저 끊자고 한 쪽(active close, 보통 클라이언트)이 FIN을 보내도, 반대쪽은 아직 보낼 데이터가 남아 있을 수 있다. TCP 연결은 양방향(full-duplex)이라 각 방향을 독립적으로 닫아야 하기 때문이다.

그래서 종료는 이렇게 진행된다.

1. FIN (Client → Server): "나는 보낼 데이터 다 끝났어."
2. ACK (Server → Client): "알았어. 근데 나는 아직 보낼 게 남았을 수도 있어." — 이 시점이 half-close 상태다. 클라이언트→서버 방향만 닫혔다.
3. FIN (Server → Client): 서버가 남은 데이터를 다 보낸 뒤 "이제 나도 끝났어."
4. ACK (Client → Server): "확인했어. 잘 가."

서버의 ACK(2단계)와 FIN(3단계) 사이에 시간 간격이 존재할 수 있기 때문에 둘을 합칠 수 없고, 그래서 4-way가 된다. (남은 데이터가 정말 없다면 구현에 따라 ACK와 FIN이 합쳐져 3개로 끝나기도 한다.)

상태 전이로 본 4-way handshake

연결을 먼저 닫는 쪽(active close)과 받는 쪽(passive close)의 상태가 다르다.

Active close (먼저 끊는 쪽)

• FIN_WAIT_1: FIN을 보내고 ACK를 기다림
• FIN_WAIT_2: 상대의 ACK를 받음. 상대의 FIN을 기다림
• TIME_WAIT: 상대의 FIN을 받고 마지막 ACK를 보낸 뒤, 일정 시간 대기

Passive close (끊김을 당하는 쪽)

• CLOSE_WAIT: 상대의 FIN을 받고 ACK를 보냄. 내 쪽 FIN은 아직
• LAST_ACK: 내 FIN을 보내고 마지막 ACK를 기다림
• CLOSED: 마지막 ACK를 받고 연결 종료

실무에서 마주치는 두 가지 상태: TIME_WAIT과 CLOSE_WAIT

이 두 상태는 서버 운영 중 ss나 netstat로 자주 보게 되는데, 의미가 완전히 다르다.

TIME_WAIT — 정상이지만 쌓이면 문제

연결을 먼저 끊은 쪽에서 나타나는 상태다. 마지막 ACK를 보낸 뒤 바로 사라지지 않고 2MSL(Maximum Segment Lifetime의 2배) 동안 머무른다. 리눅스에서는 보통 60초다.

왜 굳이 기다릴까. 두 가지 이유다.

1. 마지막 ACK가 유실될 경우를 대비한다. 상대가 ACK를 못 받으면 FIN을 재전송하는데, 이쪽이 이미 CLOSED 되어 버렸다면 응답해 줄 주체가 없다. TIME_WAIT으로 남아 있어야 재전송된 FIN에 다시 ACK를 보내줄 수 있다.
2. 이전 연결의 패킷이 새 연결에 섞이는 것을 방지한다. 같은 (IP, port) 조합으로 새 연결이 곧바로 생기면, 네트워크를 떠돌던 옛 패킷이 새 연결로 잘못 배달될 수 있다. 2MSL은 네트워크상의 모든 잔존 패킷이 소멸하기에 충분한 시간이다.

TIME_WAIT이 쌓이는 문제: 짧은 연결을 대량으로 맺고 끊는 서버(예: 다른 서비스로 HTTP 요청을 폭발적으로 보내는 경우)에서는 클라이언트 측 로컬 포트가 고갈될 수 있다. 각 TIME_WAIT이 포트 하나를 60초간 점유하기 때문이다.

대응책:

• Keep-Alive / 커넥션 풀 사용: 매번 연결을 새로 맺지 말고 재사용한다. httpx.AsyncClient를 요청마다 만들지 않고 애플리케이션 수명 동안 하나만 두는 것이 대표적인 예다.
• net.ipv4.ip_local_port_range 확대
• tcp_tw_reuse 활성화 (아웃바운드 연결에 한해 TIME_WAIT 소켓 재사용 허용)

CLOSE_WAIT — 거의 항상 버그 신호

CLOSE_WAIT은 상대가 FIN을 보냈는데, 내 애플리케이션이 아직 close()를 호출하지 않은 상태다.

상태 전이를 다시 보면, passive close 쪽은 FIN을 받으면 커널이 자동으로 ACK를 보내 CLOSE_WAIT으로 간다. 그다음 LAST_ACK로 넘어가려면 애플리케이션이 직접 소켓을 닫아줘야 한다.

즉 CLOSE_WAIT이 계속 쌓인다면, 코드 어딘가에서 소켓(또는 응답 객체, DB 커넥션 등)을 닫지 않고 누수시키고 있다는 강력한 신호다.

• TIME_WAIT은 커널이 알아서 정리한다. → 보통 정상.
• CLOSE_WAIT은 애플리케이션이 닫아야 사라진다. → 보통 버그.

ss -tan state close-wait 결과가 계속 증가한다면 리소스 정리 로직(async with, finally, 컨텍스트 매니저)을 점검해야 한다.

직접 관찰해 보기

개념을 코드와 명령어로 확인해 보면 훨씬 분명해진다.

연결 상태를 상태별로 집계해서 보는 명령:

ss -tan | awk 'NR>1 {print $1}' | sort | uniq -c | sort -rn

특정 상태의 소켓만 보기:

ss -tan state time-wait
ss -tan state close-wait

핸드셰이크 패킷을 직접 캡처하려면 tcpdump로 플래그를 따라가 보면 된다.

sudo tcpdump -i any -nn 'tcp port 8000 and (tcp[tcpflags] & (tcp-syn|tcp-fin) != 0)'

이렇게 하면 출력에서 [S](SYN), [S.](SYN+ACK), [.](ACK), [F.](FIN+ACK) 같은 플래그가 오가는 순서를 눈으로 확인할 수 있다. 3-way로 시작해서 4-way로 끝나는 흐름이 그대로 보인다.

정리

구분 3-way handshake 4-way handshake

핵심을 다시 짚으면 이렇다.

• 3-way가 3개인 이유: 서버가 SYN과 ACK를 한 패킷으로 합칠 수 있어서.
• 4-way가 4개인 이유: TCP는 양방향 연결이고, 한쪽이 끝나도 반대쪽은 보낼 데이터가 남아 있을 수 있어서 각 방향을 따로 닫아야 함.
• TIME_WAIT은 마지막 ACK 유실과 패킷 혼선을 막기 위한 정상적인 안전장치. 다만 단명 연결이 많으면 포트 고갈을 유발하므로 커넥션 풀로 대응.
• CLOSE_WAIT이 쌓이면 애플리케이션이 소켓을 안 닫고 있다는 버그 신호.

ss 출력에 찍히는 상태값 하나하나가 이 핸드셰이크 흐름의 한 지점이라는 걸 알고 나면, 서버 트러블슈팅이 훨씬 직관적으로 다가온다.

들어가며

async/await 구문을 사용하다 보면, await 키워드 뒤에 무엇이 올 수 있는지 정확히 이해하지 못한 채 코루틴 함수 호출만 기계적으로 넘기는 경우가 많다. 그러나 await이 받아들이는 대상은 코루틴 객체만이 아니다. Python은 이를 awaitable이라는 프로토콜로 일반화하고 있으며, 이 글에서는 awaitable의 정의, 종류, 그리고 실무에서의 활용 패턴을 정리한다.

Awaitable이란 무엇인가

Python 공식 문서의 정의는 명확하다. awaitable 객체란 __await__() 메서드를 구현한 객체이다. await 표현식은 내부적으로 대상 객체의 __await__()를 호출하여 이터레이터를 얻고, 이를 통해 이벤트 루프와 값을 주고받는다.

result = await some_awaitable
# 내부적으로는 some_awaitable.__await__()가 호출된다.

collections.abc.Awaitable 추상 베이스 클래스가 이 프로토콜을 정의하고 있으며, isinstance() 검사에도 활용할 수 있다.

from collections.abc import Awaitable

print(isinstance(some_object, Awaitable))

단, 코루틴 객체는 __await__() 메서드를 C 레벨에서 구현하므로, hasattr() 검사보다는 isinstance() 또는 inspect.isawaitable()을 사용하는 것이 정확하다.

Awaitable의 세 가지 유형

Python에서 awaitable 객체는 크게 세 가지로 분류된다.

1. 코루틴 객체 (Coroutine)

async def로 정의한 코루틴 함수를 호출하면 코루틴 객체가 반환된다. 이것이 가장 흔한 awaitable이다.

import asyncio

async def fetch_data() -> str:
    await asyncio.sleep(1)
    return "data"

async def main():
    # fetch_data()는 코루틴 객체를 반환한다.
    coro = fetch_data()
    print(type(coro))  # <class 'coroutine'>

    # await로 실행을 완료하고 결과를 받는다.
    result = await coro
    print(result)       # "data"

asyncio.run(main())

주의할 점은, async def 함수 자체는 awaitable이 아니라는 것이다. 호출한 결과가 awaitable이다.

# 잘못된 사용
await fetch_data   # TypeError: 함수 객체는 awaitable이 아니다.

# 올바른 사용
await fetch_data() # 호출하여 코루틴 객체를 만든 뒤 await한다.

2. Task

asyncio.Task는 코루틴을 이벤트 루프에 등록하여 동시 실행을 스케줄링하는 래퍼이다. Future의 서브클래스이므로 당연히 awaitable이다.

import asyncio

async def download(url: str) -> str:
    await asyncio.sleep(1)  # 네트워크 I/O 시뮬레이션
    return f"content from {url}"

async def main():
    # Task를 생성하면 즉시 스케줄링된다.
    task1 = asyncio.create_task(download("https://a.com"))
    task2 = asyncio.create_task(download("https://b.com"))

    # 두 Task는 동시에 실행되며, await로 결과를 수집한다.
    result1 = await task1
    result2 = await task2
    print(result1, result2)

asyncio.run(main())

코루틴 객체를 직접 await하는 것과 Task로 감싼 뒤 await하는 것의 차이는 실행 시점에 있다.

3. Future

asyncio.Future는 아직 완료되지 않은 연산의 결과를 나타내는 저수준 객체이다. 이벤트 루프가 내부적으로 사용하며, 일반 애플리케이션 코드에서 직접 생성할 일은 드물다.

import asyncio

async def set_future_result(future: asyncio.Future):
    await asyncio.sleep(1)
    future.set_result("future resolved")

async def main():
    loop = asyncio.get_running_loop()
    future = loop.create_future()

    # Future에 결과를 설정하는 Task를 스케줄링한다.
    asyncio.create_task(set_future_result(future))

    # Future가 완료될 때까지 대기한다.
    result = await future
    print(result)  # "future resolved"

asyncio.run(main())

Future는 주로 콜백 기반 코드와 async/await 코드를 연결하는 브릿지 역할을 한다. 예를 들어 전통적인 콜백 API를 async 인터페이스로 감쌀 때 유용하다.

커스텀 Awaitable 만들기: __await__ 메서드

__await__() 메서드를 구현하면 어떤 객체든 awaitable로 만들 수 있다. 이 메서드는 이터레이터를 반환해야 한다.

기본 구조

class MyAwaitable:
    def __init__(self, value):
        self.value = value

    def __await__(self):
        yield  # 이벤트 루프에 제어를 한 번 양보한다. (뒤에서 원리를 상세히 다룬다.)
        return self.value

여기서 yield는 이벤트 루프에 제어를 넘기는 일시 정지 신호이다. return은 await 표현식의 결과값이 된다. yield의 정확한 동작 원리는 뒤의 "await의 내부 동작 원리" 섹션에서 상세히 다룬다.

import asyncio

async def main():
    result = await MyAwaitable(42)
    print(result)  # 42

asyncio.run(main())

실무 예시: 지연 실행 래퍼

실무에서 커스텀 awaitable이 유용한 대표적 사례는 지연 평가(lazy evaluation) 패턴이다.

import asyncio

class LazyQuery:
    """await할 때 비로소 쿼리를 실행하는 객체."""

    def __init__(self, collection: str, filter: dict):
        self.collection = collection
        self.filter = filter

    def __await__(self):
        return self._execute().__await__()

    async def _execute(self):
        # 실제 DB 쿼리 수행 (시뮬레이션)
        await asyncio.sleep(0.1)
        return {"collection": self.collection, "filter": self.filter, "results": []}

async def main():
    # 이 시점에서는 쿼리가 실행되지 않는다.
    query = LazyQuery("users", {"active": True})

    # await하는 순간 쿼리가 실행된다.
    result = await query
    print(result)

asyncio.run(main())

위 패턴에서 핵심은 self._execute().__await__()이다. 코루틴 객체의 __await__를 위임하는 방식으로, 내부 async 로직을 커스텀 awaitable에서 자연스럽게 활용할 수 있다. 이렇게 하면 yield를 직접 다룰 필요 없이, 기존의 검증된 async 라이브러리(Motor, aiohttp 등)가 제공하는 코루틴 체인에 자연스럽게 연결된다. Beanie ODM, SQLAlchemy 2.0 등 여러 라이브러리가 이 패턴을 채택하고 있다.

_execute().__await__()는 await _execute()와 같은가

결과적으로 하는 일은 같다. 하지만 사용할 수 있는 위치가 다르다.

await은 async def 안에서만 쓸 수 있다. 그런데 __await__는 일반 함수(def)이다.

class LazyQuery:
    # __await__는 async def가 아니라 일반 def이다.
    def __await__(self):
        await self._execute()               # ❌ SyntaxError: 일반 함수 안에서 await 사용 불가
        return self._execute().__await__()   # ✅ 이터레이터를 직접 꺼내서 반환

__await__를 async def로 선언하면 되지 않느냐는 생각이 들 수 있다.

class LazyQuery:
    async def __await__(self):              # ❌ __await__는 async def가 될 수 없다.
        return await self._execute()

이것도 불가능하다. Python 명세상 __await__는 반드시 일반 이터레이터를 반환해야 하며, async def로 선언하면 코루틴 객체를 반환하게 되어 프로토콜에 맞지 않는다.

그래서 self._execute().__await__()라는 표현이 필요한 것이다. 풀어서 쓰면 다음과 같다.

def __await__(self):
    coro = self._execute()       # 1. 코루틴 객체를 생성한다.
    iterator = coro.__await__()  # 2. 코루틴에서 이터레이터를 꺼낸다.
    return iterator              # 3. 그 이터레이터를 그대로 반환한다 (위임).

정리하면 다음과 같다.

두 표현 모두 최종적으로 같은 코루틴을 실행하지만, __await__ 메서드가 일반 함수여야 한다는 제약 때문에 후자의 패턴이 존재하는 것이다.

await의 내부 동작 원리: yield는 어디에 있는가

await을 사용할 때 자연스럽게 드는 의문이 있다. "이벤트 루프에 제어를 넘기는 yield는 대체 어디에 있는가?" 결론부터 말하면, yield는 사용자가 작성하는 코드에는 없다. asyncio 내부 깊숙한 곳에 딱 하나 숨어 있고, 여러 겹의 await이 그 yield까지 연결되는 구조이다.

await 체인: 바닥까지 따라가 보기

일반적인 async 코드를 예로 들어본다.

async def get_user():
    data = await fetch_from_db()    # await 1층
    return data

async def fetch_from_db():
    await asyncio.sleep(1)          # await 2층
    return {"name": "홍길동"}

여기서 await을 따라 안으로 계속 들어가면 다음과 같은 구조가 드러난다.

[내가 작성한 코드]
  await fetch_from_db()
    └─ await asyncio.sleep(1)            ← 여기까지는 전부 await

[asyncio 라이브러리 내부]
        └─ await future                  ← Future를 await

[Future.__await__ 내부]                  ← ★ 여기서 드디어 yield가 나온다
              └─ yield self              ← 이벤트 루프에 제어 양보

즉, await 체인의 맨 바닥에 yield가 하나 있는 것이다. CPython의 Future.__await__를 개념적으로 표현하면 다음과 같다.

# CPython의 Future.__await__를 단순화한 개념적 구현
class Future:
    def __await__(self):
        if not self.done():
            yield self      # ← asyncio에서 유일한 yield. 이벤트 루프에 자신을 넘기고 멈춘다.
        return self.result()

yield self가 하는 일은 딱 하나이다. "나(Future) 아직 안 끝났으니, 이벤트 루프에게 나를 넘기고 멈춘다." 이벤트 루프는 이 Future를 받아서 I/O 완료를 감시하다가, 완료되면 send()로 코루틴을 다시 깨운다.

yield에 I/O를 넣는 것이 아니다

흔한 오해 중 하나는 "yield 부분에서 I/O 동작을 수행한다"는 것이다. 그러나 yield 자체는 "여기서 멈추고 이벤트 루프에 제어를 돌려준다"는 일시 정지 신호일 뿐이다. 실제 I/O의 등록과 완료 감지는 이벤트 루프가 처리한다.

asyncio.sleep()의 내부 구조를 개념적으로 풀어보면 이 점이 명확해진다.

# asyncio.sleep()의 개념적 구현 (실제 CPython 코드를 단순화한 것)
async def sleep(delay):
    loop = asyncio.get_running_loop()
    future = loop.create_future()

    # 1단계: 이벤트 루프에 "delay초 후에 이 future를 완료시켜줘"라고 등록한다.
    loop.call_later(delay, future.set_result, None)

    # 2단계: future를 await한다.
    #   → future.__await__()가 호출된다.
    #   → future 내부의 __await__는 자신이 완료될 때까지 yield로 제어를 양보한다.
    return await future

핵심은 다음과 같다. yield 전에 이벤트 루프에 I/O를 등록하고, yield로 제어를 넘긴 뒤, 이벤트 루프가 I/O 완료를 감지하면 다시 깨워주는 구조이다.

yield는 딱 한 번만 실행된다

또 다른 흔한 오해는 "I/O가 오래 걸리면 yield를 여러 번 반복하며 기다리는 것 아닌가?"이다. 아니다. yield는 딱 한 번만 실행된다. I/O가 아무리 오래 걸려도 한 번이다.

yield 이후 코루틴은 완전히 멈춘 상태이다. "계속 yield를 반복하면서 다 됐나 확인한다"가 아니라, "한 번 멈추고, 누군가 깨워줄 때까지 아무것도 안 한다"이다.

DB 쿼리가 5초 걸리는 경우를 시간순으로 보면 이렇다.

[시각 0초]
코루틴:     await motor_collection.find_one({"id": 1})
              → 내부적으로 소켓에 쿼리 전송
              → 이벤트 루프에 "이 소켓에 데이터 오면 알려줘" 등록
              → yield self  ← 멈춤. 코루틴은 여기서 얼어붙는다.

[시각 0~5초]
코루틴:     (아무것도 안 함. 멈춰 있음)
이벤트 루프: epoll/kqueue로 소켓 감시 중. 그 사이에 다른 코루틴들 실행.

[시각 5초]
OS:         "소켓에 데이터 도착했다"
이벤트 루프: → 데이터 읽음
              → future.set_result(데이터)
              → 멈춰있던 코루틴에 send() 호출  ← 여기서 깨어남
코루틴:     yield 다음 줄부터 이어서 실행
              → return result

정리하면 다음과 같다.

바로 이것이 async가 효율적인 이유이다. 코루틴 하나가 I/O를 기다리는 동안 CPU를 전혀 쓰지 않으므로, 이벤트 루프가 그 시간에 다른 코루틴을 실행할 수 있다.

await 표현식의 동작 절차 (요약)

위 내용을 프로토콜 수준에서 정리하면, await 표현식은 다음과 같은 절차로 동작한다.

1. 대상 객체의 __await__()를 호출하여 이터레이터를 얻는다.
2. 이터레이터에 send(None)을 호출한다.
3. yield된 값(Future)은 이벤트 루프로 전달된다.
4. 이벤트 루프가 I/O 완료를 감지하면 send()로 이터레이터를 재개한다.
5. StopIteration 예외가 발생하면, 그 value 속성이 await 표현식의 결과가 된다.

일반 애플리케이션 코드에서 yield를 직접 다룰 일은 없다. await만 사용하면 된다. yield는 asyncio가 이벤트 루프와 소통하기 위해 내부적으로 사용하는 메커니즘이고, await은 그 yield까지 도달하는 통로 역할을 하는 것이다.

inspect.isawaitable()과 타입 검사

런타임에서 awaitable 여부를 판단해야 하는 경우, inspect 모듈을 사용한다.

import inspect
import asyncio

async def coro():
    pass

# 코루틴 객체
print(inspect.isawaitable(coro()))    # True

# Future
loop = asyncio.new_event_loop()
future = loop.create_future()
print(inspect.isawaitable(future))    # True

# 일반 함수 호출 결과
print(inspect.isawaitable(42))        # False

# async def 함수 자체 (호출하지 않은 상태)
print(inspect.isawaitable(coro))      # False

loop.close()

inspect.isawaitable()은 코루틴 객체, Future 객체, __await__ 메서드를 가진 객체를 모두 True로 판별한다. 미들웨어나 프레임워크 내부에서 "await 가능한 값이면 await하고, 아니면 그대로 반환"하는 패턴에 자주 사용된다.

import inspect

async def resolve(value):
    """awaitable이면 await하고, 아니면 그대로 반환한다."""
    if inspect.isawaitable(value):
        return await value
    return value

FastAPI의 의존성 주입 시스템도 이와 유사한 로직으로 동기/비동기 의존성을 모두 처리한다.

타이핑: Awaitable[T]

typing 모듈의 Awaitable[T]는 await 결과의 타입을 명시하는 제네릭 타입이다.

from collections.abc import Awaitable

async def process(source: Awaitable[str]) -> str:
    """어떤 awaitable이든 받아서 처리한다."""
    data = await source
    return data.upper()

Coroutine, Task, Future, 커스텀 awaitable 등 어떤 것이든 Awaitable[str]로 받을 수 있으므로, 함수의 유연성이 높아진다.

더 세분화된 타입도 존재한다.

from collections.abc import Coroutine

# Coroutine[YieldType, SendType, ReturnType]
async def fetch() -> str:
    return "data"

coro: Coroutine[None, None, str] = fetch()

대부분의 실무 코드에서는 Awaitable[T]만으로 충분하다. Coroutine의 YieldType, SendType까지 명시하는 경우는 프레임워크 내부 구현 정도에 한정된다.

자주 혼동되는 개념 정리

async def 함수 vs 코루틴 객체

async def my_func():
    return 1

# my_func      → 코루틴 함수 (callable, awaitable 아님)
# my_func()    → 코루틴 객체 (awaitable)

await과 asyncio.run()

await은 이미 실행 중인 이벤트 루프 안에서만 사용할 수 있다. 루프 바깥(동기 코드)에서 코루틴을 실행하려면 asyncio.run()을 사용한다.

# 동기 진입점
result = asyncio.run(my_func())  # 이벤트 루프를 생성하고 코루틴을 실행한다.

asyncio.gather()와 awaitable

asyncio.gather()는 여러 awaitable을 동시에 실행하고 결과를 모아 반환한다.

async def main():
    results = await asyncio.gather(
        fetch_data(),          # 코루틴 객체
        some_task,             # Task
        MyAwaitable(42),       # 커스텀 awaitable
    )

gather()는 전달받은 awaitable이 코루틴이면 자동으로 Task로 감싸서 스케줄링한다.

마무리

정리하면 다음과 같다.

await 키워드의 본질은 __await__() 프로토콜 호출이다. 이 사실을 이해하면 라이브러리가 제공하는 다양한 awaitable 객체의 동작 원리를 파악할 수 있고, 필요에 따라 직접 커스텀 awaitable을 설계하는 것도 가능해진다.

LLM(Large Language Model) 엔지니어링은 모델의 학습, 튜닝, 배포 및 실무 최적화 과정을 포괄한다. 본 포스팅에서는 LLM 분야에서 반드시 숙지해야 할 핵심 용어들을 정의하고 정리한다.

1. 모델 구성 요소 및 데이터 처리 단위

• 매개변수 (Parameters): 모델 내부의 가중치 집합을 의미한다. 인간의 뇌세포 연결 고리와 유사한 역할을 수행하며, 그 수가 많을수록 모델이 처리할 수 있는 정보의 복잡도와 논리 수준이 높아진다.
• 토큰 (Token): 모델이 텍스트를 인식하고 생성하는 최소 처리 단위이다. 단어 또는 형태소 단위로 분절되며, API 사용량 및 비용 산정의 기준이 된다.
• 컨텍스트 윈도우 (Context Window): 모델이 한 번의 추론 과정에서 기억하고 참조할 수 있는 최대 토큰 범위이다. 이 범위를 초과하는 입력값은 모델이 처리하지 못한다.

2. 모델 학습 및 최적화 기법

• 사전 학습 (Pre-training): 대규모 비정형 데이터를 모델에 주입하여 언어의 일반적인 패턴을 학습시키는 초기 단계이다.
• 파인 튜닝 (Fine-tuning): 사전 학습된 모델에 특정 도메인(의료, 법률, 금융 등)의 데이터를 추가 학습시켜 목적에 특화된 성능을 도출하는 기법이다.
• RLHF (Reinforcement Learning from Human Feedback): 인간의 피드백을 기반으로 모델을 강화 학습시키는 과정이다. 답변의 유해성을 줄이고 사용자 의도에 부합하도록 조정한다.

3. 신뢰성 확보 및 추론 기술

• 할루시네이션 (Hallucination): 모델이 학습하지 않은 내용을 사실인 것처럼 생성하거나 잘못된 정보를 제공하는 환각 현상을 의미한다.
• RAG (Retrieval-Augmented Generation): 외부 데이터베이스에서 관련 정보를 검색한 후, 이를 바탕으로 답변을 생성하는 기술이다. 모델의 최신성 유지와 할루시네이션 방지에 필수적이다.
• Chain-of-Thought (CoT): 복잡한 문제 해결을 위해 모델이 단계별로 사고하도록 유도하는 프롬프트 기법이다. 논리적 추론 능력을 극대화할 때 활용한다.

4. 아키텍처 및 시스템 구성

• 임베딩 (Embedding): 텍스트와 같은 비정형 데이터를 고차원의 수치 벡터로 변환하는 과정이다. 의미론적 유사성을 계산하는 기초가 된다.
• 벡터 데이터베이스 (Vector DB): 임베딩된 벡터 데이터를 효율적으로 저장하고 검색하기 위한 특수 목적용 데이터베이스이다. RAG 시스템 구축 시 핵심적인 역할을 수행한다.
• 에이전트 (Agent): LLM이 단순히 텍스트를 생성하는 수준을 넘어, 외부 도구(API, 검색, 코드 실행 등)를 스스로 판단하여 호출하고 작업을 완수하는 자율 시스템을 지칭한다.

요약
LLM 엔지니어링의 핵심은 모델의 물리적 체급(매개변수)을 이해하고, 효율적인 데이터 단위(토큰)를 관리하며, 신뢰할 수 있는 구조(RAG, CoT)를 설계하여 실무 서비스 환경(Vector DB, Agent)에 적용하는 것에 있다. 각 개념의 유기적인 관계를 파악하는 것이 엔지니어링의 시작이다.

1. 개요

FastAPI 애플리케이션을 프로덕션 환경에 배포할 때, ASGI 서버 선택은 성능과 안정성에 직접적인 영향을 미친다. 본 글에서는 Uvicorn 단일 프로세스 사용, Uvicorn 멀티 프로세스 사용, 그리고 Gunicorn을 프로세스 매니저로 사용하는 Gunicorn+Uvicorn 조합, 이 세 가지 방식을 비교 분석한다.

2. 각 방식의 특징

2.1 Uvicorn 단일 프로세스

Uvicorn을 워커 옵션 없이 실행하는 가장 기본적인 방식이다.

uvicorn main:app --host 0.0.0.0 --port 8000

주요 특징:

• 단일 프로세스에서 asyncio 이벤트 루프를 통해 비동기 요청을 처리한다.
• CPU 코어 하나만 사용한다.
• 메모리 사용량이 가장 적다.
• 개발 환경에서 자동 리로드 기능(--reload)을 제공한다.
• 설정이 단순하고 빠르게 시작할 수 있다.

2.2 Uvicorn 멀티 프로세스

Uvicorn에 --workers 옵션을 추가하여 여러 워커 프로세스를 실행하는 방식이다.

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

주요 특징:

• 내부적으로 Gunicorn을 사용하여 워커 프로세스를 관리한다.
• 여러 CPU 코어를 활용할 수 있다.
• 각 워커가 독립적으로 요청을 처리한다.
• 기본적인 프로세스 관리 기능을 제공한다.

중요한 사실: Uvicorn의 공식 문서를 보면, --workers 옵션을 사용할 때 실제로는 내부적으로 Gunicorn을 호출한다. 즉, Uvicorn 멀티 프로세스 모드는 사실상 Gunicorn을 래핑한 것이다.

2.3 Gunicorn + Uvicorn

Gunicorn을 명시적으로 사용하고 Uvicorn을 워커 클래스로 지정하는 방식이다.

gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000

주요 특징:

• Gunicorn의 모든 프로세스 관리 기능을 직접 제어할 수 있다.
• 워커 수명 주기 관리가 정교하다.
• Graceful reload를 지원한다.
• 워커별 헬스 체크와 자동 재시작이 가능하다.
• 설정 파일을 통한 세밀한 튜닝이 가능하다.

3. 상세 비교

3.1 아키텍처 차이

Uvicorn 단일 프로세스:

Uvicorn (단일 프로세스)
  └── asyncio 이벤트 루프
       └── 비동기 요청 처리

Uvicorn 멀티 프로세스:

Gunicorn 마스터 프로세스 (Uvicorn이 내부적으로 호출)
  ├── Uvicorn 워커 1
  ├── Uvicorn 워커 2
  ├── Uvicorn 워커 3
  └── Uvicorn 워커 4

Gunicorn + Uvicorn:

Gunicorn 마스터 프로세스 (명시적 제어)
  ├── UvicornWorker 1
  ├── UvicornWorker 2
  ├── UvicornWorker 3
  └── UvicornWorker 4

Uvicorn 멀티 프로세스와 Gunicorn+Uvicorn의 아키텍처는 유사하지만, 후자는 Gunicorn의 기능을 직접 제어할 수 있다는 점에서 차이가 있다.

3.2 성능 비교

CPU 활용:

• Uvicorn 단일 프로세스: CPU 코어 하나만 사용한다. 4코어 서버에서도 25% CPU만 활용한다.
• Uvicorn 멀티 프로세스: 워커 수만큼 CPU 코어를 활용한다. 4개 워커 = 4개 코어 활용.
• Gunicorn + Uvicorn: 워커 수만큼 CPU 코어를 활용한다. 4개 워커 = 4개 코어 활용.

처리량 (동일 서버 환경):

단일 프로세스:    1,000 req/s
멀티 프로세스 (4):  3,500 req/s
Gunicorn+Uvicorn (4): 3,500 req/s

성능 자체는 Uvicorn 멀티 프로세스와 Gunicorn+Uvicorn이 거의 동일하다. 차이는 프로세스 관리 기능에 있다.

메모리 사용량:

• Uvicorn 단일: 약 50MB
• Uvicorn 멀티 (4워커): 약 200MB
• Gunicorn+Uvicorn (4워커): 약 200MB

3.3 프로세스 관리 기능 비교

이 부분이 세 방식의 가장 큰 차이점이다.

Uvicorn 단일 프로세스

uvicorn main:app --host 0.0.0.0 --port 8000

제공 기능:

• 기본적인 프로세스 실행만 가능하다.
• 프로세스가 죽으면 서비스가 중단된다.
• 코드 업데이트 시 서비스 중단이 불가피하다.

필요한 외부 도구:

• systemd, supervisor 등의 프로세스 매니저가 필수적이다.

Uvicorn 멀티 프로세스

uvicorn main:app --workers 4

제공 기능:

• 워커 프로세스 생성 및 기본 관리
• 워커가 죽으면 자동 재시작
• SIGTERM 시그널로 종료 가능

제한 사항:

• Graceful reload 기능이 제한적이다.
• 워커별 세밀한 제어가 어렵다.
• 설정 옵션이 제한적이다.
• 워커 수명 주기 관리 기능이 부족하다.

Gunicorn + Uvicorn

gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000

제공 기능:

• Graceful reload: 무중단으로 코드 업데이트가 가능하다.
• 워커 수명 주기 관리: 메모리 누수 방지를 위해 일정 요청 수 처리 후 워커를 재시작한다.
• 헬스 체크: 응답 없는 워커를 감지하고 재시작한다.
• 시그널 처리: 다양한 시그널(SIGHUP, SIGTERM, SIGINT 등)을 지원한다.
• 설정 파일 지원: Python 설정 파일로 모든 옵션을 관리할 수 있다.
• 로깅 제어: 워커별 로그 관리가 가능하다.

3.4 실전 예제: Graceful Reload

프로덕션 환경에서 가장 중요한 기능 중 하나가 무중단 배포이다.

Uvicorn 멀티 프로세스의 한계

# 서비스 실행
uvicorn main:app --workers 4

# 코드 업데이트 후 재시작 필요
# 방법 1: 프로세스 종료 후 재시작 (서비스 중단 발생)
kill -TERM <pid>
uvicorn main:app --workers 4

# 방법 2: systemd 사용 (짧은 중단 발생)
systemctl restart myapp

두 방법 모두 서비스 중단이 발생한다.

Gunicorn + Uvicorn의 Graceful Reload

# 서비스 실행
gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000

# 코드 업데이트 후 무중단 재시작
kill -HUP <gunicorn_master_pid>

동작 방식:

1. Gunicorn 마스터 프로세스가 SIGHUP 시그널을 받는다.
2. 새로운 워커 프로세스들을 시작한다 (업데이트된 코드로).
3. 기존 워커들은 현재 처리 중인 요청을 완료한다.
4. 기존 워커들이 종료된다.
5. 서비스 중단 없이 업데이트가 완료된다.

3.5 워커 수명 주기 관리

Python 애플리케이션에서 메모리 누수는 흔한 문제다. 장시간 실행되는 워커는 메모리를 계속 소비할 수 있다.

Gunicorn의 워커 수명 주기 관리

# gunicorn.conf.py
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
max_requests = 1000  # 워커가 1000개 요청 처리 후 재시작
max_requests_jitter = 50  # 랜덤하게 950~1050 사이에서 재시작

동작 방식:

• 각 워커는 설정된 요청 수를 처리한 후 자동으로 재시작된다.
• jitter를 추가하여 모든 워커가 동시에 재시작되는 것을 방지한다.
• 메모리 누수를 주기적으로 초기화할 수 있다.

Uvicorn 멀티 프로세스에는 이러한 기능이 없다.

3.6 설정 및 실행 방법

Uvicorn 단일 프로세스

# 기본 실행
uvicorn main:app --host 0.0.0.0 --port 8000

# 개발 환경 (자동 리로드)
uvicorn main:app --reload

# SSL 지원
uvicorn main:app --ssl-keyfile key.pem --ssl-certfile cert.pem

Uvicorn 멀티 프로세스

# 기본 실행
uvicorn main:app --workers 4

# 환경 변수와 함께
uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000

제한 사항:

• --reload와 --workers를 동시에 사용할 수 없다.
• 설정 파일을 사용할 수 없다.

Gunicorn + Uvicorn

커맨드라인:

gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --access-logfile - \
  --error-logfile - \
  --timeout 30 \
  --graceful-timeout 30 \
  --keep-alive 5

설정 파일 (gunicorn.conf.py):

import multiprocessing

# 서버 소켓
bind = "0.0.0.0:8000"
backlog = 2048

# 워커 프로세스
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 50

# 타임아웃
timeout = 30
graceful_timeout = 30
keepalive = 5

# 로깅
accesslog = "-"
errorlog = "-"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# 프로세스 이름
proc_name = "fastapi_app"

# 서버 훅
def on_starting(server):
    print("서버 시작 중...")

def on_reload(server):
    print("워커 리로드 중...")

def worker_int(worker):
    print(f"워커 {worker.pid} 인터럽트됨")

실행:

gunicorn -c gunicorn.conf.py main:app

4. 시그널 처리 비교

프로덕션 환경에서는 다양한 시그널을 통해 서버를 제어한다.

4.1 Gunicorn + Uvicorn의 시그널 처리

# Graceful reload (무중단 재시작)
kill -HUP <master_pid>

# Graceful shutdown (현재 요청 완료 후 종료)
kill -TERM <master_pid>

# 즉시 종료
kill -QUIT <master_pid>

# 워커 수 증가
kill -TTIN <master_pid>

# 워커 수 감소
kill -TTOU <master_pid>

4.2 Uvicorn 멀티 프로세스의 시그널 처리

# 종료 (graceful 여부는 제한적)
kill -TERM <master_pid>

# 즉시 종료
kill -KILL <master_pid>

Uvicorn 멀티 프로세스는 기본적인 시그널만 지원한다.

5. 사용 시나리오별 권장사항

5.1 개발 환경

권장: Uvicorn 단일 프로세스

uvicorn main:app --reload --host 0.0.0.0 --port 8000

이유:

• 자동 리로드 기능으로 개발 생산성이 높다.
• 디버깅이 단순하다.
• 리소스 사용량이 적다.

5.2 소규모 프로덕션 (일일 방문자 < 1,000)

권장: Uvicorn 단일 프로세스 + systemd

트래픽이 적고 단순한 API 서버라면 단일 프로세스로도 충분하다.

systemd 설정 (/etc/systemd/system/myapp.service):

[Unit]
Description=FastAPI Application
After=network.target

[Service]
Type=simple
User=www-data
Group=www-data
WorkingDirectory=/var/www/myapp
Environment="PATH=/var/www/myapp/venv/bin"
ExecStart=/var/www/myapp/venv/bin/uvicorn main:app --host 127.0.0.1 --port 8000
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

실행:

sudo systemctl daemon-reload
sudo systemctl enable myapp
sudo systemctl start myapp

5.3 중규모 프로덕션 (일일 방문자 1,000 ~ 50,000)

권장: Uvicorn 멀티 프로세스 또는 Gunicorn + Uvicorn

이 규모에서는 두 방식 모두 사용 가능하다. 단, 무중단 배포가 필요하다면 Gunicorn+Uvicorn을 선택해야 한다.

Uvicorn 멀티 프로세스:

uvicorn main:app --workers 4 --host 127.0.0.1 --port 8000

Gunicorn + Uvicorn:

gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 127.0.0.1:8000

5.4 대규모 프로덕션 (일일 방문자 > 50,000)

권장: Gunicorn + Uvicorn + 설정 파일

세밀한 튜닝과 안정적인 운영이 필요하다.

gunicorn.conf.py:

import multiprocessing

workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
bind = "127.0.0.1:8000"

# 워커 수명 주기
max_requests = 1000
max_requests_jitter = 50

# 타임아웃
timeout = 30
graceful_timeout = 30
keepalive = 5

# 연결 설정
worker_connections = 1000
backlog = 2048

# 로깅
accesslog = "/var/log/myapp/access.log"
errorlog = "/var/log/myapp/error.log"
loglevel = "warning"

실행:

gunicorn -c gunicorn.conf.py main:app

5.5 WebSocket이 필요한 경우

권장: Gunicorn + Uvicorn

WebSocket 연결은 장시간 유지되므로, 워커 수명 주기 관리가 중요하다.

주의사항:

# gunicorn.conf.py
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
timeout = 300  # WebSocket 연결을 위해 타임아웃 증가
graceful_timeout = 60

# WebSocket 연결이 있는 경우 max_requests 주의
# 연결 중인 워커가 재시작되면 연결이 끊어짐
max_requests = 5000
max_requests_jitter = 100

Nginx 설정:

location /ws {
    proxy_pass http://127.0.0.1:8000;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_read_timeout 300s;
}

6. 실전 배포 예제

6.1 Nginx + Gunicorn + Uvicorn 구성

디렉토리 구조:

/var/www/myapp/
├── main.py
├── gunicorn.conf.py
├── requirements.txt
└── venv/

Nginx 설정 (/etc/nginx/sites-available/myapp):

upstream fastapi_backend {
    server 127.0.0.1:8000 fail_timeout=0;
}

server {
    listen 80;
    server_name example.com;
    
    # SSL 리다이렉트
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name example.com;
    
    ssl_certificate /etc/letsencrypt/live/example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/example.com/privkey.pem;
    
    # 보안 헤더
    add_header Strict-Transport-Security "max-age=31536000" always;
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    
    # 정적 파일
    location /static {
        alias /var/www/myapp/static;
        expires 30d;
    }
    
    # API 요청
    location / {
        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # 타임아웃
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }
}

Gunicorn 설정 (gunicorn.conf.py):

import multiprocessing
import os

# 워커 설정
workers = int(os.getenv("WORKERS", multiprocessing.cpu_count() * 2 + 1))
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000

# 바인딩
bind = "127.0.0.1:8000"
backlog = 2048

# 워커 수명 주기
max_requests = 1000
max_requests_jitter = 50
timeout = 30
graceful_timeout = 30
keepalive = 5

# 로깅
accesslog = "/var/log/myapp/access.log"
errorlog = "/var/log/myapp/error.log"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# 프로세스 이름
proc_name = "myapp"

# 보안
limit_request_line = 4094
limit_request_fields = 100
limit_request_field_size = 8190

systemd 설정 (/etc/systemd/system/myapp.service):

[Unit]
Description=FastAPI Application with Gunicorn
After=network.target

[Service]
Type=notify
User=www-data
Group=www-data
RuntimeDirectory=gunicorn
WorkingDirectory=/var/www/myapp
Environment="PATH=/var/www/myapp/venv/bin"
Environment="WORKERS=4"
ExecStart=/var/www/myapp/venv/bin/gunicorn -c gunicorn.conf.py main:app
ExecReload=/bin/kill -s HUP $MAINPID
KillMode=mixed
TimeoutStopSec=5
PrivateTmp=true
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

배포 스크립트 (deploy.sh):

#!/bin/bash
set -e

APP_DIR="/var/www/myapp"
VENV_DIR="$APP_DIR/venv"

echo "코드 업데이트 중..."
cd $APP_DIR
git pull origin main

echo "의존성 설치 중..."
$VENV_DIR/bin/pip install -r requirements.txt

echo "Graceful reload 수행 중..."
sudo systemctl reload myapp

echo "배포 완료!"

무중단 배포 실행:

./deploy.sh

6.2 성능 테스트

부하 테스트를 통해 최적의 워커 수를 찾아야 한다.

테스트 도구 설치:

pip install locust

테스트 시나리오 (locustfile.py):

from locust import HttpUser, task, between

class FastAPIUser(HttpUser):
    wait_time = between(1, 3)
    
    @task(3)
    def get_items(self):
        self.client.get("/items")
    
    @task(1)
    def create_item(self):
        self.client.post("/items", json={"name": "test", "price": 100})

테스트 실행:

# 1000명의 동시 사용자, 초당 100명씩 증가
locust -f locustfile.py --host=https://example.com --users 1000 --spawn-rate 100

워커 수에 따른 성능 비교:

# 워커 2개
WORKERS=2 gunicorn -c gunicorn.conf.py main:app
# 결과: 2000 req/s

# 워커 4개
WORKERS=4 gunicorn -c gunicorn.conf.py main:app
# 결과: 3500 req/s

# 워커 8개
WORKERS=8 gunicorn -c gunicorn.conf.py main:app
# 결과: 4200 req/s

# 워커 16개
WORKERS=16 gunicorn -c gunicorn.conf.py main:app
# 결과: 4300 req/s (메모리 부족, 성능 저하)

최적의 워커 수는 서버 사양과 애플리케이션 특성에 따라 다르다.

7. 모니터링 및 로깅

7.1 Gunicorn 통계 확인

# 마스터 프로세스 PID 확인
cat /run/gunicorn/gunicorn.pid

# 워커 프로세스 목록
ps aux | grep gunicorn

# 워커별 메모리 사용량
ps -o pid,ppid,%mem,rss,cmd -p $(pgrep -P $(cat /run/gunicorn/gunicorn.pid))

7.2 로그 분석

접근 로그 분석:

# 가장 많이 호출된 엔드포인트
awk '{print $7}' /var/log/myapp/access.log | sort | uniq -c | sort -rn | head -10

# 평균 응답 시간
awk '{sum+=$NF; count++} END {print sum/count " ms"}' /var/log/myapp/access.log

# 상태 코드별 통계
awk '{print $9}' /var/log/myapp/access.log | sort | uniq -c | sort -rn

에러 로그 모니터링:

# 실시간 에러 확인
tail -f /var/log/myapp/error.log

# 에러 패턴 분석
grep "ERROR" /var/log/myapp/error.log | awk '{print $5}' | sort | uniq -c | sort -rn

7.3 프로메테우스 연동

Gunicorn 메트릭을 프로메테우스로 수집할 수 있다.

pip install prometheus-client

메트릭 수집 (main.py):

from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi import FastAPI, Response

app = FastAPI()

# 메트릭 정의
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
REQUEST_DURATION = Histogram('http_request_duration_seconds', 'HTTP request duration', ['method', 'endpoint'])

@app.middleware("http")
async def metrics_middleware(request, call_next):
    with REQUEST_DURATION.labels(method=request.method, endpoint=request.url.path).time():
        response = await call_next(request)
        REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path, status=response.status_code).inc()
        return response

@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

8. 트러블슈팅

8.1 워커가 자주 재시작되는 경우

원인 분석:

# 워커 재시작 로그 확인
grep "Worker" /var/log/myapp/error.log

# 메모리 사용량 확인
watch -n 1 'ps aux | grep gunicorn'

해결 방법:

# gunicorn.conf.py
# 타임아웃 증가
timeout = 60
graceful_timeout = 60

# 메모리 누수가 의심되는 경우
max_requests = 500  # 더 자주 재시작
max_requests_jitter = 50

8.2 느린 응답 시간

원인 분석:

# 느린 요청 로그 확인
awk '$NF > 1000' /var/log/myapp/access.log

해결 방법:

# gunicorn.conf.py
# 워커 연결 수 증가
worker_connections = 2000

# 워커 수 증가
workers = 8

8.3 502 Bad Gateway 에러

원인:

• Gunicorn이 응답하지 않음
• 워커가 모두 busy 상태

해결 방법:

# Gunicorn 상태 확인
sudo systemctl status myapp

# 워커 수 증가
sudo systemctl stop myapp
# gunicorn.conf.py에서 workers 증가
sudo systemctl start myapp

# Nginx 타임아웃 증가
# /etc/nginx/sites-available/myapp
proxy_read_timeout 120s;

9. 결론

9.1 선택 가이드

Uvicorn 단일 프로세스:

• 개발 환경
• 소규모 API (일일 방문자 < 1,000)
• 리소스가 제한된 환경
• 빠른 프로토타이핑

Uvicorn 멀티 프로세스:

• 중규모 API (일일 방문자 1,000 ~ 50,000)
• 무중단 배포가 필요 없는 경우
• 간단한 설정을 원하는 경우
• Gunicorn 학습 곡선을 피하고 싶은 경우

Gunicorn + Uvicorn:

• 대규모 프로덕션 환경
• 무중단 배포가 필수적인 경우
• 세밀한 프로세스 관리가 필요한 경우
• 워커 수명 주기 관리가 필요한 경우
• 엔터프라이즈 수준의 안정성이 필요한 경우

9.2 핵심 차이점 요약

성능: 세 방식 중 멀티 프로세스를 사용하는 두 방식의 성능은 거의 동일하다. 차이는 관리 기능에 있다.

프로세스 관리:

• Uvicorn 단일: 없음 (외부 도구 필요)
• Uvicorn 멀티: 기본 기능만 제공
• Gunicorn+Uvicorn: 완전한 프로세스 관리 기능

무중단 배포:

• Uvicorn 단일: 불가능
• Uvicorn 멀티: 제한적
• Gunicorn+Uvicorn: 완벽하게 지원

설정 유연성:

• Uvicorn 단일: 커맨드라인 옵션만
• Uvicorn 멀티: 커맨드라인 옵션만
• Gunicorn+Uvicorn: 설정 파일 + Python 훅

9.3 최종 권장사항

프로덕션 환경에서는 Gunicorn + Uvicorn 조합을 강력히 권장한다. 초기 설정이 약간 복잡하지만, 안정성, 관리 용이성, 무중단 배포 등의 장점이 복잡성을 상쇄하고도 남는다.

Uvicorn 멀티 프로세스는 내부적으로 Gunicorn을 사용하므로, 직접 Gunicorn을 사용하여 더 많은 제어권을 갖는 것이 합리적이다.

개발 환경에서는 Uvicorn 단일 프로세스의 자동 리로드 기능이 생산성을 크게 향상시키므로, 개발과 프로덕션 환경을 명확히 구분하여 각각에 적합한 방식을 사용하는 것이 최선이다.

들어가며

Python에서 제네릭 프로그래밍을 할 때 TypeVar와 Generic이 자주 등장한다. 이 두 도구가 무엇이고 어떻게 함께 사용하는지 혼란스러울 수 있다. 이 글에서는 TypeVar와 Generic이 각각 무엇인지, 함수와 클래스에서 어떻게 사용하는지, 그리고 왜 구분해서 사용해야 하는지까지 전부 다루어볼 것이다.

목차

1. TypeVar란 무엇인가?
2. Generic이란 무엇인가?
3. 함수에서의 TypeVar 사용
4. 클래스에서의 Generic과 TypeVar
5. 함수에서는 Generic을 쓸 필요가 없는 이유
6. 클래스에서 Generic이 필요한 이유
7. 여러 타입 변수 사용하기
8. 타입 체커와 Generic의 중요성
9. 정리

TypeVar란 무엇인가?

TypeVar는 타입 변수이다. 일반 변수가 값을 저장하듯이, TypeVar는 타입을 변수처럼 다루게 해준다.

from typing import TypeVar

T = TypeVar('T')  # T는 어떤 타입이라도 될 수 있는 타입 변수

변수와의 비유

일반 변수를 살펴보자:

x = 5       # x는 5라는 값을 가짐
x = "hello" # x는 "hello"라는 값으로 변경 가능

TypeVar도 비슷하다:

T = TypeVar('T')  # T는 어떤 타입이라도 될 수 있음

차이점은 T는 타입을 담는다는 것이다.

Generic이란 무엇인가?

Generic은 클래스나 함수가 여러 타입을 다룰 수 있도록 만드는 도구이다. TypeVar와 함께 사용하면 더 강력해진다.

Generic의 역할

Generic은 다음과 같은 역할을 한다:

• 클래스나 함수가 제네릭하다는 것을 명시적으로 선언함
• 타입 체커가 타입 관계를 정확히 파악할 수 있도록 함
• 타입 안전성을 강화함

함수에서의 TypeVar 사용

함수에서는 TypeVar만으로 제네릭 기능을 충분히 구현할 수 있다.

from typing import TypeVar

T = TypeVar('T')

def get_first_item(items: list[T]) -> T:
    """리스트의 첫 번째 항목을 반환"""
    return items[0]

함수 사용 예시

# 정수 리스트
result = get_first_item([1, 2, 3])
print(result + 10)  # 결과: 11

# 문자열 리스트
result = get_first_item(["a", "b", "c"])
print(result.upper())  # 결과: "A"

# 리스트의 리스트
result = get_first_item([[1, 2], [3, 4]])
print(result[0])  # 결과: 1

함수는 호출할 때마다 타입이 자동으로 추론되므로 Generic을 명시적으로 상속할 필요가 없다.

클래스에서의 Generic과 TypeVar

클래스를 제네릭하게 만들려면 Generic을 상속받아야 한다.

from typing import TypeVar, Generic

T = TypeVar('T')

class Box(Generic[T]):
    """어떤 타입이든 담을 수 있는 박스"""

    def __init__(self, item: T):
        self.item = item

    def get(self) -> T:
        return self.item

클래스 사용 예시

# 정수를 담는 박스
int_box = Box[int](42)
result = int_box.get()  # result는 int

# 문자열을 담는 박스
str_box = Box[str]("hello")
result = str_box.get()  # result는 str

# 리스트를 담는 박스
list_box = Box[list]([1, 2, 3])
result = list_box.get()  # result는 list

클래스는 인스턴스를 생성할 때 타입을 명시해야 하므로 Generic을 상속받는 것이 필수적이다.

Generic 없이 쓴다면?

from typing import TypeVar

T = TypeVar("T")

class Box:
    def __init__(self, value: T):
        self.value = value

    def get(self) -> T:
        return self.value

box = Box(42)  # 이렇게만 쓸 수 있음

이 코드는 실행은 되지만 Box[int] 같은 문법을 사용할 수 없다. 타입 체커도 정확히 검사하지 못한다.

함수에서는 Generic을 쓸 필요가 없는 이유

함수는 이미 TypeVar만으로 제네릭 기능이 충분하다. Generic을 상속할 필요가 없다.

함수의 특성

함수는 호출할 때마다 타입이 자동으로 추론된다:

from typing import TypeVar

T = TypeVar('T')

def process(item: T) -> T:
    return item

# 호출할 때마다 타입이 결정됨
result1 = process(42)          # T = int
result2 = process("hello")     # T = str
result3 = process([1, 2, 3])   # T = list

따라서 Generic을 명시적으로 상속할 필요가 없다. TypeVar만으로 타입 체커가 충분히 추론할 수 있다.

클래스에서 Generic이 필요한 이유

클래스는 객체를 생성할 때 타입을 미리 정해야 한다. 그래서 Generic을 상속받는 것이 필수적이다.

클래스의 특성

클래스는 인스턴스 생성 시 타입을 명시해야 한다:

from typing import Generic, TypeVar

T = TypeVar('T')

class Box(Generic[T]):
    def __init__(self, item: T):
        self.item = item

# 인스턴스 생성 시 타입을 명시
box1 = Box[int](42)       # Box의 T는 int로 고정
box2 = Box[str]("hello")  # Box의 T는 str로 고정

만약 Generic을 상속하지 않으면:

T = TypeVar('T')

class Box:  # ❌ Generic 상속 없음
    def __init__(self, item: T):
        self.item = item

box = Box[int](42)  # ❌ 에러! Box는 제네릭이 아님
box = Box(42)       # ✅ 이렇게만 가능

Generic을 상속하지 않으면 Box[int] 같은 문법을 사용할 수 없다.

여러 타입 변수 사용하기

한 클래스나 함수에서 여러 개의 타입 변수를 사용할 수 있다.

클래스에서 여러 타입 변수 사용

from typing import TypeVar, Generic

T = TypeVar('T')
U = TypeVar('U')

class Pair(Generic[T, U]):
    """두 개의 다른 타입을 담을 수 있는 쌍"""

    def __init__(self, first: T, second: U):
        self.first = first
        self.second = second

    def get_first(self) -> T:
        return self.first

    def get_second(self) -> U:
        return self.second

사용 예시

# 정수와 문자열의 쌍
pair = Pair[int, str](10, "hello")
num = pair.get_first()    # int
text = pair.get_second()  # str

# 문자열과 리스트의 쌍
pair2 = Pair[str, list]("items", [1, 2, 3])
name = pair2.get_first()    # str
items = pair2.get_second()  # list

함수에서 여러 타입 변수 사용

from typing import TypeVar

T = TypeVar('T')
U = TypeVar('U')

def combine(a: T, b: U) -> tuple[T, U]:
    """두 개의 다른 타입 값을 조합"""
    return (a, b)

result = combine(42, "hello")  # tuple[int, str]

타입 체커와 Generic의 중요성

코드는 Generic 없이도 실행되지만, 타입 체커는 정확히 검사하지 못한다.

예시: Generic 없을 때

from typing import TypeVar

T = TypeVar("T")

class Box:
    def __init__(self, value: T):
        self.value = value

    def get(self) -> T:
        return self.value

box = Box(42)
result = box.get()
print(result + 10)  # ✅ 또는 ❌ ?

타입 체커는 혼란스러워한다:

• result가 int인지 확실하지 않음
• str일 수도 있고, dict일 수도 있음
• 따라서 result + 10이 안전한지 검증할 수 없음

예시: Generic 있을 때

from typing import Generic, TypeVar

T = TypeVar("T")

class Box(Generic[T]):
    def __init__(self, value: T):
        self.value = value

    def get(self) -> T:
        return self.value

box = Box[int](42)
result = box.get()
print(result + 10)  # ✅ 타입 체커가 안전함을 확인

이제 타입 체커는:

• box가 Box[int]임을 알 수 있음
• result가 int임을 알 수 있음
• result + 10이 안전함을 검증할 수 있음

mypy로 검사한 실제 차이

# Generic 없이
box = Box(42)
print(box.get().upper())  # ❌ mypy가 경고하지 않음 (T가 뭔지 모르니까)

# Generic 있이
box = Box[int](42)
print(box.get().upper())  # ❌ mypy가 경고함! (int에는 upper() 없다고)

Generic을 사용하면 타입 체커가 더 정확하게 오류를 발견할 수 있다.

정리

TypeVar와 Generic의 차이

• TypeVar: 타입을 변수처럼 다루는 도구. 함수와 클래스 모두에서 사용 가능.
• Generic: 클래스(또는 함수)가 제네릭하다는 것을 명시적으로 선언. 타입 체커가 정확히 검사할 수 있도록 함.

함수에서의 사용

함수는 TypeVar만으로 충분하다. 호출할 때마다 타입이 자동으로 추론되므로 Generic을 상속할 필요가 없다:

from typing import TypeVar

T = TypeVar('T')

def get_first(items: list[T]) -> T:
    return items[0]

클래스에서의 사용

클래스는 Generic을 반드시 상속해야 한다. 인스턴스를 생성할 때 타입을 명시적으로 지정할 수 있도록 하려면 Generic이 필수다:

from typing import Generic, TypeVar

T = TypeVar('T')

class Box(Generic[T]):
    def __init__(self, item: T):
        self.item = item

타입 안전성

• 코드 실행: Generic 없이도 완벽하게 작동
• 타입 검사: Generic을 상속해야 타입 체커가 제대로 검사 가능
• Best Practice: 클래스에서 제네릭을 사용할 땐 항상 Generic을 상속받자

런타임 동작과 타입 체킹은 별개의 문제다. 당신의 코드가 실행되더라도, 개발할 때 타입 체커의 도움을 받으려면 Generic을 올바르게 사용해야 한다.

TypeVar와 Generic을 제대로 이해하고 사용하면, 코드의 타입 안전성이 크게 향상되고 버그를 사전에 방지할 수 있다. 처음에는 복잡해 보일 수 있지만, 몇 번 사용하면 금방 자연스러워질 것이다.

들어가며

Python에서 TypeVar를 사용할 때마다 헷갈린다면 이 글이 도움이 될 것이다. 문법도 간단하고 원리도 어렵지 않은데, 처음에는 낯설 수 있다. 이 글에서는 TypeVar가 무엇인지, 어떻게 사용하는지, 타입 체커가 어떻게 검사하는지까지 전부 다루어볼 것이다.

목차

1. TypeVar란 무엇인가?
2. 실제로 어떻게 사용하는가?
3. 왜 필요한가?
4. 문법 정리
5. 타입 체커는 어떻게 검사하는가?
6. 인자와 리턴값 모두에서 꼭 사용해야 하는가?
7. TypeVar 사용의 핵심
8. 정리

TypeVar란 무엇인가?

TypeVar는 타입 변수이다. 일반 변수가 값을 저장하듯이, TypeVar는 타입을 변수처럼 다루게 해준다.

from typing import TypeVar

T = TypeVar('T')  # T는 어떤 타입이라도 될 수 있는 타입 변수

변수와의 비유

일반 변수를 살펴보자:

x = 5       # x는 5라는 값을 가짐
x = "hello" # x는 "hello"라는 값으로 변경 가능

TypeVar도 비슷하다:

T = TypeVar('T')  # T는 어떤 타입이라도 될 수 있음

차이점은 T는 타입을 담는다는 것이다.

실제로 어떻게 사용하는가?

제네릭(generic) 함수를 만들 때 사용한다. 제네릭 함수는 여러 타입을 유연하게 받을 수 있으면서도 타입 안전성을 유지하는 함수이다.

가장 간단한 예시

from typing import TypeVar

T = TypeVar('T')

def get_first_item(items: list[T]) -> T:
    """리스트의 첫 번째 항목을 반환"""
    return items[0]

이 함수의 의미는 다음과 같다:

• 함수에 들어오는 리스트가 int이면, 반환도 int
• 함수에 들어오는 리스트가 str이면, 반환도 str
• 함수에 들어오는 리스트가 dict이면, 반환도 dict

입력 타입과 출력 타입이 같다는 것을 명확히 표현하는 것이다.

왜 필요한가?

TypeVar 없이 작성한다면:

def get_first_item(items: list) -> any:
    return items[0]

이렇게 하면 타입 힌트가 거의 무의미하다. IDE의 자동완성이나 타입 체커가 제대로 도움을 주지 못한다.

TypeVar를 사용하면:

def get_first_item(items: list[T]) -> T:
    return items[0]

타입이 명확해지므로 IDE가 반환 타입을 정확히 알 수 있고, 코드를 작성할 때 훨씬 도움이 된다.

문법 정리

T = TypeVar('T')

• T = 변수 이름 (관례상 대문자 영어 하나 사용)
• 'T' = 문자열로 된 이름 (내부적으로 사용됨)

둘이 같은 이름이어야 한다. Python의 관례일 뿐이지만 반드시 지켜야 한다.

타입 체커는 어떻게 검사하는가?

타입 체커(예: mypy)는 코드를 실행하지 않고 타입이 맞는지 검사하는 도구이다.

pip install mypy
mypy your_file.py

검사 방식

from typing import TypeVar

T = TypeVar('T')

def get_first_item(items: list[T]) -> T:
    return items[0]

# ✅ OK - 정수 리스트 입력, 정수 반환
result_int = get_first_item([1, 2, 3])
print(result_int + 10)  # 괜찮음

# ✅ OK - 문자열 리스트 입력, 문자열 반환
result_str = get_first_item(["a", "b", "c"])
print(result_str.upper())  # 괜찮음

# ❌ ERROR - 타입 불일치 감지!
result = get_first_item([1, 2, 3])
print(result.upper())  # mypy가 에러 보고함
# "int"에는 upper() 메서드가 없다고 경고

타입 체커의 논리

1. 함수 호출 분석: get_first_item([1, 2, 3])를 분석함
2. T 추론: 리스트가 list[int]이므로, T는 int라고 결정
3. 반환 타입 확인: T가 int이므로 반환값도 int
4. 사용처 검증: result.upper()를 확인했는데, int에 upper() 없음 → ❌ 에러 보고

조금 더 복잡한 예시

from typing import TypeVar

T = TypeVar('T')

def swap(a: T, b: T) -> tuple[T, T]:
    """두 값을 바꾸기"""
    return (b, a)

# ✅ OK - 두 매개변수 타입이 같음
result = swap(1, 2)  # T = int

# ❌ ERROR - 두 매개변수 타입이 다름!
result = swap(1, "hello")  # T가 뭐지? int? str? → 에러!

타입 체커는 swap(1, "hello")를 보고 혼란스러워한다:

• 첫 번째 인자는 int → T는 int여야 함
• 두 번째 인자는 str → T는 str여야 함
• 모순! → ❌ 에러 보고

인자와 리턴값 모두에서 꼭 사용해야 하는가?

아니다. 상황에 따라 다르다.

인자만 사용

T = TypeVar('T')

def process(item: T) -> None:
    """입력받은 값을 처리하기만 하고 반환 없음"""
    print(item)

이는 완전히 괜찮다. T는 어떤 타입이든 받을 수 있지만 반환값이 없으므로 T를 리턴에 사용하지 않았다.

리턴값만 사용

T = TypeVar('T')

def create_default(default_value: T) -> T:
    """기본값을 받아서 그대로 반환"""
    return default_value

이것도 문제없다.

같은 T를 여러 곳에 사용

T = TypeVar('T')

def first_and_last(items: list[T]) -> T:
    """리스트의 마지막 항목을 반환"""
    return items[-1]

T는 리스트의 원소 타입과 반환값의 타입이 같다는 것을 표현한다.

TypeVar 사용의 핵심

TypeVar를 사용하는 목적은:

"이 여러 곳의 타입들이 서로 연관되어 있다"는 것을 표현하는 것

따라서:

• 같은 타입 관계가 없으면 사용하지 않는 것이 낫고
• 다른 타입 관계가 있으면 다른 타입 변수를 만들어서 구분한다

나쁜 예시

T = TypeVar('T')

def example(a: T) -> int:  # ❌ 굳이 T를 사용할 이유가 없음
    return len(str(a))

a가 어떤 타입이든 상관없고 무조건 int를 반환하므로 T를 사용할 이유가 없다.

def example(a: object) -> int:  # ✅ 이렇게 명확하게
    return len(str(a))

이것이 훨씬 낫고 명확하다.

정리

• TypeVar는 타입 변수이다. 일반 변수가 값을 담듯이 타입을 담는다.
• 제네릭 함수를 만들 때 사용하여 여러 타입을 유연하게 받으면서도 타입 안전성을 유지한다.
• 타입 간의 관계를 표현하는 것이 핵심이다. 무조건 둘 다 사용해야 하는 것이 아니라, 의미 있는 관계가 있을 때만 사용한다.
• 타입 체커는 TypeVar로 표현된 타입의 일관성을 검증하여 런타임 에러를 미리 방지해준다.

TypeVar를 제대로 이해하고 사용하면, 코드의 타입 안전성이 크게 향상되고 유지보수도 훨씬 쉬워진다. 처음에는 낯설 수 있지만 몇 번 사용하면 금방 익숙해질 것이다.

네트워크 서버 애플리케이션을 개발하다 보면 반드시 마주치게 되는 개념이 IP binding이다. 서버를 실행할 때 단순히 포트 번호만 지정하는 것이 아니라, 어떤 IP 주소로 바인딩할지도 결정해야 한다. 이 글에서는 IP binding의 개념과 실무에서의 활용 방법을 정리한다.

IP Binding의 정의

IP binding은 서버 프로세스가 특정 IP 주소와 포트 조합에 소켓을 연결(bind)하여, 해당 주소로 들어오는 네트워크 연결 요청만 수신하도록 설정하는 것을 의미한다.

서버가 네트워크 요청을 수신하기 위해서는 다음 두 가지 요소를 지정해야 한다.

• IP 주소: 어떤 네트워크 인터페이스를 통해 들어오는 요청을 받을 것인가
• 포트 번호: 해당 인터페이스의 어떤 포트로 들어오는 요청을 받을 것인가

주요 바인딩 주소

1. 0.0.0.0 (INADDR_ANY)

0.0.0.0은 시스템의 모든 네트워크 인터페이스에 바인딩하는 특수한 주소이다.

특징:

• 서버가 가진 모든 IP 주소로 들어오는 요청을 수신한다
• 외부 네트워크, 내부 네트워크, localhost 모두 접근 가능하다
• 운영 환경에서 가장 일반적으로 사용되는 설정이다

예시:

서버의 네트워크 인터페이스:
- eth0: 203.0.113.10 (공인 IP)
- eth1: 192.168.1.10 (사설 IP)
- lo: 127.0.0.1 (loopback)

0.0.0.0:8000 으로 바인딩 시:
→ 203.0.113.10:8000 접근 가능
→ 192.168.1.10:8000 접근 가능
→ 127.0.0.1:8000 접근 가능

2. 127.0.0.1 (localhost)

127.0.0.1은 loopback 인터페이스로, 동일한 시스템 내부에서만 통신 가능한 주소이다.

특징:

• 외부 네트워크에서 절대 접근할 수 없다
• 개발 환경이나 내부 프로세스 간 통신에 사용한다
• 보안상 외부 노출을 원하지 않는 서비스에 적합하다

예시:

127.0.0.1:5000 으로 바인딩 시:
→ 동일 서버에서 curl localhost:5000 → 성공
→ 외부에서 curl 203.0.113.10:5000 → 실패 (연결 거부)

3. 특정 IP 주소

서버의 특정 네트워크 인터페이스 IP 주소를 직접 지정하여 바인딩할 수 있다.

특징:

• 서버에 여러 네트워크 인터페이스가 있을 때 선택적으로 사용한다
• 특정 네트워크에서만 서비스를 제공하고자 할 때 유용하다
• 네트워크 트래픽을 분리하여 관리할 수 있다

예시:

서버의 네트워크 인터페이스:
- eth0: 203.0.113.10 (공인 IP - 인터넷 연결)
- eth1: 192.168.1.10 (사설 IP - 내부망 연결)

192.168.1.10:8000 으로 바인딩 시:
→ 내부망에서만 접근 가능
→ 인터넷에서 203.0.113.10:8000 접근 시도 → 실패

실무 활용 시나리오

시나리오 1: 로컬 개발 환경

개발자가 로컬 머신에서 API 서버를 개발 중인 상황이다.

바인딩: 127.0.0.1:8000
결과: 외부 접근 차단, 로컬에서만 테스트 가능
장점: 보안상 안전, 의도하지 않은 외부 접근 방지

시나리오 2: 운영 웹 서버

실제 서비스를 제공하는 웹 서버를 운영하는 상황이다.

바인딩: 0.0.0.0:80
결과: 인터넷의 모든 사용자가 접근 가능
장점: 최대 접근성 확보
주의: 방화벽, 리버스 프록시 등 추가 보안 조치 필요

시나리오 3: 내부망 전용 모니터링 서버

회사 내부망에서만 사용하는 모니터링 대시보드를 운영하는 상황이다.

바인딩: 192.168.10.50:3000
결과: 내부망에서만 접근 가능
장점: 외부 인터넷에서 접근 불가, 내부 정보 보호

시나리오 4: 멀티 인터페이스 서버

공인 IP와 사설 IP를 모두 가진 서버에서 서로 다른 서비스를 제공하는 상황이다.

서비스 A 바인딩: 203.0.113.10:443 (HTTPS API - 외부 공개)
서비스 B 바인딩: 10.0.1.10:9090 (관리자 패널 - 내부 전용)

결과: 
- 외부 사용자는 API만 접근 가능
- 관리자 패널은 내부망에서만 접근 가능
- 명확한 서비스 분리 및 보안 강화

흔한 실수와 해결 방법

실수 1: 개발 설정을 운영에 그대로 배포

문제: 로컬 개발 시 127.0.0.1로 바인딩한 설정을 그대로 운영 서버에 배포
증상: 외부에서 서버에 접근할 수 없음
해결: 환경별 설정 파일을 분리하고, 운영 환경에서는 0.0.0.0 사용

실수 2: 테스트 서버의 무분별한 외부 노출

문제: 개발 중인 기능을 테스트하는 서버를 0.0.0.0으로 바인딩
증상: 인증/인가가 구현되지 않은 상태로 외부에 노출
해결: 테스트 서버는 127.0.0.1 또는 내부 IP로 바인딩하고, VPN을 통해 접근

실수 3: 잘못된 IP 주소 지정

문제: 서버에 존재하지 않는 IP 주소로 바인딩 시도
증상: 서버 시작 실패 (Cannot assign requested address 오류)
해결: ifconfig 또는 ip addr 명령어로 사용 가능한 IP 확인 후 바인딩

보안 고려사항

IP binding은 네트워크 보안의 첫 번째 방어선이다. 다음 사항을 고려해야 한다.

1. 최소 권한 원칙: 필요한 인터페이스에만 바인딩한다. 내부 전용 서비스는 외부 인터페이스에 바인딩하지 않는다.
2. 방화벽 연동: IP binding만으로는 불충분하며, iptables, ufw 등의 방화벽 규칙과 함께 사용해야 한다.
3. 프록시 활용: 직접 0.0.0.0으로 바인딩하기보다는, nginx 등의 리버스 프록시를 앞단에 두고 애플리케이션은 127.0.0.1에 바인딩하는 것이 더 안전하다.
4. 환경 분리: 개발, 스테이징, 운영 환경마다 적절한 바인딩 설정을 적용한다.

결론

IP binding은 서버 애플리케이션이 네트워크 요청을 수신하는 첫 단계에서 결정되는 중요한 설정이다. 단순해 보이지만, 올바른 IP 주소 선택은 서비스의 접근성과 보안에 직접적인 영향을 미친다.

개발자는 각 환경의 요구사항을 정확히 파악하고, 적절한 IP 주소로 바인딩하여 불필요한 노출을 방지하고 안전한 서비스를 제공해야 한다. IP binding은 방화벽, 인증/인가 등 다른 보안 계층과 함께 사용될 때 그 효과가 극대화된다는 점을 기억해야 한다.

valid_referers란 무엇인가

valid_referers는 Nginx에서 HTTP Referer 헤더를 검증하여 특정 출처로부터의 요청만 허용하는 지시어이다. 이는 주로 핫링킹(hotlinking) 방지, 즉 외부 사이트에서 자신의 서버 리소스를 무단으로 링크하여 사용하는 것을 막기 위해 사용된다.

Referer 헤더의 이해

HTTP Referer 헤더는 클라이언트가 현재 요청을 보내기 전에 어느 페이지에 있었는지를 나타낸다. 예를 들어, 사용자가 https://example.com/page.html에서 https://mysite.com/image.jpg를 클릭했다면, 이미지 요청의 Referer 헤더는 https://example.com/page.html이 된다.

valid_referers의 동작 원리

valid_referers 지시어는 ngx_http_referer_module 모듈에서 제공한다. 이 지시어는 Referer 헤더를 검사하여 지정된 값과 일치하는지 확인하고, 그 결과를 $invalid_referer 변수에 저장한다.

• Referer가 유효한 경우: $invalid_referer 변수는 빈 문자열이 된다
• Referer가 유효하지 않은 경우: $invalid_referer 변수는 "1"이 된다

기본 문법

valid_referers none | blocked | server_names | string ...;

이 지시어는 server, location 컨텍스트에서 사용 가능하다.

파라미터 설명

none

Referer 헤더가 아예 없는 경우를 허용한다. 브라우저 주소창에 직접 URL을 입력하거나, 북마크를 통해 접근하는 경우가 여기에 해당한다.

blocked

Referer 헤더가 존재하지만 방화벽이나 프록시에 의해 "http://" 또는 "https://"가 제거된 경우를 허용한다.

• 정상 사용자 보호: 기업 네트워크나 보안이 강화된 환경의 사용자들이 정상적으로 접근할 수 있다.
• 호환성: 다양한 네트워크 환경에서 서비스가 원활하게 작동하도록 한다.
• 사용자 경험: 불필요한 접근 거부를 줄인다.

server_names

현재 서버의 server_name 지시어에 정의된 모든 도메인을 자동으로 허용한다.

문자열 패턴

특정 도메인이나 패턴을 직접 지정한다. 와일드카드(*)를 사용할 수 있으며, 정규표현식도 지원한다.

설정 예시

기본적인 핫링킹 방지

location ~* \.(jpg|jpeg|png|gif|ico|css|js)$ {
    valid_referers none blocked server_names;
    
    if ($invalid_referer) {
        return 403;
    }
}

위 설정은 이미지, CSS, JavaScript 파일에 대해 다음을 허용한다:

• Referer 헤더가 없는 직접 접근
• 방화벽에 의해 차단된 Referer
• 자신의 서버 도메인

그 외의 경우는 403 Forbidden을 반환한다.

특정 도메인 허용

location ~* \.(jpg|jpeg|png|gif)$ {
    valid_referers none blocked server_names
                   example.com
                   *.example.com
                   google.com
                   *.google.com;
    
    if ($invalid_referer) {
        return 403;
    }
}

위 설정은 example.com, google.com 및 그들의 서브도메인에서의 접근을 허용한다.

정규표현식 사용

location ~* \.(jpg|jpeg|png|gif)$ {
    valid_referers none blocked server_names
                   ~\.google\.
                   ~\.facebook\.;
    
    if ($invalid_referer) {
        return 403;
    }
}

정규표현식을 사용할 때는 ~로 시작한다. 위 예시는 .google.과 .facebook.을 포함하는 모든 도메인을 허용한다.

대체 이미지 제공

단순히 403을 반환하는 대신, 무단 링크 시 대체 이미지를 제공할 수 있다.

location ~* \.(jpg|jpeg|png|gif)$ {
    valid_referers none blocked server_names;
    
    if ($invalid_referer) {
        rewrite ^/(.*)$ /images/hotlink-forbidden.png break;
    }
}

주의사항

if 지시어의 위험성

Nginx에서 if 지시어는 예상치 못한 동작을 일으킬 수 있다. if 블록 내부에서는 제한적인 지시어만 사용해야 하며, 복잡한 로직은 map 지시어나 다른 방법을 사용하는 것이 권장된다.

Referer 헤더의 신뢰성

Referer 헤더는 클라이언트가 제공하는 정보이므로 쉽게 위조할 수 있다. 따라서 valid_referers는 보안 메커니즘이라기보다는 일반적인 핫링킹 방지 수단으로 이해해야 한다.

브라우저 정책

일부 브라우저나 확장 프로그램은 개인정보 보호를 위해 Referer 헤더를 전송하지 않거나 수정할 수 있다. 이는 정상적인 사용자의 접근도 차단될 수 있음을 의미한다.

map을 이용한 더 나은 접근

if 지시어의 한계를 극복하기 위해 map 지시어를 사용할 수 있다.

map $invalid_referer $block_hotlink {
    0  "";
    1  "blocked";
}

server {
    location ~* \.(jpg|jpeg|png|gif)$ {
        valid_referers none blocked server_names;
        
        error_page 403 /403.html;
        
        if ($block_hotlink) {
            return 403;
        }
    }
}

성능 고려사항

valid_referers 검사는 매 요청마다 수행되므로, 트래픽이 많은 서버에서는 성능에 영향을 줄 수 있다. 필요한 리소스에만 선택적으로 적용하는 것이 좋다.

# 대용량 파일에만 적용
location ~* \.(mp4|zip|pdf)$ {
    valid_referers none blocked server_names;
    
    if ($invalid_referer) {
        return 403;
    }
}

# 작은 리소스는 제한 없음
location ~* \.(css|js|ico)$ {
    # valid_referers 설정 없음
}

로깅 활용

핫링킹 시도를 모니터링하기 위해 로그를 활용할 수 있다.

location ~* \.(jpg|jpeg|png|gif)$ {
    valid_referers none blocked server_names;
    
    if ($invalid_referer) {
        access_log /var/log/nginx/hotlink.log;
        return 403;
    }
}

결론

valid_referers는 Nginx에서 Referer 헤더를 검증하여 리소스 무단 사용을 방지하는 유용한 도구이다. 완벽한 보안 솔루션은 아니지만, 대부분의 일반적인 핫링킹 시도를 효과적으로 차단할 수 있다. 서버의 특성과 요구사항에 맞게 적절히 설정하여 사용하면 된다.