Std.CRDT

View source →

Conflict-free Replicated Data Types for Cure (v0.27.0).

Every CRDT in this module satisfies three laws that make it safe for eventual-consistency replication:

  1. merge(a, b) == merge(b, a) (commutative)
  2. merge(a, merge(b, c)) == merge(merge(a, b), c) (associative)
  3. merge(a, a) == a (idempotent)

The laws are asserted in the companion Std.Proof obligations emitted by lib/std/crdt.cure when re-checked; the runtime implementation in :cure_std_crdt is hand-verified against them.

Types included:

  • GCounter -- grow-only counter, increment and value.
  • PNCounter -- positive/negative counter, supports decrement.
  • ORSet(T) -- observed-remove set, add / remove / value.
  • LWWRegister(T) -- last-write-wins register.
  • MVRegister(T) -- multi-value register (set of concurrent writes).

All CRDTs carry a node_id to tag mutations so merges can reconcile conflicting updates deterministically.

Examples

use Std.CRDT

# Two replicas tick concurrently; merge converges without loss.
let a0 = g_increment(g_empty(), :n1, 3)
let b0 = g_increment(g_empty(), :n2, 2)
g_value(g_merge(a0, b0))                # => 5

use Std.CRDT

# Add-then-remove is commutative under concurrent updates.
let s0 = or_add(or_empty(), :n1, "x")
let s1 = or_add(s0, :n1, "y")
or_value(or_remove(s1, "x"))            # => ["y"]

Functions

  • # fn __group__() -> Atom

    Group tag consumed by Cure.Stdlib.Preload.

  • # fn g_empty() -> GCounter extern

    Empty GCounter.

  • # fn g_increment(c: GCounter, node: Atom, by: Int) -> GCounter extern

    Increment c by by on replica node (positive numbers only).

  • # fn g_merge(a: GCounter, b: GCounter) -> GCounter extern

    Merge two GCounters by taking the per-replica max.

  • # fn g_value(c: GCounter) -> Int extern

    Current value of the GCounter: sum of per-replica counts.

  • # fn lww_empty(node: Atom) -> LWWRegister extern

    Empty LWWRegister bound to node.

  • # fn lww_merge(a: LWWRegister, b: LWWRegister) -> LWWRegister extern

    Merge by selecting the side with the higher timestamp; ties break on node id.

  • # fn lww_set(r: LWWRegister, value: T, stamp: Int, node: Atom) -> LWWRegister extern

    Write value with timestamp stamp from replica node.

  • # fn lww_value(r: LWWRegister) -> T extern

    Current value of the register.

  • # fn mv_empty() -> MVRegister extern

    Empty MVRegister.

  • # fn mv_merge(a: MVRegister, b: MVRegister) -> MVRegister extern

    Merge by taking the union of non-dominated writes.

  • # fn mv_values(r: MVRegister) -> List(T) extern

    List of currently-live values.

  • # fn mv_write(r: MVRegister, value: T, stamp: Int, node: Atom) -> MVRegister extern

    Write value on replica node with timestamp stamp; concurrent writes are retained.

  • # fn or_add(s: ORSet, node: Atom, elem: T) -> ORSet extern

    Add elem to the set on replica node; tags the addition with a fresh unique identifier so it survives a concurrent remove.

  • # fn or_empty() -> ORSet extern

    Empty ORSet.

  • # fn or_merge(a: ORSet, b: ORSet) -> ORSet extern

    Merge two ORSets via tag union.

  • # fn or_remove(s: ORSet, elem: T) -> ORSet extern

    Remove elem from the set; only removes tags this replica has observed, so concurrent re-adds from other replicas are retained.

  • # fn or_value(s: ORSet) -> List(T) extern

    List of current elements.

  • # fn pn_decrement(c: PNCounter, node: Atom, by: Int) -> PNCounter extern

    Decrement by by on replica node.

  • # fn pn_empty() -> PNCounter extern

    Empty PNCounter.

  • # fn pn_increment(c: PNCounter, node: Atom, by: Int) -> PNCounter extern

    Increment by by on replica node.

  • # fn pn_merge(a: PNCounter, b: PNCounter) -> PNCounter extern

    Merge two PNCounters component-wise.

  • # fn pn_value(c: PNCounter) -> Int extern

    Current value: sum(p) - sum(n).