worker/worker.go (package worker)

Source: sdks/go/worker/worker.go. Back to the Go SDK reference index.

note

This page is generated by tools/docs/gen_sdk_reference.py from the SDK sources and the OpenAPI text folders. Do not edit it by hand: change the source and regenerate.

Model Job

Job is one unit of work.

Fields:

Name	Type	Required	Description
ID	string	yes	Identifier of the job, sent as worker in the default heartbeat.
RunID	string	yes	RunID, when set, receives heartbeats on POST /v1/runs/{id}/heartbeat.
Payload	json.RawMessage	yes	Raw JSON input of the job, read by the Handler.
Attempt	int	yes	Attempt is set by the Source (1 for the first delivery).

Interface Source

Source of a chart: exactly one of chart_path and oci.

Source.Ack

Ack(ctx context.Context, job *Job, result any) error

Ack reports that a job succeeded, with the result of the Handler.

Name	Type	Default	Required	Description
ctx	context.Context		yes	Context of the call: cancellation and deadline.
job	*Job		yes	Job that succeeded.
result	any		yes	Value returned by the Handler.

Returns: error. Error of the call (a typed error of the SDK), nil on success.

Source.Fail

Fail(ctx context.Context, job *Job, cause error) error

Fail reports that a job failed, with the error of the Handler.

Name	Type	Default	Required	Description
ctx	context.Context		yes	Context of the call: cancellation and deadline.
job	*Job		yes	Job that failed.
cause	error		yes	Error returned by the Handler, or the panic error.

Returns: error. Error of the call (a typed error of the SDK), nil on success.

Source.Next

Next(ctx context.Context) (*Job, _ error)

Next returns the next job, or (nil, nil) when none is available right now.

Name	Type	Default	Required	Description
ctx	context.Context) (*Job		yes	Context of the call: cancellation and deadline.
_	error		yes	Context of the worker loop, canceled at shutdown.

Type alias Handler

Handler runs a job; its context is canceled on kill switch or forced shutdown.

Definition: func(ctx context.Context, job *Job) (any, error).

Model Config

Config configures a Worker.

Fields:

Name	Type	Required	Description
Source	Source	yes	Source of a chart: exactly one of chart_path and oci.
Handler	Handler	yes	Handler runs a job; its context is canceled on kill switch or forced shutdown.
Concurrency	int	yes	default 1
Poll	time.Duration	yes	wait when the source is empty, default 1s
HeartbeatInterval	time.Duration	yes	Heartbeat interval while a job runs, default 10s.
Client	*eai.Client	no	Client sends the default heartbeats (when Heartbeat is nil) for jobs with a RunID.
Heartbeat	func(ctx context.Context, job *Job) error	yes	Heartbeat overrides the default heartbeat.
KillSwitch	func(ctx context.Context) (bool, error)	yes	KillSwitch is polled every KillInterval (default 5s); true stops the worker immediately and cancels running jobs.
KillInterval	time.Duration	yes	Polling interval of KillSwitch, 5 seconds by default.
ShutdownTimeout	time.Duration	yes	ShutdownTimeout bounds the wait for running jobs after the context is canceled (default 30s), after which their contexts are canceled.
Logger	*slog.Logger	no	Logger of the source, heartbeat and kill switch warnings; slog.Default() when nil.

Model Stats

Stats are the counters of a Worker.

Fields:

Name	Type	Required	Description
Processed	int64	yes	Jobs whose Handler succeeded.
Failed	int64	yes	Jobs whose Handler returned an error or panicked.
Heartbeats	int64	yes	Heartbeats sent without error.

Class Worker

Worker runs the loop.

Worker.Kill

Kill()

Kill stops the worker now: no new job, running jobs are canceled.

This element takes no parameter.

Worker.Run

Run(ctx context.Context) error

Run blocks until ctx is canceled (graceful: running jobs finish within ShutdownTimeout, returns nil) or the kill switch trips (returns ErrKilled).

Name	Type	Default	Required	Description
ctx	context.Context		yes	Context of the call: cancellation and deadline.

Returns: error. Error of the call (a typed error of the SDK), nil on success.

Raises: error.

Worker.Stats

Stats() Stats

Stats returns the counters.

This element takes no parameter.

Returns: Stats. A Stats: Stats are the counters of a Worker.

Class ChanSource

ChanSource is an in-process Source fed by a channel; acks and failures are reported on callbacks.

Fields:

Name	Type	Required	Description
Jobs	<-chan *Job	yes	Channel the jobs are read from, without blocking.
OnAck	func(job *Job, result any)	yes	Called by Ack with the job and its result; may be nil.
OnFail	func(job *Job, cause error)	yes	Called by Fail with the job and the error; may be nil.

ChanSource.Ack

Ack(ctx context.Context, job *Job, result any) error

Ack reports success.

Name	Type	Default	Required	Description
ctx	context.Context		yes	Context of the call: cancellation and deadline.
job	*Job		yes	Job that succeeded, passed to OnAck.
result	any		yes	Value returned by the Handler, passed to OnAck.

Returns: error. Error of the call (a typed error of the SDK), nil on success.

Raises: error.

ChanSource.Fail

Fail(ctx context.Context, job *Job, cause error) error

Fail reports a failure.

Name	Type	Default	Required	Description
ctx	context.Context		yes	Context of the call: cancellation and deadline.
job	*Job		yes	Job that failed, passed to OnFail.
cause	error		yes	Error returned by the Handler (or a panic), passed to OnFail.

Returns: error. Error of the call (a typed error of the SDK), nil on success.

Raises: error.

ChanSource.Next

Next(ctx context.Context) (*Job, error)

Next returns the next queued job or nil when the channel is empty.

Name	Type	Default	Required	Description
ctx	context.Context		yes	Context of the call: cancellation and deadline.

Returns: (*Job, error). A Job: Job is one unit of work. The error is a typed error of the SDK, nil on success.

Raises: error.

Functions

New

New(cfg Config) (*Worker, error)

New validates the configuration.

Name	Type	Default	Required	Description
cfg	Config		yes	Configuration of the worker; Source and Handler are required, zero durations take their defaults.

Returns: (*Worker, error). A Worker: Worker runs the loop. The error is a typed error of the SDK, nil on success.

Raises: error.

AgentKillSwitch

AgentKillSwitch(client *eai.Client, agentID string) func(ctx context.Context) (bool, error)

Get an agent. Returns the agent, the manifest of its current version and runs, the number of its runs per state. Needs the ReBAC relation viewer on the agent. 404 agent_not_found.

Route: /v1/agents/{agent_id} (GET).

Name	Type	Default	Required	Description
client	*eai.Client		yes	Parent client.
agentID	string		yes	Identifier (UUID) or name of the agent.

Returns: func(ctx context.Context) (bool, error). Decoded JSON answer of the operation "Get an agent". The error is a typed error of the SDK, nil on success.

Raises: error, the typed errors of the transport (Errors).

RunKillSwitch

RunKillSwitch(client *eai.Client, runID string) func(ctx context.Context) (bool, error)

Send a run heartbeat. Records that the runtime is alive and tells it whether to go on: action is continue, or stop with a reason (run_cancelled, run_terminal, run_suspended, agent_killed, agent_suspended). The answer also gives stale_after_seconds. A terminal run is not updated. Needs tenant_admin, developer or agent. Audited only for the first heartbeat of a run and when the runtime is told to stop.

Route: /v1/runs/{run_id}/heartbeat (POST).

Name	Type	Default	Required	Description
client	*eai.Client		yes	Parent client.
runID	string		yes	Identifier (UUID) of the run.

Returns: func(ctx context.Context) (bool, error). Decoded JSON answer of the operation "Send a run heartbeat". The error is a typed error of the SDK, nil on success.

Raises: error, the typed errors of the transport (Errors).