management

package
v1.0.1 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Overview

Package management defines stable worker/control-plane contracts without implementing queue delivery or backend operations.

Index

Examples

Constants

View Source 
const (
	// FailureCodeUnsupportedPayloadVersion identifies a syntactically valid
	// payload whose declared schema version cannot be processed.
	FailureCodeUnsupportedPayloadVersion = "unsupported_payload_version"
	// FailureCodeLeaseLost identifies a delivery that can no longer be settled
	// by the current owner.
	FailureCodeLeaseLost = "lease_lost"
	// FailureCodeDeadLetterDestinationUnavailable identifies a failed durable
	// append to the configured terminal destination.
	FailureCodeDeadLetterDestinationUnavailable = "dead_letter_destination_unavailable"
	// FailureCodeAdministrativeQuarantine identifies an explicit operator
	// decision to prevent normal redelivery.
	FailureCodeAdministrativeQuarantine = "administrative_quarantine"
)
View Source 
const (
	// MaxReasonBytes bounds actor-supplied administrative reasons.
	MaxReasonBytes = 1_024
	// MaxBulkSelection bounds one destructive bulk-retry command.
	MaxBulkSelection uint32 = 1_000
)
View Source 
const (
	// CurrentEnvelopeVersion is the first complete dead-letter record contract.
	CurrentEnvelopeVersion uint16 = 1
	// MaxAdministrativePayloadBytes bounds privileged payload inspection.
	MaxAdministrativePayloadBytes = 1 << 20
	// MaxFailureSummaryBytes bounds deliberately redacted operator diagnostics.
	MaxFailureSummaryBytes = 1_024
	// MaxRecordTags bounds user-supplied dead-letter record dimensions.
	MaxRecordTags = 32
	// MaxPageSize bounds one failure or dead-letter listing response.
	MaxPageSize uint32 = 200
	// MaxCursorBytes bounds opaque pagination state accepted from a client.
	MaxCursorBytes = 1_024
	// MaxSearchBytes bounds backend-neutral administrative search input.
	MaxSearchBytes = 256
)
View Source 
const (
	// MaxIdentityBytes bounds worker, backend, queue, and capability labels.
	MaxIdentityBytes = 256
	// MaxQueuesPerWorker bounds one heartbeat's queue cardinality.
	MaxQueuesPerWorker = 256
	// MaxCapabilitiesPerWorker bounds one heartbeat's capability cardinality.
	MaxCapabilitiesPerWorker = 64
	// MaxWorkerConcurrency bounds reported goroutine concurrency.
	MaxWorkerConcurrency = 1_000_000
	// MaxStatusPageSize bounds one worker or queue status response.
	MaxStatusPageSize = 200
)
View Source 
const MaxDesiredStateTargets = 256
View Source 
const MaxLifecycleCommandResults = 10_000
View Source 
const MaxStatusProviders = 1_000

Variables

View Source 
var (
	// ErrDesiredStateNotFound reports a target without an authored state. A
	// worker keeps its current local state and does not infer an active state.
	ErrDesiredStateNotFound = errors.New("management: desired state not found")
	// ErrInvalidDesiredStateConfiguration reports an unsafe reconciler graph.
	ErrInvalidDesiredStateConfiguration = errors.New("management: invalid desired state configuration")
	// ErrInvalidDesiredStateContext reports a nil reconciliation context.
	ErrInvalidDesiredStateContext = errors.New("management: invalid desired state context")
	// ErrInvalidDesiredStateOutput reports malformed or mismatched source data.
	ErrInvalidDesiredStateOutput = errors.New("management: invalid desired state output")
	// ErrDesiredStateRegression prevents an older revision from replacing state.
	ErrDesiredStateRegression = errors.New("management: desired state revision regression")
	// ErrDesiredStateConflict prevents one revision from describing two states.
	ErrDesiredStateConflict = errors.New("management: desired state revision conflict")
)
View Source 
var (
	// ErrRecordNotFound reports that the selected management record is absent.
	ErrRecordNotFound = errors.New("management: record not found")
	// ErrUnsupportedCapability reports an operation the backend did not advertise.
	ErrUnsupportedCapability = errors.New("management: unsupported capability")
	// ErrManagementUnavailable reports a temporarily unreachable data plane.
	ErrManagementUnavailable = errors.New("management: unavailable")
	// ErrMalformedCursor reports opaque pagination state that cannot be decoded.
	ErrMalformedCursor = errors.New("management: malformed cursor")
	// ErrInvalidFilter reports filtering the backend cannot safely honor.
	ErrInvalidFilter = errors.New("management: invalid filter")
	// ErrStaleRecord reports a record changed after it was selected or inspected.
	ErrStaleRecord = errors.New("management: stale record")
	// ErrMutationConflict reports an incompatible concurrent or duplicate mutation.
	ErrMutationConflict = errors.New("management: mutation conflict")
	// ErrPartialMutation reports a mutation with both confirmed and failed effects.
	ErrPartialMutation = errors.New("management: partial mutation")
	// ErrUnknownMutation reports a mutation whose durable outcome is ambiguous.
	ErrUnknownMutation = errors.New("management: unknown mutation outcome")
)
View Source 
var (
	// ErrInvalidWorkerLifecycleConfiguration reports incomplete target identity.
	ErrInvalidWorkerLifecycleConfiguration = errors.New("management: invalid worker lifecycle configuration")
	// ErrInvalidLifecycleCounter reports unbalanced admission or job accounting.
	ErrInvalidLifecycleCounter = errors.New("management: invalid lifecycle counter")
	// ErrDesiredStateTargetMismatch prevents state from crossing worker scopes.
	ErrDesiredStateTargetMismatch = errors.New("management: desired state target mismatch")
	// ErrInvalidDesiredStateTransition prevents unsupported target lifecycle state.
	ErrInvalidDesiredStateTransition = errors.New("management: invalid desired state transition")
	// ErrInvalidWorkerLifecycleContext reports a nil command context.
	ErrInvalidWorkerLifecycleContext = errors.New("management: invalid worker lifecycle context")
)
View Source 
var (
	// ErrInvalidStatusProviders reports an empty, nil, or oversized source set.
	ErrInvalidStatusProviders = errors.New("management: invalid status providers")
	// ErrInvalidStatusCursor reports pagination state outside the source set.
	ErrInvalidStatusCursor = errors.New("management: invalid status cursor")
	// ErrInvalidStatusProviderOutput reports malformed or duplicate observations.
	ErrInvalidStatusProviderOutput = errors.New("management: invalid status provider output")
)

Functions

This section is empty.

Types

type Capability 

type Capability string

Capability identifies an optional management or visibility operation. 

const (
	CapabilityWorkerStatus   Capability = "worker_status"
	CapabilityQueueStatus    Capability = "queue_status"
	CapabilityPause          Capability = "pause"
	CapabilityResume         Capability = "resume"
	CapabilityDrain          Capability = "drain"
	CapabilityTerminate      Capability = "terminate"
	CapabilityFailures       Capability = "failures"
	CapabilityDeadLetters    Capability = "dead_letters"
	CapabilityRetry          Capability = "retry"
	CapabilityBulkRetry      Capability = "bulk_retry"
	CapabilityDelete         Capability = "delete"
	CapabilityPurge          Capability = "purge"
	CapabilityReplay         Capability = "replay"
	CapabilityRetentionCount Capability = "retention_count"
	CapabilityRetentionTime  Capability = "retention_time"
	CapabilityRetentionBytes Capability = "retention_bytes"
)

type Classification 

type Classification string

Classification is the stable backend-neutral failure disposition used by workers, dead-letter records, and management clients. 

const (
	// ClassificationRetryable permits another delivery attempt.
	ClassificationRetryable Classification = "retryable"
	// ClassificationPermanent prevents policy-driven handler retries.
	ClassificationPermanent Classification = "permanent"
	// ClassificationMalformed identifies an undecodable or unsupported input.
	ClassificationMalformed Classification = "malformed"
	// ClassificationCanceled identifies interrupted work that is not terminal
	// unless an explicit backend policy says otherwise.
	ClassificationCanceled Classification = "canceled"
	// ClassificationInfrastructure identifies broker or settlement uncertainty.
	ClassificationInfrastructure Classification = "infrastructure"
)

func ClassifyFailure 

func ClassifyFailure(err error) Classification

ClassifyFailure returns the classification selected by ResolveFailure.

type Command 

type Command struct {
	ID             string
	IdempotencyKey string
	Actor          string
	Reason         string
	Protocol       ProtocolVersion
	Action         CommandAction
	Target         Target
	RequestedAt    time.Time
	Deadline       time.Time
	Confirmed      bool
	Selection      *Selection
	Replay         *ReplayOptions
}

Command is the stable envelope enforced by a queue management adapter.

func (Command) Validate 

func (c Command) Validate() error

Validate rejects malformed, unsupported, or unsafe control commands.

type CommandAction 

type CommandAction string

CommandAction identifies a data-plane management operation. 

const (
	CommandPause     CommandAction = "pause"
	CommandResume    CommandAction = "resume"
	CommandDrain     CommandAction = "drain"
	CommandTerminate CommandAction = "terminate"
	CommandRetry     CommandAction = "retry"
	CommandBulkRetry CommandAction = "bulk_retry"
	CommandDelete    CommandAction = "delete"
	CommandPurge     CommandAction = "purge"
	CommandReplay    CommandAction = "replay"
)

type CommandResult 

type CommandResult struct {
	CommandID      string
	IdempotencyKey string
	WorkerID       string
	Protocol       ProtocolVersion
	Status         CommandResultStatus
	FailureCode    string
	CompletedAt    time.Time
}

CommandResult is the redacted data-plane acknowledgement for one command. FailureCode must be a stable code and must never contain payloads, credentials, tokens, or backend endpoints.

func (CommandResult) Validate 

func (r CommandResult) Validate() error

Validate rejects incomplete or disclosure-prone command results.

type CommandResultStatus 

type CommandResultStatus string

CommandResultStatus describes an acknowledged or safely unknown outcome. 

const (
	CommandAcknowledged CommandResultStatus = "acknowledged"
	CommandRejected     CommandResultStatus = "rejected"
	CommandFailed       CommandResultStatus = "failed"
	CommandUnsupported  CommandResultStatus = "unsupported"
	CommandTimedOut     CommandResultStatus = "timed_out"
	CommandPartial      CommandResultStatus = "partial"
	CommandUnknown      CommandResultStatus = "unknown"
)

type Compatibility 

type Compatibility struct {
	State            CompatibilityState
	Enabled          []Capability
	WorkerOnly       []Capability
	ControlPlaneOnly []Capability
}

Compatibility is the deterministic result of version and capability negotiation.

func Negotiate 

func Negotiate(
	supported ProtocolRange,
	worker ProtocolVersion,
	workerCapabilities []Capability,
	controlPlaneCapabilities []Capability,
) Compatibility

Negotiate enables only capabilities both peers report at a compatible protocol version. Unknown and incompatible peers remain visible but are not controllable.

type CompatibilityState 

type CompatibilityState string

CompatibilityState describes a worker relative to a supported protocol range. 

const (
	CompatibilityCompatible  CompatibilityState = "compatible"
	CompatibilityWorkerOlder CompatibilityState = "worker_older"
	CompatibilityWorkerNewer CompatibilityState = "worker_newer"
	CompatibilityUnknown     CompatibilityState = "unknown"
)

type Controller 

type Controller interface {
	Execute(context.Context, Command) (CommandResult, error)
}

Controller is implemented by queue adapters that can safely enforce management commands. A control plane depends on this interface rather than issuing backend commands directly.

type DesiredRecord 

type DesiredRecord struct {
	Target    Target
	State     DesiredState
	Revision  uint64
	ChangedAt time.Time
	CommandID string
}

DesiredRecord is one attributed, monotonic target-state revision.

func (DesiredRecord) Validate 

func (r DesiredRecord) Validate() error

Validate rejects unauthored, unbounded, or unsupported desired state.

type DesiredState 

type DesiredState string

DesiredState is the durable lifecycle state authored by a control plane. 

const (
	DesiredActive      DesiredState = "active"
	DesiredPaused      DesiredState = "paused"
	DesiredDraining    DesiredState = "draining"
	DesiredTerminating DesiredState = "terminating"
)

type DesiredStateApplier 

type DesiredStateApplier interface {
	ApplyDesiredState(context.Context, DesiredRecord) error
}

DesiredStateApplier changes local queue admission or lifecycle state.

type DesiredStateReader 

type DesiredStateReader interface {
	GetDesiredState(context.Context, Target) (DesiredRecord, error)
}

DesiredStateReader reads authoritative tenant-scoped desired state.

type DesiredStateReconciler 

type DesiredStateReconciler struct {
	// contains filtered or unexported fields
}

DesiredStateReconciler applies each revision at most once per target. Scheduling and retry policy remain caller-owned; Reconcile starts no goroutines and a source failure cannot stop normal queue delivery.

func NewDesiredStateReconciler 

func NewDesiredStateReconciler(
	config DesiredStateReconcilerConfig,
) (*DesiredStateReconciler, error)

NewDesiredStateReconciler creates a bounded revision-aware reconciler.

func (*DesiredStateReconciler) Reconcile 

func (r *DesiredStateReconciler) Reconcile(ctx context.Context) error

Reconcile reads and applies every configured target in deterministic order. Successfully applied revisions remain committed when a later target fails.

Example 
package main

import (
	"context"
	"fmt"
	"time"

	"github.com/faustbrian/go-queue/management"
)

func main() {
	target := management.Target{Kind: management.TargetQueue, Name: "critical"}
	reconciler, err := management.NewDesiredStateReconciler(
		management.DesiredStateReconcilerConfig{
			Reader: desiredExampleReader{record: management.DesiredRecord{
				Target: target, State: management.DesiredPaused, Revision: 3,
				ChangedAt: time.Date(2026, 7, 16, 10, 0, 0, 0, time.UTC),
				CommandID: "pause-critical-3",
			}},
			Applier: desiredExampleApplier{},
			Targets: []management.Target{target},
		},
	)
	if err != nil {
		panic(err)
	}
	if err := reconciler.Reconcile(context.Background()); err != nil {
		panic(err)
	}
}

type desiredExampleReader struct {
	record management.DesiredRecord
}

func (r desiredExampleReader) GetDesiredState(
	context.Context,
	management.Target,
) (management.DesiredRecord, error) {
	return r.record, nil
}

type desiredExampleApplier struct{}

func (desiredExampleApplier) ApplyDesiredState(
	_ context.Context,
	record management.DesiredRecord,
) error {
	fmt.Printf(
		"apply %s to %s %s at revision %d\n",
		record.State,
		record.Target.Kind,
		record.Target.Name,
		record.Revision,
	)
	return nil
}

Output:
apply paused to queue critical at revision 3

type DesiredStateReconcilerConfig 

type DesiredStateReconcilerConfig struct {
	Reader  DesiredStateReader
	Applier DesiredStateApplier
	Targets []Target
}

DesiredStateReconcilerConfig defines one bounded pull reconciliation set.

type DrainState 

type DrainState string

DrainState describes progress of a graceful drain request. 

const (
	DrainNotRequested DrainState = "not_requested"
	DrainRequested    DrainState = "requested"
	DrainInProgress   DrainState = "in_progress"
	DrainCompleted    DrainState = "completed"
	DrainTimedOut     DrainState = "timed_out"
)

type Failure 

type Failure struct {
	Classification Classification
	Code           string
	// contains filtered or unexported fields
}

Failure attaches a stable classification and safe code while preserving the original cause for errors.Is and errors.As.

func NewFailure 

func NewFailure(classification Classification, code string, cause error) *Failure

NewFailure classifies cause with a stable, non-sensitive failure code.

func (*Failure) Error 

func (e *Failure) Error() string

Error implements error.

func (*Failure) Unwrap 

func (e *Failure) Unwrap() error

Unwrap preserves the classified cause for errors.Is and errors.As.

func (*Failure) Validate 

func (e *Failure) Validate() error

Validate rejects unknown dispositions and unsafe failure codes.

type FailureResolution 

type FailureResolution struct {
	Classification Classification
	Code           string
}

FailureResolution is the stable classification and safe code selected from a wrapped or joined failure graph. Code is empty when no classified failure supplied one.

func ResolveFailure 

func ResolveFailure(err error) FailureResolution

ResolveFailure resolves wrapped and joined failures with deterministic safety precedence: infrastructure, canceled, malformed, permanent, then retryable. Codes come only from the winning classification; equal-rank codes use lexical order so errors.Join argument order cannot change persistence.

type InspectRequest 

type InspectRequest struct {
	Kind       RecordKind
	ID         string
	Visibility PayloadVisibility
}

InspectRequest asks for one record at an explicit payload visibility.

func (InspectRequest) Validate 

func (r InspectRequest) Validate() error

Validate rejects ambiguous or unbounded inspection requests.

type JobRecord 

type JobRecord struct {
	Kind        RecordKind
	ID          string
	Backend     string
	Queue       string
	OccurredAt  time.Time
	Attempts    uint32
	FailureCode string
	Payload     Payload

	EnvelopeVersion      uint16
	PayloadSchemaVersion string
	OriginalID           string
	Topic                string
	Stream               string
	RoutingKey           string
	ConsumerGroup        string
	SourceRecordID       string
	EnqueuedAt           *time.Time
	FirstDeliveryAt      *time.Time
	LastDeliveryAt       *time.Time
	DeadLetteredAt       *time.Time
	RetryPolicy          string
	Classification       Classification
	FailureSummary       string
	Diagnostics          Payload
	HandlerType          string
	JobType              string
	Tags                 map[string]string
	TraceID              string
	TenantID             string
	ProducerVersion      string
	WorkerVersion        string
	OriginalDeadLetterID string
	PriorDeadLetterID    string
	ReplayGeneration     uint32
	RetentionDeadline    *time.Time
}

JobRecord is the backend-neutral failure or dead-letter representation.

func (JobRecord) Validate 

func (r JobRecord) Validate() error

Validate rejects malformed metadata and unsafe payload disclosure.

type Measurement 

type Measurement[T any] struct {
	Value     T
	Supported bool
}

Measurement distinguishes an honestly measured zero from an unsupported backend metric.

type PageRequest 

type PageRequest struct {
	Cursor    string
	Limit     uint32
	Search    string
	Sort      SortField
	Direction SortDirection
}

PageRequest defines bounded cursor pagination, sorting, and search.

func (PageRequest) Validate 

func (r PageRequest) Validate() error

Validate rejects unbounded or unsupported record-list requests.

type Payload 

type Payload struct {
	Visibility  PayloadVisibility
	ContentType string
	Size        int64
	Data        []byte
}

Payload is a bounded administrative representation. Data must remain empty unless privileged access explicitly requested revealed content.

type PayloadVisibility 

type PayloadVisibility string

PayloadVisibility describes the disclosure applied by an authorized inspection. Its zero value is hidden so payloads are never exposed by default. 

const (
	PayloadHidden   PayloadVisibility = ""
	PayloadRedacted PayloadVisibility = "redacted"
	PayloadRevealed PayloadVisibility = "revealed"
)

type ProtocolRange 

type ProtocolRange struct {
	Minimum ProtocolVersion
	Maximum ProtocolVersion
}

ProtocolRange is an inclusive set of protocol versions supported during a rolling upgrade.

type ProtocolVersion 

type ProtocolVersion struct {
	Major uint16
	Minor uint16
}

ProtocolVersion identifies a worker/control-plane management protocol.

type ProviderStatusReader 

type ProviderStatusReader struct {
	// contains filtered or unexported fields
}

ProviderStatusReader composes bounded native adapter observations.

func NewStatusReader 

func NewStatusReader(config StatusReaderConfig) (*ProviderStatusReader, error)

NewStatusReader creates a deterministic provider-backed status reader.

func (*ProviderStatusReader) ListQueues 

func (r *ProviderStatusReader) ListQueues(
	ctx context.Context,
	request StatusPageRequest,
) (QueueStatusPage, error)

ListQueues returns one queue-name-sorted bounded provider page.

func (*ProviderStatusReader) ListWorkers 

func (r *ProviderStatusReader) ListWorkers(
	ctx context.Context,
	request StatusPageRequest,
) (WorkerStatusPage, error)

ListWorkers returns one worker-ID-sorted bounded provider page.

type QueueMetrics 

type QueueMetrics struct {
	Depth            Measurement[int64]
	Lag              Measurement[int64]
	Pending          Measurement[int64]
	OldestAge        Measurement[time.Duration]
	Throughput       Measurement[float64]
	Runtime          Measurement[time.Duration]
	Succeeded        Measurement[uint64]
	Failed           Measurement[uint64]
	Retried          Measurement[uint64]
	Reclaimed        Measurement[uint64]
	DeadLettered     Measurement[uint64]
	SettlementErrors Measurement[uint64]
}

QueueMetrics reports current gauges and monotonic lifecycle counters where the backend can measure them honestly.

type QueueStatus 

type QueueStatus struct {
	Backend    string
	Queue      string
	ObservedAt time.Time
	Metrics    QueueMetrics
}

QueueStatus is a point-in-time backend-neutral queue observation.

func (QueueStatus) Validate 

func (s QueueStatus) Validate() error

Validate rejects malformed queue observations.

type QueueStatusPage 

type QueueStatusPage struct {
	Items      []QueueStatus
	NextCursor string
}

QueueStatusPage is one bounded page of queue observations.

func (QueueStatusPage) Validate 

func (p QueueStatusPage) Validate() error

Validate rejects malformed or oversized queue adapter output.

type QueueStatusProvider 

type QueueStatusProvider interface {
	ObserveQueue(context.Context) (QueueStatus, error)
}

QueueStatusProvider emits one logical backend queue observation.

type RecordKind 

type RecordKind string

RecordKind distinguishes failed jobs from terminal dead letters. 

const (
	RecordFailure    RecordKind = "failure"
	RecordDeadLetter RecordKind = "dead_letter"
)

type RecordPage 

type RecordPage struct {
	Items      []JobRecord
	NextCursor string
}

RecordPage is one bounded page of failure or dead-letter metadata.

func (RecordPage) Validate 

func (p RecordPage) Validate() error

Validate rejects oversized cursors, pages, and malformed adapter records.

type RecordReader 

type RecordReader interface {
	ListFailures(context.Context, PageRequest) (RecordPage, error)
	ListDeadLetters(context.Context, PageRequest) (RecordPage, error)
	Inspect(context.Context, InspectRequest) (JobRecord, error)
}

RecordReader is implemented by queue adapters that expose safe failure and dead-letter inspection without leaking native backend clients.

type ReplayOptions 

type ReplayOptions struct {
	Destination       string
	IdempotencyPolicy ReplayPolicy
}

ReplayOptions contains the mandatory data-plane replay safeguards.

type ReplayPolicy 

type ReplayPolicy string

ReplayPolicy declares how an explicit replay destination handles duplicate job identities. It does not claim exactly-once execution. 

const (
	ReplayRejectDuplicate  ReplayPolicy = "reject_duplicate"
	ReplayReplaceDuplicate ReplayPolicy = "replace_duplicate"
)

type Selection 

type Selection struct {
	Limit uint32
}

Selection bounds a backend-owned bulk administrative operation. Adapters must apply their own deterministic selection semantics within this limit.

type SortDirection 

type SortDirection string

SortDirection identifies ascending or descending ordering. 

const (
	SortAscending  SortDirection = "asc"
	SortDescending SortDirection = "desc"
)

type SortField 

type SortField string

SortField identifies bounded record-list ordering. 

const (
	SortOccurredAt SortField = "occurred_at"
	SortQueue      SortField = "queue"
	SortAttempts   SortField = "attempts"
)

type StatusMetadata 

type StatusMetadata struct {
	ID          string
	Version     string
	Concurrency uint32
	Protocol    ProtocolVersion
}

StatusMetadata configures stable identity for a native worker reporter.

func (StatusMetadata) Validate 

func (m StatusMetadata) Validate() error

Validate rejects incomplete or unbounded native reporter metadata.

type StatusPageRequest 

type StatusPageRequest struct {
	Cursor string
	Limit  uint32
}

StatusPageRequest defines bounded opaque-cursor status pagination.

func (StatusPageRequest) Validate 

func (r StatusPageRequest) Validate() error

Validate rejects unbounded status requests.

type StatusProvider 

type StatusProvider interface {
	WorkerStatusProvider
	QueueStatusProvider
}

StatusProvider emits both worker and queue observations.

type StatusReader 

type StatusReader interface {
	ListWorkers(context.Context, StatusPageRequest) (WorkerStatusPage, error)
	ListQueues(context.Context, StatusPageRequest) (QueueStatusPage, error)
}

StatusReader is implemented by queue adapters that expose bounded backend-neutral worker and queue observations.

type StatusReaderConfig 

type StatusReaderConfig struct {
	Workers []WorkerStatusProvider
	Queues  []QueueStatusProvider
}

StatusReaderConfig separates worker cardinality from logical queue sources.

type Target 

type Target struct {
	Kind TargetKind
	Name string
}

Target identifies a resource without exposing backend addressing.

type TargetKind 

type TargetKind string

TargetKind identifies the data-plane resource affected by a command. 

const (
	TargetQueue       TargetKind = "queue"
	TargetWorker      TargetKind = "worker"
	TargetWorkerGroup TargetKind = "worker_group"
	TargetFailure     TargetKind = "failure"
	TargetDeadLetter  TargetKind = "dead_letter"
)

type ValidationError 

type ValidationError struct {
	Field   string
	Problem string
}

ValidationError identifies one invalid management-contract field.

func (*ValidationError) Error 

func (e *ValidationError) Error() string

Error implements error.

type WorkerLifecycle 

type WorkerLifecycle struct {
	// contains filtered or unexported fields
}

WorkerLifecycle owns queue admission, in-flight job accounting, desired revisions, and bounded duplicate command results. It never touches a backend.

func NewWorkerLifecycle 

func NewWorkerLifecycle(config WorkerLifecycleConfig) (*WorkerLifecycle, error)

NewWorkerLifecycle creates an active queue lifecycle with bounded command history. Command history is never evicted; capacity exhaustion fails closed.

func (*WorkerLifecycle) ApplyDesiredState 

func (l *WorkerLifecycle) ApplyDesiredState(
	ctx context.Context,
	record DesiredRecord,
) error

ApplyDesiredState applies one durable revision and waits for the requested safe admission boundary. A canceled wait leaves the state active for retry.

func (*WorkerLifecycle) BeginAdmission 

func (l *WorkerLifecycle) BeginAdmission() bool

BeginAdmission reserves one backend request when the lifecycle is active.

func (*WorkerLifecycle) DecorateWorkerStatus 

func (l *WorkerLifecycle) DecorateWorkerStatus(
	status WorkerStatus,
) (WorkerStatus, error)

DecorateWorkerStatus overlays queue-owned admission state and capabilities on a native backend observation after verifying both describe one worker.

func (*WorkerLifecycle) EndAdmission 

func (l *WorkerLifecycle) EndAdmission() error

EndAdmission releases a request that did not produce a runnable job.

func (*WorkerLifecycle) EndJob 

func (l *WorkerLifecycle) EndJob() error

EndJob releases one in-flight job after handler settlement completes.

func (*WorkerLifecycle) Execute 

func (l *WorkerLifecycle) Execute(
	ctx context.Context,
	command Command,
) (CommandResult, error)

Execute enforces lifecycle commands and returns a bounded correlated result. Failure/dead-letter and backend mutation actions remain unsupported.

func (*WorkerLifecycle) PromoteAdmissionToJob 

func (l *WorkerLifecycle) PromoteAdmissionToJob() error

PromoteAdmissionToJob atomically converts one request into in-flight work.

func (*WorkerLifecycle) Snapshot 

func (l *WorkerLifecycle) Snapshot() WorkerLifecycleSnapshot

Snapshot returns synchronized worker lifecycle presentation state.

type WorkerLifecycleConfig 

type WorkerLifecycleConfig struct {
	Metadata          StatusMetadata
	WorkerGroup       string
	Queue             string
	MaxCommandResults int
	Now               func() time.Time
}

WorkerLifecycleConfig identifies the queue admission scope controlled by one Queue instance and bounds its in-memory duplicate-command protection.

type WorkerLifecycleSnapshot 

type WorkerLifecycleSnapshot struct {
	State       WorkerState
	DrainStatus DrainState
	CurrentJobs uint32
	Terminating bool
}

WorkerLifecycleSnapshot is one synchronized admission and drain view.

type WorkerState 

type WorkerState string

WorkerState is the state reported by a live worker. 

const (
	WorkerRunning  WorkerState = "running"
	WorkerPaused   WorkerState = "paused"
	WorkerDraining WorkerState = "draining"
	WorkerStopped  WorkerState = "stopped"
)

type WorkerStatus 

type WorkerStatus struct {
	ID           string
	Version      string
	StartedAt    time.Time
	HeartbeatAt  time.Time
	Queues       []string
	Concurrency  uint32
	State        WorkerState
	CurrentJobs  uint32
	DrainStatus  DrainState
	Backend      string
	Protocol     ProtocolVersion
	Capabilities []Capability
}

WorkerStatus is the bounded status contract emitted by a worker.

func (WorkerStatus) Validate 

func (s WorkerStatus) Validate() error

Validate rejects malformed or unbounded worker reports.

type WorkerStatusPage 

type WorkerStatusPage struct {
	Items      []WorkerStatus
	NextCursor string
}

WorkerStatusPage is one bounded page of worker observations.

func (WorkerStatusPage) Validate 

func (p WorkerStatusPage) Validate() error

Validate rejects malformed or oversized worker adapter output.

type WorkerStatusProvider 

type WorkerStatusProvider interface {
	ObserveWorker(context.Context) (WorkerStatus, error)
}

WorkerStatusProvider emits one native worker observation.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.