jonnifyechomq

The events channel · dive 03

Fire-and-forget

The channel makes no delivery promise — and that is the contract, not a defect.

A publish with no live subscriber is gone, and so is one issued in the gap between a socket drop and the reconnect. The channel is at-most-once. Two mitigations answer the two losses: the connector's resubscribe MapSet narrows the reconnect window, and the durable receipt is the stream. Its control-plane twin is EchoMQ.Cancel — a cooperative cancel with no wire identity. Step the cancel handshake on the right, then read the code.

At-most-once The two mitigations The cancel sibling Delivery model Pattern & implementation References

The cooperative cancel · step the handshake

step 1 new/0 → token = make_ref() (no wire identity) step 2 cancel/3 → send {:emq_cancel, token, reason} step 3 check/1 → receive after 0 (at a safe point) step 4 ^token match → {:cancelled, reason}; else continue a handler that never checks completes normally

At-most-once, stated plainly

PUBLISH delivers to whoever is subscribed at that instant. There is no buffer, no replay, no acknowledgement. So an event is delivered at most once to each subscriber, and possibly zero times:

No live subscriber. If nothing is subscribed to emq:{q}:events when the publish lands, the message is dropped. publish/5 reads {:ok, 0} from the wire and reports :ok — zero receivers is success, because the channel never promised one.

The reconnect window. A listener's connection can drop and reconnect. In the gap between the drop and the connector re-issuing its SUBSCRIBE, the channel is momentarily unsubscribed, and any publish in that window is lost to that listener.

The moduledoc states this directly rather than papering over it: the push channel is fire-and-forget (at-most-once); a publish with no live subscriber, or one in the window between a socket drop and the resubscribe, is lost.

No buffer, no replay, no ack. The events channel broadcasts the present and keeps nothing — the guarantee is "at most once," and zero is a legal count.

The two mitigations — one for each loss

The loss has two causes, and each has its own answer.

For the reconnect window — the resubscribe MapSet. The connector remembers every channel it is subscribed to in a MapSet and re-issues SUBSCRIBE for all of them on reconnect. That shrinks the unsubscribed window to the reconnect latency: the feed comes back automatically, and a long-lived listener never has to re-subscribe by hand. It narrows the window; it does not replay what was missed inside it.

For "I need the event to survive" — the stream log. When an event must not be lost — replayed by a reader that was not running yet, audited, consumed at-least-once — you do not harden the channel. You write to the stream log (EchoMQ.Stream): an append-only, durable, replayable log. The events channel is the reflex; the stream is the memory. The same branded id threads both — an event names a JOB; a stream record IS an EVT.

So the design splits one concern in two: a fast, lossy broadcast for "react now," and a durable log for "remember." Trying to make the broadcast durable would rebuild the log badly; the channel stays simple on purpose.

Narrow the window with the resubscribe; keep the memory in the log. Two losses, two answers — and the broadcast stays simple.

The control-plane sibling — EchoMQ.Cancel

The events channel is a fire-and-forget signal about work that happened. Its mirror on the control plane is EchoMQ.Cancel: a fire-and-forget signal asking work to stop. Same posture — best-effort, no acknowledgement, cooperative — but it carries no wire identity at all. The token is a plain make_ref(), and cancellation is a process message to the worker's mailbox.

# echo_mq — EchoMQ.Cancel (the worker-side cooperative cancel; NO wire identity)
# A token is a local make_ref(); cancellation is a process MESSAGE, not a wire op.
# Cooperative: a handler that never checks check/1 completes normally.
@type t :: reference()

# mint a token — a unique reference identifying this cancellation
def new, do: make_ref()

# flag it cancelled: send {:emq_cancel, token, reason} to the worker's mailbox.
# best-effort, fire-and-forget — answers :ok regardless of what the worker does.
def cancel(pid, token, reason \ nil) when is_pid(pid) do
  send(pid, {:emq_cancel, token, reason})
  :ok
end

# non-blocking check (O(1), `receive after 0`): is a cancel for THIS token waiting?
# the ^token match ensures a handler only catches its OWN cancellation.
def check(token) do
  receive do
    {:emq_cancel, ^token, reason} -> {:cancelled, reason}
  after
    0 -> :ok
  end
end

The mechanics make the word "cooperative" precise:

new/0 mints a make_ref() — globally unique within the node, never minted from the wire. cancel/3 sends {:emq_cancel, token, reason} to a pid's mailbox and answers :ok. It does not stop anything; it asks. Best-effort, like a publish. check/1 is a non-blocking receive after 0: it returns {:cancelled, reason} if a cancel for this token is already in the mailbox, else :ok. The ^token pin means a handler only ever catches its own cancellation. check!/1 is the same check that raises EchoMQ.Cancel.Cancelled instead, for a checkpoint-style handler that aborts by exception.

The defining property: a handler that never calls check/1 completes normally. Cancellation is something the worker chooses to honor at a safe point, between units of work — never a forced kill mid-transaction. That is why it is "cooperative," and why it needs no wire identity: both ends are processes on the same node, and the token is no more than a shared reference.

(This is the worker-side primitive only. A cancel issued from another node and coordinated across the cluster is a different, larger surface, beyond this module.)

A cancel that asks, never forces. A make_ref() token, a mailbox message, a check at a safe point — and a handler that ignores it runs to completion.

The at-most-once delivery model

Set the number of live subscribers and whether the publish lands in the reconnect window, then publish and read the outcome: the receiver count, who got the message, and — when it is lost — which mitigation applies. This is the channel's real contract over a fixed publish; the right column names the answer for that loss.

Publish once · at-most-once outcome

2
PUBLISH receivers · outcome ·

A lost message is never an error to publish/5 — it returns :ok on a zero receiver count. Durability is not the channel's job; it is the stream's.

Pattern & implementation

the pattern (Redis Patterns Applied)

Fire-and-forget messaging trades delivery guarantees for simplicity and speed; durability, when needed, is a separate, deliberate mechanism — an append-only log. R5 · Streams & Events teaches both halves.

the implementation (echo_mq)

EchoMQ.Events is at-most-once on emq:{q}:events (the resubscribe MapSet narrows the window; the stream is the durable receipt), and EchoMQ.Cancel is its control-plane twin — a cooperative, no-wire-identity make_ref() signal honored at a checkpoint.

Fire-and-forget is a choice, not a defect. Keep the broadcast and the cancel simple; put durability in the log that is built for it.

Recap — lossy on purpose

The events channel is at-most-once: a publish with no live subscriber, or one in the reconnect window, is lost. The resubscribe MapSet narrows the window; the stream log is the durable receipt. EchoMQ.Cancel is the control-plane sibling — a worker-side cooperative cancel with no wire identity: new/0 mints a make_ref(), cancel/3 sends {:emq_cancel, token, reason} to the worker's mailbox, check/1 is a non-blocking check a handler honors at a safe point, and a handler that never checks completes normally. Next, the pillar moves from the reflex to the memory: the stream log.

Next in the pillar: The stream log — the append-only, replayable receipt the events channel keeps pointing at, ordered by the branded EVT mint.

References

Where the at-most-once guarantee and the cooperative cancel are grounded, and the durable answer to the loss.

Sources

Related in this course