brazier - v0.1.0

brazier/presence/state

The presence CRDT, ported from Phoenix.Tracker.State.

An observed-remove set of (topic, pid, key) -> meta entries with a causal context. Every local change bumps this replica’s clock and tags the entry with #(replica, clock). Removes leave no tombstone: a peer infers one when a tag it holds is covered by the sender’s context or clouds but absent from the sender’s values. Clouds hold tags seen out of order; compact folds contiguous ones into the context.

Every change is also recorded in the delta, which the shard broadcasts on its next heartbeat and then resets. A peer whose clock falls behind asks for a transfer and gets a delta generation or a snapshot (see brazier/presence/delta_generation).

This module is pure. brazier_presence_shard owns one state per shard and mirrors the visible entries into ETS for reads; the read functions here scan every value and exist for tests and introspection.

Differences from Phoenix, none of which change the merge results:

Types

Delta

</>

The changes since the last reset, covering clocks range_start to range_end (exclusive start, inclusive end). Phoenix keeps this as a State in :delta mode.

pub type Delta(r, m) {
  Delta(
    values: dict.Dict(Tag(r), #(Key, m)),
    clouds: dict.Dict(r, set.Set(Int)),
    range_start: dict.Dict(r, Int),
    range_end: dict.Dict(r, Int),
  )
}

Constructors

DeltaError

</> 
pub type DeltaError {
  NotContiguous
}

Constructors 

  • NotContiguous

    The two deltas leave a gap between their ranges.

Element

</>

One tracked entry, as returned in joins, leaves and reads.

pub type Element(r, m) {
  Element(key: Key, meta: m, tag: Tag(r))
}

Constructors 

  • Element(key: Key, meta: m, tag: Tag(r))

Key

</>

What is tracked. A pid can track several keys per topic.

pub type Key {
  Key(topic: String, pid: process.Pid, key: String)
}

Constructors

ReplicaStatus

</> 
pub type ReplicaStatus {
  Up
  Down
}

Constructors 

  • Up
  • Down

Snapshot

</>

A full state extracted for one peer.

pub type Snapshot(r, m) {
  Snapshot(
    context: dict.Dict(r, Int),
    clouds: dict.Dict(r, set.Set(Int)),
    values: dict.Dict(Tag(r), #(Key, m)),
  )
}

Constructors

State

opaque </> 
pub opaque type State(r, m)

Tag

</>

A unique event id: the replica that made the change and its clock value.

pub type Tag(r) {
  Tag(replica: r, clock: Int)
}

Constructors 

  • Tag(replica: r, clock: Int)

Transfer

</> 
pub type Transfer(r, m) {
  DeltaTransfer(Delta(r, m))
  SnapshotTransfer(Snapshot(r, m))
}

Constructors

Values

clocks

</> 
pub fn clocks(state: State(r, m)) -> #(r, dict.Dict(r, Int))

This replica and its context without down replicas. Down replicas are left out because deltas leave them out too: a peer that came up after a replica went down would otherwise ask for a full transfer forever.

clouds

</> 
pub fn clouds(state: State(r, m)) -> dict.Dict(r, set.Set(Int))

compact

</> 
pub fn compact(state: State(r, m)) -> State(r, m)

Fold cloud tags that continue the context into it, and drop those it already covers. Empty clouds are removed.

context

</> 
pub fn context(state: State(r, m)) -> dict.Dict(r, Int)

delta

</> 
pub fn delta(state: State(r, m)) -> Delta(r, m)

delta_remove_replica

</> 
pub fn delta_remove_replica(
  delta: Delta(r, m),
  replica: r,
) -> Delta(r, m)

Drop a replica from a delta’s range, clouds and values.

delta_size

</> 
pub fn delta_size(delta: Delta(r, m)) -> Int

Values plus cloud tags: what bounds a delta generation.

extract

</> 
pub fn extract(
  state: State(r, m),
  remote_ref: r,
  remote_context: dict.Dict(r, Int),
) -> Snapshot(r, m)

The full state a peer with remote_context needs: the replicas it knows, without its own values.

extract_delta

</> 
pub fn extract_delta(
  delta: Delta(r, m),
  remote_ref: r,
  remote_context: dict.Dict(r, Int),
) -> Delta(r, m)

A delta restricted the same way as extract.

get_by_key

</> 
pub fn get_by_key(
  state: State(r, m),
  topic: String,
  key: String,
) -> List(Element(r, m))

Visible entries on topic under key, from any pid.

get_by_pid

</> 
pub fn get_by_pid(
  state: State(r, m),
  pid: process.Pid,
) -> List(Element(r, m))

This replica’s entries for pid.

get_by_pid_key

</> 
pub fn get_by_pid_key(
  state: State(r, m),
  pid: process.Pid,
  topic: String,
  key: String,
) -> Result(Element(r, m), Nil)

This replica’s entry for pid under topic and key.

get_by_topic

</> 
pub fn get_by_topic(
  state: State(r, m),
  topic: String,
) -> List(Element(r, m))

Visible entries (not from down replicas) on topic.

has_delta

</> 
pub fn has_delta(state: State(r, m)) -> Bool

Whether there are changes to broadcast.

join

</> 
pub fn join(
  state: State(r, m),
  pid: process.Pid,
  topic: String,
  key: String,
  meta: m,
) -> Result(#(State(r, m), Element(r, m)), Nil)

Track key for pid on topic. Error(Nil) if already tracked.

leave

</> 
pub fn leave(
  state: State(r, m),
  pid: process.Pid,
  topic: String,
  key: String,
) -> Result(#(State(r, m), Element(r, m)), Nil)

Stop tracking key for pid on topic. Error(Nil) if not tracked.

leave_join

</> 
pub fn leave_join(
  state: State(r, m),
  pid: process.Pid,
  topic: String,
  key: String,
  meta: m,
) -> Result(#(State(r, m), Element(r, m), Element(r, m)), Nil)

Replace the meta of a tracked entry: a leave of the old tag and a join under a new one. Returns the previous and the new element. Error(Nil) if not tracked.

leave_pid

</> 
pub fn leave_pid(
  state: State(r, m),
  pid: process.Pid,
) -> #(State(r, m), List(Element(r, m)))

Stop tracking everything pid tracks.

merge

</> 
pub fn merge(
  state: State(r, m),
  transfer: Transfer(r, m),
) -> #(State(r, m), List(Element(r, m)), List(Element(r, m)))

Merge a delta or snapshot from a peer. Returns the entries that joined and left. Joins are only taken from replicas this state knows, and leaves only touch other replicas’ entries: a tag the sender has seen (in its context or clouds) but no longer holds was removed.

merge_deltas

</> 
pub fn merge_deltas(
  local: Delta(r, m),
  remote: Delta(r, m),
) -> Result(Delta(r, m), DeltaError)

Merge two deltas into one covering both ranges. Phoenix: merge_deltas.

new

</> 
pub fn new(replica: r) -> State(r, m)

online_list

</> 
pub fn online_list(state: State(r, m)) -> List(Element(r, m))

Every visible entry.

pid_entry_count

</> 
pub fn pid_entry_count(
  state: State(r, m),
  pid: process.Pid,
) -> Int

The number of this replica’s entries for pid.

remove_down_replicas

</> 
pub fn remove_down_replicas(
  state: State(r, m),
  replica: r,
) -> State(r, m)

Forget a permanently down replica: its values, context and clouds.

replica

</> 
pub fn replica(state: State(r, m)) -> r

replica_down

</> 
pub fn replica_down(
  state: State(r, m),
  replica: r,
) -> #(State(r, m), List(Element(r, m)), List(Element(r, m)))

Mark replica down. Its values are hidden and returned as leaves; they stay in the state until remove_down_replicas, so a quick recovery does not need a transfer.

replica_status

</> 
pub fn replica_status(
  state: State(r, m),
  replica: r,
) -> Result(ReplicaStatus, Nil)

replica_up

</> 
pub fn replica_up(
  state: State(r, m),
  replica: r,
) -> #(State(r, m), List(Element(r, m)), List(Element(r, m)))

Mark replica up. Its values become visible and are returned as joins.

reset_delta

</> 
pub fn reset_delta(state: State(r, m)) -> State(r, m)

Start a new empty delta at the current clock.

size

</> 
pub fn size(state: State(r, m)) -> Int

The number of values held, from every replica.

transfer_size

</> 
pub fn transfer_size(transfer: Transfer(r, m)) -> Int

The number of values a transfer carries: what max_transfer_entries bounds.

✨ Search Document