🥮 mooncakes.io

moontxnkit

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

mvcc
transaction
atomic-state
workflow
wal
recovery
moon add black-duck666/moontxnkit@0.4.0
Download zip
Version
0.4.0
License
Apache-2.0
Repository
Last updated
last month
Downloads
13
README

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

© 2026 mooncakes.io