日志复制与安全性#

public class LogEntry {
    private long term;       // 创建该条目时的任期号
    private int index;       // 该条目在日志中的唯一位置
    private Command command; // 待执行的操作命令

    // 省略 getter/setter
}

public class RaftLeader {
    private int commitIndex = 0;  // 已知的被多数派接受的日志位置
    private int lastApplied = 0;  // 已应用到状态机的日志位置

    public void handleClientRequest(Command command) {
        // 1. 创建新日志条目
        LogEntry entry = new LogEntry(
            currentTerm,
            nextIndex++,  // 下一个日志索引
            command
        );
        log.append(entry);

        // 2. 并行发送给所有 Follower
        for (Follower f : followers) {
            int prevIndex = entry.index - 1;
            long prevTerm = log.get(prevIndex).term;

            AppendEntriesRequest req = new AppendEntriesRequest(
                currentTerm,
                leaderId,
                prevIndex,
                prevTerm,
                List.of(entry),  // 增量日志
                commitIndex
            );

            // 异步发送，不阻塞
            sendAsync(f, req);
        }
    }

    // 当收到 Follower 的 ACK 时调用
    public void onAppendEntriesAck(Follower f, AppendEntriesResponse resp) {
        if (resp.success()) {
            // 更新 matchIndex
            f.setMatchIndex(resp.matchIndex());
            f.setNextIndex(resp.matchIndex() + 1);

            // 检查是否可以提交
            if (countMatchingReplicas(f.getMatchIndex()) > majority) {
                // 多数派已接受，可以提交
                if (log.get(resp.matchIndex()).term == currentTerm) {
                    commitIndex = resp.matchIndex();
                    applyEntries();
                }
            }
        } else {
            // 拒绝：回退 nextIndex，重试
            f.setNextIndex(f.getNextIndex() - 1);
            retryAppendEntries(f);
        }
    }

    private void applyEntries() {
        while (lastApplied < commitIndex) {
            lastApplied++;
            LogEntry entry = log.get(lastApplied);
            stateMachine.apply(entry.command());
        }
    }
}

public class RaftFollower {
    public AppendEntriesResponse handleAppendEntries(AppendEntriesRequest req) {
        // 一致性检查
        if (req.prevLogIndex > 0) {
            // 如果 prevLogIndex 超出本地日志范围
            if (req.prevLogIndex > log.size()) {
                return new AppendEntriesResponse(false, log.size());
            }

            // 如果 prevLogIndex 处的 term 不匹配
            if (log.get(req.prevLogIndex - 1).term != req.prevLogTerm) {
                // 冲突：需要回退日志
                truncateLog(req.prevLogIndex);
                return new AppendEntriesResponse(false, log.size());
            }
        }

        // 通过检查，追加新日志
        for (LogEntry entry : req.entries) {
            log.append(entry);
        }

        // 更新 commitIndex
        if (req.leaderCommit > commitIndex) {
            commitIndex = Math.min(req.leaderCommit, log.size());
        }

        return new AppendEntriesResponse(true, log.size());
    }
}

public class RaftNode {
    private int lastSnapshotIndex = 0;
    private byte[] snapshotData;

    public void takeSnapshot(int lastIncludedIndex) {
        // 1. 状态机生成快照
        snapshotData = stateMachine.snapshot();

        // 2. 丢弃快照之前的日志
        log.truncateBefore(lastIncludedIndex);

        // 3. 记录快照位置
        lastSnapshotIndex = lastIncludedIndex;
    }

    // 发送快照给落后太远的 Follower
    public void sendSnapshot(Follower f) {
        SnapshotRequest req = new SnapshotRequest(
            lastSnapshotIndex,
            snapshotData
        );
        sendAsync(f, req);
    }

    public SnapshotResponse handleSnapshotRequest(SnapshotRequest req) {
        // 用快照覆盖本地状态
        stateMachine.restore(req.snapshotData());
        log.restoreFrom(req.lastIncludedIndex());
        return new SnapshotResponse(true);
    }
}

public class RaftLeader {
    // 幂等处理
    public Response handleClientRequest(Request req) {
        // 检查请求 ID 是否已处理
        if (processedRequests.contains(req.id())) {
            return getCachedResponse(req.id());
        }

        // 正常处理
        Response resp = processRequest(req);
        processedRequests.put(req.id(), resp);
        return resp;
    }
}

public class RaftLeader {
    public void onAppendEntriesFailure(Follower f, AppendEntriesResponse resp) {
        int gap = log.size() - f.getMatchIndex();

        if (gap > SNAPSHOT_THRESHOLD) {
            // 差距太大，发送快照
            sendSnapshot(f);
        } else {
            // 差距不大，逐步回退重试
            f.setNextIndex(f.getNextIndex() - 1);
            retryAppendEntries(f);
        }
    }
}