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

exec

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: 12 Imported by: 0

Documentation

Overview

Package exec defines the execution boundary between the Dispatch worker and a job handler.

Today a handler is an ordinary Go function called in-process, sharing the worker's memory, credentials, and network. Handlers that parse untrusted bytes with memory-unsafe native libraries need more than that, so exec generalises the call into an Executor with implementations forming an escalating ladder: in-process, subprocess, OCI container, and Kubernetes Job-per-task.

exec is a leaf package. It imports only id and artifact within this module — never job, worker, engine, scope, or the root dispatch package — so that job.Options can carry an execution Policy without an import cycle. This mirrors how artifact is itself positioned for input declarations. See deps_test.go for the guard that keeps this list accurate.

Index

Examples

Constants

View Source
const DefaultGracePeriod = 30 * time.Second

DefaultGracePeriod is how long a sandbox is given to exit after being asked politely, before it is killed outright.

Variables

View Source
var (
	// ErrHandler marks an error the handler itself returned.
	ErrHandler = errors.New("handler error")
	// ErrTimeout marks an attempt killed for exceeding its deadline.
	ErrTimeout = errors.New("execution timeout")
	// ErrOOMKilled marks an attempt killed for exceeding a memory limit.
	ErrOOMKilled = errors.New("out of memory")
	// ErrKilled marks an attempt whose process died on a signal.
	ErrKilled = errors.New("killed by signal")
	// ErrLaunchFailed marks a sandbox that never started.
	ErrLaunchFailed = errors.New("launch failed")
)

Status sentinels, so callers can classify a failure with errors.Is rather than by comparing strings.

View Source
var ErrInvalidRequest = errors.New("invalid execution request")

ErrInvalidRequest marks a Request that cannot be executed as given.

View Source
var ErrNoExecutor = errors.New("no executor satisfies the policy")

ErrNoExecutor marks a policy no configured executor can satisfy.

Functions

func Fingerprint

func Fingerprint(names []string) string

Fingerprint derives the identifier for a handler set using this binary's VCS revision. When the revision is unavailable — a build without VCS stamping — it falls back to the empty revision, so the fingerprint still covers the handler names.

func FingerprintOf

func FingerprintOf(names []string, revision string) string

FingerprintOf derives a stable identifier for a handler set and the build that contains it.

A sandbox verifies this before running anything. When the sandbox re-execs the worker's own binary the check always passes and costs one comparison; its purpose is the Policy.Image override, where a stale image would otherwise run an old handler and report success. Drift becomes an immediate, correctly-classified launch failure instead of a silent wrong answer.

Types

type Error

type Error struct {
	Status   Status
	Msg      string
	ExitCode int
	Signal   int

	// Permanent means the job must not be retried, however the attempt was
	// carried. The worker treats it exactly as it treats an error wrapping
	// dispatch.ErrPermanent.
	Permanent bool

	// Cause is the handler's own error when the attempt ran in this
	// process. It is what errors.Is and errors.As reach through Unwrap.
	Cause error
}

Error is a failed execution attempt. It carries the Status so retry policy can branch on how the attempt failed rather than parsing text.

func (*Error) Error

func (e *Error) Error() string

Error implements the error interface.

When a cause survived — an in-process attempt — the handler's own text is returned verbatim. The job's LastError, its DLQ entry, and the worker's logs then read exactly as they did before execution isolation existed, and the exec framing is not stacked on top of an error the caller already understands. Only a failure with no cause to speak for it, which is every out-of-process shape, is rendered with the status.

func (*Error) Unwrap

func (e *Error) Unwrap() []error

Unwrap returns the status sentinel together with the handler's own error when one survived, so errors.Is reaches ErrHandler and errors.Is/errors.As reach the caller's own sentinels and error types through the same value.

The multi-error form (Go 1.20) is what lets both hold at once. Returning only the sentinel would silently destroy user error identity for every in-process attempt, and dispatch.ErrPermanent with it.

type Executor

type Executor interface {
	// Name identifies the executor in configuration, logs, and metrics.
	Name() string

	// Level reports the isolation this executor actually provides, which
	// is what Registry.Select matches a Policy against.
	Level() Level

	// Run executes one attempt.
	//
	// The returned error is reserved for failures to launch — the handler
	// never ran. A handler that ran and failed is reported through
	// Result.Status, so the caller can tell a business failure from a
	// dead sandbox without inspecting error text.
	Run(ctx context.Context, req *Request) (*Result, error)

	// Reclaim releases sandboxes this worker leaked across a restart. The
	// pool calls it once at startup, for every registered executor, and a
	// failure is logged rather than fatal.
	//
	// A later phase runs the same sweep on the leader's behalf for workers
	// the cluster has declared dead; that caller does not exist yet.
	Reclaim(ctx context.Context, workerID id.WorkerID) error

	// Close releases the executor's own resources. The engine calls it for
	// every registered executor when it stops, after the pool has finished
	// its in-flight attempts.
	Close() error
}

Executor runs one job attempt. Implementations form an escalating ladder of isolation, and every one of them must pass the shared conformance suite in exec/exectest.

type InputSlot

type InputSlot struct {
	Name string
	Path string
}

InputSlot maps a declared input name to its location within InputDir. The path is relative, so the same Request describes the inputs whether the sandbox mounts them at /dispatch/in or reads them where they lie.

type Level

type Level int

Level is the minimum isolation a job definition requires. The levels are ordered, so a deployment offering a stronger level satisfies a definition asking for a weaker one.

const (
	// LevelNone runs the handler in the worker process. This is the
	// default and it provides no isolation of any kind.
	LevelNone Level = iota

	// LevelProcess runs the handler in a separate address space, so an
	// exploited parser cannot read the worker's credentials.
	LevelProcess

	// LevelSandboxed adds mount, network, PID, and user namespaces, a
	// seccomp filter, and dropped capabilities.
	LevelSandboxed

	// LevelVM adds an independent kernel — gVisor or Kata — so a Linux
	// privilege escalation is not by itself an escape.
	LevelVM
)

func (Level) String

func (l Level) String() string

String renders the level for configuration, logs, and errors.

type OutputFile

type OutputFile struct {
	Name        string
	Size        int64
	Hash        string
	ContentType string
}

OutputFile describes one artifact the handler produced, as claimed by the sandbox. The worker never reads this to decide what to commit: it is driven entirely by collectOutputEntries walking req.OutputDir on disk (worker/runner.go's commitOutputs), so a claim listed here that does not correspond to a real file on disk is simply never consulted, not checked and rejected. This field exists for whatever future consumer wants the sandbox's own account of what it wrote — logging, a mismatch warning — not as an input to the commit decision.

type Policy

type Policy struct {
	// Level is the minimum isolation required.
	Level Level

	// GracePeriod is how long the sandbox has to exit after SIGTERM
	// before it is killed.
	GracePeriod time.Duration

	// AllowDowngrade permits running at a weaker level than Level when
	// the deployment cannot provide it. Without it, a deployment that
	// cannot satisfy the policy fails at registration rather than
	// silently running the handler unisolated.
	AllowDowngrade bool

	// Image overrides the container image for out-of-process rungs.
	// Empty means the worker's own image, which is the correct default
	// because the sandbox re-execs the same binary.
	Image string
}

Policy is a job definition's execution declaration. It states the minimum isolation the handler requires, not the executor it runs on: which rung satisfies the requirement is a deployment decision.

func NewPolicy

func NewPolicy(opts ...PolicyOption) Policy

NewPolicy builds a Policy from options, starting from the defaults: no isolation and a 30-second grace period.

type PolicyOption

type PolicyOption func(*Policy)

PolicyOption configures a Policy.

func AllowDowngrade

func AllowDowngrade() PolicyOption

AllowDowngrade permits running below the declared level when the deployment cannot satisfy it.

func GracePeriod

func GracePeriod(d time.Duration) PolicyOption

GracePeriod sets how long the sandbox has to exit cleanly after being signalled. Non-positive durations are ignored, because a zero grace period reduces the kill ladder to an immediate SIGKILL and loses any chance of a clean shutdown.

func Image

func Image(ref string) PolicyOption

Image overrides the container image used by out-of-process rungs.

func Isolate

func Isolate(l Level) PolicyOption

Isolate sets the minimum isolation level the handler requires.

type PriorOutput

type PriorOutput struct {
	Name string
	Ref  artifact.Ref
}

PriorOutput is an artifact an earlier attempt of this job committed.

A sandbox keeps its artifact rows in memory and cannot query the store, so without these Accessor.Existing would always answer "no" and a retried handler would redo work it had already finished. The output would still be correct, which is exactly why this is worth carrying explicitly: nothing would fail, it would just quietly cost twice.

type Registry

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

Registry holds the executors a deployment has configured and matches job policies against them.

It is safe for concurrent use, though in practice it is built once at startup and only read afterwards.

func NewRegistry

func NewRegistry(def Executor) *Registry

NewRegistry creates a registry with a default executor, which is the one used by any job that declares no isolation requirement.

func (*Registry) Add

func (r *Registry) Add(e Executor)

Add registers an executor, replacing any existing one with the same name.

func (*Registry) Default

func (r *Registry) Default() Executor

Default returns the executor used when a job declares no requirement.

func (*Registry) Executors

func (r *Registry) Executors() []Executor

Executors returns every registered executor, ordered by name so callers and tests see a stable list.

func (*Registry) Select

func (r *Registry) Select(p Policy) (Executor, error)

Select returns the executor that should run a job with this policy.

It picks the weakest executor that still satisfies the declared level, so a job needing a separate process is not handed a Kubernetes pod merely because one is configured. When nothing satisfies the policy the call fails rather than quietly running the handler with less isolation than it asked for — unless the policy opted into a downgrade.

Example

ExampleRegistry_Select shows how a definition's declared isolation chooses the executor that runs it.

package main

import (
	"context"
	"fmt"

	"github.com/xraph/dispatch/exec"
	"github.com/xraph/dispatch/exec/inproc"
	"github.com/xraph/dispatch/job"
)

type modelInput struct {
	Detail int `json:"detail"`
}

// ExampleRegistry_Select shows how a definition's declared isolation
// chooses the executor that runs it.
func main() {
	registry := job.NewRegistry()

	// A handler that parses untrusted geometry declares that it needs a
	// separate address space at minimum.
	job.NewDefinition("tessellate.model",
		func(context.Context, modelInput) error { return nil },
		job.WithExecution(exec.Isolate(exec.LevelProcess)),
	).Register(registry)

	executors := exec.NewRegistry(inproc.New(registry))

	_, err := executors.Select(registry.Policy("tessellate.model"))
	fmt.Println(err != nil)

	// A handler that declares nothing runs in-process, as it always has.
	e, err := executors.Select(registry.Policy("send.email"))
	fmt.Println(e.Name(), err)

}
Output:
true
inprocess <nil>

type Request

type Request struct {
	JobID   id.JobID
	Name    string
	Payload []byte
	Attempt int

	// Deadline is when the attempt must be killed. Zero means no deadline.
	Deadline time.Time

	// Fingerprint identifies the handler set the caller expects.
	Fingerprint string

	// InputDir holds staged inputs and is read-only to the handler.
	InputDir string
	// OutputDir is where the handler writes artifacts.
	OutputDir string

	Inputs       []InputSlot
	PriorOutputs []PriorOutput

	Policy Policy

	// ResourceLimits is the job's resolved enforcement ceiling
	// (job.Job.ResourceLimits, see job.WithResourceLimits), carried
	// across the execution boundary so a rung that can enforce something
	// has the per-job numbers to enforce it with. A key absent or zero
	// here means the job declared no ceiling for that dimension; it is
	// not this type's business to say what a rung does about that —
	// exec/subprocess.Executor falls back to its own deployment-wide
	// default in that case, but exec itself has no opinion.
	ResourceLimits resource.Set

	// ScopeAppID and ScopeOrgID label the attempt for logs and metrics.
	// They are identifiers, never credentials.
	ScopeAppID string
	ScopeOrgID string

	// Env is passed to out-of-process rungs. It is not the worker's
	// os.Environ() handed through: exec/subprocess.Executor.buildEnv
	// builds the child's environment from this map plus its own
	// configured base, never starting from the worker's full environment.
	// It does still copy a fixed allowlist of PATH, HOME, and TMPDIR from
	// the worker's own environment ahead of this map — HOME in
	// particular is what locates ~/.aws and similar credential paths, so
	// the dedicated uid this rung requires, not environment exclusion
	// alone, is what actually keeps the child from reading them.
	Env map[string]string
}

Request is one execution attempt, fully described. Everything the handler needs crosses the boundary in this value. The environment is not inherited wholesale from the worker: an out-of-process rung constructs the child's environment from Env plus a small fixed allowlist (PATH, HOME, TMPDIR) copied from the worker's own — see Env below and exec/subprocess.Executor.buildEnv.

func (*Request) InputPath

func (r *Request) InputPath(name string) string

InputPath returns the relative path of a declared input, or an empty string when the request carries no such input.

func (*Request) Validate

func (r *Request) Validate() error

Validate reports whether the request is well formed.

type Result

type Result struct {
	// Status classifies the outcome.
	Status Status

	// HandlerErr is the handler's error string, or a diagnostic for a
	// launch failure. Empty on success.
	HandlerErr string

	// ExitCode is the sandbox process's exit status, where one applies.
	ExitCode int

	// Signal is the signal number that killed the process, or zero.
	// Stored as an int rather than a syscall.Signal so this leaf package
	// stays free of syscall.
	Signal int

	// Usage records what the attempt consumed.
	Usage Usage

	// Outputs lists the artifacts the sandbox claims to have written.
	Outputs []OutputFile

	// Permanent means retrying cannot change the outcome, so the job skips
	// its remaining attempts. This is the wire-visible half of permanence:
	// a rung that ran the handler in another process cannot send back a Go
	// error chain, but it can send back this flag.
	Permanent bool

	// Cause is the handler's own error, kept whole. It exists so an
	// in-process attempt loses nothing: errors.Is and errors.As against a
	// caller's sentinels keep working through the exec layer.
	//
	// It is deliberately not serialised — an error chain does not survive a
	// process boundary — which is why Permanent exists alongside it. A rung
	// that marshals a Result leaves this nil and sets Permanent instead.
	Cause error `json:"-"`
}

Result reports how one execution attempt ended.

func (*Result) Err

func (r *Result) Err() error

Err converts a Result into the error the worker propagates. It returns nil for StatusOK and an *Error otherwise.

type Status

type Status string

Status classifies how an execution attempt ended.

A bare error cannot express this. In-process, a handler returning an error and a handler dying are the same value; out-of-process they are different events needing different handling, and only some of them are the handler's fault.

const (
	// StatusOK means the handler ran and returned nil.
	StatusOK Status = "ok"

	// StatusHandlerError means the handler ran and returned an error.
	// This is a business failure and follows the normal retry path.
	StatusHandlerError Status = "handler_error"

	// StatusTimeout means the deadline expired and the sandbox was
	// killed. Unlike a cancelled context, this is enforced.
	StatusTimeout Status = "timeout"

	// StatusOOMKilled means a memory limit was hit. The handler did not
	// choose this and may succeed with a larger allocation.
	StatusOOMKilled Status = "oom_killed"

	// StatusKilled means the process died on a signal — a SIGSEGV from a
	// memory-unsafe parser, or a seccomp trap. It is security-relevant.
	StatusKilled Status = "killed"

	// StatusLaunchFailed means the sandbox never started: an image pull
	// failure, an exhausted quota, a missing runtime. The handler never
	// ran, so this is infrastructure rather than work.
	StatusLaunchFailed Status = "launch_failed"
)

func (Status) CountsAgainstRetries

func (s Status) CountsAgainstRetries() bool

CountsAgainstRetries reports whether an attempt ending in this status should consume the job's retry budget.

Launch failures do not. An ImagePullBackOff or a FailedScheduling says nothing about the work, and burning three retries on one bad node would send healthy jobs to the DLQ.

func (Status) IsFailure

func (s Status) IsFailure() bool

IsFailure reports whether the status represents anything other than success.

type Usage

type Usage struct {
	WallTime    time.Duration
	CPUTime     time.Duration
	PeakRSS     int64
	DiskWritten int64
}

Usage records what an attempt consumed. Every rung above in-process accounts these anyway, so collecting them costs nothing and gives the resource model its measurements.

Directories

Path Synopsis
Package exectest is the conformance suite every exec.Executor must pass.
Package exectest is the conformance suite every exec.Executor must pass.
Package inproc runs job handlers in the worker process.
Package inproc runs job handlers in the worker process.
Package shim is the child side of out-of-process execution.
Package shim is the child side of out-of-process execution.
Package subprocess runs job handlers in a re-exec'd child process.
Package subprocess runs job handlers in a re-exec'd child process.
Package wire carries execution requests and results across a process boundary.
Package wire carries execution requests and results across a process boundary.

Jump to

Keyboard shortcuts

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