🥮 mooncakes.io

    moontxnkit

    Deterministic atomic state transitions, MVCC transactions and logical recovery for MoonBit.

    mvcc
    transaction
    atomic-state
    workflow
    wal
    recovery
    Download zip
    Version
    0.4.0
    License
    Apache-2.0
    Repository
    Last updated
    3 months ago
    Downloads
    47

    No dependencies

    #MoonTxnKit

    MoonTxnKit 是面向 MoonBit 的确定性 MVCC 状态事务与恢复内核，适用于工作流、 规则引擎、模拟器、内存服务和测试替身。

    #Serializable 前缀扫描

    let engine = @moontxnkit.Engine::new()
    let transaction = engine.begin(
      isolation=@moontxnkit.IsolationLevel::Serializable,
    )

    let rows = transaction.scan_prefix("task:queued:")
    ignore(transaction.put("audit:scan", "\{rows.length()}"))

    match transaction.commit() {
      @moontxnkit.CommitResult::CommittedAt(version) =>
        println("committed at \{version}")
      @moontxnkit.CommitResult::Rejected(conflict) =>
        println(conflict.to_json())
    }

    扫描结果来自事务快照，并叠加事务自己的新增、修改和删除，最终按键稳定排序。 Serializable 提交会检测前缀内的并发新增、更新和删除，返回结构化 PredicateWrite 冲突。

    #其他能力

    • Snapshot Isolation 和点读 Serializable 校验；
    • 声明式原子状态计划；
    • 保存点和局部回滚；
    • 逻辑 WAL、校验、幂等恢复；
    • 历史读取、低水位压缩和 JSON 报告。

    完整说明见仓库 README.md 与 docs/。

    AtomicPlan

    source
    pub struct AtomicPlan {
      name : String
      isolation : IsolationLevel
      conditions : Array[PlanCondition]
      writes : Array[WriteIntent]
    }

    AtomicPlan::condition_count

    source
    fn AtomicPlan::condition_count(self : AtomicPlan) -> Int

    AtomicPlan::delete

    source
    fn AtomicPlan::delete(self : AtomicPlan, key : String) -> AtomicPlan

    AtomicPlan::expect

    source
    fn AtomicPlan::expect(self : AtomicPlan, key : String, value : String) -> AtomicPlan

    AtomicPlan::expect_missing

    source
    fn AtomicPlan::expect_missing(self : AtomicPlan, key : String) -> AtomicPlan

    AtomicPlan::name

    source
    fn AtomicPlan::name(self : AtomicPlan) -> String

    AtomicPlan::new

    source
    fn AtomicPlan::new(name : String, isolation? : IsolationLevel) -> AtomicPlan

    AtomicPlan::put

    source
    fn AtomicPlan::put(self : AtomicPlan, key : String, value : String) -> AtomicPlan

    AtomicPlan::write_count

    source
    fn AtomicPlan::write_count(self : AtomicPlan) -> Int

    CommitResult

    source
    pub(all) enum CommitResult {
      CommittedAt(Int)
      Rejected(TxnConflict)
    } derive(Eq, 
    Debug    )

    CommitResult::to_json

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

    CompactionReport

    source
    pub(all) struct CompactionReport {
      low_watermark : Int
      removed_versions : Int
      retained_versions : Int
      removed_keys : Int
    } derive(Eq, 
    Debug    )

    CompactionReport::to_json

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

    ConditionFailure

    source
    pub(all) struct ConditionFailure {
      plan : String
      key : String
      expected : String?
      actual : String?
      snapshot_version : Int
    } derive(Eq, 
    Debug    )

    ConditionFailure::to_json

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

    ConflictKind

    source
    pub(all) enum ConflictKind {
      WriteWrite
      ReadWrite
      PredicateWrite
      InvalidState
      InvalidSavepoint
      CorruptWal
    } derive(Eq, 
    Debug    )

    Engine

    source
    pub struct Engine {
      current_version : Int
      next_txn_id : Int
      histories : Map[String, Array[VersionedValue]]
      active_snapshots : Map[Int, Int]
      committed_transactions : Int
      aborted_transactions : Int
      wal_records : Array[WalRecord]
    }

    Engine::begin

    source
    fn Engine::begin(self : Engine, isolation? : IsolationLevel) -> Transaction

    Engine::compact

    source
    fn Engine::compact(self : Engine) -> CompactionReport

    Engine::execute

    source
    fn Engine::execute(self : Engine, plan : AtomicPlan) -> PlanResult

    Engine::from_snapshot

    source
    fn Engine::from_snapshot(snapshot : EngineSnapshot) -> Engine

    Rebuilds an engine from an adapter-owned checkpoint. The caller can inspect validate() afterwards and choose its own serialization, encryption, and distributed-locking policy without adding platform I/O to the core.

    Engine::get

    source
    fn Engine::get(self : Engine, key : String) -> String?

    Engine::history

    source
    fn Engine::history(self : Engine, key : String) -> Array[VersionedValue]

    Engine::history_json

    source
    fn Engine::history_json(self : Engine, key : String) -> String

    Engine::new

    source
    fn Engine::new() -> Engine

    Engine::oldest_active_snapshot

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

    Engine::read_at

    source
    fn Engine::read_at(self : Engine, key : String, version : Int) -> String?

    Engine::recover

    source
    fn Engine::recover(records : Array[WalRecord]) -> (Engine, ReplayReport)

    Engine::replay

    source
    fn Engine::replay(self : Engine, records : Array[WalRecord]) -> ReplayReport

    Engine::snapshot

    source
    fn Engine::snapshot(self : Engine) -> EngineSnapshot

    Captures committed MVCC state for an adapter-owned checkpoint. Active transactions are deliberately excluded: callers must abort or finish them before persisting, then start new transactions against the restored engine.

    Engine::stats

    source
    fn Engine::stats(self : Engine) -> EngineStats

    Engine::validate

    source
    fn Engine::validate(self : Engine) -> Array[ValidationIssue]

    Engine::version

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

    Engine::wal

    source
    fn Engine::wal(self : Engine) -> Array[WalRecord]

    EngineSnapshot

    source
    pub(all) struct EngineSnapshot {
      current_version : Int
      next_txn_id : Int
      committed_transactions : Int
      aborted_transactions : Int
      histories : Array[SnapshotHistory]
      wal_records : Array[WalRecord]
    } derive(Eq, 
    Debug    )

    Deterministic committed-state checkpoint. It contains no active transaction handles and therefore has a clear recovery boundary for storage adapters.

    EngineStats

    source
    pub(all) struct EngineStats {
      current_version : Int
      keys : Int
      versions : Int
      active_transactions : Int
      committed_transactions : Int
      aborted_transactions : Int
      wal_records : Int
    } derive(Eq, 
    Debug    )

    EngineStats::to_json

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

    IsolationLevel

    source
    pub(all) enum IsolationLevel {
      Snapshot
      Serializable
    } derive(Eq, 
    Debug    )

    KeyValue

    source
    pub(all) struct KeyValue {
      key : String
      value : String
    } derive(Eq, 
    Debug    )

    PlanCondition

    source
    pub(all) struct PlanCondition {
      key : String
      expected : String?
    } derive(Eq, 
    Debug    )

    PlanResult

    source
    pub(all) enum PlanResult {
      AppliedAt(Int)
      ConditionFailed(ConditionFailure)
      CommitRejected(TxnConflict)
    } derive(Eq, 
    Debug    )

    PlanResult::to_json

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

    PrefixObservation

    source
    pub(all) struct PrefixObservation {
      prefix : String
    } derive(Eq, 
    Debug    )

    ReadObservation

    source
    pub(all) struct ReadObservation {
      key : String
      observed_version : Int
    } derive(Eq, 
    Debug    )

    ReplayReport

    source
    pub(all) struct ReplayReport {
      applied_records : Int
      skipped_records : Int
      last_version : Int
      valid : Bool
      error : String
    } derive(Eq, 
    Debug    )

    ReplayReport::to_json

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

    Savepoint

    source
    pub(all) struct Savepoint {
      name : String
      write_count : Int
      read_count : Int
      writes : Array[WriteIntent]
      reads : Array[ReadObservation]
      prefix_reads : Array[PrefixObservation]
    } derive(Eq, 
    Debug    )

    SnapshotHistory

    source
    pub(all) struct SnapshotHistory {
      key : String
      versions : Array[VersionedValue]
    } derive(Eq, 
    Debug    )

    A serializable history entry used by EngineSnapshot.

    Transaction

    source
    pub struct Transaction {
      engine : Engine
      id : Int
      snapshot_version : Int
      isolation : IsolationLevel
      state : TxnState
      writes : Array[WriteIntent]
      reads : Array[ReadObservation]
      prefix_reads : Array[PrefixObservation]
      savepoints : Array[Savepoint]
    }

    Transaction::abort

    source
    fn Transaction::abort(self : Transaction) -> Bool

    Transaction::commit

    source
    fn Transaction::commit(self : Transaction) -> CommitResult

    Transaction::delete

    source
    fn Transaction::delete(self : Transaction, key : String) -> Bool

    Transaction::get

    source
    fn Transaction::get(self : Transaction, key : String) -> String?

    Transaction::id

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

    Transaction::pending_writes

    source
    fn Transaction::pending_writes(self : Transaction) -> Int

    Transaction::put

    source
    fn Transaction::put(self : Transaction, key : String, value : String) -> Bool

    Transaction::release_savepoint

    source
    fn Transaction::release_savepoint(self : Transaction, name : String) -> Bool

    Transaction::rollback_to

    source
    fn Transaction::rollback_to(self : Transaction, name : String) -> Bool

    Transaction::savepoint

    source
    fn Transaction::savepoint(self : Transaction, name : String) -> Bool

    Transaction::savepoint_count

    source
    fn Transaction::savepoint_count(self : Transaction) -> Int

    Transaction::scan_prefix

    source
    fn Transaction::scan_prefix(self : Transaction, prefix : String) -> Array[KeyValue]

    Returns a stable, sorted prefix view at the transaction snapshot.

    Pending writes are overlaid so a transaction observes its own inserts, updates, and deletes. Serializable transactions validate the whole prefix at commit time to detect phantoms.

    Transaction::snapshot

    source
    fn Transaction::snapshot(self : Transaction) -> Int

    Transaction::status

    source
    fn Transaction::status(self : Transaction) -> TxnState

    TxnConflict

    source
    pub(all) struct TxnConflict {
      kind : ConflictKind
      key : String
      snapshot_version : Int
      current_version : Int
      message : String
    } derive(Eq, 
    Debug    )

    TxnConflict::to_json

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

    TxnState

    source
    pub(all) enum TxnState {
      Active
      Committed
      Aborted
    } derive(Eq, 
    Debug    )

    ValidationIssue

    source
    pub(all) struct ValidationIssue {
      code : String
      key : String
      message : String
    } derive(Eq, 
    Debug    )

    ValidationIssue::to_json

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

    VersionedValue

    source
    pub(all) struct VersionedValue {
      version : Int
      value : String?
    } derive(Eq, 
    Debug    )

    VersionedValue::to_json

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

    WalRecord

    source
    pub(all) struct WalRecord {
      transaction_id : Int
      snapshot_version : Int
      commit_version : Int
      operations : Array[WriteIntent]
      checksum : Int
    } derive(Eq, 
    Debug    )

    WalRecord::is_valid

    source
    fn WalRecord::is_valid(self : WalRecord) -> Bool

    WalRecord::new

    source
    fn WalRecord::new(transaction_id : Int, snapshot_version : Int, commit_version : Int, operations : Array[WriteIntent]) -> WalRecord

    WalRecord::to_json

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

    WriteIntent

    source
    pub(all) struct WriteIntent {
      key : String
      value : String?
    } derive(Eq, 
    Debug    )

    WriteIntent::delete

    source
    fn WriteIntent::delete(key : String) -> WriteIntent

    WriteIntent::put

    source
    fn WriteIntent::put(key : String, value : String) -> WriteIntent

    WriteIntent::to_json

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

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io