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
- func ForeignRecord(rec *runstate.Record, uid v1.UID) bool
- func IsKVMarker(r v1.OwnerReference) bool
- func KVMarker(wf *v1.Workflow) v1.OwnerReference
- func MarkKVStoresOnce(ctx context.Context, s store.Store, log *slog.Logger) error
- func Permanent(err error) error
- func PinString(p v1.RevisionPin) string
- func StepFunctions(workflow v1.ObjectName, spec v1.WorkflowSpec) []v1.ObjectName
- type ChildResolver
- type ChildWorkflowResolver
- type Config
- type ContractResolver
- type Deps
- type DispatchDeps
- type DispatchRequest
- type Dispatcher
- type Engine
- func (e *Engine) Cancel(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) error
- func (e *Engine) Execute(ctx context.Context, ns v1.NamespaceName, runName, workflow v1.ObjectName, ...) (*runstate.Record, error)
- func (e *Engine) Pause(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) error
- func (e *Engine) Replay(ctx context.Context, ns v1.NamespaceName, runName, workflow v1.ObjectName, ...) (*runstate.Record, error)
- func (e *Engine) Resume(ctx context.Context, ns v1.NamespaceName, runName v1.ObjectName) (*runstate.Record, error)
- func (e *Engine) Run(ctx context.Context, drain time.Duration)
- func (e *Engine) SweepExpired(ctx context.Context, retention time.Duration, ...) (int, error)
- type Granter
- type HTTPDispatcher
- type Materializer
- type RunReconciler
- type RuntimeResolver
- type StartOptions
- type Waker
- type WorkflowReconciler
Constants ¶
const ( KVMigrationNamespace v1.NamespaceName = "funcd-system" KVMigrationRecord v1.ObjectName = "kvstore-marker-migration" )
The completion record of the KVStore marker migration (ADR-0180 Decision 3): written after the last mark and the binding strip, so a crash before either finishes re-runs it. The control plane refuses API writes to it.
Variables ¶
This section is empty.
Functions ¶
func ForeignRecord ¶ added in v0.6.1
ForeignRecord reports whether rec belongs to a WorkflowRun other than the one with uid: an earlier one of the same name. A record without a uid predates the stamp and is matched by name.
func IsKVMarker ¶ added in v0.5.0
func IsKVMarker(r v1.OwnerReference) bool
IsKVMarker reports whether r has the shape of a KVStore marker: a non-controller Workflow ref.
func KVMarker ¶ added in v0.5.0
func KVMarker(wf *v1.Workflow) v1.OwnerReference
KVMarker is kvMarker for the control plane's handover, so the marker has one definition.
func MarkKVStoresOnce ¶ added in v0.5.0
MarkKVStoresOnce marks, once, each unmarked KVStore the previous materializer made for a Workflow, then strips every Workflow-controlled Function's binding to a store it left unmarked and records completion. funcd calls it at boot before the controllers, the collector and the control plane start (ADR-0178). It logs through log (nil means slog.Default()).
func PinString ¶ added in v0.7.0
func PinString(p v1.RevisionPin) string
PinString names a revision pin in a fault message.
func StepFunctions ¶ added in v0.7.0
func StepFunctions(workflow v1.ObjectName, spec v1.WorkflowSpec) []v1.ObjectName
StepFunctions names the Functions spec's function steps dispatch to, the onFailure handler included: an image step's materialized <workflow>-<step>, a ref step's referenced Function (ADR-0190 pins each).
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
// MaxStepsInFlight bounds the function-step dispatch attempts in flight across all runs; 0 ⇒ no cap (ADR-0146).
MaxStepsInFlight int
// DefaultRetryBackoff is the first retry gap of a step whose retry.backoff is unset or 0, doubled per attempt
// (workflow.defaultRetryBackoff, ADR-0163 Decision 8); 0 ⇒ back to back.
DefaultRetryBackoff time.Duration
}
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
// Notify: a top-level run's key after each record write and on goroutine exit; nil ⇒ none; must not block
// (ADR-0146: wired to Controller.Enqueue, so the run reconciler mirrors the record).
Notify func(ns v1.NamespaceName, name v1.ObjectName)
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
// Revision is Target's pin from the run record (ADR-0190); nil ⇒ by-name dispatch (a pre-ADR-0190 record).
Revision *v1.RevisionPin
}
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 (*Engine) Cancel ¶
func (e *Engine) Cancel(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) error
Cancel abandons a run (ADR-0146 Decision 3). A live run's context is cancelled with a cancel cause: its in-flight calls close and its goroutine records the run Cancelled. Otherwise the record is written here: pending and running steps Cancelled (a running one with its end time and the cancel error) and the run Cancelled. A terminal record 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 until it is terminal, paused or halted, and returns the record. The run reconciler calls it on the run's engine-owned goroutine (start, ADR-0146). Every ready step is dispatched at once (ADR-0094: fan-out is parallel dispatch). ctx ends the steps; record writes outlive it.
func (*Engine) Pause ¶
func (e *Engine) Pause(ctx context.Context, ns v1.NamespaceName, name v1.ObjectName) error
Pause requests a graceful pause (ADR-0146 Decision 5). A live run is only signalled: its goroutine lets the in-flight calls finish, starts nothing new and persists Paused. Otherwise the record is marked Paused, so the next drive dispatches nothing. A terminal or already Paused record 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) Run ¶ added in v0.4.0
Run blocks until ctx ends, then drains (ADR-0146 Decision 6) for at most drain: start refuses, every drive starts no new step or attempt, and the in-flight calls may finish; at the bound they are cancelled with errRunStopped and return to Pending. Run returns when every run goroutine exited.
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, pin *v1.RevisionPin) 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. A non-nil pin is the target's revision pin (ADR-0190): it is granted only while that revision is the pinned one.
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 ¶
func (d *HTTPDispatcher) Dispatch(ctx context.Context, req DispatchRequest) (json.RawMessage, error)
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, supervisionPeriod time.Duration) *Materializer
NewMaterializer builds the materializer. supervisionPeriod 0 ⇒ controller.SupervisionPeriod.
func (*Materializer) Materialize ¶
Materialize brings the workflow's owned resources to the desired state. It first strips every binding of a Function this incarnation controls to a store it did not make, then binds a step only to stores this incarnation made (ADR-0178 Decision 2).
type RunReconciler ¶
type RunReconciler struct {
// contains filtered or unexported fields
}
RunReconciler drives WorkflowRun resources: it starts, signals and stops their engine runs, mirrors the coarse run state into WorkflowRun.status (its only writer, ADR-0146), 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, referentPollInterval time.Duration) *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). referentPollInterval re-checks a run waiting for its Workflow; 0 ⇒ waitRequeue.
func (*RunReconciler) Reconcile ¶
func (r *RunReconciler) Reconcile(ctx context.Context, req controller.Request) (controller.Result, error)
Reconcile starts, signals or stops one WorkflowRun and mirrors its run record into status, never waiting for a step (ADR-0146 Decision 1): the run executes on an engine-owned goroutine, whose record writes and exit enqueue the run again.
func (*RunReconciler) SweepExpired ¶ added in v0.2.1
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
ChildPins map[v1.ObjectName]runstate.ChildPin // the run's child tree; execute copies it onto the record
// Pins are the run's step Function revision pins (ADR-0190): execute stores them; a child record gets its subtree's.
Pins map[v1.ObjectName]v1.RevisionPin
}
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, artifactPollInterval time.Duration) *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). artifactPollInterval re-checks a Workflow whose step artifact is not pushed; 0 ⇒ contractRequeue.
func (*WorkflowReconciler) Reconcile ¶
func (r *WorkflowReconciler) Reconcile(ctx context.Context, req controller.Request) (controller.Result, error)
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).
Source Files
¶
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. |