스케줄러 없이 매시간 작업 돌리기: 웹 서버 안 데몬 스레드와 PostgreSQL advisory lock
이번 프로젝트는 고객사가 기존에 사용 중인 외부 주문 관리 서비스의 API를 저희 서비스와 연동하는 작업이었습니다. 외부 서비스에 쌓이는 주문을 저희 서비스에서도 볼 수 있게 하는 기능인데요, 기능 자체는 간단했지만 외부 API와 인프라의 제약 때문에 생각보다 고민할 게 많았습니다. 이 글에서는 그 고민을 어떻게 풀었는지 정리해 보려고 합니다.
무엇이 문제였나
처음 떠올린 건 조회 요청이 올 때마다 외부 API를 바로 호출하는 방식이었습니다. 제일 단순하니까요. 그런데 API 문서를 보니 응답이 느린 데다 호출 제한도 꽤 빡빡했습니다. 같은 API는 1초에 한 번만 부를 수 있고, 데이터를 많이 받은 뒤에는 수십 초를 쉬어야 했습니다. 사용자가 목록을 열 때마다 이걸 기다리게 할 수는 없었죠.
그래서 주문 데이터를 매시간 미리 받아 DB에 쌓아 두고, 조회 API는 DB만 읽게 하기로 했습니다. 여기까지는 금방 정했는데, 막상 구현하려니 걸리는 게 세 가지 있었습니다.
첫 번째는 매시간 작업을 돌릴 수단이 없다는 점이었습니다. 이 프로젝트가 사용할 수 있는 인프라에는 cron도, Celery도, Redis 같은 메시지 브로커도 없었습니다. 그렇다고 한 시간에 한 번 도는 작업 하나 때문에 추가 인프라 구축을 제안하자니 배보다 배꼽이 더 커 보였습니다.
두 번째는 실행을 놓치면 안 된다는 점이었습니다. 배포나 재시작 때문에 정각을 놓치면 서버가 뜨자마자 다시 돌아야 했고(catch-up), 사용자가 바로 동기화를 요청할 수 있는 수동 동기화 API도 필요했습니다.
세 번째는 동기화가 한 번에 하나만 돌아야 한다는 점이었습니다. 이 프로젝트의 서버는 평소 컨테이너 1개에 워커 2개로 뜨고, 롤링 배포 중에는 잠깐 컨테이너 2개(워커 4개)가 함께 떠 있습니다. 이 프로세스들이 정각에 동시에 동기화를 시작하면 외부 API 호출 제한을 넘기고 데이터가 꼬일 수 있습니다.
다행히 데이터 파싱 작업 자체는 가벼웠습니다. 한 번 도는 데 보통 몇 초, 길어도 몇 분이고, 그 시간 대부분은 외부 API 응답을 기다리는 시간입니다(I/O 바운드). 참고로 서버는 Django + gunicorn(uvicorn 워커) + PostgreSQL 구성이고, AWS ECS에 컨테이너로 배포합니다.
이렇게 풀었습니다
후보로 검토한 방법은 이렇습니다.
| 방법 | 추가로 필요한 것 | 걸린 점 |
|---|---|---|
| cron 전용 컨테이너 | 컨테이너, CI 배포 잡 | 작업 하나 때문에 배포 단위가 늘어난다 |
| Celery beat + worker | 브로커, beat·worker 프로세스 | 매시간 1회 작업에 프로세스 2개와 브로커는 과하다 |
| EventBridge 등으로 매시간 컨테이너 실행 | 스케줄 규칙, IAM 권한, 태스크 정의 | 매번 컨테이너를 새로 띄우고, 설정이 코드 밖 여러 곳에 흩어진다 |
| APScheduler | 라이브러리 의존성 | 결국 프로세스 안 스레드라 워커 수만큼 중복 실행되는 문제는 그대로다 |
| 별도 프로세스(multiprocessing) | 워커마다 프로세스 1개 | I/O 대기 위주 작업이라 프로세스를 나눠도 이점이 없다 |
| asyncio 태스크 | 없음 | Django ASGI는 lifespan 이벤트를 지원하지 않아 시작 시점에 태스크를 등록할 곳이 마땅치 않고, ORM·잠금 코드가 전부 동기다 |
| 조회 API 호출 시 마지막 동기화 시각 확인 후 실행 | 없음 | 요청이 없으면 실행되지 않는다 |
고민 끝에 고른 방법은 의외로 단순합니다. 별도 스케줄러를 두지 않고, 이미 떠 있는 웹 서버 프로세스 안에서 스레드를 하나 띄우는 겁니다.
threading.Thread(target=run_forever, name="hourly-sync", daemon=True).start()
이 방법을 고른 이유는 세 가지입니다.
첫째, 새로 들일 인프라가 없습니다. 같은 이미지, 같은 환경변수, 같은 배포 파이프라인을 그대로 씁니다.
둘째, 스레드로 충분한 작업입니다. 파이썬은 GIL(한 번에 한 스레드만 파이썬 코드를 실행하게 하는 잠금) 때문에 CPU 작업은 스레드로 나눠도 빨라지지 않지만, 네트워크 대기나 sleep 중에는 GIL이 풀립니다. 앞에서 말했듯 이 작업은 대부분 외부 API 응답을 기다리는 시간이라 웹 요청 처리에 주는 부담이 적습니다. 측정해 보니 대기 중인 스레드의 메모리도 16KB 정도였습니다.
셋째, 수동 동기화 API가 같은 코드를 재사용합니다. 정각 실행과 같은 함수를 별도 스레드에서 한 번 더 실행하면 됩니다.
APScheduler를 써도 중복 실행은 어차피 직접 막아야 했기 때문에, 라이브러리 없이 수십 줄로 직접 작성했습니다.
코드는 한 줄이지만, 웹 서버와 함께 도는 만큼 챙겨야 할 부분이 있었습니다. 실행 누락은 DB에 남기는 실행 이력으로, 중복 실행은 이미 쓰고 있던 PostgreSQL의 advisory lock으로 풀었는데요, 스케줄러 코드부터 차례로 이야기해 보겠습니다.
스케줄러 코드
설명을 위해 단순하게 다시 쓴 예시입니다.
import atexit
import threading
from datetime import datetime, timedelta
from django.utils import timezone
STARTUP_DELAY = 90 # 초. 배포 직후 헬스 체크와 겹치지 않게 잠시 기다린다
_stop = threading.Event()
def seconds_until_next_hour(now: datetime) -> float:
next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
return (next_hour - now).total_seconds()
def run_forever():
_stop.wait(STARTUP_DELAY)
while not _stop.is_set():
run_sync_safely(only_if_due=True) # 이번 시간에 아직 안 했으면 실행
_stop.wait(seconds_until_next_hour(timezone.now()))
def start_scheduler():
threading.Thread(target=run_forever, name="hourly-sync", daemon=True).start()
atexit.register(_stop.set)
Thread에 넘긴 인자에는 각각 이유가 있습니다.
target=run_forever는 다음 정각까지 대기하고, 깨어나서 실행 여부를 확인한 뒤, 다시 대기하는 루프입니다. 대기에는 time.sleep() 대신 _stop.wait(timeout)을 씁니다. Event.wait()는 timeout이 지나거나 _stop.set()이 호출되면 바로 반환하므로, 종료 신호를 받으면 최대 1시간을 기다리지 않고 즉시 루프를 빠져나옵니다.
name="hourly-sync"는 로그나 스레드 덤프(py-spy 등)에서 이 스레드를 바로 찾기 위한 이름입니다. 지정하지 않으면 Thread-3 같은 이름이 붙어, 문제가 생겼을 때 어떤 스레드인지 구분하기 어렵습니다.
daemon=True면 프로세스가 종료될 때 이 스레드를 기다리지 않습니다. 데몬이 아닌 스레드가 살아 있으면 파이썬은 그 스레드가 끝날 때까지 프로세스를 종료하지 않습니다. 그러면 배포나 재시작 때 워커가 제때 내려가지 못하고, gunicorn이 graceful timeout(기본 30초) 뒤 강제 종료할 때까지 기다리게 됩니다.
예시에서는 뺐지만, start_scheduler()가 여러 번 불려도 스레드가 하나만 뜨도록 막는 플래그도 넣어 두었습니다. 다음 정각까지 남은 시간처럼 시각을 계산하는 부분은 순수 함수로 빼 두면 스레드 없이 테스트할 수 있습니다.
def test_seconds_until_next_hour():
assert seconds_until_next_hour(datetime(2026, 1, 1, 10, 59, 30)) == 30
assert seconds_until_next_hour(datetime(2026, 1, 1, 10, 0, 0)) == 3600
웹 서버 안 스레드의 약점과 대응
장점만 있는 건 아닙니다. 웹 서버 프로세스의 생명주기를 그대로 따라가기 때문에 다음과 같은 약점이 있습니다.
- 스레드가 예외로 죽어도 알기 어렵다
- 재시작하면 메모리 상태가 사라진다
- 실행 도중 강제 종료될 수 있다
- 웹 서버가 아닌 곳(마이그레이션, 테스트)에서도 시작될 수 있다
- gunicorn
--preload와 함께 쓸 수 없다 - 요청 밖에서 쓰는 DB 연결을 직접 관리해야 한다
- 프로세스 수만큼 스레드가 떠서 동시에 실행된다
1~6은 이 장에서, 7은 다음 장에서 다룹니다.
1. 스레드가 예외로 죽어도 알기 어렵다
스레드에서 처리되지 않은 예외가 발생하면 그 스레드만 종료됩니다. 프로세스와 웹 요청 처리는 정상이라 겉으로는 문제가 보이지 않고, 동기화만 다음 배포 때까지 멈춥니다.
그래서 동기화 1회 실행을 감싸는 함수에서 모든 예외를 잡습니다. 보통은 except Exception:처럼 넓게 잡는 걸 피하지만, 이곳은 어떤 오류가 나도 루프가 계속 돌아야 하므로 예외로 두고 이유를 주석으로 남겼습니다.
def run_sync_safely(only_if_due=False):
try:
if only_if_due and synced_this_hour(timezone.now()):
return
sync_once() # 잠금을 잡고 실제 동기화를 한다 (뒤에서 설명)
except ExternalApiError:
logger.warning("외부 API 일시 장애로 동기화 실패", exc_info=True)
except Exception as exc: # 스케줄러 스레드는 어떤 오류에도 멈추면 안 된다
logger.exception("동기화 실패")
sentry_sdk.capture_exception(exc)
finally:
connection.close() # 요청 밖에서 도는 스레드라 연결을 직접 닫는다
외부 API 일시 장애는 예상 가능한 실패라 warning 로그만 남기고, 예상하지 못한 오류만 Sentry로 보냅니다. 그렇지 않으면 외부 서비스 점검 때마다 알림이 쏟아집니다.
sentry_sdk.capture_exception()을 직접 호출하는 이유도 있습니다. 저희 프로젝트는 Sentry 로깅 연동이 CRITICAL 레벨만 이벤트로 보내도록 설정돼 있어서, logger.exception()(ERROR 레벨)만으로는 Sentry에 기록되지 않습니다. 백그라운드 스레드의 실패는 HTTP 500처럼 드러나지도 않으니 직접 보내야 했습니다.
2. 재시작하면 메모리 상태가 사라진다
마지막 실행 시각을 메모리에만 두면 재시작할 때 사라집니다. 배포할 때마다 프로세스가 새로 뜨기 때문에, 정각 무렵에 배포하면 그 시간의 동기화를 건너뛰게 됩니다.
그래서 실행 이력을 DB에 남깁니다. 동기화가 끝날 때마다 이력 테이블에 한 줄을 쓰고, 스레드가 깨어나면 이번 정각 이후에 성공한 이력이 있는지 조회합니다. 있으면 건너뛰고, 없으면 실행합니다.
def synced_this_hour(now: datetime) -> bool:
hour_start = now.replace(minute=0, second=0, microsecond=0)
return SyncHistory.objects.filter(status="SUCCESS", started_at__gte=hour_start).exists()
예를 들어 12시 정각에 떠 있는 서버가 없었다면, 새로 뜬 서버가 90초 뒤 12시 이후 성공 이력이 없는 것을 확인하고 바로 실행합니다. 시작 직후가 아니라 90초 뒤에 확인하는 건 배포 직후의 헬스 체크와 겹치지 않게 하기 위해서입니다.
서버가 내려가 있는 동안에는 동기화도 멈추는데, 이 부분은 감수했습니다. 서버가 죽어 있으면 어차피 조회 API도 응답할 수 없고, 복구되면 90초 안에 놓친 동기화가 실행되기 때문입니다.
3. 실행 도중 강제 종료될 수 있다
daemon=True의 대가입니다. 파이썬 공식 문서에도 데몬 스레드는 종료 시 갑자기 중단되며, 열린 파일이나 DB 트랜잭션 같은 자원이 제대로 정리되지 않을 수 있다고 나와 있습니다. 동기화 도중에 배포가 시작되면 작업이 중간에 끊깁니다.
그래서 중간에 끊겨도 다시 실행하면 되는 구조로 만들었습니다.
- 조회 시작 시점은 마지막으로 성공한 동기화의 종료 시각입니다. 실패하거나 끊긴 실행은 성공 이력을 남기지 않으므로, 다음 실행이 같은 구간을 다시 가져옵니다.
- 가져온 데이터는 외부 서비스의 고유 ID 기준으로 upsert합니다. 이미 있으면 바뀐 필드만 갱신하고, 없으면 추가합니다. 같은 데이터를 여러 번 받아도 결과가 같은, 멱등(idempotent)한 저장입니다.
- 구간 경계에서 데이터가 누락되지 않도록 조회 시작 시점을 10분 앞당깁니다. 겹치는 구간은 upsert라 중복되지 않습니다.
- 이력은 실행이 끝날 때만 씁니다. 시작할 때 '진행 중' 이력을 먼저 남기면, 도중에 프로세스가 죽었을 때 그 이력이 계속 '진행 중'으로 남기 때문입니다.
마지막 항목은 뒤의 advisory lock과도 연결됩니다.
atexit.register(_stop.set)은 프로세스 종료 시 대기 중인 루프를 깨워 정상적으로 빠져나오게 합니다. 동기화 중이었다면 끊기지만, 위 구조 덕분에 다음 실행이 조금 더 넓은 구간을 가져오는 정도로 끝납니다.
4. 웹 서버가 아닌 곳에서 시작되면 안 된다
Django에서 앱 시작 시 코드를 실행할 때는 보통 AppConfig.ready()를 씁니다. 하지만 ready()는 웹 서버뿐 아니라 manage.py migrate, manage.py shell, pytest에서도 호출됩니다. 여기서 스케줄러를 시작하면 마이그레이션 중에 외부 API를 호출하고, 테스트 중에 실제 데이터를 가져오게 됩니다.
그래서 스케줄러는 웹 서버 진입점인 asgi.py에서만 시작합니다.
# asgi.py
application = get_asgi_application()
# 스케줄러는 웹 서버 프로세스에서만 시작한다
from sync.scheduler import start_scheduler # noqa: E402
start_scheduler()
asgi.py는 gunicorn이 앱을 로드할 때만 import됩니다. management command와 테스트는 이 파일을 거치지 않으므로 스레드가 뜨지 않습니다. 테스트에서는 스레드 없이 동기화 함수를 직접 호출하고, 외부 API 호출은 conftest의 autouse fixture로 모든 테스트에서 막았습니다.
설정으로도 한 번 더 막습니다. start_scheduler()는 활성화 플래그(AUTO_SYNC=true)와 API 키 등 필요한 설정이 모두 있을 때만 스레드를 띄우고, 하나라도 없으면 서버는 정상 기동하되 스케줄러만 꺼집니다. 로컬과 CI는 기본값이 꺼짐입니다.
5. gunicorn --preload와 함께 쓸 수 없다
gunicorn은 마스터 프로세스가 fork로 워커 프로세스를 만듭니다. fork는 메모리는 그대로 복사하지만, 스레드는 fork를 호출한 스레드 하나만 복사합니다.
--preload 옵션을 켜면 gunicorn은 앱을 마스터에서 먼저 로드한 뒤 fork합니다. 그러면 asgi.py가 마스터에서 실행되어 스케줄러 스레드는 마스터에만 생기고, 워커에는 스레드가 복사되지 않습니다. 워커마다 스레드가 하나씩 뜬다는 전제가 깨집니다.
그래서 --preload는 쓰지 않고, 그 이유를 서비스 문서에 적어 두었습니다. 메모리를 아끼려고 무심코 켜기 쉬운 옵션이라 문서화가 중요했습니다.
6. DB 연결을 직접 관리해야 한다
Django는 요청 시작과 종료 시점(request_started, request_finished 시그널)에 오래되거나 끊긴 DB 연결을 정리합니다. 요청과 무관하게 도는 스레드에서는 이 정리가 일어나지 않아, 직접 닫지 않으면 연결이 계속 열려 있습니다.
그래서 run_sync_safely의 finally에서 connection.close()를 직접 호출합니다. 실행 여부만 확인하고 건너뛸 때도 이 finally를 거치므로, 스레드가 대기하는 1시간 동안은 DB 연결을 점유하지 않습니다.
트랜잭션 범위도 주의했습니다. 운영 DB 연결에는 타임아웃이 두 가지 있습니다. 쿼리 하나가 30초를 넘으면 취소되고(statement_timeout), 트랜잭션을 연 채 60초 동안 아무 쿼리도 실행하지 않으면 세션이 끊깁니다(idle_in_transaction_session_timeout). 동기화 전체를 트랜잭션 하나로 묶으면 외부 API 응답을 기다리는 동안 60초 제한에 걸립니다. 그래서 API를 기다리는 동안에는 트랜잭션을 열지 않고, 가져온 데이터를 저장할 때만 짧게 엽니다.
이 60초 제한은 잠금 방식을 고를 때도 결정적이었습니다.
중복 실행 막기: PostgreSQL advisory lock
왜 중복 실행이 문제인가
앞에서 말했듯 스케줄러 스레드는 프로세스마다 하나씩 뜨므로, 평소 2개에서 배포 중에는 4개까지 늘어납니다. 이 스레드들이 정각에 동시에 깨어나 이력을 조회하면, 아직 아무도 성공 이력을 남기지 않았으므로 모두 실행을 시작합니다. 그 순간 수동 동기화 요청까지 들어오면 하나가 더 늘어납니다.
동시에 여러 개가 실행되면 이런 문제가 생깁니다.
- 외부 API 호출 제한을 넘깁니다. '같은 API는 1초에 한 번' 같은 제한은 프로세스 안에서만 지킬 수 있습니다. 프로세스 4개가 각자 지켜도 외부 서비스 입장에서는 4배로 호출되는 셈이고, 비정상 호출로 차단될 수 있습니다.
- 오래된 데이터가 최신 데이터를 덮어씁니다. A가 먼저 조회하고 B가 나중에 조회했는데 저장은 B가 먼저, A가 나중에 끝나면 최신 상태가 이전 상태로 덮입니다. '출고 완료'였던 주문이 '출고 대기'로 돌아가는 식입니다.
- 같은 신규 데이터를 동시에 INSERT하다 하나가 실패합니다. 저희 upsert는 조회 → 비교 → 저장 순서라서, 두 프로세스가 동시에 '없음'을 확인하고 INSERT하면 하나는 unique 제약 위반으로 실패합니다.
INSERT ... ON CONFLICT를 쓰면 이 경합을 피할 수 있지만, 저희 테이블은 '삭제되지 않은 행끼리만 유일'한 조건부 unique 제약(partial unique index)을 쓰고 있어서 Django의bulk_create(update_conflicts=True)로는 처리할 수 없었습니다.
그래서 필요한 건 프로세스와 컨테이너를 넘어 '동기화는 한 번에 하나만' 실행되게 하는 방법이었습니다.
잠금 방법 비교
| 방법 | 문제 |
|---|---|
threading.Lock | 같은 프로세스 안에서만 유효하다. 다른 워커나 컨테이너는 알 수 없다 |
| 파일 잠금 | 같은 컨테이너 안에서만 유효하다 |
| 테이블에 '진행 중' 행 기록 | 프로세스가 죽으면 '진행 중'이 그대로 남는다. 'N분 지나면 죽은 것으로 간주' 같은 만료 규칙이 필요하다 |
SELECT ... FOR UPDATE, pg_advisory_xact_lock | 잠금이 트랜잭션에 묶인다. 작업 내내 트랜잭션을 열어 둬야 해서 60초 제한에 걸린다 |
| Redis 분산 락 | 잠금 하나 때문에 Redis를 도입해야 하고, TTL을 작업 시간에 맞춰 정해야 한다 |
advisory lock 자체는 이미 쓰고 있었습니다. 채번처럼 짧은 구간에는 트랜잭션 단위 잠금(pg_advisory_xact_lock)을 썼는데, 트랜잭션이 끝나면 자동으로 풀려 편하지만 몇 분씩 외부 API를 기다리는 작업에는 맞지 않았습니다.
세션 단위 advisory lock
advisory lock은 테이블이나 행이 아니라 애플리케이션이 정한 숫자(키)에 거는 잠금입니다. PostgreSQL은 그 숫자의 의미는 모르고, 누가 그 키를 잡고 있는지만 관리합니다. 테이블을 잠그지 않으므로 동기화 중에도 조회 쿼리는 영향을 받지 않습니다.
PostgreSQL 함수 pg_try_advisory_lock과 pg_advisory_unlock을 컨텍스트 매니저로 감싸면 이렇게 쓸 수 있습니다.
from contextlib import contextmanager
from django.db import connection
@contextmanager
def advisory_lock(key: int):
"""잠금을 얻었는지 넘겨주고, 블록이 끝나면 반드시 해제한다"""
with connection.cursor() as cursor:
cursor.execute("SELECT pg_try_advisory_lock(%s)", [key])
acquired = cursor.fetchone()[0]
try:
yield acquired
finally:
if acquired:
with connection.cursor() as cursor:
cursor.execute("SELECT pg_advisory_unlock(%s)", [key])
이름에 xact가 없는 함수는 잠금이 트랜잭션이 아니라 세션(DB 연결)에 묶입니다. 이 성질이 저희 조건에 맞았습니다.
- 트랜잭션 없이도 잠금이 유지되어 60초 제한에 걸리지 않습니다.
- 프로세스가 죽으면 자동으로 풀립니다. 연결이 끊기면 PostgreSQL이 세션 종료로 판단해 그 세션의 잠금을 모두 해제합니다. '진행 중' 행처럼 흔적이 남지 않아, 데몬 스레드가 강제 종료되는 상황과도 잘 맞습니다. (네트워크 단절이라면 PostgreSQL이 감지하기까지 시간이 걸릴 수 있습니다.)
pg_try_advisory_lock은 잠금을 얻지 못하면 대기하지 않고 즉시false를 반환합니다.- 추가 인프라가 필요 없습니다. 모든 프로세스가 이미 같은 PostgreSQL을 쓰고 있습니다.
정각에는 이렇게 동작합니다.
11:00:00 워커 1 try_lock 성공 → 동기화 → 성공 이력 기록 → unlock
11:00:00 워커 2 try_lock 실패 → 12:00까지 대기
11:00:00 워커 3 try_lock 실패 → 12:00까지 대기
11:00:00 워커 4 try_lock 실패 → 12:00까지 대기
동기화 1회 실행 함수는 이 잠금을 얻었을 때만 일합니다. 키(SYNC_LOCK_KEY)를 정하는 방법은 바로 아래에서 다룹니다.
def sync_once():
with advisory_lock(SYNC_LOCK_KEY) as acquired:
if not acquired:
return # 다른 프로세스가 이미 동기화 중
sync_orders()
잠금 키 만들기
advisory lock의 키는 64비트 정수이고, 데이터베이스 전체가 같은 키 공간을 공유합니다. 여러 서비스가 DB 하나를 함께 쓰거나 다른 기능에서도 advisory lock을 쓴다면, 키를 아무렇게나 정했다가 다른 잠금과 충돌할 수 있습니다.
그래서 용도를 나타내는 이름을 해시해 64비트 정수로 만듭니다. 대상별로 따로 잠가야 한다면 f"order-sync:{tenant_id}"처럼 이름에 대상 ID를 붙이면 됩니다.
import hashlib
def lock_key(name: str) -> int:
"""이름을 PostgreSQL bigint(부호 있는 64비트) 범위의 정수로 바꾼다"""
digest = hashlib.blake2b(name.encode(), digest_size=8).digest()
return int.from_bytes(digest, "big", signed=True)
SYNC_LOCK_KEY = lock_key("order-sync")
파이썬 내장 hash()는 쓰면 안 됩니다. 문자열 해시는 보안상 이유로 프로세스마다 시드가 달라져서, 워커마다 다른 키를 잡게 되고 잠금이 무의미해집니다. 그래서 항상 같은 값이 나오는 blake2b를 씁니다.
함정: 연결 풀과 세션 잠금
세션 잠금은 세션이 끝날 때까지 유지됩니다. 그런데 연결 풀을 쓰면 이 성질이 함정이 됩니다.
저희는 운영에서 psycopg3 연결 풀을 씁니다(Django 5.1부터 OPTIONS["pool"]로 지원). 풀을 켜면 connection.close()는 연결을 실제로 닫지 않고 풀에 반납합니다. Django 5.2 코드를 보면 이렇습니다.
# django/db/backends/postgresql/base.py
def _close(self):
if self.connection is not None:
with self.wrap_database_errors:
if self.pool:
self.connection._pool.putconn(self.connection)
self.connection = None
else:
return self.connection.close()
PostgreSQL 입장에서는 세션이 계속 살아 있으므로 잠금도 유지됩니다. 풀은 반납된 연결에 열린 트랜잭션이 있으면 롤백하지만, 세션 잠금은 롤백으로 풀리지 않습니다. unlock 없이 연결을 반납하면 이런 일이 생깁니다.
- 잠금을 가진 연결이 풀에 계속 남아, 다른 워커는 매번
try_lock에 실패하고 동기화가 조용히 멈춥니다. - 그 연결을 다음에 가져간 요청이 잠금을 물려받습니다. 게다가 advisory lock은 같은 세션이 다시 요청하면 또 성공하고, 잡은 횟수만큼 풀어야 해제됩니다. 새어 나간 잠금이 엉뚱한 곳에서 중첩될 수 있습니다.
더 까다로운 건 로컬에서는 재현되지 않는다는 점입니다. 로컬과 테스트에서는 연결 풀을 끄기 때문에 close()가 실제로 연결을 닫고 잠금도 함께 풀립니다. 운영에서만 드러나는 버그입니다.
그래서 연결을 반납하기 전에 반드시 잠금부터 해제합니다. 앞의 예시에서 advisory_lock은 블록이 끝나는 즉시 잠금을 풀고, 연결 반납(connection.close())은 그 바깥인 run_sync_safely의 finally에서 합니다. 순서가 '잠금 해제 → 연결 반납'으로 고정되는 거죠. Django의 DB 연결은 스레드마다 따로라서, 잠금도 그 잠금을 잡은 스레드에서만 풀 수 있다는 점도 기억해 두면 좋습니다.
연결이 끊겨서 해제 쿼리가 실패하는 경우라면, 세션이 끝나면서 잠금도 함께 풀립니다. 더 확실히 하려면 풀의 reset 콜백에서 pg_advisory_unlock_all()을 실행해, 반납되는 모든 연결의 세션 잠금을 비우는 방법도 있습니다(저희는 아직 적용하지 않았습니다).
덤: 잠금으로 '동기화 중' 여부 확인하기
advisory lock은 pg_locks 시스템 뷰에서 조회할 수 있고, 다른 프로세스가 잡은 잠금도 보입니다. 그래서 API 응답에 '동기화 중' 여부를 담을 때 별도 상태 컬럼 없이 잠금 여부만 확인하면 됩니다.
def is_locked(key: int) -> bool:
"""다른 프로세스가 잡은 잠금도 보인다"""
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT EXISTS (
SELECT 1 FROM pg_locks
WHERE locktype = 'advisory' AND granted
AND classid = %s::oid AND objid = %s::oid AND objsubid = 1
)
""",
[(key >> 32) & 0xFFFFFFFF, key & 0xFFFFFFFF],
)
return cursor.fetchone()[0]
pg_locks는 64비트 키를 상위 32비트(classid)와 하위 32비트(objid)로 나눠 보여 주고, bigint 키 하나로 잡은 잠금은 objsubid가 1입니다. 그래서 키를 둘로 나눠 비교합니다.
'진행 중'을 컬럼으로 관리하면 서버가 죽었을 때 값이 남는 문제가 다시 생기지만, 잠금으로 판단하면 세션이 끝나는 순간 '동기화 중' 상태도 함께 사라집니다.
수동 동기화 API: 프로세스 내부 잠금 추가
수동 동기화 API에는 프로세스 내부 잠금을 하나 더 둡니다. 같은 워커에 요청이 연달아 들어오면(더블 클릭 등) 두 요청 모두 '실행 중 아님'을 확인하고 각자 스레드를 만들 수 있습니다. 실제 동기화는 advisory lock이 하나만 실행되게 막아 주지만, 그 전에 쓸모없는 스레드가 생기고 두 요청 모두 '시작함'으로 응답하게 됩니다. 그래서 프로세스 내부는 threading.Lock, 프로세스 간은 advisory lock이 담당합니다.
_manual_lock = threading.Lock()
def trigger_manual_sync() -> bool:
"""이미 실행 중이면 False, 아니면 백그라운드에서 시작하고 바로 True를 돌려준다"""
if is_locked(SYNC_LOCK_KEY) or not _manual_lock.acquire(blocking=False):
return False
def run():
try:
run_sync_safely()
finally:
_manual_lock.release()
try:
threading.Thread(target=run, name="manual-sync", daemon=True).start()
except RuntimeError: # 스레드를 못 띄우면 잠금을 바로 돌려놓는다
_manual_lock.release()
raise
return True
acquire(blocking=False)는 잠금을 얻지 못하면 대기 없이 False를 반환합니다. 또 threading.Lock은 획득한 스레드가 아니어도 해제할 수 있어서(RLock은 불가), 요청 스레드가 획득하고 동기화 스레드가 끝날 때 해제합니다.
수동 동기화 API는 동기화 완료를 기다리지 않고 즉시 응답합니다. 몇 분 걸릴 수 있는 작업을 요청 안에서 처리하면 로드밸런서 idle timeout(AWS ALB 기본 60초)에 걸릴 수 있기 때문입니다. 클라이언트는 몇 초 뒤 목록을 다시 조회해 결과를 확인합니다.
언제 다른 방식으로 옮길까
이 구조는 지금 조건에서 맞는 선택입니다. 다음 중 하나라도 해당되면 별도 스케줄러로 옮길 계획입니다.
- 주기 작업이 서너 개 이상으로 늘어날 때. 작업마다 루프·잠금·이력을 따로 관리하면 금방 복잡해집니다.
- 작업이 CPU를 많이 쓰게 될 때. GIL 때문에 같은 프로세스의 웹 요청까지 느려집니다.
- 재시도 정책, 실행 모니터링 대시보드, 작업 큐 같은 기능이 본격적으로 필요할 때. Celery나 클라우드 스케줄러가 이미 잘하는 일입니다.
- 정확한 시각에 정확히 한 번 실행되어야 할 때. 지금 구조가 보장하는 건 '정각 무렵 한 번 실행되고, 중복 실행되더라도 결과는 같다'까지입니다.
옮기기 쉽도록 '언제 실행할지'(스케줄러)와 '무엇을 실행할지'(동기화 함수)를 처음부터 분리했습니다. 스케줄러는 run_sync_safely()를 호출하기만 하므로, 이 함수를 management command로 감싸 cron이나 클라우드 스케줄러에서 호출하면 됩니다. 잠금과 이력 기록은 동기화 함수 쪽에 있어서 그대로 따라갑니다.
마치며
돌아보면 시간이 든 건 스레드를 띄우는 한 줄이 아니라, 그 스레드가 웹 서버 프로세스 안에서 돌면서 생기는 문제를 하나씩 막는 일이었습니다. 예외로 죽는 스레드, 재시작, 실행 중 강제 종료, 의도하지 않은 시작 위치, --preload, DB 연결 관리, 그리고 중복 실행까지요. 그 결과 별도 스케줄러 없이 코드 수십 줄과 몇 가지 규칙만으로 매시간 동기화를 구현할 수 있었습니다. 작업 하나를 위해 인프라를 늘리는 것보다 훨씬 가벼운 선택이었습니다.
작업이 가볍고 I/O 대기 위주, 중간에 끊기거나 중복 실행돼도 괜찮, 여러 프로세스 중 하나만 실행. 셋 다 그렇다면 웹 서버 안 스레드도 충분히 쓸 만한 선택입니다.