Documentation
¶
Overview ¶
Package controlplane coordinates flow execution. See design docs/plans/2026-06-24-flow-engine-design.md §18.5.
Index ¶
Constants ¶
const DefaultNackBackoff = 2 * time.Millisecond
DefaultNackBackoff is the base delay before a Nack'd Work is requeued (H1b), unless WithNackBackoff overrides it. It is small enough to be invisible to a healthy worker yet large enough to convert a transiently-failing dependency's redelivery from a 100%-CPU spin into a bounded-rate retry.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ClosedError ¶
type ClosedError struct{}
ClosedError reports an operation attempted after the control plane has been shut down (its lifetime ctx cancelled via Close). It is fail-secure: a shut-down control plane rejects new work rather than silently dropping it (§18.5). It carries no fields — the failure is context-free (the plane is simply gone) — but remains a typed struct so callers can errors.As for it.
type MemControlPlane ¶
type MemControlPlane struct {
// contains filtered or unexported fields
}
MemControlPlane is the in-process, channel-based control plane (§18.5). It is local and ephemeral: enqueued Work lives only in memory and is lost on process exit (durability is the CheckpointStore's job, separate). Construct it with Mem.
func Mem ¶
func Mem(opts ...MemOption) *MemControlPlane
Mem constructs and starts an in-process control plane (§18.5). The returned *MemControlPlane runs a background dispatcher goroutine bound to a fresh lifetime ctx; the dispatcher exits (closing all consumer channels) when that ctx is cancelled — see Close. Submit/Consume ctxs passed to the methods scope individual calls/subscriptions and are independent of the lifetime ctx. Options (e.g. WithNackBackoff) tune behavior at the composition root.
func (*MemControlPlane) Close ¶
func (cp *MemControlPlane) Close()
Close cancels the control plane's lifetime ctx, closing every consumer's Delivery channel and stopping the dispatcher. It blocks until the dispatcher has fully exited, so after Close returns no control-plane goroutine remains. It is idempotent (cancelling an already-cancelled ctx is a no-op).
func (*MemControlPlane) Consume ¶
func (cp *MemControlPlane) Consume(ctx context.Context, serves []flow.GraphVersionKey) (<-chan flow.Delivery, error)
Consume registers a worker serving the given version keys and returns a channel delivering only Work whose Key is in serves (§18.5). Registration is implicit: consuming IS registering — there is no separate RPC. The channel is closed when ctx is done (this subscription ends) or when the control plane is shut down (Close), so a `for d := range ch` worker loop terminates cleanly with no goroutine leak. A nil/empty serves registers a consumer that receives nothing.
func (*MemControlPlane) Submit ¶
Submit enqueues w for consumers serving w.Key (§18.5). It honors ctx and the control plane's lifetime: it returns ctx.Err() if ctx is done, or a typed *ClosedError if the control plane has been shut down (fail-secure — a shut-down control plane rejects new work rather than silently dropping it). It does not block on consumer availability: enqueue is decoupled from delivery.
type MemOption ¶
type MemOption func(*MemControlPlane)
MemOption configures a MemControlPlane at construction (§18.5). It is the functional-options seam (CLAUDE.md: wire dependencies at the composition root); a new knob is a new MemOption with zero edits to existing callers (open/closed).
func WithNackBackoff ¶
WithNackBackoff sets the base delay before a Nack'd Work is requeued (H1b). A value <= 0 is ignored (DefaultNackBackoff stays in force), so a caller cannot accidentally disable the spin guard and reintroduce the hot-loop.
Source Files