scheduler

type Cache ¶

type Cache struct {
	Schedules     []LocalSchedule      `json:"schedules"`
	LastFiredAt   map[string]time.Time `json:"last_fired_at"`
	LastFetchedAt time.Time            `json:"last_fetched_at"`
}

Cache is the disk-persisted shape — exported so an agent's debug endpoint can render it directly.

type Chain ¶

type Chain struct {
	// contains filtered or unexported fields
}

Chain turns an agent's job-completion stream into ordered, event-triggered schedule fires. It is the generic primitive behind "run the maintenance jobs after the nightly backup wave finishes": cron-fired backup/copy schedules land their data, and when the agent has no more of those in flight the Chain fires the next stage's On-schedules (e.g. prune), then the stage after that (e.g. check) once the prune jobs drain — strictly one stage per day.

The Chain is deliberately agent-agnostic:

• It does not know about duplicacy actions. The agent supplies a Spec whose callbacks classify a completed job (is it part of the wave? which stage's work just finished?) and report whether any of a stage's jobs are still active.
• Firing a stage means "FireMatching the schedules this stage owns". The agent's Spec provides the predicate (typically: Schedule.On == sentinel && Action == "<stage action>").

Idempotency is per calendar day (agent-local timezone). Each stage records the date it last fired in an atomic on-disk marker file, so an agent restart mid-morning neither re-fires a stage that already ran today nor skips a stage whose turn had come. First boot with no marker does NOT retroactively fire a wave that completed before the process started — OnJobComplete only ever acts on transitions it personally witnesses.

func NewChain(s *Scheduler, spec Spec) *Chain

func (c *Chain) OnJobComplete(ctx context.Context, action string)

type Config ¶

type Config struct {
	// PullURL is the full URL the scheduler GETs to refresh the local
	// schedule list, e.g.
	// "https://controller.example.com/api/schedules?node=node-a".
	// The kit does not format this — callers substitute their own node.
	PullURL string

	// NodeName identifies this agent in log lines. Not used in the
	// PullURL — that's the caller's responsibility.
	NodeName string

	// CachePath is the disk path for the JSON cache file. The kit writes
	// atomically (path + ".tmp" → rename) so a crash mid-write doesn't
	// corrupt the file. Parent directory must exist.
	CachePath string

	// HTTPClient does the pull. Defaults to http.DefaultClient. Pass a
	// pre-configured client when the call needs bearer auth, a custom
	// dial path (e.g. through Traefik), or non-default timeouts.
	HTTPClient *http.Client

	// Fire is the per-schedule callback. Required.
	Fire FireFunc

	// Timezone for cron evaluation. Defaults to America/New_York to
	// match the rest of the stack.
	Timezone *time.Location

	// FireInterval is the cron tick. Defaults to 1 * time.Minute. Tests
	// can lower this; production should not.
	FireInterval time.Duration

	// PullInterval is the reconcile tick. Defaults to 5 * time.Minute.
	PullInterval time.Duration
}

Config configures the scheduler. PullURL, NodeName, CachePath, and Fire are required; the rest have safe defaults.

type FireFunc ¶

type FireFunc func(ctx context.Context, sch LocalSchedule, triggerKey string, missedRecovery bool) error

FireFunc is the agent-supplied callback invoked when a schedule matches the current minute. The kit only handles the bookkeeping — the agent is responsible for the actual work (CLI invocation, API call, etc).

`triggerKey` is set by the kit:

• "schedule" for a normal fire from the 1-minute tick.
• "schedule-recovery" for a missed-run catch-up fire from Start.

`missedRecovery` mirrors triggerKey for convenience and matches the agents' existing telemetry shape.

Returning an error does NOT stop the scheduler — it is logged and the next tick proceeds normally. A non-nil error means the fire did NOT happen: the kit leaves LastFiredAt un-stamped so the window is retried (next matching minute and by the periodic missedRunCheck). Return nil ONLY when the work was actually dispatched; return an error for any transient failure (secret vend timeout, repo not yet loaded, job failed to start) so it gets another attempt rather than being silently lost.

type LocalSchedule ¶

type LocalSchedule struct {
	ID         string                 `json:"id"`
	Name       string                 `json:"name"`
	RepoID     string                 `json:"repo_id"`
	Action     string                 `json:"action"`
	Storage    string                 `json:"storage,omitempty"`
	Schedule   Schedule               `json:"schedule"`
	MissPolicy MissPolicy             `json:"miss_policy"`
	Enabled    bool                   `json:"enabled"`
	Params     map[string]interface{} `json:"params,omitempty"`
	Version    int64                  `json:"version"`
}

LocalSchedule is one row pulled from the controller plus the agent-side bookkeeping the kit maintains.

Action is an opaque string — each agent defines its own action vocabulary ("backup" / "check" / "prune" for duplicacy; "backup" / "prune" for gdrive). The kit does not interpret it.

Params is a free-form map that the agent's FireFunc decodes. Use the Param* helpers in this package to extract typed values without panicking on JSON-decoded interface{} variations.

type MissPolicy ¶

type MissPolicy string

MissPolicy controls catch-up behaviour when an agent boots after missing a fire window.

const (
	// MissSkip: never replay missed windows.
	MissSkip MissPolicy = "skip"
	// MissRunOnceWithin24 fires once on boot if the most recent window
	// within the last 24 h was missed (lastFired before that window).
	MissRunOnceWithin24 MissPolicy = "run-once-within-24h"
	// MissRunAll is currently treated identically to MissRunOnceWithin24
	// to avoid a missed-prune storm; the truncation is logged.
	MissRunAll MissPolicy = "run-all"
)

type Schedule ¶

type Schedule struct {
	Days            []string `json:"days"`
	Hours           []string `json:"hours"`
	Minute          string   `json:"minute"`
	IntervalMinutes *int     `json:"interval_minutes,omitempty"`
	On              string   `json:"on,omitempty"`
}

Schedule is the cron-like shape pulled from the controller. Mirrors the controller's scheduler so the JSON wire format is identical whether the schedule fires from the controller or from an agent.

On is a non-cron trigger sentinel: when set (e.g. {"on":"after_wave"}), the schedule NEVER fires on the clock — Matches always returns false and missed-run recovery skips it. Such rows are fired on demand by an agent's chain controller via FireSchedule when an upstream condition is met (e.g. "the nightly backup+copy wave finished"). Days/Hours/Minute are ignored when On is set. Empty On means a normal cron schedule.

type Scheduler ¶

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler is the running scheduler. Create with New, drive with Start + Stop, accept early-reconcile pokes via Pull (intended for an HTTP /schedules/refresh handler).

func New(cfg Config) (*Scheduler, error)

func (s *Scheduler) Cache() Cache

func (s *Scheduler) FireMatching(ctx context.Context, triggerKey string, pred func(LocalSchedule) bool) int

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

func (s *Scheduler) Schedules() []LocalSchedule

func (s *Scheduler) Start(ctx context.Context)

func (s *Scheduler) Stop()

type Spec ¶

type Spec struct {
	// Stages run in order. Stage[i+1] fires once Stage[i] has fired today AND
	// Stage[i].Active() reports false (its jobs drained). Stage[0] fires once
	// the wave is complete (see WaveComplete).
	Stages []Stage

	// IsWaveJob reports whether a completed job is part of the nightly wave
	// (backup/copy) that gates Stage[0]. Completions of non-wave jobs (e.g.
	// the chain's own prune/check) still drive stage advancement but do not
	// re-arm the wave.
	IsWaveJob func(action string) bool

	// WaveComplete reports whether the wave has fully drained: at least one
	// wave job ran since arming and none are now running/pending. The agent
	// implements this against its job registry (e.g. "no backup/copy job is
	// running or pending AND we saw ≥1 finish"). The Chain calls it only on a
	// wave-job completion.
	WaveComplete func() bool

	// MarkerPath is the on-disk file where per-stage last-fired dates persist
	// (atomic write). Parent dir must exist. Required.
	MarkerPath string
}

Spec configures a Chain for a specific agent.

type Stage ¶

type Stage struct {
	// Name labels the stage in logs and is the key under which its
	// last-fired date is persisted. Stable across releases — changing it
	// resets that stage's idempotency marker.
	Name string

	// TriggerKey labels the jobs this stage starts (e.g. "chain-prune").
	// Surfaced on the job row so operators can see a job was chain-fired.
	TriggerKey string

	// Match selects the schedules this stage fires. Called for every cached
	// schedule; return true for the ones this stage owns (typically
	// sch.Schedule.On == "after_wave" && sch.Action == "<stage action>").
	Match func(LocalSchedule) bool

	// Active reports whether any job belonging to THIS stage is still
	// running or pending. The Chain advances to the next stage only when this
	// returns false after the stage has fired. For a stage that fires
	// "prune", Active is "any prune job running/pending?".
	Active func() bool
}

Stage is one step in the chain, fired after the previous stage's work drains. Stage 0's "previous" is the backup+copy wave itself.