Documentation
¶
Overview ¶
Package jobagent supervises a single Mailer job in-process and exposes its lifecycle over HTTP. One agent owns one *mailer.StreamExecutionEnv: it runs Execute in a goroutine, tracks lifecycle transitions, serves /state, /describe, /metrics, /healthz, and translates POST /cancel into a graceful context cancellation.
This is the control contract every runner embeds (see the mailer-runner binary and the job-orchestration plan, phase P1). It adds no container or scheduling concerns — those live in the separate control plane.
Index ¶
- type Agent
- func (a *Agent) Cancel()
- func (a *Agent) DescribeJSON() string
- func (a *Agent) Handler() http.Handler
- func (a *Agent) RequestSavepoint(label string) bool
- func (a *Agent) Run(ctx context.Context) error
- func (a *Agent) SavepointRequest() (label string, requested bool)
- func (a *Agent) Serve(ctx context.Context, addr string) error
- func (a *Agent) State() State
- type Phase
- type State
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Agent ¶
type Agent struct {
// contains filtered or unexported fields
}
Agent supervises one job. It is safe for concurrent use: State/Cancel may be called from HTTP handlers while Run executes on another goroutine.
func New ¶
func New(env *mailer.StreamExecutionEnv) *Agent
New creates an agent that will run the given configured environment. The env must already have its source, sink, and any operators wired (e.g. from the workflow compiler); the agent does not build pipelines.
func (*Agent) Cancel ¶
func (a *Agent) Cancel()
Cancel requests a graceful shutdown. It is idempotent and a no-op once the job has reached a terminal phase. The pipeline drains in-flight records and takes a final checkpoint before Run returns.
func (*Agent) DescribeJSON ¶
DescribeJSON returns the pipeline topology as indented JSON.
func (*Agent) Handler ¶
Handler returns the agent's HTTP surface. Routes use Go 1.22+ method patterns so wrong-method requests get 405 automatically.
GET /healthz liveness + current phase GET /state lifecycle snapshot (JSON State) GET /describe pipeline topology (env.DescribeJSON) GET /metrics Prometheus exposition POST /cancel request graceful shutdown POST /savepoint stop-with-savepoint (?label=<name>)
func (*Agent) RequestSavepoint ¶
RequestSavepoint asks for a stop-with-savepoint: the job drains and writes its final checkpoint (like Cancel), and the label is recorded so the runner can promote that checkpoint to a named savepoint once Run returns. It is a no-op once the job is terminal; returns whether the request was accepted.
func (*Agent) Run ¶
Run executes the job to completion, blocking until it finishes, fails, or a cancel drains it. It registers the checkpoint listener, moves the phase to Running, then calls Execute. The returned error is Execute's error (nil on a clean finish or a completed cancel). Run is intended to be called once.
func (*Agent) SavepointRequest ¶
SavepointRequest reports whether a savepoint was requested and its label. The runner reads this after Run returns to decide whether to promote the final checkpoint.
type Phase ¶
type Phase string
Phase is the coarse lifecycle state of a supervised job. It only ever moves forward: Starting → Running → (Cancelling) → Finished | Failed.
const ( // PhaseStarting is set from construction until Execute is entered. PhaseStarting Phase = "starting" // PhaseRunning means Execute is in progress. PhaseRunning Phase = "running" // PhaseCancelling means a cancel was requested and the pipeline is // draining; it still ends in Finished (clean drain) or Failed. PhaseCancelling Phase = "cancelling" // PhaseFinished is a clean terminal state: the source drained or a // cancel completed its graceful shutdown without error. PhaseFinished Phase = "finished" // PhaseFailed is a terminal state: Execute returned a non-cancel error. PhaseFailed Phase = "failed" )
type State ¶
type State struct {
Phase Phase `json:"phase"`
StartedAt time.Time `json:"startedAt"`
Uptime string `json:"uptime"`
RecordsIn int64 `json:"recordsIn"`
RecordsOut int64 `json:"recordsOut"`
// CurrentCheckpointID is the ID of the most recently completed
// checkpoint, empty until the first one completes.
CurrentCheckpointID string `json:"currentCheckpointId,omitempty"`
// LastCheckpointAt is when the agent observed that checkpoint
// complete (nil until the first one).
LastCheckpointAt *time.Time `json:"lastCheckpointAt,omitempty"`
// LastError is the terminal error message when Phase is Failed.
LastError string `json:"lastError,omitempty"`
}
State is a point-in-time snapshot of a job, returned by GET /state. Record counts are read from the process Prometheus counters at snapshot time (one job per process, so the cumulative counters are the job's totals); checkpoint fields are populated by the engine checkpoint listener.
Source Files