Documentation
¶
Overview ¶
Package scheduler provides task run scheduling.
WorkerRunner abstracts how a worker is started for a task run (local process or k8s Job).
Scheduler polls for pending task runs and runs the worker via a WorkerRunner (local process or k8s Job). It does not perform run execution; the worker process calls runtime.RunTask.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AuditRetainer ¶
type AuditRetainer struct {
// contains filtered or unexported fields
}
AuditRetainer expires audit events older than the configured window.
It is the only thing in the system that removes a governance record, and it leaves one behind when it does: every sweep that deleted anything writes an AuditEventsPruned event naming the cutoff and the count. That event is what distinguishes a trail shortened by policy from a trail somebody truncated — without it, the two look identical to a reader, and the wrong one of them is the kind of thing an audit trail exists to make visible.
func NewAuditRetainer ¶
func NewAuditRetainer(store coreaudit.PruneStore, writer coreaudit.Writer, retentionDays int, interval time.Duration) *AuditRetainer
NewAuditRetainer returns a retainer, or nil when nothing should be removed.
A nil store or a window of zero returns nil: keeping every event is the default, and a deployment that has not chosen a retention policy must not get one by accident. Use 0 for the default interval.
func (*AuditRetainer) Start ¶
func (a *AuditRetainer) Start()
Start launches the sweep loop. Calling it on a nil retainer is a no-op, so a deployment that keeps everything needs no branch at the call site.
func (*AuditRetainer) Stop ¶
func (a *AuditRetainer) Stop()
Stop signals the loop to exit and blocks until it has finished.
type CredentialCleaner ¶
type CredentialCleaner struct {
// contains filtered or unexported fields
}
CredentialCleaner periodically removes expired login codes and refresh tokens.
Neither table is correctness-critical to sweep — expiry is enforced on read, not by absence — so a failed sweep is logged and retried on the next tick rather than stopping the server.
func NewCredentialCleaner ¶
func NewCredentialCleaner(store ExpiredCredentialStore, interval time.Duration) *CredentialCleaner
NewCredentialCleaner returns a cleaner for store. A nil store returns nil, so a deployment without a database simply has nothing to sweep. Use 0 for the default interval.
func (*CredentialCleaner) Start ¶
func (c *CredentialCleaner) Start()
Start launches the sweep loop in a background goroutine. Calling it on a nil cleaner is a no-op, so the caller does not need to check.
func (*CredentialCleaner) Stop ¶
func (c *CredentialCleaner) Stop()
Stop signals the loop to exit and blocks until it has finished.
type ExpiredCredentialStore ¶
type ExpiredCredentialStore interface {
DeleteExpiredLoginCodes(ctx context.Context, before time.Time) (int64, error)
DeleteExpiredRefreshTokens(ctx context.Context, before time.Time) (int64, error)
}
ExpiredCredentialStore deletes credential rows that can no longer be redeemed. Both methods return how many rows they removed.
type LocalRunner ¶
type LocalRunner struct {
// contains filtered or unexported fields
}
LocalRunner runs the worker binary as a local process (blocks until exit).
func NewLocalRunner ¶
func NewLocalRunner(workerPath string, env []string, runTokenEnvKey string, stopGrace time.Duration) *LocalRunner
NewLocalRunner returns a runner that exec's the worker binary with --task-run-id.
env is the child's complete environment. It is supplied rather than inherited because a worker runs model-chosen shell commands, and the server's own environment holds credentials — the JWT signing secret, the database password — that a worker never reads. Deciding what a worker may hold belongs to the layer that assembles the process, not to the one that spawns it.
runTokenEnvKey names the variable a run token is delivered in. It arrives with the environment for the same reason: this package cannot import config, which owns every environment variable name. Empty means the deployment delivers no run token.
stopGrace is how long a worker gets to stop in order once its dispatch is cancelled. Zero uses defaultWorkerStopGrace.
func (*LocalRunner) Run ¶
func (r *LocalRunner) Run(ctx context.Context, run coretask.Run, runToken string) (workerType string, k8sJobName *string, k8sJobCreatedAt *time.Time, err error)
Run executes the worker process; on success returns ("local_process", nil, nil, nil). On failure returns error.
The run token is placed in the child's environment rather than on its command line, where every process on the machine could read it.
Cancelling ctx asks the worker to stop rather than killing it, so the run it is executing can report what it produced. It is killed if it does not manage that inside stopGrace — see docs/design/graceful-shutdown.md §6.1.
type MintRunToken ¶
MintRunToken signs the credential a worker presents for one run's managed inference calls.
It only signs. The scheduler builds the claims from the run and its task, so a worker's identity comes from what the server already knows about the run rather than from anything the worker or its model could influence.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler polls the task run store for PENDING runs and runs the worker via the configured runner.
func NewScheduler ¶
func NewScheduler(taskRunStore coretask.RunStore, runner WorkerRunner, mint MintRunToken) (*Scheduler, error)
NewScheduler creates a Scheduler that polls for pending task runs and runs the worker via the given runner. Call Start() to begin polling.
mint may be nil, which is every deployment that has not enabled managed worker inference: its workers reach a provider directly and have nothing to authenticate to.
func NewSchedulerWithPollInterval ¶
func NewSchedulerWithPollInterval(taskRunStore coretask.RunStore, runner WorkerRunner, mint MintRunToken, pollInterval time.Duration) (*Scheduler, error)
NewSchedulerWithPollInterval is like NewScheduler but allows setting the poll interval (e.g. for tests). Use 0 for default.
func (*Scheduler) Start ¶
func (s *Scheduler) Start()
Start launches the poll loop in a background goroutine.
func (*Scheduler) Stop ¶
Stop stops claiming runs, then gives the dispatches already in flight until ctx expires to finish. It reports whether they all finished.
Two phases because they need different answers. Claiming must stop at once — a run started by a process that is going away is a run nobody will report on. A dispatch already made is the opposite: in local_process mode it *is* the running agent, and the worker it holds needs the server's API alive long enough to say what it produced. Cancelling the dispatch context is what asks that worker to stop; see docs/design/graceful-shutdown.md §6.1.
func (*Scheduler) WithUserStore ¶
func (s *Scheduler) WithUserStore(users coreidentity.UserStore) *Scheduler
WithUserStore lets the scheduler refuse work for a disabled account.
It is a setter rather than a constructor parameter because the check is optional: a deployment with no user store schedules exactly as it did before, and the three existing call sites do not have to learn about accounts to keep compiling.
type StaleRunReaper ¶
type StaleRunReaper struct {
// contains filtered or unexported fields
}
StaleRunReaper finishes runs that nothing else will finish.
Two cases, one loop. A run whose worker never reported an outcome stays SCHEDULED or RUNNING forever, because the only thing that moves it is the worker itself: that was already possible when a pod was evicted or a process was killed, and run tokens add one more way, since a run outliving its token can no longer report anything. A run someone canceled has the same problem for the same reason — the cancel is a request its worker honors, and a worker that is gone honors nothing. Either way the run is over and the record should say so.
See docs/design/worker-run-token.md.
func NewStaleRunReaper ¶
func NewStaleRunReaper(runs StaleRunStore, timeout, interval time.Duration) *StaleRunReaper
NewStaleRunReaper returns a reaper for runs, or nil when there is no store to sweep — so a caller does not need to check before starting it. Zero values use the defaults.
func (*StaleRunReaper) Start ¶
func (c *StaleRunReaper) Start()
Start launches the sweep loop. Calling it on a nil reaper is a no-op.
func (*StaleRunReaper) Stop ¶
func (c *StaleRunReaper) Stop()
Stop signals the loop to exit and blocks until it has finished.
func (*StaleRunReaper) Sweep ¶
func (c *StaleRunReaper) Sweep(ctx context.Context, now time.Time)
Sweep finishes every run the reaper is responsible for: those dispatched longer ago than the timeout, and those asked to stop longer ago than the grace. It is exported so a test can drive one pass without a clock.
type StaleRunStore ¶
type StaleRunStore interface {
ListStaleTaskRuns(ctx context.Context, cutoff time.Time, limit int) ([]coretask.Run, error)
ListCancelRequestedTaskRuns(ctx context.Context, cutoff time.Time, limit int) ([]coretask.Run, error)
TransitionTaskRun(ctx context.Context, in coretask.TransitionRunInput) (bool, error)
}
StaleRunStore is the narrow store surface the reaper needs.
type WorkerRunner ¶
type WorkerRunner interface {
Run(ctx context.Context, run coretask.Run, runToken string) (workerType string, k8sJobName *string, k8sJobCreatedAt *time.Time, err error)
}
WorkerRunner starts a worker for a task run. On success returns worker info to persist; on failure returns an error (caller should revert run to PENDING).
runToken is this run's credential for the managed LLM gateway, or "" when the deployment issues none. It is passed per run rather than held by the runner because it names one run and expires: a runner built once at startup has nowhere to put a value that changes on every dispatch. See docs/design/worker-run-token.md.
Source Files