Documentation
¶
Overview ¶
Package job is Atlas's in-process worker harness: it bridges the engine's activatable jobs to worker handlers and feeds their results back as commands (ADR-0007, streaming pull with completion-as-command).
This is the in-process form. Workers register a Handler per job type; the Runner pulls activatable jobs of those types from the state store, runs the handler, and submits CompleteJob back through the processor — the processor never blocks on the handler. A handler that returns an error is routed into the incident model (ADR-0061): the job is failed (retried while retries remain, then an incident parks its token) rather than aborting the whole Drive, so one failing job cannot poison the run loop. The gRPC streaming transport and job leases with timeout/backoff are later milestones.
Index ¶
- Constants
- type CompletingHandler
- type Completion
- type Engine
- type Handler
- type Job
- type Outcome
- type OutputHandler
- type Runner
- func (r *Runner) Claim() ([]Job, error)
- func (r *Runner) Drive() error
- func (r *Runner) Handle(jobType int32, build func(state.Reader) Handler)
- func (r *Runner) HandleCompleting(jobType int32, build func(state.Reader) CompletingHandler)
- func (r *Runner) HandleWithOutput(jobType int32, build func(state.Reader) OutputHandler)
- func (r *Runner) Handles(jobType int32) bool
- func (r *Runner) PollOnce() (int, error)
- func (r *Runner) Submit(outcomes []Outcome)
- func (r *Runner) Unhandle(jobType int32)
- func (r *Runner) Work(jobs []Job, reader state.Reader) []Outcome
Constants ¶
const DefaultConcurrency = 16
DefaultConcurrency is how many handlers a round runs at once when nothing says otherwise. It is well above serial, so a burst still drains quickly, and well below the point where the sockets and memory of one round are a problem.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CompletingHandler ¶
type CompletingHandler func(Job) (Completion, error)
CompletingHandler does a job's work and returns its full Completion — outputs plus any decision evaluation. It is the widest handler shape; Handler and OutputHandler are the output-less and decision-less special cases. Returning an error fails the job (retry, then an incident) exactly as for the others.
type Completion ¶
type Completion struct {
Outputs []model.VariableValue
Decision *model.DecisionEvaluationValue
}
Completion is everything a worker hands back when a job succeeds: the output variables to write into the instance and, for a business rule task, the decision evaluation to retain for debugging (ADR-0066). Decision is nil for workers that do not produce one.
type Engine ¶
type Engine interface {
RunUntilIdle() error
CompleteJob(jobKey uint64, outputs ...model.VariableValue)
CompleteJobWithDecision(jobKey uint64, decision *model.DecisionEvaluationValue, outputs ...model.VariableValue)
FailJob(jobKey uint64, retries int32, message string, backoff int64)
}
Engine is the slice of the processor the runner drives: process queued commands, accept job completions with their output variables and (for a decision) the evaluation to retain, and accept job failures (which retry or raise an incident, ADR-0061).
type Handler ¶
Handler does a job's work with no output. Returning nil completes the job; returning an error fails it (retry while retries remain, then an incident, ADR-0061).
type Job ¶
type Job struct {
Key uint64
Type int32 // interned job-type index
ProcessInstanceKey uint64
ElementInstanceKey uint64
Retries int32
}
Job is the unit of work handed to a worker.
type Outcome ¶ added in v0.3.0
type Outcome struct {
Job Job
Completion Completion
Err error
}
Outcome is what one job's handler produced: the job, what it completed with, and the error that failed it. Exactly one of Completion and Err is meaningful.
type OutputHandler ¶
type OutputHandler func(Job) ([]model.VariableValue, error)
OutputHandler does a job's work and returns the variables to write back into the job's process instance on completion (nil for none) — e.g. a business rule task's decision result. As for Handler, returning an error fails the job (retry, then an incident).
type Runner ¶
type Runner struct {
// Concurrency bounds how many handlers a round runs at once.
//
// Serial dispatch had one virtue: a backlog could not become a thundering herd.
// Running handlers concurrently (ADR-0157 step 6) is what ended the amplification
// of a burst, but unbounded it would trade one failure for another — a thousand
// parked jobs becoming a thousand simultaneous outbound calls, exhausting file
// descriptors here and hammering whatever is on the other end. The cap keeps the
// throughput and drops the herd.
Concurrency int
// contains filtered or unexported fields
}
Runner dispatches activatable jobs to registered handlers.
A handler is registered as a *factory* rather than a handler, because handlers read state and no longer run on the goroutine that owns it: each round binds its handlers to that round's state.Reader, a consistent read view taken while the run loop was held (ADR-0157 step 6). The connector packages are unchanged by this — their Handler constructors already take a reader, so the factory is one closure at the registration site.
func (*Runner) Claim ¶ added in v0.3.0
Claim collects every activatable job of a registered type. It reads state, so it runs on the goroutine that owns it — the run loop — and it does nothing slow: the work itself happens in Runner.Work, off the loop.
func (*Runner) Drive ¶
Drive runs the engine and dispatches jobs alternately until the system is idle: no pending commands and no activatable jobs for registered types. It is the in-process equivalent of workers streaming alongside a running processor.
func (*Runner) Handle ¶
Handle registers an output-less worker for a job type. The type is the interned index the compiler assigned (cross-process, globally consistent job-type interning is a later concern).
func (*Runner) HandleCompleting ¶
func (r *Runner) HandleCompleting(jobType int32, build func(state.Reader) CompletingHandler)
HandleCompleting registers a worker whose completion carries both output variables and a decision evaluation to retain (the DMN worker, ADR-0066). Same dispatch as the others; its Completion rides along on the CompleteJob command.
func (*Runner) HandleWithOutput ¶
func (r *Runner) HandleWithOutput(jobType int32, build func(state.Reader) OutputHandler)
HandleWithOutput registers a worker whose completion writes output variables back into the instance (e.g. a service-task worker that returns variables). Same dispatch as Handle; the only difference is that its returned variables ride along on the CompleteJob command.
func (*Runner) Handles ¶ added in v0.3.0
Handles reports whether an in-process worker is registered for a job type.
It is what keeps one job from being worked twice: an external worker leasing by type must be refused a type this runner is already draining, because nothing else separates them — the runner does not lease, it simply dispatches whatever is activatable (ADR-0157). Relocating a kind to an external worker therefore means not registering its handler here.
func (*Runner) PollOnce ¶
PollOnce claims, works and submits one round on the calling goroutine. It is the simple synchronous form — used by Runner.Drive and by callers that own the loop themselves; a server that must keep the loop free drives the three steps itself.
func (*Runner) Submit ¶ added in v0.3.0
Submit turns a round's outcomes into commands. It mutates engine state, so like Runner.Claim it runs on the run loop.
A failure is routed into the incident model (ADR-0061) rather than returned: one job that cannot complete — a business rule task whose decision model is not deployed, say — must not poison the run, or every later deploy and completion that drives jobs would fail with it. FailJob retries while retries remain, then parks the token on an incident. The in-process runner retries immediately (backoff 0); a worker-supplied backoff arrives through the HTTP fail endpoint instead (ADR-0111).
func (*Runner) Unhandle ¶ added in v0.3.0
Unhandle removes the in-process worker for a job type, so its jobs park for an external one instead (ADR-0168).
It exists as a removal rather than a condition at each registration site because the registrations are spread across the server — a managed connector kind registers through its own descriptor, the script languages through their loop, the rest inline — and a switch that has to be remembered at ten places is a switch that will be missed at one.
func (*Runner) Work ¶ added in v0.3.0
Work runs the handlers for claimed jobs and returns what each produced. This is the part that must NOT hold the run loop: a handler makes the outbound call a connector exists for, and holding the single writer for its duration is the stall ADR-0155 measured and ADR-0157 step 6 removes.
Handlers run concurrently, one goroutine per job. That is also what ends the amplification of the serial runner: a burst of parked jobs against a dead host used to cost one timeout after another, and now costs the slowest of them.
reader is the round's consistent read view. Handlers are built against it here, so each sees one coherent state rather than whatever the writer has reached.
Source Files