#读修复与反熵

public class ReadRepairExecutor {
    private final DynamoClient dynamoClient;
    private final int numReplicas;

    public byte[] readWithRepair(String key) {
        // 并发读取所有副本
        List<CompletableFuture<ReplicaResult>> futures = new ArrayList<>();
        for (Replica replica : replicas) {
            futures.add(dynamoClient.getAsync(replica, key));
        }

        // 收集响应
        List<ReplicaResult> results = futures.stream()
            .map(CompletableFuture::join)
            .filter(r -> r.isSuccess())
            .collect(Collectors.toList());

        // 选择最新版本
        ReplicaResult latest = results.stream()
            .max(Comparator.comparing(ReplicaResult::getVersion))
            .orElseThrow(() -> new NoDataException(key));

        // 修复旧版本节点（异步，不阻塞读取）
        for (ReplicaResult result : results) {
            if (result.getVersion() < latest.getVersion()) {
                asyncRepair(result.getReplica(), key, latest);
            }
        }

        return latest.getData();
    }

    private void asyncRepair(Replica replica, String key, ReplicaResult latest) {
        // 异步写回最新值，不等待响应
        dynamoClient.putAsync(replica, key, latest.getData(), latest.getVersion())
            .thenAccept(r -> LOG.debug("Repaired replica {} for key {}", replica, key))
            .exceptionally(e -> {
                LOG.warn("Failed to repair replica {} for key {}", replica, key, e);
                return null;
            });
    }
}

public class MerkleTreeComparison {
    /**
     * 比较两棵 Merkle 树，返回需要同步的数据范围
     */
    public DiffResult compare(MerkleNode a, MerkleNode b) {
        // 如果两个节点哈希相同，数据一致
        if (a.getHash().equals(b.getHash())) {
            return DiffResult.NO_DIFF;
        }

        // 如果是叶子节点且哈希不同，需要同步这个范围的数据
        if (a.isLeaf() && b.isLeaf()) {
            return DiffResult.createRangeDiff(a.getRange(), b.getRange());
        }

        // 如果是内部节点，递归比较子树
        DiffResult diff = new DiffResult();
        for (int i = 0; i < a.getChildren().size(); i++) {
            diff.merge(compare(a.getChildren().get(i), b.getChildren().get(i)));
        }
        return diff;
    }
}

// Cassandra Anti-Entropy 修复流程（简化）
public class AntiEntropyRepair {
    public void repair(String keyspace, String table) {
        // 1. 计算 Merkle 树
        MerkleTree tree = calculateMerkleTree(keyspace, table);

        // 2. 与其他副本交换 Merkle 树
        for (Replica replica : replicas) {
            MerkleTree remoteTree = exchangeMerkleTrees(replica);

            // 3. 比较 Merkle 树
            List<Range> diffs = compareTrees(tree, remoteTree);

            // 4. 对差异范围进行数据同步
            for (Range range : diffs) {
                syncRange(range, replica);
            }
        }
    }

    private void syncRange(Range range, Replica target) {
        // 读取本地数据，按范围分片发送给目标节点
        List<ByteBuffer> data = readRange(range);
        for (ByteBuffer row : data) {
            target.put(row.getKey(), row.getValue(), row.getVersion());
        }
    }
}

// Cassandra 数据同步优先级
public enum SyncStrategy {
    // 1. 最高优先级：Gossip 传播近期变更
    //    节点间每秒交换 Gossip 消息
    //    变更立即同步到其他节点

    // 2. 次优先级：Hinted Handoff
    //    节点故障期间的临时写入
    //    节点恢复后归还数据

    // 3. 普通优先级：Read Repair
    //    读取时被动修复
    //    只修复被读取的数据

    // 4. 最低优先级：Anti-Entropy
    //    定期深度检查
    //    后台批量修复不一致
}