headgate

package module
v0.1.4 Latest Latest
Warning

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

 Go to latest Published: Aug 31, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

Documentation

Overview

Package headgate is a distributed job queue whose dequeue is an admission decision.

Every other queue asks the store "give me N jobs". headgate asks "given the fleet's policy state and my capacity, what may I run?" — evaluated atomically inside the store. That single change is what makes fleet-wide rate limiting, tenant fairness, global concurrency ceilings, and poison-pill quarantine one mechanism instead of four missing features.

Index

Constants

View Source 
const (
	MaxProgressValue        uint64 = 1<<53 - 1
	MaxProgressMessageBytes        = 512
)
View Source 
const (
	UniqueReplacePayload     uint32 = 1 << 0
	UniqueReplaceScheduledAt uint32 = 1 << 1
	UniqueReplacePriority    uint32 = 1 << 2
	UniqueReplaceMaxAttempts uint32 = 1 << 3
	UniqueReplaceAll                = UniqueReplacePayload | UniqueReplaceScheduledAt | UniqueReplacePriority | UniqueReplaceMaxAttempts
)
View Source 
const (
	// TraceparentHeader is the RESERVED envelope header carrying W3C Trace Context's
	// traceparent.
	//
	// The name is specified here because an unwritten convention becomes multiple
	// incompatible conventions across SDKs. The key is lowercase because
	// W3C Trace Context defines these as HTTP header field names, which are
	// case-insensitive on the wire and canonically lowercase; the envelope's header map
	// is NOT case-insensitive, so the spec has to pick one spelling and this is it.
	TraceparentHeader = "traceparent"
	// TracestateHeader is the RESERVED envelope header carrying W3C Trace Context's
	// tracestate. Opaque: headgate never parses, validates, or truncates it.
	TracestateHeader = "tracestate"
)
View Source 
const (
	IsolatedProtocolPrefix = "HEADGATE/1 "
)
View Source 
const MaxOpaqueSchemaVersion uint32 = 1<<31 - 1

MaxOpaqueSchemaVersion is the largest result/output schema version portable across every backend, including PostgreSQL's signed integer columns.

View Source 
const Version = "0.1.4"

Version is the release version shared by the Go modules.

Variables

View Source 
var (
	ErrDuplicate    = errors.New("headgate: duplicate unique key")
	ErrIDConflict   = errors.New("headgate: id conflict")
	ErrQuarantined  = errors.New("headgate: fingerprint is quarantined")
	ErrBackpressure = errors.New("headgate: enqueue backpressure")
	ErrNoUpcastPath = errors.New("headgate: no upcast path for schema version")
	ErrLeaseLost    = errors.New("headgate: lease lost; stop work immediately")
	// the three variants the store port was missing. The Rust `StoreError`
	// enum has had NotFound / Invalid / Unavailable since Phase 2; Go expressed all
	// three as `errors.New("headgate: …")` strings, so `headgateapi.storeErr` had to
	// classify by string PREFIX — which meant exactly one shape ("not found: ") was
	// recognized and EVERYTHING else, a dropped Postgres connection included, fell
	// through to 400. A 400 on a dead store silently defeats 5xx-based client retry.
	ErrNotFound    = errors.New("headgate: not found")
	ErrInvalid     = errors.New("headgate: invalid request")
	ErrUnavailable = errors.New("headgate: store unavailable")
)
View Source 
var (
	// ErrSkipJob: stop retrying, archive. The branch apalis shipped commented out.
	ErrSkipJob = errors.New("headgate: skip: archive without retrying")
	// ErrRevokeJob: drop the job entirely.
	ErrRevokeJob = errors.New("headgate: revoke: drop entirely")
	// ErrRateLimited (surveyed policy behavior): the upstream said 429 — requeue without consuming an
	// attempt and without recording a failure.
	ErrRateLimited = errors.New("headgate: rate limited upstream")
)
View Source 
var (
	ErrTaskTrackerUnavailable = errors.New("headgate: tracked tasks are only available inside a handler")
	ErrTaskTrackerClosed      = errors.New("headgate: job attempt is no longer accepting tracked tasks")
)
View Source 
var ErrCircuitRejected = errors.New("headgate: enqueue circuit rejected call")
View Source 
var ErrClientFromContextUnavailable = errors.New("headgate: client is only available inside a handler")
View Source 
var ErrEnqueueForbidden = errors.New("headgate: enqueue forbidden")
View Source 
var ErrInvalidPlugin = errors.New("headgate: invalid plugin")

ErrInvalidPlugin is returned when a plugin has no useful identity or an invalid kind scope. Components themselves use the existing middleware and hook contracts.

View Source 
var ErrTaskDataUnavailable = errors.New("headgate: task data is only available inside a handler")

ErrTaskDataUnavailable means a task-data operation was attempted outside a handler context. Worker data is configured explicitly through Config.Extensions instead.

View Source 
var ErrWaitUnsupported = errors.New("headgate: insert-and-await is unsupported")

Functions

func AuthorizeEnqueueBatch 

func AuthorizeEnqueueBatch(
	ctx context.Context,
	authorizer EnqueueAuthorizer,
	source EnqueueSource,
	batch []Envelope,
) error

AuthorizeEnqueueBatch performs the complete authorization pass before any I/O. A nil authorizer means the documented allow-all default.

func CanonicalTags 

func CanonicalTags(tags []string) []string

func Data 

func Data[T any](ctx context.Context) (value T, ok bool)

Data resolves T from this attempt first, then from the worker's shared defaults. Job-local shadowing lets middleware specialize one dependency without mutating the value seen by concurrently-running siblings.

func DecodeArgs 

func DecodeArgs[T Args](e Envelope) (T, error)

DecodeArgs exposes the same version-aware decode path used by RegisterFunc for opt-in raw-envelope adapters.

func DecodeHeaders 

func DecodeHeaders(b []byte) map[string]string

DecodeHeaders parses that JSON back. Non-string values are DROPPED rather than stringified: the envelope's header map is string->string, and silently coercing {"a":1} into "1" would make a round trip lossy in a way nothing else here is.

func Deref 

func Deref(p *string) string

Deref reads a filter field with nil meaning "absent". Drivers use it where the SQL or the Lua wants a plain string and the nil case has already been handled.

func EffectiveUniqueKey 

func EffectiveUniqueKey(e Envelope) []byte

EffectiveUniqueKey returns the versioned store key. Kind is included by default; ExcludeKind uses a separate namespace so the two scopes cannot alias.

func EffectiveWeight 

func EffectiveWeight(weight uint32) uint32

EffectiveWeight turns proto3 omission and Go's zero-value struct literal into the documented default cost of one. HTTP APIs reject an explicitly supplied zero; the store accepts zero only as the compatibility sentinel.

func EncodeHeaders 

func EncodeHeaders(h map[string]string) string

EncodeHeaders renders an envelope's headers as the JSON object every adapter stores. ONE implementation for all four Go adapters, in the core module, because the Redis keyspace byte-diff in scripts/test-admission.sh compares a Go-driven store against a Rust-driven one: the two encodings must agree to the byte, not merely to the value.

SetEscapeHTML(false) is load-bearing for exactly that reason — Go's encoding/json escapes <, > and & to </>/& by default and Rust's serde_json does not, so a header value containing one would have diffed. Empty renders as "" so the Redis adapter can omit the field entirely rather than writing "{}".

func EnqueueQueue 

func EnqueueQueue(e Envelope) string

EnqueueQueue is the queue an envelope actually lands in. Every backend defaults an empty queue to "default" on write, so the idempotent enqueue identity id comparison must normalize the same way or a replay that omitted the queue would read as a conflict against its own row.

func Extension 

func Extension[T any](extensions *Extensions) (value T, ok bool)

Extension returns the value stored under exactly T. Asking for another type is a miss; callers never receive an untyped value to cast.

func Fingerprint 

func Fingerprint(kind string, payload []byte) string

Fingerprint derives the crash quarantine fingerprint of (kind, payload), specified in ARCHITECTURE.md content fingerprinting and nowhere else: 

lowercase_hex( SHA256( u32_le(len(kind)) || kind || u32_le(len(payload)) || payload )[0..16] )

Length-prefixed so ("a","bc") and ("ab","c") cannot collide; truncated to 128 bits because a collision over-quarantines. Derived CLIENT-SIDE at enqueue when the caller does not supply one; stores pass the value through untouched. The Rust implementation must produce identical output — the content fingerprinting test vectors are the conformance scenario.

func Invalidf 

func Invalidf(format string, args ...any) error

Invalidf is the constructor the drivers use: Invalidf("unknown action `%s`", a).

func IsUnavailable 

func IsUnavailable(err error) bool

IsUnavailable reports whether err is a lost store connection rather than a rejected request — typed availability errors's distinction, which decides between a 503 the caller should retry and a 4xx it must not.

It answers true for an explicit *UnavailableError, and otherwise identifies a transport failure by STANDARD-LIBRARY error identity: net.Error (every dial failure and timeout from pgx, go-redis and database/sql wraps a *net.OpError), the three socket errnos a peer death produces, io.EOF from a connection closed mid-reply, and database/sql's ErrBadConn. That is deliberately not a string match, and it is the reason the API layer can classify a dropped connection without importing a single database driver — which it must not do (invariant 8's spirit: one module per driver, so nobody's go.mod pulls every database).

It is conservative on purpose: an error it does not recognize is answered 500 by the API, not 400. Unclassified is a server fault until someone proves otherwise.

func JobData 

func JobData[T any](ctx context.Context) (value T, ok bool)

func Log 

func Log(ctx context.Context, msg string)

Log records one execution-log line onto THIS attempt (attempt-log contract, River's riverlog): it lands inside the attempt's error-history entry when the runner acks, so the console can answer "why did attempt 3 fail" without a log aggregator. Bounded: 100 lines per attempt, 2KB per line (truncated) — the history is a timeline, not a log store. Outside a running job (no runner context) it is a no-op.

func Logf 

func Logf(ctx context.Context, format string, args ...any)

Logf is Log with formatting.

func NoisyPartitionKeys 

func NoisyPartitionKeys(loads map[string]int64) map[string]bool

NoisyPartitionKeys classifies noisy neighbours from observed in-flight skew (tenant fairness/backlog metrics). A partition needs at least two in-flight jobs and more than twice the mean load of all peers. A lone partition is never noisy. Integer 128-bit products keep the threshold equivalent to Rust without float rounding at the boundary.

func NotFoundf 

func NotFoundf(format string, args ...any) error

NotFoundf is the constructor the drivers use: NotFoundf("job %s", id).

func Ptr 

func Ptr[T any](v T) *T

Ptr is the constructor for the pointer-valued filter fields above (and for `Counts`' queue argument). `headgate.Ptr("")` is how a caller asks for the empty value, which is the whole reason those fields are pointers.

func RecordResult 

func RecordResult(ctx context.Context, schemaVersion uint32, bytes []byte) error

RecordResult stores versioned opaque bytes in the attempt. The runtime commits them atomically with success; retry/error outcomes discard them.

func RegisterBatchFunc 

func RegisterBatchFunc[T Args](
	r *Registry,
	maxSize int,
	maxDelay time.Duration,
	work func([]BatchJob[T]) []error,
) error

RegisterBatchFunc registers a typed chunk handler. Same-kind admitted attempts wait until maxSize or maxDelay, then one call receives them. Results are positional and still flow through the ordinary per-job ack/fence/death-handler path.

func RegisterExtracted1 

func RegisterExtracted1[T Args, A any](r *Registry, a HandlerExtractor[A], work func(context.Context, *Job[T], A) error) error

RegisterExtractedN keeps extraction compile-time typed without reflection or a service locator. All N extractors finish before work is called.

func RegisterExtracted2 

func RegisterExtracted2[T Args, A, B any](r *Registry, a HandlerExtractor[A], b HandlerExtractor[B], work func(context.Context, *Job[T], A, B) error) error

func RegisterExtracted3 

func RegisterExtracted3[T Args, A, B, C any](r *Registry, a HandlerExtractor[A], b HandlerExtractor[B], c HandlerExtractor[C], work func(context.Context, *Job[T], A, B, C) error) error

func RegisterExtracted4 

func RegisterExtracted4[T Args, A, B, C, D any](r *Registry, a HandlerExtractor[A], b HandlerExtractor[B], c HandlerExtractor[C], d HandlerExtractor[D], work func(context.Context, *Job[T], A, B, C, D) error) error

func RegisterExtracted5 

func RegisterExtracted5[T Args, A, B, C, D, E any](r *Registry, a HandlerExtractor[A], b HandlerExtractor[B], c HandlerExtractor[C], d HandlerExtractor[D], e HandlerExtractor[E], work func(context.Context, *Job[T], A, B, C, D, E) error) error

func RegisterFunc 

func RegisterFunc[T Args](r *Registry, work func(context.Context, *Job[T]) error) error

func RegisterIsolated 

func RegisterIsolated[T Args](r *Registry, cfg IsolatedProcessConfig) error

RegisterIsolated registers T's kind and aliases for child-process execution. Job bytes travel only through stdin; they are never interpolated into a shell command.

func RegisterRaw 

func RegisterRaw[T Args](r *Registry, work func(context.Context, Claim) error) error

RegisterRaw registers T's kind and aliases without decoding its envelope. Opt-in layers such as encrypted payloads transform bytes here before typed dispatch.

func RegisterWorker 

func RegisterWorker[T Args](r *Registry, w Worker[T]) error

RegisterWorker registers w for T's kind and aliases. Payloads decode via the default JSON codec (payload codecs); a Versioned T gets its Upcast called for foreign schema versions.

func RemoveExtension 

func RemoveExtension[T any](extensions *Extensions) (value T, ok bool)

func ReportActualWeight 

func ReportActualWeight(ctx context.Context, actual uint32) error

ReportActualWeight records the final surveyed policy behavior rate-budget cost after an upstream call. Admission already charged the envelope's estimate; ack reconciles this total under the same fence. Zero is valid, and the last report wins. Calling outside a running handler returns an error instead of silently dropping the correction.

func SameJobContent 

func SameJobContent(e Envelope, kind, fingerprint, queue string) bool

SameJobContent answers idempotent enqueue identity's question: does the row that already owns this id hold the SAME job? The comparison set is (kind, content fingerprinting fingerprint, queue). The fingerprint is content identity over kind+payload by construction — length-prefixed SHA-256, derived client-side, passed through untouched by every store — so comparing it compares the payload without shipping the payload back. Kind is compared as well as hashed so two envelopes that both omit the fingerprint cannot pass as each other. The queue is in the set because routing is part of what a replay must not silently change.

func ScheduleDueTicks 

func ScheduleDueTicks(spec string, firstMs, nowMs int64, cap int) ([]int64, error)

ScheduleDueTicks mirrors Rust's schedule_spec::due_ticks: firstMs (the stored next_run, inclusive) plus every successor up to and including nowMs, capped at the `cap` MOST RECENT ticks, oldest first.

func ScheduleNextAfter 

func ScheduleNextAfter(spec string, afterMs int64) (int64, error)

ScheduleNextAfter returns the next tick STRICTLY AFTER afterMs. Mirrors Rust's schedule_spec::next_after for "@every:<ms>".

func SchedulerSweep 

func SchedulerSweep(ctx context.Context, insp InspectStore) (uint64, error)

SchedulerSweep is one pass: fire everything due, advance. Returns jobs enqueued. Errors on a single schedule are logged and skipped, never fatal to the sweep.

func SchedulerSweepWithHooks 

func SchedulerSweepWithHooks(
	ctx context.Context,
	insp InspectStore,
	hooks ...PeriodicEnqueueHook,
) (uint64, error)

SchedulerSweepWithHooks runs one pass and emits schedule-aware begin/end events around every actual tick enqueue. The legacy entry point delegates with no hooks.

func SetCursor 

func SetCursor[C any](ctx context.Context, cursor C) error

SetCursor records progress inside a cursor step, durably and fence-verified — synchronous by design in v0.1, correctness before the ride-the-renewal batching.

func SetExtension 

func SetExtension[T any](extensions *Extensions, value T) (previous T, replaced bool)

SetExtension stores value under exactly T and returns the prior value of T, if any.

func SetJobData 

func SetJobData[T any](ctx context.Context, value T) error

SetJobData inserts scratch data into this attempt only. Concurrent jobs always have different job maps; derived contexts and goroutines for this job share the same map.

func Snooze 

func Snooze(d time.Duration) error

func Step 

func Step(ctx context.Context, name string, fn func(context.Context) error) error

Step runs a named unit of work once per JOB, not once per attempt. On retry, steps already recorded in the checkpoint are skipped without running.

func StepCursor 

func StepCursor[C any](ctx context.Context, name string, fn func(context.Context, C) error) error

StepCursor resumes a loop at a saved position — Sidekiq's IterableJob shape. The cursor is JSON-serialized (payload codecs's default codec); fn receives the zero C on a first run and the last durable cursor on resume, and calls SetCursor as it progresses.

func StepOnce 

func StepOnce(ctx context.Context, name string, fn func(context.Context, Tx) error) error

StepOnce (step replay × transactional effects) is a step whose SIDE EFFECTS and completion marker commit in ONE transaction, keyed "{job_id}/{name}" — the step's writes happen exactly once even though the job may be admitted many times. On retry a completed step is skipped like any other. Requires a transactional store; Redis declines (runtime capability boundary).

func Track 

func Track(ctx context.Context, work func(context.Context) error) error

Track starts work concurrently and attaches it to the current job attempt. The work receives the handler's cancellation/deadline context and may itself call Track. The first tracked error fails the attempt after all sibling work has been cancelled and joined. Calling Track outside dispatch, or after the handler has returned, is an explicit error—there is no process-global tracker.

func Unavailablef 

func Unavailablef(format string, args ...any) error

Unavailablef is the constructor the drivers use.

func ValidateEnqueue 

func ValidateEnqueue(batch []Envelope) error

ValidateEnqueue is the boundary validation every backend's Enqueue runs before it writes anything — ONE function so the rule cannot drift between four adapters, and the layer is the store because the API and the harnesses call Store.Enqueue directly, never through the runtime. Batch-level: a repeated id WITHIN one batch is an IDConflictError on every backend rather than a constraint error from whichever row the database reached first.

func ValidateKind 

func ValidateKind(kind string) error

ValidateKind is the one kind-format rule (typed dispatch), enforced identically at handler registration, at enqueue in every backend, and at the HTTP API.

[A-Za-z0-9_] first, then word characters or one of `- [ ] < > / . : +`, 1..=128 bytes. That is River's charset (\A[\w][\w\-\[\]<>/.·:+]+\z) with three deliberate differences, each with a reason:

  • ASCII-only word characters. Go's \w is ASCII and Rust's regex \w is Unicode-aware; a rule written as \w would mean two different things in the two languages, which is exactly the drift the conformance suite exists to catch.
  • Minimum length ONE, where River requires two. headgate's own conformance corpus enqueues kind "w", and a one-letter kind is a short name, not a hazard.
  • No · (U+00B7). It follows from ASCII-only; nothing in the corpus uses it.

Whitespace and control characters are rejected by construction — neither is in the permitted set.

func ValidateProgress 

func ValidateProgress(update ProgressUpdate) error

func WithEnqueueIdentity 

func WithEnqueueIdentity(ctx context.Context, identity EnqueueIdentity) context.Context

WithEnqueueIdentity attaches an authenticated identity for the library client or HTTP authorizer. The value is copied so later caller mutation cannot change an in-flight decision.

func WorkerData 

func WorkerData[T any](ctx context.Context) (value T, ok bool)

func WrapUnavailable 

func WrapUnavailable(err error) error

WrapUnavailable is the driver-boundary half of the typed availability errors error contract. Database libraries necessarily return their own transport errors; a Store implementation must not leak those concrete types to callers or force the API to import every driver. Recognized connection failures become *UnavailableError. Validation, uniqueness, quarantine, and every other typed domain error pass through unchanged.

Drivers call this at the OUTER Enqueue boundary, after ValidateEnqueue has run. That ordering is load-bearing: an invalid job is still invalid while the store is down.

Types

type AdmissionExplain 

type AdmissionExplain struct {
	State      string
	Admissible bool
	// BlockedBy: rate_class | concurrency_limit | fairness | quarantine | schedule |
	// queue_paused; empty when nothing blocks.
	BlockedBy string
	Detail    map[string]string
	// EstimatedAdmissionMs is nil when the block will not clear on its own.
	EstimatedAdmissionMs *int64
}

type AdmissionUnit 

type AdmissionUnit struct {
	Claims []Claim
}

AdmissionUnit is ordinarily one job and occasionally a group admitted as one decision (batch-shaped admission). v0.1 always returns units of one, but the contract is group-shaped now because batched execution changes the gate's accounting in four places, and retrofitting that means reopening the atomic claim after it has traffic.

func GroupAdmissionClaims 

func GroupAdmissionClaims(claims []Claim, maxUnitSize int) []AdmissionUnit

GroupAdmissionClaims turns the flat atomically claimed result into deterministic same-kind handler units. Policy accounting remains per member, so a unit of N spends N units of concurrency/fairness and each member's own rate weight.

type AdmitRequest 

type AdmitRequest struct {
	Worker   string
	LeaseID  string
	Queues   []string
	Capacity int
	Lease    time.Duration
	Quantum  int64 // tenant fairness per-partition fair share for this call
}

type Aliased 

type Aliased interface {
	Args
	KindAliases() []string
}

Aliased is optional. Kinds this worker also answers to — typed dispatch. Enqueue always uses Kind(); dispatch matches Kind() or any alias. Renaming a task without this strands every already-enqueued job of the old kind.

type AllowAllEnqueues 

type AllowAllEnqueues struct{}

AllowAllEnqueues is the backward-compatible default. Authentication and identity remain the embedding application's responsibility.

func (AllowAllEnqueues) AuthorizeEnqueue 

func (AllowAllEnqueues) AuthorizeEnqueue(context.Context, EnqueueAuthorization, Envelope) bool

type Args 

type Args interface {
	Kind() string
}

Args is a job payload. Kind is the dispatch key and is wire state: changing it strands every already-enqueued job of that type.

type Attempt 

type Attempt struct {
	ReturnedErrors uint32
	Crashes        uint32
	MaxAttempts    uint32
}

Attempt keeps returned errors and crash-attributed losses distinct.

type BackoffConfig 

type BackoffConfig struct {
	Floor      time.Duration
	Ceiling    time.Duration
	Multiplier float64
	Jitter     float64
}

type BackpressureError 

type BackpressureError struct {
	Queue                    string
	Limit, Current, Incoming uint64
}

BackpressureError is a producer policy rejection, not a backend failure. The store evaluated Current + Incoming against Limit atomically for Queue. Callers can retry after capacity is released, route elsewhere, or shed explicitly.

func (*BackpressureError) Error 

func (e *BackpressureError) Error() string

func (*BackpressureError) Unwrap 

func (e *BackpressureError) Unwrap() error

type BatchJob 

type BatchJob[T Args] struct {
	Context context.Context
	Job     *Job[T]
}

BatchJob is one independently fenced attempt delivered to a chunk handler. Context remains per member so cancellation, checkpoints, logs, and actual rate usage cannot leak across jobs merely because application work is coalesced.

type BulkOp 

type BulkOp struct {
	ID, Action         string
	Queue, State, Kind string
	PartitionKey       string
	OlderThanMs        *int64
	DryRun             bool
}

BulkOp is control API contract's asynchronous bulk mutation as data. An empty selector is rejected.

type Caps 

type Caps uint32
const (
	CapTransactional Caps = 1 << iota
	CapNotifying
	CapInspect
)

func (Caps) Has 

func (c Caps) Has(x Caps) bool

type Checkpoint 

type Checkpoint struct {
	LastCompletedStep string
	// CompletedSteps is the completed steps IN ORDER. Replay compares positionally: the
	// step at index i of the new attempt must match CompletedSteps[i], or the step set
	// changed under the checkpoint and the job goes to undecodable — never a silent
	// restart.
	CompletedSteps []string
	// InProgressStep is the step that was running when the checkpoint was last written
	// (written BEFORE the step's side effects); the reclaimer attributes a crash to it.
	InProgressStep string
	CursorStep     string
	Cursor         []byte
	// payload versioning × step replay — the step set this was written against. A resumed job whose step set
	// no longer matches goes to Undecodable rather than silently restarting from step one.
	SchemaVersion uint32
	StepSetHash   string
	// crash quarantine — crash counts per step. "Always dies at transcode" beats "dies".
	CrashesByStep map[string]uint32
}

type CheckpointInspectStore added in v0.1.4 

type CheckpointInspectStore interface {
	// A nil checkpoint means the job does not exist. Existing jobs with no resumable
	// progress return an empty Checkpoint.
	GetJobCheckpoint(ctx context.Context, id string) (*Checkpoint, error)
}

CheckpointInspectStore explicitly exposes resumable-step state. Cursor bytes may carry application data, so ordinary job/list reads never include them.

type CircuitBreaker 

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

CircuitBreaker is concurrency-safe. It counts only IsUnavailable errors; policy and domain rejections are reachable-store results and reset/complete a recovery probe.

func NewCircuitBreaker 

func NewCircuitBreaker(config CircuitBreakerConfig) (*CircuitBreaker, error)

func (*CircuitBreaker) Snapshot 

func (b *CircuitBreaker) Snapshot() CircuitSnapshot

Snapshot applies the recovery timer before returning the current state.

type CircuitBreakerConfig 

type CircuitBreakerConfig struct {
	FailureThreshold uint32
	RecoveryTimeout  time.Duration
	HalfOpenMaxCalls uint32
}

CircuitBreakerConfig controls an opt-in, process-local enqueue circuit breaker.

func DefaultCircuitBreakerConfig 

func DefaultCircuitBreakerConfig() CircuitBreakerConfig

DefaultCircuitBreakerConfig returns the conservative prior-art shape. Constructing a client does not install it automatically: callers explicitly choose this failure boundary and may share one CircuitBreaker across clients.

type CircuitOpenError 

type CircuitOpenError struct {
	State      CircuitState
	RetryAfter time.Duration
}

CircuitOpenError means the store was not called. RetryAfter is positive while open; it is zero when the half-open probe budget is currently occupied.

func (*CircuitOpenError) Error 

func (e *CircuitOpenError) Error() string

func (*CircuitOpenError) Unwrap 

func (e *CircuitOpenError) Unwrap() error

type CircuitSnapshot 

type CircuitSnapshot struct {
	State               CircuitState
	ConsecutiveFailures uint32
	HalfOpenSuccesses   uint32
	HalfOpenInFlight    uint32
	RetryAfter          time.Duration
}

CircuitSnapshot is a read-only view suitable for telemetry and readiness details.

type CircuitState 

type CircuitState string

CircuitState is the observable producer availability state. 

const (
	CircuitClosed   CircuitState = "closed"
	CircuitOpen     CircuitState = "open"
	CircuitHalfOpen CircuitState = "half_open"
)

type Claim 

type Claim struct {
	Envelope Envelope
	LeaseID  string
	Fence    uint64
	Expires  time.Time
	// step replay step progress persisted by earlier attempts; zero-valued on a first attempt.
	Checkpoint Checkpoint
}

type Client 

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

Client is the producer-facing enqueue boundary. Raw Store remains a trusted low-level port for workers and adapters; applications accepting untrusted input should expose a Client instead.

func NewClient 

func NewClient(store Store, options ...ClientOption) *Client

NewClient preserves existing behavior with an allow-all default. Installing an authorizer is explicit.

func (*Client) Enqueue 

func (c *Client) Enqueue(ctx context.Context, batch []Envelope) error

func (*Client) EnqueueAndWait 

func (c *Client) EnqueueAndWait(ctx context.Context, envelope Envelope) (Completion, error)

EnqueueAndWait subscribes before enqueue, then reconciles durable state after enqueue and periodically. This closes fast-completion, dropped-event, and reconnect races.

func (*Client) EnqueueTx 

func (c *Client) EnqueueTx(ctx context.Context, tx Tx, batch []Envelope) error

EnqueueTx runs the identical authorization pass before touching the caller's transaction, so the transactional path is not a policy bypass.

func (*Client) EnqueueWithSource 

func (c *Client) EnqueueWithSource(
	ctx context.Context,
	source EnqueueSource,
	batch []Envelope,
) error

EnqueueWithSource lets trusted adapters (notably the control API) preserve the source supplied to authorization while using the identical circuit and store call.

type ClientOption 

type ClientOption func(*Client)

ClientOption configures the producer client.

func WithCircuitBreaker 

func WithCircuitBreaker(breaker *CircuitBreaker) ClientOption

WithCircuitBreaker installs a local availability circuit. Sharing one breaker across clients gives one process a coherent outage view; nil disables it.

func WithEnqueueAuthorizer 

func WithEnqueueAuthorizer(authorizer EnqueueAuthorizer) ClientOption

WithEnqueueAuthorizer installs application policy. The same policy guards ordinary, bulk, and transactional enqueue.

func WithEnqueueMiddleware 

func WithEnqueueMiddleware(middlewares ...EnqueueMiddleware) ClientOption

WithEnqueueMiddleware appends ordered producer middleware. Registration order is nesting order: the first middleware runs its before half first and after half last.

func WithEventBus 

func WithEventBus(eventBus *EventBus) ClientOption

WithEventBus installs the same process-local bus configured on the worker. Events provide latency; durable Inspect reads provide correctness.

func WithInsertHooks 

func WithInsertHooks(hooks ...InsertHook) ClientOption

WithInsertHooks appends non-wrapping observers of every actual enqueue store attempt.

func WithPlugins 

func WithPlugins(plugins ...Plugin) ClientOption

WithPlugins installs producer bundles. Standalone components always run first, followed by global plugins and then matching scoped plugins. Install order is stable within each plugin class, even when global and scoped plugins are supplied interleaved.

type Clock 

type Clock interface{ NowMs() int64 }

Clock is injectable so scheduling and lease expiry are testable without sleeping.

type Codec 

type Codec interface {
	Encode(Args) ([]byte, error)
	Decode(kind string, version uint32, b []byte) (Args, error)
}

type Completion 

type Completion struct {
	JobID  string
	State  string
	Result *JobResult
	Error  string
}

Completion is the durable terminal state returned by EnqueueAndWait. Result is nil for jobs that completed without bytes and for non-success terminal states.

type ConcurrencyLimit 

type ConcurrencyLimit struct {
	Name          string
	Queue         string
	PartitionBy   string
	MaxConcurrent uint64
	// surveyed policy behavior what happens when the key is saturated. Hatchet and Solid Queue both make
	// this explicit; everyone else leaves users to reimplement it badly.
	OnSaturated SaturationStrategy
}

type Config 

type Config struct {
	Queues            map[string]QueueConfig
	RateClasses       []RateClass // admission policy FLEET-WIDE, not per process
	ConcurrencyLimits []ConcurrencyLimit
	Quantum           int64  // tenant fairness default per-partition fair share
	CrashLimit        uint32 // crash quarantine crashes before quarantine. default 3
	LeaseDuration     time.Duration
	ShutdownTimeout   time.Duration
	// MemoryLimitBytes enables the process memory guard. Zero disables it. Crossing the
	// limit stops admission and uses the ordinary bounded graceful drain; the process
	// supervisor is responsible for starting the replacement.
	MemoryLimitBytes uint64
	// MemoryCheckInterval defaults to 30 seconds when the guard is enabled.
	MemoryCheckInterval time.Duration
	// MemorySampler is injectable so tests never depend on allocator or OS timing.
	// Nil selects the platform process sampler.
	MemorySampler MemorySampler
	Telemetry     Telemetry
	Clock         Clock
	RetryPolicy   RetryPolicy
	// Extensions contains type-safe process-local dependencies shared by all attempts
	// on this runner. Each attempt receives a separate empty job-local map. Neither map
	// is part of Envelope, so values disappear across retry, restart, or another worker.
	Extensions *Extensions
	// Producer is the complete client stack exposed to handlers for follow-on work.
	// Nil builds an allow-all client over this Runner's Store.
	Producer *Client
	// PeriodicEnqueueHooks observe the elected scheduler's actual durable tick enqueues.
	PeriodicEnqueueHooks []PeriodicEnqueueHook
	// DeathHandlers run only after a fence-verified transition to archived succeeds.
	DeathHandlers []DeathHandler
	// StuckJobHandler runs only if timeout/cancellation has not stopped the handler and
	// its tracked work within StuckJobThreshold. It is an operational escalation point,
	// not lifecycle middleware.
	StuckJobHandler   StuckJobHandler
	StuckJobThreshold time.Duration
	// EventBus receives bounded application-facing lifecycle events after Store success.
	EventBus *EventBus

	// IsFailure decides whether an error consumes a retry attempt. failure classification — asynq's
	// generalization of the RateLimited special case. Returning false re-queues without
	// incrementing Attempt and without recording a queue failure. Default: all errors
	// are failures.
	IsFailure func(error) bool

	// Pool is a caller-supplied connection pool. failure classification — headgate never closes a pool it
	// did not open. asynq accepts an existing client on every entry point for this
	// reason, and Oban's scaling guide is largely about connection pressure.
	//
	// EmptyPollBackoff controls the idle path: a fixed interval across N idle workers is
	// N wasted queries per tick, and on MySQL (no LISTEN/NOTIFY) the idle path is the
	// only path. A notify resets the backoff to its floor.
	EmptyPollBackoff BackoffConfig

	// WorkerID is a stable identity; generated from pid + time when empty.
	WorkerID string
	// panic-recovery contract panic recovery is ON by default; this is the EXPLICIT opt-out, and it shifts
	// a panic from "retry with a recorded error" to "crash-attributed via the reclaimer".
	DisablePanicRecovery bool
	// singleton duties the reclaimer and promoter run under duty leases unless disabled.
	DisableDuties bool
	DutyInterval  time.Duration
}

type DeathEvent 

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

DeathEvent is emitted only after a fence-verified transition to archived succeeds. Envelope returns a deep copy so one callback cannot alter a later callback's view.

func (DeathEvent) Envelope 

func (e DeathEvent) Envelope() Envelope

func (DeathEvent) ErrorMessage 

func (e DeathEvent) ErrorMessage() string

func (DeathEvent) Reason 

func (e DeathEvent) Reason() DeathReason

func (DeathEvent) TerminalState 

func (e DeathEvent) TerminalState() string

type DeathHandler 

type DeathHandler interface {
	HandleDeath(context.Context, DeathEvent)
}

DeathHandler observes a job once when it becomes permanently archived, never once per ordinary retry. The durable transition has completed before this method is called.

type DeathHandlerFunc 

type DeathHandlerFunc func(context.Context, DeathEvent)

func (DeathHandlerFunc) HandleDeath 

func (f DeathHandlerFunc) HandleDeath(ctx context.Context, event DeathEvent)

type DeathReason 

type DeathReason string
const (
	DeathAttemptsExhausted DeathReason = "attempts_exhausted"
	DeathSkipped           DeathReason = "skipped"
	DeathDeadlineExceeded  DeathReason = "deadline_exceeded"
)

type DuplicateError 

type DuplicateError struct {
	ExistingID string
	Replaced   bool
}

DuplicateError carries the existing job's ID so the caller can join rather than guess. job uniqueness — one semantic across every backend, not silent-skip here and a hard error there.

func (*DuplicateError) Error 

func (e *DuplicateError) Error() string

func (*DuplicateError) Unwrap 

func (e *DuplicateError) Unwrap() error

type EnqueueAuthorization 

type EnqueueAuthorization struct {
	Source   EnqueueSource
	Identity *EnqueueIdentity
}

EnqueueAuthorization is supplied to every per-envelope authorization decision.

type EnqueueAuthorizeFunc 

type EnqueueAuthorizeFunc func(context.Context, EnqueueAuthorization, Envelope) bool

EnqueueAuthorizeFunc adapts a function into an EnqueueAuthorizer.

func (EnqueueAuthorizeFunc) AuthorizeEnqueue 

func (f EnqueueAuthorizeFunc) AuthorizeEnqueue(
	ctx context.Context,
	authorization EnqueueAuthorization,
	envelope Envelope,
) bool

type EnqueueAuthorizer 

type EnqueueAuthorizer interface {
	AuthorizeEnqueue(context.Context, EnqueueAuthorization, Envelope) bool
}

EnqueueAuthorizer is application policy. Returning false rejects the entire batch before any store call.

type EnqueueForbiddenError 

type EnqueueForbiddenError struct {
	Kind string
}

EnqueueForbiddenError is a typed policy rejection. It is neither a store outage nor a job failure.

func (*EnqueueForbiddenError) Error 

func (e *EnqueueForbiddenError) Error() string

func (*EnqueueForbiddenError) Unwrap 

func (e *EnqueueForbiddenError) Unwrap() error

type EnqueueIdentity 

type EnqueueIdentity struct {
	Subject    string
	Attributes map[string]string
}

EnqueueIdentity is established by the embedding application before headgate sees a request. Attributes are deliberately application-defined: the queue does not invent a role model. HTTP middleware attaches this value after authentication; headgate never trusts a caller-controlled identity header.

func EnqueueIdentityFromContext 

func EnqueueIdentityFromContext(ctx context.Context) (EnqueueIdentity, bool)

EnqueueIdentityFromContext returns the identity installed by trusted application middleware. ok=false means anonymous; the configured policy decides whether that is allowed.

type EnqueueMiddleware 

type EnqueueMiddleware interface {
	HandleEnqueue(context.Context, EnqueueRequest, EnqueueNext) error
}

EnqueueMiddleware wraps one logical producer call. The first registered middleware is the outermost wrapper: its before half runs first and its after half runs last. A middleware can mutate request, return without calling next to veto the operation, or invoke next more than once to implement an explicit retry.

type EnqueueMiddlewareFunc 

type EnqueueMiddlewareFunc func(context.Context, EnqueueRequest, EnqueueNext) error

EnqueueMiddlewareFunc adapts a function to EnqueueMiddleware.

func (EnqueueMiddlewareFunc) HandleEnqueue 

func (f EnqueueMiddlewareFunc) HandleEnqueue(
	ctx context.Context,
	request EnqueueRequest,
	next EnqueueNext,
) error

type EnqueueNext 

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

EnqueueNext is the remainder of the chain. It is reusable: retry middleware can call Run again with an owned request. A transactional terminal remains serialized by the caller's synchronous invocation.

func (EnqueueNext) Run 

func (n EnqueueNext) Run(ctx context.Context, request EnqueueRequest) error

type EnqueueOperation 

type EnqueueOperation string

EnqueueOperation identifies the terminal selected by the client. It is metadata for middleware: changing the field does not turn a direct call into a transactional one. 

const (
	EnqueueOperationDirect        EnqueueOperation = "direct"
	EnqueueOperationTransactional EnqueueOperation = "transactional"
)

type EnqueueRequest 

type EnqueueRequest struct {
	Source    EnqueueSource
	Operation EnqueueOperation
	Batch     []Envelope
}

EnqueueRequest is owned by the producer chain. Client clones the caller's batch before invoking middleware, so mutations affect what reaches authorization and the store without changing caller memory.

type EnqueueSource 

type EnqueueSource string

EnqueueSource tells an authorizer whether a decision came from an internal library client or the HTTP API. Policies never need to infer that distinction from headers. 

const (
	EnqueueSourceLibrary EnqueueSource = "library"
	EnqueueSourceHTTP    EnqueueSource = "http"
)

type Envelope 

type Envelope struct {
	ID, Kind, Queue         string
	SchemaVersion           uint32
	Payload                 []byte
	PartitionKey, RateClass string
	// Weight is the estimated rate-budget cost. Zero is the backward-compatible
	// omitted value and is normalized to one at every store boundary.
	Weight                             uint32
	Fingerprint                        string
	Priority                           int32
	Attempt, CrashAttempt, MaxAttempts uint32
	EnqueuedAtMs, ScheduledAtMs        int64
	TimeoutMs, DeadlineMs              int64
	UniqueKey                          []byte
	UniqueStates                       uint32
	// UniqueWindowMs selects the job uniqueness uniqueness mode. 0 = LIFECYCLE: one live job per
	// key, released by terminal state. > 0 = THROTTLE: at most one per window, released
	// by the clock. A caller-side duration that rounds to zero must be REJECTED (boundary validation),
	// never clamped into lifecycle mode.
	UniqueWindowMs int64
	// UniqueReplace is the request-only allowlist used when UniqueKey conflicts. It is
	// not persisted. Replacement is restricted to single-job enqueue calls.
	UniqueReplace uint32
	// UniqueDebounceMs is a trailing-edge store-clock debounce window. It requires
	// UniqueKey and reschedules the holder on every conflict.
	UniqueDebounceMs int64
	// UniqueExcludeKind removes Kind from the effective uniqueness key.
	UniqueExcludeKind bool
	// RetentionMs is how long a successful job's row is kept. 0 means DELETE on
	// completion (retention policy) — ephemeral, not keep-forever.
	RetentionMs int64
	// PeriodicScheduleID and PeriodicTickMs are typed durable origin. Empty/zero means
	// an ordinary enqueue; both fields are set together.
	PeriodicScheduleID string
	PeriodicTickMs     int64
	// Headers is opaque caller metadata carried with the job (proto field 20). The
	// store never interprets these bytes — it round-trips them. Two keys are RESERVED:
	// TraceparentHeader and TracestateHeader (W3C Trace Context, telemetry and trace context).
	Headers map[string]string
	// Tags are canonical operator-indexed labels, separate from opaque headers.
	Tags []string
	// Pending inserts durably but cannot be admitted until PromoteJob.
	Pending bool
	// StickyWorker is the exact stable worker identity allowed to claim this job.
	// Empty means any worker. Stores enforce it inside atomic admission.
	StickyWorker string
}

type Event 

type Event struct {
	// Type is one of: admitted | rejected | completed | quarantined | evicted
	// | job_span | worker_saturation | worker_memory.
	Type        string
	Queue, Kind string
	Fingerprint string
	Count       int
	Duration    time.Duration

	// Policy names the clause that refused the job when Type == "rejected", from the
	// admission-explain vocabulary (rate_class | concurrency_limit | fairness |
	// quarantine | schedule | queue_paused) — so a dashboard counting rejections by
	// policy and GET /jobs/{id}/admission use one word for one thing. This field matches
	// the Rust `Event::Rejected { queue, policy, count }` variant already carried.
	Policy string

	// Job-span fields (Type == "job_span").
	// Emitted exactly once per attempt, after the handler returns, carrying everything
	// an OTel-bridged deployment needs to build one span: identity, outcome, and — the
	// point of the addition — the traceparent the PRODUCER put on the envelope, already
	// parsed.
	//
	// It fires at the END and carries StartedAtMs + Duration rather than firing at the
	// start, because a facade has no span object to hand back: a start-only callback
	// would force every bridge to keep its own job-id -> span map and to leak one
	// whenever a worker is killed mid-attempt. An OTel span builder takes explicit start
	// and end timestamps, so one event is enough and nothing has to be remembered.
	JobID   string
	Attempt uint32
	// Outcome is success | retry | skip | revoke | snooze | undecodable | rate_limited.
	Outcome     string
	StartedAtMs int64
	// Trace is the parsed trace context from the envelope's reserved headers. The
	// ZERO VALUE (Trace.Valid() == false) means the envelope carried no traceparent OR
	// carried an invalid one — see ParseTraceparent. A bridge then starts a root span.
	Trace TraceContext

	// Worker-saturation fields (Type == "worker_saturation").
	// Emitted by the runner on every heartbeat, alongside the registry upsert that
	// already happens — so the same numbers reach a metrics exporter and GET /cluster
	// from one place and cannot disagree. This is a SIGNAL, not an autoscaler: headgate
	// never sizes a fleet, it only publishes the two numbers that decide the direction.
	//
	//   Utilization    = Inflight / Capacity — scale UP when high AND the backlog's
	//                    time-to-drain is growing (backlog metrics).
	//   EmptyPollRatio = admits returning zero / total admits over the rolling window —
	//                    scale DOWN when high: the fleet is asking for work that is not
	//                    there.
	Worker         string
	Inflight       uint32
	Capacity       uint32
	Utilization    float64
	EmptyPollRatio float64
	// Polls / EmptyPolls are the window totals behind the ratio, so an exporter can
	// publish counters too.
	Polls, EmptyPolls uint64

	// ----- rolling restart / memory guard (Type == "worker_memory") -----
	// Emitted on every configured sample. RestartRequested is true exactly once: the
	// sample that crossed the limit and sent the runner through graceful shutdown.
	MemoryBytes      uint64
	MemoryLimitBytes uint64
	RestartRequested bool
}

Event is the facade's payload. It is a struct rather than a sum type so new signals remain additive: bridges switch on Type and ignore fields they do not understand. Adding a field is compatible; renaming or repurposing one is not.

type EventBus 

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

EventBus is a bounded, non-blocking, process-local lifecycle fanout.

func NewEventBus 

func NewEventBus() *EventBus

func (*EventBus) Subscribe 

func (bus *EventBus) Subscribe(ctx context.Context, cfg SubscriptionConfig) (*Subscription, error)

type Extensions 

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

Extensions is a concurrency-safe heterogeneous type map. It is process-local and is not a field of Envelope: no extension can be serialized into a queued job by the runtime. Pass one instance in Config.Extensions for worker-shared dependencies; the runner creates a different empty instance for every job attempt.

func NewExtensions 

func NewExtensions() *Extensions

func (*Extensions) Len 

func (extensions *Extensions) Len() int

type ExtractionError 

type ExtractionError struct {
	Extractor string
	Message   string
}

ExtractionError reports which pre-handler extractor failed. It travels through the normal attempt error path, but the registered user function is never entered.

func (*ExtractionError) Error 

func (e *ExtractionError) Error() string

type ExtractorFunc 

type ExtractorFunc[T any] func(context.Context) (T, error)

func (ExtractorFunc[T]) Extract 

func (f ExtractorFunc[T]) Extract(ctx context.Context) (T, error)

type HandlerExtractor 

type HandlerExtractor[T any] interface {
	Extract(context.Context) (T, error)
}

HandlerExtractor constructs one typed handler parameter from the dispatch context. Applications can implement it directly; ExtractorFunc is the adapter for a function.

func ExtractAttempt 

func ExtractAttempt() HandlerExtractor[Attempt]

func ExtractClient 

func ExtractClient() HandlerExtractor[*JobClient]

func ExtractData 

func ExtractData[T any]() HandlerExtractor[T]

ExtractData resolves T from job data first, then worker data. Asking for the wrong concrete type is a missing-data error, never an untyped cast inside the handler.

func ExtractMeta 

func ExtractMeta[T any](decode func(Metadata) (T, error)) HandlerExtractor[T]

ExtractMeta validates the durable metadata into an application type before the handler runs. A missing or malformed header should be returned by decode as an error.

func ExtractMetadata 

func ExtractMetadata() HandlerExtractor[Metadata]

func ExtractTaskID 

func ExtractTaskID() HandlerExtractor[TaskID]

func ExtractWorkerContext 

func ExtractWorkerContext() HandlerExtractor[WorkerContext]

type HistoryBucket 

type HistoryBucket struct {
	AtMs, Arrived, Completed int64
}

type IDConflictError 

type IDConflictError struct{ JobID string }

IDConflictError is idempotent enqueue identity: the caller supplied an Envelope.ID that already names a row whose CONTENT differs. Distinct from DuplicateError, which is best-effort uniqueness over a key the caller opted into; this is the strict per-id guarantee asynq separates as TaskID(id) + ErrTaskIDConflict. Its own type because the two map to different API responses — folding it into a plain error, where it lived before, surfaced a 409 condition as a 400.

func (*IDConflictError) Error 

func (e *IDConflictError) Error() string

func (*IDConflictError) Unwrap 

func (e *IDConflictError) Unwrap() error

type IDGen 

type IDGen interface{ New() string }

type InsertAttempt 

type InsertAttempt struct {
	Source    EnqueueSource
	Operation EnqueueOperation
	// contains filtered or unexported fields
}

InsertAttempt is an immutable snapshot immediately before an enqueue store call. Batch returns a deep copy so a hook cannot mutate what this or a later hook observes, or what the store will persist. Request mutation belongs in enqueue middleware.

func (InsertAttempt) Batch 

func (a InsertAttempt) Batch() []Envelope

Batch returns an independently owned view of the attempted atomic batch.

type InsertHook 

type InsertHook interface {
	OnInsert(context.Context, InsertHookEvent)
}

InsertHook observes an actual store attempt without receiving a next function. Hooks run in registration order at both phases and cannot veto, mutate, retry, or replace the result. Expensive work should be handed to an asynchronous exporter.

type InsertHookEvent 

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

InsertHookEvent is one non-wrapping lifecycle point. Outcome returns false for Begin and the exact classified store result for End.

func (InsertHookEvent) Attempt 

func (e InsertHookEvent) Attempt() InsertAttempt

func (InsertHookEvent) Outcome 

func (e InsertHookEvent) Outcome() (InsertOutcome, bool)

func (InsertHookEvent) Phase 

func (e InsertHookEvent) Phase() InsertHookPhase

type InsertHookFunc 

type InsertHookFunc func(context.Context, InsertHookEvent)

func (InsertHookFunc) OnInsert 

func (f InsertHookFunc) OnInsert(ctx context.Context, event InsertHookEvent)

type InsertHookPhase 

type InsertHookPhase string
const (
	InsertHookBegin InsertHookPhase = "begin"
	InsertHookEnd   InsertHookPhase = "end"
)

type InsertOutcome 

type InsertOutcome struct {
	Kind       InsertOutcomeKind
	ExistingID string
	Replaced   bool
	JobID      string
	Err        error
}

InsertOutcome preserves the actual store result. Duplicate and ID conflict expose their useful identifiers; Err is the original typed error for every non-success.

type InsertOutcomeKind 

type InsertOutcomeKind string
const (
	// InsertOutcomeSucceeded includes a new insert and an idempotent same-ID replay;
	// Store intentionally returns nil for both.
	InsertOutcomeSucceeded  InsertOutcomeKind = "succeeded"
	InsertOutcomeDuplicate  InsertOutcomeKind = "duplicate"
	InsertOutcomeIDConflict InsertOutcomeKind = "id_conflict"
	InsertOutcomeRejected   InsertOutcomeKind = "rejected"
)

type InspectStore 

type InspectStore interface {
	Store
	GetJob(ctx context.Context, id string, includePayload bool) (*JobSummary, error)
	ListJobs(ctx context.Context, f JobFilter, cursor string, limit uint32) (JobPage, error)
	// Counts: nil queue = every queue; a non-nil pointer to "" = the queue literally
	// named "" — the same Option<&str> contract Rust's `counts` has. See JobFilter.
	Counts(ctx context.Context, queue *string) (StateCounts, error)
	QueueStats(ctx context.Context) ([]QueueStatsView, error)
	SetQueuePaused(ctx context.Context, queue string, paused bool) error
	SetQueueWeight(ctx context.Context, queue string, weight uint32) error
	SetEnqueueLimit(ctx context.Context, queue string, maxUnfinishedJobs *uint64) error
	RateClasses(ctx context.Context) ([]RateClassState, error)
	UpsertRateClass(ctx context.Context, cfg RateClassConfig) error
	ConcurrencyLimits(ctx context.Context) ([]ConcurrencyLimit, error)
	UpsertConcurrencyLimit(ctx context.Context, cfg ConcurrencyLimit) error
	Partitions(ctx context.Context, queue string) ([]PartitionState, error)
	QuarantineList(ctx context.Context) ([]QuarantineEntry, error)
	QuarantineRelease(ctx context.Context, fingerprint string) (released uint64, err error)
	OperatorRetry(ctx context.Context, id string) error
	OperatorCancel(ctx context.Context, id string) error
	PromoteJob(ctx context.Context, id string) error
	DeleteJob(ctx context.Context, id string) error
	ExplainAdmission(ctx context.Context, id string) (*AdmissionExplain, error)
	History(ctx context.Context, queue string, sinceMs, bucketMs int64) ([]HistoryBucket, error)
	// QuarantineSweep (crash quarantine): waiting jobs whose fingerprint is quarantined move to
	// the terminal quarantined state, VISIBLY — never an invisible gate-skip forever.
	QuarantineSweep(ctx context.Context, limit int64) (int64, error)
	RescheduleJob(ctx context.Context, id string, atMs int64) error
	EditPayload(ctx context.Context, id string, payload []byte, schemaVersion uint32, fingerprint string) error
	UpsertSchedule(ctx context.Context, s ScheduleEntry) error
	DeleteSchedule(ctx context.Context, id string) error
	ListSchedules(ctx context.Context) ([]ScheduleEntry, error)
	DueSchedules(ctx context.Context, limit int64) (due []ScheduleEntry, storeNowMs int64, err error)
	AdvanceSchedule(ctx context.Context, id string, fromNextRunMs, toNextRunMs int64) (bool, error)
	RecordScheduleEvent(ctx context.Context, event ScheduleEvent) error
	ListScheduleEvents(ctx context.Context, scheduleID string, beforeEventID uint64, limit uint32) ([]ScheduleEvent, error)
	// HeartbeatWorker upserts the worker row and returns any pending operator COMMAND —
	// the surveyed policy behavior control channel riding the heartbeat (Faktory's BEAT): "quiet" stops
	// admitting, "resume" resumes, "restart" drains without a timeout, "terminate"
	// performs a bounded shutdown, and "resign" releases the worker's singleton duties.
	// "" = none.
	HeartbeatWorker(ctx context.Context, w WorkerMeta) (command string, err error)
	ListWorkers(ctx context.Context, staleAfterMs int64) ([]WorkerMeta, error)
	// SignalWorker sets (or clears, with "") a worker's pending command. Runtimes
	// clear every command after applying it, then publish the acknowledged state.
	SignalWorker(ctx context.Context, workerID, command string) error
	// DistinctKinds: kinds present among waiting jobs (bounded sample), for typed dispatch's
	// startup warning about kinds no registered handler answers.
	DistinctKinds(ctx context.Context, limit int64) ([]string, error)
	CreateOperation(ctx context.Context, req BulkOp) error
	GetOperation(ctx context.Context, id string) (*OperationStatus, error)
	RunPendingOperations(ctx context.Context, batch int64) (uint64, error)
	DeleteQueue(ctx context.Context, queue string, force bool) (operationID string, err error)
	SampleQueueMemory(ctx context.Context, limit uint32) (sampled uint32, err error)
}

InspectStore is the control API's store surface, separate from Store the way TransactionalStore is (runtime capability boundary): a backend that cannot answer these does not have them. Every read is bounded — no method may be O(queue depth) (invariant 6).

type InvalidError 

type InvalidError struct{ Msg string }

InvalidError: a request rejected at the boundary — a bad cursor, a duration that rounds to zero (boundary validation), a transition the table does not define. 400.

NOTE THE ABSENT PREFIX. Rust's `StoreError::Invalid(m)` renders as "invalid request: {m}" through Display but the API serves the RAW `m`, because the 400 already says "invalid request". `Error()` here is "headgate: {m}" so that the API's existing TrimPrefix produces the same bytes on both servers — control API contract's raw-message contract.

func (*InvalidError) Error 

func (e *InvalidError) Error() string

func (*InvalidError) Unwrap 

func (e *InvalidError) Unwrap() error

type IsolatedOutcome 

type IsolatedOutcome string
const (
	IsolatedSuccess     IsolatedOutcome = "success"
	IsolatedRetry       IsolatedOutcome = "retry"
	IsolatedSkip        IsolatedOutcome = "skip"
	IsolatedRevoke      IsolatedOutcome = "revoke"
	IsolatedSnooze      IsolatedOutcome = "snooze"
	IsolatedRateLimited IsolatedOutcome = "rate_limited"
	IsolatedUndecodable IsolatedOutcome = "undecodable"
)

type IsolatedProcessConfig 

type IsolatedProcessConfig struct {
	Program        string
	Args           []string
	Env            map[string]string
	InheritEnv     bool
	MaxOutputBytes int64
}

IsolatedProcessConfig is a fixed executable invocation. The parent environment is cleared by default so child handlers receive only explicitly configured values.

type IsolatedRequest 

type IsolatedRequest struct {
	Version       uint32 `json:"version"`
	JobID         string `json:"job_id"`
	Kind          string `json:"kind"`
	SchemaVersion uint32 `json:"schema_version"`
	PayloadBase64 string `json:"payload_base64"`
	Queue         string `json:"queue"`
	PartitionKey  string `json:"partition_key"`
	RateClass     string `json:"rate_class"`
	Weight        uint32 `json:"weight"`
	Attempt       uint32 `json:"attempt"`
	CrashAttempt  uint32 `json:"crash_attempt"`
	MaxAttempts   uint32 `json:"max_attempts"`
	Fence         uint64 `json:"fence"`
	DeadlineMs    int64  `json:"deadline_ms"`
}

IsolatedRequest is the immutable versioned document written to the child stdin.

func (IsolatedRequest) Payload 

func (r IsolatedRequest) Payload() ([]byte, error)

type IsolatedResponse 

type IsolatedResponse struct {
	Version uint32          `json:"version"`
	Outcome IsolatedOutcome `json:"outcome"`
	Error   string          `json:"error,omitempty"`
	DelayMs int64           `json:"delay_ms,omitempty"`
}

IsolatedResponse follows IsolatedProtocolPrefix on one stdout line.

type Job 

type Job[T Args] struct {
	ID           string
	Args         T
	Queue        string
	Attempt      uint32 // failures the handler RETURNED
	CrashAttempt uint32 // crash quarantine failures where the worker DIED — counted separately
	MaxAttempts  uint32
	Fence        uint64 // rejects writes from a superseded lease holder
	PartitionKey string
	RateClass    string
	// Weight is the estimated surveyed policy behavior rate-budget cost. It is unrelated to weighted
	// queue selection: queue weight chooses a queue, this spends the chosen job's
	// rate-class budget.
	Weight   uint32
	Deadline time.Time
}

func (*Job[T]) Once 

func (j *Job[T]) Once(ctx context.Context, fn func(tx Tx) error) error

Once runs fn AT MOST ONCE per job ID, ever, committing atomically with the job's completion — transactional effects, the thing all three surveyed queues tell you to build yourself. Inside fn, do your writes on the given transaction (Unwrap to the driver's handle for raw access). If a previous delivery already committed the effect, fn is skipped and Once returns nil.

The guarantee comes from three things in ONE transaction: the effect-key claim, your writes, and the fence-verified completion. A superseded holder fails the completion, rolls everything back, and stops (ErrLeaseLost) — its half-done writes never commit. Requires a transactional store; Redis declines rather than approximating (runtime capability boundary).

type JobClient 

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

func ClientFromContext 

func ClientFromContext(ctx context.Context) (*JobClient, bool)

ClientFromContext returns the producer bound to this handler. There is no package global fallback: ok is false outside runtime dispatch.

func (*JobClient) Context 

func (client *JobClient) Context() context.Context

func (*JobClient) Enqueue 

func (client *JobClient) Enqueue(batch []Envelope) error

Enqueue submits follow-on work through the configured producer stack using the exact handler context. Parent cancellation/deadline therefore reaches middleware, hooks, authorization, and the Store. A valid W3C carrier is inherited only when the child did not explicitly set that header.

type JobEvent 

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

JobEvent is an immutable process-local snapshot of a persisted runtime outcome.

func (JobEvent) AtMs 

func (e JobEvent) AtMs() int64

func (JobEvent) Envelope 

func (e JobEvent) Envelope() Envelope

func (JobEvent) ErrorMessage 

func (e JobEvent) ErrorMessage() string

func (JobEvent) Kind 

func (e JobEvent) Kind() JobEventKind

func (JobEvent) State 

func (e JobEvent) State() string

type JobEventKind 

type JobEventKind string
const (
	JobEventCompleted JobEventKind = "completed"
	JobEventFailed    JobEventKind = "failed"
	JobEventCancelled JobEventKind = "cancelled"
)

type JobFilter 

type JobFilter struct {
	Queue, State, Kind, KindPrefix *string
	PartitionKey, ID               *string
	Fingerprint, RateClass         *string
	Priority                       *int32
	TagsAll, TagsAny               []string
}

JobFilter is control API contract's list/search predicate. Every field is a POINTER, and that is load-bearing rather than stylistic ; PORT CHANGE, reason recorded in the register's "Job search / filter" row).

An EMPTY value is a real, filterable value here. `partition_key` is the case that forces it: the empty string is the DEFAULT partition, so it is the single most common partition in any store that never set one — and `?partition_key=` is the only way to ask for it. Rust models these as `Option<String>`, so `?partition_key=` arrives as `Some("")` and filters FOR the empty value; Go's plain `string` collapsed that into "no filter" and answered with the WHOLE queue. Same divergence on `?queue=`, `?state=`, `?kind=` and every `field:` term of the `q=` grammar (`q=queue:` asks for the empty queue name). Rust is right and Go's port could not express the question at all, which is why the port moved rather than the semantic.

Nil means "no filter". A non-nil pointer to "" means "match the empty value".

type JobOutput 

type JobOutput struct {
	SchemaVersion uint32
	Bytes         []byte
	Fence         uint64
	UpdatedAtMs   int64
}

JobOutput is the latest opaque output persisted by a running fenced attempt. Fence identifies the attempt that wrote it; UpdatedAtMs comes from the store clock.

func PersistOutput 

func PersistOutput(ctx context.Context, schemaVersion uint32, bytes []byte) (*JobOutput, error)

PersistOutput replaces the job's versioned mid-run output under the current running lease and fence. It is durable before this call returns; a stolen holder is rejected and cannot overwrite output written by the new attempt.

type JobPage 

type JobPage struct {
	Jobs       []JobSummary
	NextCursor string
}

type JobProgress 

type JobProgress struct {
	Current     uint64
	Total       uint64
	Message     string
	Fence       uint64
	UpdatedAtMs int64
}

JobProgress is the latest report accepted from a fenced running attempt.

func ReportProgress 

func ReportProgress(ctx context.Context, current, total uint64, message string) (*JobProgress, error)

ReportProgress replaces this job's operator-facing progress under the current running lease. The store stamps the report and rejects a superseded holder.

type JobResult 

type JobResult struct {
	SchemaVersion uint32
	Bytes         []byte
}

type JobSummary 

type JobSummary struct {
	ID, Kind, Queue, State                string
	SchemaVersion                         uint32
	Priority                              int32
	Attempt, CrashAttempt, MaxAttempts    uint32
	PartitionKey, RateClass, StickyWorker string
	Weight                                uint32
	Fingerprint                           string
	EnqueuedAtMs, ScheduledAtMs           int64
	ClaimedAtMs                           *int64
	PeriodicScheduleID                    string
	PeriodicTickMs                        int64
	FinalizedAtMs                         *int64
	// Payload is nil unless explicitly requested (invariant 9).
	Payload []byte
	// Headers are opaque producer metadata and accompany only an explicit detail read.
	Headers    map[string]string
	ErrorsJSON string
	Tags       []string
}

func (JobSummary) IsOrphaned 

func (j JobSummary) IsOrphaned() bool

IsOrphaned reports durable provenance: the store has reclaimed this job from an expired worker lease at least once. It is derived from CrashAttempt, not a state.

type LeaseRef 

type LeaseRef struct {
	JobID   string
	LeaseID string
	Fence   uint64
}

LeaseRef identifies one claimed job for Ack/Renew. Admit writes ONE lease id for every job claimed in the same call, and Fence counts per job — so (leaseID, fence) alone is ambiguous: two jobs on their first claim in one call are both fence=1. JobID selects the row; LeaseID + Fence still gate the write (lease fencing) so a superseded holder is rejected, never silently no-opped.

type LeaseRejectedError 

type LeaseRejectedError struct{ JobID string }

LeaseRejectedError: the caller no longer holds this lease (reclaimed, or superseded by a newer fence). The worker must stop this job immediately (lease fencing).

func (*LeaseRejectedError) Error 

func (e *LeaseRejectedError) Error() string

func (*LeaseRejectedError) Unwrap 

func (e *LeaseRejectedError) Unwrap() error

type MemorySampler 

type MemorySampler interface {
	MemoryBytes() (uint64, error)
}

MemorySampler reports the worker process's memory footprint in bytes. The default sampler uses the process resident-set high-water mark where the standard library exposes it. Tests and unusual platforms can inject an equivalent process sampler.

type MemorySamplerFunc 

type MemorySamplerFunc func() (uint64, error)

func (MemorySamplerFunc) MemoryBytes 

func (f MemorySamplerFunc) MemoryBytes() (uint64, error)

type Metadata 

type Metadata struct {
	Queue, PartitionKey, RateClass string
	Weight                         uint32
	Priority                       int32
	SchemaVersion                  uint32
	Headers                        map[string]string
}

Metadata is the durable, non-payload envelope metadata visible at dispatch.

type MissedPolicy 

type MissedPolicy int

MissedPolicy decides what happens to periodic runs missed during downtime. surveyed policy behavior — NOTHING in the surveyed field backfills, including River, whose schedules live in the leader's memory and can skip a tick entirely across an election. 

const (
	MissedSkip     MissedPolicy = iota // default, matches every other queue
	MissedRunOnce                      // one catch-up run
	MissedBackfill                     // up to N catch-up runs
)

type NotFoundError 

type NotFoundError struct{ What string }

NotFoundError: the addressed job/schedule/worker/fingerprint does not exist. 404. `Error()` reproduces the literal every driver already wrote — "headgate: not found: job x" — so typing these changes the STATUS classification and not one wire byte.

func (*NotFoundError) Error 

func (e *NotFoundError) Error() string

func (*NotFoundError) Unwrap 

func (e *NotFoundError) Unwrap() error

type NotifyingStore 

type NotifyingStore interface {
	Store
	// WaitWakeup blocks up to timeout for a hint that work may be available. An empty
	// queues slice matches ANY queue (bounded live-control contract's one-subscription case). Returns the
	// waking queue's name ("" on burst overflow) and ok=true, or ok=false on timeout.
	// Wakeups may be spurious; a MISSED one costs latency, never correctness — the
	// poll fallback always stands. Mirrors Rust's Notifying::wait_wakeup.
	WaitWakeup(ctx context.Context, queues []string, timeout time.Duration) (queue string, ok bool, err error)
}

NotifyingStore provides push wakeup (push wakeups). MySQL never implements it — its wakeup latency floor is the poll interval — and PgBouncer in transaction pooling breaks it, which is why poll-only remains a first-class mode.

type OperationStatus 

type OperationStatus struct {
	ID, Status     string
	Affected       int64
	TotalEstimated int64
	DryRun         bool
	Error          string
}

type Outcome 

type Outcome int
const (
	OutcomeSuccess     Outcome = iota
	OutcomeRetry               // handler returned an error
	OutcomeSkip                // stop retrying, archive
	OutcomeRevoke              // drop entirely
	OutcomeSnooze              // reschedule without consuming an attempt
	OutcomeLeaseLost           // crash quarantine crash-attributed
	OutcomeUndecodable         // payload versioning
	// OutcomeRateLimited is NOT a failure: the job returns to available and `Attempt`
	// is not incremented. surveyed policy behavior — BullMQ and Sidekiq both treat it this way.
	OutcomeRateLimited
)

type OutputInspectStore 

type OutputInspectStore interface {
	GetJobOutput(ctx context.Context, id string) (*JobOutput, error)
}

type OutputStore 

type OutputStore interface {
	WriteJobOutput(ctx context.Context, lease LeaseRef, output JobResult) (*JobOutput, error)
}

OutputStore persists replace-style mid-run output without transitioning the job. The write must match running state, lease id, and fence atomically.

type PartitionState 

type PartitionState struct {
	PartitionKey string
	Deficit      int64
	Waiting      int64
}

type Performed 

type Performed struct {
	JobID, Kind string
	// Outcome is the telemetry and trace context outcome name the runtime acked (or would have): success |
	// retry | skip | revoke | snooze | undecodable | rate_limited | lease_lost.
	Outcome string
}

Performed is what PerformOne observed: which job ran, and what the runtime did with it.

type PeriodicEnqueueAttempt 

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

PeriodicEnqueueAttempt is an immutable snapshot of one durable schedule tick. Accessors return owned copies so a hook cannot alter the schedule, tick ID, unique key, or request.

func (PeriodicEnqueueAttempt) Envelope 

func (a PeriodicEnqueueAttempt) Envelope() Envelope

func (PeriodicEnqueueAttempt) Schedule 

func (a PeriodicEnqueueAttempt) Schedule() ScheduleEntry

func (PeriodicEnqueueAttempt) ScheduleID 

func (a PeriodicEnqueueAttempt) ScheduleID() string

func (PeriodicEnqueueAttempt) TickMs 

func (a PeriodicEnqueueAttempt) TickMs() int64

type PeriodicEnqueueHook 

type PeriodicEnqueueHook interface {
	OnPeriodicEnqueue(context.Context, PeriodicEnqueueHookEvent)
}

PeriodicEnqueueHook is a synchronous, schedule-aware observer. It cannot mutate or replace the durable tick request or Store result.

type PeriodicEnqueueHookEvent 

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

PeriodicEnqueueHookEvent surrounds one actual Store enqueue from the scheduler duty.

func (PeriodicEnqueueHookEvent) Attempt 

func (e PeriodicEnqueueHookEvent) Attempt() PeriodicEnqueueAttempt

func (PeriodicEnqueueHookEvent) Outcome 

func (e PeriodicEnqueueHookEvent) Outcome() (InsertOutcome, bool)

func (PeriodicEnqueueHookEvent) Phase 

func (e PeriodicEnqueueHookEvent) Phase() PeriodicEnqueueHookPhase

type PeriodicEnqueueHookFunc 

type PeriodicEnqueueHookFunc func(context.Context, PeriodicEnqueueHookEvent)

func (PeriodicEnqueueHookFunc) OnPeriodicEnqueue 

func (f PeriodicEnqueueHookFunc) OnPeriodicEnqueue(
	ctx context.Context,
	event PeriodicEnqueueHookEvent,
)

type PeriodicEnqueueHookPhase 

type PeriodicEnqueueHookPhase string
const (
	PeriodicEnqueueHookBegin PeriodicEnqueueHookPhase = "begin"
	PeriodicEnqueueHookEnd   PeriodicEnqueueHookPhase = "end"
)

type Plugin 

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

Plugin is an installable bundle of enqueue middleware and insert hooks. An empty kind set means global; a scoped plugin activates when any envelope in an atomic batch matches and then observes the whole batch.

func NewPlugin 

func NewPlugin(name string, options ...PluginOption) (Plugin, error)

NewPlugin constructs a global plugin unless WithPluginKinds is supplied.

func (Plugin) Kinds 

func (p Plugin) Kinds() []string

Kinds returns nil for a global plugin and an owned, deterministically sorted slice for a scoped plugin.

func (Plugin) Name 

func (p Plugin) Name() string

type PluginConfigError 

type PluginConfigError struct {
	Field string
	Msg   string
}

PluginConfigError identifies the invalid plugin field while preserving a stable sentinel for errors.Is.

func (*PluginConfigError) Error 

func (e *PluginConfigError) Error() string

func (*PluginConfigError) Unwrap 

func (e *PluginConfigError) Unwrap() error

type PluginOption 

type PluginOption func(*Plugin) error

PluginOption configures a Plugin while allowing validation to remain at construction.

func WithPluginEnqueueMiddleware 

func WithPluginEnqueueMiddleware(middlewares ...EnqueueMiddleware) PluginOption

WithPluginEnqueueMiddleware appends middleware inside the plugin's contiguous wrapper.

func WithPluginInsertHooks 

func WithPluginInsertHooks(hooks ...InsertHook) PluginOption

WithPluginInsertHooks appends sequential insert observers inside the plugin bundle.

func WithPluginKinds 

func WithPluginKinds(kinds ...string) PluginOption

WithPluginKinds scopes the plugin to a sorted-set-equivalent collection of task kinds. At least one kind is required; duplicate kinds are harmless.

type ProgressInspectStore 

type ProgressInspectStore interface {
	GetJobProgress(ctx context.Context, id string) (*JobProgress, error)
}

type ProgressStore 

type ProgressStore interface {
	WriteJobProgress(ctx context.Context, lease LeaseRef, update ProgressUpdate) (*JobProgress, error)
}

ProgressStore persists replace-style operator progress without transitioning the job. The write must match running state, lease id, and fence atomically.

type ProgressUpdate 

type ProgressUpdate struct {
	Current uint64
	Total   uint64
	Message string
}

ProgressUpdate is an exact operator-facing progress fraction with an optional short status message. Use Total=100 for a percentage; progress is not a log channel.

type QuarantineEntry 

type QuarantineEntry struct {
	Fingerprint, Kind, Reason string
	CrashCount                int64
	QuarantinedAtMs           int64
}

type QuarantinedError 

type QuarantinedError struct{ Fingerprint string }

QuarantinedError carries the fingerprint an enqueue was rejected for (crash quarantine).

func (*QuarantinedError) Error 

func (e *QuarantinedError) Error() string

func (*QuarantinedError) Unwrap 

func (e *QuarantinedError) Unwrap() error

type QueueConfig 

type QueueConfig struct {
	MaxWorkers int
}

type QueueStatsView 

type QueueStatsView struct {
	Queue string
	// Weight selects BETWEEN queues inside the atomic gate. It is unrelated to the
	// envelope Weight that spends one selected job's rate budget.
	Weight uint32
	// UnfinishedJobs is exact O(1) producer depth, unlike bounded/approximate ByState.
	UnfinishedJobs uint64
	// nil disables producer backpressure; zero rejects every new unfinished job.
	MaxUnfinishedJobs *uint64
	ByState           map[string]int64
	CountsApproximate bool
	ArrivalRate       float64
	DrainRate         float64
	// TimeToDrainMs is nil when arrival >= drain — the alert condition (backlog metrics).
	TimeToDrainMs *int64
	// OldestAvailableMs is the store-clock age of the oldest currently available job.
	// Nil means there is no available job; it is an age so it is directly SLO-shaped.
	OldestAvailableMs *int64
	QuietGroups       QuietGroupMetrics
	Paused            bool
	// nil until an explicit bounded sampler has stored an estimate.
	MemoryBytes *uint64
}

type QuietGroupMetrics 

type QuietGroupMetrics struct {
	ArrivalRate       float64
	DrainRate         float64
	TimeToDrainMs     *int64
	OldestAvailableMs *int64
	NoisyPartitions   uint32
	Approximate       bool
}

type RateClass 

type RateClass struct {
	Name  string
	Limit uint64
	Per   time.Duration
	Burst uint64
}

type RateClassConfig 

type RateClassConfig struct {
	Name     string
	Limit    int64
	WindowMs int64
	Burst    int64
	// Paused is the invariant-16 kill switch: admit nothing until unpaused.
	Paused bool
}

type RateClassState 

type RateClassState struct {
	Name            string
	TokensAvailable int64
	Burst           int64
	LimitPerWindow  int64
	WindowMs        int64
	JobsWaiting     int64
	Paused          bool
}

type Reclaimed 

type Reclaimed struct {
	JobID        string
	Fingerprint  string
	CrashAttempt uint32
	Quarantined  bool
}

Reclaimed is a job the lease reclaimer swept. Quarantined tells the caller which counter and event to emit — eviction and quarantine are never silent (retention and eviction contract).

type Registry 

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

Registry maps kind -> handler. Registration enforces typed dispatch's invariant — every kind and alias unique — at startup, not one failing job at a time in production.

func NewRegistry 

func NewRegistry() *Registry

type ResultInspectStore 

type ResultInspectStore interface {
	GetJobResult(ctx context.Context, id string) (*JobResult, error)
}

type ResultStore 

type ResultStore interface {
	AckSuccessWithResult(
		ctx context.Context,
		lease LeaseRef,
		logs []string,
		actualWeight *uint32,
		result JobResult,
	) error
}

ResultStore atomically records versioned bytes with the fenced success transition. It is separate so a backend that cannot honor results does not silently accept them.

type RetryPolicy 

type RetryPolicy interface {
	NextRetry(attempt uint32, err error) time.Duration
}

type Runner 

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

func NewRunner 

func NewRunner(store Store, reg *Registry, cfg Config) *Runner

func (*Runner) Drain 

func (r *Runner) Drain(ctx context.Context, n int) ([]string, error)

Drain admits up to n jobs and runs each straight through its handler and ack, synchronously — Oban's drain_queue, the most useful helper in an integration test. Due scheduled/retryable jobs are promoted first so "fail, then drain again" exercises a real retry without sleeping through the backoff.

func (*Runner) PerformOne 

func (r *Runner) PerformOne(ctx context.Context) (Performed, bool, error)

PerformOne runs EXACTLY ONE job through the real dispatch path and says what happened to it — River's rivertest.Worker.Work / Oban's perform_job, the second helper every serious queue ships. The Rust twin is headgate::testing::perform_job.

the register claimed this and had nothing behind it. Drain(n) runs a batch and returns ids, so a test that wanted "run this one job and tell me the outcome" had to drain and then re-read the store to infer what the runtime decided — which asserts the STORE's opinion, not the runtime's, and cannot see the outcomes that never reach a row at all (lease_lost).

It is the real path, not a shortcut: the same Admit the run loop makes (capacity ONE, so the gate really chooses the job), the same processOne, the same ack. ok is false when the gate admitted nothing — which is itself an assertable fact.

func (*Runner) Run 

func (r *Runner) Run(ctx context.Context) error

Run until Shutdown() (or ctx cancellation). Store outages degrade to backoff-and- retry, never a crash of the loop.

func (*Runner) Shutdown 

func (r *Runner) Shutdown()

type SaturationStrategy 

type SaturationStrategy string

SaturationStrategy is the wire/storage spelling read by every atomic gate. 

const (
	SaturateQueue          SaturationStrategy = "queue" // wait (default)
	SaturateDiscard        SaturationStrategy = "discard"
	SaturateCancelRunning  SaturationStrategy = "cancel_running"  // newest wins
	SaturateCancelIncoming SaturationStrategy = "cancel_incoming" // oldest wins
)

func (SaturationStrategy) Valid 

func (s SaturationStrategy) Valid() bool

type ScheduleEntry 

type ScheduleEntry struct {
	ID, Kind      string
	Payload       []byte
	Queue         string
	PartitionKey  string
	RateClass     string
	Priority      int32
	MaxAttempts   uint32
	RetentionMs   int64
	Spec          string
	NextRunMs     int64
	LastEnqueued  *int64
	OnMissed      MissedPolicy
	BackfillLimit uint32
	Paused        bool
}

ScheduleEntry is a surveyed policy behavior periodic entry — durable in the store, never in a leader's memory. Spec is "@every:<ms>" (epoch-aligned) or a UTC cron expression; both languages must derive identical tick times, because ticks feed unique keys.

type ScheduleEvent 

type ScheduleEvent struct {
	EventID      uint64
	ScheduleID   string
	TickMs       int64
	JobID        string
	Outcome      ScheduleEventOutcome
	Reason       string
	RecordedAtMs int64
}

ScheduleEvent is one durable scheduler enqueue attempt. Reason is a stable, low-cardinality classification, never a raw backend error or payload.

type ScheduleEventOutcome 

type ScheduleEventOutcome string
const (
	ScheduleEventEnqueued     ScheduleEventOutcome = "enqueued"
	ScheduleEventDeduplicated ScheduleEventOutcome = "deduplicated"
	ScheduleEventFailed       ScheduleEventOutcome = "failed"
	ScheduleEventSkipped      ScheduleEventOutcome = "skipped"
	ScheduleEventLimit                             = uint32(100)
)

func (ScheduleEventOutcome) Valid 

func (o ScheduleEventOutcome) Valid() bool

type SnoozeError 

type SnoozeError struct{ Delay time.Duration }

SnoozeError re-schedules without consuming an attempt. Return Snooze(d) from a handler. A duration that rounds to zero milliseconds is a handler bug and is acked as a retry with an explanatory error (boundary validation — never clamped).

func (*SnoozeError) Error 

func (e *SnoozeError) Error() string

type StaleCheckpointError 

type StaleCheckpointError struct{ Expected, Got string }

StaleCheckpointError: the step set changed under the checkpoint (payload versioning × step replay). The runner acks Undecodable — silently restarting would re-run completed side effects with no signal that a deploy caused it.

func (*StaleCheckpointError) Error 

func (e *StaleCheckpointError) Error() string

type StateCounts 

type StateCounts struct {
	Counts      map[string]int64
	Approximate bool
}

type Store 

type Store interface {
	Admit(ctx context.Context, req AdmitRequest) ([]AdmissionUnit, error)
	// Ack applies the transition table. delayMs: required for OutcomeSnooze (> 0); for
	// OutcomeRetry it overrides the store's default backoff (0 = default); ignored
	// otherwise. OutcomeLeaseLost is never acked — it is the reclaimer's transition.
	// Equivalent to AckAttempt with no logs.
	Ack(ctx context.Context, lease LeaseRef, outcome Outcome, errMsg string, delayMs int64) error
	// AckAttempt is Ack plus attempt-log contract per-attempt execution logs (River's riverlog):
	// captured handler log lines land INSIDE the attempt's error-history entry.
	// Recorded for success/retry/skip/undecodable (non-empty logs on success write a
	// success entry — the only time one exists); dropped for snooze/rate_limited/
	// revoke, which by design record no attempt entry.
	AckAttempt(ctx context.Context, lease LeaseRef, outcome Outcome, errMsg string, delayMs int64, logs []string) error
	// AckAttemptWithActualWeight atomically applies the transition and reconciles the
	// envelope's estimated rate-budget charge. nil means estimate == actual; a pointer
	// to zero is a real full refund. Keeping this in ack means a rejected fence can never
	// leave a separately committed correction behind.
	AckAttemptWithActualWeight(ctx context.Context, lease LeaseRef, outcome Outcome, errMsg string, delayMs int64, logs []string, actualWeight *uint32) error
	// Renew extends leases and returns the JOB IDS whose lease was lost. A worker that
	// lost a lease must be able to stop — a silent no-op here is how asynq stranded
	// jobs in ACTIVE since 2022.
	Renew(ctx context.Context, leases []LeaseRef, lease time.Duration) (lostJobIDs []string, err error)
	Enqueue(ctx context.Context, batch []Envelope) error
	// Checkpoint persists step progress, fence-verified: it succeeds only while the
	// caller still holds the lease, so it doubles as the step boundary's lease check.
	// ErrLeaseLost here means STOP before the next step's side effects. Durable BEFORE
	// the step runs, never after the worker returns (step replay — River's mistake).
	Checkpoint(ctx context.Context, lease LeaseRef, cp Checkpoint) error
	// ReclaimExpired turns expired leases into OutcomeLeaseLost — NEVER OutcomeRetry:
	// crash_attempt increments, attempt does not, and quarantine depends on the
	// difference (crash quarantine). Safe under contention; run it under a duty lease.
	ReclaimExpired(ctx context.Context, limit int64) ([]Reclaimed, error)
	// PromoteDue is the schedule_due/backoff_due sweep: due scheduled and retryable
	// jobs become available. Returns how many were promoted.
	PromoteDue(ctx context.Context, limit int64) (int64, error)
	// EvictRetained is the retention and eviction contract retention sweep: terminal jobs whose
	// finalized_at_ms + retention_ms has lapsed are deleted (retention 0 was already
	// deleted at ack time). quarantined is exempt — it parks visibly until an
	// operator acts. Bounded per call; run under the retention duty lease.
	EvictRetained(ctx context.Context, limit int64) (int64, error)
	// ClaimDuty claims (or renews) a singleton duties singleton duty — the same compare-and-set as
	// claiming a job, on store time. false = someone else holds it; skip the tick.
	ClaimDuty(ctx context.Context, name, holder string, lease time.Duration) (bool, error)
	// ReleaseDuty steps down by expiring the duty immediately, so takeover is fast.
	ReleaseDuty(ctx context.Context, name, holder string) error
	Caps() Caps
}

Store is the whole port. Four methods, deliberately coarse: the admission decision must be atomic inside the store, so a fine-grained get/set/claim port would force the gate back into the worker — which is the mistake this design exists to avoid.

type StuckJobEvent 

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

StuckJobEvent is emitted only after cancellation has remained unobserved for the configured threshold. Envelope returns a deep copy so the callback cannot mutate the attempt's metadata.

func (StuckJobEvent) Envelope 

func (e StuckJobEvent) Envelope() Envelope

func (StuckJobEvent) Reason 

func (e StuckJobEvent) Reason() StuckReason

func (StuckJobEvent) Threshold 

func (e StuckJobEvent) Threshold() time.Duration

type StuckJobHandler 

type StuckJobHandler interface {
	HandleStuck(context.Context, StuckJobEvent)
}

StuckJobHandler is the singular operational escalation point for attempts that fail to cooperate with timeout, lease-loss, or shutdown cancellation.

type StuckJobHandlerFunc 

type StuckJobHandlerFunc func(context.Context, StuckJobEvent)

func (StuckJobHandlerFunc) HandleStuck 

func (f StuckJobHandlerFunc) HandleStuck(ctx context.Context, event StuckJobEvent)

type StuckReason 

type StuckReason string
const (
	StuckCancellation StuckReason = "cancellation"
	StuckTimeout      StuckReason = "timeout"
)

type Subscription 

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

func (*Subscription) Close 

func (subscription *Subscription) Close()

func (*Subscription) Dropped 

func (subscription *Subscription) Dropped() uint64

func (*Subscription) Events 

func (subscription *Subscription) Events() <-chan JobEvent

type SubscriptionConfig 

type SubscriptionConfig struct {
	// ChanSize defaults to 64. Negative values are invalid.
	ChanSize int
	// Empty means every event kind.
	Kinds []JobEventKind
}

type TaskID 

type TaskID string

type Telemetry 

type Telemetry interface{ OnEvent(Event) }

Telemetry is a facade. telemetry and trace context: core links against no exporter, ever.

type TraceContext 

type TraceContext struct {
	TraceID    string // 32 lowercase hex characters, never all zero
	SpanID     string // 16 lowercase hex characters, never all zero — the PARENT span
	TraceFlags uint8  // bit 0 is `sampled`
	TraceState string // verbatim tracestate, empty when absent. Never parsed.
}

TraceContext is a parsed traceparent (plus the unparsed tracestate).

Producers set the headers at enqueue; the runtime parses traceparent at DISPATCH and hands the result to the handler (TraceContextFrom) and to the telemetry facade. See ParseTraceparent for what "lenient" means here.

func ParseTraceparent 

func ParseTraceparent(value string) (TraceContext, bool)

ParseTraceparent parses a W3C traceparent value: `00-{32 lowercase hex}-{16 lowercase hex}-{2 hex}`.

LENIENT MEANS LENIENT ABOUT THE CONSEQUENCE, STRICT ABOUT THE FORMAT. An unparseable value is treated as ABSENT (ok == false) and is never an enqueue error and never a dispatch failure. The headers stay opaque bytes to the store either way, so a malformed trace header can lose you a trace link and can never lose you a job. The Rust runtime implements this identically (headgate_core::parse_traceparent); a divergence would mean one runtime silently drops a parent the other honours.

Rejected, each for a reason W3C names: a version other than 00 (this specification pins one version rather than guessing at a future one's field layout); uppercase hex (W3C mandates lowercase, and accepting both would make two producers disagree about whether two ids are the same id); an all-zero trace-id or span-id (explicitly invalid in the spec); any field of the wrong length, or extra/missing `-`-separated fields.

func TraceContextFrom 

func TraceContextFrom(ctx context.Context) (TraceContext, bool)

TraceContextFrom returns the W3C trace context the PRODUCER put on the envelope, parsed at dispatch (telemetry and trace context). ok is false when the reserved `traceparent` header was absent OR malformed — the two are deliberately indistinguishable, because a handler that behaved differently for a typo'd header would be a worse bug than a missing trace link. Outside a running job (no runner context) it is also false.

Use it to parent a span, or to propagate the trace into a downstream call: tc.Traceparent() re-emits the producer's exact bytes.

func TraceContextOf 

func TraceContextOf(headers map[string]string) (TraceContext, bool)

TraceContextOf is the dispatch-time read: pull TraceparentHeader out of an envelope's headers and parse it, attaching TracestateHeader verbatim. ok is false when the header is absent OR invalid — the two are deliberately indistinguishable to callers.

func (TraceContext) Sampled 

func (t TraceContext) Sampled() bool

Sampled is W3C's sampled flag (bit 0 of trace-flags).

func (TraceContext) Traceparent 

func (t TraceContext) Traceparent() string

Traceparent re-renders the header value. Round-trips ParseTraceparent exactly, so a runtime that re-injects the context into a downstream call emits the same bytes the producer sent.

func (TraceContext) Valid 

func (t TraceContext) Valid() bool

Valid reports whether this is a real parsed context rather than the zero value.

type TransactionalStore 

type TransactionalStore interface {
	Store
	// BeginTx/CommitTx/RollbackTx are the dyn path (transactional API): for code that only knows
	// TransactionalStore, like Job.Once. Callers with their own driver transaction
	// wrap it instead (caller-owned transaction contract).
	BeginTx(ctx context.Context) (Tx, error)
	CommitTx(ctx context.Context, tx Tx) error
	RollbackTx(ctx context.Context, tx Tx) error
	EnqueueTx(ctx context.Context, tx Tx, batch []Envelope) error
	CompleteTx(ctx context.Context, tx Tx, lease LeaseRef) error
	// CompleteTxWithActualWeight is the Once/transactional counterpart of
	// AckAttemptWithActualWeight. Correction, caller effects, and fenced completion are
	// one commit or one rollback.
	CompleteTxWithActualWeight(ctx context.Context, tx Tx, lease LeaseRef, actualWeight *uint32) error
	// ClaimEffect (transactional effects) claims an effect key inside the caller's transaction. false
	// means a COMMITTED transaction already claimed it — the effect ran; skip the work.
	// The claim commits (or vanishes) with everything else in the transaction, which
	// is the entire mechanism behind at-most-once effects.
	ClaimEffect(ctx context.Context, tx Tx, key string) (bool, error)
	// CheckpointTx (step replay × transactional effects) writes the checkpoint inside the caller's
	// transaction, fence-verified — what makes a step's effects and its completion
	// marker ONE commit (see StepOnce).
	CheckpointTx(ctx context.Context, tx Tx, lease LeaseRef, cp Checkpoint) error
}

TransactionalStore exists separately so a backend that cannot honor it does not have it (runtime capability boundary). Redis implements Store and not this — no silent no-ops, no runtime surprise.

type Tx 

type Tx interface{ Unwrap() any }

Tx is a caller-owned store transaction. Drivers wrap their concrete handle and recover it via Unwrap — the Go mirror of Rust's TxHandle::as_any (transactional API): the compile-time path is typed, the dyn path downcasts, and a foreign handle is a hard error, never a silent no-op. (An unexported method here would seal the interface and make TransactionalStore unimplementable outside this package.)

type UnavailableError 

type UnavailableError struct{ Msg string }

UnavailableError: typed availability errors, the store is unreachable — a refused dial, a closed pool, a reset connection. 503, and typed APART from a validation failure precisely so a caller can tell "your request was wrong" from "come back later". This is the variant whose absence made a dropped connection a 400.

func (*UnavailableError) Error 

func (e *UnavailableError) Error() string

func (*UnavailableError) Unwrap 

func (e *UnavailableError) Unwrap() error

type UndecodableError 

type UndecodableError struct{ Cause error }

UndecodableError: the payload cannot decode into the registered type and never will (payload versioning). The runner acks Undecodable rather than retrying a decode error 25 times.

func (*UndecodableError) Error 

func (e *UndecodableError) Error() string

func (*UndecodableError) Unwrap 

func (e *UndecodableError) Unwrap() error

type Versioned 

type Versioned interface {
	Args
	Version() uint32
	// Upcast decodes an older payload into the current shape. Returning
	// ErrNoUpcastPath sends the job to `undecodable` instead of retrying it 25 times.
	Upcast(version uint32, payload []byte) (Args, error)
}

Versioned is optional. Implement it the day you ship, not the day you need it — payload versioning: a schema_version cannot be added retroactively to jobs already in the queue.

type Worker 

type Worker[T Args] interface {
	Work(ctx context.Context, job *Job[T]) error
}

type WorkerContext 

type WorkerContext struct {
	WorkerID string
	Queues   []string
	Capacity int
}

WorkerContext contains stable facts about the runner, not its dependency container.

type WorkerMeta 

type WorkerMeta struct {
	WorkerID, Host string
	PID            int32
	Queues         []string
	// Concurrency is the worker's configured capacity — the denominator of
	// Inflight / Concurrency.
	Concurrency   uint32
	StartedAtMs   int64
	HeartbeatAtMs int64

	// Inflight is how many jobs this worker is running right now.
	Inflight uint32
	// Polls is admissions attempted in the runner's rolling window.
	Polls uint64
	// EmptyPolls is how many of those returned zero jobs. The RATIO is the scale-down
	// signal; the two counters ride the wire instead of a float so the aggregate is
	// exact and so neither language has to agree with the other about float formatting.
	EmptyPolls uint64

	// Status is the worker-acknowledged control state: running, quiet, restarting,
	// or terminating. PendingCommand is the store mailbox and is populated only by
	// inspection reads; workers never write it in a heartbeat.
	Status         string
	DutiesActive   bool
	PendingCommand string
}

func (WorkerMeta) EmptyPollRatio 

func (w WorkerMeta) EmptyPollRatio() float64

EmptyPollRatio is backlog metrics's empty admissions / total admissions over the reported window. 0 when the window is empty — an idle-since-startup worker has no evidence either way, and reporting 1.0 there would signal "scale down" from no data at all.

func (WorkerMeta) Utilization 

func (w WorkerMeta) Utilization() float64

Utilization is backlog metrics's Inflight / Concurrency. 0 when capacity is 0 — never a division by zero, and never 1.0 for a worker that cannot run anything.

Source Files

View all Source files

Directories

Path Synopsis
driver
headgatemysql module
headgatepgx module
headgateredis module
headgateapi module
headgatectl module
headgateotel module
headgatetest module
headgateui module
Package postgressql owns safe, explicit Postgres object qualification shared by the driver and migrator.
 Package postgressql owns safe, explicit Postgres object qualification shared by the driver and migrator.
proto
headgatev1

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.