workflow

package
v0.3.0 Latest Latest
Warning

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

Go to latest
Published: Oct 5, 2026 License: Apache-2.0 Imports: 26 Imported by: 0

Documentation

Overview

Package workflow implements the funcd workflow engine core (ADR-0094, FEAT-0005/F64): the state-machine orchestrator that drives a WorkflowRun over its steps. This file is the pure scheduling core — no I/O — so the semantics (implicit chaining, fan-out, join all/any, skip cascade, fail-fast, run-phase derivation) are unit-testable without Badger, dispatch, or containerd.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Permanent

func Permanent(err error) error

Permanent wraps err as a non-retryable dispatch failure.

Types

type ChildResolver

type ChildResolver interface {
	Child(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) (v1.WorkflowSpec, map[v1.ObjectName]string, error)
}

ChildResolver resolves a child workflow's pinned spec + its resolved step-image cache for a `workflow:` step's EXECUTION (store-backed in prod; a fake in tests). Reconcile-time typing does NOT use this — the reconciler reads the child's status.contract from the store it already has (ADR-0099). The step-image map (ADR-0107, the child's ADR-0098 `status.steps[].Image`) digest-pins the inline child run's record, so a standalone replay of that child run gates on real digests instead of false-positiving.

type ChildWorkflowResolver added in v0.2.3

type ChildWorkflowResolver interface {
	ChildWorkflow(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) (*v1.Workflow, error)
}

ChildWorkflowResolver is an optional ChildResolver extension, additive to ADR-0107's Child: a resolver that implements it returns the child Workflow itself, so the inline child run pins the per-step contracts of its ADR-0098 cache with its step images, from one read, and a when: binds a schema default there (ADR-0095).

type Config

type Config struct {
	DefaultMaxAttempts  int           // per-step, when a step sets no retry (default 1 = no retry)
	DefaultStepTimeout  time.Duration // per-step invocation bound (0 = none)
	PayloadLimit        int64         // max bytes for a run input (at admission and run start) and a step output; 0 = unbounded
	MaxSubworkflowDepth int           // ADR-0099: max sub-workflow nesting (default 8); a deeper chain fails cleanly
}

Config holds the engine's tunables (ADR-0094 workflow.* keys).

type ContractResolver

type ContractResolver interface {
	Contract(ctx context.Context, image string) (contract v1.WorkflowContract, digest string, err error)
}

ContractResolver reads a step image's I/O contract from OCI metadata (ADR-0059/0098), mirroring RuntimeResolver. Production wraps artifact.InspectContract; tests inject a fake. NotFound ⇒ not pushed.

type Deps

type Deps struct {
	Runs     runstate.Store // durable run state (the port; Badger driver in prod, in-memory in tests)
	Dispatch Dispatcher     // the step-invocation seam
	Config   Config
	Clock    clock.Clock   // stamps run timestamps (retention GC input); defaults to the system clock
	Children ChildResolver // ADR-0099: resolves a child workflow's spec for a `workflow:` step (nil ⇒ rejected)
	// Traces is the shared funclog trace sink (ADR-0104): the engine emits the run-root span for INLINE
	// sub-workflow child runs (the reconciler drives only top-level runs). nil ⇒ no child run-root span.
	Traces funclog.TraceSink
	Logger *slog.Logger
}

Deps wires the engine (internal component, ADR-0002 §1).

type DispatchDeps

type DispatchDeps struct {
	Endpoints activator.Endpoints
	Waker     Waker
	Grant     Granter
	Client    *http.Client
	Logger    *slog.Logger
}

DispatchDeps wires the HTTP dispatcher.

type DispatchRequest

type DispatchRequest struct {
	Namespace v1.NamespaceName
	Run       v1.ObjectName
	Step      v1.ObjectName
	Target    v1.ObjectName // the function to invoke (materialized name or referenced)
	Attempt   int
	Input     json.RawMessage
	// MaxOutput is the payload limit on the step's output (ADR-0094); 0 ⇒ unbounded. A dispatcher need
	// read no more than one byte past it, since a longer output is rejected anyway.
	MaxOutput int64
	// TraceID / ParentSpanID are the run's W3C trace context (ADR-0102): the dispatcher sets a
	// traceparent header from them so the step-function invocation's span joins the run's trace.
	// Empty TraceID ⇒ no header (additive). ParentSpanID is the step's PRIMARY predecessor (ADR-0105;
	// the run root for a true root step) — the parent edge the step nests under.
	TraceID      string // 32 hex
	ParentSpanID string // 16 hex
	// SpanID is the engine-minted span-id the step function uses as its own (ADR-0105, X-Funcd-Span-Id) so a
	// successor can parent on it. Links are the non-primary fan-in predecessors' span-ids (X-Funcd-Span-Links).
	SpanID string   // 16 hex; "" ⇒ the shim mints its own (direct invoke / additive)
	Links  []string // 16-hex span-ids, same trace
}

DispatchRequest is one step invocation. Attempt is the 1-based attempt number (surfaced to the function as the idempotency key, at-least-once).

type Dispatcher

type Dispatcher interface {
	Dispatch(ctx context.Context, req DispatchRequest) (json.RawMessage, error)
}

Dispatcher is the step-invocation seam (ADR-0094): it delivers a step's input to its function and returns the output. Production wraps activator wake + HTTP; tests inject a fake. A returned error is a step failure; a Permanent error is not retried.

type Engine

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

Engine executes workflow runs against durable state and the dispatcher.

func New

func New(d Deps) (*Engine, error)

New builds the engine.

func (*Engine) Cancel

func (e *Engine) Cancel(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) error

Cancel abandons a run: pending/running steps are marked Cancelled and the run ends Cancelled immediately (the in-flight invocation is abandoned; idempotency covers it). A run that is already terminal is left unchanged (WorkflowRunSpec.Cancel).

func (*Engine) Execute

func (e *Engine) Execute(ctx context.Context, ns v1.NamespaceName, runName, workflow v1.ObjectName, spec v1.WorkflowSpec, input json.RawMessage, opts StartOptions) (*runstate.Record, error)

Execute runs a workflow synchronously to a terminal phase and returns the final record. It is the engine core; the controller reconciler drives it asynchronously (wiring is a separate layer). Every ready step is dispatched at once (ADR-0094: fan-out is parallel dispatch).

func (*Engine) Pause

func (e *Engine) Pause(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) error

Pause requests a graceful pause: the persisted run is marked Paused so the next drive dispatches nothing new (in-flight steps, in the async model, finish first). A run that is already terminal is left unchanged.

func (*Engine) Replay

func (e *Engine) Replay(ctx context.Context, ns v1.NamespaceName, runName, workflow v1.ObjectName, seed v1.ReplaySeed, current map[v1.ObjectName]string) (*runstate.Record, error)

Replay seeds runName from a finished source run's checkpoint and drives it (ADR-0107): a NEW run that re-runs seed.From + its descendants, reusing the source's recorded upstream outputs verbatim. workflow is the replay run's declared workflow (must match the source's). current is the workflow's live resolved step-image map (the ADR-0098 cache) — the drift-gate comparison key + the re-run steps' revision. Faults: NotFound (source absent); Invalid whose message leads with reason token SeedInvalid or DigestDrift (naming the offending step). The source record is never mutated.

func (*Engine) Resume

func (e *Engine) Resume(ctx context.Context, ns v1.NamespaceName, runName v1.ObjectName) (*runstate.Record, error)

Resume continues a persisted run after a crash, pause, or cancel-race (ADR-0094): it rebuilds the scheduling state from the durable record and re-dispatches any step that was in-flight (with a fresh attempt), so no state is lost. It rebuilds from the record's PINNED spec (rec.Spec), never the live Workflow — an in-flight run is immune to a mid-run spec edit or artifact re-push. The caller passes no spec; the pinned one is the truth.

func (*Engine) SweepExpired

func (e *Engine) SweepExpired(ctx context.Context, retention time.Duration, reclaim func(context.Context, *runstate.Record) error) (int, error)

SweepExpired deletes terminal run records whose last update is older than retention (ADR-0094: terminal runs "swept after workflow.retention"). retention ≤ 0 disables the sweep. reclaim, when non-nil, runs before each expired record is deleted (the run reconciler deletes the run's WorkflowRun there); its error keeps the record for the next sweep. Returns the number of records reclaimed. Non-terminal runs are never swept. Callers invoke it periodically (pkg/funcd lifecycle); it is idempotent and safe to run concurrently with reconciles (the store is the single writer per run and a terminal run is immutable).

type Granter

type Granter interface {
	Allow(ns v1.NamespaceName, target v1.ObjectName) bool
}

Granter authorizes a step dispatch: only functions declared in a run's pinned spec may be invoked (spec-as-grant, ADR-0094). An undeclared target is Forbidden.

type HTTPDispatcher

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

HTTPDispatcher is the production Dispatcher: it resolves a step function's ready upstream (waking a scaled-to-zero one), POSTs the step input, and returns the response body as the step output. A 4xx is a permanent failure (contract rejection / Forbidden); a 5xx or transport error is retryable. It enforces spec-as-grant fail-closed before dispatching.

func NewHTTPDispatcher

func NewHTTPDispatcher(d DispatchDeps) (*HTTPDispatcher, error)

NewHTTPDispatcher builds the production dispatcher.

func (*HTTPDispatcher) Dispatch

Dispatch resolves, wakes, and invokes the step function.

type Materializer

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

Materializer creates and updates a Workflow's owned step Functions and KVStores (the Deployment→ReplicaSet analogy, ADR-0094). It owns the ADR-0073 apply-order cycle internally: Functions without kv → KVStores → patch Functions with kv.

func NewMaterializer

func NewMaterializer(s store.Store, r RuntimeResolver, log *slog.Logger) *Materializer

NewMaterializer builds the materializer.

func (*Materializer) Materialize

func (m *Materializer) Materialize(ctx context.Context, wf *v1.Workflow) error

Materialize brings the workflow's owned resources to the desired state.

type RunReconciler

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

RunReconciler drives WorkflowRun resources: it runs the engine, mirrors the coarse run state into WorkflowRun.status, and maintains the parent Workflow's status.runs link (ADR-0094). It is a controller.Reconciler.

func NewRunReconciler

func NewRunReconciler(s store.Store, e *Engine, traces funclog.TraceSink, log *slog.Logger) *RunReconciler

NewRunReconciler builds the run reconciler. traces is the shared funclog trace sink (ADR-0103): when non-nil, the reconciler emits one INTERNAL run-root span per terminal run so the run's step spans (F51, parented on the run root) nest under it. nil ⇒ no run-root span (additive).

func (*RunReconciler) Reconcile

Reconcile drives one WorkflowRun toward its terminal phase.

func (*RunReconciler) SweepExpired added in v0.2.1

func (r *RunReconciler) SweepExpired(ctx context.Context, retention time.Duration) (int, error)

SweepExpired reclaims the closed runs older than retention (ADR-0094): each expired engine record and the WorkflowRun object that `workflow runs` lists, whose status is swept with the run (ADR-0100). Returns the number of runs reclaimed.

type RuntimeResolver

type RuntimeResolver interface {
	Runtime(ctx context.Context, image string) (v1.RuntimeName, error)
}

RuntimeResolver resolves a step image's runtime class from its OCI manifest (the dev.funcd.runtime.v1 annotation, ADR-0059 family). Production reads it via internal/artifact.Inspect; tests inject a fake.

type StartOptions

type StartOptions struct {
	Contract   *v1.WorkflowContract     // pinned derived contract; nil ⇒ no run-start input check
	StepImages map[v1.ObjectName]string // step name → resolved digest-pinned image; stamps stepNode.revision (function steps)
	// StepContracts is the ADR-0098 cache's per-step I/O contract, pinned on the record: a when: binds an
	// absent parent-output field to its schema default (ADR-0095). Nil ⇒ no defaults are bound.
	StepContracts map[v1.ObjectName]v1.WorkflowContract
	RunUID        v1.UID // the starting WorkflowRun's uid, stamped on the record; empty for an inline child run
}

StartOptions consolidates run-start inputs (ADR-0107, replacing Execute's variadic contract param): the pinned contract (ADR-0098) + the per-step resolved digest-pinned image map (the ADR-0098 cache), which stamps each function step's revision so a replay can prove it re-runs the same artifact.

type Waker

type Waker interface {
	Wake(ctx context.Context, fn activator.FunctionRef) (upstream string, err error)
}

Waker wakes a scaled-to-zero function and returns its ready upstream (ADR-0033). A structural match for eventing.Waker, declared locally so the workflow package does not import a sibling feature.

type WorkflowReconciler

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

WorkflowReconciler is the controller.Reconciler for the Workflow kind: it brings a Workflow's owned step Functions and KVStores to the desired state (the Deployment→ ReplicaSet analogy, ADR-0094) via the Materializer. Run execution is the RunReconciler's job; this reconciler only maintains the materialized fleet.

func NewWorkflowReconciler

func NewWorkflowReconciler(s store.Store, m *Materializer, contracts ContractResolver, log *slog.Logger) *WorkflowReconciler

NewWorkflowReconciler builds the Workflow reconciler over a Materializer + a ContractResolver (the F65 typed-edge gate, ADR-0098). A nil ContractResolver skips the gate (materialization-only).

func (*WorkflowReconciler) Reconcile

Reconcile materializes one Workflow's owned resources, then runs the F65 typed-edge contract gate: resolve each step's contract from OCI metadata, type-check the edges + when: predicates, derive + cache the graph in status, and set Ready / SchemaMismatch (ADR-0098).

Directories

Path Synopsis
Package runstate is the durable-run-state PORT for the workflow engine (ADR-0094).
Package runstate is the durable-run-state PORT for the workflow engine (ADR-0094).
badger
Package badger is the run-state runstate driver (ADR-0094), backed by Badger v4.
Package badger is the run-state runstate driver (ADR-0094), backed by Badger v4.

Jump to

Keyboard shortcuts

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