tasks

package
v0.11.0 Latest Latest
Warning

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

 Go to latest Published: Aug 2, 2026 License: LGPL-3.0 Imports: 12 Imported by: 0

Documentation

Overview

Package tasks defines the engine's external-worker execution contract: an asynchronous fetch-and-lock job queue (ADR-021 §2.4, SRD-036). The engine Enqueues a job and parks the ServiceTask; a worker FetchAndLocks it, executes, and reports (Complete/Fail); the report re-enters the instance loop as a WorkerOutcome and resumes the parked track. The in-process default lives in the localdispatcher sibling subpackage; a remote adapter (HTTP/gRPC) is a future extension (ADR-004).

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ApplyOutputMapping added in v0.9.0 

func ApplyOutputMapping(
	ctx context.Context,
	ee expression.Engine,
	rules []OutputRule,
	body *data.ItemDefinition,
) ([]data.Data, error)

ApplyOutputMapping evaluates each rule's Path over body and returns the extracted data to commit as the ServiceTask's output, one datum per Var head. A plain Var ("orderId") emits its evaluated value whole; a set of structural Vars sharing a head ("order.total", "order.items[0].price") assembles ONE record for that head via values.SetPath (ADR-011 v.6 §2.9.3, §2.9.5). Mixing a whole-value and a structural rule on one head, or a malformed Var path, is a classified mapping error; a required path that fails to evaluate is a fault (a worker response that violates the contract → technical fault), an optional one is skipped. The body is exposed to Path as the "body" datum through the same transient Source the ErrorMapper classifier uses (SRD-037 §4.5).

Types

type BpmnError added in v0.9.0 

type BpmnError struct {
	Code    string
	Message string
}

BpmnError yields a Business Error: the engine raises Code as a BPMN error (interrupting), caught by a matching Error boundary event (ADR-018). Message is an optional diagnostic.

type ErrorMapper added in v0.9.0 

type ErrorMapper interface {
	Classify(ctx context.Context, ee expression.Engine, f Fault) (MappedOutcome, error)
}

ErrorMapper classifies a raw worker Fault into a MappedOutcome. The declarative RuleMapper covers the common cases; a custom implementation covers imperative ones the rule list can't express (ADR-021 §2.6). It is evaluated at resume with the execution's expression engine (SRD-037 §4.1).

type ExpressionEngineBinder added in v0.9.0 

type ExpressionEngineBinder interface {
	BindExpressionEngine(ee expression.Engine)
}

ExpressionEngineBinder is an optional dispatcher capability: the engine binds its expression engine at startup so the dispatcher can run a Job's ErrorMapper (which evaluates FormalExpressions) when it classifies a raw fault engine-side under EngineAuthoritative (SRD-038). A dispatcher that never classifies engine-side (e.g. a WorkerTrusted-only remote adapter) need not implement it.

type ExternalWorker added in v0.9.0 

type ExternalWorker interface {
	// WorkerTopic reports the external-worker topic and whether the node is
	// worker-dispatched.
	WorkerTopic() (topic Topic, ok bool)

	// BindJobInput binds the node's operation input message from r (without
	// executing it) — the payload the engine puts in the enqueued Job.Input.
	BindJobInput(
		ctx context.Context, r service.DataReader,
	) (*data.ItemDefinition, error)
}

ExternalWorker is implemented by a node whose work is dispatched to an external worker rather than run in-process. The instance loop diverts a node whose WorkerTopic reports ok == true to the wait-node park path (it enqueues a job and resumes on the worker's report); ok == false runs the node in-process as usual.

type Fault added in v0.9.0 

type Fault struct {
	Body  *data.ItemDefinition
	Cause error
	Code  string
}

Fault is a worker's raw (unclassified) terminal fault (ADR-021 §2.6, SRD-037). Code is a protocol/domain status (an HTTP status once a remote transport exists, ADR-004); Body is the response payload; Cause is the diagnostic Go error. The engine ErrorMapper classifies {Code, Body}; an all-empty Fault (only Cause) matches no rule and falls through to the default technical outcome.

type Job 

type Job struct {
	Input  *data.ItemDefinition
	Policy *Policy
	ID     JobID
	Topic  Topic
}

Job is the unit the engine Enqueues. Input is the single bound input-message item (nil if the operation has no inMessage), per the operation contract.

type JobCompletionSink added in v0.9.0 

type JobCompletionSink interface {
	ReportJobCompletion(ctx context.Context, outcome *WorkerOutcome) error
}

JobCompletionSink routes a worker's terminal report to the owning instance. The engine implements it; a dispatcher calls it from Complete/Fail. Keeping this separate from WorkerDispatcher keeps the queue decoupled from instance internals (SRD-036 §4.1).

type JobID added in v0.9.0 

type JobID string

JobID identifies one execution of a worker-dispatched ServiceTask; a worker treats it as an idempotency key. It embeds the owning instance's id (see MakeJobID) so a completion routes back to that instance without a separate registry (SRD-036 §4.5).

func MakeJobID added in v0.9.0 

func MakeJobID(instanceID string) JobID

MakeJobID composes a JobID for a worker-dispatched ServiceTask on instanceID, embedding that id so ReportJobCompletion routes the outcome back to the owning instance without a registry (SRD-036 §4.5). The suffix is a fresh unique id, so two jobs on the same instance never collide.

func (JobID) InstanceID added in v0.9.0 

func (j JobID) InstanceID() string

InstanceID returns the owning instance id embedded in the JobID (the segment before the first separator), or the whole string if it carries none.

type LockedJob added in v0.9.0 

type LockedJob struct {
	Deadline time.Time
	Job
	WorkerID WorkerID
}

LockedJob is a Job a worker received from FetchAndLock, together with its lock: WorkerID holds it until Deadline, extendable via ExtendLock.

type LoggerBinder added in v0.9.0 

type LoggerBinder interface {
	BindLogger(logger observability.Logger)
}

LoggerBinder is an optional dispatcher capability: the engine binds its configured logger (from the runtime config) at startup, so a dispatcher's own lifecycle logging uses the embedder's logger rather than a private default. A dispatcher that manages its own logging need not implement it.

type MappedOutcome added in v0.9.0 

type MappedOutcome interface {
	// contains filtered or unexported methods
}

MappedOutcome is the classification an ErrorMapper yields for a raw fault: one of BpmnError (business, interrupting), Status (business, non-interrupting), or Technical (retry/terminal). It is a sealed interface — an unexported marker, mirroring options.Option — so the set of outcomes is closed to this package.

type OutcomeKind added in v0.9.0 

type OutcomeKind uint8

OutcomeKind classifies a WorkerOutcome into one of the four ADR-021 §2.6 kinds. The parked ServiceTask dispatches on it at resume (SRD-037 §3.5). 

const (
	// OutcomeComplete is a success — bind Output, complete.
	OutcomeComplete OutcomeKind = iota
	// OutcomeBpmnError is a worker-declared Business Error — raise BpmnCode, caught
	// by an Error boundary (interrupting).
	OutcomeBpmnError
	// OutcomeStatus is a worker-declared Business Status — write StatusValue to the
	// WithStatus variable, complete normally.
	OutcomeStatus
	// OutcomeFault is a raw fault — the engine ErrorMapper classifies Fault{code,body}
	// at resume.
	OutcomeFault
)

func (OutcomeKind) String added in v0.9.0 

func (k OutcomeKind) String() string

String returns the outcome-kind name for logging.

type OutputRule added in v0.9.0 

type OutputRule struct {
	Path     data.FormalExpression
	Var      string
	Required bool
}

OutputRule extracts a value from a Complete's raw response body into a named output variable (ADR-021 §2.5, SRD-037 FR-7). Path is a body-path expression (in-process binding: gobpm expressions over Go values, reading the body as the "body" datum); Var is the output variable it fills — a plain name ("orderId") or a structural path ("order.items[0].price", ADR-011 v.6 §2.9.3) that shapes one nested output value; Required makes a path the body doesn't satisfy a fault rather than a skipped mapping.

type Policy added in v0.9.0 

type Policy struct {
	// ErrorMapper classifies a raw fault; nil = raw faults fall through to the
	// default Technical outcome.
	ErrorMapper ErrorMapper
	// RetryPolicy drives technical-fault retries; wired in SRD-038 M7 (nil when
	// unset).
	RetryPolicy RetryPolicy
	// OutputMapping shapes a completion's raw body into the final committed output;
	// applied by the policy owner (dispatcher/worker), not the track (SRD-039 M8).
	// Empty = the raw output is committed directly.
	OutputMapping []OutputRule
	// Trust selects the policy locus — the worker (WorkerTrusted) or the engine's
	// dispatcher (EngineAuthoritative). Resolved at enqueue; never trustUnset on a
	// shipped job (SRD-039 M9).
	Trust TrustMode
}

Policy is a worker-dispatched ServiceTask's resolved outcome policy. Under EngineAuthoritative (SRD-038) the dispatcher uses it to classify a raw fault (ErrorMapper) and drive technical retries (RetryPolicy); under WorkerTrusted (SRD-039) it is shipped to the worker. The instance resolves it (two-level: per-service over engine-wide) at enqueue. A nil Policy = an in-process task or a not-yet-populated job.

type ReporterBinder added in v0.9.0 

type ReporterBinder interface {
	BindReporter(sink observability.Reporter)
}

ReporterBinder is an optional dispatcher capability: the engine binds its observable-event sink (ADR-013 v.2 §2.7) at startup so the dispatcher can emit JobState events (enqueue/lock/report/retry/exhaust/reclaim) onto the one engine-wide seam. A third-party dispatcher that does not implement it simply does not emit — the optional-capability pattern.

type RetryPolicy added in v0.9.0 

type RetryPolicy interface {
	Retry(attempt int, cause error) (backoff time.Duration, retry bool)
}

RetryPolicy decides whether a technical fault is retried and the backoff before the next attempt (ADR-021 §2.7). attempt is the 1-based number of the attempt that just failed; cause is its technical error.

func DefaultRetryPolicy added in v0.9.0 

func DefaultRetryPolicy() RetryPolicy

DefaultRetryPolicy is the engine default when neither a per-service nor an engine-wide RetryPolicy is set: 3 attempts with jittered exponential backoff from 500ms capped at 30s — improving on Camunda's zero-wait default (ADR-021 §2.7).

func ExponentialBackoff added in v0.9.0 

func ExponentialBackoff(
	maxAttempts int, base, maxBackoff time.Duration, jitter bool,
) RetryPolicy

ExponentialBackoff returns a RetryPolicy that retries until maxAttempts executions have run, doubling the backoff (base, 2·base, 4·base, …) capped at maxBackoff. With jitter, each backoff is randomized into [d/2, d] to spread retries.

func FixedDelay added in v0.9.0 

func FixedDelay(maxAttempts int, delay time.Duration) RetryPolicy

FixedDelay returns a RetryPolicy that retries a technical fault until maxAttempts executions have been made, waiting delay before each retry.

func NoRetry added in v0.9.0 

func NoRetry() RetryPolicy

NoRetry returns a RetryPolicy that never retries: a technical fault is terminal on first report (a fail-fast worker).

type Rule added in v0.9.0 

type Rule struct {
	BodyClause data.FormalExpression
	Yield      MappedOutcome
	Code       string
}

Rule is one ErrorMapper classification rule (first match wins). Code is an exact code match ("" matches any code); BodyClause is an optional predicate over the fault's {code, body} (nil = code-only); Yield is the outcome on a match.

type RuleMapper added in v0.9.0 

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

RuleMapper is the declarative ErrorMapper: an ordered rule list, first match wins, falling through to Technical when none match.

func NewRuleMapper added in v0.9.0 

func NewRuleMapper(rules ...Rule) (*RuleMapper, error)

NewRuleMapper builds a declarative ErrorMapper from rules (evaluated in order). A rule with a nil Yield is rejected — every rule must classify to some outcome.

func (*RuleMapper) Classify added in v0.9.0 

func (m *RuleMapper) Classify(
	ctx context.Context,
	ee expression.Engine,
	f Fault,
) (MappedOutcome, error)

Classify returns the first rule's Yield whose code matches and whose BodyClause (if any) evaluates true over the fault's {code, body}; no match → Technical.

type SinkBinder added in v0.9.0 

type SinkBinder interface {
	BindSink(sink JobCompletionSink)
}

SinkBinder is an optional dispatcher capability: the engine binds its completion sink at startup so the dispatcher can deliver outcomes back. A dispatcher that reaches the engine another way (e.g. a remote adapter) need not implement it.

type Status added in v0.9.0 

type Status struct {
	Value data.Value
}

Status yields a Business Status: the engine writes Value to the ServiceTask's WithStatus variable and the task completes normally.

type Technical added in v0.9.0 

type Technical struct{}

Technical yields a technical fault: it feeds the retry policy (SRD-038) and is terminal for now (SRD-037). It is also the implicit default when no rule matches.

type Topic added in v0.9.0 

type Topic string

Topic is a job's type/fetch key — a worker fetches by topic, and it equals a ServiceTask's WithWorker topic.

type TrustMode added in v0.9.0 

type TrustMode uint8

TrustMode governs where a worker-dispatched ServiceTask's policy bundle (output mapping, classification, retry) executes (ADR-021 §2.6). The zero value is "unset" — an internal resolution sentinel; a resolved job is always one of the two exported modes, defaulting to WorkerTrusted. 

const (

	// WorkerTrusted (default) ships the policy to the worker, which maps its
	// output, self-classifies its faults, and retries technical faults
	// internally, reporting only a final verdict (ADR-021 §2.6).
	WorkerTrusted TrustMode
	// EngineAuthoritative keeps the policy engine-side: the worker returns the raw
	// {code, body} and the engine maps / classifies / retries (ADR-021 §2.6).
	EngineAuthoritative
)

func (TrustMode) Resolve added in v0.9.0 

func (m TrustMode) Resolve(fallback TrustMode) TrustMode

Resolve returns m when it is a configured mode, else fallback. It composes the two-level resolution — per-service over engine-wide over the WorkerTrusted default — while keeping the unset sentinel unexported: 

perService.Resolve(engineWide.Resolve(WorkerTrusted))

func (TrustMode) String added in v0.9.0 

func (m TrustMode) String() string

String returns the trust-mode name for logging.

type WorkerConfig added in v0.9.0 

type WorkerConfig interface {
	WorkerConfig() (perService Policy, ok bool)
}

WorkerConfig is implemented by a node whose worker outcome the engine classifies, maps, and retries. WorkerConfig returns the node's per-service policy (a partial Policy — a nil/empty field means "fall back to the engine-wide default" resolved at enqueue); ok == false for an in-process (non-worker) node.

type WorkerDispatcher 

type WorkerDispatcher interface {
	// Enqueue adds a job to the queue (non-blocking); the engine then parks the
	// ServiceTask.
	Enqueue(ctx context.Context, job Job) error

	// FetchAndLock returns and locks (for lockDuration, to workerID) the next
	// available jobs for the given topics, blocking until at least one is
	// available or ctx is done.
	FetchAndLock(
		ctx context.Context,
		workerID WorkerID,
		topics []Topic,
		lockDuration time.Duration,
	) ([]LockedJob, error)

	// ExtendLock extends the lock on jobID (held by workerID) by newDuration
	// from now. Holder-only; bounded by the configured maxLockDuration.
	ExtendLock(
		ctx context.Context,
		jobID JobID,
		workerID WorkerID,
		newDuration time.Duration,
	) error

	// Complete reports a successful outcome: output is the operation's result
	// item (nil if the operation has no outMessage).
	Complete(
		ctx context.Context,
		jobID JobID,
		workerID WorkerID,
		output *data.ItemDefinition,
	) error

	// ReportBpmnError reports a worker-declared Business Error (Camunda
	// handleBpmnError): the engine raises code (message is an optional
	// diagnostic), caught by a matching Error boundary event (interrupting).
	ReportBpmnError(
		ctx context.Context,
		jobID JobID,
		workerID WorkerID,
		code, message string,
	) error

	// ReportStatus reports a worker-declared Business Status: the engine writes
	// value to the ServiceTask's WithStatus variable and the task completes.
	ReportStatus(
		ctx context.Context,
		jobID JobID,
		workerID WorkerID,
		value data.Value,
	) error

	// Fail reports a raw fault the engine ErrorMapper classifies (§2.6). A
	// pure-technical fault carries only Fault.Cause (empty code, nil body) and
	// falls through to the default technical outcome.
	Fail(ctx context.Context, jobID JobID, workerID WorkerID, fault Fault) error
}

WorkerDispatcher is an asynchronous fetch-and-lock job queue (ADR-021 §2.4). Enqueue is engine-facing; FetchAndLock / ExtendLock and the four terminal reports (Complete / ReportBpmnError / ReportStatus / Fail) are worker-facing. The reports mirror the four outcome kinds (SRD-037 §2.6): a worker either self-classifies (ReportBpmnError / ReportStatus) or reports a raw Fault the engine ErrorMapper classifies.

type WorkerError added in v0.9.0 

type WorkerError struct {
	Cause         error      // the technical cause (a plain-technical verdict)
	Status        data.Value // non-nil → a Business Status verdict
	BpmnErrorCode string     // non-empty → a Business Error verdict
	Message       string     // optional Business Error diagnostic
}

WorkerError is a WorkerFunc's rich, self-classifying error (ADR-021 §2.6): a WorkerTrusted worker returns it to declare its own outcome. Precedence is BpmnErrorCode → Status → technical (Cause). A plain (non-*WorkerError) error is an unclassified technical fault the worker runs through the fallback ErrorMapper.

func (*WorkerError) Error added in v0.9.0 

func (e *WorkerError) Error() string

Error implements error, reporting the declared classification.

type WorkerID added in v0.9.0 

type WorkerID string

WorkerID identifies the worker holding a job's lock.

type WorkerOutcome added in v0.9.0 

type WorkerOutcome struct {
	foundation.BaseElement
	// contains filtered or unexported fields
}

WorkerOutcome is the synthetic event a worker's report rides back into the instance loop: it implements flow.EventDefinition, so it flows through the parked track's event channel exactly like a UserTask completion (ADR-021 §2.4, SRD-036 §3.2). Its Kind selects which field is meaningful. It never reaches the EventHub or correlation, so its Type is an internal sentinel and it exposes no ItemDefinitions.

func NewWorkerBpmnError added in v0.9.0 

func NewWorkerBpmnError(jobID JobID, code, message string) *WorkerOutcome

NewWorkerBpmnError builds a worker-declared Business Error outcome: the engine raises code (message is an optional diagnostic), caught by an Error boundary.

func NewWorkerComplete added in v0.9.0 

func NewWorkerComplete(jobID JobID, output []data.Data) *WorkerOutcome

NewWorkerComplete builds a successful outcome for job jobID carrying the final committed output (nil if the operation produced none). The output is already shaped — the policy owner (the dispatcher under EngineAuthoritative, the worker under WorkerTrusted) applied WithOutputMapping before this outcome is built, so the track only commits it (SRD-039 §3.4).

func NewWorkerFault added in v0.9.0 

func NewWorkerFault(jobID JobID, fault Fault) *WorkerOutcome

NewWorkerFault builds a raw-fault outcome for job jobID; the engine ErrorMapper classifies fault's {code, body} at resume.

func NewWorkerStatus added in v0.9.0 

func NewWorkerStatus(jobID JobID, value data.Value) *WorkerOutcome

NewWorkerStatus builds a worker-declared Business Status outcome carrying value (written to the ServiceTask's WithStatus variable).

func (*WorkerOutcome) BpmnError added in v0.9.0 

func (o *WorkerOutcome) BpmnError() (code, message string)

BpmnError returns the worker-declared business-error code and message (empty on other kinds).

func (*WorkerOutcome) Fault added in v0.9.0 

func (o *WorkerOutcome) Fault() Fault

Fault returns the raw fault (zero on other kinds); the engine ErrorMapper classifies its {code, body}.

func (*WorkerOutcome) GetItemsList added in v0.9.0 

func (o *WorkerOutcome) GetItemsList() []*data.ItemDefinition

GetItemsList returns nil — a WorkerOutcome carries its data via its kind-specific accessors, not ItemDefinitions.

func (*WorkerOutcome) JobID added in v0.9.0 

func (o *WorkerOutcome) JobID() JobID

JobID returns the job this outcome reports.

func (*WorkerOutcome) Kind added in v0.9.0 

func (o *WorkerOutcome) Kind() OutcomeKind

Kind returns the outcome's classification.

func (*WorkerOutcome) Output added in v0.9.0 

func (o *WorkerOutcome) Output() []data.Data

Output returns the final committed output on a completion — already shaped by the policy owner (nil otherwise). The track commits it as-is.

func (*WorkerOutcome) StatusValue added in v0.9.0 

func (o *WorkerOutcome) StatusValue() data.Value

StatusValue returns the worker-declared business-status value (nil on other kinds).

func (*WorkerOutcome) Type added in v0.9.0 

func (o *WorkerOutcome) Type() flow.EventTrigger

Type returns the internal worker-outcome sentinel.

Source Files

View all Source files

Directories

Path Synopsis
Package localdispatcher provides the engine's default WorkerDispatcher (ADR-021 §2.4, SRD-036): an in-memory fetch-and-lock job store with per-job lock state and a local worker pool.
 Package localdispatcher provides the engine's default WorkerDispatcher (ADR-021 §2.4, SRD-036): an in-memory fetch-and-lock job store with per-job lock state and a local worker pool.

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.