ACID Properties

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

# pip install psycopg[binary]
import psycopg
conn1 = psycopg.connect("dbname=demo user=postgres")
conn2 = psycopg.connect("dbname=demo user=postgres")
c1 = conn1.cursor(); c2 = conn2.cursor()

# Tx1: 특정 행을 배타락으로 고정(UPDATE 의도) → 2PL에서 Growing phase
c1.execute("BEGIN")
c1.execute("SELECT * FROM accounts WHERE id=1 FOR UPDATE")
# Tx2: 같은 행을 잠그려 하면 블로킹(직렬가능성에 맞게 직렬화)
c2.execute("BEGIN")
try:
    c2.execute("UPDATE accounts SET balance=balance+10 WHERE id=1")  # 여기서 Tx1이 커밋/롤백 전까지 대기
except Exception as e:
    print("Tx2 blocked or error:", e)

# Tx1 커밋 → Shrinking phase에서 락 해제
conn1.commit()
# 이제 Tx2 진행 가능
conn2.commit()

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

import psycopg
# 두 세션 각각 REPEATABLE READ(스냅샷 고정)
connA = psycopg.connect("dbname=demo user=postgres", autocommit=False)
connB = psycopg.connect("dbname=demo user=postgres", autocommit=False)
connA.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
connB.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")

# TxA: 스냅샷 시점의 합계를 읽는다
curA = connA.cursor()
curA.execute("SELECT SUM(balance) FROM accounts")
print("TxA sum snapshot:", curA.fetchone()[0])

# TxB: 한 행 갱신 후 커밋(새 버전 생성, 기존 버전은 TxA에 계속 보임)
curB = connB.cursor()
curB.execute("UPDATE accounts SET balance=balance+100 WHERE id=2")
connB.commit()

# TxA: 같은 쿼리를 다시 실행해도 동일 스냅샷 → 동일 결과
curA.execute("SELECT SUM(balance) FROM accounts")
print("TxA sum still snapshot:", curA.fetchone()[0])
connA.commit()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# WAL 모드에서 커밋 = 로그에 선기록 → 장애 시 로그 재생으로 지속성 보강
import sqlite3, os

db = sqlite3.connect("demo.db")
db.execute("PRAGMA journal_mode=WAL")        # WAL 활성화
db.execute("PRAGMA synchronous=FULL")        # fsync 보장(지속성↑)
db.execute("BEGIN")
db.execute("CREATE TABLE IF NOT EXISTS t(k INTEGER PRIMARY KEY, v TEXT)")
db.execute("INSERT INTO t(v) VALUES ('wal-durable')")
db.commit()                                  # 먼저 WAL에 기록 후 안정화
db.close()

1
2
3
4
5
6
7
8

-- 참여자 DB에서
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
PREPARE TRANSACTION 'tx-100';  -- prepare(잠정 커밋)
-- 코디네이터가 모든 참여자 성공 확인 후
COMMIT PREPARED 'tx-100';      -- 최종 커밋
-- 실패 시
-- ROLLBACK PREPARED 'tx-100';

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

import json, os, shutil

DATA = "data.json"
SHADOW = "data.shadow.json"

def read_state(path=DATA):
    if not os.path.exists(path):
        return {"counter": 0}
    return json.load(open(path))

def shadow_commit(mutator):
    # 1) 원본 읽기
    state = read_state(DATA)
    # 2) 섀도 사본에서 변경
    new_state = mutator(state)
    json.dump(new_state, open(SHADOW, "w"))
    # 3) 커밋: 포인터(파일) 스위치 = 원자적 rename
    os.replace(SHADOW, DATA)  # 대부분의 OS에서 rename은 원자적
    # 실패 시 원본(DATA) 그대로 → 롤백 효과

shadow_commit(lambda s: {"counter": s["counter"] + 1})
print(read_state())

 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

// go run
// 의도: 짧은 간격으로 들어오는 커밋들을 모아 한 번의 fsync(=Commit)로 처리 → 처리량↑, 지연은 약간↑
package main

import (
	"context"
	"fmt"
	"time"
)

type commitReq struct {
	done chan struct{}
}

func main() {
	commitCh := make(chan commitReq, 1024)
	// 그룹 커밋 루프: 5ms마다 들어온 요청을 모아 한 번에 커밋
	go func() {
		ticker := time.NewTicker(5 * time.Millisecond)
		defer ticker.Stop()
		var batch []commitReq
		doCommit := func() {
			// 실제 DB 커밋/로그 fsync 호출이 들어갈 자리
			time.Sleep(200 * time.Microsecond) // fsync 비용 흉내
			for _, r := range batch {
				close(r.done)
			}
			batch = batch[:0]
		}
		for {
			select {
			case r := <-commitCh:
				batch = append(batch, r)
			case <-ticker.C:
				if len(batch) > 0 {
					doCommit()
				}
			}
		}
	}()

	// 클라이언트: 커밋 요청
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	req := commitReq{done: make(chan struct{})}
	select {
	case commitCh <- req:
	default:
		fmt.Println("commit queue full")
	}
	select {
	case <-req.done:
		fmt.Println("group-committed")
	case <-ctx.Done():
		fmt.Println("commit timeout")
	}
}

 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

# ACID와 Isolation이 반영된 주문 트랜잭션 예시
import psycopg2
from contextlib import closing

def place_order(conn, user_id, sku, qty, payment_token):
    with conn:
        with conn.cursor() as cur:
            # 1) 재고 행을 잠금하여 Lost Update/Phantom 방지 (Isolation 관련)
            cur.execute("SELECT stock FROM inventory WHERE sku = %s FOR UPDATE", (sku,))
            row = cur.fetchone()
            if not row or row[0] < qty:
                raise ValueError("재고 부족")

            # 2) 주문 생성 (Atomicity: 예외 시 전체 롤백)
            cur.execute("""
                INSERT INTO orders(user_id, status) VALUES (%s, %s) RETURNING id
            """, (user_id, "PENDING"))
            order_id = cur.fetchone()[0]

            # 3) 주문항목/재고차감
            cur.execute("""
                INSERT INTO order_items(order_id, sku, qty) VALUES (%s, %s, %s)
            """, (order_id, sku, qty))
            cur.execute("UPDATE inventory SET stock = stock - %s WHERE sku = %s", (qty, sku))

            # 4) 결제 승인 기록 (Consistency: 도메인 규칙 반영)
            cur.execute("""
                INSERT INTO payments(order_id, token, status) VALUES (%s, %s, %s)
            """, (order_id, payment_token, "AUTHORIZED"))

            # with conn: 블록 종료 시 커밋 -> WAL에 Commit 레코드 플러시(Durability)
    return order_id

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

// 격리수준 설정과 트랜잭션 경계
const { Pool } = require('pg');
const pool = new Pool({ /* conn info */ });

async function withTx(fn, isolation = 'READ COMMITTED') {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    await client.query(`SET TRANSACTION ISOLATION LEVEL ${isolation}`);
    const result = await fn(client);
    await client.query('COMMIT'); // WAL 플러시 -> Durability
    return result;
  } catch (e) {
    await client.query('ROLLBACK'); // Atomicity 보장
    throw e;
  } finally {
    client.release();
  }
}

  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

import psycopg2
from psycopg2 import sql
import logging

class BookOrderService:
    def __init__(self, db_connection):
        self.conn = db_connection
        
    def process_order(self, customer_id, book_id, quantity, payment_amount):
        """
        ACID 트랜잭션으로 도서 주문 처리
        - Atomicity: 모든 단계가 성공하거나 모두 롤백
        - Consistency: 비즈니스 규칙 준수 (재고 > 0, 결제 금액 정확)
        - Isolation: 동시 주문 간 간섭 방지
        - Durability: 커밋된 주문 정보 영구 저장
        """
        cursor = self.conn.cursor()
        
        try:
            # 트랜잭션 시작 - BEGIN 자동 실행
            cursor.execute("BEGIN;")
            
            # 1. 재고 확인 및 락 (Isolation 보장)
            cursor.execute(
                "SELECT stock_quantity, price FROM books WHERE book_id = %s FOR UPDATE;",
                (book_id,)
            )
            book_info = cursor.fetchone()
            
            if not book_info:
                raise ValueError("Book not found")
                
            stock_quantity, book_price = book_info
            total_price = book_price * quantity
            
            # 2. 비즈니스 규칙 검증 (Consistency 보장)
            if stock_quantity < quantity:
                raise ValueError("Insufficient stock")
                
            if payment_amount != total_price:
                raise ValueError("Payment amount mismatch")
            
            # 3. 재고 차감 (Atomicity의 일부)
            cursor.execute(
                "UPDATE books SET stock_quantity = stock_quantity - %s WHERE book_id = %s;",
                (quantity, book_id)
            )
            
            # 4. 결제 처리 (Atomicity의 일부)
            cursor.execute(
                "INSERT INTO payments (customer_id, amount, payment_date) VALUES (%s, %s, NOW());",
                (customer_id, payment_amount)
            )
            payment_id = cursor.lastrowid
            
            # 5. 주문 생성 (Atomicity의 일부)
            cursor.execute(
                "INSERT INTO orders (customer_id, book_id, quantity, total_price, payment_id, order_date) VALUES (%s, %s, %s, %s, %s, NOW());",
                (customer_id, book_id, quantity, total_price, payment_id)
            )
            order_id = cursor.lastrowid
            
            # 모든 작업 성공 시 커밋 (Durability 보장)
            cursor.execute("COMMIT;")
            
            logging.info(f"Order {order_id} processed successfully")
            return {"order_id": order_id, "status": "success"}
            
        except Exception as e:
            # 오류 발생 시 롤백 (Atomicity 보장)
            cursor.execute("ROLLBACK;")
            logging.error(f"Order processing failed: {str(e)}")
            return {"error": str(e), "status": "failed"}
            
        finally:
            cursor.close()

# 사용 예시
def main():
    # 데이터베이스 연결
    conn = psycopg2.connect(
        host="localhost",
        database="bookstore",
        user="postgres", 
        password="password"
    )
    
    order_service = BookOrderService(conn)
    
    # ACID 트랜잭션으로 주문 처리
    result = order_service.process_order(
        customer_id=123,
        book_id=456, 
        quantity=2,
        payment_amount=59.98
    )
    
    print(f"Order result: {result}")
    conn.close()

if __name__ == "__main__":
    main()

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

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# DB 연결 및 세션 생성
engine = create_engine('postgresql://user:pw@localhost/bankdb')
Session = sessionmaker(bind=engine)

def transfer_money(sender_id, receiver_id, amount):
    """은행 이체에서 ACID 트랜잭션 (원자성, 일관성, 격리성, 지속성) 보장 예시"""
    session = Session()
    try:
        # 출금
        sender = session.query(Account).get(sender_id)
        receiver = session.query(Account).get(receiver_id)
        if sender.balance < amount:
            raise Exception("잔액 부족")
        sender.balance -= amount
        receiver.balance += amount
        session.commit()  # 모든 로직이 성공하면 커밋(Durability 보장)
    except Exception as e:
        session.rollback()  # 에러 발생 시 전체 원자적 롤백
        print("이체 실패:", e)
    finally:
        session.close()

  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

import asyncio
import aioredis
import asyncpg
from typing import Dict, Any
import json
import logging
from dataclasses import dataclass

@dataclass
class OrderRequest:
    customer_id: int
    items: list
    payment_info: dict
    shipping_address: dict

class DistributedOrderProcessor:
    """
    분산 환경에서 ACID 트랜잭션을 구현한 주문 처리 시스템
    - Saga 패턴으로 분산 트랜잭션 관리
    - 각 서비스별로 로컬 ACID 보장
    - 전체적으로 최종 일관성 달성
    """
    
    def __init__(self, db_pools: Dict[str, Any], redis_pool):
        self.db_pools = db_pools  # 서비스별 DB 연결 풀
        self.redis = redis_pool   # 분산 세션 저장소
        
    async def process_order(self, order_request: OrderRequest) -> Dict[str, Any]:
        """
        분산 ACID 트랜잭션으로 주문 처리
        """
        saga_id = f"order_saga_{order_request.customer_id}_{int(time.time())}"
        
        try:
            # 1. Saga 시작 - 분산 트랜잭션 상태 관리
            await self._start_saga(saga_id, order_request)
            
            # 2. 재고 예약 (Inventory Service에서 로컬 ACID)
            reservation_result = await self._reserve_inventory(saga_id, order_request.items)
            if not reservation_result['success']:
                await self._compensate_saga(saga_id, ['inventory'])
                return {"status": "failed", "reason": "inventory_unavailable"}
            
            # 3. 결제 처리 (Payment Service에서 로컬 ACID)  
            payment_result = await self._process_payment(saga_id, order_request.payment_info)
            if not payment_result['success']:
                await self._compensate_saga(saga_id, ['inventory', 'payment'])
                return {"status": "failed", "reason": "payment_failed"}
            
            # 4. 주문 생성 (Order Service에서 로컬 ACID)
            order_result = await self._create_order(saga_id, order_request)
            if not order_result['success']:
                await self._compensate_saga(saga_id, ['inventory', 'payment', 'order'])
                return {"status": "failed", "reason": "order_creation_failed"}
            
            # 5. 배송 준비 (Shipping Service에서 로컬 ACID)
            shipping_result = await self._prepare_shipping(saga_id, order_request.shipping_address)
            
            # 6. Saga 완료 - 모든 서비스 커밋
            await self._complete_saga(saga_id)
            
            return {
                "status": "success", 
                "order_id": order_result['order_id'],
                "saga_id": saga_id
            }
            
        except Exception as e:
            # 전체 보상 트랜잭션 실행
            await self._compensate_saga(saga_id, ['inventory', 'payment', 'order', 'shipping'])
            return {"status": "error", "message": str(e)}
    
    async def _reserve_inventory(self, saga_id: str, items: list) -> Dict[str, Any]:
        """
        재고 서비스에서 ACID 트랜잭션으로 재고 예약
        """
        async with self.db_pools['inventory'].acquire() as conn:
            async with conn.transaction():  # PostgreSQL 트랜잭ション 시작
                try:
                    reserved_items = []
                    
                    for item in items:
                        # 재고 확인 및 락 (Isolation)
                        result = await conn.fetchrow(
                            "SELECT stock_quantity FROM inventory WHERE product_id = $1 FOR UPDATE",
                            item['product_id']
                        )
                        
                        if not result or result['stock_quantity'] < item['quantity']:
                            raise ValueError(f"Insufficient stock for product {item['product_id']}")
                        
                        # 재고 예약 (Atomicity)
                        await conn.execute(
                            "UPDATE inventory SET reserved_quantity = reserved_quantity + $1 WHERE product_id = $2",
                            item['quantity'], item['product_id']
                        )
                        
                        reserved_items.append(item)
                    
                    # Saga 상태 저장 (Durability)
                    await self.redis.hset(f"saga:{saga_id}", "inventory_reserved", json.dumps(reserved_items))
                    
                    return {"success": True, "reserved_items": reserved_items}
                    
                except Exception as e:
                    # 자동 롤백 (Atomicity)
                    logging.error(f"Inventory reservation failed: {e}")
                    return {"success": False, "error": str(e)}
    
    async def _process_payment(self, saga_id: str, payment_info: dict) -> Dict[str, Any]:
        """
        결제 서비스에서 ACID 트랜잭션으로 결제 처리
        """
        async with self.db_pools['payment'].acquire() as conn:
            async with conn.transaction():
                try:
                    # 결제 정보 검증 (Consistency)
                    if not self._validate_payment_info(payment_info):
                        raise ValueError("Invalid payment information")
                    
                    # 결제 처리 (Atomicity)
                    payment_result = await conn.fetchrow(
                        "INSERT INTO payments (customer_id, amount, payment_method, status, created_at) VALUES ($1, $2, $3, 'completed', NOW()) RETURNING payment_id",
                        payment_info['customer_id'], payment_info['amount'], payment_info['method']
                    )
                    
                    # Saga 상태 저장 (Durability)
                    await self.redis.hset(f"saga:{saga_id}", "payment_id", payment_result['payment_id'])
                    
                    return {"success": True, "payment_id": payment_result['payment_id']}
                    
                except Exception as e:
                    logging.error(f"Payment processing failed: {e}")
                    return {"success": False, "error": str(e)}
    
    async def _compensate_saga(self, saga_id: str, services_to_compensate: list):
        """
        Saga 패턴의 보상 트랜잭션 - 실패한 작업들을 되돌림
        """
        for service in reversed(services_to_compensate):  # 역순으로 보상
            try:
                if service == 'inventory':
                    await self._compensate_inventory(saga_id)
                elif service == 'payment':
                    await self._compensate_payment(saga_id)
                elif service == 'order':
                    await self._compensate_order(saga_id)
                elif service == 'shipping':
                    await self._compensate_shipping(saga_id)
                    
            except Exception as e:
                logging.error(f"Compensation failed for {service}: {e}")
    
    async def _compensate_inventory(self, saga_id: str):
        """재고 예약 취소"""
        reserved_items = json.loads(await self.redis.hget(f"saga:{saga_id}", "inventory_reserved") or "[]")
        
        async with self.db_pools['inventory'].acquire() as conn:
            async with conn.transaction():
                for item in reserved_items:
                    await conn.execute(
                        "UPDATE inventory SET reserved_quantity = reserved_quantity - $1 WHERE product_id = $2",
                        item['quantity'], item['product_id']
                    )

# 사용 예시
async def main():
    # 데이터베이스 연결 풀 설정
    db_pools = {
        'inventory': await asyncpg.create_pool("postgresql://user:pass@inventory-db:5432/inventory"),
        'payment': await asyncpg.create_pool("postgresql://user:pass@payment-db:5432/payment"),
        'order': await asyncpg.create_pool("postgresql://user:pass@order-db:5432/orders"),
        'shipping': await asyncpg.create_pool("postgresql://user:pass@shipping-db:5432/shipping"),
    }
    
    # Redis 연결
    redis_pool = await aioredis.create_redis_pool("redis://redis:6379")
    
    # 분산 주문 처리기 초기화
    processor = DistributedOrderProcessor(db_pools, redis_pool)
    
    # 주문 요청
    order_request = OrderRequest(
        customer_id=12345,
        items=[{"product_id": 101, "quantity": 2}, {"product_id": 102, "quantity": 1}],
        payment_info={"customer_id": 12345, "amount": 199.99, "method": "credit_card"},
        shipping_address={"street": "123 Main St", "city": "Seoul", "country": "KR"}
    )
    
    # 분산 ACID 트랜잭션으로 주문 처리
    result = await processor.process_order(order_request)
    print(f"Order processing result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

  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

import threading
import time
from contextlib import contextmanager
from typing import Optional

class BankAccount:
    """ACID Properties를 보장하는 은행 계좌 클래스"""
    
    def __init__(self, account_id: str, initial_balance: float = 0.0):
        self.account_id = account_id
        self._balance = initial_balance
        self._lock = threading.RLock()  # 계좌별 락 (고립성 보장)
        self._transaction_log = []  # 거래 로그 (지속성 보장)
    
    @property
    def balance(self) -> float:
        """현재 잔액 조회"""
        with self._lock:
            return self._balance
    
    def _log_transaction(self, operation: str, amount: float, 
                        timestamp: float, success: bool) -> None:
        """거래 로그 기록 (지속성 구현)"""
        log_entry = {
            'operation': operation,
            'amount': amount,
            'timestamp': timestamp,
            'balance_after': self._balance,
            'success': success
        }
        self._transaction_log.append(log_entry)
    
    def deposit(self, amount: float) -> bool:
        """입금 처리 (일관성 보장)"""
        if amount <= 0:
            return False  # 비즈니스 규칙 위반 방지
        
        with self._lock:  # 고립성 보장
            timestamp = time.time()
            try:
                # 원자성: 모든 연산이 성공해야 함
                old_balance = self._balance
                self._balance += amount
                
                # 일관성: 잔액이 음수가 될 수 없음 (비즈니스 규칙)
                if self._balance < 0:
                    self._balance = old_balance  # 롤백
                    self._log_transaction('DEPOSIT', amount, timestamp, False)
                    return False
                
                # 지속성: 성공적인 거래 로그 기록
                self._log_transaction('DEPOSIT', amount, timestamp, True)
                return True
                
            except Exception as e:
                # 원자성: 오류 시 롤백
                self._balance = old_balance
                self._log_transaction('DEPOSIT', amount, timestamp, False)
                return False
    
    def withdraw(self, amount: float) -> bool:
        """출금 처리 (모든 ACID 특성 적용)"""
        if amount <= 0:
            return False
        
        with self._lock:  # 고립성 보장
            timestamp = time.time()
            try:
                # 일관성: 잔액 부족 확인
                if self._balance < amount:
                    self._log_transaction('WITHDRAW', amount, timestamp, False)
                    return False
                
                # 원자성: 출금 실행
                old_balance = self._balance
                self._balance -= amount
                
                # 지속성: 성공적인 거래 로그 기록
                self._log_transaction('WITHDRAW', amount, timestamp, True)
                return True
                
            except Exception as e:
                # 원자성: 오류 시 롤백
                self._balance = old_balance
                self._log_transaction('WITHDRAW', amount, timestamp, False)
                return False

class TransactionManager:
    """트랜잭션 매니저 (원자성 구현)"""
    
    def __init__(self):
        self._active_transactions = {}
        self._global_lock = threading.Lock()
    
    @contextmanager
    def transaction(self, transaction_id: str):
        """트랜잭션 컨텍스트 매니저"""
        try:
            self._begin_transaction(transaction_id)
            yield self
            self._commit_transaction(transaction_id)
        except Exception as e:
            self._rollback_transaction(transaction_id)
            raise e
    
    def _begin_transaction(self, transaction_id: str) -> None:
        """트랜잭션 시작"""
        with self._global_lock:
            self._active_transactions[transaction_id] = {
                'start_time': time.time(),
                'operations': [],
                'locks_held': []
            }
    
    def _commit_transaction(self, transaction_id: str) -> None:
        """트랜잭션 커밋 (지속성 보장)"""
        with self._global_lock:
            if transaction_id in self._active_transactions:
                # 모든 변경사항을 영구적으로 적용
                transaction = self._active_transactions[transaction_id]
                # 로그에 커밋 기록
                print(f"Transaction {transaction_id} committed successfully")
                del self._active_transactions[transaction_id]
    
    def _rollback_transaction(self, transaction_id: str) -> None:
        """트랜잭션 롤백 (원자성 보장)"""
        with self._global_lock:
            if transaction_id in self._active_transactions:
                # 모든 변경사항을 원래 상태로 복구
                transaction = self._active_transactions[transaction_id]
                print(f"Transaction {transaction_id} rolled back")
                del self._active_transactions[transaction_id]
    
    def transfer(self, from_account: BankAccount, to_account: BankAccount, 
                amount: float) -> bool:
        """계좌 이체 (모든 ACID 특성 적용)"""
        transaction_id = f"transfer_{time.time()}"
        
        try:
            with self.transaction(transaction_id):
                # 원자성: 출금과 입금이 모두 성공해야 함
                if not from_account.withdraw(amount):
                    raise Exception("출금 실패")
                
                if not to_account.deposit(amount):
                    # 롤백: 출금을 취소해야 함
                    from_account.deposit(amount)  # 보상 트랜잭션
                    raise Exception("입금 실패")
                
                return True
                
        except Exception as e:
            print(f"이체 실패: {e}")
            return False

# 사용 예시
def demonstrate_acid_properties():
    """ACID Properties 데모"""
    
    # 계좌 생성
    account_a = BankAccount("ACC_001", 1000.0)
    account_b = BankAccount("ACC_002", 500.0)
    
    # 트랜잭션 매니저 생성
    tm = TransactionManager()
    
    print("=== 초기 상태 ===")
    print(f"계좌 A 잔액: {account_a.balance}")
    print(f"계좌 B 잔액: {account_b.balance}")
    print(f"총 잔액: {account_a.balance + account_b.balance}")
    
    print("\n=== 정상 이체 (300원) ===")
    success = tm.transfer(account_a, account_b, 300.0)
    print(f"이체 결과: {'성공' if success else '실패'}")
    print(f"계좌 A 잔액: {account_a.balance}")
    print(f"계좌 B 잔액: {account_b.balance}")
    print(f"총 잔액: {account_a.balance + account_b.balance}")
    
    print("\n=== 잔액 부족 이체 시도 (2000원) ===")
    success = tm.transfer(account_a, account_b, 2000.0)
    print(f"이체 결과: {'성공' if success else '실패'}")
    print(f"계좌 A 잔액: {account_a.balance}")
    print(f"계좌 B 잔액: {account_b.balance}")
    print(f"총 잔액: {account_a.balance + account_b.balance}")

if __name__ == "__main__":
    demonstrate_acid_properties()