The highest tagged major version is
Documentation
¶
Overview ¶
Package control orchestrates authorization, durable idempotent admission, dispatch and lifecycle audit without enforcing queue delivery or supervising workers. NewService borrows its authorizer, lifecycle journal, dispatcher and clock; they must support synchronous shared use and bounded operations. NextDesiredState plans convergence revisions, not worker acknowledgements.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrNotDesiredStateAction identifies mutations that do not alter durable // worker or queue operational state. ErrNotDesiredStateAction = errors.New("control: action has no desired state") // ErrInvalidDesiredTarget identifies an action/resource combination that // workers cannot enforce through desired state. ErrInvalidDesiredTarget = errors.New("control: invalid desired-state target") // ErrDesiredTargetMismatch prevents a record from being reused for another // resource. ErrDesiredTargetMismatch = errors.New("control: desired-state target mismatch") // ErrDesiredTenantMismatch prevents cross-tenant state reuse. ErrDesiredTenantMismatch = errors.New("control: desired-state tenant mismatch") // ErrInvalidDesiredTransition identifies corrupt or irreversible state. ErrInvalidDesiredTransition = errors.New("control: invalid desired-state transition") // ErrDesiredRevisionExhausted prevents revision wraparound. ErrDesiredRevisionExhausted = errors.New("control: desired-state revision exhausted") )
var ( // ErrInvalidDispatcherConfiguration reports a missing dispatch boundary. ErrInvalidDispatcherConfiguration = errors.New("control: invalid dispatcher configuration") ErrDataPlaneUnavailable = errors.New("control: data plane unavailable") )
ErrCommandIDUnavailable reports failure to allocate a durable operation ID.
ErrLifecycleJournalUnavailable reports a journal that cannot durably record dispatch and acknowledgement recovery boundaries.
var ErrOutcomeUnknown = errors.New("control: command outcome unknown")
ErrOutcomeUnknown means dispatch may have occurred but its result could not be durably recorded. Clients must inspect the command before retrying.
Functions ¶
This section is empty.
Types ¶
type Authorizer ¶
type Authorizer interface {
// Authorize must fail closed for the tenant, authenticated actor,
// permission and target, returning nil only when access is allowed.
Authorize(
context.Context,
string,
string,
controlplane.Permission,
controlplane.Target,
) error
}
Authorizer enforces actor permissions at the target-resource boundary.
type DesiredRecord ¶
type DesiredRecord struct {
// TenantID is the durable tenant scope and must match subsequent commands.
TenantID string
// Target is the exact durable resource scope and must match subsequent
// commands.
Target controlplane.Target
// State is the requested operational state, not an observed worker
// acknowledgement.
State DesiredState
// Revision increments for each planned change and must not wrap uint64.
Revision uint64
// CommandKey stores the initiating CommandID, despite the field name; it is
// not the IdempotencyKey.
CommandKey string
// ChangedAt retains the command's RequestedAt time.
ChangedAt time.Time
}
DesiredRecord is one revisioned durable state for a queue or worker scope.
func NextDesiredState ¶
func NextDesiredState( current *DesiredRecord, command controlplane.Command, ) (DesiredRecord, error)
NextDesiredState validates and plans the next monotonic desired-state revision. It performs no persistence and sends no worker command.
type DesiredState ¶
type DesiredState string
DesiredState is the durable operational state workers converge toward.
const ( // DesiredActive allows future worker admission for the selected scope. DesiredActive DesiredState = "active" // DesiredPaused requests paused admission, without proving active jobs have // drained. DesiredPaused DesiredState = "paused" // DesiredDraining requests completion of active work without new admission. DesiredDraining DesiredState = "draining" // DesiredTerminating is a terminal desired transition that cannot return to // another state. DesiredTerminating DesiredState = "terminating" )
type DispatchOutcome ¶
type DispatchOutcome struct {
// Status is a terminal backend outcome validated by Service, not merely
// command acceptance.
Status controlplane.CommandStatus
// Failure is a secret-safe public failure code for unsuccessful terminal
// outcomes.
Failure string
// WorkerID optionally attributes an acknowledgement and must accompany
// Protocol.
WorkerID string
// Protocol optionally identifies the acknowledging worker protocol; paired
// with WorkerID.
Protocol *controlplane.ProtocolVersion
// CapabilityAvailable optionally reports negotiated capability
// availability; nil means unspecified.
CapabilityAvailable *bool
// CompletedAt is the backend completion time and must not precede
// RequestedAt.
CompletedAt time.Time
}
DispatchOutcome is a terminal, redacted data-plane acknowledgement. It deliberately omits tenant and idempotency identity, which Service owns.
type Dispatcher ¶
type Dispatcher interface {
// Dispatch synchronously enforces the borrowed command; failure may follow
// partial or accepted external work.
Dispatch(context.Context, controlplane.Command) error
}
Dispatcher sends a validated, authorized, and durably accepted command to its explicitly selected data-plane or workload adapter.
type Journal ¶
type Journal interface {
// Accept must atomically claim the tenant/key and initial audit, returning
// existing results without new dispatch for duplicates and rejecting
// conflicting envelopes.
Accept(
context.Context,
controlplane.Command,
) (result controlplane.CommandResult, created bool, err error)
// Complete must atomically persist the terminal result and corresponding
// audit; errors do not undo external dispatch.
Complete(context.Context, controlplane.CommandResult) error
}
Journal provides durable idempotency and audit persistence. Accept must atomically persist a newly accepted command and its initial audit event. It returns the existing result with created=false for duplicate keys. Complete must atomically update the result and append its completion audit event.
type LifecycleJournal ¶
type LifecycleJournal interface {
Journal
// MarkDispatched must durably record the pre-dispatch recovery boundary and
// audit before backend I/O.
MarkDispatched(context.Context, controlplane.CommandResult) error
// MarkAcknowledged must durably record acknowledgement and audit before
// terminal completion.
MarkAcknowledged(context.Context, controlplane.CommandResult) error
}
LifecycleJournal adds durable dispatch and acknowledgement boundaries while leaving the original Journal interface source-compatible.
type ResultDispatcher ¶
type ResultDispatcher interface {
// DispatchResult returns terminal acknowledgement detail; errors may leave
// external effects and must not fabricate certainty.
DispatchResult(context.Context, controlplane.Command) (DispatchOutcome, error)
}
ResultDispatcher reports an honest terminal data-plane acknowledgement. Legacy Dispatcher implementations remain supported for workload adapters.
type RoutingDispatcher ¶
type RoutingDispatcher struct {
// contains filtered or unexported fields
}
RoutingDispatcher keeps Kubernetes scaling separate from queue control commands.
func NewRoutingDispatcher ¶
func NewRoutingDispatcher(dataPlane Dispatcher, workloads Dispatcher) (*RoutingDispatcher, error)
NewRoutingDispatcher creates an action router with explicit boundaries.
func (*RoutingDispatcher) Dispatch ¶
func (dispatcher *RoutingDispatcher) Dispatch(ctx context.Context, command controlplane.Command) error
Dispatch routes scaling to Kubernetes and every data-plane command to its queue adapter.
func (*RoutingDispatcher) DispatchResult ¶
func (dispatcher *RoutingDispatcher) DispatchResult( ctx context.Context, command controlplane.Command, ) (DispatchOutcome, error)
DispatchResult preserves structured queue acknowledgements while assigning an orchestration completion time to legacy workload adapters.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service sequences authorization, idempotency, audit, dispatch, and outcome persistence without implementing worker/backend semantics or owning shutdown. Construct before use with nonnil collaborators and a LifecycleJournal. Panics and blocking calls are not isolated; safe shared use follows collaborators.
func NewInstrumentedService ¶
func NewInstrumentedService( authorizer Authorizer, journal Journal, dispatcher Dispatcher, now func() time.Time, meter metric.Meter, ) (*Service, error)
NewInstrumentedService creates a service with bounded command telemetry.
func NewService ¶
func NewService( authorizer Authorizer, journal Journal, dispatcher Dispatcher, now func() time.Time, ) *Service
NewService retains borrowed dependencies without validating the whole graph. Execute refuses journals without LifecycleJournal before dispatch; other missing or unusable collaborators can panic when invoked.
func (*Service) Execute ¶
func (s *Service) Execute( ctx context.Context, command controlplane.Command, ) (controlplane.CommandResult, error)
Execute makes at most one dispatch attempt per atomically admitted journal record, provided Journal honors its contract. Duplicate keys return stored outcomes; they do not retry external work. Authorization precedes admission, including separate replay-destination permission. Cancellation is recorded only before dispatch; later failure can leave external effects or an unknown outcome that must be inspected/reconciled. Command option pointers are borrowed for this synchronous call and must remain stable.
type UnavailableDispatcher ¶
type UnavailableDispatcher struct{}
UnavailableDispatcher explicitly represents a disconnected data plane.
func (UnavailableDispatcher) Dispatch ¶
func (UnavailableDispatcher) Dispatch(context.Context, controlplane.Command) error
Dispatch fails closed without issuing any backend operation.
Source Files