brazier/store/crdt

Add-wins last-writer-wins map: the delta-state CRDT behind replicated stores. A port of the AWLWWMap of delta_crdt_ex (DeltaCrdt.AWLWWMap), itself the “add-wins observed-remove map” of Almeida, Shoker and Baquero, “Delta State Replicated Data Types” (2018), with a timestamp on every value.

Every value written gets a fresh dot (replica, counter). The state is the live dots of every key plus a causal context: the set of every dot this replica has ever observed, live or removed. The context is a version vector (clock: replica -> highest counter, meaning counters 1..n were all seen) plus a cloud of dots seen out of order.

A write never sends the whole state, only a delta: a state fragment with the new dots and, in its context, the dots the write replaced or removed. Joining two states keeps a dot when

• both sides have it, or
• one side has it and the other side’s context does not contain it (the other side never saw it, so it cannot have removed it).

So a removal only removes the dots it observed, and a concurrent write on another node survives it (add wins). Two concurrent writes of one key both survive as dots; reads pick the value with the greatest (timestamp, dot) (last writer wins), the same on every node, and the next write of the key replaces both.

Join is commutative, associative and idempotent, so replicas converge whatever order, duplication or batching deltas and full states arrive in.

Every state carries a digest of itself for anti-entropy, kept up to date by every join instead of recomputed (see digest).

Floors: a context bounded across restarts

A restarted store server is a new replica (a new incarnation of its node), so without more every restart would add a clock entry that no replica could ever drop. The context therefore also holds floors: floors[node] = i covers every dot of every incarnation of node below i, so the clock entries and cloud dots of those incarnations are dropped. Only an incarnation raises its own node’s floor, to itself (raise_floor), and only once every replica it can reach holds the same state, so every dot of an earlier incarnation any of them holds is in its own state: a live one stays live (a floor covers a dot, it does not remove it), a removed one stays removed. The context then holds at most one incarnation per node name that is still compacting, plus the incarnations that started since a node’s last raise.

The cost is a narrow loss: a dot of an earlier incarnation that only a replica out of reach held (partitioned, or a node whose replica had not yet joined), or that was still in flight, is covered without having been seen, so it is dropped wherever it arrives. The earlier incarnation itself is gone, and its writes are fire and forget, so such a write was already at risk; docs/CLUSTERING.md says when it happens.

Against delta_crdt_ex

The join of dot sets is delta_crdt_ex’s AWLWWMap.join_dot_sets/4 (the formula on join_keys); the differences are:

• The context is exact: delta_crdt_ex’s Dots.compress folds a dot set into one counter per replica, which also covers counters below it that it never saw. Here a dot beyond a gap stays in the cloud until the gap fills, and only a contiguous run moves into the clock.
• Values are ordered by a hybrid logical clock (brazier/store/hlc) and ties break on the dot, a total order, so every node reads the same value. delta_crdt_ex stamps a value with the writer’s System.monotonic_time, which is not comparable across nodes, and reads the greatest with Enum.max_by, with no tie-break.
• Floors (above): delta_crdt_ex never shrinks its context, and its node id is a fresh :rand.uniform(1_000_000_000) on every start, so its context grows with every restart (and two starts can draw one id).
• Anti-entropy compares one incremental digest (see digest) and a mismatch pulls the whole state; delta_crdt_ex walks a Merkle tree down to the differing keys and sends only those, one sync in flight per neighbour until it is acknowledged.

pub type Change(value) {
  Added(key: String, dot: Dot, entry: Entry(value))
  Removed(key: String, dot: Dot, entry: Entry(value))
}

pub type Context {
  Context(
    clock: dict.Dict(Replica, Int),
    cloud: set.Set(Dot),
    floors: dict.Dict(String, Int),
  )
}

pub type Delta(value) {
  Delta(
    entries: dict.Dict(String, dict.Dict(Dot, Entry(value))),
    context: Context,
    topics: List(String),
  )
}

pub type Dot {
  Dot(replica: Replica, counter: Int)
}

pub type Dots(value) =
  dict.Dict(Dot, Entry(value))

pub type Entry(value) {
  Entry(value: value, ts: hlc.Timestamp, topics: List(String))
}

pub type Replica {
  Replica(node: String, incarnation: Int)
}

pub type State(value) {
  State(
    entries: dict.Dict(String, dict.Dict(Dot, Entry(value))),
    context: Context,
    size: Int,
    digest: Int,
  )
}

pub fn append(
  state: State(value),
  replica: Replica,
  key: String,
  items: List(#(value, hlc.Timestamp)),
  topics: List(String),
) -> #(State(value), Delta(value), List(Change(value)))

pub fn compare_dots(a: Dot, b: Dot) -> order.Order

pub fn context_size(state: State(value)) -> Int

pub fn covers(context: Context, dot: Dot) -> Bool

pub fn delta_size(delta: Delta(value)) -> Int

pub fn digest(state: State(value)) -> Int

pub fn floor(state: State(value), node: String) -> Int

pub fn has_older_incarnation(
  state: State(value),
  replica: Replica,
) -> Bool

pub fn join(
  state: State(value),
  delta: Delta(value),
) -> #(State(value), List(Change(value)))

pub fn join_full(
  state: State(value),
  full: Delta(value),
) -> #(State(value), List(Change(value)))

pub fn key_dots(
  state: State(value),
  key: String,
) -> dict.Dict(Dot, Entry(value))

pub fn max_timestamp(
  delta: Delta(value),
) -> Result(hlc.Timestamp, Nil)

pub fn new() -> State(value)

pub fn put(
  state: State(value),
  replica: Replica,
  key: String,
  value: value,
  ts: hlc.Timestamp,
  topics: List(String),
) -> #(State(value), Delta(value), List(Change(value)))

pub fn raise_floor(
  state: State(value),
  replica: Replica,
) -> Result(#(State(value), Delta(value)), Nil)

pub fn recompute_digest(state: State(value)) -> Int

pub fn remove(
  state: State(value),
  key: String,
  topics: List(String),
) -> #(State(value), Delta(value), List(Change(value)))

pub fn replica_to_string(replica: Replica) -> String

pub fn size(state: State(value)) -> Int

pub fn to_delta(state: State(value)) -> Delta(value)

pub fn value(
  state: State(value),
  key: String,
) -> Result(value, Nil)

pub fn values(state: State(value), key: String) -> List(value)