Bitzsoft.Integrations.Sync 1.0.0-alpha.10

Bitzsoft.Integrations.Sync

跨 net5.0、net8.0、net10.0 的可恢复同步引擎，面向目录、ERP、CRM、 云盘和其他 ToB 数据连接器。

安装与兼容性

dotnet add package Bitzsoft.Integrations.Sync

包支持 net5.0、net8.0 和 net10.0。它是调度无关的同步状态机，不创建 Hosted Service、不决定运行频率，也不提供数据库 Adapter。

核心语义

• SyncCursor 是 opaque 值，ToString() 永不输出原文；
• CommittedCursor 表示上次完整成功运行提交的 delta cursor；
• ContinuationCursor 只表示当前运行的下一页；
• 每一页先由幂等 Sink 应用，再原子推进 checkpoint；
• 最终页先落 Sink，再 reconcile，最后提交 delta cursor；
• 进程在任意一步退出都只会造成安全重放，不会越过未完成数据；
• SyncRecord 支持 revision、sequence、modified time 和 tombstone；
• Provider RetryAfter、页预算和取消都保存 checkpoint，不在引擎内阻塞等待；
• 全量运行可生成 missing/unexpected/conflict reconcile 结果；
• tenant/provider/stream/sink/region/environment 全部进入 SyncJobKey。

注册

生产宿主应实现具有数据库原子租约和持久化 checkpoint 的 ISyncStateStore：

services.AddBitzsoftIntegrationSync();
services.AddSingleton<ISyncStateStore, SqlSyncStateStore>();

开发和单实例测试可显式使用内存实现：

services.AddInMemoryIntegrationSync();

基础注册不会隐式选择内存 Store。

Source 与 Sink

Source 只负责分页读取，不自行提交 cursor：

public Task<SyncPage> ReadPageAsync(
    SyncReadRequest request,
    CancellationToken cancellationToken)
{
    // Full 时 request.CommittedCursor 为 null。
    // Incremental 时它是上次成功运行的 delta token。
    // request.ContinuationCursor 只用于当前运行翻页。
}

Sink 必须对相同 revision 的重放幂等，并按 Provider 的 sequence、版本或修改时间 拒绝乱序旧数据：

public Task<SyncApplyResult> ApplyAsync(
    SyncApplyContext context,
    IReadOnlyCollection<SyncRecord> records,
    CancellationToken cancellationToken);

如果一页部分落库后发生异常，下一次会重放该页，因此生产 Sink 必须以 job + record id + revision 建唯一约束或实现等价幂等语义。

运行与恢复

var key = new SyncJobKey(
    tenantId,
    "Directory:scim",
    streamName: "Users",
    sinkName: "identity-db",
    region: "global");

var result = await engine.RunAsync(
    new SyncRunRequest(
        key,
        mode: SyncMode.Automatic,
        pageSize: 200,
        maximumPages: 20),
    source,
    sink,
    reconciler,
    cancellationToken);

MaximumPages 默认 null，不强制限制总记录或正文大小。设置后只限制单次 Worker 调用的公平性；checkpoint 会立即可恢复。

结果可能为：

• Completed：最终 delta cursor 已提交；
• Paused：等待限流时间或可立即从 continuation 恢复；
• Failed：非瞬时错误，checkpoint 保留；
• LeaseUnavailable：另一个 Worker 正在运行。

生产 Adapter 要求

• 数据库唯一任务键；
• compare-and-swap 租约 token 与过期接管；
• checkpoint 与运行计数原子更新；
• cursor 静态加密和日志脱敏；
• Sink 幂等、tombstone 和 Provider 特定版本排序；
• Reset/强制全量操作审计；
• 指标区分 records read/applied/deleted/duplicate/out-of-order；
• reconcile 差异必须由业务策略决定是否自动删除，平台不会默认破坏性清理。