conflate

Documentation

Overview

Package conflate provides a single-producer, multi-consumer keyed latest-value fan-out bus.

A Hub hands out a singleton Sender and any number of [Receiver]s. Where github.com/amorey/gochan/watch keeps one latest-value slot and github.com/amorey/gochan/broadcast keeps a fixed ring of every value (returning ErrLagged on overflow), conflate keeps the latest value *per key*: each receiver holds one slot per key plus an insertion-ordered queue, and a Sender.Send for a key that is already pending coalesces into that slot rather than appending. A slow receiver therefore never lags-as-loss — it catches up to the latest value of every key, in first-touch order, and its memory stays bounded by the live key set rather than by write volume.

Coalescing policy is supplied by the caller as a Merge function so the bus stays domain-agnostic: Merge decides how an undelivered pending value combines with a newly sent one, and may annihilate the slot entirely (e.g. a create followed by a delete the consumer never observed).

Typical uses

Streaming resource state to watchers (Kubernetes-style informers), UI or dashboard update feeds where only the current state of each entity matters, cache invalidation fan-out, incremental index maintenance, and any "coalesce writes per entity, deliver at consumer pace" pipeline.

Semantics

Latest-value-per-key delivery. Each receiver owns an insertion-ordered queue of keys plus one value slot per key. A Send for a key with no pending slot appends the key at the back of the queue; a Send for a key that is already pending coalesces into the existing slot via Merge and leaves the key's queue position unchanged. Delivery order is therefore first-touch order, and a hot key does not starve a cold one by repeatedly jumping the queue.

Bounded by the key set, not the write volume. A thousand sends across four keys leave at most four pending entries. This is what makes conflate safe for an unbounded producer feeding a slow consumer: there is no capacity argument and no lag error because there is no unbounded backlog to grow.

Merge may annihilate. Returning keep == false drops the key entirely — queue entry and value slot both — so a create/delete pair the consumer never observed leaves no residue at all.

Send never blocks. Slow receivers cannot apply backpressure to the publisher; they simply coalesce more aggressively.

A send with no receivers is cheap. Sender.Send returns without taking the bus lock when no receiver is registered, so a hot producer does not pay for a hub nobody is watching. The result is the same one it always gave: nil, with nothing published. The corollary is a subscriber-side ordering requirement — register with Hub.Receiver before taking a snapshot of the producer's state, never after — because a value published in that gap reaches no receiver and conflate has no replay.

Per-receiver policy overrides, passed as options to Hub.Receiver. Hub.WithKeyFilter filters keys at enqueue time, so a receiver interested in one key out of a high-cardinality producer stays bounded by the keys it actually wants. Hub.WithMerge lets a single consumer coalesce by its own policy without affecting the rest of the bus — necessary when consumers of the same producer disagree about what may be dropped, since one hub-wide Merge cannot express that. The options compose.

The backlog head is observable without consuming it. Receiver.Peek returns the oldest pending event and leaves it in place, under the same closed > value precedence the popping paths use.

Sender close drains. Sender.Close lets each receiver drain its pending values once before subsequent reads report gobus.ErrClosed. Hub.Close is hard tear-down with no drain.

A single Receiver is intended for one consumer goroutine.

Index

• type Hub
  • func New[K comparable, V any](merge Merge[V]) *Hub[K, V]
  • func (h *Hub[K, V]) Close()
  • func (h *Hub[K, V]) Receiver(opts ...ReceiverOption[K, V]) *Receiver[K, V]
  • func (h *Hub[K, V]) Sender() *Sender[K, V]
  • func (h *Hub[K, V]) WithKeyFilter(keep func(K) bool) ReceiverOption[K, V]
  • func (h *Hub[K, V]) WithMerge(merge Merge[V]) ReceiverOption[K, V]
• type Merge
• type Receiver
  • func (rx *Receiver[K, V]) Chan() <-chan gobus.Event[K, V]
  • func (rx *Receiver[K, V]) Close()
  • func (rx *Receiver[K, V]) Peek() (gobus.Event[K, V], error)
  • func (rx *Receiver[K, V]) Recv() (gobus.Event[K, V], error)
  • func (rx *Receiver[K, V]) RecvContext(ctx context.Context) (gobus.Event[K, V], error)
  • func (rx *Receiver[K, V]) TryRecv() (gobus.Event[K, V], error)
• type ReceiverOption
• type Sender
  • func (tx *Sender[K, V]) Close()
  • func (tx *Sender[K, V]) Send(k K, v V) error
  • func (tx *Sender[K, V]) SendContext(ctx context.Context, k K, v V) error
  • func (tx *Sender[K, V]) TrySend(k K, v V) error

Types

type Hub

type Hub[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Hub is the construction handle for a conflate pipeline.

func New

func New[K comparable, V any](merge Merge[V]) *Hub[K, V]

New creates a hub whose receivers coalesce per key using merge. It panics if merge is nil — the coalescing policy is the whole point of the bus, so there is no implicit default.

func (*Hub[K, V]) Close

func (h *Hub[K, V]) Close()

Close is hard tear-down: the sender and every live receiver are closed immediately, with no final drain of pending values. Use Sender.Close for the soft path. Future Hub.Receiver calls return pre-closed handles. Idempotent.

func (*Hub[K, V]) Receiver

func (h *Hub[K, V]) Receiver(opts ...ReceiverOption[K, V]) *Receiver[K, V]

Receiver returns a new receiver bound to the hub, configured by opts. A receiver starts empty: it observes values sent after it was created, not the producer's history. If the hub is already closed the receiver is pre-closed and reports gobus.ErrClosed on use.

Options are minted by the hub itself — Hub.WithKeyFilter and Hub.WithMerge — and compose freely:

rx := hub.Receiver()                           // every key, hub's merge
rx := hub.Receiver(hub.WithKeyFilter(wanted))  // one key subset
rx := hub.Receiver(hub.WithMerge(stricter))    // own coalescing policy
rx := hub.Receiver(hub.WithKeyFilter(wanted), hub.WithMerge(stricter))

Later options win over earlier ones for the same setting.

func (*Hub[K, V]) Sender

func (h *Hub[K, V]) Sender() *Sender[K, V]

Sender returns the singleton send-side handle. Repeated calls return the same handle. After the hub has been closed the returned handle reports gobus.ErrClosed on use.

func (*Hub[K, V]) WithKeyFilter

func (h *Hub[K, V]) WithKeyFilter(keep func(K) bool) ReceiverOption[K, V]

WithKeyFilter returns an option restricting a receiver to keys for which keep returns true; all other keys are dropped at Send time and never buffered. Filtering at enqueue keeps a selective receiver's memory bounded by the keys it actually wants rather than the producer's whole key space — important for a receiver interested in a single key out of a high-cardinality producer. keep is called under the bus lock, so it must not call back into the hub. Panics if keep is nil.

It is a method on Hub rather than a package-level function so that K and V are fixed by the hub: callers write no type arguments, and an option built for the wrong key type is a compile error.

func (*Hub[K, V]) WithMerge

func (h *Hub[K, V]) WithMerge(merge Merge[V]) ReceiverOption[K, V]

WithMerge returns an option making a receiver coalesce with its own merge instead of the hub's shared one. This lets a single consumer apply a different policy — e.g. annihilating pending values others must retain — without affecting the rest of the bus. Use it when consumers of the same producer have genuinely different requirements about what may be dropped, which a single hub-wide Merge cannot express. merge is called under the bus lock, like the shared one. Panics if merge is nil.

type Merge

type Merge[V any] func(prev, next V) (merged V, keep bool)

Merge combines an undelivered pending value with a newly sent value for the same key. It is invoked only when the key already has a pending (not yet delivered) slot. It returns the surviving value and whether to keep the slot; keep == false drops the key entirely (annihilation).

Merge is called under the bus lock, so it must not call back into the hub.

type Receiver

type Receiver[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Receiver is a receive-side handle, intended for one consumer goroutine. It holds an insertion-ordered key queue plus a per-key value slot; coalescing happens on Send into these structures, so a read is a plain pop under the lock.

func (*Receiver[K, V]) Chan

func (rx *Receiver[K, V]) Chan() <-chan gobus.Event[K, V]

Chan returns a per-receiver native channel that yields pending events in first-touch key order, carrying the same gobus.Event values the Recv methods return. The channel is unbuffered: coalescing continues to happen in the receiver's slots while the consumer is busy, so a fast publisher produces no backlog beyond the live key set. Repeated calls on the same receiver return the same channel.

Note that an event already handed to the feeder has left the receiver's slots, so a Send for that key while the feeder is parked on delivery enqueues the key afresh rather than coalescing into the in-flight event.

The channel is closed when the feeder observes receiver-close, or sender/hub-close with nothing left to drain. Abandoning the channel without calling Receiver.Close pins the feeder goroutine — it will park forever waiting for the next event. Always Close the receiver when you stop reading.

func (*Receiver[K, V]) Close

func (rx *Receiver[K, V]) Close()

Close closes this receiver only; other receivers and the sender are unaffected. Any pending values are abandoned and the Chan feeder, if started, shuts down and closes the channel. Idempotent.

func (*Receiver[K, V]) Peek

func (rx *Receiver[K, V]) Peek() (gobus.Event[K, V], error)

Peek returns the oldest pending event without removing it, so a subsequent Recv or TryRecv still returns it. It returns gobus.ErrEmpty if nothing is pending, or gobus.ErrClosed if the receiver or hub is closed (or the sender is closed and the pending values have drained) — the same precedence Receiver.TryRecv applies. Peek is TryRecv without the pop, not a raw read of the queue: a closed handle reports ErrClosed even with a value at the head. Note that ErrClosed is therefore not a statement that the backlog was empty: Hub.Close and Receiver.Close abandon whatever is still queued.

The returned value is the current merged contents of the head key's slot. A Send that coalesces into that slot between two Peeks changes what the second Peek reports but leaves the key's queue position — and so its identity as the head — unchanged. A Send whose Merge annihilates the head key removes it, so the next Peek reports a different key: the head key is stable under coalescing, not under annihilation.

Peek takes the hub lock, the same one that serializes the whole Send fan-out, so polling it in a loop slows every publisher and every other receiver on the hub. Call it once per unit of work, not as a spin.

Peek is safe to call from any goroutine, but like the rest of the receive side it is only meaningful on the receiver's single consuming goroutine: a concurrent Recv, TryRecv or Receiver.Chan feeder may take the peeked event before the caller can act on it. An event already handed to the Chan feeder has left the queue, so Peek reports ErrEmpty while it is in flight.

func (*Receiver[K, V]) Recv

func (rx *Receiver[K, V]) Recv() (gobus.Event[K, V], error)

Recv blocks until an event is pending, then pops and returns the oldest one: the key that was touched least recently, carrying its latest merged value. It returns gobus.ErrClosed once the receiver or hub is closed (with the sender's soft close, after the pending values have drained).

func (*Receiver[K, V]) RecvContext

func (rx *Receiver[K, V]) RecvContext(ctx context.Context) (gobus.Event[K, V], error)

RecvContext blocks like Recv but returns ctx.Err() if ctx is cancelled first. Cancellation does not close this receiver.

It implements the closed > cancelled > value precedence documented on gobus.Receiver — including that a cancelled ctx never consumes the event it declined, and that reaching ctx.Err() neither closes nor deregisters the receiver. What is conflate-specific is the cost of ignoring the latter: an abandoned handle keeps coalescing, so it holds one slot per live key for the hub's lifetime. `defer rx.Close()` covers it, as it does for any abandoned receiver.

To consume what is left before closing, loop on Receiver.TryRecv until it reports any error, then Close. The flush alone is not a substitute for the Close: against a still-open sender it ends on ErrEmpty, which is not terminal and does not deregister — only a drain that reaches ErrClosed does that on its own.

func (*Receiver[K, V]) TryRecv

func (rx *Receiver[K, V]) TryRecv() (gobus.Event[K, V], error)

TryRecv pops the oldest pending event without blocking. It returns gobus.ErrEmpty if nothing is pending, or gobus.ErrClosed if the receiver or hub is closed (or the sender is closed and the pending values have drained).

type ReceiverOption

type ReceiverOption[K comparable, V any] func(*receiverConfig[K, V])

ReceiverOption configures a receiver minted by Hub.Receiver. Options are built by the hub's own Hub.WithKeyFilter and Hub.WithMerge methods, which fix K and V from the hub — so option call sites need no type arguments and a mismatched option fails to compile rather than at run time.

The set of options is closed: the parameter type is unexported, so code outside this package cannot name it to write one of its own.

type Sender

type Sender[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Sender is the singleton send-side handle. Safe to share across goroutines.

func (*Sender[K, V]) Close

func (tx *Sender[K, V]) Close()

Close closes the sender. Receivers drain their pending values once before subsequent reads return gobus.ErrClosed, and their Chan feeders close the channel after the same drain. Further Send calls return ErrClosed. Idempotent.

func (*Sender[K, V]) Send

func (tx *Sender[K, V]) Send(k K, v V) error

Send publishes v under key k to every receiver. Never blocks. For a receiver that already has k pending, the caller's Merge coalesces into that slot; otherwise k is appended at the back of the receiver's queue.

A Send to a hub with no live receiver does nothing and returns nil. That has always been the answer — there is nobody to fan out to — but it is now reached without taking the bus lock at all, so a hot producer pays nothing for a bus nobody is reading. Only the cost changes, never the result.

This makes one existing requirement load-bearing in a place a reader would not think to look. A subscriber must call Hub.Receiver *before* it takes its own snapshot of the producer's state, never after: a value published in the gap reaches no receiver, and conflate has no replay to recover it. The requirement is not new — a receiver has always observed only what was sent after it was created — but a publisher that skips the lock cannot notice a subscriber that arrives late, so getting the order wrong loses the value permanently rather than merely racily.

func (*Sender[K, V]) SendContext

func (tx *Sender[K, V]) SendContext(ctx context.Context, k K, v V) error

SendContext behaves like Send, but reports a cancelled ctx instead of publishing. Send never blocks, so ctx is consulted exactly once — there is no parked state in which a cancellation could arrive — and gates the call on nothing beyond reaching the point where the send is resolved.

That single check is made where the send is *resolved*, not on entry. A live ctx on a hub with no live receiver resolves at the lock-free receiver-count read, with nothing to publish and nothing to report. Every other send — including every cancelled one — resolves under s.mu, so that txClosed and ctxDone are read from one consistent view and closed > cancelled is derived rather than assembled from two reads taken at different moments. The rest of this comment is about that path.

So a ctx that was live when SendContext was called but is cancelled by the time this send reaches the front of the lock reports ctx.Err() and publishes nothing. This is deliberate: a caller passing a context is asking for the publish to be bounded by it, and the wait for s.mu is real work — the caller's own Merge and key filters run under that lock. Enqueueing a value on behalf of a context that has since expired, on the grounds that it was live a moment earlier, is the weaker answer. It also keeps every verdict in this package derived from state read at the decision point: txClosed and rx.done are re-read under the lock for the same reason, and recvLoop places its own cancellation check under s.mu rather than trusting an entry-time snapshot.

A cancellation racing this call may therefore land either side of the lock, and the two outcomes — published, or ctx.Err() — are both correct resolutions of that race; the caller cannot have been relying on which.

Precedence is closed > cancelled: a closed sender reports gobus.ErrClosed even for an already-cancelled ctx, since that is the durable answer and a retry with a fresh context would only return it again. A cancelled ctx on a live sender still reports ctx.Err().

Only ctx's Done channel is consulted under the bus lock; ctx.Err() is called after it is released, so a context implementation that locks cannot deadlock against another goroutine's Send or Close. See sendLocked.

func (*Sender[K, V]) TrySend

func (tx *Sender[K, V]) TrySend(k K, v V) error

TrySend is equivalent to Send for conflate: Send never blocks, so there is no separate non-blocking path. Provided to satisfy the common gobus.Sender interface.