BeamConsole.Collector (beam_console v0.3.0)

Owns a shared, bounded, non-overlapping runtime sampling loop.

The collector remains idle without subscribers in the default lazy mode, runs scans outside its own mailbox, retains the last completed snapshot, and emits coalesced version invalidations to each monitored subscriber. Explicit always-recording mode keeps the same bounded scan loop active without viewers.

Summary

Types

changes_result()

state()

t()

Functions

acknowledge(sequence, server \\ __MODULE__, timeout \\ 5000)

Acknowledges the caller's outstanding snapshot version and releases its newest pending version.

changes_since(sequence, server \\ __MODULE__)

Returns the current bounded diff when it directly follows sequence, otherwise requests resync.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

current_diff(server \\ __MODULE__)

Returns the bounded diff produced by the most recent completed scan.

latest_snapshot(server \\ __MODULE__, timeout \\ 5000)

Returns the most recent completed snapshot.

refresh(server \\ __MODULE__)

Requests a scan, coalescing the request when a scan is already running.

request_reconciliation(server \\ __MODULE__, timeout \\ 5000)

Requests an acknowledged internal reconciliation scan without operator rate limiting.

request_refresh(server \\ __MODULE__, timeout \\ 5000)

Requests an operator scan while enforcing the configured refresh cooldown.

start_link(options)

Starts a collector with optional runtime, interval, and limit overrides.

status(server \\ __MODULE__, timeout \\ 5000)

Returns bounded collector health and freshness information.

subscribe(server \\ __MODULE__, timeout \\ 5000)

Subscribes the caller and starts sampling when it is the first subscriber.

unsubscribe(server \\ __MODULE__)

Unsubscribes the caller and stops scheduled sampling when no subscribers remain.

Types

changes_result()

@type changes_result() ::
  {:ok, [BeamConsole.Diff.t()]} | {:resync, BeamConsole.Snapshot.t() | nil}

state()

@type state() :: t()

t()

@type t() :: %BeamConsole.Collector{
  always_record?: boolean(),
  diff: BeamConsole.Diff.t() | nil,
  diff_limit: non_neg_integer(),
  interval: non_neg_integer(),
  last_error: BeamConsole.ReasonSummary.t() | nil,
  last_failure_at: DateTime.t() | nil,
  last_operator_refresh_ms: integer() | nil,
  lifecycle_recorder: GenServer.server() | nil,
  monotonic_clock: (-> integer()),
  name: GenServer.name() | nil,
  pending_refresh?: boolean(),
  recorder_epoch: String.t(),
  recorder_gap?: boolean(),
  refresh_cooldown: non_neg_integer(),
  runtime: module(),
  runtime_options: keyword(),
  scan: Task.t() | nil,
  scan_timeout: non_neg_integer(),
  scan_timeout_ref: reference() | nil,
  sequence: non_neg_integer(),
  snapshot: BeamConsole.Snapshot.t() | nil,
  subscribers: %{required(pid()) => BeamConsole.Collector.Subscriber.t()},
  task_supervisor: Supervisor.supervisor(),
  tick_ref: {reference(), reference()} | nil
}

Functions

acknowledge(sequence, server \\ MODULE, timeout \\ 5000)

@spec acknowledge(non_neg_integer(), GenServer.server(), timeout()) :: :ok

Acknowledges the caller's outstanding snapshot version and releases its newest pending version.

changes_since(sequence, server \\ MODULE)

@spec changes_since(non_neg_integer(), GenServer.server()) :: changes_result()

Returns the current bounded diff when it directly follows sequence, otherwise requests resync.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

current_diff(server \\ MODULE)

@spec current_diff(GenServer.server()) :: BeamConsole.Diff.t() | nil

Returns the bounded diff produced by the most recent completed scan.

latest_snapshot(server \\ MODULE, timeout \\ 5000)

@spec latest_snapshot(GenServer.server(), timeout()) :: BeamConsole.Snapshot.t() | nil

Returns the most recent completed snapshot.

refresh(server \\ MODULE)

@spec refresh(GenServer.server()) :: :ok

Requests a scan, coalescing the request when a scan is already running.

request_reconciliation(server \\ MODULE, timeout \\ 5000)

@spec request_reconciliation(GenServer.server(), timeout()) :: :ok

Requests an acknowledged internal reconciliation scan without operator rate limiting.

request_refresh(server \\ MODULE, timeout \\ 5000)

@spec request_refresh(GenServer.server(), timeout()) :: :ok | {:error, :rate_limited}

Requests an operator scan while enforcing the configured refresh cooldown.

start_link(options)

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

Starts a collector with optional runtime, interval, and limit overrides.

status(server \\ MODULE, timeout \\ 5000)

@spec status(GenServer.server(), timeout()) :: BeamConsole.Collector.Status.t()

Returns bounded collector health and freshness information.

unsubscribe(server \\ MODULE)

@spec unsubscribe(GenServer.server()) :: :ok

Unsubscribes the caller and stops scheduled sampling when no subscribers remain.