WormDb.Cluster 1.0.2

WormDb.Cluster

WormDb 的 gRPC 集群复制模块，实现 Follower → Leader 单向推流复制。

写节点（Follower）先将日志本地落盘，再异步推送到汇聚节点（Leader）归档。支持断线自动重连、补偿推送、背压保护。

功能

核心 API

Leader 端

// LeaderNode
int AggregateBatch(PushBatch push);      // 汇聚批次（含去重），返回写入数
void OnFollowerConnected(uint id);        // 跟踪连接
void OnFollowerDisconnected(uint id);
IReadOnlyList<LogEntry> QueryRange(long startNs, long endNs);

// 统计
long TotalReceived / TotalBatches / DuplicateSkipped
int ConnectedFollowers
uint NodeId
ulong GetFollowerWatermark(uint followerId)

Follower 端

// FollowerReplicationClient
bool Write(LogEntry entry);                          // 单条：本地写入 + 推流入队
bool WriteBatch(IReadOnlyList<LogEntry> entries);    // 批量写入（false = 背压）
Task StartAsync();                                    // 启动后台推流循环
Task ShutdownAsync(TimeSpan timeout);                 // 优雅排空 + 关闭
Task RequestReplayAsync(ulong fromSeq);               // 向 Leader 请求回放

// 统计
long SentCount / LastAckSeq / DroppedCount
int PushErrors

gRPC 服务定义

service WormReplication {
    rpc PushEntries (stream PushBatch) returns (PushAck);
    rpc ReplayFrom (ReplayRequest) returns (stream LogEntry);
}

核心组件

数据流

客户端 HTTP 写入
    |
    v
Follower: Write() -> 本地 LogDatabase.Append() -> Channel<T>.TryWrite()
    |
    | gRPC Client Streaming (PushEntries)
    v
Leader: PushEntries() -> LeaderNode.AggregateBatch()
        -> 去重检查(序列号水位) -> db.Append()
        -> 更新持久化水位

依赖

<ProjectReference Include="..\WormDb.Core\WormDb.Core.csproj" />


<PackageReference Include="Google.Protobuf" Version="3.35.0" />
<PackageReference Include="Grpc.AspNetCore" Version="2.80.0" />
<PackageReference Include="Grpc.Net.Client" Version="2.80.0" />
<PackageReference Include="Grpc.Tools" Version="2.81.0" PrivateAssets="All" />


<FrameworkReference Include="Microsoft.AspNetCore.App" />

使用

Leader 模式

using WormDb.Cluster;

var leader = new LeaderNode(nodeId: 0, db, stateFile: "./leader-state.json");
var grpcService = new LeaderReplicationService(leader);
// 注册到 gRPC 服务...

Follower 模式

using WormDb.Cluster;

var follower = new FollowerReplicationClient(
    nodeId: 1,
    localDb,
    leaderAddress: "http://localhost:5100",
    stateFile: "./follower-state.json"
);
await follower.StartAsync();

// 写入（本地落盘 + 异步推流）
follower.Write(new LogEntry { ... });

// 优雅退出
await follower.ShutdownAsync(TimeSpan.FromSeconds(10));