scheduledtask

Documentation

Overview

Package scheduledtask runs operator-defined jobs on a cron schedule, in the running container (exec mode) or in a fresh one-off container built from the app's own spec (fresh mode), mirroring internal/backup's Runner/Scheduler split: Runner (this file) does one run and records its outcome, Scheduler (scheduler.go) decides when a run is due. The same Runner backs the cron loop, "run now" and the API trigger, so there is one place that executes a command and records what happened.

Index

• func NewRunID() (string, error)
• func Occurrences(sched *cronexpr.Schedule, loc *time.Location, after, upTo time.Time, limit int) (out []time.Time, truncated bool)
• type AppStore
• type HistoryStore
• type Limits
  • func DefaultLimits() Limits
  • func LimitsFromEnv(lookup func(string) (string, bool)) Limits
• type NodeRuntimeResolver
• type Recoverer
• type RunStore
• type Runner
  • func (r *Runner) Cleanup(ctx context.Context, runs []store.ScheduledTaskRun)
  • func (r *Runner) ContainerName(service, runID, taskID string) string
  • func (r *Runner) Run(ctx context.Context, task store.ScheduledTask) error
  • func (r *Runner) RunTrigger(ctx context.Context, task store.ScheduledTask, trig Trigger) error
  • func (r *Runner) Start(ctx context.Context, task store.ScheduledTask, trigger string) (string, error)
• type ScheduleStore
• type Scheduler
  • func NewScheduler(st ScheduleStore, runner TaskRunner, logger *slog.Logger) *Scheduler
  • func (s *Scheduler) Recover(ctx context.Context, rec Recoverer) error
  • func (s *Scheduler) Run(ctx context.Context, interval time.Duration) error
  • func (s *Scheduler) Tick(ctx context.Context) error
  • func (s *Scheduler) Wait()
• type SpecBuilder
• type TaskRunner
• type Trigger

Functions

func NewRunID

func NewRunID() (string, error)

NewRunID mints a run ID (jr_ plus 9 random bytes, URL-safe base64).

func Occurrences

func Occurrences(sched *cronexpr.Schedule, loc *time.Location, after, upTo time.Time, limit int) (out []time.Time, truncated bool)

Occurrences returns the cron occurrences in (after, upTo], evaluated on loc's wall clock, oldest first. A wall time skipped by a DST jump runs once at the shifted time; one repeated by a fall-back runs once.

Types

type AppStore

type AppStore interface {
	GetDesiredService(ctx context.Context, name string) (*store.DesiredService, error)
}

AppStore resolves a job's target service.

type HistoryStore

type HistoryStore interface {
	CreateScheduledTaskRun(ctx context.Context, run store.ScheduledTaskRun) error
	FinishScheduledTaskRun(ctx context.Context, run store.ScheduledTaskRun) error
	PruneScheduledTaskRuns(ctx context.Context, taskID string, keep int, outputCutoff time.Time) error
}

HistoryStore persists per-run history. Optional on Runner: nil keeps only the last-run snapshot RunStore writes.

type Limits

type Limits struct {
	DefaultTimeout      time.Duration
	MaxRetries          int
	RetryBackoff        time.Duration
	MaxRetryBackoff     time.Duration
	CatchUpLookback     time.Duration
	HistoryKeep         int
	OutputRetention     time.Duration
	MaxPerApp           int
	MaxConcurrentPerApp int
	MaxOutputBytes      int
}

Limits are the control-plane-wide knobs for scheduled jobs, all environment-driven so no threshold is hardcoded at a call site.

func DefaultLimits

func DefaultLimits() Limits

DefaultLimits is LimitsFromEnv with an empty environment.

func LimitsFromEnv

func LimitsFromEnv(lookup func(string) (string, bool)) Limits

LimitsFromEnv reads APP_JOBS_DEFAULT_TIMEOUT, APP_JOBS_MAX_RETRIES, APP_JOBS_RETRY_BACKOFF, APP_JOBS_RETRY_BACKOFF_MAX, APP_JOBS_CATCH_UP_LOOKBACK, APP_JOBS_HISTORY_KEEP, APP_JOBS_OUTPUT_RETENTION, APP_JOBS_MAX_PER_APP, APP_JOBS_MAX_CONCURRENT_PER_APP and APP_JOBS_MAX_OUTPUT_BYTES. A value that does not parse or is not positive falls back to its default.

type NodeRuntimeResolver

type NodeRuntimeResolver func(nodeID string) (docker.Runtime, error)

NodeRuntimeResolver resolves the docker.Runtime for a node ID, the same shape internal/api's own resolver has.

type Recoverer

type Recoverer interface {
	Cleanup(ctx context.Context, runs []store.ScheduledTaskRun)
}

Recoverer cleans up what a previous process left behind. *Runner satisfies it; optional.

type RunStore

type RunStore interface {
	RecordScheduledTaskRun(ctx context.Context, id string, ranAt time.Time, status, output string) error
}

RunStore persists a run's outcome onto the task's own row.

type Runner

type Runner struct {
	Apps     AppStore
	Runs     RunStore
	History  HistoryStore
	Specs    SpecBuilder
	Resolver NodeRuntimeResolver
	Logger   *slog.Logger
	Limits   Limits
	// NamePrefix namespaces fresh job containers (the brand short name).
	NamePrefix string

	// Now returns the current time; nil falls back to time.Now.
	Now func() time.Time
	// Timeout overrides every attempt's timeout when positive; a test seam.
	Timeout time.Duration
	// Sleep waits between retries; nil uses a context-aware timer.
	Sleep func(ctx context.Context, d time.Duration) error
	// contains filtered or unexported fields
}

Runner runs one job and records the outcome.

func (*Runner) Cleanup

func (r *Runner) Cleanup(ctx context.Context, runs []store.ScheduledTaskRun)

Cleanup force-removes the containers of runs a previous process left in flight. Best effort: a node that is unreachable now is simply skipped.

func (*Runner) ContainerName

func (r *Runner) ContainerName(service, runID, taskID string) string

ContainerName is the deterministic name of a fresh job container, so a restart can find and remove one the previous process left behind. The run suffix is deliberately not hex, so no service-container parser takes it for a replica.

func (*Runner) Run

func (r *Runner) Run(ctx context.Context, task store.ScheduledTask) error

Run executes task for a manual "run now".

func (*Runner) RunTrigger

func (r *Runner) RunTrigger(ctx context.Context, task store.ScheduledTask, trig Trigger) error

RunTrigger runs task once. task.ConcurrencyPolicy governs a still in-flight previous run of the same task (forbid skips, replace cancels it), and the per-app cap skips a run when the app already has Limits.MaxConcurrentPerApp jobs running. Retries with backoff happen inside the one run. The returned error is only an infrastructure failure; a failing command is a recorded outcome, not an error.

func (*Runner) Start

func (r *Runner) Start(ctx context.Context, task store.ScheduledTask, trigger string) (string, error)

Start records a run's history row synchronously, then runs it in the background, and returns the run ID so a caller can follow it. Without a history store it still runs, with an empty ID. ctx should already be detached from the request that asked.

type ScheduleStore

type ScheduleStore interface {
	ListEnabledScheduledTasks(ctx context.Context) ([]store.ScheduledTask, error)
	ClaimScheduledTaskRun(ctx context.Context, run store.ScheduledTaskRun) (bool, error)
	InterruptRunningScheduledTaskRuns(ctx context.Context, now time.Time) ([]store.ScheduledTaskRun, error)
}

ScheduleStore is the store surface Scheduler needs. *store.DB satisfies it structurally.

type Scheduler

type Scheduler struct {
	Store  ScheduleStore
	Runner TaskRunner
	Logger *slog.Logger
	// Now returns the current time; nil falls back to time.Now.
	Now func() time.Time
	// Grace is how late an occurrence may be and still count as on time;
	// zero means twice the interval passed to Run.
	Grace time.Duration
	// Lookback is the default catch-up window when a job sets none.
	Lookback time.Duration
	// NewID mints run IDs; nil uses NewRunID.
	NewID func() (string, error)
	// contains filtered or unexported fields
}

Scheduler is level-triggered: every tick it derives, from the persisted high-water mark (store.ScheduledTask.LastScheduledFor) and the current time, which cron occurrence is due, claims it atomically in the database and only then runs it. Because the claim is persisted before execution, a restart can neither run an occurrence twice nor lose track of one: an occurrence found overdue is run once (catch_up run_once, inside the lookback), or recorded as missed (skip), never silently dropped.

func NewScheduler

func NewScheduler(st ScheduleStore, runner TaskRunner, logger *slog.Logger) *Scheduler

NewScheduler builds a Scheduler. logger defaults to slog.Default().

func (*Scheduler) Recover

func (s *Scheduler) Recover(ctx context.Context, rec Recoverer) error

Recover marks runs a previous process left in flight as interrupted and removes their containers. They are never re-run: re-running an occurrence whose first execution may have done real work is how double runs happen.

func (*Scheduler) Run

func (s *Scheduler) Run(ctx context.Context, interval time.Duration) error

Run recovers interrupted runs, ticks once immediately (so a restart catches up without waiting a full interval), then ticks every interval until ctx is done.

func (*Scheduler) Tick

func (s *Scheduler) Tick(ctx context.Context) error

Tick evaluates every enabled task once. A broken task never blocks the rest; errors are joined.

func (*Scheduler) Wait

func (s *Scheduler) Wait()

Wait blocks until every run started by Tick has returned. Tests use it; production never needs to.

type SpecBuilder

type SpecBuilder interface {
	BuildJobSpec(ctx context.Context, svc *store.DesiredService, rt docker.Runtime, name string, command []string, res *docker.Resources) (application.JobSpec, error)
	JobSecretValues(ctx context.Context, svc *store.DesiredService, rt docker.Runtime) ([]string, error)
}

SpecBuilder derives a one-off container spec from a service, so a fresh job gets the service's resolved env, volumes and network. Optional: without it fresh mode reports itself unavailable.

type TaskRunner

type TaskRunner interface {
	RunTrigger(ctx context.Context, task store.ScheduledTask, trig Trigger) error
}

TaskRunner runs one claimed occurrence. *Runner satisfies it; a test substitutes a fake.

type Trigger

type Trigger struct {
	Kind string
	// RunID is the pre-claimed history row, set by Scheduler; empty makes
	// the runner create one.
	RunID        string
	ScheduledFor *time.Time
	CatchUp      bool
}

Trigger says why a run happens and, for a scheduled one, which cron occurrence it was claimed for.