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 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) 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 WithNamespace(ns string) Option
- func WithReconfigure() Option
- type Result
- type Store
- 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 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) 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 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.
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
}
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 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