Help us shape the future of Go! Take the 2026 Go Developer Survey

engine

package
v1.9.3 Latest Latest
Warning

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

 Go to latest Published: Oct 26, 2026 License: MIT Imports: 40 Imported by: 3

Documentation

Overview

Package engine wires all Dispatch subsystems together and provides the primary application-level API for registering and enqueuing work.

The engine package exists to break a fundamental import cycle: the root dispatch package defines Entity (imported by job, workflow, cron, etc.) and therefore cannot import those packages back. Engine sits above all subsystem packages and below the application layer.

Building an Engine 

d, err := dispatch.New(
    dispatch.WithStore(pgStore),
    dispatch.WithConcurrency(20),
)

eng := engine.Build(d,
    engine.WithExtension(myExtension),
    engine.WithMiddleware(middleware.Logging(logger)),
    engine.WithBackoff(backoff.Exponential()),
    engine.WithQueueConfig(queue.Config{
        Name:      "critical",
        RateLimit: 100,
    }),
)

Registering Work 

// Jobs
engine.Register(eng, SendEmail)

// Workflows
engine.RegisterWorkflow(eng, ProcessOrder)

// Crons
engine.RegisterCron(ctx, eng, "daily-report", "0 9 * * *", GenerateReport, ReportInput{})

Enqueuing Jobs 

engine.Enqueue(ctx, eng, SendEmail, EmailInput{To: "user@example.com"})

// With options
engine.EnqueueAt(ctx, eng, SendEmail, input, time.Now().Add(5*time.Minute))
engine.EnqueueWithPriority(ctx, eng, SendEmail, input, 10)

Options

Package engine wires all Dispatch subsystems together. It creates the extension registry, job registry, middleware chain, worker pool, and provides Register/Enqueue operations.

This package exists to break the import cycle: the root dispatch package defines Entity (imported by job, workflow, etc.) and so cannot import those packages back. The engine package sits above all subsystem packages and below the application layer.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrDurableUnsupported = errors.New("dispatch: store does not implement durable.Store")
	ErrDurableDisabled    = errors.New("dispatch: durable workflows are not enabled")
)

Durable configuration errors prevent fallback to checkpoint execution.

Functions

func Bind 

func Bind(name string, ref artifact.Ref) job.Option

Bind attaches an artifact to a declared input at enqueue.

Bindings are validated against the job's declarations before the job is persisted, so an oversized or undeclared input is a caller error rather than a job that fails on a worker.

func Enqueue 

func Enqueue[T any](ctx context.Context, eng *Engine, name string, payload T, opts ...job.Option) (*job.Job, error)

Enqueue creates and enqueues a job.

func Register 

func Register[T any](eng *Engine, def *job.Definition[T])

Register registers a typed job definition with the engine.

Use RegisterChecked when the definition declares artifact inputs, or an execution policy (job.WithExecution), and you want either validated — the staging budget for the former, and for the latter, that some configured executor can actually satisfy it. Register itself performs neither check: a definition declaring exec.Isolate(exec.LevelProcess) with no rung configured to provide it registers cleanly here and only fails the first time a worker actually tries to run it, which is exactly the startup-error-versus-silently-hung-job choice RegisterChecked exists to take out of a caller's hands.

func RegisterAll 

func RegisterAll(eng *Engine, defs ...job.Registrable) error

RegisterAll registers a set of definitions.

It takes job.Registrable rather than a typed definition so a single handler list can be shared between the worker and an out-of-process entrypoint, which cannot be handed an engine.

func RegisterChecked 

func RegisterChecked[T any](eng *Engine, def *job.Definition[T]) error

RegisterChecked registers a definition and validates its artifact declarations and execution policy, so a job that could never be staged or could never be isolated as it requires fails here — at registration, on a developer's machine — rather than on every worker that picks it up.

func RegisterCron 

func RegisterCron[T any](ctx context.Context, eng *Engine, def *cron.Definition[T]) error

RegisterCron registers a typed cron definition with the engine. It validates the schedule expression, computes the initial NextRunAt, and persists the entry. Re-registration of the same name is idempotent.

func RegisterWorkflow 

func RegisterWorkflow[T any](eng *Engine, def *workflow.Definition[T])

RegisterWorkflow registers a typed workflow definition with the engine.

func StartWorkflow 

func StartWorkflow[T any](ctx context.Context, eng *Engine, name string, input T) (*workflow.Run, error)

StartWorkflow starts a workflow run with a typed input.

Types

type Engine 

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

Engine wraps a Dispatcher with typed subsystem access. Use Build() to create one from a Dispatcher.

func Build 

func Build(d *dispatch.Dispatcher, opts ...Option) (*Engine, error)

Build creates an Engine from an existing Dispatcher. The Dispatcher's store must implement job.Store.

func (*Engine) ArtifactCache 

func (eng *Engine) ArtifactCache() *cache.Cache

ArtifactCache returns the staging cache, or nil when the plane is off.

func (*Engine) Artifacts 

func (eng *Engine) Artifacts() *artifact.Service

Artifacts returns the artifact service, or nil when the plane is off.

func (*Engine) BeginDurableDrain 

func (eng *Engine) BeginDurableDrain(ctx context.Context, request drt.DrainRequest) (drt.DrainHandle, error)

BeginDurableDrain closes polling with an explicit operation-owned deadline. Publisher delivery, authorized queries and storage remain available.

func (*Engine) CancelJob 

func (eng *Engine) CancelJob(ctx context.Context, jobID id.JobID) (*job.Job, error)

CancelJob moves a pending, retrying or running job to cancelled.

A pending or retrying job is cancelled on the spot and JobCancelled fires before CancelJob returns. A running job is written cancelled while its worker still holds it, and the worker finds out through the lease: its next RenewLease, or its terminal write, comes back job.ErrLeaseLost because the row is no longer running. The pool then cancels the handler's context and the runner reports JobCancelled instead of JobFailed once it reads the row and sees why. Any other state wraps dispatch.ErrInvalidState.

The running write is fenced on the holder and epoch read here, through job.LeaseStore.UpdateLeasedJob, rather than a plain UpdateJob. A plain write landing just after the worker completed the job would overwrite completed with cancelled. The fenced one is refused instead, and CancelJob reads the row again and decides on what it finds. A store without the lease capability, or a running job claimed without a lease, falls back to UpdateJob: there is no fence to write through.

The returned job is the row as written. One OperatorAction of kind ext.ActionJobCancelled is emitted on success.

func (*Engine) ClusterStore 

func (eng *Engine) ClusterStore() cluster.Store

ClusterStore returns the cluster store for worker management.

func (*Engine) CompleteDurableActivity 

func (eng *Engine) CompleteDurableActivity(ctx context.Context, request drt.AsyncCompletionRequest) (durable.Receipt, error)

CompleteDurableActivity publishes an asynchronous result or failure. Authorize callers before exposing this trusted Go API through a remote endpoint.

func (*Engine) CountDLQPurge 

func (eng *Engine) CountDLQPurge(ctx context.Context, before time.Time) (int64, error)

CountDLQPurge is PurgeDLQ's dry run: how many entries a purge with the same before would delete, counted on the same boundary. It changes nothing and emits nothing. A zero before is refused, as PurgeDLQ refuses it.

func (*Engine) CronStore 

func (eng *Engine) CronStore() cron.Store

CronStore returns the cron store.

func (*Engine) DLQService 

func (eng *Engine) DLQService() *dlq.Service

DLQService returns the engine's DLQ service for replay and inspection.

func (*Engine) DeleteCron 

func (eng *Engine) DeleteCron(ctx context.Context, cronID id.CronID) error

DeleteCron removes a cron entry. dispatch.ErrCronNotFound when it does not exist.

func (*Engine) DeleteDLQ 

func (eng *Engine) DeleteDLQ(ctx context.Context, entryID id.DLQID) error

DeleteDLQ removes one dead letter entry for good. Unknown entries return dispatch.ErrDLQNotFound. One OperatorAction of kind ext.ActionDLQDeleted is emitted on success.

func (*Engine) DisableCron 

func (eng *Engine) DisableCron(ctx context.Context, cronID id.CronID) (*cron.Entry, error)

DisableCron turns a cron entry off and returns it as stored. Its next_run_at is left as it was. dispatch.ErrCronNotFound when the entry does not exist.

func (*Engine) Dispatcher 

func (eng *Engine) Dispatcher() *dispatch.Dispatcher

Dispatcher returns the underlying Dispatcher.

func (*Engine) DurableReadiness 

func (eng *Engine) DurableReadiness(ctx context.Context) (drt.WorkerReadiness, error)

DurableReadiness reports polling and persisted deployment compatibility.

func (*Engine) DurableStatus 

func (eng *Engine) DurableStatus() (drt.WorkerStatus, error)

DurableStatus reports worker readiness independently of Health's database ping.

func (*Engine) DurableWorker 

func (eng *Engine) DurableWorker() *drt.Worker

DurableWorker returns the configured runtime, or nil when disabled. Engine owns its polling lifecycle.

func (*Engine) EnableCron 

func (eng *Engine) EnableCron(ctx context.Context, cronID id.CronID) (*cron.Entry, error)

EnableCron turns a cron entry on and returns it as stored.

The next fire time is computed from now, so an entry that was off past its old next_run_at does not fire a catch-up the moment it comes back. For an @every schedule that restarts the interval, and enabling an entry that is already on does the same. A schedule with no fire time in the next five years (30 February) is refused with a wrapped dispatch.ErrInvalidState, since the scheduler would have nothing to wait for. dispatch.ErrCronNotFound when the entry does not exist.

func (*Engine) EnqueueRaw 

func (eng *Engine) EnqueueRaw(ctx context.Context, name string, payload []byte, opts ...job.Option) (*job.Job, error)

EnqueueRaw enqueues a job with a pre-serialized payload.

func (*Engine) EventBus 

func (eng *Engine) EventBus() *event.Bus

EventBus returns the event bus.

func (*Engine) Executors 

func (eng *Engine) Executors() *exec.Registry

Executors returns the configured executor registry.

func (*Engine) Extensions 

func (eng *Engine) Extensions() *ext.Registry

Extensions returns the extension registry.

func (*Engine) Health 

func (eng *Engine) Health(ctx context.Context) error

Health checks the health of the engine by pinging the dispatcher's store. It also fails while the durable worker has stopped or its claims go unanswered, since durable work isn't being picked up then.

func (*Engine) HeartbeatDurableActivity 

func (eng *Engine) HeartbeatDurableActivity(ctx context.Context, request drt.AsyncHeartbeatRequest) (durable.Receipt, error)

HeartbeatDurableActivity persists progress for a deferred activity. The caller supplies a stable request ID and the next consecutive heartbeat sequence.

func (*Engine) Inspect 

func (eng *Engine) Inspect() Inspection

Inspect returns detached configuration and effective subsystem settings.

func (*Engine) MaxWorkerCapacity 

func (eng *Engine) MaxWorkerCapacity(ctx context.Context) resource.Set

MaxWorkerCapacity returns the per-key maximum capacity the unschedulable check may compare a job against, or an empty Set when the check is off.

An empty result disables the check rather than rejecting everything.

It is empty unless an operator called WithWorkerCapacity, and that gate is the whole correctness argument. The registry cannot supply the fleet maximum on its own: a worker's row carries only the capacity that worker declared through WithWorkerCapacity (and the k8s provider does not persist it at all), so a fleet whose largest worker has 64 GiB but declared nothing reads back as a fleet of workers with no capacity. Deriving the ceiling from whatever this process happens to know therefore does not converge on the truth; it converges on THIS process, and rejects at enqueue every job bigger than the pod that enqueued it. Requiring the declaration also keeps the common path free of a ListWorkers round trip per enqueue.

When the declaration is present the registry is still consulted, so the ceiling can only rise toward the real fleet maximum on a backend that carries capacity. "Live" means both an active state and a recent heartbeat. State alone is not enough: nothing in Dispatch ever writes WorkerDead — a worker registers active and is either deregistered on clean shutdown or its row is deleted by DeleteStaleWorkers. A worker killed by SIGKILL, an OOM or a pod eviction therefore stays "active" in the registry until something sweeps it, and counting its capacity would admit jobs no live worker can run.

func (*Engine) PlanWorkflowReplay 

func (eng *Engine) PlanWorkflowReplay(ctx context.Context, runID id.RunID, fromStep string) (*workflow.ReplayPlan, error)

PlanWorkflowReplay reports what ReplayWorkflowFrom would do for the run and step, without changing anything. See workflow.Runner.PlanReplay for the refusals; a running run plans fine and its State says a replay would be refused.

func (*Engine) PurgeDLQ 

func (eng *Engine) PurgeDLQ(ctx context.Context, before time.Time) (int64, error)

PurgeDLQ deletes every dead letter entry that failed strictly before before, and returns how many went. A zero before is refused: it matches nothing, so it can only be a caller that forgot to set it. One OperatorAction of kind ext.ActionDLQPurged, with Count set, is emitted on success, including when nothing matched. A store that reports partial progress with an error retains that count and emits it before returning.

func (*Engine) QueryDurableWorkflow 

func (eng *Engine) QueryDurableWorkflow(ctx context.Context, request drt.QueryRequest) (drt.QueryResult, error)

QueryDurableWorkflow selects and reconstructs one run without claiming tasks or writing history. Authorize callers before exposing this trusted Go API remotely.

func (*Engine) QueueManager 

func (eng *Engine) QueueManager() *queue.Manager

QueueManager returns the queue manager, or nil if no queue configs were provided.

func (*Engine) Registry 

func (eng *Engine) Registry() *job.Registry

Registry returns the job registry.

func (*Engine) ReplayAllDLQ 

func (eng *Engine) ReplayAllDLQ(ctx context.Context, opts ReplayAllOpts) (ReplayAllResult, error)

ReplayAllDLQ replays every entry nobody has replayed yet, newest first, up to opts.Limit, each one exactly as ReplayDLQ would. It is not a transaction: entries are claimed one at a time, so a concurrent replay of one of them shows up as a Conflict, and a failure partway leaves the earlier replays in place.

It returns an error only when the entries cannot be listed or ctx ends, together with what it had done by then. One OperatorAction records the committed replay count, including when a later read fails. A completed operation with no replays also emits a zero count. An interrupted operation that changed no entries emits nothing.

func (*Engine) ReplayDLQ 

func (eng *Engine) ReplayDLQ(ctx context.Context, entryID id.DLQID) (*job.Job, error)

ReplayDLQ turns a dead letter entry back into work: a new pending job carrying everything the failed one ran with, under a fresh ID.

The entry is claimed before the job is enqueued (dlq.Service.ReplayEntry), so of two concurrent replays exactly one creates a job and the other gets an error wrapping dispatch.ErrDLQAlreadyReplayed, as does a replay of an entry whose job was already retried. A failed enqueue releases the claim. Unknown entries return dispatch.ErrDLQNotFound, and a job too big for every worker wraps resource.ErrUnschedulable.

One OperatorAction of kind ext.ActionDLQReplayed is emitted on success, naming the entry, the failed job and the new one.

func (*Engine) ReplayWorkflowFrom 

func (eng *Engine) ReplayWorkflowFrom(ctx context.Context, runID id.RunID, fromStep string) (*workflow.ReplayPlan, error)

ReplayWorkflowFrom re-runs a finished run from a step on the run's stamped version, keeping the checkpoints up to and including fromStep. It returns once the replay has started on the runner's background launcher; Stop waits for it. Refusals are workflow.Runner.ReplayFrom's: dispatch.ErrRunNotFound, dispatch.ErrInvalidState (running, claimed by another replay, no checkpoint for the step, version not registered), or workflow.ErrRunnerShutdown after Stop.

On success it emits one ext.ActionWorkflowReplayed with the run and step.

func (*Engine) ReplayWorkflowFromGeneration 

func (eng *Engine) ReplayWorkflowFromGeneration(ctx context.Context, runID id.RunID, fromStep string, generation int64) (*workflow.ReplayPlan, error)

ReplayWorkflowFromGeneration starts replay only if the reviewed generation still matches. It emits an operator action only after a successful launch.

func (*Engine) RequestCancelDurableWorkflow 

func (eng *Engine) RequestCancelDurableWorkflow(ctx context.Context, request durable.CancelExecutionRequest) (durable.CancelExecutionReceipt, error)

RequestCancelDurableWorkflow records cancellation acceptance for an explicit or current run. Authorize callers before exposing this trusted API remotely.

func (*Engine) Resources 

func (eng *Engine) Resources() resource.Manager

Resources returns the shared admission ledger, or nil when the resource model is off. It is the instance the staging cache must have been built with.

func (*Engine) RetryJob 

func (eng *Engine) RetryJob(ctx context.Context, jobID id.JobID) (*job.Job, error)

RetryJob puts a failed job back to pending, as if it had just been enqueued: no retries spent, no error, no owner. Any other state wraps dispatch.ErrInvalidState.

A failed job usually has a dead letter entry, and replaying that entry would run the same failure a second time. So when one exists the retry claims it first, with the retried job as the claim's job, and a claim that is already taken refuses the retry with dispatch.ErrDLQAlreadyReplayed. If the job write then fails the claim is released, so the entry can be replayed or retried later. A failed job with no entry retries without a claim.

Without an entry there is nothing to serialise two concurrent retries on. Both write the same pending row, which is harmless unless a worker claims the job between the two writes.

The returned job is the row as written. One OperatorAction of kind ext.ActionJobRetried is emitted on success, carrying the claimed entry's ID when there was one.

func (*Engine) Scheduler 

func (eng *Engine) Scheduler() *cron.Scheduler

Scheduler returns the cron scheduler.

func (*Engine) SignalDurableWorkflow 

func (eng *Engine) SignalDurableWorkflow(ctx context.Context, request durable.SignalRequest) (durable.SignalReceipt, error)

SignalDurableWorkflow accepts a message for an explicit or current open run. You must authorize callers before exposing this trusted Go method remotely.

func (*Engine) SignalWithStartDurableWorkflow 

func (eng *Engine) SignalWithStartDurableWorkflow(ctx context.Context, request durable.SignalWithStartRequest) (durable.SignalReceipt, error)

SignalWithStartDurableWorkflow accepts a message and creates a run when needed. Reuse the whole request to recover its original target after a lost response.

func (*Engine) Start 

func (eng *Engine) Start(ctx context.Context) (startErr error)

Start begins job processing by starting the worker pool and cron scheduler. It also resumes any workflow runs left in "running" state (crash recovery).

func (*Engine) StartDurableWorkflow 

func (eng *Engine) StartDurableWorkflow(ctx context.Context, request durable.StartRequest) (durable.Receipt, error)

StartDurableWorkflow persists the run and its initial task. Calls may precede Engine.Start. The caller supplies stable IDs so ambiguous outcomes are retryable.

func (*Engine) Stop 

func (eng *Engine) Stop(ctx context.Context) error

Stop confirms worker quiescence before attempting publisher drain and storage closure. A timed-out caller leaves one quiescence task running and can retry.

func (*Engine) StopWorkers 

func (eng *Engine) StopWorkers(ctx context.Context) error

StopWorkers permanently stops this engine's execution producers and waits for both worker pools to finish. Publisher delivery, final extension hooks and storage remain active. This is a terminal worker stop, not a resumable pause. Concurrent Stop/StopWorkers calls share the same quiescence task. A caller's timeout never establishes completion and can be retried with a fresh context.

func (*Engine) StreamBroker 

func (eng *Engine) StreamBroker() *stream.Broker

StreamBroker returns the real-time event broker, or nil if not enabled.

func (*Engine) Telemetry 

func (eng *Engine) Telemetry() *telemetry.Telemetry

Telemetry returns the spans and metrics the engine records through, built from WithMetricFactory and WithTracerProvider. The Forge extension hands it to the outbox publisher and artifact sweeper it builds; do the same for any subsystem you assemble beside the engine.

func (*Engine) TriggerCron 

func (eng *Engine) TriggerCron(ctx context.Context, cronID id.CronID) (*job.Job, error)

TriggerCron enqueues the entry's job now, with its payload and queue, the same job a scheduled fire makes. The schedule is untouched: next_run_at and last_run_at stay as they were, so the entry still fires at its next scheduled time. A disabled entry can be triggered too, since running it by hand is exactly what an operator asks for here. Hooks see CronFired with the new job, as for a scheduled fire, and then the operator action. dispatch.ErrCronNotFound when the entry does not exist.

func (*Engine) ValidateArtifactInputs 

func (eng *Engine) ValidateArtifactInputs(name string, specs []artifact.InputSpec) error

ValidateArtifactInputs checks a definition's declarations against the engine's staging capacity.

A definition whose declared inputs could never fit the cache budget is rejected here, at registration, rather than failing every time such a job runs. The point is to surface the mistake on a developer's machine instead of at 3am in production.

func (*Engine) WaitDurableDrain 

func (eng *Engine) WaitDurableDrain(ctx context.Context, handle drt.DrainHandle) (drt.DrainResult, error)

WaitDurableDrain only waits. Its context never shortens an accepted drain.

func (*Engine) WorkerID 

func (eng *Engine) WorkerID() id.WorkerID

WorkerID returns this engine's unique worker identifier.

func (*Engine) WorkflowRunner 

func (eng *Engine) WorkflowRunner() *workflow.Runner

WorkflowRunner returns the workflow runner.

type Inspection 

type Inspection struct {
	Config                                        dispatch.Config
	Pool                                          worker.Settings
	Cron                                          cron.Settings
	ResourceDefaults                              resource.Set
	QueueResources                                map[string]resource.Set
	WorkerCapacity                                resource.Set
	WorkerCustomKeys                              []string
	ResourceManagerEnabled, EstimatorConfigured   bool
	ScratchRoot                                   string
	WakeNotifierSupported                         bool
	WorkerHeartbeatInterval, WorkerStaleThreshold time.Duration
}

Inspection describes the serving process. It makes no fleet-wide health claim.

type Option 

type Option func(*Engine)

Option configures an Engine.

func WithArtifacts 

func WithArtifacts(svc *artifact.Service, c *cache.Cache) Option

WithArtifacts enables the artifact plane.

The staging middleware is appended to the chain, so declared inputs are materialised before a handler runs and every cache lease is released afterwards. Without this option Dispatch behaves exactly as it did before artifacts existed.

func WithBackoff 

func WithBackoff(b backoff.Strategy) Option

WithBackoff sets the retry backoff strategy for the engine. If not set, backoff.DefaultStrategy() (exponential with jitter) is used.

func WithDurableWorkflows 

func WithDurableWorkflows(options drt.Options) Option

WithDurableWorkflows enables the history-based runtime alongside the existing checkpoint workflow API. Its store must explicitly implement durable.Store.

func WithEstimator 

func WithEstimator(e resource.Estimator) Option

WithEstimator installs the resource estimator consulted at enqueue.

The estimator sits above a definition's static declaration and below a per-enqueue override. It receives the declaration in the request and may return it unchanged, so installing one is an explicit opt-in to letting inference override declaration. An estimator that errors is ignored: it must never fail an enqueue.

func WithExecutor 

func WithExecutor(e exec.Executor) Option

WithExecutor registers an additional executor, making a stronger isolation level available to job definitions that ask for it.

The in-process executor is always present as the default, so a deployment that adds nothing behaves exactly as it always has.

func WithExtension 

func WithExtension(e ext.Extension) Option

WithExtension registers an extension with the engine.

func WithMetricFactory 

func WithMetricFactory(factory gu.MetricFactory) Option

WithMetricFactory sets the go-utils MetricFactory for engine-level metrics. Use fapp.Metrics() in forge applications to integrate with the forge metrics system. If not set, a default metrics collector is used for the legacy counters and the labelled dispatch_* metrics (see the telemetry package) are off: the default collector keys series by name alone and would fold their labels.

func WithMiddleware 

func WithMiddleware(m mw.Middleware) Option

WithMiddleware adds middleware to the engine's chain.

func WithQueueConfig 

func WithQueueConfig(configs ...queue.Config) Option

WithQueueConfig registers queue-level rate limiting and concurrency configurations. Queues not listed have no limits.

func WithResourceDefaults 

func WithResourceDefaults(global resource.Set, perQueue map[string]resource.Set) Option

WithResourceDefaults sets the fleet-wide default requirement and any per-queue overrides. Both are the lowest-precedence sources, below a definition's own declaration.

func WithResourceManager 

func WithResourceManager(m resource.Manager) Option

WithResourceManager installs the admission ledger the worker pool admits jobs against.

The manager passed here MUST be the same instance the staging cache was built with (cache.WithManager). One ledger is the whole design: the cache holds a lease per cached entry and registers itself as the manager's disk reclaimer, and the pool's dequeue budget offers disk as free PLUS what that reclaimer could evict. Give the cache a private manager — which it constructs for itself when none is supplied — and the pool's Reclaimable() is permanently zero, staged bytes are never offered back to the budget, and the disk path quietly does nothing. It presents as a worker that went quiet, not as an error.

Leaving this unset is the supported default: the pool passes an unbounded DequeueOpts, every backend skips its fit predicate, and no leases are taken. That is exactly how Dispatch behaved before the resource model existed.

Installing a manager does NOT declare a fleet capacity, and does not turn the enqueue-time unschedulable check on. See WithWorkerCapacity for why that is opt-in and separate.

Leases compose with this. The grant travels on job.DequeueOpts (WorkerID and LeaseUntil), so a claim that takes a lease is an ordinary claim that also writes the lease columns — it carries the budget, the custom-key containment and the locality preference like any other. There is one dequeue path per backend and it is the one this manager constrains. Turning leases and resources on together is the natural upgrade, and it is a supported one.

func WithScratchRoot 

func WithScratchRoot(dir string) Option

WithScratchRoot sets the root directory an out-of-process attempt's scratch OutputDir is created under (worker.Runner.WithArtifacts).

It only has an effect once the artifact plane is also configured (WithArtifacts): Build only ever passes this value to the Runner alongside eng.artifacts (see Build's own runner.WithArtifacts call), so with no artifact plane configured this sets a value Build never reads and the Runner's scratch directories fall back to os.TempDir() regardless. That is not because scratch-dir creation or its startup sweep are themselves gated on the artifact plane — they are not, and both run for any out-of-process attempt either way (see worker.Runner.prepareOutputDir and Reclaim) — it is purely this option's own value never reaching the Runner. The extension package's resolveExecutionOptions logs a startup warning for exactly this configuration (a configured scratch_dir with the artifact plane off), so that stays a deliberate, visible choice rather than a silent one.

Leaving it unset defaults to os.TempDir(), exactly as worker.Runner does on its own.

func WithStreamBroker 

func WithStreamBroker(opts ...stream.BrokerOption) Option

WithStreamBroker enables the real-time stream broker for event pub/sub. The broker is automatically registered as an extension so it receives all job and workflow lifecycle events. It is required for DWP (Dispatch Wire Protocol) real-time subscriptions.

func WithTracerProvider 

func WithTracerProvider(tp trace.TracerProvider) Option

WithTracerProvider sets a custom OTel TracerProvider for the engine. When set, every span the engine and its subsystems start (job attempts, enqueues, cron fires, workflow runs, durable tasks, the tracing middleware) uses this provider instead of the global one. If not set, the global otel.GetTracerProvider() is used.

func WithWorkerCapacity 

func WithWorkerCapacity(c resource.Set) Option

WithWorkerCapacity declares the largest single-worker capacity in the fleet, and is the ONLY thing that turns the enqueue-time unschedulable check on. Leave it unset — the default — and Enqueue never rejects a job for being too big for any worker.

It is a fleet-wide statement, not a description of this process, and the distinction is the whole reason the check is opt-in. Declare it and a job requiring more than this on any dimension fails Enqueue with ErrUnschedulable, wherever it was enqueued from: a light API pod that declared its own 2 GiB would hard-reject the tessellation job the heavy tier runs perfectly well.

The check cannot derive the fleet maximum for itself. Every store backend now persists cluster.Worker.Capacity (the k8s provider still drops it), but a worker's row carries only what that worker passed here, so a fleet in which nobody declared reads back with no capacity at all. MaxWorkerCapacity therefore starts from this value and lets live workers' published capacities raise it, never lower it. Declaring the ceiling is the operator's job, or the check stays off.

It is deliberately NOT defaulted from WithResourceManager's capacity. That default read as a convenience and behaved as a silent rescope of a fleet-wide question to one process.

func WithWorkerCustomKeys 

func WithWorkerCustomKeys(keys []string) Option

WithWorkerCustomKeys narrows the custom resource keys this worker advertises at dequeue.

The default — every custom key the manager has capacity for — is usually right. This exists to shrink it, so a worker draining a device can stop attracting work for it without being reconfigured. Keep the list a subset of the manager's custom capacity: dequeue matches custom keys by containment and never by quantity, so a key advertised here with no capacity behind it passes the store's filter and is then refused locally, on every attempt.

type ReplayAllOpts 

type ReplayAllOpts struct {
	// Queue limits the replay to one queue. Empty means every queue.
	Queue string
	// Limit caps how many entries are tried. Zero or less means 1000.
	Limit int
}

ReplayAllOpts selects what ReplayAllDLQ replays.

type ReplayAllResult 

type ReplayAllResult struct {
	// Replayed is how many entries became new jobs.
	Replayed int
	// Conflicts is how many were claimed by somebody else between the
	// listing and their own replay.
	Conflicts int
	// Failed is how many could not be replayed for any other reason.
	// Each of those entries was released and can be replayed again.
	Failed int
	// Errors holds a message for the first failures, "<entry id>: <error>",
	// at most 20 of them.
	Errors []string
}

ReplayAllResult counts what ReplayAllDLQ did with each entry it tried.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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