1. 这一 Lab 在学什么

Lab 2 是 单机 KV Server，看起来很简单：

Get(key) → value
Put(key, value)
Append(key, value) → oldValue

但有一个关键挑战：网络是不可靠的。Clerk（客户端）发的请求可能：

• 丢失（请求或响应包都可能丢）
• 长时间延迟后才到
• 被打乱顺序

如果 Clerk 检测到 RPC 失败就重试，那同一个 Append("x", "abc") 可能
在服务器上被执行两次——结果就变成了 "abcabc"。这是不能接受的。

这一 Lab 的目标就是用会话语义（at-most-once） 让重试安全：
即便 Clerk 把同一个请求发了 100 次，服务器最多真正应用一次。

为什么不靠 TCP 的"有序可靠传输"解决？因为 TCP 只在一条连接内保证。
RPC 失败重连时是新的 TCP 连接，应用层必须自己做幂等。这是几乎所有
真实分布式系统都要处理的问题。

2. 实现关键：ClerkID + 单调 Seq

sequenceDiagram participant C as Clerk id=42 participant S as KVServer Note over C: seq=1 C->>S: Append k=a clerk=42 seq=1 Note over S: log[42] = (seq=1, oldValue="") S-->>C: oldValue="" Note over C: ackSeq=1, seq=2 Note over C: 重试场景 C->>S: Append k=b clerk=42 seq=2 ackSeq=1 Note over S: 收到 ackSeq=1，删除 log[42] 旧条目 Note over S: 应用 v=ab，log[42]=(seq=2, oldValue=a) S-->>C: oldValue=a（网络丢了，Clerk 没收到） Note over C: ack 还是 1，seq 不变 C->>S: Append k=b clerk=42 seq=2 ackSeq=1（重试） Note over S: log[42].seq == 2 == args.seq → 重复 S-->>C: oldValue=a（回放上次结果） Note over C: ackSeq=2

关键点：

1. ClerkID 在 Clerk 创建时一次性随机产生（63 位），全局唯一。
2. SeqID 在 Clerk 内单调递增。重试同一请求时复用 SeqID——这一点
   非常重要，是去重的基础。
3. 服务端 log[clerkID] 只存该 Clerk 最近一条 已应用的写。原因：
   Clerk 是顺序发请求的，下一条新请求到达时一定意味着上一条已经被
   Clerk 收到过结果（否则 Clerk 还在重试）。
4. AckSeq 让服务端可以及时清理 log——否则 log 会随 Clerk 永久增长。

3. 代码

common

package kvsrv

// Lab 2 单机 KV Server 的 RPC 协议定义。
//
// 核心难点：在不可靠网络下，客户端会重试 RPC，因此服务端**必须**判重，
// 否则 Append "x" 会被多次执行，最终值变成 "xxx"。
//
// 解法：每个 Clerk 在初始化时随机一个 64 位 ClerkID，请求按 SeqID
// 单调递增编号。服务端记下 (ClerkID → 最近一次完成的 Seq + 结果) 即可
// 安全地把重复请求 *直接返回上次结果*，而不重新执行。
//
// 内存优化：客户端在收到响应后，会下一次 RPC 里把 AckSeq 带回来，
// 服务端据此清理"已经被客户端确认收到"的旧结果。

// Op 标记一次写入是 Put 还是 Append（Get 不走这个 RPC）。
type Op int

const (
	OpPut    Op = iota // 覆盖
	OpAppend           // 拼接到旧值末尾，并返回拼接前的旧值（at-most-once 必需）
)

// PutAppendArgs 是 Put / Append 的统一参数。
type PutAppendArgs struct {
	Key   string
	Value string
	Op    Op

	ClerkID int64 // 唯一标识一个 Clerk
	SeqID   int64 // 该 Clerk 内单调递增
	AckSeq  int64 // 客户端已经"看到结果"的最大 SeqID，用于服务端清理缓存
}

type PutAppendReply struct {
	Value string // Append 返回拼接前的旧值；Put 留空
}

type GetArgs struct {
	Key string

	ClerkID int64
	SeqID   int64
	AckSeq  int64
}

type GetReply struct {
	Value string
}

Server

package kvsrv

// 单机 KV Server 实现。
//
// 关键不变量：对于同一个 (ClerkID, SeqID)，服务端最多只把它**真正应用**到
// KV map 一次，但可以"返回上次的结果"任意多次。
//
// 内存模型：
//   - kv  map[string]string                : 实际数据
//   - log map[int64]lastEntry               : 每个 ClerkID 的最近一次结果（用于幂等）
//
// 当客户端在新请求里 AckSeq=N，意味着"我已经收到 SeqID ≤ N 的回复"，
// 此时服务端就能安全地从 log 中删掉 SeqID ≤ N 的条目。
// （但只删等于 lastEntry.Seq 的，因为 log 每个 Clerk 只存最近一条。）

import (
	"sync"
)

type lastEntry struct {
	Seq   int64
	Value string // 上次返回给客户端的值（Append 用）
}

type KVServer struct {
	mu  sync.Mutex
	kv  map[string]string
	log map[int64]lastEntry // ClerkID → 最近一次完成的写
}

// StartKVServer 由测试代码调用。
func StartKVServer() *KVServer {
	return &KVServer{
		kv:  map[string]string{},
		log: map[int64]lastEntry{},
	}
}

// Get 是只读操作，理论上幂等，不需要 dedup；
// 但客户端仍可能把 AckSeq 带过来，我们顺便清理。
func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	kv.mu.Lock()
	defer kv.mu.Unlock()

	kv.gcLocked(args.ClerkID, args.AckSeq)
	reply.Value = kv.kv[args.Key]
}

// PutAppend 是核心写路径，必须做幂等。
//
// 三种情况：
//   1) SeqID == log[clerk].Seq —— 重复请求，返回上次结果；
//   2) SeqID >  log[clerk].Seq —— 新请求，应用并记录；
//   3) SeqID <  log[clerk].Seq —— 不应该出现（Clerk 顺序发请求），
//                                 保险起见也直接 ignore。
func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	kv.mu.Lock()
	defer kv.mu.Unlock()

	kv.gcLocked(args.ClerkID, args.AckSeq)

	if last, ok := kv.log[args.ClerkID]; ok && last.Seq >= args.SeqID {
		if last.Seq == args.SeqID {
			// 已应用过，回放结果
			reply.Value = last.Value
		}
		// last.Seq > args.SeqID：迟到的旧请求，直接丢
		return
	}

	switch args.Op {
	case OpPut:
		kv.kv[args.Key] = args.Value
		// Put 不需要返回旧值，但仍记录 SeqID 让重复请求快速返回
		kv.log[args.ClerkID] = lastEntry{Seq: args.SeqID}
	case OpAppend:
		old := kv.kv[args.Key]
		kv.kv[args.Key] = old + args.Value
		reply.Value = old
		kv.log[args.ClerkID] = lastEntry{Seq: args.SeqID, Value: old}
	}
}

// gcLocked 在客户端确认 AckSeq 后清理 log 项。
// 调用前必须持有 kv.mu。
func (kv *KVServer) gcLocked(clerkID, ackSeq int64) {
	if ackSeq <= 0 {
		return
	}
	if last, ok := kv.log[clerkID]; ok && last.Seq <= ackSeq {
		delete(kv.log, clerkID)
	}
}

Client

package kvsrv

// Clerk —— KV Server 的客户端封装。
//
// 行为：
//   - Get 直接转发；服务器无状态，不需要去重也能正确。
//   - Put / Append 在不可靠网络下需要重试，所以每次请求带 ClerkID + 单调
//     递增的 SeqID，让服务器幂等去重。
//   - 上一次 RPC 成功后，把 SeqID 写到 ackSeq；下一次发新请求时一并带过去，
//     这样服务器可以安全清理 log，避免无限增长。

import (
	"crypto/rand"
	"math/big"

	"6.5840/labrpc"
)

type Clerk struct {
	server *labrpc.ClientEnd

	clerkID int64
	seq     int64 // 下一次请求要使用的 SeqID（递增分配）
	ackSeq  int64 // 已经成功收到响应的最大 SeqID
}

func MakeClerk(server *labrpc.ClientEnd) *Clerk {
	return &Clerk{
		server:  server,
		clerkID: nrand(),
		seq:     0,
		ackSeq:  0,
	}
}

// Get 一直重试，直到服务器返回。
func (ck *Clerk) Get(key string) string {
	args := GetArgs{Key: key, ClerkID: ck.clerkID, SeqID: ck.nextSeq(), AckSeq: ck.ackSeq}
	for {
		reply := GetReply{}
		if ok := ck.server.Call("KVServer.Get", &args, &reply); ok {
			ck.ackSeq = args.SeqID
			return reply.Value
		}
		// 网络失败：重试。复用同一个 SeqID 是安全的（服务器幂等）。
	}
}

// Put 不返回值。
func (ck *Clerk) Put(key, value string) {
	ck.putAppend(key, value, OpPut)
}

// Append 返回拼接 *前* 的旧值（论文里 at-most-once 的标准接口）。
func (ck *Clerk) Append(key, value string) string {
	return ck.putAppend(key, value, OpAppend)
}

func (ck *Clerk) putAppend(key, value string, op Op) string {
	args := PutAppendArgs{
		Key:     key,
		Value:   value,
		Op:      op,
		ClerkID: ck.clerkID,
		SeqID:   ck.nextSeq(),
		AckSeq:  ck.ackSeq,
	}
	for {
		reply := PutAppendReply{}
		if ok := ck.server.Call("KVServer.PutAppend", &args, &reply); ok {
			ck.ackSeq = args.SeqID
			return reply.Value
		}
	}
}

func (ck *Clerk) nextSeq() int64 {
	ck.seq++
	return ck.seq
}

// nrand 返回 63 位随机非负整数。Clerk 用它生成自己的全局唯一 ID。
func nrand() int64 {
	max := big.NewInt(int64(1) << 62)
	x, _ := rand.Int(rand.Reader, max)
	return x.Int64()
}