🥮 mooncakes.io
    Sign in

    tideweave

    Correction-aware event-time windows with atomic retractions, session splits, finality guards and deterministic replay.

    event-time
    retraction
    session-window
    streaming
    changelog
    Download zip
    Author
    Version
    0.1.0
    License
    MIT
    Repository
    Last updated
    5 hours ago
    Downloads
    2

    #TideWeave

    让迟到与撤回的数据得到完整、一致、可解释的修正。

    MoonBit 事件时间窗口与修正日志引擎。迟到事件可以合并会话，撤回桥接事件可以重新拆分会话；每次原子修正输出 before/after 变更，显式水位线保护已终结事实，下游按版本检查漏接与旧值冲突。

    在线实验室 · Mooncakes · English · 语义契约 · 自检证据

    跨平台 CI

    实验室中的迟到事件合并与变更日志

    #一个容易出错的真实问题

    同一设备先上报 at=0 和 at=16 的两条记录，gap=10，因此它们属于两个会话。设备恢复网络后补传 at=8 的记录，两个会话应合并。随后发现该记录有误并撤回，窗口必须再次拆分。只对 SUM 做加减会留下错误的边界、成员与 min/max。

    TideWeave 将每次修正视为原子状态变更：计算候选事实 → 重建受影响 key → 校验终结约束 → 提交版本 → 输出完整变更。事务若失败，窗口、事件索引、去重状态与版本一起保持原状。

    #已实现

    • 滚动、带相位偏移的滑动与会话窗口；半开区间；负窗口起点的数学 floor 对齐。
    • 按事件 ID 插入、替换、撤回，支持跨 key 移动与多项原子事务。
    • 会话迟到合并、撤回拆分，count / Int64 sum / min / max 与完整成员来源。
    • open → closing → final；保护已终结窗口及原始事件，拒绝将旧事实重新打开。
    • 成功命令幂等、冲突检测；可重放检查点恢复后继续去重。
    • 稳定顺序的 before/after 日志，以及校验版本连续性、旧值与原子应用的下游 View。
    • 严格 JSON 协议，拒绝重复解码键、未知字段、被浮点舍入掩盖的小数和不适用参数。
    • Node CLI、三个可重现用例、交互式浏览器实验室，共用 MoonBit 核心。

    #在 MoonBit 中使用

    要求编译器 0.10.16 或更新兼容版本。

    moon add LMK-ai-nb/tideweave@0.1.0

    在消费项目的 moon.pkg 加入：

    import { "LMK-ai-nb/tideweave" @tide }

    在可抛出错误的函数中：

    let cfg = @tide.config(@tide.Session(gap=10), lateness=20)
    let engine = @tide.Engine::new(cfg)
    let view = @tide.View::new()
    let outcome = engine.apply(@tide.Transact(id="batch-1", changes=[
      @tide.Put({ id: "left", key: "device-A", at: 0, value: 20 }),
      @tide.Put({ id: "right", key: "device-A", at: 16, value: 35 }),
    ]))
    view.apply(outcome)
    let restored = @tide.restore(engine.checkpoint())
    assert_true(restored.rows() == engine.rows())

    完整公开接口由 moon info 生成，见 pkg.generated.mbti。aggregate 可批处理；Engine 可持续接收容量内的修正；replay_json / resume_json 提供严格文本边界。

    #运行与验证

    moon version --all
moon check --target all --deny-warn
moon test --target all --deny-warn
moon build --target all --deny-warn
moon fmt --check
moon info
moon run cmd/demo
node scripts/build.mjs
node --test tests/cli.test.mjs
node scripts/serve.mjs

    浏览器打开 http://127.0.0.1:4174。Node 要求 22+；没有第三方 npm 依赖。native 构建还需要平台 C 工具链，Windows CI 使用 MSVC。

    node bin/tideweave.mjs run examples/order-amendment.json --checkpoint checkpoint.json
node bin/tideweave.mjs run examples/session-correction.json
node scripts/benchmark.mjs

    CLI 退出码：0 表示所有命令 accepted/duplicate；1 表示存在 rejected/conflict（演示中有意触发）；2 表示输入／IO／运行错误。恢复时传入 {schema_version:1,checkpoint:<保存的对象>,commands:[...]} 并使用 resume 子命令。完整格式见 FORMAT.md。

    可选浏览器检查：安装 Python Playwright 和 Chromium 后，在预览服务运行时执行 python tests/browser.py。CI 在 Linux 实际运行此检查。

    #质量与边界

    37 个 MoonBit 测试包括独立区间枚举 oracle、独立图连通 oracle、混合 key 状态差分、原子回滚、终结保护和检查点恢复。8 个 Node 集成测试覆盖公共协议与真实 CLI；Chromium 检查交互、导出、注入文本与 390px 手机布局。GitHub CI 在 Linux、macOS、Windows 检查 wasm、wasm-gc、js、native 四个目标。

    这是有界、同步、单进程的可嵌入状态机，适用于离线对账、边缘批次及流处理回归测试。最多 5000 活跃事件、10000 成功命令、20000 窗口、2 Mi 规范 journal 单元；不自动遗忘去重或终结事实。连续无限流需要宿主设计分区／批次生命周期，宿主还负责消息 IO、持久化、授权与外部消费一致性。这里不提供分布式运行时或外部 exactly-once 承诺。

    Windows / Node 24 合成基准：4096 事件、64 key、32 批次，包含解析、状态修正、日志、检查点和完整 JSON 输出，10 次样本中位约 84 ms。环境、范围与复现命令见 BENCHMARK.md，不将其外推为生产吞吐承诺。

    #原创性与申报

    维护者：李明坤 / LMK-ai-nb。独立新仓库与新业务实现，不复用 submit-guard、permscope 或 AgentCanary 的核心。事件时间、窗口和 changelog 是已有概念；本项目的独立贡献是修正、会话拆分、事务终结保护、下游旧值校验和重启去重的组合实现，不声称发明流处理。

    开发使用 AI 辅助；维护者须理解实现并承担验证责任。原创代码 MIT，标准库编译分发保留许可与 NOTICE，所有示例为合成数据，见 第三方说明。

    十月报名表要求一页内 Markdown 申报书，且注明“不要使用 AI 编写”。项目事实文档是参考材料，正式申报书须由李明坤理解项目后亲自撰写。技术交付不等于完成官方报名／通过审核。见 申报说明 与 三分钟演示。

    TideError

    source
    pub(all) suberror TideError {
      TideError(String)
    } derive(Eq)

    TideError::equal

    source
    fn TideError::equal(TideError, TideError) -> Bool

    TideError::not_equal

    source
    fn TideError::not_equal(x : TideError, y : TideError) -> Bool

    Command

    source
    pub(all) enum Command {
      Transact(id~ : String, changes~ : Array[Mutation])
      Advance(id~ : String, to~ : Int)
    } derive(Eq)

    impl ToJson for Command

    Command::equal

    source
    fn Command::equal(Command, Command) -> Bool

    Command::not_equal

    source
    fn Command::not_equal(x : Command, y : Command) -> Bool

    Command::to_json

    source
    fn Command::to_json(self : Command) -> Json

    Config

    source
    pub(all) struct Config {
      window : WindowSpec
      lateness : Int
      max_events : Int
      max_operations : Int
    } derive(Eq)

    impl ToJson for Config

    Config::equal

    source
    fn Config::equal(Config, Config) -> Bool

    Config::not_equal

    source
    fn Config::not_equal(x : Config, y : Config) -> Bool

    Config::to_json

    source
    fn Config::to_json(self : Config) -> Json

    Delta

    source
    pub(all) struct Delta {
      kind : String
      before : Window?
      after : Window?
    } derive(Eq, ToJson)

    Apply all deltas in an outcome atomically. Deletes precede upserts.

    Delta::equal

    source
    fn Delta::equal(Delta, Delta) -> Bool

    Delta::not_equal

    source
    fn Delta::not_equal(x : Delta, y : Delta) -> Bool

    Delta::to_json

    source
    fn Delta::to_json(Delta) -> Json

    Engine

    source
    pub struct Engine {
      cfg : Config
      records : Map[String, Event]
      materialized : Array[Window]
      clock : Int
      revision : Int
      committed : Map[String, Command]
      journal : Array[Command]
      journal_units : Int
    }

    Opaque mutable engine. Public queries and outcomes return detached arrays.

    Engine::apply

    source
    fn Engine::apply(self : Engine, command : Command) -> Outcome

    All failures leave state, version, successful IDs and journal unchanged.

    Engine::checkpoint

    source
    fn Engine::checkpoint(self : Engine) -> String

    A portable validated replay checkpoint, not a signed proof of history.

    Engine::events

    source
    fn Engine::events(self : Engine) -> Array[Event]

    Engine::new

    source
    fn Engine::new(cfg : Config) -> Engine raise TideError

    Engine::rows

    source
    fn Engine::rows(self : Engine) -> Array[Window]

    Detached, sorted materialized view. Mutating its arrays cannot alter state.

    Engine::run

    source
    fn Engine::run(engine : Engine, commands : Array[Command]) -> Replay raise TideError

    Resume on an existing engine; commands are processed in supplied order.

    Engine::version

    source
    fn Engine::version(self : Engine) -> Int

    Engine::watermark

    source
    fn Engine::watermark(self : Engine) -> Int

    Event

    source
    pub(all) struct Event {
      id : String
      key : String
      at : Int
      value : Int
    } derive(Eq, ToJson)

    Event ID is global within one engine. Keys and IDs are opaque Unicode.

    Event::equal

    source
    fn Event::equal(Event, Event) -> Bool

    Event::not_equal

    source
    fn Event::not_equal(x : Event, y : Event) -> Bool

    Event::to_json

    source
    fn Event::to_json(Event) -> Json

    Mutation

    source
    pub(all) enum Mutation {
      Put(Event)
      Remove(String)
    } derive(Eq)

    impl ToJson for Mutation

    Mutation::equal

    source
    fn Mutation::equal(Mutation, Mutation) -> Bool

    Mutation::not_equal

    source
    fn Mutation::not_equal(x : Mutation, y : Mutation) -> Bool

    Mutation::to_json

    source
    fn Mutation::to_json(self : Mutation) -> Json

    Outcome

    source
    pub(all) struct Outcome {
      id : String
      status : String
      version : Int
      watermark : Int
      deltas : Array[Delta]
      message : String
    } derive(Eq, ToJson)

    Outcome::equal

    source
    fn Outcome::equal(Outcome, Outcome) -> Bool

    Outcome::not_equal

    source
    fn Outcome::not_equal(x : Outcome, y : Outcome) -> Bool

    Outcome::to_json

    source
    fn Outcome::to_json(Outcome) -> Json

    Replay

    source
    pub(all) struct Replay {
      schema_version : Int
      config : Config
      steps : Array[Outcome]
      rows : Array[Window]
      events : Array[Event]
      watermark : Int
      version : Int
      checkpoint : Json
    } derive(ToJson)

    Replay::to_json

    source
    fn Replay::to_json(Replay) -> Json

    View

    source
    pub struct View {
      materialized : Array[Window]
      revision : Int
      clock : Int
      last_delivery : String
    }

    A downstream materialized view that rejects gaps and stale before-images.

    View::apply

    source
    fn View::apply(self : View, outcome : Outcome) -> Unit raise TideError

    Validates transport consistency; authenticity is the host's responsibility.

    View::new

    source
    fn View::new() -> View

    View::rows

    source
    fn View::rows(self : View) -> Array[Window]

    View::version

    source
    fn View::version(self : View) -> Int

    Window

    source
    pub(all) struct Window {
      key : String
      start : Int
      end : Int
      count : Int
      sum : Int64
      min : Int
      max : Int
      members : Array[String]
      phase : String
    } derive(Eq, ToJson)

    Finality is materialized so downstream consumers need no hidden clock.

    Window::equal

    source
    fn Window::equal(Window, Window) -> Bool

    Window::not_equal

    source
    fn Window::not_equal(x : Window, y : Window) -> Bool

    Window::to_json

    source
    fn Window::to_json(Window) -> Json

    WindowSpec

    source
    pub(all) enum WindowSpec {
      Tumbling(size~ : Int, offset~ : Int)
      Sliding(size~ : Int, slide~ : Int, offset~ : Int)
      Session(gap~ : Int)
    } derive(Eq)

    Window parameters use caller-defined integer time units.

    WindowSpec::equal

    source
    fn WindowSpec::equal(WindowSpec, WindowSpec) -> Bool

    WindowSpec::not_equal

    source
    fn WindowSpec::not_equal(x : WindowSpec, y : WindowSpec) -> Bool

    WindowSpec::to_json

    source
    fn WindowSpec::to_json(self : WindowSpec) -> Json

    aggregate

    source
    fn aggregate(c : Config, events : Array[Event], watermark? : Int) -> Array[Window] raise TideError

    Batch aggregation API. Validates the same bounds used by the state machine.

    assign_fixed

    source
    fn assign_fixed(spec : WindowSpec, at : Int) -> Array[(Int, Int)] raise TideError

    Return start/end pairs in ascending order, including negative starts.

    config

    source
    fn config(window : WindowSpec, lateness? : Int, max_events? : Int, max_operations? : Int) -> Config raise TideError

    decode_commands

    source
    fn decode_commands(source : String) -> Array[Command] raise TideError

    decode_config

    source
    fn decode_config(source : String) -> Config raise TideError

    replay

    source
    fn replay(c : Config, commands : Array[Command]) -> Replay raise TideError

    replay_json

    source
    fn replay_json(source : String) -> String raise TideError

    Validate all input before running; domain rejections are report steps.

    restore

    source
    fn restore(source : String) -> Engine raise TideError

    resume_json

    source
    fn resume_json(source : String) -> String raise TideError

    Resume input has a checkpoint object and an array of new commands.

    validate_event

    source
    fn validate_event(e : Event) -> Unit raise TideError

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io