🥮 mooncakes.io

    moonreconcile

    Transport-neutral set reconciliation with invertible sketches

    iblt
    set-reconciliation
    sync
    distributed
    Download zip
    Author
    Version
    0.2.1
    License
    Apache-2.0
    Repository
    Last updated
    6 days ago
    Downloads
    37

    No dependencies

    #MoonReconcile：MoonBit 集合差异摘要与恢复库

    源码与文档： GitHub · 从空目录安装 · 源码结构 · 九项验收对照。

    git clone https://github.com/apoloe4/moonreconcile.git
cd moonreconcile

    两端都有大量记录、只有少数 ID 不同时，先交换紧凑摘要，再恢复双方缺少的 ID，减少全量清单传输。核心算法、协议处理、分区协商和测试均以 MoonBit 实现，无 JS 包装层、C FFI 或外部运行库依赖。

    当前状态：0.2.1，实验性基础库。 面向合作节点的集合同步；不是数据库、文件同步软件或已通过生产审计的安全协议。摘要格式是本项目自定义格式，不兼容 minisketch、Negentropy 或其他 IBLT 实现。

    #可以解决什么问题

    场景输入处理过程输出
    边缘缓存与中心索引同步双方缓存对象的稳定 UInt64 ID交换快照摘要、恢复差异、拉取缺少对象新增与移除 ID 清单
    备份清单核对本地与远端备份块 ID分区清单定位变化、只交换变化分区摘要需要补传或复核的块 ID
    离线节点重新连接双方事件/文档版本 ID固定快照版本、重试扩大摘要、校验结果同步计划或全量回退建议

    业务负责获取、验证和存储实际内容；本库同步的是 ID 集合。同一记录内容变化必须产生不同 ID，或由应用另行核对版本。

    #已实现

    • 三个互不重叠的哈希分带，带有计数、ID 异或与 64 位校验摘要的 IBLT；支持插入、删除、相减及正负差异恢复。
    • Inventory 提供精确本地去重、幂等变更、快照版本号和不可变同步计划；apply 返回新集合，失败不会修改源集合。
    • reconcile 明确返回 Ready / Retry / FullTransfer；恢复不完整时不发布同步计划。
    • MRCL 二进制快照、MRPM 分区清单、CRC-32、版本/保留字段检查、长度检查、分配上限及分带一致性检查。
    • FrameReader 支持分块输入、截断检测和错误后重置。
    • 有序数据库游标可直接构建快照，校验重复和逆序 ID，无需保存完整成员 Map；容量规划器按字节预算给出初始宽度与重试上限。
    • Layout / Manifest / BucketOffer / Collector 支持先定位变化分区、任意顺序收集、幂等重传及全局差异数量限制。
    • Session 累计限制收发帧字节、轮数和 peeling 步数，精确重传记账，固定双方快照并返回明确停止原因。
    • 基础、分区、容量和双端交换示例；60 项测试，四目标 CI；4500 项核心恢复率、160 项会话成本与240项 minisketch 对照，保留失败数据。

    #立即运行

    安装 MoonBit 工具链，确认 moonc -v 不低于 0.10.17（满足赛事0.10.16下限），进入仓库根目录：

    moon check --deny-warn
moon test --deny-warn
moon run examples/basic
moon run examples/partitioned
moon run examples/capacity
moon run examples/exchange

    基础示例输出新增 [5]、移除 [1, 4]，同步结果 [2, 3, 5]。不需要服务器、密钥、账号或网络。

    本地使用本仓库的方式：把自己的示例包放入同一模块并在 moon.pkg 中导入：

    import { "apoloe4/moonreconcile" @reconcile }

    在独立 MoonBit 模块中安装固定版本：moon add apoloe4/moonreconcile@0.2.1。已发布到 Mooncakes，独立安装证据见验收进度；工具链建议使用本次验证的 MoonBit 0.10.17 或更新的兼容版本。

    #核心 API

    以下语句放在可抛出异常的函数中，完整版本见 examples/basic/main.mbt：

    let source = @reconcile.Inventory::from_ids([1UL, 2UL, 3UL], scope=42UL, epoch=7UL)
    let target = @reconcile.Inventory::from_ids([2UL, 3UL, 4UL], scope=42UL, epoch=9UL)
    let config = @reconcile.Config::new(16)
    let wire = target.snapshot(config).encode()
    let received = @reconcile.Snapshot::decode(wire, @reconcile.DecodeLimits::new())
    match @reconcile.reconcile(source.snapshot(config), received, @reconcile.Policy::new()) {
      Ready(plan) => {
        let result = source.apply(plan)
        assert_eq(result.ids(), target.ids())
      }
      Retry(next) => { next.width() |> ignore }
      FullTransfer(_) => { /* 由应用执行全量 ID 清单交换 */ }
    }

    scope 必须标识同一个数据集；epoch 是各节点自己的版本标签，双方不需要相等。目标版本可以小于源版本，库不会自行判断哪个节点更权威。应用必须决定同步方向并验证授权。Inventory 幂等变更不会增加版本号，实际增删会增加。

    #网络接入流程

    1. 冻结双方快照，协商同一个 scope、宽度和 seed。
    2. 经应用的 HTTP/WebSocket/其他通道传输 Snapshot::encode() 数据。
    3. 用 DecodeLimits 和 Snapshot::decode，或分块 FrameReader 接收。
    4. Ready 后根据新增 ID 拉取并验证内容，再在业务事务中提交变更。
    5. Retry 时双方针对同一冻结快照重新生成摘要；FullTransfer 时应用交换全量清单。

    分区模式先交换 Manifest::encode()，再由 Collector::pending() 给出需交换摘要的分区编号；每个分区独立重试，最后 finish() 合并成一个经全局校验的计划。分区编号和布局作为应用消息元数据传输，用 BucketOffer::new 包装收到的快照。完整流程见 examples/partitioned。

    #容量与成本

    width 为每个分带的单元数量，总单元数为 3 * width。快照线长为 68 + 60 * width 字节。容量应该随预计差异数量选择，并用实际负载测量成功率；不承诺某个差异数必然成功。容量不足时加倍宽度，达到策略上限后回退。

    本地 Inventory 精确保存全部 ID，内存为 O(n)；摘要是 O(width)，构建 O(n)，恢复工作量受预算限制，结果排序 O(d log d)。分区清单 O(b)，批量分区摘要构建只扫描一次本地集合；按单个分区重复生成摘要会多次扫描集合。

    固定示例：10,000 条 ID、2 个差异、64 分区，接收清单 1,604 字节，接收变化分区摘要 616 字节，合计 2,220 字节；全量 ID 数组为 80,000 字节。这里只比较该示例的接收数据，不包含请求、传输协议、发送端摘要或实际对象内容，不是普遍性能保证。

    #使用边界

    • IBLT 恢复及指纹校验属于概率方法，存在碰撞或恢复失败的可能。恢复失败会明确返回，但不能宣称不存在指纹碰撞。
    • 内部混合函数和 CRC-32 不是密码学认证。恶意对端、数据隐私和防重放由应用的认证通道、版本策略、内容哈希与授权处理。
    • 不实现网络连接、数据库事务、持久化、定时任务、业务冲突解决、全量传输或对象下载。
    • Sketch::insert/erase 是底层代数操作，调用者需保证合法集合增删；需要去重时使用 Inventory。
    • 共享实例不提供线程同步；并发应用应冻结快照或在外部加锁。

    #文档和来源

    #接口契约与验收证据

    核心契约 · 会话接入 · 实测与原始数据 · 查重更新 · 版本记录。

    #验收说明

    按参赛者2026-10-30提供的验收指南，初审通过后无需再提交验收表格，只需持续开发并上传GitHub。九项要求及对应证据见 验收对照；项目申报书保持不变。

    ReconcileError

    source
    pub(all) suberror ReconcileError {
      InvalidConfig(String)
      Incompatible(String)
      InvalidFrame(String)
      LimitExceeded(String)
      StaleSnapshot
      VerificationFailed
      InvalidPlan(String)
    } derive(Eq, 
    Debug    )

    BucketOffer

    source
    pub struct BucketOffer {
      // private fields
    }

    BucketOffer::new

    source
    fn BucketOffer::new(index : Int, layout : Layout, snapshot : Snapshot) -> BucketOffer raise ReconcileError

    Reuses a decoded snapshot; its fingerprint and routing are checked by Collector.

    BucketOffer::snapshot

    source
    fn BucketOffer::snapshot(self : BucketOffer) -> Snapshot

    CapacityPlan

    source
    pub struct CapacityPlan {
      // private fields
    }

    Conservative starting-size heuristic, not a probabilistic success guarantee. Allocate one cell per estimated difference in each of the three bands.

    CapacityPlan::initial

    source
    fn CapacityPlan::initial(self : CapacityPlan) -> Config

    CapacityPlan::max_width

    source
    fn CapacityPlan::max_width(self : CapacityPlan) -> Int

    Maximum retry width that still fits the supplied snapshot wire budget.

    CapacityPlan::new

    source
    fn CapacityPlan::new(expected_differences : Int, max_wire_bytes? : Int, seed? : UInt64) -> CapacityPlan raise ReconcileError

    CapacityPlan::smaller_than_full_ids

    source
    fn CapacityPlan::smaller_than_full_ids(self : CapacityPlan, members : Int) -> Bool raise ReconcileError

    Compare only summary vs raw UInt64 ID bytes, excluding protocol/payload costs.

    Cell

    source

    Collector

    source
    pub struct Collector {
      // private fields
    }

    Collect validated per-bucket plans in any order, with idempotent retransmits. No global Plan is exposed until every changed bucket has completed.

    Collector::accept

    source
    fn Collector::accept(self : Collector, source : BucketOffer, target : BucketOffer, policy : Policy) -> Outcome raise ReconcileError

    Collector::finish

    source
    fn Collector::finish(self : Collector) -> Plan raise ReconcileError

    Collector::new

    source
    fn Collector::new(source : Manifest, target : Manifest, max_changes? : Int) -> Collector raise ReconcileError

    Collector::pending

    source
    fn Collector::pending(self : Collector) -> Array[Int]

    Config

    source
    pub struct Config {
      // private fields
    } derive(Eq, 
    Debug    )

    Three disjoint hash bands, each of width cells. Immutable configuration.

    Config::cells

    source
    fn Config::cells(self : Config) -> Int

    Config::new

    source
    fn Config::new(width : Int, seed? : UInt64) -> Config raise ReconcileError

    Config::seed

    source
    fn Config::seed(self : Config) -> UInt64

    Config::width

    source
    fn Config::width(self : Config) -> Int

    Config::wire_bytes

    source
    fn Config::wire_bytes(self : Config) -> Int

    DecodeLimits

    source
    pub struct DecodeLimits {
      // private fields
    }

    Per-call limits are checked before allocating the cell array.

    DecodeLimits::new

    source
    fn DecodeLimits::new(max_width? : Int, max_members? : Int) -> DecodeLimits raise ReconcileError

    DecodeReport

    source
    pub struct DecodeReport {
      // private fields
    }

    Decode may be partial; only a complete result is a usable synchronization plan.

    DecodeReport::complete

    source
    fn DecodeReport::complete(self : DecodeReport) -> Bool

    DecodeReport::negative

    source
    fn DecodeReport::negative(self : DecodeReport) -> Array[UInt64]

    DecodeReport::positive

    source
    fn DecodeReport::positive(self : DecodeReport) -> Array[UInt64]

    DecodeReport::residual_cells

    source
    fn DecodeReport::residual_cells(self : DecodeReport) -> Int

    DecodeReport::steps

    source
    fn DecodeReport::steps(self : DecodeReport) -> Int

    Fingerprint

    source
    pub struct Fingerprint {
      // private fields
    } derive(Eq, 
    Debug    )

    Commutative, non-cryptographic set fingerprint. This is not authentication.

    Fingerprint::count

    source
    fn Fingerprint::count(self : Fingerprint) -> Int

    FrameReader

    source
    pub struct FrameReader {
      // private fields
    }

    Incremental framing for a single MRCL snapshot received in network chunks. Failure poisons the reader until reset; trailing bytes are never ignored.

    FrameReader::buffered_bytes

    source
    fn FrameReader::buffered_bytes(self : FrameReader) -> Int

    FrameReader::feed

    source
    fn FrameReader::feed(self : FrameReader, chunk : Bytes) -> Snapshot? raise ReconcileError

    FrameReader::finish

    source
    fn FrameReader::finish(self : FrameReader) -> Unit raise ReconcileError

    Signal transport EOF; truncated input is an error, never an empty set.

    FrameReader::new

    source
    fn FrameReader::new(limits : DecodeLimits) -> FrameReader

    FrameReader::reset

    source
    fn FrameReader::reset(self : FrameReader) -> Unit

    Inventory

    source
    pub struct Inventory {
      // private fields
    }

    Exact source set tracker: idempotent changes, immutable snapshot exports.

    Inventory::apply

    source
    fn Inventory::apply(self : Inventory, plan : Plan) -> Inventory raise ReconcileError

    Atomically applies a plan to a copy, so failures never alter the source. This transfers IDs only. Applications fetch/authenticate payloads separately.

    Inventory::bucket_offer

    source
    fn Inventory::bucket_offer(self : Inventory, layout : Layout, index : Int, config : Config) -> BucketOffer raise ReconcileError

    Single pass for one requested bucket. Calling for many buckets costs O(n*b). For all-bucket exports use bucket_offers, which scans the inventory once.

    Inventory::bucket_offers

    source
    fn Inventory::bucket_offers(self : Inventory, layout : Layout, indexes : Array[Int], config : Config, max_cells? : Int) -> Array[BucketOffer] raise ReconcileError

    Allocate only requested buckets; constrain total cells before allocating.

    Inventory::contains

    source
    fn Inventory::contains(self : Inventory, id : UInt64) -> Bool

    Inventory::epoch

    source
    fn Inventory::epoch(self : Inventory) -> UInt64

    Inventory::fingerprint

    source
    fn Inventory::fingerprint(self : Inventory) -> Fingerprint

    Inventory::from_ids

    source
    fn Inventory::from_ids(ids : Array[UInt64], scope? : UInt64, epoch? : UInt64) -> Inventory

    Initial duplicates are discarded. All UInt64 values, including zero, are valid IDs.

    Inventory::ids

    source
    fn Inventory::ids(self : Inventory) -> Array[UInt64]

    Inventory::insert

    source
    fn Inventory::insert(self : Inventory, id : UInt64) -> Bool raise ReconcileError

    Inventory::manifest

    source
    fn Inventory::manifest(self : Inventory, layout : Layout) -> Manifest

    Inventory::new

    source
    fn Inventory::new(scope? : UInt64, epoch? : UInt64) -> Inventory

    Inventory::remove

    source
    fn Inventory::remove(self : Inventory, id : UInt64) -> Bool raise ReconcileError

    Inventory::snapshot

    source
    fn Inventory::snapshot(self : Inventory, config : Config) -> Snapshot raise ReconcileError

    Layout

    source
    pub struct Layout {
      // private fields
    } derive(Eq, 
    Debug    )

    Hash partitioning permits exchanging sketches only for changed buckets. Both peers must negotiate identical layout; seeds are not secret keys.

    Layout::bucket

    source
    fn Layout::bucket(self : Layout, key : UInt64) -> Int

    Layout::buckets

    source
    fn Layout::buckets(self : Layout) -> Int

    Layout::new

    source
    fn Layout::new(buckets : Int, seed? : UInt64) -> Layout raise ReconcileError

    Manifest

    source
    pub struct Manifest {
      // private fields
    }

    Manifest::changed_buckets

    source
    fn Manifest::changed_buckets(self : Manifest, remote : Manifest) -> Array[Int] raise ReconcileError

    Returns changed buckets in increasing order. Fingerprints are probabilistic.

    Manifest::decode

    source
    fn Manifest::decode(data : Bytes, max_buckets? : Int, max_members? : Int) -> Manifest raise ReconcileError

    Manifest::encode

    source
    fn Manifest::encode(self : Manifest) -> Bytes

    MRPM v1: 64-byte header, 24 bytes per partition fingerprint, CRC-32.

    Manifest::epoch

    source
    fn Manifest::epoch(self : Manifest) -> UInt64

    Manifest::fingerprint

    source
    fn Manifest::fingerprint(self : Manifest) -> Fingerprint

    Outcome

    source
    pub enum Outcome {
      Ready(Plan)
      Retry(Config)
      FullTransfer(Int)
    }

    Plan

    source
    pub struct Plan {
      // private fields
    }

    Private fields prevent unvalidated plans being constructed by consumers.

    Plan::additions

    source
    fn Plan::additions(self : Plan) -> Array[UInt64]

    Plan::removals

    source
    fn Plan::removals(self : Plan) -> Array[UInt64]

    Plan::target_epoch

    source
    fn Plan::target_epoch(self : Plan) -> UInt64

    Policy

    source
    pub struct Policy {
      // private fields
    }

    Policy::new

    source
    fn Policy::new(max_width? : Int, max_differences? : Int) -> Policy raise ReconcileError

    Session

    source
    pub struct Session {
      // private fields
    }

    Transport-neutral observer. Each session binds one frozen source and target epoch. Source snapshots passed on later rounds must retain the same identity. Accepted frames are retained for exact replay detection, bounded by max_bytes.

    Session::decode_steps

    source
    fn Session::decode_steps(self : Session) -> Int

    Session::new

    source
    fn Session::new(source : Snapshot, target_epoch : UInt64, limits : SessionLimits) -> Session raise ReconcileError

    Session::receive

    source
    fn Session::receive(self : Session, source : Snapshot, frame : Bytes) -> SessionState raise ReconcileError

    Every call accounts actual received bytes, even malformed/duplicate frames. Stopped sessions reject further input. Exact accepted replays do not consume additional decode rounds/steps. Errors terminally invalidate the session.

    Session::received_bytes

    source
    fn Session::received_bytes(self : Session) -> Int64

    Session::record_sent

    source
    fn Session::record_sent(self : Session, frame : Bytes) -> SessionState raise ReconcileError

    Call immediately BEFORE sending a local snapshot. A returned Stopped means do not send it. Retransmissions are charged again. This does not perform IO.

    Session::rounds

    source
    fn Session::rounds(self : Session) -> Int

    Session::sent_bytes

    source
    fn Session::sent_bytes(self : Session) -> Int64

    Session::state

    source
    fn Session::state(self : Session) -> SessionState

    SessionLimits

    source
    pub struct SessionLimits {
      // private fields
    }

    Limits apply to one observer's actual sent+received MRCL frames, across retries. Transport headers, payload downloads and full-transfer fallback are separate.

    SessionLimits::new

    source
    fn SessionLimits::new(max_rounds? : Int, max_bytes? : Int64, max_steps? : Int, max_width? : Int, max_members? : Int) -> SessionLimits raise ReconcileError

    SessionState

    source
    pub enum SessionState {
      Awaiting(Config)
      Complete(Plan)
      Stopped(SessionStop)
    }

    SessionStop

    source
    pub enum SessionStop {
      ByteLimit
      RoundLimit
      WorkLimit
      WidthLimit
      SnapshotChanged
      ProtocolError
    } derive(Eq, 
    Debug    )

    SessionStop::label

    source
    fn SessionStop::label(self : SessionStop) -> String

    Sketch

    source
    pub struct Sketch {
      // private fields
    }

    Mutable algebraic sketch. Insertion/removal require valid set semantics. Prefer Inventory when callers cannot guarantee uniqueness.

    Sketch::config

    source
    fn Sketch::config(self : Sketch) -> Config

    Sketch::copy

    source
    fn Sketch::copy(self : Sketch) -> Sketch

    Sketch::decode

    source
    fn Sketch::decode(self : Sketch, budget? : Int) -> DecodeReport raise ReconcileError

    Peels on a private copy. An explicit budget bounds work on untrusted tables. The queue has at most cells + 3*budget entries; stale entries are rechecked.

    Sketch::erase

    source
    fn Sketch::erase(self : Sketch, key : UInt64) -> Unit raise ReconcileError

    Sketch::insert

    source
    fn Sketch::insert(self : Sketch, key : UInt64) -> Unit raise ReconcileError

    Sketch::new

    source
    fn Sketch::new(config : Config) -> Sketch

    Sketch::subtract

    source
    fn Sketch::subtract(self : Sketch, other : Sketch) -> Sketch raise ReconcileError

    Returns a new difference table, without changing either input.

    Snapshot

    source
    pub struct Snapshot {
      // private fields
    }

    Scope separates datasets; epoch identifies an immutable snapshot version.

    Snapshot::config

    source
    fn Snapshot::config(self : Snapshot) -> Config

    Snapshot::decode

    source
    fn Snapshot::decode(data : Bytes, limits : DecodeLimits) -> Snapshot raise ReconcileError

    Wire accepts inventory snapshots, not arbitrary signed difference tables. Strict length/version/reserved-byte checks precede CRC and allocation.

    Snapshot::encode

    source
    fn Snapshot::encode(self : Snapshot) -> Bytes

    MRCL v1, little-endian, fixed 64-byte header and 20-byte cells.

    Snapshot::encoded_size

    source
    fn Snapshot::encoded_size(self : Snapshot) -> Int

    Snapshot::epoch

    source
    fn Snapshot::epoch(self : Snapshot) -> UInt64

    Snapshot::fingerprint

    source
    fn Snapshot::fingerprint(self : Snapshot) -> Fingerprint

    Snapshot::from_sorted_ids

    source
    fn Snapshot::from_sorted_ids(ids : Iter[UInt64], config : Config, scope? : UInt64, epoch? : UInt64, max_members? : Int) -> Snapshot raise ReconcileError

    Build from a strictly increasing database/index cursor without an exact Map. Duplicate/order checks need O(1) state; sketch storage is O(config.width). The cursor must read a consistent application snapshot throughout the scan.

    Snapshot::scope

    source
    fn Snapshot::scope(self : Snapshot) -> UInt64

    reconcile

    source
    fn reconcile(source : Snapshot, remote : Snapshot, policy : Policy) -> Outcome raise ReconcileError

    Reconcile source -> remote. A Retry means both peers must rebuild summaries at the returned configuration for the same frozen snapshots.

    synchronize

    source
    fn synchronize(source : Inventory, remote : Inventory, initial : Config, policy : Policy) -> Outcome raise ReconcileError

    Convenience for source simulations. Network applications should call reconcile on received frames, send Retry requests, and implement their own transport.

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io