component

type Base ¶

type Base struct {
	// contains filtered or unexported fields
}

Base is a reusable event-loop implementation. It subscribes on construction (so components are guaranteed to receive events published after EventBus.Start()), wraps each dispatch in a recover, and supports graceful shutdown via either a cancelled context or an explicit Stop.

func New(cfg *Config) *Base

func (b *Base) EventBus() *busevents.EventBus

func (b *Base) FlushPending()

func (b *Base) LifecycleContext() context.Context

func (b *Base) Logger() *slog.Logger

func (b *Base) Name() string

func (b *Base) Start(ctx context.Context) error

func (b *Base) Stop()

type CoalescingHandler ¶

type CoalescingHandler interface {
	CoalescesOn() []string
}

CoalescingHandler is an optional interface implemented by handlers whose events of the declared types have latest-wins semantics. When it returns a non-empty list, Base runs in MAILBOX mode: a dedicated intake goroutine drains the subscription channel immediately into an internal unbounded queue, so the bus-side buffer can never fill and the bus never drops this subscriber's events — no matter how slow the handler is. Uninterrupted runs of coalescible events of a declared type (i.e. event.(busevents.CoalescibleEvent).Coalescible() == true) collapse to their latest element at the queue tail; any other event is appended, preserving arrival order across event types. The worker dispatches from the queue head at its own pace.

This exists because slow handlers (e.g. status appliers doing SSA round-trips per event) otherwise stall the channel long enough under burst for the bus to overflow the subscriber buffer and drop events — including non-coalescible ones and the final event of a burst, whose loss leaves stale state until the next external trigger.

WHEN A COMPONENT MUST RUN IN MAILBOX MODE. Declare a type when BOTH hold:

1. The handler can block — a Kubernetes API round-trip, an agent call, any network I/O — for longer than the gap between two events.
2. At least one subscribed type can arrive faster than the handler drains it: published per render, per pod, per resource change, or by a watcher the chart runs undebounced.

One without the other does not qualify: a blocking handler on a per-config-change stream cannot outrun its input, and a fast handler on a firehose keeps up. Every adopter so far was retrofitted after a measured incident; the point of stating the rule is to identify the next one first.

If a component meets both but must NOT collapse — because its event type string is shared across sources it has to see individually, as resource.index.updated is across every watched kind — it needs a level-triggered re-read instead, not this. Coalescing collapses by event TYPE, so declaring such a type discards the sources the run did not end on.

Two gates hold this together:

• pkg/controller/subscriber_inventory_test.go records the mailbox decision for every non-lossy subscriber and fails when a new one appears without one.
• pkg/controller/events/coalescible_inventory_test.go records why each type that implements CoalescibleEvent is safe to collapse. Declaring a type that is not armed is inert, so arming a type is the change that can start losing events — that is the one gated.

Declaring a type is a per-component statement that ONLY the latest queued event of that type matters to THIS component. Never declare a type whose every instance carries per-event bookkeeping for the component (e.g. the deployer must see every deployment.completed to clear its in-flight flag, so it declares only deployment.scheduled).

An empty list disables coalescing — handlers that conditionally need it can return nil to opt out at runtime (plain channel loop, no mailbox).

type Config ¶

type Config struct {
	EventBus   *busevents.EventBus
	Logger     *slog.Logger
	Name       string
	BufferSize int
	Handler    EventHandler
	EventTypes []string

	// EventFilter, when set, is asked about every event that matched
	// EventTypes, before it reaches the buffer. Use it where one event type
	// carries many subjects and this component wants a few: filtering in the
	// handler still spends a buffer slot on each event, and for a non-lossy
	// subscriber a full buffer ends the controller iteration. Runs on the
	// publishing path — field comparisons only.
	EventFilter func(busevents.Event) bool
}

Config wires up a new Base.

If EventTypes is empty the component receives every event on the bus; if it is non-empty the component receives only the listed types (preferred for components that only react to a handful of events).

type EventHandler ¶

type EventHandler interface {
	HandleEvent(event busevents.Event)
}

EventHandler dispatches a single event received on the subscription channel. Implementations should not block the goroutine for longer than the processing budget documented on the event.

type PanicHandler ¶

type PanicHandler interface {
	HandlePanic(recovered any, event busevents.Event)
}

PanicHandler is an optional interface that a component implements when it needs to publish a domain-specific response if event handling panics. The base always logs the panic and keeps the loop alive regardless of whether this interface is implemented.

type QueueCoalescingHandler ¶

type QueueCoalescingHandler interface {
	CoalescingHandler
	CoalescesAcrossQueue() bool
}

QueueCoalescingHandler is an optional refinement of CoalescingHandler for components whose declared types are latest-wins across the WHOLE queue, not just within an uninterrupted run. When CoalescesAcrossQueue reports true, a coalescible event of a declared type supersedes any earlier queued coalescible event of the same type wherever it sits, so a stream that strictly alternates two declared types (render → resources.applied, deploy → deployment.completed, …) still keeps the queue bounded by the number of declared types instead of growing without limit while the handler is slower than the arrival rate. Relative order of the surviving entries is preserved; a non-coalescible event still separates nothing — it simply keeps its place. Only declare this when the component's own per-type semantics are "the newest event carries the complete state"; the status applier is the canonical adopter (each event carries the full patch set for its phase, and a 2048-deep alternating backlog was measured at 6 GB of live heap on a 1,900-route churn).

type ReadySignal ¶

type ReadySignal struct {
	// contains filtered or unexported fields
}

ReadySignal is a per-Start signal used by leader-only components to let callers wait until the component has finished subscribing to the event bus. Leader-only components subscribe in Start() (not in their constructor), and the controller start-up sequence waits on SubscriptionReady() before proceeding so published events are not lost.

Embed *ReadySignal in a leader-only component to pick up the accessor and Mark helpers instead of repeating the channel plumbing in each file.

Example:

type Component struct {
    *component.ReadySignal
    // ... other fields
}

func New(...) *Component {
    return &Component{ReadySignal: component.NewReadySignal(), ...}
}

func (c *Component) Start(ctx context.Context) error {
    c.eventChan = bus.SubscribeTypesLeaderOnly(...)
    c.MarkReady()
    // ... event loop
}

func NewReadySignal() *ReadySignal

func (r *ReadySignal) MarkReady()

func (r *ReadySignal) Rearm()

func (r *ReadySignal) SubscriptionReady() <-chan struct{}