TinyKv Project4 Transactions

// GetValue finds the value for key, valid at the start timestamp of this transaction.
// I.e., the most recent value committed before the start of this transaction.
//查询当前事务下，传入 key 对应的 Value。
func (txn *MvccTxn) GetValue(key []byte) ([]byte, error) {
  // Your Code Here (4A).
  // 首先需要根据 write 确定最近版本是不是有效的
  // 如果是有效的，则从 default 中取出 value
  // Your Code Here (4A).
  // 1. 遍历 Write 通过 iter.Seek(EncodeKey(key, txn.StartTS)) 查找遍历 Write，找到 commitTs <= ts 最新 Write；
  iter := txn.Reader.IterCF(engine_util.CfWrite)
  defer iter.Close()
  iter.Seek(EncodeKey(key, txn.StartTS)) // 找到的总是最新版本的 user key
  if !iter.Valid() {
    return nil, nil
  }
  // 2. 判断找到的 key 是不是自己需要的 key
  userKey := DecodeUserKey(iter.Item().KeyCopy(nil))
  if !bytes.Equal(userKey, key) {
    return nil, nil
  }
  // 3. 判断 Write 的 Kind 是不是 WriteKindPut
  value, err := iter.Item().ValueCopy(nil)
  if err != nil {
    return nil, err
  }
  write, err := ParseWrite(value)
  if err != nil {
    return nil, err
  }
  // 4. 从 Default 中获取值
  if write.Kind == WriteKindPut {
    return txn.Reader.GetCF(engine_util.CfDefault, EncodeKey(key, write.StartTS))
  }
  return nil, nil
}

// CurrentWrite searches for a write with this transaction's start timestamp. It returns a Write from the DB and that
// write's commit timestamp, or an error.
//查询当前事务下，传入 key 的最新 Write。
func (txn *MvccTxn) CurrentWrite(key []byte) (*Write, uint64, error) {
  // Your Code Here (4A).
  iter := txn.Reader.IterCF(engine_util.CfWrite)
  defer iter.Close()
  //1. 通过 iter.Seek(EncodeKey(key, math.MaxUint64)) 查询该 key 的最新 Write；
  for iter.Seek(EncodeKey(key, ^uint64(0))); iter.Valid(); iter.Next() {
    item := iter.Item()
    gotKey := item.KeyCopy(nil)
    userKey := DecodeUserKey(gotKey)
    if !bytes.Equal(userKey, key) {
      return nil, 0, nil
    }
    value, err := item.ValueCopy(nil)
    if err != nil {
      return nil, 0, err
    }
    write, err := ParseWrite(value)
    if err != nil {
      return nil, 0, err
    }
    //2. 如果 write.StartTS > txn.StartTS，继续遍历，直到找到 write.StartTS == txn.StartTS 的 Write
    if write.StartTS == txn.StartTS {
      //3. 返回这个 Write 和 commitTs；
      return write, decodeTimestamp(gotKey), nil
    }
  }
  return nil, 0, nil
}

// MostRecentWrite finds the most recent write with the given key. It returns a Write from the DB and that
// write's commit timestamp, or an error.
//查询传入 key 的最新 Write，这里不需要考虑事务的 startTs。
func (txn *MvccTxn) MostRecentWrite(key []byte) (*Write, uint64, error) {
  // Your Code Here (4A).
  iter := txn.Reader.IterCF(engine_util.CfWrite)
  defer iter.Close()
  // 1. 通过 iter.Seek(EncodeKey(key, math.MaxUint64)) 查找；
  iter.Seek(EncodeKey(key, ^uint64(0)))
  if !iter.Valid() {
    return nil, 0, nil
  }
  userKey := DecodeUserKey(iter.Item().KeyCopy(nil))
  if !bytes.Equal(userKey, key) {
    return nil, 0, nil
  }
  value, err := iter.Item().ValueCopy(nil)
  if err != nil {
    return nil, 0, err
  }
  //2. 判断目标 Write 的 key 是不是我们需要的，不是返回空
  write, err := ParseWrite(value)
  if err != nil {
    return nil, 0, err
  }
  // 是直接返回该 Write；
  return write, decodeTimestamp(iter.Item().KeyCopy(nil)), nil
}

// PutValue adds a key/value write to this transaction.
func (txn *MvccTxn) PutValue(key []byte, value []byte) {
  // Your Code Here (4A).
  modify := storage.Modify{
    Data: storage.Put{
      Key:   EncodeKey(key, txn.StartTS),
      Cf:    engine_util.CfDefault,
      Value: value,
    },
  }
  txn.writes = append(txn.writes, modify)
}
// DeleteValue removes a key/value pair in this transaction.
func (txn *MvccTxn) DeleteValue(key []byte) {
  // Your Code Here (4A).
  modify := storage.Modify{
    Data: storage.Delete{
      Key: EncodeKey(key, txn.StartTS),
      Cf:  engine_util.CfDefault,
    },
  }
  txn.writes = append(txn.writes, modify)
}
// GetLock returns a lock if key is locked. It will return (nil, nil) if there is no lock on key, and (nil, err)
// if an error occurs during lookup.
func (txn *MvccTxn) GetLock(key []byte) (*Lock, error) {
  // Your Code Here (4A).
  value, err := txn.Reader.GetCF(engine_util.CfLock, key)
  if err != nil {
    return nil, err
  }
  if value == nil {
    return nil, nil
  }
  lock, err := ParseLock(value) // 反序列化 Lock 结构体
  if err != nil {
    return nil, err
  }
  return lock, nil
}
// PutLock adds a key/lock to this transaction.
func (txn *MvccTxn) PutLock(key []byte, lock *Lock) {
  // Your Code Here (4A).
  modify := storage.Modify{
    Data: storage.Put{
      Key:   key,
      Cf:    engine_util.CfLock,
      Value: lock.ToBytes(), // 序列化 Lock 结构体
    },
  }
  txn.writes = append(txn.writes, modify)
}
// DeleteLock adds a delete lock to this transaction.
func (txn *MvccTxn) DeleteLock(key []byte) {
  // Your Code Here (4A).
  modify := storage.Modify{
    Data: storage.Delete{
      Key: key,
      Cf:  engine_util.CfLock,
    },
  }
  txn.writes = append(txn.writes, modify)
}
// PutWrite records a write at key and ts.
func (txn *MvccTxn) PutWrite(key []byte, ts uint64, write *Write) {
  // Your Code Here (4A).
  modify := storage.Modify{
    Data: storage.Put{
      Key:   EncodeKey(key, ts),
      Cf:    engine_util.CfWrite,
      Value: write.ToBytes(),
    },
  }
  txn.writes = append(txn.writes, modify)
}

// KvGet 只需要判断一下锁的状态，锁存在并且时间戳小于 txn.StartTS 就等待锁释放，返回给客户端
func (server *Server) KvGet(_ context.Context, req *kvrpcpb.GetRequest) (*kvrpcpb.GetResponse, error) {
  // Your Code Here (4B).
  resp := &kvrpcpb.GetResponse{}
  // 1. 获取 Reader
  reader, err := server.storage.Reader(req.Context)
  if regionErr, ok := err.(*raft_storage.RegionError); ok {
    resp.RegionError = regionErr.RequestErr
    return resp, nil
  }
  defer reader.Close()
  // 2. 创建事务，获取 Lock
  txn := mvcc.NewMvccTxn(reader, req.Version)
  lock, err := txn.GetLock(req.Key)
  if regionErr, ok := err.(*raft_storage.RegionError); ok {
    resp.RegionError = regionErr.RequestErr
    return resp, nil
  }
  // 3. Percolator 为了保证快照隔离的时候总是能读到已经 commit 的数据
  // 当发现准备读取的数据被锁定的时候，会等待解锁
  if lock != nil && req.Version >= lock.Ts {
    resp.Error = &kvrpcpb.KeyError{
      Locked: &kvrpcpb.LockInfo{
        PrimaryLock: lock.Primary,
        LockVersion: lock.Ts,
        Key:         req.Key,
        LockTtl:     lock.Ttl,
      },
    }
  }
  // 4. 获取 value
  value, err := txn.GetValue(req.Key)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  if value == nil {
    resp.NotFound = true
  }
  resp.Value = value
  return resp, nil
}

// KvPrewrite 检测 key 是否出现冲突，如果没有的话写入 lock 和 default
// 冲突检测：1. 事务开始之后 write 列是否有数据 2. lock 列是否有数据，不需要在意 lock 的时间
func (server *Server) KvPrewrite(_ context.Context, req *kvrpcpb.PrewriteRequest) (*kvrpcpb.PrewriteResponse, error) {
  // Your Code Here (4B).
  resp := &kvrpcpb.PrewriteResponse{}
  reader, err := server.storage.Reader(req.Context)
  if regionErr, ok := err.(*raft_storage.RegionError); ok {
    resp.RegionError = regionErr.RequestErr
    return resp, nil
  }
  defer reader.Close()
  // 检测事务需要修改的 key 是否出现冲突
  txn := mvcc.NewMvccTxn(reader, req.StartVersion)
  var keyErrors []*kvrpcpb.KeyError
  for _, operation := range req.Mutations {
    write, ts, err := txn.MostRecentWrite(operation.Key)
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    // 检测在 StartTS 之后是否有已经提交的 Write，如果有的话说明写冲突，需要 abort 当前的事务
    if write != nil && ts >= req.StartVersion {
      keyErrors = append(keyErrors, &kvrpcpb.KeyError{
        Conflict: &kvrpcpb.WriteConflict{
          StartTs:    req.StartVersion,
          ConflictTs: ts,
          Key:        operation.Key,
          Primary:    req.PrimaryLock,
        },
      })
      continue
    }
    // 检测 Key 是否有 Lock 锁住，如果有的话则说明别的事务可能正在修改
    lock, err := txn.GetLock(operation.Key)
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    if lock != nil {
      keyErrors = append(keyErrors, &kvrpcpb.KeyError{
        Locked: &kvrpcpb.LockInfo{
          PrimaryLock: req.PrimaryLock,
          LockVersion: lock.Ts,
          Key:         operation.Key,
          LockTtl:     lock.Ttl,
        },
      })
      continue
    }
    // 暂存修改到 txn 中，然后对需要修改的 Key 进行加锁
    var kind mvcc.WriteKind
    switch operation.Op {
    case kvrpcpb.Op_Put:
      kind = mvcc.WriteKindPut
      txn.PutValue(operation.Key, operation.Value)
    case kvrpcpb.Op_Del:
      kind = mvcc.WriteKindDelete
      txn.DeleteValue(operation.Key)
    default:
      return nil, nil
    }
    txn.PutLock(operation.Key, &mvcc.Lock{
      Primary: req.PrimaryLock,
      Ts:      req.StartVersion,
      Ttl:     req.LockTtl,
      Kind:    kind,
    })
  }
  // 判断是否有 key 出错了，如果有的话需要 abort 事务
  if len(keyErrors) > 0 {
    resp.Errors = keyErrors
    return resp, nil
  }
  // 写入事务中暂存的修改到 storage 中
  err = server.storage.Write(req.Context, txn.Writes())
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  return resp, nil
}

// KvCommit 检查 lock，没有冲突的话就写入 write 并清除 lock
// lock 必须要存在，并且时间戳需要等于当前事务的开始时间戳
func (server *Server) KvCommit(_ context.Context, req *kvrpcpb.CommitRequest) (*kvrpcpb.CommitResponse, error) {
  // Your Code Here (4B).
  resp := &kvrpcpb.CommitResponse{}
  reader, err := server.storage.Reader(req.Context)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  defer reader.Close()
  txn := mvcc.NewMvccTxn(reader, req.StartVersion)
  server.Latches.WaitForLatches(req.Keys)
  defer server.Latches.ReleaseLatches(req.Keys)
  for _, key := range req.Keys {
    // 1. 检测是否是重复提交（测试程序需要）
    write, _, err := txn.CurrentWrite(key)
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    // Rollback 类型的 Write 表示是已经正确提交的，这个时候按照重复提交处理
    if write != nil && write.Kind != mvcc.WriteKindRollback && write.StartTS == req.StartVersion {
      return resp, nil
    }
    // 2. 检查每个 Key 的 Lock 是否还存在
    lock, err := txn.GetLock(key)
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    // 如果 Lock 不存在，有两种情况：第一种是事务已经正确提交了，这次是一个重复提交；第二种是这个事务被别的事务清除了
    if lock == nil || lock.Ts != req.StartVersion {
      resp.Error = &kvrpcpb.KeyError{Retryable: "true"}
      return resp, nil
    }
    // 3. 第一次提交事务，正常处理
    txn.PutWrite(key, req.CommitVersion, &mvcc.Write{
      StartTS: req.StartVersion,
      Kind:    lock.Kind,
    })
    txn.DeleteLock(key)
  }
  err = server.storage.Write(req.Context, txn.Writes())
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  return resp, nil
}

func (server *Server) KvScan(_ context.Context, req *kvrpcpb.ScanRequest) (*kvrpcpb.ScanResponse, error) {
  // Your Code Here (4C).
  resp := &kvrpcpb.ScanResponse{}
  reader, err := server.storage.Reader(req.Context)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  txn := mvcc.NewMvccTxn(reader, req.Version)
  scanner := mvcc.NewScanner(req.StartKey, txn)
  defer reader.Close()
  defer scanner.Close()
  var pairs []*kvrpcpb.KvPair // 存储扫描到的 {key, value}
  // 查找 key
  for i := 0; i < int(req.Limit); {
    key, value, err := scanner.Next()
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    // key 为空表示没有数据了
    if key == nil {
      break
    }
    lock, err := txn.GetLock(key)
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    if lock != nil && req.Version >= lock.Ts {
      pairs = append(pairs, &kvrpcpb.KvPair{
        Error: &kvrpcpb.KeyError{
          Locked: &kvrpcpb.LockInfo{
            PrimaryLock: lock.Primary,
            LockVersion: lock.Ts,
            Key:         key,
            LockTtl:     lock.Ttl,
          },
        },
        Key: key,
      })
      i++
      continue
    }
    if value != nil {
      pairs = append(pairs, &kvrpcpb.KvPair{Key: key, Value: value})
      i++
    }
  }
  resp.Pairs = pairs
  return resp, nil
}

// Scanner is used for reading multiple sequential key/value pairs from the storage layer. It is aware of the implementation
// of the storage layer and returns results suitable for users.
// Invariant: either the scanner is finished and cannot be used, or it is ready to return a value immediately.
type Scanner struct {
  // Your Data Here (4C).
  nextKey  []byte
  txn      *MvccTxn
  iter     engine_util.DBIterator
  finished bool
}
// NewScanner creates a new scanner ready to read from the snapshot in txn.
func NewScanner(startKey []byte, txn *MvccTxn) *Scanner {
  // Your Code Here (4C).
  return &Scanner{
    nextKey: startKey,
    txn:     txn,
    iter:    txn.Reader.IterCF(engine_util.CfWrite),
  }
}
func (scan *Scanner) Close() {
  // Your Code Here (4C).
  scan.iter.Close()
}
// Next returns the next key/value pair from the scanner. If the scanner is exhausted, then it will return `nil, nil, nil`.
func (scan *Scanner) Next() ([]byte, []byte, error) {
  // Your Code Here (4C).
  if scan.finished {
    return nil, nil, nil
  }
  // 查找 CfWrite
  key := scan.nextKey
  scan.iter.Seek(EncodeKey(key, scan.txn.StartTS))
  if !scan.iter.Valid() {
    scan.finished = true
    return nil, nil, nil
  }
  item := scan.iter.Item()
  gotKey := item.KeyCopy(nil)
  userKey := DecodeUserKey(gotKey)
  if !bytes.Equal(userKey, key) {
    scan.nextKey = userKey
    return scan.Next()
  }
  // 跳过所有相同的 key
  for {
    scan.iter.Next()
    if !scan.iter.Valid() {
      scan.finished = true
      break
    }
    item := scan.iter.Item()
    gotKey := item.KeyCopy(nil)
    userKey := DecodeUserKey(gotKey)
    if !bytes.Equal(userKey, key) {
      scan.nextKey = userKey
      break
    }
  }
  writeVal, err := item.ValueCopy(nil)
  if err != nil {
    return key, nil, err
  }
  write, err := ParseWrite(writeVal)
  if err != nil {
    return key, nil, err
  }
  if write.Kind == WriteKindDelete {
    return key, nil, nil
  }
  // 查找 CfDefault
  value, err := scan.txn.Reader.GetCF(engine_util.CfDefault, EncodeKey(key, write.StartTS))
  return key, value, err
}

// KvCheckTxnStatus 检查超时时间，删除过期的锁并返回锁的状态
func (server *Server) KvCheckTxnStatus(_ context.Context, req *kvrpcpb.CheckTxnStatusRequest) (*kvrpcpb.CheckTxnStatusResponse, error) {
  // Your Code Here (4C).
  resp := &kvrpcpb.CheckTxnStatusResponse{}
  reader, err := server.storage.Reader(req.Context)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  defer reader.Close()
  txn := mvcc.NewMvccTxn(reader, req.LockTs)
  // 1. 检查 PrimaryKey 是否存在 Write，Write 的开始时间戳与 req.LockTs 相等
  write, ts, err := txn.CurrentWrite(req.PrimaryKey)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  if write != nil {
    // WriteKind 如果不是 WriteKindRollback 则说明已经被 commit
    if write.Kind != mvcc.WriteKindRollback {
      resp.CommitVersion = ts
    }
    return resp, nil
  }
  // 2. 检查 lock 是否存在
  lock, err := txn.GetLock(req.PrimaryKey)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  // lock 不存在，说明 primary key 已经被回滚了，创建一个 WriteKindRollback
  if lock == nil {
    txn.PutWrite(req.PrimaryKey, req.LockTs, &mvcc.Write{
      StartTS: req.LockTs,
      Kind:    mvcc.WriteKindRollback,
    })
    err = server.storage.Write(req.Context, txn.Writes())
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    resp.Action = kvrpcpb.Action_LockNotExistRollback
    return resp, nil
  }
  // lock 不为空，检查 lock 是否超时，如果超时则移除 Lock 和 Value，创建一个 WriteKindRollback
  if mvcc.PhysicalTime(lock.Ts)+lock.Ttl <= mvcc.PhysicalTime(req.CurrentTs) {
    txn.DeleteLock(req.PrimaryKey)
    txn.DeleteValue(req.PrimaryKey)
    txn.PutWrite(req.PrimaryKey, req.LockTs, &mvcc.Write{
      StartTS: req.LockTs,
      Kind:    mvcc.WriteKindRollback,
    })
    err = server.storage.Write(req.Context, txn.Writes())
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    resp.Action = kvrpcpb.Action_TTLExpireRollback
  }
  return resp, nil
}

// KvBatchRollback 批量回滚 key
func (server *Server) KvBatchRollback(_ context.Context, req *kvrpcpb.BatchRollbackRequest) (*kvrpcpb.BatchRollbackResponse, error) {
  // Your Code Here (4C).
  resp := &kvrpcpb.BatchRollbackResponse{}
  reader, err := server.storage.Reader(req.Context)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  defer reader.Close()
  txn := mvcc.NewMvccTxn(reader, req.StartVersion)
  server.Latches.WaitForLatches(req.Keys)
  defer server.Latches.ReleaseLatches(req.Keys)
  for _, key := range req.Keys {
    // 1. 获取 key 的 Write，如果已经是 WriteKindRollback 则跳过这个 key
    write, _, err := txn.CurrentWrite(key)
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    if write != nil {
      if write.Kind == mvcc.WriteKindRollback {
        continue
      } else {
        resp.Error = &kvrpcpb.KeyError{Abort: "true"}
        return resp, nil
      }
    }
    // 获取 Lock，如果 Lock 被清除或者 Lock 不是当前事务的 Lock，则中止操作
    // 这个时候说明 key 被其他事务占用
    // 否则的话移除 Lock、删除 Value，写入 WriteKindRollback 的 Write
    lock, err := txn.GetLock(key)
    if err != nil {
      if regionErr, ok := err.(*raft_storage.RegionError); ok {
        resp.RegionError = regionErr.RequestErr
        return resp, nil
      }
      return nil, err
    }
    if lock == nil || lock.Ts != req.StartVersion {
      txn.PutWrite(key, req.StartVersion, &mvcc.Write{
        StartTS: req.StartVersion,
        Kind:    mvcc.WriteKindRollback,
      })
      continue
    }
    txn.DeleteLock(key)
    txn.DeleteValue(key)
    txn.PutWrite(key, req.StartVersion, &mvcc.Write{
      StartTS: req.StartVersion,
      Kind:    mvcc.WriteKindRollback,
    })
  }
  err = server.storage.Write(req.Context, txn.Writes())
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  return resp, nil
}

func (server *Server) KvResolveLock(_ context.Context, req *kvrpcpb.ResolveLockRequest) (*kvrpcpb.ResolveLockResponse, error) {
  // Your Code Here (4C).
  resp := &kvrpcpb.ResolveLockResponse{}
  reader, err := server.storage.Reader(req.Context)
  if err != nil {
    if regionErr, ok := err.(*raft_storage.RegionError); ok {
      resp.RegionError = regionErr.RequestErr
      return resp, nil
    }
    return nil, err
  }
  iter := reader.IterCF(engine_util.CfLock)
  defer reader.Close()
  defer iter.Close()
  var keys [][]byte
  for ; iter.Valid(); iter.Next() {
    item := iter.Item()
    value, err := item.ValueCopy(nil)
    if err != nil {
      return resp, err
    }
    lock, err := mvcc.ParseLock(value)
    if err != nil {
      return resp, nil
    }
    if lock.Ts == req.StartVersion {
      key := item.KeyCopy(nil)
      keys = append(keys, key)
    }
  }
  if len(keys) == 0 {
    return resp, nil
  }
  // 根据 request.CommitVersion 提交或者 Rollback
  if req.CommitVersion == 0 {
    resp1, err := server.KvBatchRollback(nil, &kvrpcpb.BatchRollbackRequest{
      Context:      req.Context,
      StartVersion: req.StartVersion,
      Keys:         keys,
    })
    resp.Error, resp.RegionError = resp1.Error, resp1.RegionError
    return resp, err
  } else {
    resp1, err := server.KvCommit(nil, &kvrpcpb.CommitRequest{
      Context:       req.Context,
      StartVersion:  req.StartVersion,
      Keys:          keys,
      CommitVersion: req.CommitVersion,
    })
    resp.Error, resp.RegionError = resp1.Error, resp1.RegionError
    return resp, err
  }
}