Backpressure

Delivery and acknowledgements

Fanout does not call process_info/2 on each recipient. Each session has a counter Wiregrid maintains, and each durable send inserts a reservation {session_id, delivery_id} before the message is placed in the owner mailbox.

The envelope

receive do
  {:"$wiregrid", envelope} ->
    envelope.instance
    envelope.session_id
    envelope.delivery_id
    envelope.topic
    envelope.class
    envelope.event
    Wiregrid.ack(envelope.instance, envelope.session_id, envelope.delivery_id)
end

The message is always the two-tuple {:"$wiregrid", envelope}. The map always has instance, session_id, delivery_id, topic, and class (:durable or :ephemeral). context is present only when the subscription or fanout supplied a non-empty context map. delivery_format chooses the body:

• :term adds event
• :encoded adds payload (codec bytes)
• :both adds event and payload

Wiregrid.decode_payload/2 turns an encoded payload back into a term for the instance’s codec. Wiregrid.Chat.ack/2 needs session_id and delivery_id on that map.

Soft queue and hard queue

Defaults are soft_queue: 500 and hard_queue: 2_000. soft_queue must be strictly below hard_queue or startup returns :soft_queue_must_be_below_hard_queue.

When the pending count already exceeds the hard queue, Wiregrid rolls the counter back and may isolate the owner. slow_consumer_action is :disconnect by default, or :exit_owner. Capacity for reservations is also globally capped by max_delivery_reservations (100,000 / 2,000,000 / 20,000,000 on the three profiles). A full table rejects the send before the mailbox grows.

Acknowledgements

{:ok, pending} = Wiregrid.ack(:chat, session_id, delivery_id)
{:ok, %{acked: n, unknown: missing, pending: left}} =
  Wiregrid.ack_many(:chat, session_id, ids)
{:ok, pending} = Wiregrid.ack(:chat, session_id)
count = Wiregrid.pending(:chat, session_id)

ack/3 takes one reservation and returns the new pending count, or {:error, :unknown_delivery}. ack_many/3 accepts up to max_batch_items ids and reports how many were released. Unknown ids are listed; they are not an error for the whole call. ack/2 acknowledges one of the session’s reservations, the first key in the table, and returns {:ok, 0} when none remain. Protocol adapters should call ack/3 with the id from the envelope so the client and the server agree on which message was committed. pending/2 returns 0 for an invalid session id or a missing instance.

Do the application work, then ack. Acknowledging before the write commits tells Wiregrid the consumer is healthy while the data is still in flight. Grouped helpers consume_delivery/3, consume_many_deliveries/3, consume_ordered_deliveries/3, and consume_projection/5 run a handler and acknowledge only what the handler committed. Failed items stay reserved.

Replay

{:ok, stats} = Wiregrid.replay_session(:chat, session_id, {:channel, "general"}, limit: 50)

replay_session/4 reads one bounded storage page and redelivers it through authorization and the same pressure rules. The returned resume cursor moves only past rows delivery accepted. A row that was refused for pressure is not skipped. Stats include resume_cursor, next_cursor, delivered, dropped, and stopped. Cowboy exposes this as the command {:replay, stream, opts}, and the router binds it to the socket’s session. The client cannot name another session id.

Request and reply

{:ok, _} = Wiregrid.request_session(:chat, from_session, to_session, event)
context = Wiregrid.request_context(envelope)
{:ok, _} = Wiregrid.reply(:chat, to_session, envelope, response)

Request and reply are ordinary deliveries with a context the callee can read. They use the same reservation model. There is no hidden side channel with different reliability.