Documentation
¶
Overview ¶
Package worker is the Go SDK for packtrail workers. A worker serves one kind: it pulls jobs from <ns>.work.<kind>, runs a handler, and reports the outcome as a command (complete, fail, interrupt) to the job's reply subject. The protocol is plain NATS + JSON; SDKs in other languages implement the same steps.
w, _ := worker.New(nc, "summarize", func(ctx context.Context, j *worker.Job) (*worker.Result, error) {
var in struct{ Text string }
if err := j.Input(&in); err != nil {
return nil, worker.Permanent(err)
}
return &worker.Result{Output: map[string]any{"summary": in.Text[:10]}}, nil
})
_ = w.Run(ctx)
Index ¶
- Constants
- Variables
- func Interrupt(payload any) error
- func IsPermanent(err error) bool
- func Permanent(err error) error
- func TraceID(tp string) string
- func WithSummary(err error, summary any) error
- func WithUsage(err error, usage map[string]float64) error
- type Context
- type Handler
- type InterruptError
- type Job
- func (j *Job) Channel(name string, v any) error
- func (j *Job) DecodeMeta(v any) error
- func (j *Job) Input(v any) error
- func (j *Job) Interrupted(v any) (bool, error)
- func (j *Job) Item(v any) error
- func (j *Job) Logger() *slog.Logger
- func (j *Job) Progress(v any) error
- func (j *Job) Result(node string, v any) error
- func (j *Job) Resumed(v any) (bool, error)
- func (j *Job) Signal(name string, v any) error
- func (j *Job) Store() *Store
- type NodeError
- type Option
- func WithAckWait(d time.Duration) Option
- func WithBlobTimeout(d time.Duration) Option
- func WithConcurrency(n int) Option
- func WithDrainTimeout(d time.Duration) Option
- func WithLiveCheck(d time.Duration) Option
- func WithLogger(l *slog.Logger) Option
- func WithMaxAckPending(n int) Option
- func WithMaxDeliver(n int) Option
- func WithMeterProvider(mp metric.MeterProvider) Option
- func WithNamespace(ns string) Option
- func WithReconfigure() Option
- func WithTracerProvider(tp trace.TracerProvider) Option
- type Result
- type Store
- type SummaryError
- type UsageError
- type Worker
Examples ¶
Constants ¶
const ( DefaultConcurrency = 4 DefaultAckWait = 30 * time.Second DefaultMaxDeliver = 5 // DefaultMaxAckPending bounds the jobs of one kind in flight across all // worker processes. Each process takes only as many as its concurrency. DefaultMaxAckPending = 10000 )
Defaults.
const DefaultLiveCheck = 5 * time.Second
DefaultLiveCheck is how often a running job checks the head of its execution's log, in case a stop notice was missed.
const HeaderBusy = "Pt-Busy"
HeaderBusy counts how many times a job was handed back because its concurrency key was busy.
Variables ¶
var ( // ErrNotFound means a store key has no value. ErrNotFound = apierr.ErrNotFound // ErrInvalidArgument wraps an invalid namespace, key or value. ErrInvalidArgument = apierr.ErrInvalidArgument )
Store errors. They are the same values as packtrail.ErrNotFound and packtrail.ErrInvalidArgument: test with errors.Is against either.
var ErrCancelled = errors.New("worker: job cancelled")
ErrCancelled is the cause of a job context cancelled because the work is no longer wanted: its execution ended (cancelled, failed, completed elsewhere), its task was cancelled (a join settled, a map aborted), or its attempt timed out and was retried. Check it with context.Cause(ctx). The worker drops a cancelled job's result — it would be stale — and acks the job.
var ErrNoProgress = errors.New("worker: progress needs a running job")
ErrNoProgress is returned by Job.Progress outside a running job.
var ErrNotProvisioned = apierr.ErrNotProvisioned
ErrNotProvisioned is returned by Run when the namespace does not exist yet (no engine has run Init or Run on it). It is the same value as packtrail.ErrNotProvisioned; a worker started before the engine may retry.
Functions ¶
func Interrupt ¶
Interrupt returns an error that interrupts the node with payload (shown to whoever resumes it). On resume the node runs again with Job.Resumed true and the payload in Job.Interrupted.
func IsPermanent ¶
IsPermanent reports whether err was marked Permanent.
func TraceID ¶
TraceID returns the trace id of a W3C traceparent ("00-<trace id>-<parent id>-<flags>"), or "" when tp is not one.
func WithSummary ¶
WithSummary attaches a summary to the error a handler returns, so a run that interrupts or fails still records what it did, as Result.Summary does for a run that completes. It composes with WithUsage, Interrupt, Permanent and %w wrapping in any order. WithSummary(nil, …) is nil.
func WithUsage ¶
WithUsage attaches usage counters to the error a handler returns, so a run that interrupts or fails still adds what it spent to the execution's counters (budgets), as Result.Usage does for a run that completes. It composes with Interrupt, Permanent and %w wrapping in any order; when several are nested, the outermost counts. Every attempt's usage is added, once. WithUsage(nil, …) is nil: a successful run reports Result.Usage.
Types ¶
type Context ¶
type Context struct {
Input json.RawMessage `json:"input"`
Channels map[string]json.RawMessage `json:"channels"`
Results map[string]json.RawMessage `json:"results"`
LastNode string `json:"last_node"`
Visits map[string]int `json:"visits"`
Signals map[string]json.RawMessage `json:"signals"`
Branches map[string]string `json:"branches"`
Counters map[string]float64 `json:"counters"`
// Errors holds the last permanent failure of each node: what an
// on_failure handler reacts to (errors[last_node] for the node that
// routed here).
Errors map[string]NodeError `json:"errors"`
Item json.RawMessage `json:"item,omitempty"`
Index *int `json:"index,omitempty"`
Resume json.RawMessage `json:"resume,omitempty"`
// Interrupt is the payload the instance interrupted with, on every
// attempt of its resumed run; absent otherwise.
Interrupt json.RawMessage `json:"interrupt,omitempty"`
}
Context is the execution view handed to a worker.
type Handler ¶
Handler runs one job. Return a Result on success; an error fails the attempt (retried by the node's policy unless Permanent); Interrupt pauses the execution. A panic is recovered and treated as a retryable failure: it never takes the worker process down (I-21).
type InterruptError ¶
type InterruptError struct {
Payload any
}
InterruptError pauses the execution at this node until Client.Resume.
func (*InterruptError) Error ¶
func (e *InterruptError) Error() string
type Job ¶
type Job struct {
ExecID string
Flow string
FlowHash string
Node string
Kind string
Key string
Index int
Generation int
Attempt int
// Deliveries counts deliveries of this job message (redelivery after a
// crash or nak).
Deliveries uint64
Traceparent string
// Context is the execution view: input, channels, results, last_node,
// visits, signals, branches, counters, errors, and item/index/resume when
// set.
Context Context
// Meta is the node's meta (flow.Node.Meta) as defined in the flow version
// the execution runs, so a redeploy never changes it under a running
// execution, a fork or a rerun. Nil when the node has none.
Meta json.RawMessage
// contains filtered or unexported fields
}
Job is one attempt of a task instance.
func (*Job) DecodeMeta ¶
DecodeMeta decodes the node's meta into v.
func (*Job) Interrupted ¶
Interrupted reports whether the job is a resumed run carrying the payload its instance interrupted with (Interrupt(payload)), and decodes it into v (when v is non-nil). An interrupt without payload resumes with false.
func (*Job) Logger ¶
Logger returns the worker's logger with this attempt bound: exec, flow, node, key, kind, generation and attempt, plus trace_id when the job carries a W3C traceparent. Handlers log through it so their lines join the engine's and the worker's on the same keys.
func (*Job) Progress ¶
Progress publishes v (anything JSON-encodable) as an intermediate result of this job: partial output, a percentage, a log line. It is delivered to clients watching the execution's progress (Client.Progress) and never stored — the log only records the final result. Delivery is best effort, in order per job; a message larger than the server's max payload is refused.
func (*Job) Resumed ¶
Resumed reports whether the job is the resumption of an interrupt, and decodes the resume value into v (when v is non-nil).
type NodeError ¶
type NodeError struct {
Error string `json:"error"`
Reason string `json:"reason"`
Attempt int `json:"attempt,omitempty"`
// Index is the failed item of a map.
Index *int `json:"index,omitempty"`
}
NodeError is how a node last failed permanently.
type Option ¶
Option configures a Worker.
func WithAckWait ¶
WithAckWait sets how long a job may go without a heartbeat before it is redelivered. Values under one second are raised to one second (a nanosecond ack wait would break the heartbeat ticker).
func WithBlobTimeout ¶
WithBlobTimeout bounds one claim-check transfer (large job contexts and results) when the handler context has no deadline (default 2m).
func WithConcurrency ¶
WithConcurrency bounds parallel jobs; the consumer never prefetches more than this: prefetched jobs would expire while waiting for a slot.
func WithDrainTimeout ¶
WithDrainTimeout bounds the graceful drain on shutdown.
func WithLiveCheck ¶
WithLiveCheck sets how often a running job checks whether its execution ended (default 5s); 0 disables the check, leaving only stop notices.
func WithMaxAckPending ¶
WithMaxAckPending bounds the jobs of this kind in flight across every worker process (default 10000). It is a kind-wide setting: see WithReconfigure.
func WithMaxDeliver ¶
WithMaxDeliver sets how many deliveries a job gets before it is dead-lettered and its node failed (redelivery happens on crash or when the command cannot be published).
func WithMeterProvider ¶
func WithMeterProvider(mp metric.MeterProvider) Option
WithMeterProvider sets where the worker's metrics go (default: the global OpenTelemetry provider).
func WithNamespace ¶
WithNamespace sets the namespace (default "packtrail").
func WithReconfigure ¶
func WithReconfigure() Option
WithReconfigure makes this worker apply its kind-wide settings (ack wait, max deliver, max ack pending) to the shared consumer. Without it the consumer is created once by the first worker and then used as is, so processes started with different options do not overwrite each other.
func WithTracerProvider ¶
func WithTracerProvider(tp trace.TracerProvider) Option
WithTracerProvider sets where the worker's spans go (default: the global OpenTelemetry provider). Each job runs in a span whose context the handler gets, so spans the handler starts are its children.
type Result ¶
type Result struct {
// Output is the node output: a JSON object (a map, a struct, or a
// json.RawMessage holding an object), or nil.
Output any
// Writes are deltas for state channels, folded by each channel's reducer.
Writes map[string]any
// Next routes to a declared dynamic successor (Command(goto)).
Next string
// Usage reports generic counters (budgets).
Usage map[string]float64
// Summary describes what the attempt did (any value that encodes to
// JSON): it is recorded on the attempt's event, for timelines and
// audits, and never read by the flow. Keep it small.
Summary any
}
Result is the outcome of a successful job.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the long-term key-value store of the worker's namespace: the one packtrail.Client.Store reads and writes, for application data that outlives executions (per-user memory, learned facts). Namespaces and keys are tokens.
Store writes are not part of the execution log: a redelivered or retried job runs its writes again, so make them idempotent (write a value derived from the job's context rather than appending blindly).
type SummaryError ¶
SummaryError carries the summary of a run that ended with an error (see WithSummary). It unwraps to Err, so the error is classified as Err is.
func (*SummaryError) Error ¶
func (e *SummaryError) Error() string
func (*SummaryError) Unwrap ¶
func (e *SummaryError) Unwrap() error
type UsageError ¶
UsageError carries the usage of a run that ended with an error (see WithUsage). It unwraps to Err, so the error is classified as Err is.
func (*UsageError) Error ¶
func (e *UsageError) Error() string
func (*UsageError) Unwrap ¶
func (e *UsageError) Unwrap() error
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker serves one kind.
func New ¶
New returns a worker for kind. It performs no I/O.
Example ¶
package main
import (
"context"
"errors"
"fmt"
"github.com/nats-io/nats.go"
"github.com/henomis/packtrail/worker"
)
func main() {
var nc *nats.Conn // a connected NATS client
handler := func(_ context.Context, j *worker.Job) (*worker.Result, error) {
var in struct{ Text string }
if err := j.Input(&in); err != nil {
return nil, worker.Permanent(err) // bad input: do not retry
}
if in.Text == "" {
return nil, errors.New("empty text") // retried by the node's policy
}
return &worker.Result{
Output: map[string]any{"length": len(in.Text)},
Writes: map[string]any{"seen": 1},
}, nil
}
_, err := worker.New(nc, "measure", handler, worker.WithConcurrency(8))
fmt.Println(err)
}
Output: worker: nil connection or handler
func (*Worker) Ready ¶
func (w *Worker) Ready() <-chan struct{}
Ready returns a channel closed once Run is subscribed to stop notices and pulling jobs: the signal for a readiness probe. It stays open if Run fails first, and once Run returns Ready gives a new, open channel: call it at each probe. It reports start-up only, not later connectivity.
Source Files