The highest tagged major version is v6.

controlplane

package module

v4.0.1 Latest Latest Go to latest Published: Oct 22, 2026 License: MIT Imports: 6 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/faustbrian/go-queue-control-plane

Links

Open Source Insights

README ¶

queue-control-plane

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. The released kubernetes path remains as a deprecated compatibility path. 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

Main now selects the candidate /v4 module for bounded rejection detail and explicit known-state reporting. V4 is not published by this source change; see the v4 migration guide and confirm release availability before adoption. The commands below retain existing published v3 choices.

Published v3.0.0 uses root v3 with Authentication v2. The commands below pin that release and retain the supported historical v1 and v2 choices. See the v3 migration guide for the coherent nominal import migration and finite authentication admission. Published v3.1.1 also includes PostgreSQL v2 pool integration. Select that release or a later compatible v3 maintenance release when adopting the pool integration; the v3.0.0 pins below do not include it.

Use the official major suffix for published v3.0.0:

go get github.com/faustbrian/go-queue-control-plane/v3@v3.0.0
go install github.com/faustbrian/go-queue-control-plane/v3/cmd/queue-control-plane@v3.0.0
go install github.com/faustbrian/go-queue-control-plane/v3/cmd/queue-control@v3.0.0

The published v1 line remains available. Add that module to an application with:

go get github.com/faustbrian/go-queue-control-plane@latest

Install the server and administrative CLI for local evaluation with:

go install github.com/faustbrian/go-queue-control-plane/cmd/queue-control-plane@latest
go install github.com/faustbrian/go-queue-control-plane/cmd/queue-control@latest

Pin the published v1.1.2 release exactly with:

go get github.com/faustbrian/go-queue-control-plane@v1.1.2
go install github.com/faustbrian/go-queue-control-plane/cmd/queue-control-plane@v1.1.2
go install github.com/faustbrian/go-queue-control-plane/cmd/queue-control@v1.1.2

The supported security-related v2.0.1 release is published. Use the official major suffix:

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

The public migration helpers now compose Migrations v2 types. See the v2 migration guide for the nominal API change and unchanged schema history. The Go floor remains 1.27.0.

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"
    }
  ]
}

Start the API and apply its embedded migration:

export DATABASE_URL='postgres://user:password@localhost/control_plane?sslmode=disable'
export QUEUE_CONTROL_ACCESS_FILE=/tmp/queue-control-access.json
export QUEUE_CONTROL_RUN_MIGRATIONS=true
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 run a one-shot QUEUE_CONTROL_MIGRATE_ONLY=true Job 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 separate migrate-only and bounded retention-only Jobs. 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.

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.

Source Files ¶

View all Source files

command.go

Directories ¶

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

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 exposes the target-oriented entry point for the narrow Kubernetes Deployment integration.	Package kubernetes exposes the target-oriented entry point for the narrow Kubernetes Deployment integration.
alerts 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.
apihttp Package apihttp provides versioned administrative HTTP routes and public JSON models.	Package apihttp provides versioned administrative HTTP routes and public JSON models.
authz Package authz maps an already authenticated context principal to scoped authorization decisions.	Package authz maps an already authenticated context principal to scoped authorization decisions.
cli Package cli implements the administrative command-line workflow.	Package cli implements the administrative command-line workflow.
client 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 one-shot migration/retention modes.	Package main implements the deployable administrative service and one-shot migration/retention modes.
control 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.
dataplane 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.
fleet 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.
history Package history seals and verifies value-based administrative audit hash chains.	Package history seals and verifies value-based administrative audit hash chains.
kubernetes Package kubernetes preserves the released path for the deliberately narrow Kubernetes integration.	Package kubernetes preserves the released path for the deliberately narrow Kubernetes integration.
postgres 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.
server Package server composes explicit authentication/browser admission and an HTTP server lifecycle.	Package server composes explicit authentication/browser admission and an HTTP server lifecycle.
ui Package ui serves the optional embedded administrative web console.	Package ui serves the optional embedded administrative web console.

README ¶

queue-control-plane

Install

Five-minute local start

Documentation

Development

License

Documentation ¶

Overview ¶

Index ¶

Constants ¶

Variables ¶

Functions ¶

func NewCommandID ¶

Types ¶

type Action ¶

type Command ¶

func (Command) Validate ¶

type CommandResult ¶

func (CommandResult) Validate ¶

type CommandStatus ¶

type Permission ¶

type ProtocolVersion ¶

type Replay ¶

type ReplayPolicy ¶

type Scale ¶

type Selection ¶

type SensitiveAccess ¶

func (SensitiveAccess) Validate ¶

type Target ¶

type TargetKind ¶

type ValidationError ¶

func (*ValidationError) Error ¶

Source Files ¶

Directories ¶