README
¶
pkg/controller/coalesce
Generic "latest-wins" channel drain helper used by event-driven components that produce work faster than they can consume it.
Overview
When an upstream burst of events floods a component's subscription channel, processing each one individually wastes work — only the most recent event is meaningful for downstream consumers. DrainLatest lets a component finish processing the current event, then non-blockingly drain the rest of the channel: uninterrupted runs of coalescible events collapse to their latest element (delivered via the flush callback), while every other event is a run boundary — the held run is flushed first, then the event is routed to the component's regular handler, preserving arrival order across event types.
This is the same pattern several components implement; pulling it into one place avoids re-implementing the type assertion + channel-drain dance everywhere. Flushing at run boundaries (instead of holding the latest until the channel empties) is load-bearing: under sustained mixed traffic the channel may never empty, and a hold-until-empty drain starves the coalesced type for the entire burst while dispatching newer other-type events ahead of the older held one.
API
func DrainLatest[T busevents.Event](
eventChan <-chan busevents.Event,
handleOther func(busevents.Event),
flush func(latest T, supersededCount int),
)
Drains until the channel is momentarily empty; flush is invoked once per run of coalescible events (with how many earlier events the delivered one superseded). Non-matching event types and matching events whose Coalescible() bool returns false are routed to handleOther in arrival order.
Coalescibility
Only events that implement the pkg/events.CoalescibleEvent interface and return Coalescible() == true are eligible for coalescing. The flag is set by the publisher per event — e.g. ReconciliationTriggeredEvent is coalescible when triggered by a debounced resource change but not when triggered by drift prevention (a missed drift-prevention pass would silently swallow the corrective action).
Quick Start
import (
"gitlab.com/haproxy-haptic/haptic/pkg/controller/coalesce"
"gitlab.com/haproxy-haptic/haptic/pkg/controller/events"
)
func (c *Component) handleSomeEvent(event *events.SomeEvent) {
c.performWork(event)
// After work completes, drain queued events: consecutive coalescible
// SomeEvents collapse to their latest, everything else is handled in
// arrival order.
coalesce.DrainLatest(
c.eventChan,
c.handleEvent, // route non-coalescible / other types here
func(latest *events.SomeEvent, superseded int) {
c.logger.Debug("processing coalesced event", "superseded_count", superseded)
c.performWork(latest)
},
)
}
See Also
pkg/events— defines theCoalescibleEventinterfacepkg/controller/reconciler— primary producer: marksReconciliationTriggeredEventcoalescible (or not, depending on the trigger reason) before publishing- Consumers (grep
coalesce.DrainLatest[): onlypkg/controller/deployer'sDeploymentScheduler.handlePodsDiscovered(hand-rolled loop that can't embedcomponent.Base); Base-embedded components coalesce via Base's mailbox mode (CoalescesOn() []string) instead
License
Apache-2.0 — see root LICENSE.
Documentation
¶
Overview ¶
Package coalesce provides utilities for event coalescing in controller components.
Event coalescing implements the "latest wins" pattern where intermediate events are skipped when newer events of the same type are available. This prevents queue backlog when events arrive faster than they can be processed.
Only events that implement CoalescibleEvent and return Coalescible() == true are coalesced, and only within an uninterrupted run of such events. Any other event (different type, or same type but not coalescible) is a run boundary: the held latest event of the current run is flushed FIRST, then the boundary event is passed to handleOther. This preserves arrival order across event types and guarantees the coalesced type cannot be starved: an earlier design held the run's latest back until the channel drained empty, and under sustained mixed traffic (e.g. deployment-completed status applies each taking longer than the event arrival gap) that point never came — rendered status patches were starved for the entire burst (54s observed in gateway-api conformance) while newer other-type events were dispatched ahead of the older held event.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DrainLatest ¶
func DrainLatest[T busevents.Event]( eventChan <-chan busevents.Event, handleOther func(busevents.Event), flush func(latest T, supersededCount int), )
DrainLatest drains the event channel until it is momentarily empty. Uninterrupted runs of coalescible events of type T collapse to their latest element, delivered via flush together with the count of superseded events. Every other event is a run boundary: the held run is flushed first, then the event is passed to handleOther, preserving arrival order.
An event joins the current run if:
- It matches type T
- It implements CoalescibleEvent interface
- Its Coalescible() method returns true
Usage pattern:
func (c *Component) handleSomeEvent(event *events.SomeEvent) {
c.performWork(event)
// After work completes, drain queued events: consecutive coalescible
// SomeEvents collapse to their latest, everything else is handled in
// arrival order.
coalesce.DrainLatest(
c.eventChan,
c.handleEvent, // Handle non-coalescible and other event types
func(latest *events.SomeEvent, superseded int) {
c.performWork(latest)
},
)
}
func DrainLatestByType ¶
func DrainLatestByType( eventChan <-chan busevents.Event, eventType string, handleOther func(busevents.Event), flush func(latest busevents.Event, supersededCount int), )
DrainLatestByType is the runtime-typed sibling of DrainLatest. Instead of a compile-time type parameter it matches events whose EventType() equals eventType. Components that select on a dynamic event-type string (e.g. component.Base's coalescing loop, which reads the type from a CoalescingHandler) use this; consumers with a static type use the generic DrainLatest.
Types ¶
This section is empty.
Source Files