Documentation
¶
Overview ¶
Package scheduler is the in-process job scheduler that hosts the background tasks of the `da service` runtime. It is deliberately small (interval + fsnotify triggers, panic-recovering task dispatch, last-run bookkeeping, drain-on-stop) rather than a durable queue — see .agents/workflow/specs/r3-background-worker-service/design.md (D2) for the rationale and rejected alternatives.
The package owns goroutine lifecycle only. It deliberately knows nothing about scoring, watermarks, HTTP, or the event bus: tasks are plain funcs and the bus plugs in at the call site (the runtime composes them). This keeps the scheduler reusable and trivially testable.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( ErrNoName = errors.New("scheduler: task name is required") ErrNoTrigger = errors.New("scheduler: task trigger is required") ErrNoRunFn = errors.New("scheduler: task RunFn is required") ErrDuplicateName = errors.New("scheduler: duplicate task name") ErrAlreadyStarted = errors.New("scheduler: already started") ErrNotStarted = errors.New("scheduler: not started") )
Errors returned by Register / Start.
Functions ¶
This section is empty.
Types ¶
type FSNotifyTrigger ¶
type FSNotifyTrigger struct {
// Paths are the files or directories to watch. At least one is required.
Paths []string
// Debounce is the quiet period that must elapse after the last raw event
// before a coalesced tick is emitted. Zero defaults to 50ms.
Debounce time.Duration
}
FSNotifyTrigger fires when any watched path emits a write/create/rename event. Rapid bursts are coalesced: while a debounce window is open, further events extend it instead of producing a tick, so a flurry of writes to one file yields a single fire after the burst settles.
func FSNotify ¶
func FSNotify(path string) *FSNotifyTrigger
FSNotify is a convenience constructor for an FSNotifyTrigger over one path.
type IntervalTrigger ¶
IntervalTrigger fires once per Every duration. The first tick is emitted one interval after Start (it does not fire immediately).
func Interval ¶
func Interval(d time.Duration) *IntervalTrigger
Interval is a convenience constructor for an IntervalTrigger.
type RunFunc ¶
RunFunc is the unit of work a Task performs on each trigger fire. The context is derived from the scheduler's run context and is additionally bounded by the task's Timeout when one is set. A returned error is recorded as the task's last-error and increments its consecutive-failure count; a nil error resets that count.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler hosts a set of tasks, each driven by its own trigger. It is safe for concurrent use: Register before Start, State at any time, Stop once.
func (*Scheduler) Register ¶
Register adds a task. It must be called before Start. Returns an error for missing required fields or a duplicate name.
func (*Scheduler) Start ¶
Start launches every registered task's trigger loop. The provided context governs the whole scheduler: cancelling it stops all triggers and (via Stop's drain semantics) the in-flight runs. Start returns once all triggers are live, or an error if a trigger fails to start (in which case any triggers already started are torn down).
func (*Scheduler) Stop ¶
Stop performs a graceful shutdown: it stops accepting new ticks (cancelling trigger loops), then waits up to timeout for in-flight RunFns to finish. If the timeout elapses first, the run context is cancelled so RunFns honouring ctx can abort, and Stop returns context.DeadlineExceeded. A nil error means a clean drain. Stop is idempotent-safe to call once; a second call returns ErrNotStarted.
type Task ¶
type Task struct {
// Name uniquely identifies the task within a scheduler. Required.
Name string
// Trigger drives when RunFn executes. Required.
Trigger Trigger
// RunFn is the work performed on each fire. Required.
RunFn RunFunc
// Timeout bounds a single RunFn execution. Zero means no timeout.
Timeout time.Duration
}
Task binds a name to a trigger and a unit of work.
type TaskState ¶
type TaskState struct {
Name string `json:"name"`
LastRunAt *time.Time `json:"last_run_at,omitempty"`
NextRunAt *time.Time `json:"next_run_at,omitempty"`
LastError string `json:"last_error,omitempty"`
ConsecutiveFailures int `json:"consecutive_failures"`
Runs int `json:"runs"`
// Dropped counts trigger fires skipped because a previous run of this task
// was still in flight (OQ2: drop-on-overrun, single goroutine per task).
Dropped int `json:"dropped"`
// Running reports whether a RunFn is currently executing.
Running bool `json:"running"`
}
TaskState is an immutable snapshot of one task's health, surfaced to the status CLI / HTTP endpoint via Scheduler.State.
type Trigger ¶
type Trigger interface {
// Start begins emitting ticks until ctx is cancelled. The returned channel
// is closed when the trigger has fully stopped (after ctx cancellation).
// An error is returned only for setup failures (e.g. a watch path that
// cannot be registered); a nil error means the channel is live.
Start(ctx context.Context) (<-chan time.Time, error)
}
Trigger emits ticks that drive a Task's RunFn. A trigger owns its own goroutine lifecycle: Start spins up the source and returns a channel that receives one value per fire; closing the returned context (or calling the returned stop func) tears it down. Implementations must be safe to Start at most once.
Triggers do not run the task themselves — the scheduler reads from the channel and dispatches. This keeps trigger logic (timing, file watching) independent of task lifecycle (panic recovery, bookkeeping, drop-on-overrun).
Source Files