Documentation
¶
Overview ¶
Package scheduler is the daemon's job-scheduling engine: it drains each held pipeline's durable queue in this process — workers, the --max-runs slots, the circuit breaker, pause, abort, the shutdown grace — and holds the workspace provider each pipeline's runs materialize in.
Every trigger lands in that one queue — a poll, a webhook, a browser click, an approval release — and is drained by the same pipeline.RunJob every other path calls. There is deliberately no second execution route: a job started from a browser must be the same job, with the same caching, hooks, serial groups, and recording, as one started from a terminal. internal/web reaches this package only through its Runner interface; internal/cli wires the two.
Index ¶
- Variables
- func PrepareQueue(ctx context.Context, target *Pipeline)
- func SyncQueueLimits(ctx context.Context, target *Pipeline)
- type ExportHooks
- type ExportServer
- type ExportWho
- type LocalRunner
- func (r *LocalRunner) Abort(target *Pipeline, runID string) bool
- func (r *LocalRunner) AbortQueued(ctx context.Context, target *Pipeline, jobName string) (bool, error)
- func (r *LocalRunner) Close()
- func (r *LocalRunner) Drain(ctx context.Context, pipelines []*Pipeline)
- func (r *LocalRunner) DrainOne(ctx context.Context, target *Pipeline) bool
- func (r *LocalRunner) DrainPipeline(ctx context.Context, target *Pipeline)
- func (r *LocalRunner) Enqueue(ctx context.Context, target *Pipeline, jobName, reason string, force bool) (int64, error)
- func (r *LocalRunner) EnqueueRerun(ctx context.Context, target *Pipeline, jobName, runID string) error
- func (r *LocalRunner) LimitRuns(n int)
- func (r *LocalRunner) NameAnswerDB(db string)
- func (r *LocalRunner) RemoveProvider(slug string)
- func (r *LocalRunner) RunCapacity() (int, int)
- func (r *LocalRunner) Running(slug string) []RunningBuild
- func (r *LocalRunner) ServeExport(ctx context.Context, target *Pipeline, export config.MCPExport, ...) (*ExportServer, error)
- func (r *LocalRunner) SetProvider(slug string, provider workspace.Provider)
- func (r *LocalRunner) StopWith(process context.Context)
- type Pipeline
- type RunningBuild
Constants ¶
This section is empty.
Variables ¶
var ErrRateLimited = pipeline.ErrRateLimited //nolint:gochecknoglobals // a re-export, not state
ErrRateLimited is a call past an export's rate:.
var WithExportWho = pipeline.WithExportWho //nolint:gochecknoglobals // a re-export, not state
WithExportWho carries who is calling to the rules checking the call.
Functions ¶
func PrepareQueue ¶
PrepareQueue is the startup recovery `steps web` does, mirrored here so recovery happens the same way regardless of which front end drains the queue. It is recovery ONLY: what a configuration says about who may run is SyncQueueLimits' job, because that has to happen again on every reload while this must happen exactly once.
The caller runs this BEFORE starting any drain or poll goroutine, which is a requirement rather than a convention: ResetStaleRunning is three statements with no transaction around them, and an enqueue landing between two of them leaves a row no later poll re-queues.
It recovers unconditionally, which is a statement about deployment rather than a shortcut: recovery reads every `running` row as an abandoned leftover, and `steps web` is the only daemon there is — so a row it finds running at startup belongs to a process that is gone. Two of them against one state file would each undo the other, which is the deployment mistake the one-process-per-database rule names. It was once answered by a file lock this process raced for, and then by --no-watch, which went with the separate watcher; see store.ResetStaleRunning for why the lock went.
func SyncQueueLimits ¶
SyncQueueLimits mirrors the served configuration's admission rules into the tables ClaimNextJob reads.
NOT part of PrepareQueue, though it once was: it belongs to ADOPTING a configuration, which happens at startup and at every reload, while recovery belongs to starting the process. ResetStaleRunning reads every `running` row as an abandoned leftover, which is true once, at startup, and false while this process is mid-build. Everything a swap changes about who may run — a job joining a serial group, a max_in_flight raised or removed — lives in SQL rather than in the Config, so a configuration adopted without this serves pages promising a `serial:` the queue goes on ignoring.
Types ¶
type ExportHooks ¶
type ExportHooks = pipeline.ExportHooks
ExportHooks are what the mounting server decides about each call.
type ExportServer ¶
type ExportServer = pipeline.ExportServer
ExportServer is one MCP export, served; see pipeline.ExportServer.
type LocalRunner ¶
type LocalRunner struct {
// contains filtered or unexported fields
}
LocalRunner drains each pipeline's queue in this process. It is the Runner the `steps web` command installs; a read-only server has none.
func NewLocalRunner ¶
func NewLocalRunner( initial map[string]workspace.Provider, base pipeline.Run, concurrent int, ) *LocalRunner
NewLocalRunner builds a runner over per-pipeline workspace providers, keyed by pipeline slug. concurrent below 1 means one job at a time.
func (*LocalRunner) Abort ¶
func (r *LocalRunner) Abort(target *Pipeline, runID string) bool
Abort only cancels: the run still unwinds through its on_abort and ensure hooks, and gives back its serial slot when it actually ends.
func (*LocalRunner) AbortQueued ¶
func (r *LocalRunner) AbortQueued(ctx context.Context, target *Pipeline, jobName string) (bool, error)
AbortQueued drops a job's queued run before it starts, and the force it was queued with.
func (*LocalRunner) Close ¶
func (r *LocalRunner) Close()
Close retires every workspace this runner holds. Only a provider's own Close removes the tree it created, so a shutdown that closed the stores alone left a steps-* directory per pipeline behind with nothing to reap it.
func (*LocalRunner) Drain ¶
func (r *LocalRunner) Drain(ctx context.Context, pipelines []*Pipeline)
Drain runs each pipeline's queue until ctx is canceled. One goroutine per pipeline: they have separate databases, so they never contend, and a slow job in one pipeline must not stall another's queue.
func (*LocalRunner) DrainOne ¶
func (r *LocalRunner) DrainOne(ctx context.Context, target *Pipeline) bool
DrainOne claims at most one queued job and runs it. It reports whether it ran anything, so the caller knows whether to back off.
func (*LocalRunner) DrainPipeline ¶
func (r *LocalRunner) DrainPipeline(ctx context.Context, target *Pipeline)
DrainPipeline runs one pipeline's queue with `concurrent` workers.
Workers rather than one loop because a daemon that is mid-build stops noticing everything else: the queue this drains is also where a browser trigger and an approval release land, and one long agent step used to make all of them wait. What it does NOT do is loosen the pipeline's own limits — ClaimNextJob admits in a single statement against serial: and max_in_flight, so extra workers find nothing to claim rather than running what the pipeline forbade.
func (*LocalRunner) Enqueue ¶
func (r *LocalRunner) Enqueue(ctx context.Context, target *Pipeline, jobName, reason string, force bool) (int64, error)
Enqueue puts a person's trigger on the pipeline's queue.
func (*LocalRunner) EnqueueRerun ¶
func (r *LocalRunner) EnqueueRerun(ctx context.Context, target *Pipeline, jobName, runID string) error
EnqueueRerun puts a person's retry of a run on the pipeline's queue.
func (*LocalRunner) LimitRuns ¶
func (r *LocalRunner) LimitRuns(n int)
LimitRuns is --max-runs. Call it before draining; n <= 0 is no limit.
func (*LocalRunner) NameAnswerDB ¶
func (r *LocalRunner) NameAnswerDB(db string)
NameAnswerDB is the --db a parked step's printed answer command needs. Call it before draining.
func (*LocalRunner) RemoveProvider ¶
func (r *LocalRunner) RemoveProvider(slug string)
RemoveProvider retires a destroyed pipeline's workspace, once nothing is running in it.
func (*LocalRunner) RunCapacity ¶
func (r *LocalRunner) RunCapacity() (int, int)
RunCapacity is the --max-runs limit (0 when unlimited) and how many builds hold a slot.
func (*LocalRunner) Running ¶
func (r *LocalRunner) Running(slug string) []RunningBuild
Running is every run of one pipeline in flight, by job and then run id.
func (*LocalRunner) ServeExport ¶
func (r *LocalRunner) ServeExport(ctx context.Context, target *Pipeline, export config.MCPExport, hooks ExportHooks) (*ExportServer, error)
ServeExport builds one of target's exports under its current configuration. The caller owns the result and closes it when the configuration moves on.
func (*LocalRunner) SetProvider ¶
func (r *LocalRunner) SetProvider(slug string, provider workspace.Provider)
SetProvider installs the workspace a pipeline's runs materialize in, retiring whatever it replaces once the runs holding it finish.
func (*LocalRunner) StopWith ¶
func (r *LocalRunner) StopWith(process context.Context)
StopWith names the context whose end is the process going down, as opposed to one pipeline's drain being stopped. Call it before draining.
type Pipeline ¶
type Pipeline struct {
// Slug is the name `steps pipeline set` was given: the route, the store's pipelines.name and the Config's name are ONE identity, not three kept in agreement.
Slug string
Store store.Store
// Bus carries live run events for runs this process itself executes.
// Runs started by a separate `steps run` land in the store but not on
// this bus, which is why every live view falls back to replaying the
// stored events rather than assuming the bus saw everything.
Bus *events.Bus
// contains filtered or unexported fields
}
Pipeline is one held pipeline as the drain sees it: its name, its store handle, the bus its runs publish on, and the configuration being served. Two held pipelines may share a state FILE (see --db), but never a store handle: each one is scoped to its own pipeline row, which is what keeps their histories and caches apart.
func NewPipeline ¶
NewPipeline is a constructor rather than a struct literal because a literal would leave cfg nil for whatever reads it before the caller filled it in.
type RunningBuild ¶
RunningBuild is a run in flight: what a refusal names so a person can find it and abort it.
Source Files