lattice_presence/presence_state

Presence State - Pure CRDT for distributed presence tracking

A causal-context add-wins observed-remove set, inspired by Phoenix.Tracker.State. This module is a pure data structure with no actors or side effects.

Each node (replica) tracks its own presences authoritatively. State is replicated by extracting deltas and merging them at remote replicas. Conflicts are resolved causally: adds win over concurrent removes. Replica identities must be unique per process incarnation. Use new_incarnation when a stable node name can restart.

Example

import gleam/json
import lattice_presence/presence_state as state

let a = state.new_incarnation("node-a")
  |> state.join("pid-1", "room:lobby", "alice", json.object([]))
let b = state.new_incarnation("node-b")
  |> state.join("pid-2", "room:lobby", "bob", json.object([]))
let assert Ok(merged) = state.merge(a, b)
state.get_by_topic(merged, "room:lobby")
// -> [#("pid-1", "alice", _), #("pid-2", "bob", _)]

Types

Clock

</>

Monotonically increasing counter per replica

pub type Clock =
  Int

Diff

</>

A diff representing changes between two states

pub type Diff {
  Diff(
    joins: dict.Dict(String, List(#(String, String, json.Json))),
    leaves: dict.Dict(String, List(#(String, String, json.Json))),
  )
}

Constructors

Diff(
  joins: dict.Dict(String, List(#(String, String, json.Json))),
  leaves: dict.Dict(String, List(#(String, String, json.Json))),
)

Entry

</>

A tracked presence entry

pub type Entry {
  Entry(topic: String, key: String, pid: String, meta: json.Json)
}

Constructors

```
Entry(topic: String, key: String, pid: String, meta: json.Json)
```
Arguments

pid

Unique identifier for the tracked entity (e.g., socket ID, user ID)

meta

Arbitrary metadata

MergeError

</>

Error returned when replicated data conflicts with the local replica identity.

This includes divergent states claiming the same name and unseen local-owned causal history echoed by another replica. It indicates a stale state after a restart or multiple live nodes configured with the same replica name. Assign every live node a unique identity and discard stale state before retrying.

pub type MergeError {
  SameReplica(replica: String)
}

Constructors

```
SameReplica(replica: String)
```

Replica

</>

Unique identifier for a running node incarnation in the cluster

pub type Replica =
  String

ReplicaStatus

</>

Replica status

pub type ReplicaStatus {
  Up
  Down
}

Constructors

```
Up
```
```
Down
```

State

opaque </>

The CRDT state.

Why not reuse `lattice_core/version_vector` or `dot_context`?

The causal context here is the Phoenix.Tracker-style pair of (context, clouds): a compacted vector-clock prefix plus per-replica sets of observed-but-not-yet-contiguous clocks. merge and compact rely on the gap-tracking that clouds provides — that is what makes the add-wins observed-remove semantics work with a constant-size header in the common case.

lattice_core/version_vector is a plain Dict(ReplicaId, Int) with no gap tracking, and lattice_core/dot_context stores every observed dot individually (no compaction). Neither captures the invariant “every clock <= context[replica] has been observed AND any clock listed in clouds[replica] has been observed”, which tag_is_in, compact, and next_clock all depend on. Adopting either type would either lose information or change the on-the-wire shape; reuse is possible only after extending lattice_core with a compacted variant, which is intentionally deferred.

pub opaque type State

SupersedeError

</>

Error returned when superseding would retire this state’s local writer.

Create a fresh local state after a restart instead of changing the identity of an existing writer. The error includes the local and selected identities.

Examples

let local = new_incarnation("node-a")
let current = new_incarnation("node-a")
let assert Error(CannotSupersedeLocalReplica(local_replica, current_replica)) =
  supersede(local, replica(current))

pub type SupersedeError {
  CannotSupersedeLocalReplica(
    local_replica: String,
    current_replica: String,
  )
}

Constructors

CannotSupersedeLocalReplica(
  local_replica: String,
  current_replica: String,
)

Tag

</>

A tag uniquely identifies when and where an entry was created

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

Constructors

```
Tag(replica: String, clock: Int)
```

Values

base_replica

</>

pub fn base_replica(replica: String) -> String

Get the stable replica name from an incarnation identity.

Replica values not created by new_incarnation are returned unchanged.

Examples

let state = new_incarnation("node-a")
base_replica(replica(state))
// -> "node-a"

cloud_count

</>

pub fn cloud_count(state: State) -> Int

Return the number of uncompacted cloud entries retained by the state.

compact

</>

pub fn compact(state: State) -> State

Compact clouds into context where possible

Remove cloud clocks already covered by context, then advance context through the remaining contiguous prefix.

compacted_clocks

</>

pub fn compacted_clocks(state: State) -> dict.Dict(String, Int)

Get the compacted vector clock.

decoder

</>

pub fn decoder() -> decode.Decoder(State)

Decode replicated state, including when embedded in a sync envelope.

Local replica liveness is reset, as with from_json.

Examples

let envelope_decoder = {
  use state <- decode.field("state", decoder())
  decode.success(state)
}
let payload = json.object([#("state", to_json(new("node-a")))])
json.parse(json.to_string(payload), envelope_decoder)
// -> Ok(new("node-a"))

entry_count

</>

pub fn entry_count(state: State) -> Int

Return the number of entries retained by the CRDT state.

extract_full_state

</>

pub fn extract_full_state(state: State) -> State

Extract state for sending to a remote replica.

Currently returns the full local state. Remote’s merge handles deduplication of entries it already has, and absence of an entry combined with coverage in context represents an observed removal.

A future delta-extraction variant will use the remote’s known context to filter to only the tags the remote hasn’t seen — that will be exposed as a separate function rather than retrofitted onto this one.

from_json

</>

pub fn from_json(
  json_string: String,
) -> Result(State, json.DecodeError)

Decode a JSON string into a state with only its own replica marked Up.

The serialized replica identity is retained. Merge a remote snapshot into the local state before making local edits.

Examples

new("node-a") |> to_json_string |> from_json
// -> Ok(new("node-a"))

get_by_key

</>

pub fn get_by_key(
  state state: State,
  topic topic: String,
  key key: String,
) -> List(#(String, json.Json))

Get presences for a specific key within a topic

get_by_topic

</>

pub fn get_by_topic(
  state: State,
  topic: String,
) -> List(#(String, String, json.Json))

Get all presences for a topic (from non-down replicas)

join

</>

pub fn join(
  state state: State,
  pid pid: String,
  topic topic: String,
  key key: String,
  meta meta: json.Json,
) -> State

Add a tracked presence. Increments the local clock.

leave

</>

pub fn leave(
  state state: State,
  pid pid: String,
  topic topic: String,
  key key: String,
) -> State

Remove a specific presence by pid, topic, and key.

Only entries owned by this replica are removable — leaving a foreign replica’s entry would not be causally observed (this node’s context doesn’t cover the foreign tag), so it would silently reappear on the next merge. Foreign entries are filtered out at the source instead.

leave_by_pid

</>

pub fn leave_by_pid(state: State, pid: String) -> State

Remove all presences for a pid owned by this replica.

As with leave, only locally-owned entries are eligible — see that function’s docs for the rationale.

merge

</>

pub fn merge(
  local: State,
  remote: State,
) -> Result(State, MergeError)

Merge remote state into local state.

replicas (per-node liveness view) is not merged because it is local-only view state, not part of the replicated CRDT payload.

Returns Error(SameReplica(...)) when the states claim the same replica name but their replicated data differs, or when remote carries local-owned tags or causal history that local has not observed, even via another peer. History for removed entries is checked too. Echoes of already-known local tags remain valid, and identical same-replica states are an idempotent no-op. This check does not replace the requirement for unique incarnation identities.

Examples

let assert Ok(merged) = merge(new("node-a"), new("node-b"))

merge_with_diff

</>

pub fn merge_with_diff(
  local: State,
  remote: State,
) -> Result(#(State, Diff), MergeError)

Merge remote state into local state and return a diff of what changed.

Returns Error(SameReplica(...)) under the same conditions as merge. Values owned by an earlier incarnation of the local state’s stable replica are not admitted. Their causal context is still merged so syncing the restarted state back to peers removes any cached entries from that earlier incarnation.

Examples

let assert Ok(#(merged, diff)) =
  merge_with_diff(new("node-a"), new("node-b"))

new

</>

pub fn new(replica: String) -> State

Create a new empty state for a globally unique replica incarnation.

Reusing replica after a process or node restart is unsafe because peers may retain causal history for the previous incarnation. Use new_incarnation when the same stable replica name can restart.

Examples

let state = new("node-a-019d449c-2c82-71bb-b4bf-6505df7ad7c2")
replica(state)
// -> "node-a-019d449c-2c82-71bb-b4bf-6505df7ad7c2"

new_incarnation

</>

pub fn new_incarnation(base: String) -> State

Create a new empty state with a fresh incarnation of a stable replica name.

The generated replica identity is safe to use in the existing string-valued replication and JSON formats. Use base_replica to recover base.

Examples

let state = new_incarnation("node-a")
base_replica(replica(state))
// -> "node-a"

online_list

</>

pub fn online_list(
  state: State,
) -> List(#(String, String, String, json.Json))

List all online presences across all topics (from non-down replicas)

remove_down_replica

</>

pub fn remove_down_replica(
  state: State,
  replica: String,
) -> State

Permanently remove all entries for a downed replica.

The replica’s causal high-water mark is retained so entries held by a lagging peer cannot be re-admitted later. If the replica is not marked Down, the state is returned unchanged.

replica

</>

pub fn replica(state: State) -> String

Get the replica name this state was created with.

replica_down

</>

pub fn replica_down(
  state: State,
  replica: String,
) -> #(State, Diff)

Mark a replica as down. Returns entries that are now invisible (leaves).

Idempotent: if the replica is already Down, the state is unchanged and the returned diff is empty.

replica_up

</>

pub fn replica_up(
  state: State,
  replica: String,
) -> #(State, Diff)

Mark a replica as up. Returns entries that are now visible again (joins).

Idempotent: if the replica is already Up (or unknown — unknown replicas are assumed up), the state is unchanged and the returned diff is empty.

same_base

</>

pub fn same_base(first: String, second: String) -> Bool

Return whether two replica identities share the same stable name.

Examples

let first = new_incarnation("node-a")
let second = new_incarnation("node-a")
same_base(replica(first), replica(second))
// -> True

supersede

</>

pub fn supersede(
  state: State,
  current_replica: String,
) -> Result(#(State, Diff), SupersedeError)

Retire locally known alternatives to a caller-selected replica identity.

The caller must choose the authoritative current_replica, for example through cluster membership. An inbound sync alone does not establish that authority; UUIDs and message arrival order do not rank incarnations.

Returns Error(CannotSupersedeLocalReplica(...)) before any work if the selected identity differs from the local writer but shares its base. This applies even when the local writer has no entries, is Down, or has no liveness entry. Selecting the local identity itself or an identity of another base is valid.

On Ok(#(state, diff)), other known identities with the same base have been marked Down and pruned using remove_down_replica. Leaves are combined by topic as #(key, pid, meta) tuples, preserving duplicates without an ordering guarantee. Joins are empty, and already-Down identities emit no new leaves. The selected identity and unrelated bases are unchanged: the selected identity need not be known and is neither inserted nor marked Up.

Pruning retains the existing context/cloud high-water marks, not clocks from uncovered value tags. Covered stale tags cannot return, but unseen higher tags can: this is not a permanent ban on an identity. Repeating the call without intervening changes leaves the state unchanged with no diff.

Examples

let old = new_incarnation("node-a")
  |> join("pid-1", "lobby", "alice", json.null())
let current = new_incarnation("node-a")
let assert Ok(peer) = merge(new("observer"), old)
let assert Ok(#(peer, diff)) = supersede(peer, replica(current))
dict.get(diff.leaves, "lobby")
// -> Ok([#("alice", "pid-1", json.null())])

to_json

</>

pub fn to_json(state: State) -> json.Json

Encode replicated state to JSON, omitting local replica liveness.

Examples

new("node-a") |> to_json |> json.to_string
// -> "{\"replica\":\"node-a\",\"context\":{},\"clouds\":{},\"values\":[]}"

to_json_string

</>

pub fn to_json_string(state: State) -> String

Encode replicated state to a JSON string.

Examples

new("node-a") |> to_json_string
// -> "{\"replica\":\"node-a\",\"context\":{},\"clouds\":{},\"values\":[]}"

Example

Constructors

Constructors

Arguments

Constructors

Constructors

Why not reuse lattice_core/version_vector or dot_context?

Examples

Constructors

Constructors

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Why not reuse `lattice_core/version_vector` or `dot_context`?