jonnifyechomq

Coherence · dive 02

The two
lanes

One payload, two carriers — the choice is what a lost message costs.

The 29-byte message can ride two different lanes, and the table chooses by declaring its coherence mode. The choice is not about throughput — both are cheap — it is about what a lost message costs. The broadcast lane is a single PUBLISH, applied by EchoStore.Ring: at-most-once, for surfaces where a lost message costs one TTL of staleness. The job lane is an enqueue over EchoMQ’s fair lanes, remembered by EchoStore.Journal: at-least-once, crash-surviving, for surfaces where a stale read costs money. Same truth, different promise.

One payload, two addresses The broadcast lane The job lane Pick a lane Pattern & implementation References

One payload, two addresses

The same payload(id, version) from the previous dive feeds both lanes; only the carrier differs. Each lane has its own address — a channel for broadcast, a queue for the job lane — and both share the cache’s {table} hashtag, so a broadcast channel and the L2 keys it invalidates land on the same Valkey Cluster slot.

# EchoStore.Coherence — the two lanes and their addresses
# Same 29-byte payload(id, version); two carriers; the table picks by the
# cost of a missed invalidation.
def channel(table), do: "ecc:{" <> table <> "}:coh"   # the broadcast address
def queue(table),   do: "ecc.coh." <> table           # the job-lane address

# Broadcast: fire-and-forget, one wire hop, returns the receiver count.
def broadcast(conn, table, id, version) do
  Connector.command(conn, ["PUBLISH", channel(table), payload(id, version)])
end

# Job lane: at-least-once over EchoMQ's fair lanes. The lane mints a JOB id
# and carries the same coherence payload as the job's body.
def enqueue(conn, table, group, id, version) do
  Lanes.enqueue(conn, queue(table), group, BrandedId.generate!("JOB"), payload(id, version))
end

The job-lane queue is named ecc.coh.<table> — its own queue under EchoMQ — so a coherence job is an ordinary job: claimed, retried, and deduplicated by its JOB id like any other.

One payload, two addresses · pick a table

payload(id,ver) 29 bytes channel/1 · broadcast ecc:{board}:coh queue/1 · job lane ecc.coh.board {table} hashtag

The broadcast lane — at-most-once, for cheap staleness

broadcast/4 is a single PUBLISH. There is no acknowledgement and no retry: if a subscriber is briefly disconnected when the message goes out, it never sees it — and that is acceptable precisely because a lost broadcast costs one TTL of staleness. The missed row expires on its jittered clock and re-fills with the current value. For a board, a leaderboard, a presence count — read-mostly data where a second of staleness is invisible — that is the right trade.

The applier on the receiving side is EchoStore.Ring: a bounded ring with one producer and one applier, the Disruptor’s shape on the BEAM. Two :atomics carry the head/tail sequences, an ETS table holds the preallocated slots, and occupancy = tail − head is the backpressure gauge anyone can read at any time. Wakes are edge-triggered — one :wake only on the empty→non-empty transition — so a busy burst costs one message however many items flow through it.

# EchoStore.Ring — publish/2 (single producer)
# When the ring is full the publish is refused and counted — never blocked,
# never overwritten. The broadcast lane is at-most-once by its substrate's
# contract, and a counted drop under storm keeps that contract honest where
# silent overwriting or unbounded queueing would trade it for a worse one.
def publish(name, item) do
  rt = :persistent_term.get({__MODULE__, name})
  tail = :atomics.get(rt.seq, 1)
  head = :atomics.get(rt.seq, 2)

  if tail - head >= rt.capacity do
    :counters.add(rt.counters, @counters[:dropped], 1)
    :dropped                       # full: refuse and count, do not block
  else
    next = tail + 1
    :ets.insert(rt.slots, {rem(next, rt.capacity), item})
    :atomics.put(rt.seq, 1, next)
    :counters.add(rt.counters, @counters[:published], 1)
    # one wake only on the empty -> non-empty edge
    if tail == :atomics.get(rt.seq, 2) do
      send(rt.applier, :wake)
      :counters.add(rt.counters, @counters[:wakes], 1)
    end
    :ok
  end
end

A full ring drops and counts; it does not block the table’s owner and it does not overwrite a slot another item is using. Surfaces that cannot accept a drop do not ride this lane — they ride the job lane, which does not pass through the ring at all.

The job lane — at-least-once, for staleness that costs money

enqueue/5 puts the same payload on ecc.coh.<table> over EchoMQ’s fair lanes. Now the message is a job: admitted once (the bus deduplicates the job id), claimed by a worker, retried on failure, surviving a crash. This is the lane for data where a stale read costs money — a wallet balance, an inventory count, a settlement — where “the row will expire eventually” is not good enough.

The memory behind the job lane is EchoStore.Journal, a transactional outbox standing beside the bus (the bus stays volatile by decision D-2). Its applied table records the last version applied per name and survives the node, the cache, and the bus — so even a coherence intent replayed after a crash answers correctly. The deep durability of that lane is the subject of the next dive and the Echo Persistence floor.

Choosing the lane

The table’s declared spec carries one field, coherence: :none | :broadcast | :tracking:

  • :none — no coherence lane; the cache relies on TTL expiry alone (fine for derived, short-lived data).
  • :broadcast — the PUBLISH + Ring lane: at-most-once, cheap, for read-mostly data.
  • :tracking — RESP3 server-assisted client tracking, where Valkey itself pushes the invalidation; the store opts into the server-push instead of carrying its own message.

Pick a lane by the cost of staleness

Pick a data class and read the lane the coherence: mode selects, the verb it runs, the substrate that applies it, and its delivery guarantee. The model maps each cost-of-staleness to the real lane over a fixed set of codemojex-style read surfaces.

coherence: · the cost picks the lane

payload(id,ver) 29 bytes broadcast lane PUBLISH ecc:{t}:coh EchoStore.Ring · at-most-once job lane Lanes.enqueue ecc.coh.t EchoStore.Journal · at-least-once broadcast at-most-once

Same 29-byte truth, two promises: the cheap lane lets a lost message cost one TTL; the durable lane makes the message survive a crash. The data class — not a default — picks which.

Pattern & implementation

the pattern (pub/sub invalidation vs durable queue)

An invalidation can be a fire-and-forget signal or a guaranteed-delivery job. You pick per data class by what a missed message costs — a stale board self-heals on TTL; a stale balance does not.

the implementation (Coherence + Ring + Journal)

broadcast/4 → PUBLISH → Ring (at-most-once, refuse-and-count under storm); enqueue/5 → Lanes.enqueue → Journal (at-least-once, crash-surviving). One payload/2 feeds both; the coherence: field selects.

The cheap lane and the durable lane carry the identical truth; what differs is the promise.

What the receiver does

The next dive is what lands on the far side of either lane: the one conditional drop that makes both lanes safe to deliver twice or out of order, and the journal’s durable memory that folds to Echo Persistence. The fair lanes the job lane rides are the Queue.

References

Sources

Related in this course