Sagents.PresenceWriter (Sagents v0.17.1)

Copy Markdown

Makes presence writes on behalf of agent processes, so an agent never waits on Phoenix.Tracker.

Every Phoenix.Tracker write (track, update, untrack) is a GenServer.call to a tracker shard with a 5 second timeout, and list is one too. A shard falls behind under load, and all agents share one shard for the "agent_server:presence" discovery topic, because the tracker shards by topic. An agent that made those calls itself would stall for 5 seconds per status change while the shard is backed up, and the timeout exit would terminate it.

Agents instead hand their writes to this process with a cast, and it makes the tracker call for them. The tracker accepts any pid as the tracked process and links to it, so an entry is still removed when its agent exits.

Semantics

  • Desired state, not commands. put/5 says "this entry should exist with exactly this metadata", and remove/4 says "this entry should not exist". A put/5 updates the entry, and tracks it if the tracker does not have it ({:error, :nopresence}). An entry lost to a failed call is therefore recreated by the agent's next status change.
  • Coalesced. Each entry (presence module, topic, key, pid) holds at most one pending write, and a newer write replaces an older one that has not been applied yet. While the tracker is slow, the backlog is bounded by the number of entries, not by the number of status changes.
  • Ordered per entry. Writes for one entry are applied in the order they were made.
  • Never fatal. A write that exits or raises is logged and dropped. A put/5 for a local pid that is no longer alive is skipped.

Supervision

Sagents.Supervisor starts this process before anything that hosts agents. Children stop in reverse start order, so it stops after the agents, and it applies the writes still pending when it stops, which includes the remove/4 each agent makes on an orderly shutdown. A host that starts its Phoenix.Presence before Sagents.Supervisor (the usual order) keeps the tracker available for that.

When this process is not running, as in a test that starts an Sagents.AgentServer without Sagents.Supervisor, a write is applied directly in the caller, with the same error handling.

Summary

Functions

Apply every pending write before returning.

Ensure pid is present on topic under key with exactly meta.

Ensure pid is not present on topic under key.

Start the writer. Sagents.Supervisor starts it; hosts do not.

Types

entry()

@type entry() ::
  {presence_module :: module(), topic :: String.t(), key :: String.t(), pid()}

Functions

flush()

@spec flush() :: :ok

Apply every pending write before returning.

For tests that need to observe a write: a write cast before this call is applied by the time it returns. Returns :ok when the writer is not running.

put(presence_module, pid, topic, key, meta)

@spec put(module(), pid(), String.t(), String.t(), map()) :: :ok

Ensure pid is present on topic under key with exactly meta.

Returns immediately.

remove(presence_module, pid, topic, key)

@spec remove(module(), pid(), String.t(), String.t()) :: :ok

Ensure pid is not present on topic under key.

Returns immediately.

start_link(opts \\ [])

Start the writer. Sagents.Supervisor starts it; hosts do not.