ForgeOpsTracker.DeliveryQueue (forge_ops_tracker v0.15.0)

Copy Markdown View Source

A small bounded queue, processed one event at a time, so delivery never blocks the caller that raised the error. The BEAM equivalent of the Ruby/Python/Java clients' own background-thread- backed queue: except here the "queue manager" is an ordinary supervised GenServer (started once, see ForgeOpsTracker.Application, not spun up lazily on first push, since BEAM processes are cheap enough that there's no meaningful cost to keeping one alive idle) and each actual delivery runs in its own short-lived Task, not inline in the GenServer's own callback.

That split matters, confirmed directly (not a hypothetical): a GenServer that calls a slow, blocking operation (:httpc.request, here) directly from inside handle_info cannot process any other message (including a push from another process, or even :sys.get_state) until that call returns, because a GenServer only ever handles one message at a time. Doing the actual HTTP call in a separate Task instead keeps this GenServer responsive to new pushes the whole time a delivery is in flight, while the :DOWN monitor message below still enforces the same one-delivery-at-a-time ordering every other client's own queue has.

Bounded via Configuration.queue_size (:queue has no size limit of its own, so this tracks the length itself): a burst of exceptions must never apply backpressure to the host app by growing this queue without limit.

Nothing reported just before the VM stops is lost, mirroring the Ruby gem's DeliveryQueue#drain. On a normal supervised shutdown (Application.stop/1, System.stop/0, a release stopping), terminate/2 delivers whatever is still queued and waits for the delivery already under way, within drain_timeout/0; this needs Process.flag(:trap_exit, true), like MetricBuffer and PerformanceFlusher, and a shutdown timeout longer than the drain. A mix run script, an elixir script or an escript usually ends by halting the VM without stopping applications, so the first push also registers a System.at_exit/1 callback (once per VM) that calls flush/1. A VM that never reports anything never waits.

Summary

Functions

Returns a specification to start this module under a supervisor.

The longest a drain at shutdown, or flush/0 with no argument, waits, in milliseconds.

Waits until everything queued so far has been delivered (or given up on), for at most timeout milliseconds. Returns :ok, or {:error, :timeout} if something was still waiting when the time ran out. Returns :ok straight away when nothing is queued, or when the client isn't running.

Enqueues payload for delivery. Always returns :ok: a full queue drops the payload (logged via Configuration.logger) rather than raising or blocking the caller.

Enqueues one explicit change, delivered to the changes endpoint; shares this queue's bound.

Enqueues the startup change snapshot, delivered to the change snapshots endpoint.

Enqueues one masked EXPLAIN plan, delivered to the query plans endpoint.

Enqueues a finished trace, delivered to the spans endpoint; shares this queue's bound.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

drain_timeout()

@spec drain_timeout() :: pos_integer()

The longest a drain at shutdown, or flush/0 with no argument, waits, in milliseconds.

flush(timeout \\ 5000)

@spec flush(timeout()) :: :ok | {:error, :timeout}

Waits until everything queued so far has been delivered (or given up on), for at most timeout milliseconds. Returns :ok, or {:error, :timeout} if something was still waiting when the time ran out. Returns :ok straight away when nothing is queued, or when the client isn't running.

push(payload)

@spec push(map()) :: :ok

Enqueues payload for delivery. Always returns :ok: a full queue drops the payload (logged via Configuration.logger) rather than raising or blocking the caller.

push_change(change)

@spec push_change(map()) :: :ok

Enqueues one explicit change, delivered to the changes endpoint; shares this queue's bound.

push_change_snapshot(snapshot)

@spec push_change_snapshot(map()) :: :ok

Enqueues the startup change snapshot, delivered to the change snapshots endpoint.

push_query_plan(plan)

@spec push_query_plan(map()) :: :ok

Enqueues one masked EXPLAIN plan, delivered to the query plans endpoint.

push_trace(trace)

@spec push_trace(map()) :: :ok

Enqueues a finished trace, delivered to the spans endpoint; shares this queue's bound.