🥮 mooncakes.io

    MoonCRDTKit

    A MoonBit CRDT toolkit for offline-first state merge and collaborative data structures.

    crdt
    offline
    sync
    merge
    collaboration
    Download zip
    Author
    Version
    0.2.0
    License
    Apache-2.0
    Repository
    Last updated
    3 months ago
    Downloads
    31

    No dependencies

    #MoonCRDTKit

    MoonCRDTKit 是一个面向 MoonBit 的离线协同 CRDT 状态合并基础库。

    #安装与最小接入

    在你的 MoonBit 项目根目录安装已发布包：

    moon add ddd-1234d/MoonCRDTKit

    最小离线任务同步示例：

    import { "ddd-1234d/MoonCRDTKit" @crdt }

    let left = @crdt.ReplicaState::new(1)
    left.add_task("draft")
    let right = @crdt.ReplicaState::new(2)
    right.add_task("review")
    let converged = left.merge(right)
    println(converged.tasks.contains("draft"))
    println(converged.tasks.contains("review"))

    ReplicaState 为每次本地操作分配副本内单调序号；ORSet::add_from 使用 (replica, sequence) 作为 Dot 身份，避免不同副本相同序号的并发添加碰撞。旧的 ORSet::add 仅保留给单写者兼容场景。

    项目聚焦弱网与离线场景下的多端状态收敛，提供向量时钟、LWW 寄存器、G-Counter、PN-Counter、OR-Set、变更日志和同步摘要等能力，适合协作文档、离线表单、边缘设备状态同步、多人编辑器和教学算法示例。

    #创新点

    • 离线优先：副本可以先在本地修改状态，恢复连接后再合并。
    • 无中心合并：不依赖服务端仲裁，多个副本按同一规则收敛。
    • 因果可解释：向量时钟和变更日志记录状态来源，便于同步调试。
    • 组合式 CRDT：寄存器、计数器、集合和摘要可以组合成应用状态。
    • 增量同步积木：提供 DeltaBatch 与 SyncPlan，便于上层协议判断是否需要交换状态。

    #与已有 MoonBit CRDT 工作的关系

    MoonCRDTKit 不是对 mizchi/crdt_db 或 mizchi/converge 的重复实现。mizchi/converge 更接近 EG-Walker inspired Local-First DB Sync Engine，关注数据库同步、事件图、operation log、WASM/JS SDK 和后端同步部署。MoonCRDTKit 的边界更小：只做状态型 CRDT 原语和轻量同步辅助，不提供数据库 CRUD、网络传输、WASM Component、Cloudflare/Deno 后端或 BFT 签名层。

    MoonBit 官方文章《Implementing CRDT Algorithms with MoonBit and Building Real-time Collaborative Applications》主要介绍 OT、RGA、EG-Walker、Lomo 等协同文本算法路线。MoonCRDTKit 不实现文本 CRDT，也不做 RGA/EG-Walker/Lomo 富文本编辑器；它面向离线表单、任务集合、计数器、边缘设备状态和教学示例这类非文本状态同步场景。

    因此，本项目定位是 MoonBit 生态中的 CRDT 基础积木库：上层项目可以在它之上自行选择存储、网络协议和应用模型。

    #当前能力

    • VectorClock：记录副本计数，支持 tick、merge、dominates、concurrent 判断。
    • LwwRegister：最后写入胜出寄存器，支持时间戳和副本号确定性裁决。
    • GCounter：只增计数器，按副本取最大值合并。
    • PNCounter：正负计数器，支持离线增减后收敛。
    • ORSet：观察删除集合；多副本调用 add_from(element, replica, sequence)，以 (replica, sequence) 区分并发添加，支持离线添加、观察后删除和合并。
    • ChangeLog：记录副本变更事件并按事件 id 去重。
    • DeltaBatch：根据对端向量时钟提取缺失事件，支持幂等应用。
    • SyncSummary：输出副本同步摘要，便于 CLI、调试面板和状态心跳使用。
    • SyncPlan：比较两个摘要中的完整 CRDT 状态，而非只比较数量；可识别“元素数相同但内容不同”的副本，判断是否需要 push / pull / 双向交换。
    • ReplicaState：将因果时钟、PN-Counter、OR-Set 与 ChangeLog 组合为可直接用于离线任务和计数场景的副本状态。

    #快速示例

    let left = @MoonCRDTKit.VectorClock::new()
    let right = @MoonCRDTKit.VectorClock::new()
    left.tick(1)
    right.tick(2)
    let merged = left.merge(right)

    let counter = @MoonCRDTKit.PNCounter::new()
    counter.increment(1, amount=8)
    counter.decrement(2, amount=3)
    let log = @MoonCRDTKit.ChangeLog::new()
    ignore(log.append(@MoonCRDTKit.ChangeEvent::new(1, 1, "set", "title")))
    let delta = log.delta_after(left, origin=1)

    println(merged.concurrent_with(left))
    println(counter.value())
    println(delta.to_json())

    运行演示：

    moon run cmd/main

    运行测试：

    moon test

    #设计原则

    1. 核心库保持后端中立，不依赖网络、数据库、浏览器或平台 API。
    2. 所有合并函数都保持确定性，便于在多端同步中复现结果。
    3. 用简单 MoonBit 结构体表达 CRDT 状态，优先保证教学性和可测试性。
    4. 不和数据库同步引擎、协同文本编辑器绑定，避免污染基础库抽象。

    #仓库

    更完整的竞品/参考关系说明见 docs/RELATED_WORK.md。

    ChangeEvent

    source
    pub(all) struct ChangeEvent {
      replica : Int
      sequence : Int
      kind : String
      target : String
    } derive(Eq, 
    Debug    )

    ChangeEvent::id

    source
    fn ChangeEvent::id(self : ChangeEvent) -> Int

    ChangeEvent::new

    source
    fn ChangeEvent::new(replica : Int, sequence : Int, kind : String, target : String) -> ChangeEvent

    ChangeEvent::same_identity

    source
    fn ChangeEvent::same_identity(self : ChangeEvent, other : ChangeEvent) -> Bool

    ChangeLog

    source
    pub(all) struct ChangeLog {
      events : Array[ChangeEvent]
    } derive(
    Debug    )

    ChangeLog::append

    source
    fn ChangeLog::append(self : ChangeLog, event : ChangeEvent) -> Bool

    ChangeLog::apply_delta

    source
    fn ChangeLog::apply_delta(self : ChangeLog, delta : DeltaBatch) -> ChangeLog

    ChangeLog::delta_after

    source
    fn ChangeLog::delta_after(self : ChangeLog, known : VectorClock, origin? : Int) -> DeltaBatch

    ChangeLog::length

    source
    fn ChangeLog::length(self : ChangeLog) -> Int

    ChangeLog::merge

    source
    fn ChangeLog::merge(self : ChangeLog, other : ChangeLog) -> ChangeLog

    ChangeLog::new

    source
    fn ChangeLog::new() -> ChangeLog

    ClockEntry

    source
    pub(all) struct ClockEntry {
      replica : Int
      counter : Int
    } derive(Eq, 
    Debug    )

    ClockEntry::new

    source
    fn ClockEntry::new(replica : Int, counter : Int) -> ClockEntry

    CounterEntry

    source
    pub(all) struct CounterEntry {
      replica : Int
      value : Int
    } derive(Eq, 
    Debug    )

    CounterEntry::new

    source
    fn CounterEntry::new(replica : Int, value : Int) -> CounterEntry

    DeltaBatch

    source
    pub(all) struct DeltaBatch {
      origin : Int
      events : Array[ChangeEvent]
    } derive(
    Debug    )

    DeltaBatch::is_empty

    source
    fn DeltaBatch::is_empty(self : DeltaBatch) -> Bool

    DeltaBatch::length

    source
    fn DeltaBatch::length(self : DeltaBatch) -> Int

    DeltaBatch::new

    source
    fn DeltaBatch::new(origin : Int) -> DeltaBatch

    DeltaBatch::to_json

    source
    fn DeltaBatch::to_json(self : DeltaBatch) -> String

    GCounter

    source
    pub(all) struct GCounter {
      entries : Array[CounterEntry]
    } derive(
    Debug    )

    GCounter::get

    source
    fn GCounter::get(self : GCounter, replica : Int) -> Int

    GCounter::increment

    source
    fn GCounter::increment(self : GCounter, replica : Int, amount? : Int) -> Unit

    GCounter::merge

    source
    fn GCounter::merge(self : GCounter, other : GCounter) -> GCounter

    GCounter::new

    source
    fn GCounter::new() -> GCounter

    GCounter::set

    source
    fn GCounter::set(self : GCounter, replica : Int, value : Int) -> Unit

    GCounter::value

    source
    fn GCounter::value(self : GCounter) -> Int

    LwwRegister

    source
    pub(all) struct LwwRegister {
      value : String
      timestamp : Int
      replica : Int
    } derive(Eq, 
    Debug    )

    LwwRegister::assign

    source
    fn LwwRegister::assign(self : LwwRegister, value : String, timestamp : Int, replica : Int) -> LwwRegister

    LwwRegister::merge

    source
    fn LwwRegister::merge(self : LwwRegister, other : LwwRegister) -> LwwRegister

    LwwRegister::new

    source
    fn LwwRegister::new(value : String, timestamp : Int, replica : Int) -> LwwRegister

    LwwRegister::to_json

    source
    fn LwwRegister::to_json(self : LwwRegister) -> String

    ORSet

    source
    pub(all) struct ORSet {
      entries : Array[ORSetEntry]
    } derive(
    Debug    )

    ORSet::add

    source
    fn ORSet::add(self : ORSet, element : String, dot : Int) -> Unit

    Compatibility helper for single-writer sets. Multi-replica callers should use add_from.

    ORSet::add_from

    source
    fn ORSet::add_from(self : ORSet, element : String, replica : Int, dot : Int) -> Unit

    Adds with an explicit replica-scoped sequence number. The pair (replica, dot) is the identity of an observed add.

    ORSet::contains

    source
    fn ORSet::contains(self : ORSet, element : String) -> Bool

    ORSet::merge

    source
    fn ORSet::merge(self : ORSet, other : ORSet) -> ORSet

    ORSet::new

    source
    fn ORSet::new() -> ORSet

    ORSet::remove

    source
    fn ORSet::remove(self : ORSet, element : String) -> Unit

    ORSet::size

    source
    fn ORSet::size(self : ORSet) -> Int

    ORSetEntry

    source
    pub(all) struct ORSetEntry {
      element : String
      replica : Int
      dot : Int
      removed : Bool
    } derive(Eq, 
    Debug    )

    ORSetEntry::new

    source
    fn ORSetEntry::new(element : String, replica : Int, dot : Int, removed? : Bool) -> ORSetEntry

    PNCounter

    source
    pub(all) struct PNCounter {
      positive : GCounter
      negative : GCounter
    } derive(
    Debug    )

    PNCounter::decrement

    source
    fn PNCounter::decrement(self : PNCounter, replica : Int, amount? : Int) -> Unit

    PNCounter::increment

    source
    fn PNCounter::increment(self : PNCounter, replica : Int, amount? : Int) -> Unit

    PNCounter::merge

    source
    fn PNCounter::merge(self : PNCounter, other : PNCounter) -> PNCounter

    PNCounter::new

    source
    fn PNCounter::new() -> PNCounter

    PNCounter::value

    source
    fn PNCounter::value(self : PNCounter) -> Int

    ReplicaState

    source
    pub(all) struct ReplicaState {
      replica : Int
      clock : VectorClock
      counter : PNCounter
      tasks : ORSet
      log : ChangeLog
    } derive(
    Debug    )

    A complete, mergeable replica state for offline task and counter workloads.

    ReplicaState::add_task

    source
    fn ReplicaState::add_task(self : ReplicaState, task : String) -> Unit

    Adds a task using a replica-scoped OR-Set dot, avoiding cross-replica collisions.

    ReplicaState::decrement

    source
    fn ReplicaState::decrement(self : ReplicaState, amount? : Int) -> Unit

    Records a counter decrement that can later be merged with an offline peer.

    ReplicaState::increment

    source
    fn ReplicaState::increment(self : ReplicaState, amount? : Int) -> Unit

    Records a counter increment that can later be merged with an offline peer.

    ReplicaState::merge

    source
    fn ReplicaState::merge(self : ReplicaState, other : ReplicaState) -> ReplicaState

    Merges independently mutated replica state and operation history.

    ReplicaState::new

    source
    fn ReplicaState::new(replica : Int) -> ReplicaState

    Creates an offline-first state machine whose mutations carry causal metadata.

    ReplicaState::remove_task

    source
    fn ReplicaState::remove_task(self : ReplicaState, task : String) -> Unit

    Removes all adds for a task that this replica has observed.

    ReplicaState::summary

    source
    fn ReplicaState::summary(self : ReplicaState) -> SyncSummary

    Produces an exact-state sync summary for peer exchange planning.

    SyncPlan

    source
    pub(all) struct SyncPlan {
      source : Int
      target : Int
      needs_push : Bool
      needs_pull : Bool
      source_events : Int
      target_events : Int
    } derive(Eq, 
    Debug    )

    SyncPlan::should_exchange

    source
    fn SyncPlan::should_exchange(self : SyncPlan) -> Bool

    SyncPlan::to_json

    source
    fn SyncPlan::to_json(self : SyncPlan) -> String

    SyncSummary

    source
    pub(all) struct SyncSummary {
      replica : Int
      clock : VectorClock
      counter : PNCounter
      set : ORSet
      log : ChangeLog
      clock_entries : Int
      counter_value : Int
      set_size : Int
      log_events : Int
    } derive(
    Debug    )

    SyncSummary::new

    source
    fn SyncSummary::new(replica : Int, clock : VectorClock, counter : PNCounter, set : ORSet, log : ChangeLog) -> SyncSummary

    SyncSummary::plan_with

    source
    fn SyncSummary::plan_with(self : SyncSummary, remote : SyncSummary) -> SyncPlan

    SyncSummary::to_json

    source
    fn SyncSummary::to_json(self : SyncSummary) -> String

    VectorClock

    source
    pub(all) struct VectorClock {
      entries : Array[ClockEntry]
    } derive(
    Debug    )

    VectorClock::concurrent_with

    source
    fn VectorClock::concurrent_with(self : VectorClock, other : VectorClock) -> Bool

    VectorClock::dominates

    source
    fn VectorClock::dominates(self : VectorClock, other : VectorClock) -> Bool

    VectorClock::get

    source
    fn VectorClock::get(self : VectorClock, replica : Int) -> Int

    VectorClock::merge

    source
    fn VectorClock::merge(self : VectorClock, other : VectorClock) -> VectorClock

    VectorClock::new

    source

    VectorClock::set

    source
    fn VectorClock::set(self : VectorClock, replica : Int, counter : Int) -> Unit

    VectorClock::tick

    source
    fn VectorClock::tick(self : VectorClock, replica : Int) -> Unit

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io