AMQP consumer that feeds broker deliveries into the PhiAccrual core detector.

Opens an AMQP connection, opens a channel, subscribes to one configured queue, and on every delivery calls PhiAccrual.observe(detector_key, receipt_ts) where receipt_ts comes from :erlang.monotonic_time(:millisecond) at the instant the delivery is processed — that is, when it is dequeued from this GenServer's mailbox, not when it arrived there. Under a backlog the two differ by the queueing delay. The detector key is extracted from the envelope by PhiAccrualAmqp.Envelope.extract/2.

Clock discipline (read this)

phi_accrual's estimator works on local monotonic time only. The publisher's BasicProperties.timestamp and any broker-stamped header (e.g., from the rabbitmq_message_timestamp plugin) are cross-process wall clocks; using them to feed the EWMA breaks the detector. This module passes them through as diagnostic telemetry metadata but never as the value handed to observe/2.

Liveness caveat (read this too)

In AMQP, "delivery received" proves three things are alive in combination: publisher, broker, and the network paths between them and you. A high phi value does NOT pin the fault on the publisher. If you need publisher-only liveness, choose a transport with no intermediary (e.g., phi_accrual_udp).

Options

• :queue (required) — the queue to consume from. Must already exist; this package declares no topology.
• :url — broker URL, default "amqp://localhost".
• :connection_opts — a URL or a keyword list passed to AMQP.Connection.open/1,2. Takes precedence over :url: when set, :url is ignored entirely. A keyword list is merged over the connection defaults, so anything given there wins; a binary is a URL, leaving the defaults in force.

• :key_resolver — (meta -> PhiAccrual.detector_key() | nil), default Envelope.default_key_resolver/1.

• :reconnect_min_ms — backoff floor, default 1000.
• :reconnect_max_ms — backoff ceiling, default 30_000. Must not be below :reconnect_min_ms.
• :max_tracked_keys — cap on remembered detector keys, default 1000.
• :name — registers the process. Omitted, the consumer runs unnamed so several can coexist in one supervision tree.
• :connect — whether to connect on start, default true. Setting it to false starts a consumer that opens no connection; it exists so the delivery and lifecycle paths can be exercised without a broker.

Unknown keys, mistyped values and a :reconnect_min_ms above :reconnect_max_ms raise ArgumentError from start_link/1.

Mapping deliveries to detector keys

See PhiAccrualAmqp.Envelope for the resolver contract. The default extracts meta.routing_key. For static N-queues-per-node topologies pass a constant resolver per consumer:

Consumer.start_link(
  queue: "heartbeats.node_a",
  key_resolver: fn _meta -> :node_a end
)

Connection lifecycle

The consumer manages its own connection, channel, and subscription. On startup it schedules an async :connect so the supervisor can come up before the broker is reachable. On any failure — broker unreachable, channel error, server-initiated basic.cancel, connection or channel process death — the consumer tears down what it has and reconnects with jittered exponential backoff between :reconnect_min_ms and :reconnect_max_ms. The ceiling doubles per attempt; the delay is drawn uniformly between the floor and that ceiling, so consumers attached to a restarting broker spread their retries instead of arriving in lockstep.

This deliberately differs from PhiAccrualUdp.Listener's fail-fast socket open: an AMQP connection is a remote-broker contract that can blip during normal operation, while a UDP socket open is a local syscall that essentially never fails after success.

Telemetry

[:phi_accrual_amqp, :connection, :up]
  measurements: %{system_time}
  metadata:     %{queue}

[:phi_accrual_amqp, :connection, :down]
  measurements: %{tracked}
  metadata:     %{queue, reason, keys}
  # also fires on a server-initiated cancel, with
  # reason: :server_cancelled, since that path tears the
  # connection down too
  # tracked counts what :keys lists. The list is for policy; the
  # count is for a gauge, since Telemetry.Metrics cannot aggregate
  # a list.
  # keys lists the detector keys this consumer was feeding when
  # delivery stopped. φ for those keys will climb while the
  # transport is down; a policy layer can use this to decide the
  # excursion is not evidence about the entities themselves.

[:phi_accrual_amqp, :keys, :evicted]
  measurements: %{tracked}
  metadata:     %{queue, key, incoming_key, max_tracked_keys}
  # emitted when the tracked-key set is at :max_tracked_keys and a
  # new key displaces the least recently seen one.

[:phi_accrual_amqp, :consumer, :registered]
  measurements: %{system_time}
  metadata:     %{queue, consumer_tag}

[:phi_accrual_amqp, :consumer, :cancelled]
  measurements: %{system_time}
  metadata:     %{queue, consumer_tag, reason}

[:phi_accrual_amqp, :sample, :received]
  measurements: %{monotonic_time, system_time}
  # monotonic_time is the exact value handed to
  # PhiAccrual.observe/2, so a handler can derive inter-arrival
  # intervals directly instead of reconstructing them — and the
  # clock-discipline promise becomes inspectable rather than
  # merely documented.
  metadata:     %{detector_key, envelope_timestamp, routing_key, exchange, queue}
  # envelope_timestamp may be nil if the publisher didn't set one;
  # it is NEVER what gets passed to PhiAccrual.observe/2.

[:phi_accrual_amqp, :extract, :error]
  measurements: %{system_time}
  metadata:     %{reason, routing_key, exchange, queue}
  # reason ∈ [:no_detector_key, :resolver_raised]

The :sample, :received event name is shared with phi_accrual_udp, but the payload shape differs (identity key detector_key vs node; timestamp in metadata vs measurements). Handlers are not cross-transport drop-in.