Help us shape the future of Go! Take the 2026 Go Developer Survey

controlplane

package module
v6.0.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 28, 2026 License: MIT Imports: 6 Imported by: 0

README

queue-control-plane

CI CodeQL Documentation Go Reference Release Go License

queue-control-plane is the administrative control plane for queue. It provides durable, tenant-scoped commands, desired state, audit history, an HTTP API, an administrative CLI, and an optional narrow Kubernetes Deployment adapter.

See Releases for the latest published version. A backend-neutral adapter maps tenant-scoped commands and acknowledgements through published queue management contracts. The optional tenant management document now enables the authenticated status, command, and record transport; endpoints supply the managed root queue as their native lifecycle controller and a native management.RecordReader for failure workflows. Kubernetes scale commands work only when the canonical adapters/kubernetes integration is configured. Redis Streams and Valkey Streams workers can publish native worker and queue status through the same queue management HTTP handler. Managed queues can also consume durable desired state through the typed client. See Current capability status before evaluating a rollout.

Install

The current major is /v6. It adopts Authentication v3, Authorization v4 and Identifier v2 on top of the v5 Queue v2, PostgreSQL v3, Migrations v4 and Telemetry v4 cohort; see the v6 migration guide:

go get github.com/faustbrian/go-queue-control-plane/v6@v6.0.1
go install github.com/faustbrian/go-queue-control-plane/v6/cmd/queue-control-plane@v6.0.1
go install github.com/faustbrian/go-queue-control-plane/v6/cmd/queue-control@v6.0.1

Earlier majors remain published at their own module paths but receive no further releases; only the latest major receives security fixes.

The Go-installed server does not carry the release pipeline's commit and build time metadata, so its /version response uses the development build identity. The public v2.0.1 assets are source-only, not deployable binaries or images; see the release process for their provenance scope.

Five-minute local start

Prerequisites: Go 1.27.0 or newer and an empty PostgreSQL database reachable through DATABASE_URL. The installed queue-control-plane and queue-control binaries must be on PATH.

Create /tmp/queue-control-access.json outside version control:

{
  "keys": [
    {"id": "local-cli", "key": "replace-this-secret", "subject": "operator-1"}
  ],
  "acl": [
    {
      "id": "view-audit",
      "subject": "operator-1",
      "tenant": "tenant-1",
      "action": "audit_view",
      "resource_type": "workload",
      "resource_id": "audit",
      "effect": "allow"
    }
  ]
}

Apply the embedded schema with a go-migrations v4 runner owned by your application, for example:

// database is a *sql.DB opened with sql.Open("pgx", dsn) after importing
// github.com/jackc/pgx/v5/stdlib.
source, err := controlpostgres.MigrationSource()
if err != nil {
	return err
}
backend, err := migrationpostgres.New(database) // go-migrations/v4/postgres
if err != nil {
	return err
}
runner, err := migrations.NewRunner(source, backend)
if err != nil {
	return err
}
if _, err := runner.Up(ctx); err != nil {
	return err
}

Then start the API:

export DATABASE_URL='postgres://user:password@localhost/control_plane?sslmode=disable'
export QUEUE_CONTROL_ACCESS_FILE=/tmp/queue-control-access.json
queue-control-plane

In another shell, verify the public probes and authenticated CLI:

curl --fail http://localhost:8080/health/live
curl --fail http://localhost:8080/health/ready

export QUEUE_CONTROL_URL=http://localhost:8080
export QUEUE_CONTROL_KEY_ID=local-cli
export QUEUE_CONTROL_KEY=replace-this-secret
queue-control audit list --tenant tenant-1

Do not commit the local access document. For production, inject it from a secret volume and apply the schema through the deploying application's migration runner before starting serving replicas.

Documentation

For ecosystem-wide package selection and ownership guidance, see the versioned Golib ecosystem index and its Persistence and durability family.

Development

For standalone untagged tests:

GOWORK=off go test ./... -count=1 -timeout=3m

make check and make ci are bounded public-verifier routes with the actual PostgreSQL, Redis, Valkey and integration-tag prerequisites. Both retain the cheap no-Go documentation boundary; hosted Required also needs that boundary to pass. Explicit make check-full retains full integration, disaster recovery, browser, benchmark and release-risk qualification. See CONTRIBUTING.md for focused fuzzing, cancellation/shutdown checks, and fixture requirements.

The same server image supports a separate bounded retention-only Job. See Deployment and configuration for their strict inputs and failure semantics. Retention verifies and advances the audit anchor before removing unreferenced old terminal commands; active and current desired-state commands remain durable.

The API, typed client, and CLI expose newest-first tenant command history with bounded opaque-cursor pagination in addition to point lookup by idempotency key.

Full qualification compares the exported Go module surface with the reviewed API baseline.

License

This project is licensed under the MIT License.

Continue with the documentation index.

Documentation

Overview

Package controlplane defines administrative commands, permissions and result validation for the HTTP/CLI/control-service boundaries. Start with Command and its Validate method; it validates shape, not authentication or permission. Optional PostgreSQL, queue-management and Kubernetes composition own their external resources. Domain values containing pointers are not automatically immutable or safe to mutate while a borrowed operation is in progress.

Index

Constants

View Source
const (
	// MaxIdentityBytes matches the durable and data-plane identity bounds.
	MaxIdentityBytes = 256
	// MaxReasonBytes bounds actor-supplied administrative audit reasons.
	MaxReasonBytes = 1_024
	// MaxFailureBytes bounds stable public command failure codes.
	MaxFailureBytes = 256
)
View Source
const (
	// DefaultCommandLifetime is the enforcement window assigned when a caller
	// supplies no deadline; explicit windows may extend to MaxCommandLifetime.
	DefaultCommandLifetime time.Duration = 30_000_000_000
	// MaxCommandLifetime prevents administrative work from remaining live
	// indefinitely across request and adapter boundaries.
	MaxCommandLifetime time.Duration = 300_000_000_000
)
View Source
const (
	// FailureDispatch is the redacted public code for a data-plane dispatch
	// failure. Raw adapter errors must not be exposed through administrative API
	// models because they may contain credentials or backend endpoints.
	FailureDispatch = "dispatch_failed"
	// FailureOutcomeUnknown reports that enforcement may have occurred but no
	// reliable acknowledgement was available.
	FailureOutcomeUnknown = "outcome_unknown"
	// FailureInvalidDispatchResult reports malformed adapter output without
	// exposing its contents.
	FailureInvalidDispatchResult = "invalid_dispatch_result"
	// FailureDeadlineExceeded reports a command whose enforcement window closed
	// before it reached a tenant controller.
	FailureDeadlineExceeded = "deadline_exceeded"
	// FailureCanceled reports cancellation before a command crossed the
	// durable dispatch boundary. Once dispatched, cancellation is not claimed.
	FailureCanceled = "canceled"
)
View Source
const MaxBulkSelection uint32 = 1_000

MaxBulkSelection is the largest destructive selection accepted by one command. Callers must paginate larger administrative workflows.

View Source
const MaxScaleReplicas uint32 = 10_000

MaxScaleReplicas bounds one explicitly authorized scaling request.

Variables

This section is empty.

Functions

func NewCommandID

func NewCommandID() (string, error)

NewCommandID allocates one opaque lowercase ULID operation identifier.

Types

type Action

type Action string

Action identifies an administrative mutation.

const (
	// ActionPause requests a queue or worker-group admission pause; it does
	// not join active jobs.
	ActionPause Action = "pause"
	// ActionResume requests resumption for a queue or worker group.
	ActionResume Action = "resume"
	// ActionDrain requests worker or worker-group draining without immediate
	// termination.
	ActionDrain Action = "drain"
	// ActionTerminate requests worker or worker-group termination.
	ActionTerminate Action = "terminate"
	// ActionRetry retries one failure or dead-letter record.
	ActionRetry Action = "retry"
	// ActionBulkRetry retries a confirmed bounded record selection.
	ActionBulkRetry Action = "bulk_retry"
	// ActionDelete deletes one failure or dead-letter record.
	ActionDelete Action = "delete"
	// ActionPurge purges a confirmed queue or record target.
	ActionPurge Action = "purge"
	// ActionReplay replays a confirmed record into an explicitly authorized
	// destination.
	ActionReplay Action = "replay"
	// ActionScale requests a bounded Kubernetes workload replica count.
	ActionScale Action = "scale"
)

type Command

type Command struct {
	// CommandID is an optional bounded operation identity; Service/Journal
	// generate it when absent.
	CommandID string
	// IdempotencyKey is the required tenant-scoped duplicate/conflict key, not
	// authorization.
	IdempotencyKey string
	// TenantID is required bounded scope; a trusted ingress must authenticate
	// and authorize it.
	TenantID string
	// Actor is required bounded attribution; authenticated ingress owns its
	// binding to a principal.
	Actor string
	// AuthenticationMethod is optional bounded audit attribution; orchestration
	// defaults it to internal, without authenticating anyone.
	AuthenticationMethod string
	// Reason is required nonblank audit text bounded by MaxReasonBytes; callers
	// must omit credentials.
	Reason string
	// Action selects a supported mutation and its required
	// target/confirmation/options.
	Action Action
	// Capability is optional bounded audit attribution; orchestration defaults
	// it to the action text.
	Capability string
	// Target identifies the scoped resource; action/target combinations are
	// validated.
	Target Target
	// RequestedAt is required and anchors the enforcement lifetime; validation
	// does not replace caller time.
	RequestedAt time.Time
	// Deadline is optional; orchestration defaults it to RequestedAt plus
	// DefaultCommandLifetime, and supplied values must follow RequestedAt
	// within MaxCommandLifetime.
	Deadline time.Time
	// Confirmed must be true for purge, bulk retry, replay and scaling to zero.
	Confirmed bool
	// Selection is required for bulk retry; its pointer remains caller-owned
	// outside explicit copying boundaries.
	Selection *Selection
	// Replay is required for replay, including an explicit destination and
	// duplicate policy.
	Replay *Replay
	// Scale is required for scaling; zero replicas additionally require
	// confirmation.
	Scale *Scale
}

Command is the mandatory envelope for every administrative mutation.

func (Command) Validate

func (c Command) Validate() error

Validate rejects incomplete and unsupported mutation envelopes.

type CommandResult

type CommandResult struct {
	// CommandID is an optional bounded operation identity retained for legacy
	// result compatibility.
	CommandID string `json:"command_id"`
	// IdempotencyKey is the required scoped journal lookup key.
	IdempotencyKey string `json:"idempotency_key"`
	// TenantID is the required bounded journal scope.
	TenantID string `json:"tenant_id"`
	// Status selects a validated lifecycle state; a returned unknown result
	// need not itself have been durably saved.
	Status CommandStatus `json:"status"`
	// Failure is required bounded nonblank text for unsuccessful terminal
	// states and absent otherwise; use secret-safe public codes.
	Failure string `json:"failure,omitempty"`
	// WorkerID is optional bounded acknowledgement attribution and must
	// accompany Protocol.
	WorkerID string `json:"worker_id,omitempty"`
	// Protocol is optional worker protocol attribution; when present its major
	// must be nonzero and WorkerID must be present.
	Protocol *ProtocolVersion `json:"protocol,omitempty"`
	// CapabilityAvailable optionally distinguishes a reported capability from
	// an unspecified acknowledgement; nil means not supplied.
	CapabilityAvailable *bool `json:"capability_available,omitempty"`
	// DispatchedAt records the pre-dispatch boundary, not proof that a backend
	// accepted the operation.
	DispatchedAt time.Time `json:"dispatched_at,omitempty"`
	// AcknowledgedAt records acknowledgement after dispatch;
	// accepted/pending/dispatched states may not carry it.
	AcknowledgedAt time.Time `json:"acknowledged_at,omitempty"`
	// CompletedAt is required for terminal states and absent before completion.
	CompletedAt time.Time `json:"completed_at,omitempty"`
}

CommandResult is the durable result associated with an idempotency key.

func (CommandResult) Validate

func (r CommandResult) Validate() error

Validate rejects malformed or internally inconsistent durable results.

type CommandStatus

type CommandStatus string

CommandStatus is the durable administrative outcome presented to clients.

const (
	// CommandPending identifies pending dispatch; the native Journal stores
	// this state as its initial durable admission result.
	CommandPending CommandStatus = "pending"
	// CommandAccepted identifies durable acceptance before dispatch.
	CommandAccepted CommandStatus = "accepted"
	// CommandDispatched identifies the durable pre-dispatch recovery boundary,
	// not enforcement proof.
	CommandDispatched CommandStatus = "dispatched"
	// CommandAcknowledged identifies recorded backend acknowledgement before
	// completion.
	CommandAcknowledged CommandStatus = "acknowledged"
	// CommandSucceeded identifies a successful terminal result.
	CommandSucceeded CommandStatus = "succeeded"
	// CommandFailed identifies an unsuccessful terminal result; backend effects
	// may be partial.
	CommandFailed CommandStatus = "failed"
	// CommandUnsupported identifies a terminal unsupported capability.
	CommandUnsupported CommandStatus = "unsupported"
	// CommandTimedOut identifies an expired terminal enforcement outcome.
	CommandTimedOut CommandStatus = "timed_out"
	// CommandPartial identifies a terminal operation with partial effects.
	CommandPartial CommandStatus = "partial"
	// CommandUnknown identifies a terminal result without reliable enforcement
	// certainty.
	CommandUnknown CommandStatus = "unknown"
	// CommandCanceled identifies cancellation before dispatch, not cancellation
	// of previously dispatched work.
	CommandCanceled CommandStatus = "canceled"
)

type Permission

type Permission string

Permission names an authorization capability required by an operation. Naming a permission does not authenticate a principal or grant access.

const (
	// PermissionView permits ordinary administrative views.
	PermissionView Permission = "view"
	// PermissionPause permits pausing supported resources.
	PermissionPause Permission = "pause"
	// PermissionResume permits resuming supported resources.
	PermissionResume Permission = "resume"
	// PermissionDrain permits worker draining.
	PermissionDrain Permission = "drain"
	// PermissionTerminate permits worker termination.
	PermissionTerminate Permission = "terminate"
	// PermissionRetry permits one record retry.
	PermissionRetry Permission = "retry"
	// PermissionBulkRetry permits bounded bulk retries.
	PermissionBulkRetry Permission = "bulk_retry"
	// PermissionDelete permits record deletion.
	PermissionDelete Permission = "delete"
	// PermissionPurge permits confirmed purging.
	PermissionPurge Permission = "purge"
	// PermissionReplay permits replay, including separate destination
	// authorization.
	PermissionReplay Permission = "replay"
	// PermissionScale permits workload scaling.
	PermissionScale Permission = "scale"
	// PermissionRecordList permits safe record metadata listings.
	PermissionRecordList Permission = "record_list"
	// PermissionRecordInspect permits safe record metadata inspection.
	PermissionRecordInspect Permission = "record_inspect"
	// PermissionPayloadView permits separately audited payload disclosure.
	PermissionPayloadView Permission = "payload_view"
	// PermissionDiagnosticsView permits separately audited diagnostic
	// disclosure.
	PermissionDiagnosticsView Permission = "diagnostics_view"
	// PermissionAuditView permits tenant audit history.
	PermissionAuditView Permission = "audit_view"
	// PermissionRetentionConfigure permits retention-policy configuration.
	PermissionRetentionConfigure Permission = "retention_configure"
)

type ProtocolVersion

type ProtocolVersion struct {
	// Major is the nonzero major identity required when attached to a
	// CommandResult.
	Major uint16 `json:"major"`
	// Minor is the worker's reported minor protocol revision.
	Minor uint16 `json:"minor"`
}

ProtocolVersion identifies the data-plane protocol that acknowledged a command without coupling this package to an adapter implementation.

type Replay

type Replay struct {
	// Destination is a required nonblank queue identity; Service separately
	// authorizes this destination.
	Destination string
	// IdempotencyPolicy must explicitly select rejection or replacement of
	// destination duplicates.
	IdempotencyPolicy ReplayPolicy
}

Replay contains the explicit destination and idempotency semantics required for a replay command.

type ReplayPolicy

type ReplayPolicy string

ReplayPolicy declares how duplicate destination identities are handled.

const (
	// ReplayRejectDuplicate refuses an existing destination identity.
	ReplayRejectDuplicate ReplayPolicy = "reject_duplicate"
	// ReplayReplaceDuplicate explicitly permits replacement of an existing
	// destination identity.
	ReplayReplaceDuplicate ReplayPolicy = "replace_duplicate"
)

type Scale

type Scale struct {
	// Replicas is at most MaxScaleReplicas; zero requires command confirmation.
	Replicas uint32
}

Scale declares the desired Kubernetes workload replica count.

type Selection

type Selection struct {
	// Limit must be positive and at most MaxBulkSelection; it is a per-command
	// selection bound.
	Limit uint32
}

Selection bounds a bulk administrative mutation.

type SensitiveAccess

type SensitiveAccess struct {
	// CommandID is a required bounded identity linking this privileged access
	// to its audit record.
	CommandID string
	// TenantID is required tenant scope; the record itself does not
	// authenticate it.
	TenantID string
	// Actor is required bounded authenticated attribution supplied by ingress.
	Actor string
	// Permission must be PermissionPayloadView or PermissionDiagnosticsView.
	Permission Permission
	// Target must identify a bounded failure or dead-letter record.
	Target Target
	// OccurredAt is required observation time for the privileged access.
	OccurredAt time.Time
}

SensitiveAccess is one fail-closed audit record for privileged record data.

func (SensitiveAccess) Validate

func (a SensitiveAccess) Validate() error

Validate rejects unscoped or non-sensitive access audit records.

type Target

type Target struct {
	// Kind selects the resource class; supported classes depend on the action.
	Kind TargetKind
	// Name is required nonblank identity text bounded by MaxIdentityBytes;
	// validation does not trim it.
	Name string
}

Target identifies an administrative resource without exposing backend addressing or queue serialization details.

type TargetKind

type TargetKind string

TargetKind identifies the resource affected by a command.

const (
	// TargetQueue identifies a tenant-scoped queue.
	TargetQueue TargetKind = "queue"
	// TargetWorker identifies one tenant-scoped worker.
	TargetWorker TargetKind = "worker"
	// TargetWorkerGroup identifies a tenant-scoped worker group.
	TargetWorkerGroup TargetKind = "worker_group"
	// TargetFailure identifies a failure record or its supported administrative
	// selection.
	TargetFailure TargetKind = "failure"
	// TargetDeadLetter identifies a dead-letter record or its supported
	// administrative selection.
	TargetDeadLetter TargetKind = "dead_letter"
	// TargetWorkload identifies a Kubernetes workload, not a queue worker
	// process.
	TargetWorkload TargetKind = "workload"
)

type ValidationError

type ValidationError struct {
	// Field names the invalid input boundary and is printed by Error; keep
	// labels non-sensitive.
	Field string
	// Problem is printed by Error; use rule descriptions rather than rejected
	// values.
	Problem string
}

ValidationError is a machine-readable public contract for invalid input.

func (*ValidationError) Error

func (e *ValidationError) Error() string

Error implements error.

Directories

Path Synopsis
_browser
testserver command
Package main hosts synthetic loopback surfaces for the optional Chromium assurance harness.
Package main hosts synthetic loopback surfaces for the optional Chromium assurance harness.
adapters
kubernetes
Package kubernetes is the narrow Kubernetes Deployment integration.
Package kubernetes is the narrow Kubernetes Deployment integration.
Package alerts derives bounded alert inputs from validated operational snapshots without delivering notifications or storing time series.
Package alerts derives bounded alert inputs from validated operational snapshots without delivering notifications or storing time series.
Package apihttp provides versioned administrative HTTP routes and public JSON models.
Package apihttp provides versioned administrative HTTP routes and public JSON models.
Package authz maps an already authenticated context principal to scoped authorization decisions.
Package authz maps an already authenticated context principal to scoped authorization decisions.
Package cli implements the administrative command-line workflow.
Package cli implements the administrative command-line workflow.
Package client provides typed administrative HTTP access.
Package client provides typed administrative HTTP access.
cmd
queue-control command
Package main implements the queue-control client command.
Package main implements the queue-control client command.
queue-control-plane command
Package main implements the deployable administrative service and its one-shot retention mode.
Package main implements the deployable administrative service and its one-shot retention mode.
Package control orchestrates authorization, durable idempotent admission, dispatch and lifecycle audit without enforcing queue delivery or supervising workers.
Package control orchestrates authorization, durable idempotent admission, dispatch and lifecycle audit without enforcing queue delivery or supervising workers.
Package dataplane adapts stable queue management contracts without acquiring backend-native clients or reimplementing queue semantics.
Package dataplane adapts stable queue management contracts without acquiring backend-native clients or reimplementing queue semantics.
Package fleet models reported worker liveness and protocol/capability compatibility without supervising processes or implementing delivery.
Package fleet models reported worker liveness and protocol/capability compatibility without supervising processes or implementing delivery.
Package history seals and verifies value-based administrative audit hash chains.
Package history seals and verifies value-based administrative audit hash chains.
Package postgres provides command/desired-state transactions and tenant audit history for explicit control-plane composition.
Package postgres provides command/desired-state transactions and tenant audit history for explicit control-plane composition.
Package server composes explicit authentication/browser admission and an HTTP server lifecycle.
Package server composes explicit authentication/browser admission and an HTTP server lifecycle.
Package ui serves the optional embedded administrative web console.
Package ui serves the optional embedded administrative web console.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL