StatifierUI.Trace.Subscriber (StatifierUI v0.1.0)

Copy Markdown View Source

The only GenServer in this bead - it owns attach and detach, seq stamping, the session.start manifest emission, session-death handling, and the bounded buffer, and fans every message out to registered listeners. Everything else under lib/statifier_ui/trace/ is a pure module; this is the process that wires them to a live Statifier.Session.

The only path that sees the initialize burst - Statifier.Session.start_link/2 runs Statifier.Interpreter.initialize/2 to quiescence before returning, so a subscriber added afterward has already missed the opening EntrySet/ContentExecuted/InvokePass/MacrostepStable (GAP 6 / st-uqo4). Passing this subscriber's pid in :subscribers at Statifier.Session.start_link/2 time is the only way around that:

{:ok, sub} = StatifierUI.Trace.Subscriber.start_link(machine: machine, source: source)

{:ok, session} =
  Statifier.Session.start_link(machine, trace: true, subscribers: [sub], session_id: "sess_1")

:ok = StatifierUI.Trace.Subscriber.attach(sub, session, subscribe: false)

The session already has sub in its monitored subscriber set from :subscribers, so attach/3 with subscribe: false only takes the monitor - it never calls Statifier.Session.subscribe/2.

The late attach path

attach(sub, session) (default subscribe: true) calls Statifier.Session.subscribe/2 itself, in addition to monitoring. A :late_attach diagnostic is recorded in stats/1's diagnostics list - the initialize burst is gone by the time this call can run, because Statifier.Session.start_link/2 already returned.

The catch-up attach path

attach(sub, session, catch_up: true) closes the late-attach gap for a session started with record: true (statifier ADR-0049): the subscription and the recording snapshot happen in the same session handle_call, the missed prefix is Statifier.Replay.run/1's stream, and this subscriber folds that prefix into its buffer inside its own attach call - before any live suffix from its mailbox is processed. Prefix and suffix are one uniform stream with no overlap, no gap, and no dedup key (the ADR's mid-run invariant; trust the seam). The session id is read from the recording's resolved :session_id option, so the session.start manifest is emitted as seq: 0 ahead of the prefix.

On a session started without record: true the session answers {:error, :not_recorded} and does not subscribe, so this subscriber falls back to Statifier.Session.subscribe/2 and records a :not_recorded diagnostic in stats/1 - the stream is live-only and says so; it is never silently presented as whole. A Statifier.Replay.run/1 failure records :catch_up_failed the same way (the live subscription from the catch-up call itself is already in place in that case).

Session id discovery

This subscriber never asks the session for its id (decision 9 of the plan): no GenServer.call from this process into the session it is subscribed to, because the session runs initialize/2 inside its own init/1 and a call issued while the caller's start_link chain is still unwinding can block on a process that is itself blocked (GAP 5 / st-xbaz). The session id instead arrives on the first {:statifier, session_id, _} message this process receives. At that moment the session.start manifest is built and emitted as seq: 0, and the triggering message is normalized immediately after as seq: 1. If StatifierUI.Trace.Manifest.build/3 fails (a bad :fixtures value, say), no manifest is emitted, the failure is recorded as a diagnostic in stats/1, and normalization continues - losing the index tables is bad, losing the whole trace is worse.

Halting is not end-of-stream

{:halted, reason} becomes a session.halted message and changes nothing else: this subscriber does not stop, does not unsubscribe, and does not mark the stream complete. GAP 4 / st-r6l9 shows trace.* effects arriving after {:halted, :done}, and dropping them is the failure this note exists to prevent. Only the session process actually exiting - observed as this subscriber's own monitor :DOWN - produces session.terminated and moves stats/1's status to :terminated; the buffer is kept and stays readable either way.

Summary

Types

The snapshot stats/1 returns.

Functions

Registers pid to receive {:statifier_ui, session_id, %Message{}} for every subsequent message.

Attaches this subscriber to session, monitoring it for termination.

Returns a specification to start this module under a supervisor.

Detaches this subscriber from its session: unsubscribes, demonitors with [:flush], and sets stats/1's status to :detached. The buffer is kept and stays readable.

Returns every %Message{} currently held, oldest first.

Removes pid from the listener set. A no-op if it was never registered.

Starts a subscriber.

Returns this subscriber's current stats() snapshot.

Types

server()

@type server() :: GenServer.server()

stats()

@type stats() :: %{
  session: String.t() | nil,
  status: :detached | :attached | :terminated,
  seq: non_neg_integer(),
  buffered: non_neg_integer(),
  dropped: non_neg_integer(),
  errors: non_neg_integer(),
  foreign: non_neg_integer(),
  diagnostics: [StatifierUI.Fixtures.diagnostic()]
}

The snapshot stats/1 returns.

Functions

add_listener(server, pid)

@spec add_listener(server(), pid()) :: :ok

Registers pid to receive {:statifier_ui, session_id, %Message{}} for every subsequent message.

attach(server, session, opts \\ [])

@spec attach(server(), session :: pid(), opts :: keyword()) :: :ok

Attaches this subscriber to session, monitoring it for termination.

opts[:subscribe] (default true) selects the late path - Statifier.Session.subscribe/2 is called and a :late_attach diagnostic is recorded. Pass subscribe: false for the recommended early path, where session already carries this subscriber's pid in its own :subscribers start option.

opts[:catch_up] (default false) selects the catch-up path described in the moduledoc - Statifier.Session.subscribe/3 with catch_up: true, the replayed prefix folded in before this call returns. When set, opts[:subscribe] is ignored: catch-up decides its own subscription.

Idempotent about the monitor: a second attach/3 call for the same session pid does not stack a second monitor. It does not attempt to detect an existing subscription, which is why the two paths are distinguished by the explicit :subscribe flag rather than inferred.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

detach(server)

@spec detach(server()) :: :ok

Detaches this subscriber from its session: unsubscribes, demonitors with [:flush], and sets stats/1's status to :detached. The buffer is kept and stays readable.

messages(server)

@spec messages(server()) :: [StatifierUI.Trace.Message.t()]

Returns every %Message{} currently held, oldest first.

remove_listener(server, pid)

@spec remove_listener(server(), pid()) :: :ok

Removes pid from the listener set. A no-op if it was never registered.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts a subscriber.

opts:

  • :machine (required) - the compiled %Statifier.Machine{} the session.start manifest is built from.
  • :source - the SCXML text, for session.start's source field. Optional, since %Statifier.Machine{} does not retain it.
  • :fixtures - the decoded sidecar JSON object, for session.start's fixtures field. Optional.
  • :parent_session, :invokeid - invoke-tree origin fields, forwarded to session.start when supplied. Optional.
  • :capacity - the bounded buffer's capacity, default 1000.
  • :listeners - pids receiving {:statifier_ui, session_id, %Message{}} as messages arrive. Default [].
  • :name - passed to GenServer.start_link/3 unchanged.

stats(server)

@spec stats(server()) :: stats()

Returns this subscriber's current stats() snapshot.