안녕하세요, 개발자 여러분! 오늘 다룰 주제는 물류 및 재고 관리 시스템에서 매우 중요하지만 동시에 까다로운 동시성(Concurrency) 문제입니다. 특히 FEFO(First Expired, First Out) 원칙에 따라 여러 재고 배치(Batch)에서 동시다발적인 주문에 대한 재고를 예약해야 할 때, 데이터 일관성을 유지하는 것은 쉽지 않습니다. 잘못된 구현은 초과 판매(Overselling)와 같은 심각한 비즈니스 문제로 이어질 수 있습니다.
이 글에서는 Flask, SQLAlchemy 2.x, PostgreSQL 환경에서 FEFO 재고 동시 예약 시 발생할 수 있는 문제점을 분석하고, SELECT ... FOR UPDATE와 견고한 트랜잭션 관리 전략을 통해 이를 안전하게 해결하는 방법을 구체적인 코드 예시와 함께 제시합니다.
Flask 기반의 물류 & 재고 ERP 시스템을 운영하고 있다고 가정해봅시다. 주문이 확정되면 시스템은 FEFO(만료일이 가장 빠른 재고부터) 원칙에 따라 가용 재고 배치에서 재고를 예약해야 합니다. 이때 다음과 같은 요구사항과 잠재적인 문제에 직면할 수 있습니다.
다중 주문 동시성: 여러 주문이 거의 동시에 확정되어 재고 예약 요청이 들어옵니다.
정확한 재고 확인: 각 주문은 실제로 사용 가능한 재고만을 예약해야 합니다.
다중 배치 예약: 하나의 주문에 필요한 수량이 단일 배치만으로 충족되지 않아 여러 재고 배치에서 분할 예약해야 할 수 있습니다.
원자성(Atomicity): 충분한 재고가 없으면 예약은 부분적인 변경 없이 완전히 실패해야 합니다.
멱등성(Idempotency): 네트워크 오류 등으로 인한 재시도가 발생해도 중복 예약이 생성되지 않아야 합니다.
기존 구현을 검토할 때, 이러한 동시성 요청 아래에서 재고 예약 로직이 안전한지 확인해야 합니다. 특히, 초과 판매 버그가 발생할 가능성이 있는지 면밀히 검토하는 것이 중요합니다.
이러한 상황에서 개발자들은 다음과 같은 질문에 대한 답을 필요로 합니다.
PostgreSQL의 SELECT ... FOR UPDATE를 사용하여 가용 배치 행을 락킹해야 하는가?
예약이 여러 FEFO 배치에 걸쳐 발생할 때 행은 어떻게 락킹해야 하는가?
예약 기록과 재고 잔고 업데이트가 원자적으로 커밋되거나 롤백되도록 어떻게 보장할 수 있는가?
애플리케이션 레벨 락킹 외에 어떤 데이터베이스 제약 조건이나 멱등성 보호 장치를 보충해야 하는가?
2. 명확한 발생 원인
위와 같은 요구사항을 충족하지 못하는 시스템에서 발생하는 문제의 핵심은 경합 조건(Race Condition)과 트랜잭션 일관성 부족입니다.
경합 조건 (Race Condition)
두 개 이상의 트랜잭션이 동시에 동일한 재고 배치 데이터를 읽고 수정하려고 할 때 발생합니다.
예를 들어, 트랜잭션 A가 특정 재고 배치의 수량(예: 10개)을 확인하고 예약 로직을 시작합니다. 이때 트랜잭션 B도 동일한 재고 배치의 수량을 확인하고 예약 로직을 시작할 수 있습니다. 만약 두 트랜잭션 모두 “10개 재고가 있다”고 판단하고 각각 5개씩 예약한다면, 실제로는 10개만 있는 재고가 총 10개 소진(5+5)되어 재고가 0이 되어야 하지만, 두 트랜잭션이 모두 5개씩 차감하는 업데이트를 수행하면 재고가 -10이 되는 초과 판매가 발생할 수 있습니다.
데이터 일관성 문제
Phantom Reads (환영 읽기): 트랜잭션이 특정 조건을 만족하는 행 집합을 조회했는데, 나중에 같은 트랜잭션 내에서 다시 조회했을 때 다른 트랜잭션이 추가/삭제한 행 때문에 결과 집합이 달라지는 경우. FEFO 재고 조회 시 중간에 새로운 배치가 추가되거나 사라지면 계산이 틀어질 수 있습니다.
Lost Updates (갱신 손실): 한 트랜잭션이 데이터를 수정한 후 커밋하기 전에 다른 트랜잭션이 동일한 데이터를 수정하여 앞선 트랜잭션의 변경 사항이 덮어씌워져 유실되는 경우.
원자성 부족 (Lack of Atomicity)
단일 주문이 여러 재고 배치에 걸쳐 재고를 예약해야 할 때, 이 모든 과정이 하나의 트랜잭션으로 묶이지 않으면 문제가 됩니다. 예를 들어, 100개 중 첫 번째 배치에서 50개를 성공적으로 예약하고 다음 배치에서 50개를 예약하려는데 실패하면, 중간 상태(50개만 예약된 상태)가 되어 비즈니스 로직의 일관성이 깨집니다.
멱등성 부족 (Lack of Idempotency)
클라이언트의 요청이 네트워크 문제 등으로 인해 여러 번 서버에 전달될 때, 매번 같은 결과를 보장하지 못하면 중복 예약이 발생할 수 있습니다. 예를 들어, 이미 처리된 주문에 대해 다시 예약 요청이 들어왔을 때, 시스템은 이를 감지하고 중복 예약을 생성하지 않거나 기존 예약을 다시 반환해야 합니다.
이러한 문제들은 적절한 데이터베이스 트랜잭션 관리와 락킹(Locking) 메커니즘을 통해 해결할 수 있습니다.
3. 해결 방법 및 코드 예시
동시성 재고 예약 문제를 해결하기 위한 핵심 전략은 `SELECT … FOR UPDATE` 구문을 활용하여 필요한 모든 재고 배치에 명시적인 행 레벨 락(Row-level Lock)을 걸고, 이 모든 과정을 단일 데이터베이스 트랜잭션으로 묶는 것입니다. 또한, 멱등성을 보장하기 위한 애플리케이션 및 DB 레벨의 안전장치를 마련해야 합니다.
3.1. 핵심 전략: `SELECT … FOR UPDATE` 및 트랜잭션
PostgreSQL의 SELECT ... FOR UPDATE는 조회된 행에 대해 다른 트랜잭션이 해당 행을 수정하거나 동일한 FOR UPDATE 락을 걸지 못하도록 배타적 락을 설정합니다. 이렇게 하면 현재 트랜잭션이 재고를 확인하고 업데이트하는 동안 다른 트랜잭션은 해당 행에 접근하지 못하고 대기하거나 데드락(Deadlock)이 발생할 수 있습니다 (이 경우 데드락 감지 후 한 트랜잭션은 롤백됩니다).
FEFO 로직에서는 만료일이 빠른 순서로 여러 배치를 조회하고 사용해야 하므로, 필요한 모든 잠재적 재고 배치들을 한 번에 FOR UPDATE로 락킹하여 전체 재고 계산 및 차감 과정의 일관성을 보장하는 것이 중요합니다. 모든 작업은 단일 트랜잭션 내에서 이루어져야 하며, 중간에 오류가 발생하면 전체 트랜잭션을 롤백하여 원자성을 유지합니다.
3.2. SQLAlchemy 2.x와 PostgreSQL을 활용한 코드 예시
다음은 Flask, SQLAlchemy, PostgreSQL 환경에서 FEFO 재고 예약 로직을 안전하게 구현하는 예시 코드입니다.
from sqlalchemy import create_engine, Column, Integer, String, DateTime, func, Boolean
from sqlalchemy.orm import sessionmaker, declarative_base
from datetime import datetime, timedelta
import threading
import time
# 데이터베이스 설정 (실제 환경에 맞게 수정 필요)
DATABASE_URL = "postgresql://user:password@localhost:5432/inventory_db"
engine = create_engine(DATABASE_URL)
Session = sessionmaker(bind=engine)
Base = declarative_base()
# 모델 정의
class InventoryBatch(Base):
__tablename__ = 'inventory_batches'
id = Column(Integer, primary_key=True)
product_id = Column(Integer, nullable=False)
quantity = Column(Integer, nullable=False) # 현재 재고 수량
expiry_date = Column(DateTime, nullable=False)
location = Column(String, nullable=False)
def __repr__(self):
return f"<InventoryBatch(id={self.id}, product_id={self.product_id}, quantity={self.quantity}, expiry_date={self.expiry_date.strftime('%Y-%m-%d')})>"
class Reservation(Base):
__tablename__ = 'reservations'
id = Column(Integer, primary_key=True)
order_id = Column(Integer, nullable=False, unique=True) # 주문 ID는 고유해야 함 (멱등성 보장)
product_id = Column(Integer, nullable=False)
reserved_quantity = Column(Integer, nullable=False)
reservation_date = Column(DateTime, default=func.now())
is_confirmed = Column(Boolean, default=False) # 예약 확정 상태
def __repr__(self):
return f"<Reservation(id={self.id}, order_id={self.order_id}, product_id={self.product_id}, reserved_quantity={self.reserved_quantity}, confirmed={self.is_confirmed})>"
class ReservationDetail(Base):
__tablename__ = 'reservation_details'
id = Column(Integer, primary_key=True)
reservation_id = Column(Integer, nullable=False)
batch_id = Column(Integer, nullable=False)
reserved_quantity_from_batch = Column(Integer, nullable=False)
def __repr__(self):
return f"<ReservationDetail(id={self.id}, reservation_id={self.reservation_id}, batch_id={self.batch_id}, quantity={self.reserved_quantity_from_batch})>"
# 테이블 생성 (최초 실행 시 필요)
# Base.metadata.create_all(engine)
def reserve_inventory_fefo(order_id: int, product_id: int, requested_quantity: int):
"""
FEFO 원칙에 따라 재고를 예약하고, 동시성 문제를 안전하게 처리합니다.
"""
session = Session() # 각 요청마다 새로운 세션을 생성하여 트랜잭션 격리를 보장
try:
# 1. 멱등성 보장: 이미 해당 order_id로 예약된 내역이 있는지 확인
# unique=True 제약조건이 있지만, 애플리케이션 레벨에서 미리 처리하여 불필요한 예외 방지
existing_reservation = session.query(Reservation).filter_by(order_id=order_id).first()
if existing_reservation:
if existing_reservation.is_confirmed:
print(f"[Order {order_id}] 이미 확정된 예약입니다. (Reserved: {existing_reservation.reserved_quantity})")
session.rollback() # 트랜잭션을 롤백하여 세션 정리
return existing_reservation
else:
# 확정되지 않은 예약이 있는 경우, 기존 예약을 취소하고 새로 시도
# 또는 업데이트 로직으로 변경 가능. 여기서는 단순화를 위해 삭제 후 재진행.
print(f"[Order {order_id}] 확정되지 않은 기존 예약을 발견했습니다. 재시도합니다.")
session.query(ReservationDetail).filter_by(reservation_id=existing_reservation.id).delete()
session.delete(existing_reservation)
session.flush() # 변경 사항을 DB에 즉시 반영 (아직 커밋은 아님)
# 2. FEFO 원칙에 따라 만료일이 빠른 순서로 재고 배치들을 조회 및 락킹
# .with_for_update(of=InventoryBatch)를 사용하여 조회된 InventoryBatch 행에 락을 설정합니다.
# 이렇게 하면 다른 트랜잭션이 이 배치들의 quantity를 변경하거나 추가로 락을 걸 수 없게 됩니다.
candidate_batches = session.query(InventoryBatch) \
.filter(InventoryBatch.product_id == product_id,
InventoryBatch.quantity > 0) \
.order_by(InventoryBatch.expiry_date.asc(), InventoryBatch.id.asc()) \
.with_for_update(of=InventoryBatch, nowait=False) \
.all()
# nowait=True: 락을 획득할 수 없으면 즉시 오류 발생 (LockNotAvailableError)
# nowait=False (기본값): 락을 획득할 수 있을 때까지 대기
total_available_quantity = sum(batch.quantity for batch in candidate_batches)
if total_available_quantity < requested_quantity:
print(f"[Order {order_id}] 재고 부족! 요청: {requested_quantity}, 가용: {total_available_quantity}")
session.rollback() # 재고 부족 시 트랜잭션 롤백
return None
# 3. 예약 레코드 생성 및 재고 차감 로직
new_reservation = Reservation(order_id=order_id, product_id=product_id, reserved_quantity=requested_quantity, is_confirmed=False)
session.add(new_reservation)
session.flush() # new_reservation.id를 얻기 위해 flush (DB에 즉시 쓰지만 커밋은 아님)
remaining_quantity_to_reserve = requested_quantity
reserved_batches_details = []
for batch in candidate_batches:
if remaining_quantity_to_reserve <= 0:
break
quantity_from_batch = min(batch.quantity, remaining_quantity_to_reserve)
batch.quantity -= quantity_from_batch # 재고 배치 수량 차감
reserved_batches_details.append(
ReservationDetail(
reservation_id=new_reservation.id,
batch_id=batch.id,
reserved_quantity_from_batch=quantity_from_batch
)
)
remaining_quantity_to_reserve -= quantity_from_batch
# 만약 여기까지 왔는데도 예약해야 할 수량이 남아있다면 논리적 오류 (위에서 총 가용량 검사했으므로)
if remaining_quantity_to_reserve > 0:
print(f"[Order {order_id}] 심각한 오류: 예약 로직 불일치. 남은 수량: {remaining_quantity_to_reserve}")
session.rollback()
return None
session.add_all(reserved_batches_details)
new_reservation.is_confirmed = True # 예약 완료 상태로 변경
session.commit() # 모든 변경 사항을 원자적으로 커밋
print(f"[Order {order_id}] 제품 {product_id} {requested_quantity}개 예약 성공!")
return new_reservation
except Exception as e:
print(f"[Order {order_id}] 예약 중 오류 발생: {e}")
session.rollback() # 오류 발생 시 트랜잭션 롤백
return None
finally:
session.close() # 세션 닫기
# 초기 데이터 로드 (한번만 실행하여 테스트 환경 설정)
def setup_initial_data():
session = Session()
# 데이터가 없으면 초기 데이터 추가
if not session.query(InventoryBatch).first():
print("초기 재고 데이터를 추가합니다...")
session.add_all([
InventoryBatch(product_id=101, quantity=50, expiry_date=datetime.now() + timedelta(days=30), location="Warehouse A"),
InventoryBatch(product_id=101, quantity=70, expiry_date=datetime.now() + timedelta(days=60), location="Warehouse B"),
InventoryBatch(product_id=101, quantity=30, expiry_date=datetime.now() + timedelta(days=15), location="Warehouse C"), # 가장 먼저 만료
InventoryBatch(product_id=102, quantity=100, expiry_date=datetime.now() + timedelta(days=90), location="Warehouse A")
])
session.commit()
session.close()
# 동시성 테스트를 위한 헬퍼 함수
def run_concurrent_reservations(order_prefix, num_orders, product_id, requested_quantity):
threads = []
print(f"\n--- {num_orders}개의 주문 ({product_id}번 제품 {requested_quantity}개씩) 동시 실행 ---")
for i in range(num_orders):
order_id = int(f"{order_prefix}{i}")
thread = threading.Thread(target=reserve_inventory_fefo, args=(order_id, product_id, requested_quantity))
threads.append(thread)
thread.start()
# time.sleep(0.005) # 약간의 지연으로 동시성 발생 확률 높임
for thread in threads:
thread.join()
if __name__ == '__main__':
print("--- 데이터베이스 및 초기 데이터 설정 ---")
# Base.metadata.drop_all(engine) # 테스트 시 기존 테이블 삭제 후 재생성 (선택 사항)
Base.metadata.create_all(engine) # 테이블 생성
setup_initial_data()
print("\n--- 현재 재고 상태 ---")
s = Session()
batches = s.query(InventoryBatch).order_by(InventoryBatch.expiry_date.asc()).all()
for batch in batches:
print(batch)
s.close()
# 시나리오 1: 여러 주문이 충분한 재고를 동시에 예약 시도
# product 101의 총 초기 재고는 50 + 70 + 30 = 150개.
# 각 주문이 20개씩 요청하면 최대 7개 주문(140개)까지 성공 가능.
# 8개 주문을 동시에 실행하여 동시성 처리 확인.
run_concurrent_reservations(1000, 8, 101, 20)
print("\n--- 시나리오 1 종료 후 재고 상태 ---")
s = Session()
batches = s.query(InventoryBatch).order_by(InventoryBatch.expiry_date.asc()).all()
for batch in batches:
print(batch)
print("\n--- 시나리오 1 종료 후 예약 내역 ---")
reservations = s.query(Reservation).order_by(Reservation.order_id).all()
for res in reservations:
print(res)
details = s.query(ReservationDetail).filter_by(reservation_id=res.id).all()
for detail in details:
print(f" -> 배치 {detail.batch_id}: {detail.reserved_quantity_from_batch}개")
s.close()
# 시나리오 2: 재고 부족 상황에서 재시도 및 멱등성 확인
# product 101 재고는 이제 10개 (150 - 140) 남았을 것으로 예상.
# order_id 20000과 20001이 각각 20개를 요청하며, 특히 20000은 이미 생성되었지만 확정되지 않은 상태로 가정 (예시에서 재시도).
# 이 테스트는 위에 "existing_reservation" 로직을 확인하기 위함
# 실제로는 `setup_initial_data` 함수에 다음을 추가하여 시작 상태를 만들 수 있음
# session.add(Reservation(order_id=20000, product_id=101, reserved_quantity=20, is_confirmed=False))
# run_concurrent_reservations(2000, 2, 101, 20) # 2개 주문 중 하나만 성공 or 모두 실패 예상
print("\n--- 최종 재고 상태 ---")
s = Session()
batches = s.query(InventoryBatch).order_by(InventoryBatch.expiry_date.asc()).all()
for batch in batches:
print(batch)
print("\n--- 최종 예약 내역 ---")
reservations = s.query(Reservation).order_by(Reservation.order_id).all()
for res in reservations:
print(res)
details = s.query(ReservationDetail).filter_by(reservation_id=res.id).all()
for detail in details:
print(f" -> 배치 {detail.batch_id}: {detail.reserved_quantity_from_batch}개")
s.close()
3.3. 코드 설명 및 중요 포인트
모델 정의
InventoryBatch: 실제 재고 배치를 나타내며, quantity 컬럼이 현재 수량을 저장합니다.
Reservation: 주문에 대한 전체 예약 정보를 저장합니다. order_id에 unique=True 제약 조건을 추가하여 DB 레벨에서 중복 예약을 방지합니다. is_confirmed 플래그를 통해 예약의 최종 확정 상태를 관리합니다.
ReservationDetail: 하나의 예약이 여러 재고 배치에서 이루어질 수 있으므로, 각 배치에서 얼마나 예약되었는지를 기록하는 상세 테이블입니다.
`reserve_inventory_fefo` 함수
트랜잭션 관리: 각 예약 요청은 Session()을 통해 독립적인 세션을 얻고, try...except...finally 블록 내에서 session.commit() 또는 session.rollback()으로 원자성을 보장합니다.
멱등성 처리: Reservation.order_id의 unique 제약 조건과 함께, 함수 시작 부분에서 order_id로 이미 존재하는 예약이 있는지 확인합니다. 이미 확정된 예약이라면 처리를 스킵하고, 확정되지 않은 예약이라면 기존 것을 삭제하고 새로 시도하는 로직을 포함합니다. 이는 API 재시도 시 중복 예약이 발생하지 않도록 합니다.
이 부분이 핵심입니다. with_for_update(of=InventoryBatch)를 통해 FEFO 정렬 기준에 따라 조회된 모든 InventoryBatch 행에 락을 겁니다. nowait=False는 락을 얻을 때까지 대기함을 의미하며, nowait=True로 설정하면 락을 얻을 수 없을 때 즉시 LockNotAvailableError를 발생시킬 수 있습니다. 시스템의 요구사항에 따라 선택합니다.
재고 확인 및 차감: 락이 걸린 candidate_batches의 총 가용 수량을 계산하고, 요청 수량보다 적으면 즉시 롤백합니다. 이후 FEFO 순서에 따라 각 배치에서 필요한 만큼 재고를 차감하고, ReservationDetail을 생성합니다.
상태 업데이트: 모든 예약 및 재고 차감 작업이 논리적으로 완료되면 new_reservation.is_confirmed = True로 설정하고 session.commit()을 호출합니다.
3.4. 데이터베이스 제약 조건 및 멱등성 보강
고유 제약 조건 (Unique Constraint): Reservation.order_id 컬럼에 unique=True를 설정하여 데이터베이스 레벨에서 동일한 order_id로 여러 예약이 생성되는 것을 방지합니다.
애플리케이션 레벨의 멱등성 처리: 위 코드 예시에서처럼, 예약 함수 내에서 order_id를 기반으로 기존 예약 존재 여부를 확인하고 적절히 처리하는 로직은 필수적입니다.
트랜잭션 격리 수준 (Isolation Level): PostgreSQL의 기본 격리 수준은 READ COMMITTED입니다. SELECT ... FOR UPDATE와 함께 사용하면 대부분의 동시성 문제를 해결할 수 있습니다. 하지만 매우 엄격한 데이터 일관성이 요구되는 경우 SERIALIZABLE 격리 수준을 고려할 수 있습니다. SERIALIZABLE은 동시성 트랜잭션이 마치 순차적으로 실행된 것처럼 동작하도록 보장하지만, 직렬화 실패(Serialization Failure)로 인한 롤백 비율이 높아져 성능 오버헤드가 있을 수 있으므로 신중하게 선택해야 합니다.
4. 향후 예방을 위한 팁
견고한 재고 시스템을 구축하기 위해서는 단순히 문제를 해결하는 것을 넘어, 향후 발생할 수 있는 문제들을 미리 예방하고 시스템을 지속적으로 개선하는 노력이 중요합니다.
동시성 테스트 및 부하 테스트
실제 운영 환경과 유사한 조건에서 동시성 테스트(Concurrency Test)와 부하 테스트(Load Test)를 정기적으로 수행해야 합니다. 예를 들어, Locust나 JMeter 같은 도구를 사용하여 많은 수의 동시 요청을 발생시켜 시스템의 안정성을 검증하고, 잠재적인 병목 현상이나 데드락 상황을 사전에 발견해야 합니다.
명확한 트랜잭션 경계 설정
모든 비즈니스 로직 단위, 특히 재고와 같이 민감한 데이터를 다루는 부분에서는 트랜잭션의 시작과 끝을 명확히 정의하고, 필요한 모든 작업을 트랜잭션 내에 포함해야 합니다. SQLAlchemy의 세션 관리를 통해 이를 효과적으로 수행할 수 있습니다.
로깅 및 모니터링 강화
트랜잭션 실패, 락 타임아웃, 데드락 발생 등의 이벤트를 상세하게 로깅하고 모니터링 시스템에 연동하여 비정상적인 상황을 즉시 감지하고 대응할 수 있도록 해야 합니다. PostgreSQL의 pg_locks 뷰나 Slow Query 로그 등을 활용하여 락킹 관련 이슈를 분석할 수 있습니다.
데이터베이스 스키마 최적화
재고 테이블에 적절한 인덱스를 추가하여 FEFO 정렬 및 필터링 성능을 최적화해야 합니다. 또한, 재고 상태(예: 예약 중, 출고 완료 등)를 명확하게 나타내는 컬럼이나 테이블을 추가하여 복잡한 비즈니스 로직을 단순화할 수 있습니다.
낙관적 락킹 (Optimistic Locking) 고려
SELECT ... FOR UPDATE와 같은 비관적 락킹(Pessimistic Locking)은 동시성이 높을 때 락 경합으로 인한 성능 저하를 유발할 수 있습니다. 모든 상황에서 비관적 락킹이 최선은 아니며, 읽기 작업이 많고 충돌이 적은 경우에는 버전(version) 컬럼 등을 활용한 낙관적 락킹을 고려해볼 수 있습니다. 업데이트 시 버전이 일치하는 경우에만 적용하는 방식입니다.
이 글에서 제시된 솔루션과 팁이 FEFO 재고 동시 예약 시스템을 견고하게 구축하는 데 도움이 되기를 바랍니다. 궁금한 점이나 더 나은 아이디어가 있다면 언제든지 의견을 공유해주세요!
![[문제 해결] [SQLAlchemy] 레거시 읽기 전용 DB Python 데이터 모델링: 최적의 선택과 전략](https://dev-error.com/wp-content/plugins/contextual-related-posts/default.png "[문제 해결] [SQLAlchemy] 레거시 읽기 전용 DB Python 데이터 모델링: 최적의 선택과 전략")