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 ¶
- func Permanent(err error) error
- type ChildResolver
- 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) 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 ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
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 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 step output (and run input, at admission); 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
// 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 (*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).
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). Steps of a ready batch are dispatched sequentially in V1 (correct for the DAG; concurrent fan-out is a performance optimization).
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).
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 ¶
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. 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 ¶
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) *Materializer
NewMaterializer builds the materializer.
func (*Materializer) Materialize ¶
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 ¶
func (r *RunReconciler) Reconcile(ctx context.Context, req controller.Request) (controller.Result, error)
Reconcile drives one WorkflowRun toward its terminal phase.
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)
}
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 ¶
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. |