Transaction Isolation Levels

1
2
3
4
5

-- 트랜잭션 내 갱신 대상 행을 명시적으로 잠금(팬텀 방지엔 범위조건 + 적절한 인덱스 필요)
BEGIN;
SELECT balance FROM accounts WHERE id = 100 FOR UPDATE;  -- X락
UPDATE accounts SET balance = balance - 50 WHERE id = 100;
COMMIT;

1
2
3
4
5

-- 스냅샷 격리로 같은 트랜잭션 내 읽기 일관성 확보(비차단 읽기)
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT sum(amount) FROM orders WHERE status = 'PAID'; -- 스냅샷 기준
-- 다른 세션 갱신과 무관하게 트랜잭션 종료 전까지 조회 결과 일관
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import psycopg2, time
from psycopg2 import extensions, errors

def run_serializable(txn_fn, max_retry=5):
    for i in range(max_retry):
        conn = psycopg2.connect("dbname=app user=app")
        conn.set_isolation_level(extensions.ISOLATION_LEVEL_SERIALIZABLE)
        try:
            with conn:
                with conn.cursor() as cur:
                    txn_fn(cur)  # 비즈니스 로직 수행
            return True
        except errors.SerializationFailure:  # SSI 충돌
            conn.rollback()
            time.sleep(0.05 * (2 ** i))     # 지수 백오프 후 재시도
        finally:
            conn.close()
    return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// 버전 컬럼(version)으로 조건부 업데이트: 동시 갱신 충돌 시 0행 갱신 → 재시도
async function optimisticUpdate(client, id, newData) {
  for (let attempt = 0; attempt < 4; attempt++) {
    const { rows } = await client.query(
      'SELECT payload, version FROM items WHERE id=$1', [id]
    );
    const { version } = rows[0];
    const res = await client.query(
      'UPDATE items SET payload=$1, version=$2 WHERE id=$3 AND version=$4',
      [newData, version + 1, id, version]
    );
    if (res.rowCount === 1) return true;
    await new Promise(r => setTimeout(r, 50 * (2 ** attempt))); // 백오프
  }
  throw new Error('Concurrency conflict');
}

1
2
3
4
5
6

-- 세션 격리 수준 설정(세션/글로벌): RR에서 Next-Key가 범위 잠금으로 팬텀 억제
SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
START TRANSACTION;
SELECT * FROM accounts WHERE id BETWEEN 100 AND 200 FOR UPDATE; -- 범위 잠금
UPDATE accounts SET flag = 1 WHERE id BETWEEN 120 AND 180;
COMMIT;

1
2
3
4
5
6
7

-- 1) DB 레벨로 RCSI 활성(전역 읽기 스냅샷, RC에서도 비차단 읽기)
ALTER DATABASE MyDb SET READ_COMMITTED_SNAPSHOT ON;
-- 2) SNAPSHOT 격리: DB 옵션 ON + 세션에서 선택적으로 사용
ALTER DATABASE MyDb SET ALLOW_SNAPSHOT_ISOLATION ON;
GO
SET TRANSACTION ISOLATION LEVEL SNAPSHOT;  -- 세션에서 선택
BEGIN TRAN; SELECT * FROM Sales WHERE ...; COMMIT;

1
2
3
4
5
6
7
8

-- Oracle은 기본 RC 일관성: 각 쿼리는 SCN 시점의 커밋 데이터만 읽음(Dirty 불가)
-- 갱신 충돌이 우려되는 행은 FOR UPDATE로 명시 잠금
BEGIN
  SELECT balance INTO :b FROM accounts WHERE id = :id FOR UPDATE;
  UPDATE accounts SET balance = balance - :amt WHERE id = :id;
  COMMIT;
END;
/

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# pip install psycopg2-binary
import psycopg2, threading, time

DDL = """
create table if not exists items(id serial primary key, price int, cat int);
truncate items;
insert into items(price, cat) values (100, 1), (150, 1), (200, 2);
"""

def session_read(level):
    """
    격리 수준 효과 관찰: 같은 WHERE 조건을 두 번 읽어 행 수/합계가 변하는지 확인
    """
    conn = psycopg2.connect("dbname=demo user=postgres password=postgres host=127.0.0.1")
    cur = conn.cursor()
    cur.execute(DDL); conn.commit()
    cur.execute(f"begin isolation level {level};")
    cur.execute("select count(*), sum(price) from items where cat=1;")
    before = cur.fetchone()
    time.sleep(2)  # T2 작업 대기
    cur.execute("select count(*), sum(price) from items where cat=1;")
    after = cur.fetchone()
    cur.execute("commit;")
    conn.close()
    print(level, "before/after:", before, after)  # Non-repeatable/Phantom 여부 확인

def session_write():
    conn = psycopg2.connect("dbname=demo user=postgres password=postgres host=127.0.0.1")
    cur = conn.cursor()
    time.sleep(1)
    cur.execute("begin; insert into items(price, cat) values(120, 1); commit;")
    conn.close()

# 실행: Read Committed vs Repeatable Read vs Serializable
threading.Thread(target=session_read, args=("read committed",)).start()
threading.Thread(target=session_write).start()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# pip install pymysql
import pymysql, threading, time

def reader(level):
    conn = pymysql.connect(host="127.0.0.1", user="root", password="root", database="demo", autocommit=False)
    cur = conn.cursor()
    cur.execute("create table if not exists items(id int primary key auto_increment, price int, cat int, key(cat));")
    cur.execute("truncate items;")
    cur.execute("insert into items(price, cat) values (100,1),(150,1),(200,2);")
    conn.commit()

    cur.execute(f"set session transaction isolation level {level}; start transaction;")
    cur.execute("select count(*) from items where cat between 1 and 1;")  # 인덱스 범위 스캔
    print("reader first:", cur.fetchone())
    time.sleep(2)
    cur.execute("select count(*) from items where cat between 1 and 1;")
    print("reader second:", cur.fetchone())
    conn.commit(); conn.close()

def writer():
    time.sleep(1)
    conn = pymysql.connect(host="127.0.0.1", user="root", password="root", database="demo")
    cur = conn.cursor()
    cur.execute("insert into items(price, cat) values (120,1);")
    conn.commit(); conn.close()

# RR에서 next-key/gap lock으로 삽입 대기/차단 → 팬텀 방지
threading.Thread(target=reader, args=("repeatable read",)).start()
threading.Thread(target=writer).start()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import sqlite3

# DB 연결 및 격리 수준 지정
conn = sqlite3.connect(":memory:")
conn.isolation_level = "DEFERRED"  # 실제 DB 엔진별 격리 수준 매핑 필요

def test_isolation():
    """
    격리 수준에 따른 트랜잭션 동작 예시
    - Dirty Read 발생 여부 체크
    """
    cur = conn.cursor()
    # 트랜잭션A: 데이터 읽기
    cur.execute("SELECT balance FROM accounts WHERE account_id = 1")
    print(cur.fetchone())
    # 트랜잭션B: 데이터 변경 후 미커밋
    # … (다른 커넥션에서 실행)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import psycopg2

def run_transaction(level):
    conn = psycopg2.connect("dbname=test user=postgres")
    conn.set_session(isolation_level=level)
    cur = conn.cursor()
    cur.execute("SELECT balance FROM accounts WHERE id=1;")
    print(cur.fetchone())
    conn.commit()

# 실행 예: run_transaction('SERIALIZABLE')

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299

@Service
@Slf4j
public class NaverPayPointService {
    
    @Autowired
    private PointRepository pointRepository;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    /**
     * 네이버페이 스타일 포인트 결제 처리
     * - 이 메서드는 대규모 동시 결제 처리를 담당
     * - 여기서 Transaction Isolation의 실제 상용 서비스 적용 확인 가능
     */
    @Transactional
    public PaymentResult processPointPayment(PointPaymentRequest request) {
        String userId = request.getUserId();
        BigDecimal amount = request.getAmount();
        String orderId = request.getOrderId();
        
        try {
            // 1단계: 캐시에서 빠른 잔액 확인 (Read Committed 수준)
            BigDecimal cachedBalance = getCachedPointBalance(userId);
            if (cachedBalance != null && cachedBalance.compareTo(amount) < 0) {
                throw new InsufficientPointsException("포인트 잔액 부족 (캐시)");
            }
            
            // 2단계: 결제 승인 처리 (Repeatable Read)
            PaymentApproval approval = approvePaymentWithConsistency(
                userId, amount, orderId
            );
            
            // 3단계: 실제 포인트 차감 (Serializable)
            PointTransaction transaction = deductPointsWithIsolation(
                userId, amount, orderId
            );
            
            // 4단계: 캐시 업데이트 및 이벤트 발행
            updateCacheAndPublishEvent(userId, transaction);
            
            return PaymentResult.success(approval, transaction);
            
        } catch (Exception e) {
            log.error("포인트 결제 처리 실패: userId={}, amount={}, orderId={}", 
                userId, amount, orderId, e);
            
            // 보상 트랜잭션 실행
            executeCompensationTransaction(userId, amount, orderId);
            throw new PaymentProcessingException("결제 처리 실패", e);
        }
    }
    
    /**
     * Redis 캐시를 활용한 빠른 잔액 확인
     * - 이 부분은 높은 성능의 읽기 작업을 담당
     * - 여기서 Transaction Isolation의 캐시 활용 패턴 확인 가능
     */
    private BigDecimal getCachedPointBalance(String userId) {
        String cacheKey = "point:balance:" + userId;
        
        try {
            Object cached = redisTemplate.opsForValue().get(cacheKey);
            if (cached != null) {
                return new BigDecimal(cached.toString());
            }
            
            // 캐시 미스 시 DB에서 조회 후 캐시 업데이트
            BigDecimal dbBalance = pointRepository.getPointBalance(userId);
            redisTemplate.opsForValue().set(cacheKey, dbBalance, 
                Duration.ofMinutes(5));
            
            return dbBalance;
            
        } catch (Exception e) {
            log.warn("캐시 조회 실패, DB에서 직접 조회: userId={}", userId, e);
            return null;  // 캐시 실패 시 DB 조회로 폴백
        }
    }
    
    /**
     * 결제 승인 처리 - Repeatable Read 적용
     * - 이 부분은 승인 과정의 데이터 일관성을 담당
     * - 여기서 Transaction Isolation의 반복 읽기 보장 확인 가능
     */
    @Transactional(isolation = Isolation.REPEATABLE_READ)
    public PaymentApproval approvePaymentWithConsistency(
            String userId, BigDecimal amount, String orderId) {
        
        // 사용자 정보 및 제한 사항 확인
        UserPointInfo userInfo = pointRepository.getUserPointInfo(userId);
        validateUserEligibility(userInfo);
        
        // 일일 사용 한도 확인 (첫 번째 읽기)
        BigDecimal dailyUsed = pointRepository.getDailyPointUsage(
            userId, LocalDate.now()
        );
        
        if (dailyUsed.add(amount).compareTo(userInfo.getDailyLimit()) > 0) {
            throw new DailyLimitExceededException("일일 포인트 사용 한도 초과");
        }
        
        // 비즈니스 규칙 검증 (시간이 걸리는 작업)
        validateMerchantEligibility(orderId);
        applySpecialPromotions(userId, amount);
        
        // 일일 사용량 재확인 (두 번째 읽기 - 같은 값이어야 함)
        BigDecimal recheckedDailyUsed = pointRepository.getDailyPointUsage(
            userId, LocalDate.now()
        );
        
        if (!dailyUsed.equals(recheckedDailyUsed)) {
            throw new ConcurrencyException("동시성 문제로 승인 실패");
        }
        
        // 승인 기록 생성
        PaymentApproval approval = PaymentApproval.builder()
            .userId(userId)
            .orderId(orderId)
            .amount(amount)
            .approvedAt(LocalDateTime.now())
            .status(ApprovalStatus.APPROVED)
            .build();
            
        return paymentApprovalRepository.save(approval);
    }
    
    /**
     * 포인트 차감 처리 - Serializable 적용
     * - 이 부분은 포인트 잔액의 완전한 격리를 담당
     * - 여기서 Transaction Isolation의 최고 수준 격리 확인 가능
     */
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public PointTransaction deductPointsWithIsolation(
            String userId, BigDecimal amount, String orderId) {
        
        // 포인트 계좌 잠금 및 조회
        PointAccount account = pointRepository
            .findByUserIdForUpdate(userId)
            .orElseThrow(() -> new PointAccountNotFoundException(userId));
        
        // 정확한 잔액 확인
        if (account.getBalance().compareTo(amount) < 0) {
            throw new InsufficientPointsException(
                String.format("포인트 부족: 잔액=%s, 요청=%s", 
                    account.getBalance(), amount)
            );
        }
        
        // 포인트 차감 실행
        account.deductPoints(amount);
        pointRepository.save(account);
        
        // 거래 기록 생성
        PointTransaction transaction = PointTransaction.builder()
            .userId(userId)
            .orderId(orderId)
            .amount(amount.negate())  // 차감은 음수로 기록
            .balanceAfter(account.getBalance())
            .transactionType(TransactionType.PAYMENT)
            .createdAt(LocalDateTime.now())
            .build();
            
        return pointTransactionRepository.save(transaction);
    }
    
    /**
     * 캐시 업데이트 및 이벤트 발행
     * - 이 부분은 실시간 데이터 동기화를 담당
     * - 여기서 Transaction Isolation의 이벤트 기반 아키텍처 적용 확인 가능
     */
    private void updateCacheAndPublishEvent(String userId, PointTransaction transaction) {
        // Redis 캐시 업데이트
        String cacheKey = "point:balance:" + userId;
        redisTemplate.opsForValue().set(cacheKey, transaction.getBalanceAfter(),
            Duration.ofMinutes(5));
        
        // 포인트 사용 이벤트 발행
        PointUsedEvent event = PointUsedEvent.builder()
            .userId(userId)
            .orderId(transaction.getOrderId())
            .amount(transaction.getAmount().abs())
            .balanceAfter(transaction.getBalanceAfter())
            .timestamp(transaction.getCreatedAt())
            .build();
        
        kafkaTemplate.send("point-events", userId, event);
        
        // 실시간 알림 발송 (WebSocket)
        sendRealtimeNotification(userId, event);
    }
    
    /**
     * 보상 트랜잭션 (실패 시 복구)
     * - 이 부분은 오류 상황에서의 데이터 복구를 담당
     * - 여기서 Transaction Isolation의 오류 처리 및 복구 확인 가능
     */
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public void executeCompensationTransaction(
            String userId, BigDecimal amount, String orderId) {
        
        try {
            // 포인트 복구 (차감된 포인트가 있다면)
            Optional<PointTransaction> deductedTransaction = 
                pointTransactionRepository.findByOrderIdAndType(
                    orderId, TransactionType.PAYMENT
                );
            
            if (deductedTransaction.isPresent()) {
                PointAccount account = pointRepository
                    .findByUserIdForUpdate(userId)
                    .orElseThrow(() -> new PointAccountNotFoundException(userId));
                
                account.addPoints(amount);
                pointRepository.save(account);
                
                // 보상 거래 기록
                PointTransaction compensation = PointTransaction.builder()
                    .userId(userId)
                    .orderId(orderId)
                    .amount(amount)  // 복구는 양수로 기록
                    .balanceAfter(account.getBalance())
                    .transactionType(TransactionType.COMPENSATION)
                    .relatedTransactionId(deductedTransaction.get().getId())
                    .createdAt(LocalDateTime.now())
                    .build();
                
                pointTransactionRepository.save(compensation);
                
                // 캐시 및 이벤트 업데이트
                updateCacheAndPublishEvent(userId, compensation);
            }
            
        } catch (Exception e) {
            log.error("보상 트랜잭션 실행 실패: userId={}, orderId={}", 
                userId, orderId, e);
            
            // 수동 처리를 위한 알림
            alertService.sendCriticalAlert(
                "포인트 보상 트랜잭션 실패", 
                String.format("수동 처리 필요: userId=%s, orderId=%s", userId, orderId)
            );
        }
    }
}

/**
 * 성능 모니터링 및 지표 수집
 * - 이 부분은 실시간 성능 지표 수집을 담당
 * - 여기서 Transaction Isolation의 성능 모니터링 확인 가능
 */
@Component
@Slf4j
public class PointServiceMonitor {
    
    @Autowired
    private MeterRegistry meterRegistry;
    
    @EventListener
    public void handlePointTransactionEvent(PointTransactionEvent event) {
        // 격리 수준별 성능 지표 수집
        String isolationLevel = event.getIsolationLevel();
        long duration = event.getDuration();
        
        Timer.Sample sample = Timer.start(meterRegistry);
        sample.stop(Timer.builder("point.transaction.duration")
            .tag("isolation", isolationLevel)
            .tag("type", event.getTransactionType())
            .register(meterRegistry));
        
        // 처리량 카운터
        meterRegistry.counter("point.transaction.count",
            "isolation", isolationLevel,
            "status", event.getStatus())
            .increment();
        
        // 임계값 초과 시 알림
        if (duration > getPerformanceThreshold(isolationLevel)) {
            log.warn("성능 임계값 초과: isolation={}, duration={}ms", 
                isolationLevel, duration);
                
            meterRegistry.counter("point.transaction.slow",
                "isolation", isolationLevel)
                .increment();
        }
    }
    
    private long getPerformanceThreshold(String isolationLevel) {
        switch (isolationLevel) {
            case "READ_COMMITTED": return 50;
            case "REPEATABLE_READ": return 200;
            case "SERIALIZABLE": return 500;
            default: return 100;
        }
    }
}

1
2
3
4
5
6
7

# 격리 수준 설정 예시
apiVersion: v1
kind: ConfigMap
metadata:
  name: db-isolation-config
data:
  isolation_level: "SERIALIZABLE" # 완전 격리 설정

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// npm i pg
const { Client } = require('pg');

async function reserveStock(client, sku, qty) {
  // 결제 핵심 로직: Serializable로 팬텀/쓰기왜곡 방지
  await client.query('BEGIN ISOLATION LEVEL SERIALIZABLE');
  const { rows } = await client.query(
    'SELECT available FROM inventory WHERE sku=$1 FOR UPDATE', [sku]
  );
  if (!rows.length || rows[0].available < qty) {
    await client.query('ROLLBACK'); return false;
  }
  await client.query('UPDATE inventory SET available=available-$1 WHERE sku=$2', [qty, sku]);
  await client.query('COMMIT'); return true;
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

import psycopg2
from contextlib import contextmanager
import logging

class InventoryManager:
    """
    재고 관리 클래스
    Repeatable Read 격리 수준을 사용하여 동시성 문제 해결
    """
    
    def __init__(self, connection_params):
        """
        데이터베이스 연결 매개변수 초기화
        """
        self.connection_params = connection_params
        self.logger = logging.getLogger(__name__)
    
    @contextmanager
    def get_transaction(self, isolation_level="REPEATABLE READ"):
        """
        트랜잭션 컨텍스트 매니저
        격리 수준을 설정하고 자동 커밋/롤백 처리
        """
        conn = None
        try:
            # 데이터베이스 연결 및 격리 수준 설정
            conn = psycopg2.connect(**self.connection_params)
            conn.set_isolation_level(
                psycopg2.extensions.ISOLATION_LEVEL_REPEATABLE_READ
            )
            
            # 트랜잭션 시작
            with conn.cursor() as cursor:
                cursor.execute(f"BEGIN TRANSACTION ISOLATION LEVEL {isolation_level};")
                yield cursor
                
            # 성공 시 커밋
            conn.commit()
            self.logger.info("트랜잭션이 성공적으로 커밋되었습니다.")
            
        except psycopg2.Error as e:
            # 오류 발생 시 롤백
            if conn:
                conn.rollback()
            self.logger.error(f"트랜잭션 오류: {e}")
            raise
            
        finally:
            # 연결 해제
            if conn:
                conn.close()
    
    def process_order(self, product_id, quantity, customer_id):
        """
        주문 처리 메인 로직
        격리 수준을 통한 재고 일관성 보장
        """
        try:
            with self.get_transaction() as cursor:
                # 1. 현재 재고 조회 (Repeatable Read로 일관성 보장)
                cursor.execute(
                    "SELECT stock_quantity, product_name, price FROM products WHERE id = %s",
                    (product_id,)
                )
                result = cursor.fetchone()
                
                if not result:
                    raise ValueError(f"상품 ID {product_id}를 찾을 수 없습니다.")
                
                current_stock, product_name, price = result
                self.logger.info(f"상품 '{product_name}' 현재 재고: {current_stock}")
                
                # 2. 재고 부족 검증
                if current_stock < quantity:
                    raise ValueError(
                        f"재고 부족: 요청 수량 {quantity}, 현재 재고 {current_stock}"
                    )
                
                # 3. 재고 차감 (동시성 제어를 위한 WHERE 조건 추가)
                cursor.execute("""
                    UPDATE products 
                    SET stock_quantity = stock_quantity - %s,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE id = %s AND stock_quantity >= %s
                    RETURNING stock_quantity
                """, (quantity, product_id, quantity))
                
                updated_result = cursor.fetchone()
                if not updated_result:
                    raise ValueError("다른 트랜잭션에 의해 재고가 변경되었습니다. 재시도하세요.")
                
                new_stock = updated_result[0]
                self.logger.info(f"재고 차감 완료: {current_stock} -> {new_stock}")
                
                # 4. 주문 정보 저장
                cursor.execute("""
                    INSERT INTO orders (customer_id, product_id, quantity, 
                                      unit_price, total_amount, order_status, created_at)
                    VALUES (%s, %s, %s, %s, %s, 'confirmed', CURRENT_TIMESTAMP)
                    RETURNING order_id
                """, (customer_id, product_id, quantity, price, price * quantity))
                
                order_id = cursor.fetchone()[0]
                self.logger.info(f"주문 생성 완료: Order ID {order_id}")
                
                # 5. 주문 상세 로그 기록
                cursor.execute("""
                    INSERT INTO order_logs (order_id, action, description, created_at)
                    VALUES (%s, 'order_created', %s, CURRENT_TIMESTAMP)
                """, (order_id, f"상품 {product_name} {quantity}개 주문 확정"))
                
                return {
                    'success': True,
                    'order_id': order_id,
                    'product_name': product_name,
                    'quantity': quantity,
                    'remaining_stock': new_stock,
                    'total_amount': price * quantity
                }
                
        except ValueError as e:
            # 비즈니스 로직 오류
            self.logger.warning(f"주문 처리 실패: {e}")
            return {'success': False, 'error': str(e)}
            
        except psycopg2.Error as e:
            # 데이터베이스 오류
            self.logger.error(f"데이터베이스 오류: {e}")
            return {'success': False, 'error': '시스템 오류가 발생했습니다.'}
    
    def get_inventory_status(self, product_id):
        """
        재고 상태 조회 (Read Committed 사용으로 최신 정보 제공)
        """
        try:
            with self.get_transaction("READ COMMITTED") as cursor:
                cursor.execute("""
                    SELECT product_name, stock_quantity, reserved_quantity,
                           (stock_quantity - reserved_quantity) as available_quantity
                    FROM products 
                    WHERE id = %s
                """, (product_id,))
                
                result = cursor.fetchone()
                if result:
                    return {
                        'product_name': result[0],
                        'total_stock': result[1],
                        'reserved_stock': result[2],
                        'available_stock': result[3]
                    }
                return None
                
        except Exception as e:
            self.logger.error(f"재고 조회 오류: {e}")
            return None

# 사용 예시
if __name__ == "__main__":
    # 데이터베이스 연결 설정
    db_params = {
        'host': 'localhost',
        'database': 'ecommerce',
        'user': 'postgres',
        'password': 'password'
    }
    
    # 재고 관리자 인스턴스 생성
    inventory_manager = InventoryManager(db_params)
    
    # 주문 처리 테스트
    result = inventory_manager.process_order(
        product_id=101,      # 상품 ID
        quantity=2,          # 주문 수량
        customer_id=12345    # 고객 ID
    )
    
    # 결과 출력
    if result['success']:
        print(f"주문 성공: {result}")
    else:
        print(f"주문 실패: {result['error']}")