engine

package
v0.0.8 Latest Latest
Warning

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

Go to latest
Published: Jul 31, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package engine is the shared fetch/execute/settle/maintenance machinery the queue and event runtimes are built on, neutral over driver.Source.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Backoff

func Backoff(attempt int) time.Duration

Backoff returns the retry delay for a 1-based attempt: exponential growth (2^n seconds, exponent capped at 8) with deterministic jitter (no rand dependency, so results are reproducible in tests), hard-capped at 300s. A negative attempt is treated as zero.

func ExtractTraceContext added in v0.0.4

func ExtractTraceContext(ctx context.Context, meta map[string]string) context.Context

ExtractTraceContext returns a context carrying the remote span described by meta's traceparent/tracestate, or ctx unchanged when meta carries none (or is nil).

func InjectTraceContext added in v0.0.4

func InjectTraceContext(ctx context.Context, meta map[string]string) map[string]string

InjectTraceContext writes ctx's current span, if valid, into meta as traceparent/tracestate, allocating meta when nil, and returns it. Without a valid span (no sampled trace in ctx) meta is returned unchanged, so a producer with tracing off never allocates a Meta map on ctx's account.

Types

type Classifier

type Classifier func(error) Outcome

Classifier maps a handler error to its Outcome. Each consuming package injects its own (the queue's Abort/Retry/RetryAfter/Reportable taxonomy, the event bus's Permanent).

type Config

type Config struct {
	// Store is the persistence driver the engine fetches from and settles into.
	Store driver.Store
	// Source is the job partition this engine operates; it never touches jobs of
	// another source.
	Source driver.Source
	// Logger is the structured logger. Nil means slog.Default().
	Logger *slog.Logger
	// Settings are the resolved runtime knobs.
	Settings Settings
	// Acker settles a successful job, receiving the handler's result. Nil
	// means the default: Store.Ack with the result discarded (queue and event
	// handlers produce none). The workflow runtime injects AckTaskResult here
	// so a task's output is persisted atomically with its completion. Like
	// every settlement it is fenced: a not-found error is swallowed as the
	// expected lost-lease race.
	Acker func(ctx context.Context, id, leaseToken uuid.UUID, result json.RawMessage) error
}

Config assembles an Engine for one job source. The queue runtime builds one with Source queue and the event runtime one with Source event; the engine is identical for both.

type Engine

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

Engine is the shared fetch/execute/settle/maintenance machinery, neutral over driver.Source. A runtime constructs one, registers its kinds, and calls Start; the engine owns the slot-reservation invariant (executor capacity is reserved before any job is leased), lease renewal with fencing, settlement and the maintenance loops.

func New

func New(cfg Config) *Engine

New builds an Engine from cfg. Kinds are registered afterwards with Register, before Start.

func (*Engine) Done added in v0.0.4

func (e *Engine) Done() <-chan struct{}

Done closes once Start has returned, for any reason. Calling Done before Start has ever run returns a channel that never closes; check Started first. Like Ready, Done is a one-shot gate tied to the first Start invocation: if a Start that failed during setup is retried and later succeeds, Done was already closed by the failed attempt and stays closed while the retry is running. Done is meant for the common case (wait for the one Start call to finish); a caller that retries Start after a setup failure should not rely on Done to detect the retried call's completion.

func (*Engine) Ready

func (e *Engine) Ready() <-chan struct{}

Ready closes once wakeup setup succeeded and the loops are running (or once Start has returned, on any error, so a caller blocked on Ready never hangs past a failed Start). Poll-only engines become ready immediately after Start. A retried Start after a failed setup reuses the same Engine, and Ready is a one-shot gate: after the first Start attempt resolves, Ready is permanently closed even across a later, successful retry. Callers pairing Ready with Start should inspect Start's returned error.

func (*Engine) Register

func (e *Engine) Register(k Kind) error

Register adds one kind. It fails on a duplicate name and after Start.

func (*Engine) Start

func (e *Engine) Start(ctx context.Context) error

Start runs the engine until ctx is cancelled: one fetch loop per registered kind (push wake + poll fallback) and the maintenance loop. On cancellation, in-flight handlers drain for up to Settings.ShutdownDrain, then are cancelled. Handlers run on a context derived from Background so they survive the shutdown of ctx during the drain window. Start returns nil after a graceful shutdown (ctx cancelled, drained or not); it returns a non-nil error only for a setup failure (already started, no store, wake setup failed).

func (*Engine) Started

func (e *Engine) Started() bool

Started reports whether Start has been called (and not failed during setup).

type Kind

type Kind struct {
	// Name is the fetch partition (job kind, or subscriber name for events).
	Name string
	// Concurrency caps concurrent handlers of this kind.
	Concurrency int
	// Timeout bounds one handler execution; 0 means unlimited.
	Timeout time.Duration
	// MaxAttempts is the retry budget dequeues resolve durably on a job's first
	// lease when MaxAttemptsSet is true (see driver.DequeueParams).
	MaxAttempts int
	// MaxAttemptsSet marks MaxAttempts as an explicit per-kind override.
	MaxAttemptsSet bool
	// Handler runs one leased job. Its result travels to the engine's acker on
	// success; runtimes without task results (queue, event) return nil.
	Handler func(ctx context.Context, job driver.Job) (json.RawMessage, error)
	// Classify maps a handler error to its outcome. Nil means plain retry.
	Classify Classifier
}

Kind registers one job kind on the engine: its limits, its handler and its error classifier. Decoding and error taxonomy live with the consumer; the engine only sees driver.Job in and error out.

type Outcome

type Outcome struct {
	Kind OutcomeKind
	// Delay overrides the exponential backoff for a retry and is the snooze
	// duration for OutcomeSnooze; 0 means Backoff for a retry and an
	// immediate re-check for a snooze.
	Delay time.Duration
	// Reportable flags the error for loud logging when retries are exhausted.
	Reportable bool
}

Outcome is a consuming package's classification of a handler error. The engine turns it into a settlement: Snooze parks the job budget-free after Delay; Abort or an exhausted budget dead-letters the job; anything else reschedules it after Delay (or the exponential backoff when Delay is zero).

func ClassifyOutcome added in v0.0.6

func ClassifyOutcome(err error) Outcome

ClassifyOutcome is the shared classifier every runtime's sentinel types resolve through: an OutcomeError anywhere in err's chain decides the outcome; anything else is a plain retry.

type OutcomeError added in v0.0.6

type OutcomeError interface {
	error
	// AsyncOutcome returns the engine settlement the sentinel encodes.
	AsyncOutcome() Outcome
}

OutcomeError is the cross-runtime face of a runtime's error sentinels: the private error types behind queue.Abort, dag.NotReady, event.Permanent and friends implement it, so ANY runtime can classify a sentinel minted by ANOTHER runtime's package with errors.As against this interface — a handler migrating between runtimes keeps its failure behavior instead of silently degrading (a dag.NotReady inside a workflow Operation used to burn retry budget; now it snoozes there too).

type OutcomeKind

type OutcomeKind int

OutcomeKind is the classified fate of a failed handler.

const (
	// OutcomeRetry reschedules the job (the default for plain errors).
	OutcomeRetry OutcomeKind = iota
	// OutcomeAbort sends the job straight to the dead letter.
	OutcomeAbort
	// OutcomeSnooze parks the job as scheduled after Delay via Store.Snooze,
	// without consuming a retry attempt and regardless of the remaining
	// budget: the polling-wait primitive (the DAG runtime maps NotReady to
	// it). A job carrying a snooze budget is the one exception: a snooze
	// settled past its stamped deadline dead-letters instead (see
	// driver.Store.Snooze).
	OutcomeSnooze
	// OutcomeSkip settles the job as StateSkipped via Store.Skip: terminal,
	// deliberately-no-work, distinguishable in ops from a success that did
	// work (the DAG runtime maps Skip to it).
	OutcomeSkip
)

type Settings

type Settings struct {
	// LeaseTTL is how long a claim is held; it also paces lease renewal
	// (LeaseTTL/2) and the reaper (one sweep per LeaseTTL).
	LeaseTTL time.Duration
	// ShutdownDrain is how long Start waits for in-flight handlers after ctx
	// ends before cancelling them.
	ShutdownDrain time.Duration
	// MaxConcurrency caps concurrent handlers across every kind.
	MaxConcurrency int
	// FetchBatchSize caps how many jobs one dequeue leases.
	FetchBatchSize int
	// FetchPollInterval is the minimum polling period while idle.
	FetchPollInterval time.Duration
	// FetchCooldown is the pause after a productive fetch, and the idle backoff
	// floor.
	FetchCooldown time.Duration
	// IdleBackoffMax caps the exponential idle backoff of a fetch loop.
	IdleBackoffMax time.Duration
	// MaxReaps is how many lease expirations a job survives before the reaper
	// kills it.
	MaxReaps int
	// StatsRetention bounds the daily stat counters; 0 retains forever.
	StatsRetention time.Duration
	// CompletedRetention bounds succeeded-job history; 0 retains forever.
	CompletedRetention time.Duration
	// DeadRetention bounds dead-job history; 0 retains forever (the default:
	// dead jobs are never automatically removed unless an operator opts in).
	DeadRetention time.Duration
	// PromoteInterval overrides the scheduled->pending promotion cadence.
	// Zero means the production default (1s).
	PromoteInterval time.Duration
	// VacuumInterval overrides the vacuum cadence. Zero means the production
	// default (1h).
	VacuumInterval time.Duration
}

Settings are the resolved runtime knobs an Engine runs with. The consuming runtime resolves them from the core defaults plus its own overrides.

Jump to

Keyboard shortcuts

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