controller

type ControlPlaneEventReader ¶

type ControlPlaneEventReader interface {
	ListAfter(ctx context.Context, afterRevision int64, limit int) ([]v1alpha1store.ControlPlaneEvent, error)
	OldestRevision(ctx context.Context) (revision int64, ok bool, err error)
	CurrentRevision(ctx context.Context) (int64, error)
}

ControlPlaneEventReader is the durable event-log surface the controller uses to replay source invalidations.

type ControllerConfig ¶

type ControllerConfig struct {
	Retention                  RetentionPolicy
	DiscoveryInterval          time.Duration
	DiscoveryStaleAfterMisses  int
	DiscoveryDeleteAfterMisses int
	DependencyKinds            map[string]bool
}

ControllerConfig controls optional controller maintenance loops.

type ControllerHandle ¶

type ControllerHandle struct {
	Controller *DeploymentController
	Discovery  *DeploymentDiscoveryController
	Retention  *RetentionPruner
}

ControllerHandle owns the always-on Deployment controller loops.

func StartDeploymentController(
	ctx context.Context,
	pool *pgxpool.Pool,
	stores map[string]*v1alpha1store.Store,
	adapters map[string]types.DeploymentAdapter,
	config ControllerConfig,
) (*ControllerHandle, error)

type DeploymentController ¶

type DeploymentController struct {
	Stores   map[string]*v1alpha1store.Store
	Adapters map[string]types.DeploymentAdapter
	Getter   v1alpha1.GetterFunc
	Events   ControlPlaneEventReader

	BatchLimit int
	Wakeups    <-chan struct{}
	Queue      workqueue.TypedRateLimitingInterface[deploymentQueueKey]
	// DependencyKinds extends the built-in resource kinds whose durable events
	// requeue Deployments. Fingerprint gating prevents unchanged adapter work.
	DependencyKinds map[string]bool
	// contains filtered or unexported fields
}

DeploymentController replays durable source invalidations and reconciles Deployments through an in-memory workqueue. Source state remains durable in the v1alpha1 tables and control_plane_events; queued work is intentionally process-local and rebuilt by startup/repair full reconciles.

func (c *DeploymentController) Checkpoint() int64

func (c *DeploymentController) Drain(ctx context.Context) (SyncResult, error)

func (c *DeploymentController) FullReconcile(ctx context.Context) (int, error)

func (c *DeploymentController) HandleEvent(ctx context.Context, event v1alpha1store.ControlPlaneEvent) (int, error)

func (c *DeploymentController) ReadinessError() error

func (c *DeploymentController) Ready() bool

func (c *DeploymentController) Refresh(ctx context.Context) (SyncResult, error)

func (c *DeploymentController) Run(ctx context.Context, resyncInterval time.Duration) error

func (c *DeploymentController) RunOnce(ctx context.Context) (int, error)

func (c *DeploymentController) RunWorker(ctx context.Context) error

func (c *DeploymentController) Sync(ctx context.Context, checkpoint int64) (SyncResult, error)

type DeploymentDiscoveryController ¶

type DeploymentDiscoveryController struct {
	Stores            map[string]*v1alpha1store.Store
	Adapters          map[string]types.DeploymentAdapter
	StaleAfterMisses  int
	DeleteAfterMisses int
}

DeploymentDiscoveryController materializes adapter discovery snapshots into persisted Deployment rows. It owns provider-observed state; the normal DeploymentController skips these rows so they do not become desired state.

func (c *DeploymentDiscoveryController) Run(ctx context.Context, interval time.Duration) error

func (c *DeploymentDiscoveryController) Sync(ctx context.Context) (DeploymentDiscoverySyncResult, error)

type DeploymentDiscoverySyncResult ¶

type DeploymentDiscoverySyncResult struct {
	Runtimes   int
	Discovered int
	// Stale counts discovered rows currently at or past the consecutive-miss
	// staleness threshold (their Discovered/Ready conditions are False).
	Stale int
	// Removed counts discovered rows deleted this pass — either past the
	// consecutive-miss deletion threshold or orphaned by Runtime deletion.
	Removed int
}

DeploymentDiscoverySyncResult summarizes one discovery materialization pass.

type PluginController ¶

type PluginController struct {
	Store    pluginStore
	Resolver source.Resolver
	Wakeups  <-chan struct{}
	// contains filtered or unexported fields
}

PluginController reconciles Plugin resources out of band of the API write: it resolves each plugin's pinned source pointer (a git commit or — later — an OCI digest) to a concrete commit/digest, scans the source for its manifest and inventory, and records all of it in PluginStatus. It stores NOTHING: the bundle stays at its origin and is materialized from source at deploy time.

It is level-triggered — every control-plane wakeup (and the resync tick) re-lists plugins and enqueues those whose status is behind their generation. Status writes never re-emit control-plane events (the trigger skips spec-equal updates), so the controller does not wake itself. Each controller opens its OWN control-plane LISTEN subscription (the Deployment controller has a separate one); there is no shared listen loop.

func NewPluginController(
	pool *pgxpool.Pool,
	stores map[string]*v1alpha1store.Store,
	deps PluginControllerDeps,
) (*PluginController, error)

func (c *PluginController) Run(ctx context.Context, resync time.Duration) error

func (c *PluginController) Start(ctx context.Context) error

func (c *PluginController) Stop()

type PluginControllerDeps ¶

type PluginControllerDeps struct {
	Resolver source.Resolver
}

PluginControllerDeps are the Plugin controller's dependencies. Resolver pins a plugin's source pointer and loads its bundle; it is required.

type PruneStores ¶

type PruneStores struct {
	ControlPlaneEvents interface {
		PruneBefore(ctx context.Context, before time.Time, keepAfterRevision int64, limit int) (int64, error)
	}
}

PruneStores groups the store surfaces needed by RunRetentionPrune. Keeping these as tiny interfaces lets the controller package stay independent from concrete Postgres store construction and keeps tests cheap.

type ReconcileAction ¶

type ReconcileAction string

ReconcileAction is the operation implied by a Deployment's current source state.

const (
	// ReconcileActionApply converges a Deployment toward running desired state.
	ReconcileActionApply ReconcileAction = "apply"
	// ReconcileActionDelete tears down runtime resources for an undeploy/delete.
	ReconcileActionDelete ReconcileAction = "delete"
)

type RetentionPolicy ¶

type RetentionPolicy struct {
	ControlPlaneEvents time.Duration
	EventKeepAfterRev  int64
	BatchLimit         int
}

RetentionPolicy is the bounded-history contract for the controller event replay log. Durations <= 0 disable pruning.

func (p RetentionPolicy) Enabled() bool

type RetentionPruneResult ¶

type RetentionPruneResult struct {
	ControlPlaneEvents int64
}

RetentionPruneResult reports how many event rows were removed in one maintenance pass.

func RunRetentionPrune(ctx context.Context, stores PruneStores, policy RetentionPolicy, now time.Time) (RetentionPruneResult, error)

type RetentionPruner ¶

type RetentionPruner struct {
	Stores PruneStores
	Policy RetentionPolicy
	Now    func() time.Time
}

RetentionPruner owns the periodic maintenance loop for controller event replay rows.

func (p *RetentionPruner) Enabled() bool

func (p *RetentionPruner) Run(ctx context.Context, interval time.Duration) error

func (p *RetentionPruner) RunOnce(ctx context.Context) (RetentionPruneResult, error)

type SkillController ¶

type SkillController struct {
	Store   skillStore
	Resolve SkillResolveFunc
	Wakeups <-chan struct{}
	// contains filtered or unexported fields
}

SkillController reconciles Skill resources out of band of the API write: it resolves each skill's pinned git source ref to a concrete commit and records it in SkillStatus.ResolvedSource. It stores NOTHING: the skill content stays at its origin and is materialized from source at deploy time. It mirrors the Plugin controller's resolve-and-pin model, minus the manifest/inventory scan (a skill has no bundle to enumerate).

It is level-triggered — every control-plane wakeup (and the resync tick) re-lists skills and enqueues those whose status is behind their generation. Status writes never re-emit control-plane events (the trigger skips spec-equal updates), so the controller does not wake itself. Each controller opens its OWN control-plane LISTEN subscription.

func NewSkillController(
	pool *pgxpool.Pool,
	stores map[string]*v1alpha1store.Store,
	deps SkillControllerDeps,
) (*SkillController, error)

func (c *SkillController) Run(ctx context.Context, resync time.Duration) error

func (c *SkillController) Start(ctx context.Context) error

func (c *SkillController) Stop()

type SkillControllerDeps ¶

type SkillControllerDeps struct {
	Resolve SkillResolveFunc
}

SkillControllerDeps are the Skill controller's dependencies. Resolve pins a skill's git source ref to a commit; it defaults to a git ls-remote resolver when nil.

type SkillResolveFunc ¶

type SkillResolveFunc func(ctx context.Context, repo *v1alpha1.Repository) (commit string, err error)

SkillResolveFunc resolves a skill's git source ref to a concrete commit SHA. It is the only I/O-bearing dependency of the Skill controller, so tests inject a fake instead of touching the network.

type SyncResult ¶

type SyncResult struct {
	Checkpoint   int64
	Events       int
	FullResynced bool
}

SyncResult describes one controller replay pass.