Presence

Presence is provided by the lattice_presence package. It tracks which pids are present for a topic and key, together with arbitrary JSON metadata.

Use it when multiple nodes need to maintain a shared view of online users, socket processes, sessions, or similar ephemeral membership without routing all joins and leaves through one coordinator.

Basic joins and reads

import gleam/json
import lattice_presence/presence_state as presence

pub fn main() {
  let state =
    presence.new_incarnation("node-a")
    |> presence.join("pid-1", "room:lobby", "alice", json.object([]))

  presence.get_by_topic(state, "room:lobby")
  // -> [#("pid-1", "alice", _)]
}

Each join creates a causal tag owned by the local replica. Queries hide entries from replicas you have marked down locally.

Process restarts

Replica identity uniqueness is per process incarnation, not just per stable node name. Create a fresh identity every time the presence process starts:

let state = presence.new_incarnation("node-a")

This preserves node-a as the stable name while giving each run a unique causal identity. On merge, a restarted state ignores cached values owned by an earlier incarnation of that stable name but retains their causal context. Syncing the restarted state back to peers therefore removes those stale entries.

Use new only when the supplied identity is already unique for the entire process incarnation and will never be reused after restart.

Merging replicas

merge returns the merged state on success:

let assert Ok(merged) = presence.merge(node_a, node_b)

Use merge_with_diff when an application needs Phoenix-style join and leave notifications while applying remote state:

let assert Ok(#(merged, diff)) = presence.merge_with_diff(node_a, node_b)

The diff groups joins and leaves by topic. It is for notifying subscribers; the merged state is still the source of truth.

Both merge functions return Error(SameReplica(...)) when divergent states use the same replica identity, or when a peer carries local-owned tags or causal history that the local state has not observed. This catches restart echoes relayed through another replica, even after the old entries have been removed. Gossip of already-known local tags remains valid, and identical same-replica states are an idempotent no-op. Discard stale restart echoes or assign each live replica a unique identity before retrying; the check does not replace the requirement for unique incarnation identities.

Replica visibility

Replica liveness is local view state, not replicated CRDT state. If a node sees a peer go down, it marks that peer down in its own state:

let #(state, diff) = presence.replica_down(state, "node-b")

Entries owned by a down replica become invisible to query functions and appear as leaves in the returned diff. If the replica comes back:

let #(state, diff) = presence.replica_up(state, "node-b")

Those entries become visible again and appear as joins. This keeps cluster liveness decisions in the embedding application instead of trying to replicate up/down status as CRDT data.

Leaving and cleanup

leave removes one local pid/topic/key entry:

let state = presence.leave(state, "pid-1", "room:lobby", "alice")

leave_by_pid removes all local entries for a pid. Both operations only remove entries owned by the local replica; foreign entries must be removed by their owning replica or hidden with replica liveness.

After your application decides that a replica will not return with useful state, mark it down and prune its entries:

let #(state, diff) = presence.replica_down(state, "node-b")
let state = presence.remove_down_replica(state, "node-b")

Use diff.leaves to notify subscribers. remove_down_replica requires Down status; otherwise it returns the state unchanged. It removes entries, sparse clouds, and local liveness, but retains the maximum observed context/cloud clock as a causal high-water mark. Stale gossip cannot restore tags covered by that mark.

Superseding a peer incarnation

Use supersede after your membership or restart protocol establishes which incarnation of a peer is current. Pass that full identity, not just its base name. The helper downs and prunes other known identities with the same base, including those already marked Down:

case presence.supersede(state, current_replica) {
  Ok(#(state, diff)) -> Ok(#(state, diff.leaves))
  Error(error) -> Error(error)
}

On success, keep the returned state and use the combined leaves for subscriber notifications. The diff groups #(key, pid, meta) entries by topic. Joins are empty, and entries already hidden by replica_down do not produce another leave. A repeated call with no new intervening entries returns the same state and an empty diff.

The selected incarnation need not be present. The helper leaves its data and liveness unchanged, so call replica_up separately if your membership protocol requires it. Replicas with unrelated bases remain unchanged.

The caller must select the current incarnation. UUIDs are random, and message arrival order does not establish restart order. Do not call supersede with each incoming sync identity: a delayed old sync could remove the current incarnation's entries.

The helper returns Error(CannotSupersedeLocalReplica(local_replica, current_replica)) if the selection would retire the local writer. It cannot change the writer identity of an existing state. For a local restart, create a fresh state with new_incarnation and merge peer snapshots into that state.

Like remove_down_replica, this helper retains causal high-water marks rather than banning future writes from a retired identity. Previously unseen higher clocks can still arrive. Your membership protocol must handle conflicting incarnation claims; supersede does not elect a current incarnation.

Serialization

Use lattice_presence/presence_state for cross-node payloads:

import lattice_presence/presence_state

let payload = presence_state.to_json_string(state)
let decoded = presence_state.from_json(payload)

The JSON format contains replicated CRDT data: replica name, causal context, clouds, and presence entries. Local replica visibility (replica_up / replica_down) is intentionally not serialized. Decoding validates causal clock values and limits nested metadata depth so malformed payloads fail as Error(_) instead of producing invalid state. Use presence_state.decoder() to include state inside an envelope decoder. The former state_json import is removed; existing wire payloads keep the same format.