🥮 mooncakes.io

MoonCRDTKit

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

crdt
offline
sync
merge
collaboration
moon add ddd-1234d/MoonCRDTKit@0.2.0
Download zip
Author
Version
0.2.0
License
Apache-2.0
Repository
Last updated
28 days ago
Downloads
2
README

#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 issuePackagesBuild queueSkillsStatistics

© 2026 mooncakes.io