controlplane

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.

The project is under active development. A backend-neutral adapter now 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 in-cluster adapter 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.

Five-minute local start

Prerequisites: Go 1.26.6 or newer and an empty PostgreSQL database reachable through DATABASE_URL.

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
go run ./cmd/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
go run ./cmd/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

• Architecture and trust boundaries
• HTTP API reference
• CLI reference
• Embedded web UI guide
• Deployment and configuration
• Compatibility and current capability status
• Hardening evidence and release gates
• Security and privacy
• Operations, retention, backup, and incidents
• Performance and load benchmarks
• Kubernetes, HPA, and KEDA
• Laravel Horizon migration matrix
• Troubleshooting and FAQ
• Release process
• Security reporting
• Changelog

Development

Run the deterministic local gate with:

make check

This checks formatting, module tidiness and checksums, vet, Staticcheck, strict golangci-lint, tests, the race detector, exact per-package 100% statement coverage, and builds. make nilaway runs the pinned advisory NilAway profile, and make fuzz runs the bounded fuzz smoke suite. make integration-postgres starts a disposable PostgreSQL 18 container and runs the real persistence contract under the race detector. make benchmarks runs eight single-core large-fleet, API, payload, audit, reconnect, and backend-outage samples with enforced allocation budgets. See CONTRIBUTING.md for repository expectations.

make security runs the pinned Go vulnerability scanner against the canonical Go vulnerability database and fails on reachable findings.

make mutation requires 100% mutant coverage and efficacy across the public command contract, authorization mapping, desired state, dispatch, and command orchestration.

make disaster-recovery-postgres creates isolated PostgreSQL 18 source and restore databases, takes a native logical backup, restores it, and verifies the complete control and audit state through the production repositories.

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.

make api-compatibility compares the complete exported Go module surface with the reviewed baseline and fails on compatible or incompatible drift.

License

This project is licensed under the MIT License. A production release is not ready until all release gates described in the project objective are complete.

Ecosystem

Use the Golib documentation portal to choose companion packages, supported stacks, recipes, and operations guidance.

Documentation

Overview

Package controlplane defines the public administrative domain contracts.

Index

• Constants
• func NewCommandID() (string, error)
• type Action
• type Command
  • func (c Command) Validate() error
• type CommandResult
  • func (r CommandResult) Validate() error
• type CommandStatus
• type Permission
• type ProtocolVersion
• type Replay
• type ReplayPolicy
• type Scale
• type Selection
• type SensitiveAccess
  • func (a SensitiveAccess) Validate() error
• type Target
• type TargetKind
• type ValidationError
  • func (e *ValidationError) Error() string

Constants

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
)

const (
	// DefaultCommandLifetime is the bounded enforcement window assigned when a
	// caller does not supply a narrower deadline.
	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
)

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"
)

const MaxBulkSelection uint32 = 1_000

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

const MaxScaleReplicas uint32 = 10_000

MaxScaleReplicas bounds one explicitly authorized scaling request.

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     Action = "pause"
	ActionResume    Action = "resume"
	ActionDrain     Action = "drain"
	ActionTerminate Action = "terminate"
	ActionRetry     Action = "retry"
	ActionBulkRetry Action = "bulk_retry"
	ActionDelete    Action = "delete"
	ActionPurge     Action = "purge"
	ActionReplay    Action = "replay"
	ActionScale     Action = "scale"
)

type Command

type Command struct {
	CommandID            string
	IdempotencyKey       string
	TenantID             string
	Actor                string
	AuthenticationMethod string
	Reason               string
	Action               Action
	Capability           string
	Target               Target
	RequestedAt          time.Time
	Deadline             time.Time
	Confirmed            bool
	Selection            *Selection
	Replay               *Replay
	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           string           `json:"command_id"`
	IdempotencyKey      string           `json:"idempotency_key"`
	TenantID            string           `json:"tenant_id"`
	Status              CommandStatus    `json:"status"`
	Failure             string           `json:"failure,omitempty"`
	WorkerID            string           `json:"worker_id,omitempty"`
	Protocol            *ProtocolVersion `json:"protocol,omitempty"`
	CapabilityAvailable *bool            `json:"capability_available,omitempty"`
	DispatchedAt        time.Time        `json:"dispatched_at,omitempty"`
	AcknowledgedAt      time.Time        `json:"acknowledged_at,omitempty"`
	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      CommandStatus = "pending"
	CommandAccepted     CommandStatus = "accepted"
	CommandDispatched   CommandStatus = "dispatched"
	CommandAcknowledged CommandStatus = "acknowledged"
	CommandSucceeded    CommandStatus = "succeeded"
	CommandFailed       CommandStatus = "failed"
	CommandUnsupported  CommandStatus = "unsupported"
	CommandTimedOut     CommandStatus = "timed_out"
	CommandPartial      CommandStatus = "partial"
	CommandUnknown      CommandStatus = "unknown"
	CommandCanceled     CommandStatus = "canceled"
)

type Permission

type Permission string

Permission is the explicit authorization capability required by an action.

const (
	PermissionView               Permission = "view"
	PermissionPause              Permission = "pause"
	PermissionResume             Permission = "resume"
	PermissionDrain              Permission = "drain"
	PermissionTerminate          Permission = "terminate"
	PermissionRetry              Permission = "retry"
	PermissionBulkRetry          Permission = "bulk_retry"
	PermissionDelete             Permission = "delete"
	PermissionPurge              Permission = "purge"
	PermissionReplay             Permission = "replay"
	PermissionScale              Permission = "scale"
	PermissionRecordList         Permission = "record_list"
	PermissionRecordInspect      Permission = "record_inspect"
	PermissionPayloadView        Permission = "payload_view"
	PermissionDiagnosticsView    Permission = "diagnostics_view"
	PermissionAuditView          Permission = "audit_view"
	PermissionRetentionConfigure Permission = "retention_configure"
)

type ProtocolVersion

type ProtocolVersion struct {
	Major uint16 `json:"major"`
	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       string
	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  ReplayPolicy = "reject_duplicate"
	ReplayReplaceDuplicate ReplayPolicy = "replace_duplicate"
)

type Scale

type Scale struct {
	Replicas uint32
}

Scale declares the desired Kubernetes workload replica count.

type Selection

type Selection struct {
	Limit uint32
}

Selection bounds a bulk administrative mutation.

type SensitiveAccess

type SensitiveAccess struct {
	CommandID  string
	TenantID   string
	Actor      string
	Permission Permission
	Target     Target
	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 TargetKind
	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       TargetKind = "queue"
	TargetWorker      TargetKind = "worker"
	TargetWorkerGroup TargetKind = "worker_group"
	TargetFailure     TargetKind = "failure"
	TargetDeadLetter  TargetKind = "dead_letter"
	TargetWorkload    TargetKind = "workload"
)

type ValidationError

type ValidationError struct {
	Field   string
	Problem string
}

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

func (*ValidationError) Error

func (e *ValidationError) Error() string

Error implements error.