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.

Example

import gleam/json
import lattice_presence/presence_state as state

let a = state.new("node-a")
  |> state.join("pid-1", "room:lobby", "alice", json.object([]))
let b = state.new("node-b")
  |> state.join("pid-2", "room:lobby", "bob", json.object([]))
let 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

Replica

</>

Unique identifier for a node 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

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

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

If context[replica] + 1 is in the cloud, advance context and remove from cloud. Repeat until no more compaction possible.

compacted_clocks

</>

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

Get the compacted vector clock.

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.

get_by_key

</>

pub fn get_by_key(
  state: State,
  topic: String,
  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,
  pid: String,
  topic: String,
  key: String,
  meta: json.Json,
) -> State

Add a tracked presence. Increments the local clock.

leave

</>

pub fn leave(
  state: State,
  pid: String,
  topic: String,
  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) -> State

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.

merge_with_diff

</>

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

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

new

</>

pub fn new(replica: String) -> State

Create a new empty state for this replica

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 and context for a downed replica

replica

</>

pub fn replica(state: State) -> String

Get the current vector clock

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.

Example

Constructors

Constructors

Arguments

Constructors

Why not reuse lattice_core/version_vector or dot_context?

Constructors

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