TinyKV 分布式事务实现与复习

账户 A: 扣减 100 元
账户 B: 增加 100 元

┌─────────────────────────────────────────────────────────────┐
│                        Client SDK                           │
│  (获取 TSO, 构造 PrewriteRequest/CommitRequest, 重试逻辑)    │
└───────────────────────────┬─────────────────────────────────┘
                            │ gRPC
┌───────────────────────────▼─────────────────────────────────┐
│                    TinyKV Server (server.go)                 │
│  KvGet | KvPrewrite | KvCommit | KvScan                     │
│  KvCheckTxnStatus | KvBatchRollback | KvResolveLock         │
└───────────────────────────┬─────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────┐
│                  MVCC Layer (transaction.go)                 │
│  MvccTxn: GetLock/PutLock | GetValue/PutValue               │
│           CurrentWrite | MostRecentWrite                    │
└───────────────────────────┬─────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────┐
│              RaftStore (Multi-Raft KV Storage)               │
│  三个列族: CfLock | CfWrite | CfDefault                     │
│  底层: badger (LSM-Tree)                                     │
└─────────────────────────────────────────────────────────────┘

事务执行阶段: 修改 buffer pool + 写 undo log + 写 redo log buffer
提交阶段:
  1. 写 binlog（如果开启）
  2. 写 redo log（prepare 状态）
  3. 写 binlog（commit 标记）
  4. 写 redo log（commit 状态）

约束: 值班医生数 >= 1
T1: 查询在岗医生 = 2，决定让 A 下班，修改 A.on_call = false
T2: 查询在岗医生 = 2，决定让 B 下班，修改 B.on_call = false
结果: A 和 B 都下班，值班医生 = 0，违反约束

-- 事务 T1
SELECT * FROM orders WHERE amount > 100 FOR UPDATE;
-- 对查询范围加 Next-Key Lock（临键锁 = 行锁 + 间隙锁）
-- 其他事务无法在此范围内插入新行

-- 事务 T2（被阻塞）
INSERT INTO orders (amount) VALUES (150);  -- 阻塞！

当前版本（row）:  trx_id=100, name="Alice", roll_ptr ──►
                                                         │
undo log v1:  trx_id=80, name="Bob", roll_ptr ──────────►
                                                         │
undo log v2:  trx_id=50, name="Charlie", roll_ptr ──────► NULL

type ReadView struct {
    m_ids          []uint64  // 创建时活跃的事务 ID 列表
    m_low_limit_id uint64    // 创建时最大事务 ID + 1（>=此值的事务不可见）
    m_up_limit_id  uint64    // m_ids 中最小的事务 ID（<此值的事务可见）
    m_creator_trx_id uint64  // 创建此 Read View 的事务 ID
}

1. trx_id == m_creator_trx_id  → 可见（自己修改的，当然能读）
2. trx_id < m_up_limit_id      → 可见（创建 Read View 前已提交）
3. trx_id >= m_low_limit_id    → 不可见（创建 Read View 后才开始的事务）
4. m_up_limit_id <= trx_id < m_low_limit_id:
   - trx_id 在 m_ids 中 → 不可见（创建时仍活跃，未提交）
   - trx_id 不在 m_ids 中 → 可见（已提交）

冲突率高（写密集型）  → 悲观锁（2PL）减少无效重试
冲突率低（读密集型）  → 乐观锁（OCC/MVCC）提高并发
长事务              → 避免持锁太久，考虑乐观锁
短事务/点查询        → 悲观锁开销可接受

Coordinator                    Participant A     Participant B
     │                               │                │
     │─── Prepare ──────────────────►│                │
     │─── Prepare ────────────────────────────────────►│
     │                               │                │
     │      执行事务操作，写 undo log + redo log，加锁    │
     │                               │                │
     │◄── Vote Yes/No ───────────────│                │
     │◄── Vote Yes/No ────────────────────────────────│

所有 Yes:  Coordinator ─── Commit ───► 所有参与者 ─── Ack ────► 协调者
有 No:     Coordinator ─── Abort ────► 所有参与者 ──No Ack ────► 协调者

情景：
1. 协调者向所有参与者发送 Commit
2. 参与者 A 收到并提交，协调者此时宕机
3. 参与者 B 没收到 Commit，进入无限等待

结果：A 已提交，B 处于未知状态 → 数据不一致

情景：
1. 协调者发送 Commit 后网络分区
2. 参与者 A 收到 Commit 并提交
3. 参与者 B 超时，不知道是 Commit 还是 Abort

分布式事务（跨 Region）
        │
        ▼
    2PC 协调（Percolator）
    │           │
    ▼           ▼
Region A     Region B
(Raft Group A) (Raft Group B)
    │           │
    ▼           ▼
  持久化       持久化
（Raft 保证每个 Region 内的强一致性）

-- 应用程序通过 XA 接口控制多个数据库
XA START 'xid';
  UPDATE accounts SET balance = balance - 100 WHERE id = 1;
XA END 'xid';
XA PREPARE 'xid';    -- Prepare 阶段

XA COMMIT 'xid';     -- 或 XA ROLLBACK 'xid'

Try: 冻结库存 1 件，冻结余额 100 元
Confirm: 扣减库存，扣减余额
Cancel: 解冻库存，解冻余额

正向流程: T1 → T2 → T3 → ... → Tn
补偿流程: C1 ← C2 ← C3 ← ... ← Cn

1. 本地事务：业务操作 + 写消息到本地消息表（原子操作）
2. 异步服务：扫描本地消息表，发送到消息队列
3. 消费者：消费消息，执行目标操作，幂等处理
4. 确认机制：消费成功后删除/标记消息

CfDefault:
Key                          Value
(Alice, startTS=100)    ─►   "Bob"       ← 事务 100 写入的值
(Alice, startTS=80)     ─►   "Alice"     ← 事务 80 写入的值
(Bob, startTS=95)       ─►   "100"       ← 事务 95 写入的值

message Lock {
    LockType   lock_type   // PUT / DELETE / ROLLBACK / LOCK
    bytes      primary     // Primary Key（用于崩溃恢复时查找事务状态）
    uint64     ts          // startTS（用于区分哪个事务的锁）
    uint64     ttl         // TTL，超时后可强制回滚
    uint64     txn_size    // 事务大小（决定 TTL）
}

CfLock:
Key      Value
Alice ─► Lock{type=PUT, primary="Alice", ts=100, ttl=3000}
Bob   ─► Lock{type=PUT, primary="Alice", ts=100, ttl=3000}

message Write {
    WriteKind kind           // PUT / DELETE / ROLLBACK
    uint64    start_ts       // 对应的 startTS（指向 CfDefault 中的值）
}

CfWrite:
Key                          Value
(Alice, commitTS=105)   ─►   Write{kind=PUT, startTS=100}
(Alice, commitTS=85)    ─►   Write{kind=PUT, startTS=80}
(Bob, commitTS=98)       ─►   Write{kind=PUT, startTS=95}

时间线（数字代表 TS）:

事务 80（已提交）: startTS=80, commitTS=85
事务 100（进行中）: startTS=100, commitTS 待定

CfDefault:
  Alice: TS=80 → "Alice"(旧值),  TS=100 → "Bob"(新值,尚未提交)

CfLock:
  Alice: Lock{ts=100, primary="Alice", ...}  ← Prewrite 写入，Commit 后删除

CfWrite:
  Alice: TS=85 → {PUT, startTS=80}  ← 事务80 Commit 后写入

读事务（startTS=90）读取 Alice:
  1. 查 CfLock[Alice]：有锁 ts=100 > 90，忽略（锁的事务在我的快照之后）
  2. 查 CfWrite[Alice]: 找 commitTS <= 90 的最新记录 → commitTS=85 → startTS=80
  3. 查 CfDefault[Alice, TS=80] → "Alice"  ✓

读事务（startTS=110）读取 Alice:
  1. 查 CfLock[Alice]：有锁 ts=100 <= 110，需要等待或清理
  2. 等待或通过 CheckTxnStatus 判断是否可清理

事务状态判断（通过 PrimaryKey）:
  - CfWrite[PrimaryKey] 有 WriteKind_Put 记录 → 事务已提交
  - CfWrite[PrimaryKey] 有 WriteKind_Rollback 记录 → 事务已回滚
  - CfLock[PrimaryKey] 存在 → 事务仍在进行（根据 TTL 决定是否清理）
  - CfLock[PrimaryKey] 不存在 且 CfWrite 无记录 → 事务已回滚（锁被清理）

CfLock[Alice] 存在（Primary），CfLock[Bob] 存在（Secondary）
→ 检查 CfWrite[Alice]：无记录
→ 检查 Alice.lock 的 TTL：如果超时 → 清理所有锁，视为回滚
→ 如果未超时 → 等待，原事务可能还在执行

CfWrite[Alice] 有 Write{PUT, startTS=100}（Primary 已提交）
CfLock[Bob] 存在（Secondary 未提交）
→ 其他事务遇到 Bob 的锁 → 检查 Primary Key → 已提交
→ 替原事务提交 Bob：DeleteLock[Bob] + PutWrite[Bob, TS=105]

客户端发起 Prewrite(mutations, primaryKey, startTS, ttl):

FOR each mutation in mutations:
    // 步骤1: 检查写冲突
    write, commitTS = MostRecentWrite(mutation.key)
    IF write != nil AND commitTS >= startTS:
        RETURN WriteConflict{commitTS, startTS, key}

    // 步骤2: 检查锁冲突
    lock = GetLock(mutation.key)
    IF lock != nil:
        RETURN KeyError{locked: lock}  // 提示客户端等待或清理

    // 步骤3: 写入数据（值）
    PutValue(mutation.key, mutation.value)

    // 步骤4: 写入锁
    PutLock(mutation.key, Lock{
        kind:    mutation.kind,    // PUT / DELETE
        primary: primaryKey,
        ts:      startTS,
        ttl:     ttl,
    })

// 批量提交到存储层（原子性写入）
storage.Write(txn.Writes())

例：
T1 startTS=100, T2 startTS=90
T2 先提交，commitTS=95
T1 尝试 Prewrite：MostRecentWrite 返回 commitTS=95 >= 100? 不，95 < 100
T1 没有冲突，继续

另一种情况：
T1 startTS=100, T2 startTS=90
T2 先提交，commitTS=105（发生在 T1 开始后！）
T1 尝试 Prewrite：MostRecentWrite 返回 commitTS=105 >= 100？是的
T1 返回 WriteConflict，客户端重试

客户端发起 Commit(keys, startTS, commitTS):

FOR each key in keys:
    // 步骤1: 验证锁还在（防止 Prewrite 被回滚）
    lock = GetLock(key)
    IF lock == nil OR lock.ts != startTS:
        RETURN retryable error  // 锁已被清理，事务已回滚

    // 步骤2: 写提交记录
    PutWrite(key, commitTS, Write{
        kind:     lock.kind,   // PUT / DELETE
        start_ts: startTS,
    })

    // 步骤3: 删除锁（释放）
    DeleteLock(key)

// 批量提交
storage.Write(txn.Writes())

客户端发起 Get(key, version=startTS):

// 步骤1: 检查锁冲突
lock = GetLock(key)
IF lock != nil AND lock.ts <= startTS:
    // 有事务（startTS <= version）正在修改这个 key
    // 需要等待或协助清理
    RETURN KeyError{locked: lock}

// 步骤2: 在 Write CF 中找最新可见提交
Write = GetValue(key)  // 内部实现：Seek CfWrite[key, startTS]

IF write == nil:
    RETURN nil  // key 不存在（在此快照下）

RETURN value  // 通过 write.startTS 从 CfDefault 读取实际值

Rollback(key, startTS):

// 写 Rollback 标记（重要：即使 CfDefault 无数据也要写）
PutWrite(key, startTS, Write{kind: ROLLBACK, start_ts: startTS})

// 删除 CfDefault 中的临时数据
DeleteValue(key)

// 删除锁（如果锁存在）
lock = GetLock(key)
IF lock != nil AND lock.ts == startTS:
    DeleteLock(key)

TTL = base_ttl + size_factor * txn_size

CheckTxnStatus(primaryKey, lockTS, currentTS):

// 检查事务是否已提交
write, commitTS = CurrentWrite(primaryKey)  // 找匹配 startTS=lockTS 的 write
IF write != nil AND write.kind != ROLLBACK:
    RETURN {action: NoAction, commitTs: commitTS}  // 已提交

// 检查是否已回滚
IF write != nil AND write.kind == ROLLBACK:
    RETURN {action: NoAction, commitTs: 0}  // 已回滚

// 检查锁是否存在
lock = GetLock(primaryKey)
IF lock == nil:
    // 锁不存在且无 Write 记录 → 写 Rollback 标记
    PutWrite(primaryKey, lockTS, Rollback)
    RETURN {action: LockNotExistRollback, commitTs: 0}

// 检查 TTL
IF physicalTime(currentTS) >= physicalTime(lockTS) + lock.ttl:
    // TTL 超时，强制回滚
    Rollback(primaryKey, lockTS)
    RETURN {action: TTLExpireRollback, commitTs: 0}

// TTL 未超时，等待
RETURN {action: NoAction, commitTs: 0, lockTtl: lock.ttl}

ResolveLock(startTS, commitTS):
  // 扫描所有 lockTS == startTS 的锁
  locks = AllLocksForTxn(startTS)

  FOR each lock in locks:
    IF commitTS == 0:
        // 回滚
        DeleteValue(lock.key)
        DeleteLock(lock.key)
        PutWrite(lock.key, startTS, Rollback)
    ELSE:
        // 提交
        DeleteLock(lock.key)
        PutWrite(lock.key, commitTS, Write{PUT, startTS})

// kv/transaction/mvcc/transaction.go
func EncodeKey(key []byte, ts uint64) []byte {
    encodedKey := codec.EncodeBytes(key)
    newKey := append(encodedKey, make([]byte, 8)...)
    binary.BigEndian.PutUint64(newKey[len(encodedKey):], ^ts)  // XOR 取反
    return newKey
}

正常存储（升序）: TS=100 < TS=200 < TS=300
倒序存储（降序）: ~TS=100 > ~TS=200 > ~TS=300
               (对应字节值: 大 > 中 > 小)

CfWrite（逻辑上按 key ASC, ts DESC 排序）:
  Alice/TS=~300 (commitTS=300)  ← 排最前
  Alice/TS=~200 (commitTS=200)
  Alice/TS=~100 (commitTS=100)

Seek(Alice, startTS=250):
  → 找到第一个 key=Alice 且 ~TS <= ~250 (即 TS >= 250) 的记录
  → 因为倒序，"第一个"就是 commitTS 最大且 <= 250 的记录
  → 找到 Alice/TS=~200 (commitTS=200) ✓

func DecodeUserKey(key []byte) []byte {
    userKey, _ := codec.DecodeBytes(key)
    return userKey
}

func decodeTimestamp(key []byte) uint64 {
    left, _ := codec.DecodeBytes(key)  // 跳过 user key 部分
    return ^binary.BigEndian.Uint64(key[len(left):])  // XOR 还原
}

CfDefault:
  [EncodeKey("Alice", 100)] → "Bob"
  [EncodeKey("Alice", 80)]  → "Alice"

CfLock:
  ["Alice"] → Lock{ts=100, primary="Alice", ...}

CfWrite:
  [EncodeKey("Alice", ~105)] → Write{kind=PUT, startTS=100}
  [EncodeKey("Alice", ~85)]  → Write{kind=PUT, startTS=80}

// kv/transaction/mvcc/transaction.go
type MvccTxn struct {
    StartTS uint64              // 事务的快照版本（startTS）
    Reader  storage.StorageReader  // 存储读接口（不可变）
    writes  []storage.Modify    // 写操作缓冲（Commit 时批量提交）
}

func (txn *MvccTxn) GetLock(key []byte) (*Lock, error) {
    // 直接读 CfLock，无需时间戳编码
    value, err := txn.Reader.GetCF(engine_util.CfLock, key)
    if err != nil || value == nil {
        return nil, err
    }
    lock, err := ParseLock(value)  // 反序列化 Lock 结构体
    return lock, err
}

func (txn *MvccTxn) PutLock(key []byte, lock *Lock) {
    txn.writes = append(txn.writes, storage.Modify{
        Data: storage.Put{
            Key:   key,
            Value: lock.ToBytes(),  // 序列化
            Cf:    engine_util.CfLock,
        },
    })
}

func (txn *MvccTxn) GetValue(key []byte) ([]byte, error) {
    // 在 CfWrite 中找 commitTS <= startTS 的最新 Write
    iter := txn.Reader.IterCF(engine_util.CfWrite)
    defer iter.Close()

    // Seek 到 EncodeKey(key, startTS) 处
    // 因为时间戳倒序，第一条 >= startTS 的就是我们想要的
    iter.Seek(EncodeKey(key, txn.StartTS))

    if !iter.Valid() {
        return nil, nil
    }

    item := iter.Item()
    userKey := DecodeUserKey(item.Key())

    // 验证是同一个 user key（防止跨 key）
    if !bytes.Equal(userKey, key) {
        return nil, nil
    }

    // 反序列化 Write 记录
    value, err := item.Value()
    write, err := ParseWrite(value)

    // 如果是 DELETE 或 ROLLBACK，说明此版本不是有效值
    if write.Kind != WriteKindPut {
        return nil, nil
    }

    // 通过 write.StartTS 从 CfDefault 取实际值
    return txn.Reader.GetCF(engine_util.CfDefault, EncodeKey(key, write.StartTs))
}

func (txn *MvccTxn) CurrentWrite(key []byte) (*Write, uint64, error) {
    // 从最大时间戳向小扫描，找匹配 startTS 的 Write
    iter := txn.Reader.IterCF(engine_util.CfWrite)
    defer iter.Close()

    // 从 TsMax 开始扫描（相当于最大 commitTS）
    for iter.Seek(EncodeKey(key, TsMax)); iter.Valid(); iter.Next() {
        item := iter.Item()
        userKey := DecodeUserKey(item.Key())

        if !bytes.Equal(userKey, key) {
            break  // 已扫过此 key 的所有版本
        }

        value, _ := item.Value()
        write, _ := ParseWrite(value)

        if write.StartTs == txn.StartTS {
            commitTS := decodeTimestamp(item.Key())
            return write, commitTS, nil
        }
    }
    return nil, 0, nil
}

func (txn *MvccTxn) MostRecentWrite(key []byte) (*Write, uint64, error) {
    iter := txn.Reader.IterCF(engine_util.CfWrite)
    defer iter.Close()

    iter.Seek(EncodeKey(key, TsMax))

    if !iter.Valid() {
        return nil, 0, nil
    }

    item := iter.Item()
    userKey := DecodeUserKey(item.Key())
    if !bytes.Equal(userKey, key) {
        return nil, 0, nil
    }

    value, _ := item.Value()
    write, _ := ParseWrite(value)
    commitTS := decodeTimestamp(item.Key())
    return write, commitTS, nil
}

// kv/transaction/mvcc/scanner.go
type Scanner struct {
    iter engine_util.DBIterator  // CfWrite 的迭代器
    txn  *MvccTxn
    key  []byte  // 上一次返回的 user key（用于跳过同 key 的旧版本）
}

func NewScanner(startKey []byte, txn *MvccTxn) *Scanner {
    scanner := &Scanner{
        iter: txn.Reader.IterCF(engine_util.CfWrite),
        txn:  txn,
    }
    // Seek 到 startKey 的最新版本
    scanner.iter.Seek(EncodeKey(startKey, txn.StartTS))
    return scanner
}

func (scan *Scanner) Next() ([]byte, []byte, error) {
    for {
        if !scan.iter.Valid() {
            return nil, nil, nil  // 已遍历完
        }

        item := scan.iter.Item()
        currentKey := DecodeUserKey(item.Key())

        // 跳过与上次相同的 user key（同一 key 的旧版本）
        if bytes.Equal(currentKey, scan.key) {
            scan.iter.Next()
            continue
        }

        // 检查此版本是否在快照范围内（commitTS <= startTS）
        commitTS := decodeTimestamp(item.Key())
        if commitTS > scan.txn.StartTS {
            scan.iter.Next()
            continue
        }

        scan.key = currentKey

        // 获取值（内部处理 DELETE/ROLLBACK 返回 nil）
        value, err := scan.txn.GetValue(currentKey)
        scan.iter.Next()

        if value == nil {
            continue  // 此 key 已被删除或不可见，继续扫下一个
        }

        return currentKey, value, err
    }
}

// kv/server/server.go
func (server *Server) KvGet(_ context.Context, req *kvrpcpb.GetRequest) (*kvrpcpb.GetResponse, error) {
    reader, _ := server.storage.Reader(req.Context)
    defer reader.Close()

    txn := mvcc.NewMvccTxn(reader, req.Version)

    // 检查锁冲突
    lock, _ := txn.GetLock(req.Key)
    if lock != nil && lock.Ts <= req.Version {
        // 有事务（startTS <= version）持有此 key 的锁，需要等待
        return &kvrpcpb.GetResponse{
            Error: &kvrpcpb.KeyError{
                Locked: lock.Info(req.Key),
            },
        }, nil
    }

    // 读取值
    value, _ := txn.GetValue(req.Key)
    if value == nil {
        return &kvrpcpb.GetResponse{NotFound: true}, nil
    }
    return &kvrpcpb.GetResponse{Value: value}, nil
}

func (server *Server) KvPrewrite(_ context.Context, req *kvrpcpb.PrewriteRequest) (*kvrpcpb.PrewriteResponse, error) {
    reader, _ := server.storage.Reader(req.Context)
    defer reader.Close()

    txn := mvcc.NewMvccTxn(reader, req.StartVersion)
    var keyErrors []*kvrpcpb.KeyError

    for _, mutation := range req.Mutations {
        // 1. 写冲突检测
        write, commitTS, _ := txn.MostRecentWrite(mutation.Key)
        if write != nil && commitTS >= req.StartVersion {
            keyErrors = append(keyErrors, &kvrpcpb.KeyError{
                Conflict: &kvrpcpb.WriteConflict{
                    StartTs:    req.StartVersion,
                    ConflictTs: commitTS,
                    Key:        mutation.Key,
                    Primary:    req.PrimaryLock,
                },
            })
            continue
        }

        // 2. 锁冲突检测
        lock, _ := txn.GetLock(mutation.Key)
        if lock != nil {
            keyErrors = append(keyErrors, &kvrpcpb.KeyError{
                Locked: lock.Info(mutation.Key),
            })
            continue
        }

        // 3. 写入数据和锁
        switch mutation.Op {
        case kvrpcpb.Op_Put:
            txn.PutValue(mutation.Key, mutation.Value)
            txn.PutLock(mutation.Key, &mvcc.Lock{
                Primary: req.PrimaryLock,
                Ts:      req.StartVersion,
                Ttl:     req.LockTtl,
                Kind:    mvcc.WriteKindPut,
            })
        case kvrpcpb.Op_Del:
            txn.DeleteValue(mutation.Key)
            txn.PutLock(mutation.Key, &mvcc.Lock{
                Primary: req.PrimaryLock,
                Ts:      req.StartVersion,
                Ttl:     req.LockTtl,
                Kind:    mvcc.WriteKindDelete,
            })
        }
    }

    if len(keyErrors) > 0 {
        return &kvrpcpb.PrewriteResponse{Errors: keyErrors}, nil
    }

    // 批量原子提交
    server.storage.Write(req.Context, txn.Writes())
    return &kvrpcpb.PrewriteResponse{}, nil
}

func (server *Server) KvCommit(_ context.Context, req *kvrpcpb.CommitRequest) (*kvrpcpb.CommitResponse, error) {
    reader, _ := server.storage.Reader(req.Context)
    defer reader.Close()

    txn := mvcc.NewMvccTxn(reader, req.StartVersion)

    // Latch 防止并发修改同一 key
    server.Latches.WaitForLatches(req.Keys)
    defer server.Latches.ReleaseLatches(req.Keys)

    for _, key := range req.Keys {
        // 检查幂等性：是否已提交
        write, _, _ := txn.CurrentWrite(key)
        if write != nil {
            if write.Kind == mvcc.WriteKindRollback {
                // 已回滚，不能再提交
                return &kvrpcpb.CommitResponse{
                    Error: &kvrpcpb.KeyError{Retryable: "already rolled back"},
                }, nil
            }
            // 已提交（幂等），继续下一个 key
            continue
        }

        // 验证锁
        lock, _ := txn.GetLock(key)
        if lock == nil || lock.Ts != req.StartVersion {
            // 锁不存在或 ts 不匹配（可能被其他事务抢走了）
            return &kvrpcpb.CommitResponse{
                Error: &kvrpcpb.KeyError{Retryable: "lock not found"},
            }, nil
        }

        // 写提交记录
        txn.PutWrite(key, req.CommitVersion, &mvcc.Write{
            StartTs: req.StartVersion,
            Kind:    lock.Kind,
        })

        // 删除锁
        txn.DeleteLock(key)
    }

    server.storage.Write(req.Context, txn.Writes())
    return &kvrpcpb.CommitResponse{}, nil
}

func (server *Server) KvScan(_ context.Context, req *kvrpcpb.ScanRequest) (*kvrpcpb.ScanResponse, error) {
    reader, _ := server.storage.Reader(req.Context)
    defer reader.Close()

    txn := mvcc.NewMvccTxn(reader, req.Version)
    scanner := mvcc.NewScanner(req.StartKey, txn)
    defer scanner.Close()

    var pairs []*kvrpcpb.KvPair

    for i := uint32(0); i < req.Limit; i++ {
        key, value, err := scanner.Next()
        if key == nil {
            break
        }

        if err != nil {
            // 处理 lock 错误：记录到 pairs，而非终止
            if keyErr, ok := err.(*mvcc.KeyError); ok {
                pairs = append(pairs, &kvrpcpb.KvPair{
                    Error: keyErr.Err,
                })
                continue
            }
            return nil, err
        }

        pairs = append(pairs, &kvrpcpb.KvPair{
            Key:   key,
            Value: value,
        })
    }

    return &kvrpcpb.ScanResponse{Pairs: pairs}, nil
}

func (server *Server) KvCheckTxnStatus(_ context.Context, req *kvrpcpb.CheckTxnStatusRequest) (
    *kvrpcpb.CheckTxnStatusResponse, error) {

    reader, _ := server.storage.Reader(req.Context)
    defer reader.Close()

    txn := mvcc.NewMvccTxn(reader, req.LockTs)

    // 1. 检查是否已提交
    write, commitTS, _ := txn.CurrentWrite(req.PrimaryKey)
    if write != nil && write.Kind != mvcc.WriteKindRollback {
        return &kvrpcpb.CheckTxnStatusResponse{
            CommitVersion: commitTS,
            Action:        kvrpcpb.Action_NoAction,
        }, nil
    }

    // 2. 检查是否已回滚
    if write != nil && write.Kind == mvcc.WriteKindRollback {
        return &kvrpcpb.CheckTxnStatusResponse{
            Action: kvrpcpb.Action_NoAction,
        }, nil
    }

    // 3. 检查锁是否存在
    lock, _ := txn.GetLock(req.PrimaryKey)
    if lock == nil {
        // 锁不存在，写 Rollback 标记
        txn.DeleteValue(req.PrimaryKey)
        txn.PutWrite(req.PrimaryKey, req.LockTs, &mvcc.Write{
            StartTs: req.LockTs,
            Kind:    mvcc.WriteKindRollback,
        })
        server.storage.Write(req.Context, txn.Writes())
        return &kvrpcpb.CheckTxnStatusResponse{
            Action: kvrpcpb.Action_LockNotExistRollback,
        }, nil
    }

    // 4. 检查 TTL
    currentPhysical := mvcc.PhysicalTime(req.CurrentTs)
    lockPhysical := mvcc.PhysicalTime(req.LockTs)
    if currentPhysical >= lockPhysical+lock.Ttl {
        // TTL 超时，强制回滚
        txn.DeleteValue(req.PrimaryKey)
        txn.DeleteLock(req.PrimaryKey)
        txn.PutWrite(req.PrimaryKey, req.LockTs, &mvcc.Write{
            StartTs: req.LockTs,
            Kind:    mvcc.WriteKindRollback,
        })
        server.storage.Write(req.Context, txn.Writes())
        return &kvrpcpb.CheckTxnStatusResponse{
            Action: kvrpcpb.Action_TTLExpireRollback,
        }, nil
    }

    // 5. TTL 未超时，返回锁信息
    return &kvrpcpb.CheckTxnStatusResponse{
        LockTtl: lock.Ttl,
        Action:  kvrpcpb.Action_NoAction,
    }, nil
}

func (server *Server) KvBatchRollback(_ context.Context, req *kvrpcpb.BatchRollbackRequest) (
    *kvrpcpb.BatchRollbackResponse, error) {

    reader, _ := server.storage.Reader(req.Context)
    defer reader.Close()

    txn := mvcc.NewMvccTxn(reader, req.StartVersion)

    for _, key := range req.Keys {
        // 检查是否已处理
        write, _, _ := txn.CurrentWrite(key)
        if write != nil {
            if write.Kind == mvcc.WriteKindRollback {
                continue  // 已回滚，幂等跳过
            }
            // 已提交，回滚失败
            return &kvrpcpb.BatchRollbackResponse{
                Error: &kvrpcpb.KeyError{Abort: "already committed"},
            }, nil
        }

        // 删除锁（如果存在且属于此事务）
        lock, _ := txn.GetLock(key)
        if lock != nil && lock.Ts == req.StartVersion {
            txn.DeleteLock(key)
        }

        // 删除数据，写 Rollback 标记
        txn.DeleteValue(key)
        txn.PutWrite(key, req.StartVersion, &mvcc.Write{
            StartTs: req.StartVersion,
            Kind:    mvcc.WriteKindRollback,
        })
    }

    server.storage.Write(req.Context, txn.Writes())
    return &kvrpcpb.BatchRollbackResponse{}, nil
}

func (server *Server) KvResolveLock(_ context.Context, req *kvrpcpb.ResolveLockRequest) (
    *kvrpcpb.ResolveLockResponse, error) {

    reader, _ := server.storage.Reader(req.Context)
    defer reader.Close()

    txn := mvcc.NewMvccTxn(reader, req.StartVersion)

    // 获取所有属于此事务的锁
    locks, _ := mvcc.AllLocksForTxn(txn)

    for _, lockInfo := range locks {
        txn2 := mvcc.NewMvccTxn(reader, req.StartVersion)

        if req.CommitVersion == 0 {
            // 回滚
            txn2.DeleteValue(lockInfo.Key)
            txn2.DeleteLock(lockInfo.Key)
            txn2.PutWrite(lockInfo.Key, req.StartVersion, &mvcc.Write{
                StartTs: req.StartVersion,
                Kind:    mvcc.WriteKindRollback,
            })
        } else {
            // 提交
            txn2.DeleteLock(lockInfo.Key)
            txn2.PutWrite(lockInfo.Key, req.CommitVersion, &mvcc.Write{
                StartTs: req.StartVersion,
                Kind:    lockInfo.Lock.Kind,
            })
        }

        server.storage.Write(req.Context, txn2.Writes())
    }

    return &kvrpcpb.ResolveLockResponse{}, nil
}

// 在 KvCommit 中使用 Latches
server.Latches.WaitForLatches(req.Keys)
defer server.Latches.ReleaseLatches(req.Keys)

// 错误做法：不检查 WriteKind
if write != nil {
    return txn.Reader.GetCF(CfDefault, EncodeKey(key, write.StartTs))
}

// 正确做法：必须检查 WriteKind
if write.Kind != WriteKindPut {
    return nil, nil  // DELETE 或 ROLLBACK 不返回值
}

锁不存在（lock == nil）：
  可能原因：锁已被 Rollback 清理（TTL 超时被其他事务清理）
  正确行为：返回 Retryable Error，客户端应该重新执行整个事务

锁 ts 不匹配（lock.ts != startVersion）：
  可能原因：有另一个事务在同一 key 上加了新锁（说明当前事务的锁已被清理）
  正确行为：返回 Retryable Error

// 错误做法：遇到 lock 错误直接返回
if err != nil {
    return nil, err
}

// 正确做法：将 lock 错误加入 pairs 继续扫描
pairs = append(pairs, &kvrpcpb.KvPair{Error: keyErr.Err})
continue

// 无锁时，必须写 Rollback 标记，即使 Default CF 中没有数据
txn.DeleteValue(req.PrimaryKey)   // 即使不存在也无害
txn.PutWrite(req.PrimaryKey, req.LockTs, Rollback)

// 维护 scan.key，跳过相同 user key 的多个版本
if bytes.Equal(currentKey, scan.key) {
    scan.iter.Next()
    continue
}
scan.key = currentKey

TrueTime 返回的不是单一时间点，而是一个区间：
TT.now() = [earliest, latest]
其中 |latest - earliest| ≤ 2ε（ε 通常为 1-7 毫秒）

提交流程：
1. 获取提交时间戳 s = TT.now().latest
2. 执行 Prepare（等待所有参与者就绪）
3. 等待（commit-wait）：等到 TT.now().earliest > s（即确保 s 已过去）
4. 执行 Commit

HLC 时钟 = (physical, logical)
规则：
  发送消息时：HLC = max(本地HLC, 消息HLC) + 1
  本地事件时：HLC = max(本地HLC, NTP时钟) + 1

如果读事务的 maxTS = readTS + uncertaintyInterval
遇到 commitTS ∈ (readTS, maxTS) 的写操作 → 不确定，需要重启事务

如果高优先级事务 T2 遇到低优先级事务 T1 的锁：
  T2 可以强制推进 T1：
  - 如果 T1 的优先级较低 → T1 的 commitTS 被推高（晚于 T2）
  - T2 继续执行，T1 在提交时需要重新校验

-- 开始 XA 事务
XA START 'transaction_id';
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
XA END 'transaction_id';

-- Prepare 阶段
XA PREPARE 'transaction_id';

-- Commit 或 Rollback
XA COMMIT 'transaction_id';
-- 或
XA ROLLBACK 'transaction_id';

XA RECOVER;  -- 返回所有 PREPARED 状态的事务

正向事务序列：
T1: 创建订单（订单服务）
T2: 扣减库存（库存服务）
T3: 扣款（支付服务）
T4: 更新积分（积分服务）

补偿事务序列（逆序执行）：
C4: 撤销积分更新
C3: 退款
C2: 恢复库存
C1: 取消订单

失败场景（T3 失败）：
T1 → T2 → T3（失败）→ C2 → C1

客户端等待 Prewrite 完成 → 等待 Commit 完成 → 返回结果

客户端等待 Prewrite 完成 → 立即返回（Commit 在后台异步进行）

单 key 写：PrewriteRequest → CommitRequest（2 次 RPC）

单 key 写：PrewriteRequest（含 commit 信息）（1 次 RPC）

// 现有实现：锁冲突直接返回错误
if err, ok := err.(*mvcc.KeyError); ok {
    pairs = append(pairs, &kvrpcpb.KvPair{Error: err.Err})
}

场景：医院值班系统，约束：至少 1 名医生在岗
当前状态：医生 A（on_call=true），医生 B（on_call=true）

T1（医生 A 申请下班）:
  startTS=100
  Read: SELECT count(*) FROM doctors WHERE on_call=true → 2
  决策: 2 > 1，可以下班
  Write: UPDATE doctors SET on_call=false WHERE id='A'

T2（医生 B 申请下班）:
  startTS=100（与 T1 相同快照）
  Read: SELECT count(*) FROM doctors WHERE on_call=true → 2
  决策: 2 > 1，可以下班
  Write: UPDATE doctors SET on_call=false WHERE id='B'

两个事务都提交成功（无冲突，因为写的是不同 key）
最终结果：A.on_call=false, B.on_call=false → 0 名医生在岗！违反约束

Safe Point = 100
CfDefault 中 startTS < 100 的旧版本 → 可以删除
CfWrite 中 commitTS < 100 的旧提交记录（保留每个 key 的最新一条）→ 部分删除

CfWrite（正序）:
  Alice/TS=85   （commitTS=85）
  Alice/TS=100  （commitTS=100）
  Alice/TS=200  （commitTS=200）

读操作（startTS=150）：
  Seek(Alice, TS=150)
  → 找到第一个 >= TS=150 的记录：Alice/TS=200
  → 但 200 > 150，不可见！
  → 需要向后扫，找上一个记录 Alice/TS=100 ✓

CfWrite（倒序，大 TS 排前面）:
  Alice/~TS=85   （对应 commitTS=85，字节值最大）
  Alice/~TS=100  （对应 commitTS=100）
  Alice/~TS=200  （对应 commitTS=200，字节值最小）

实际存储顺序（按字节值升序）:
  Alice/~TS=200  （字节值最小，排最前）
  Alice/~TS=100
  Alice/~TS=85

读操作（startTS=150）：
  Seek(EncodeKey("Alice", 150)) = Seek(Alice/~TS=150)
  → 第一条 >= Alice/~150 的记录：Alice/~100（因为 ~100 > ~150）
  → 对应 commitTS=100 ≤ 150，直接可用！✓

活跃事务: T1(startTS=90), T2(startTS=95), T3(startTS=110)
Safe Point = min(90, 95, 110) = 90

Safe Point = 90 意味着：
  - commitTS < 90 的历史版本：没有任何活跃事务的 startTS <= 90 能读到这些版本
  - 可以安全删除（保留每个 key 在 TS=90 时刻可见的最新版本）

场景：
大事务（写入 10000 个 key，需要 5 秒）
TTL = 3 秒

时序：
t=0: 开始 Prewrite
t=3: TTL 到期，其他事务清理了部分锁
t=5: 事务继续 Commit → 发现锁已被清理 → 事务失败，需要重试

结果：大事务被误杀，需要重试，浪费资源

场景：
事务崩溃（客户端宕机），遗留了锁
TTL = 10 分钟

影响：
读取该 key 的其他事务等待长达 10 分钟（或需要主动发起 CheckTxnStatus）
系统吞吐量大幅下降

base_ttl = 3000ms（3 秒）
size_factor = 1ms per KB

TTL = base_ttl + size_factor × txn_size_kb
    = 3000 + 1 × 10000  （10MB 事务）
    = 13000ms（13 秒）

// 错误写法（未验证 user key）
iter.Seek(EncodeKey(key, txn.StartTS))
if iter.Valid() {
    item := iter.Item()
    value, _ := item.Value()
    write, _ := ParseWrite(value)
    return txn.Reader.GetCF(CfDefault, EncodeKey(key, write.StartTs))
    // ↑ 如果 Seek 到了下一个 key 的版本，write.StartTs 是另一个 key 的 TS！
}

// 正确写法（必须验证 user key）
iter.Seek(EncodeKey(key, txn.StartTS))
if iter.Valid() {
    item := iter.Item()
    userKey := DecodeUserKey(item.Key())
    if !bytes.Equal(userKey, key) {
        return nil, nil  // 已超出此 key 的范围
    }
    // ... 继续处理
}

for iter.Seek(EncodeKey(key, TsMax)); iter.Valid(); iter.Next() {
    item := iter.Item()
    userKey := DecodeUserKey(item.Key())

    if !bytes.Equal(userKey, key) {
        break  // 已扫完此 key 的所有版本，进入下一个 key
    }

    // 处理...
}

情况1：lock.ts != startVersion（锁属于另一个事务）
  → Retryable: 当前事务的锁已被清理，客户端应重新开始事务

情况2：write.kind == WriteKindRollback（已回滚）
  → Abort: 不应该再提交（返回 Retryable: "already rolled back"）
  → 注意：有些实现在这里返回非 Retryable 错误，需根据测试要求调整

情况3：lock == nil（锁不存在）
  → Retryable: 同情况1，事务可能已超时被回滚

// 已 Rollback 的 key 直接跳过（幂等）
if write != nil && write.Kind == WriteKindRollback {
    continue  // 不重复写 Rollback
}

// 已提交的 key 报错
if write != nil && write.Kind != WriteKindRollback {
    return AbortError("already committed")
}

lock = GetLock(primaryKey)
if lock == nil {
    return {Action: LockNotExistRollback}  // 只返回，不写 Rollback 标记
}

t=1: T1 Prewrite（获得 startTS=100）
t=2: T1 的 Prewrite 网络延迟，锁还未写入
t=3: T2 读取 primaryKey，发现无锁 → CheckTxnStatus 返回 LockNotExistRollback（但未写标记）
t=4: T1 的 Prewrite 到达 → 锁成功写入！
t=5: T1 尝试 Commit → 发现锁存在 → 成功提交！
t=6: 数据不一致（T2 已经认为 T1 不存在，但 T1 实际提交了）

if lock == nil {
    txn.PutWrite(primaryKey, lockTS, WriteKindRollback)
    server.storage.Write(...)
    return {Action: LockNotExistRollback}
}

// 扫描所有 CfLock，找属于 startTS 的锁
func AllLocksForTxn(txn *MvccTxn) ([]KlPair, error) {
    iter := txn.Reader.IterCF(engine_util.CfLock)
    defer iter.Close()

    var locks []KlPair
    for iter.Seek(nil); iter.Valid(); iter.Next() {
        item := iter.Item()
        value, _ := item.Value()
        lock, _ := ParseLock(value)
        if lock.Ts == txn.StartTS {
            locks = append(locks, KlPair{Key: item.Key(), Lock: lock})
        }
    }
    return locks, nil
}

// TinyKV 中 PutWrite 参数：key, ts（commitTS）, write
txn.PutWrite(key, req.CommitVersion, &mvcc.Write{StartTs: req.StartVersion, Kind: lock.Kind})

// Column Families
engine_util.CfDefault  = "default"  // 存储实际值
engine_util.CfLock     = "lock"     // 存储锁
engine_util.CfWrite    = "write"    // 存储提交记录

// Write Kinds
mvcc.WriteKindPut      // 写入操作
mvcc.WriteKindDelete   // 删除操作
mvcc.WriteKindRollback // 回滚标记

// Lock Kinds（与 Write 相同类型）
mvcc.LockKindPut
mvcc.LockKindDelete

// 特殊时间戳
mvcc.TsMax = math.MaxUint64  // 最大时间戳，用于 Seek 最新版本

事务生命周期：

1. 客户端获取 startTS（TSO）

2. Prewrite 阶段（FOR 每个 mutation）:
   a. MostRecentWrite(key) → commitTS >= startTS? → WriteConflict
   b. GetLock(key) → lock != nil? → LockConflict
   c. PutValue(key, value) + PutLock(key, Lock{primary, startTS, TTL})
   d. storage.Write（批量原子提交）

3. 客户端获取 commitTS（TSO）

4. Commit 阶段（FOR 每个 key）:
   a. CurrentWrite(key) → 已有记录? → 幂等成功/已回滚报错
   b. GetLock(key) → 锁不存在或 ts 不匹配? → Retryable Error
   c. PutWrite(key, commitTS, Write{startTS, kind}) + DeleteLock(key)
   d. storage.Write（批量原子提交）

5. 崩溃恢复（遇到 Stale Lock）:
   a. CheckTxnStatus(primaryKey, lockTS) → 确认事务状态
   b. ResolveLock(startTS, commitTS) → 批量解决所有锁