Documentation
¶
Overview ¶
Package reconcile is the agent's desired-state execution engine: a small, in-process control loop in the Kubernetes mould (a level-triggered reconciler driven by a deduplicating work queue) but with no cluster, no apiserver, and no etcd. It runs in one binary over the resource store, the event log, and the job queue. The control-theory mechanics are borrowed (they are domain-agnostic and proven); the engine on top, reconciling agent goals toward an LLM-judged stop condition, is the agent's own.
This file is the critical primitive: the work queue. It mirrors the semantics of client-go's controller workqueue (a key is never processed concurrently; adds collapse while a key waits; delayed and rate-limited re-adds avoid hot loops) without importing it, and it schedules delays through a clock.Timing source so every delay is deterministic under a Manual clock.
Index ¶
- Constants
- type Controller
- type ControllerOption
- type Manager
- type ManagerOption
- type Queue
- func (q *Queue[T]) Add(item T)
- func (q *Queue[T]) AddAfter(item T, d time.Duration)
- func (q *Queue[T]) AddRateLimited(item T)
- func (q *Queue[T]) Done(item T)
- func (q *Queue[T]) Forget(item T)
- func (q *Queue[T]) Get() (item T, shutdown bool)
- func (q *Queue[T]) Len() int
- func (q *Queue[T]) NumRequeues(item T) int
- func (q *Queue[T]) ShutDown()
- type Reconciler
- type ReconcilerFunc
- type Ref
- type Result
Constants ¶
const DefaultResync = 30 * time.Second
DefaultResync is how often the manager re-enqueues every live resource of each registered kind when no interval is configured. Resync is the safety net: even if a change hint is lost (the bus is at-most-once), every resource is reconciled again within this window, so the system always converges.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Controller ¶
type Controller[T comparable] struct { // contains filtered or unexported fields }
Controller runs a Reconciler against keys delivered by a Queue. It owns the retry policy: a Transient error backs off and retries, a success forgets the key, an explicit RequeueAfter re-enqueues after a delay, and any other error class (Terminal, NeedsApproval, BudgetExceeded, Cancelled) is not hot-retried - the manager's periodic resync is the safety net that re-enqueues it later.
func NewController ¶
func NewController[T comparable](name string, queue *Queue[T], rec Reconciler[T], opts ...ControllerOption) *Controller[T]
NewController builds a controller pulling from queue and reconciling with rec.
func (*Controller[T]) Queue ¶
func (c *Controller[T]) Queue() *Queue[T]
Queue exposes the controller's queue so a manager can enqueue keys into it.
func (*Controller[T]) Run ¶
func (c *Controller[T]) Run(ctx context.Context)
Run starts the workers and blocks until ctx is cancelled (which shuts the queue down) and every worker has drained. It is the controller's whole lifecycle.
type ControllerOption ¶
type ControllerOption func(*controllerConfig)
ControllerOption configures a Controller.
func WithWorkers ¶
func WithWorkers(n int) ControllerOption
WithWorkers sets the number of concurrent workers (default 1). The queue guarantees one key is never reconciled by two workers at once regardless, so more workers only add cross-key parallelism.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager wires reconcilers to the resource store. It owns one Controller per registered kind, routes change hints to the right controller's queue, and runs the periodic resync that guarantees eventual convergence. It deliberately holds no watch of its own: truth is the store, and the manager only decides WHEN to reconcile, never WHAT the state is.
func NewManager ¶
func NewManager(store resource.Store, opts ...ManagerOption) *Manager
NewManager returns a manager over store. Register kinds before Start.
func (*Manager) Enqueue ¶
Enqueue schedules a reconcile of ref. This is the change-hint entry point: a bus signal, a write callback, or any "this resource may have changed" notice calls it. Hints are advisory (the reconcile re-reads truth), and a hint for an unregistered kind is ignored.
func (*Manager) Register ¶
func (m *Manager) Register(kind string, rec Reconciler[Ref], opts ...ControllerOption)
Register attaches a reconciler for a kind. It must be called before Start. Registering the same kind twice replaces the earlier reconciler.
type ManagerOption ¶
type ManagerOption func(*Manager)
ManagerOption configures a Manager.
func WithClock ¶
func WithClock(c clock.Timing) ManagerOption
WithClock sets the time source (default clock.System). A Manual clock makes the resync ticker deterministic in tests.
func WithResync ¶
func WithResync(d time.Duration) ManagerOption
WithResync sets the resync interval (default DefaultResync). A value <= 0 disables periodic resync (only the initial resync and explicit Enqueue drive reconciles), which is useful in tests that want to control every trigger.
func WithoutResync ¶
func WithoutResync() ManagerOption
WithoutResync disables the resync sweep entirely, both the initial pass and the periodic one, so the manager drives only the resources explicitly enqueued (by a completion signal or an Enqueue hint) and never adopts pre-existing resources it finds in the store. A one-shot run uses this so it drives its own submitted work to completion without re-adopting goals left non-terminal by an earlier run; that keeps each run's event stream its own and makes resuming a parked run an explicit act rather than a side effect of starting any run.
type Queue ¶
type Queue[T comparable] struct { // contains filtered or unexported fields }
Queue is a deduplicating, level-triggered work queue keyed by T. A key added any number of times before it is fetched is processed once; while a key is being processed, re-adds are remembered and the key is re-queued exactly once when processing finishes (Done). AddAfter and AddRateLimited delay a re-add without blocking, so a failing item backs off instead of spinning. T is the item's identity (for the reconciler, a resource Ref), so the queue carries no payload: the reconciler always re-reads current state, never trusting a stale enqueued value.
func NewQueue ¶
func NewQueue[T comparable](clk clock.Timing) *Queue[T]
NewQueue returns an empty queue that schedules delays on clk.
func (*Queue[T]) Add ¶
func (q *Queue[T]) Add(item T)
Add enqueues item for processing. It is a no-op if item is already waiting (dedup) or already being processed (it is remembered and re-queued on Done).
func (*Queue[T]) AddAfter ¶
AddAfter enqueues item once d has elapsed on the queue's clock, without blocking the caller. A d <= 0 adds immediately. Repeated AddAfter for the same item are harmless: each fires an Add, and Add dedups. The timer is created synchronously (registered on the clock before AddAfter returns) so a Manual clock advanced past d fires it deterministically.
Rather than one goroutine per delayed item, all delayed adds are handed to a single waiting loop (started on first use) that holds them in a deadline ordered heap and enqueues each when the clock reaches its deadline. The timer is created here under q.mu so a concurrent ShutDown either observes it (and the loop stops it) or has already set shuttingDown (and this add is dropped). Only the loop ever stops a timer it is waiting on, so there is no cross-goroutine race on the timer channel.
func (*Queue[T]) AddRateLimited ¶
func (q *Queue[T]) AddRateLimited(item T)
AddRateLimited enqueues item after its current backoff delay, growing the delay on each call (per-item exponential backoff) until Forget resets it. This is how a reconcile that errors is retried without a hot loop.
func (*Queue[T]) Done ¶
func (q *Queue[T]) Done(item T)
Done marks item as finished. If item was re-added while it was being processed, it is re-queued now (and only now), preserving the "never concurrent per key" guarantee.
func (*Queue[T]) Forget ¶
func (q *Queue[T]) Forget(item T)
Forget clears item's backoff so its next AddRateLimited starts from baseDelay again. Call it after a successful reconcile.
func (*Queue[T]) Get ¶
Get blocks until an item is ready or the queue shuts down. The returned item is marked as processing; the caller MUST call Done(item) when finished so a re-add that arrived during processing is re-queued. shutdown is true only when the queue is draining and empty.
func (*Queue[T]) NumRequeues ¶
NumRequeues is how many times item has been rate-limited since the last Forget.
type Reconciler ¶
type Reconciler[T comparable] interface { Reconcile(ctx context.Context, key T) (Result, error) }
Reconciler drives one resource of a kind toward its desired state. It is given only the key (identity), never an event or a cached object, which forces it to re-read the current observed state every time: the level-triggered discipline that makes the loop self-healing and crash-resumable. Reconcile must be idempotent and should return promptly; long or stochastic work (an LLM step) is dispatched elsewhere and observed, not run inline.
type ReconcilerFunc ¶
type ReconcilerFunc[T comparable] func(ctx context.Context, key T) (Result, error)
ReconcilerFunc adapts a function to a Reconciler.
type Ref ¶
Ref identifies the resource a reconcile acts on: kind + scope + name, the logical key of the resource store. It is the only thing a Reconciler is handed, so the reconciler must re-read the live resource itself (level-triggered).
type Result ¶
Result tells the controller what to do after a reconcile. The zero Result with a nil error means "success, nothing more to do": the key settles and is not touched again until it changes or the next resync. A positive RequeueAfter asks to be reconciled again after that delay (the idempotent way to poll long-running work), independent of backoff.
Source Files