scheduler

package
v0.12.2 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 19, 2026 License: MIT Imports: 9 Imported by: 0

Documentation

Overview

Package scheduler provides cron-based scheduling for agent runs.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Dialect added in v0.4.0

type Dialect string

Dialect selects SQL syntax for the target database.

const (
	// DialectSQLite targets SQLite (dev/test). Writers are serialized, so the
	// atomic conditional advance in ClaimDue is exactly-once without SKIP LOCKED.
	DialectSQLite Dialect = "sqlite"
	// DialectPostgres targets PostgreSQL (production). ClaimDue selects due rows
	// FOR UPDATE SKIP LOCKED so many replicas claim disjoint schedules.
	DialectPostgres Dialect = "postgres"
)

type RunFunc

type RunFunc func(ctx context.Context, agentID, input, sessionID string) error

RunFunc is called when a schedule fires. It receives the agent ID, input, and session ID.

type RunRecord

type RunRecord struct {
	ID         string    `json:"id"`
	ScheduleID string    `json:"schedule_id"`
	AgentID    string    `json:"agent_id"`
	SessionID  string    `json:"session_id"`
	Input      string    `json:"input"`
	Status     string    `json:"status"` // success, error
	Error      string    `json:"error,omitempty"`
	StartedAt  time.Time `json:"started_at"`
	FinishedAt time.Time `json:"finished_at"`
}

RunRecord is a historical record of a scheduled run.

type Runner added in v0.4.0

type Runner interface {
	Add(agentID, cronExpr, input string, newSession bool) (*Schedule, error)
	Remove(id string) error
	List() []*Schedule
	Get(id string) (*Schedule, error)
	History(scheduleID string) []RunRecord
	Start(ctx context.Context)
	Stop()
}

Runner is the scheduler control surface used by the control plane. Both the in-process Scheduler (default; single-replica / tests) and the store-backed StoreScheduler (production; exactly-once firing across replicas) implement it, so the server can hold a Runner and swap implementations by configuration.

type SQLStore added in v0.4.0

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

SQLStore implements Store over any database/sql database (SQLite or PostgreSQL, selected by Dialect). It owns the scheduler schema and never touches the shared storage package's tables.

func NewSQLStore added in v0.4.0

func NewSQLStore(db *sql.DB, dialect Dialect) *SQLStore

NewSQLStore constructs a store over an already-open *sql.DB. The caller owns the DB lifecycle unless Close is used.

func (*SQLStore) Add added in v0.4.0

func (s *SQLStore) Add(ctx context.Context, sched *Schedule) error

Add persists a new schedule.

func (*SQLStore) AddRunRecord added in v0.4.0

func (s *SQLStore) AddRunRecord(ctx context.Context, rec RunRecord) error

AddRunRecord appends a run record.

func (*SQLStore) ClaimDue added in v0.4.0

func (s *SQLStore) ClaimDue(ctx context.Context, now time.Time, nextFn func(expr string, after time.Time) time.Time) ([]*Schedule, error)

ClaimDue atomically advances due schedules and returns those claimed by this caller. The winning replica is the one whose conditional UPDATE affects the row: after it advances next_run_at into the future, a concurrent replica's UPDATE re-evaluates the `next_run_at <= now` guard against the already-advanced row and affects zero rows, so the firing is claimed exactly once.

func (*SQLStore) Close added in v0.4.0

func (s *SQLStore) Close() error

Close closes the underlying database.

func (*SQLStore) Get added in v0.4.0

func (s *SQLStore) Get(ctx context.Context, id string) (*Schedule, error)

Get returns a schedule by ID.

func (*SQLStore) History added in v0.4.0

func (s *SQLStore) History(ctx context.Context, scheduleID string) ([]RunRecord, error)

History returns run records for a schedule.

func (*SQLStore) List added in v0.4.0

func (s *SQLStore) List(ctx context.Context) ([]*Schedule, error)

List returns all schedules.

func (*SQLStore) Migrate added in v0.4.0

func (s *SQLStore) Migrate(ctx context.Context) error

Migrate creates the scheduler schema.

func (*SQLStore) Remove added in v0.4.0

func (s *SQLStore) Remove(ctx context.Context, id string) error

Remove deletes a schedule by ID.

func (*SQLStore) SetSession added in v0.4.0

func (s *SQLStore) SetSession(ctx context.Context, id, sessionID string) error

SetSession persists the reused session ID for a schedule.

type Schedule

type Schedule struct {
	ID         string    `json:"id"`
	AgentID    string    `json:"agent_id"`
	CronExpr   string    `json:"cron_expr"`
	Input      string    `json:"input"`
	NewSession bool      `json:"new_session"` // true = new session per run, false = reuse
	SessionID  string    `json:"session_id,omitempty"`
	Enabled    bool      `json:"enabled"`
	CreatedAt  time.Time `json:"created_at"`
	LastRunAt  time.Time `json:"last_run_at,omitempty"`
	NextRunAt  time.Time `json:"next_run_at,omitempty"`
	RunCount   int64     `json:"run_count"`
}

Schedule defines a cron-scheduled agent run.

type Scheduler

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

Scheduler manages cron-scheduled agent runs.

func New

func New(runFn RunFunc) *Scheduler

New creates a new Scheduler. runFn is called when a schedule fires.

func (*Scheduler) Add

func (s *Scheduler) Add(agentID, cronExpr, input string, newSession bool) (*Schedule, error)

Add creates a new schedule.

func (*Scheduler) Get

func (s *Scheduler) Get(id string) (*Schedule, error)

Get returns a schedule by ID.

func (*Scheduler) History

func (s *Scheduler) History(scheduleID string) []RunRecord

History returns run records for a schedule.

func (*Scheduler) List

func (s *Scheduler) List() []*Schedule

List returns all schedules.

func (*Scheduler) Remove

func (s *Scheduler) Remove(id string) error

Remove deletes a schedule.

func (*Scheduler) Start

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

Start begins the scheduler loop. It checks for due schedules every tick interval.

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop halts the scheduler loop.

func (*Scheduler) WithTickInterval

func (s *Scheduler) WithTickInterval(d time.Duration) *Scheduler

WithTickInterval sets the polling interval (for testing). Default is 1 minute.

type Store added in v0.4.0

type Store interface {
	// Migrate creates the scheduler schema (idempotent).
	Migrate(ctx context.Context) error
	// Add persists a new schedule.
	Add(ctx context.Context, sched *Schedule) error
	// Remove deletes a schedule by ID.
	Remove(ctx context.Context, id string) error
	// Get returns a schedule by ID.
	Get(ctx context.Context, id string) (*Schedule, error)
	// List returns all schedules.
	List(ctx context.Context) ([]*Schedule, error)
	// ClaimDue atomically advances the NextRunAt of every schedule that is due at
	// now and returns only those this caller claimed. Across replicas sharing the
	// store, a given due firing is claimed by exactly one caller. nextFn computes
	// the next fire time for a cron expression strictly after the given instant.
	ClaimDue(ctx context.Context, now time.Time, nextFn func(expr string, after time.Time) time.Time) ([]*Schedule, error)
	// SetSession persists the reused session ID for a schedule.
	SetSession(ctx context.Context, id, sessionID string) error
	// AddRunRecord appends a run record to a schedule's history.
	AddRunRecord(ctx context.Context, rec RunRecord) error
	// History returns run records for a schedule.
	History(ctx context.Context, scheduleID string) ([]RunRecord, error)
	// Close releases resources owned by the store.
	Close() error
}

Store is the durable persistence contract for the store-backed scheduler. It is defined here (rather than in the shared storage package) so the scheduler owns its own schema and can be backed by any *sql.DB via SQLStore. Timestamps are passed explicitly so behavior is deterministic and testable.

type StoreScheduler added in v0.4.0

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

StoreScheduler is a store-backed Scheduler that claims each due schedule exactly once across all replicas sharing the Store, via an atomic conditional advance of NextRunAt (SELECT ... FOR UPDATE SKIP LOCKED on Postgres; SQLite serializes writers), so two replicas polling the same store never both claim the same firing. Note this is exactly-once *claiming* / at-most-once *execution*: a crash after the claim commits but before runFn completes drops that one firing (it is not retried), rather than risking a double-run.

It implements Runner. The Runner methods match the in-process Scheduler's signatures for drop-in substitution and therefore take no ctx; they use a background context internally. Callers that need ctx/error propagation should use the *Context variants (AddContext, ListContext, HistoryContext).

func NewStoreScheduler added in v0.4.0

func NewStoreScheduler(store Store, runFn RunFunc) *StoreScheduler

NewStoreScheduler creates a store-backed Scheduler. runFn is called when a schedule fires. Call Migrate (or Start, which migrates) before use.

func (*StoreScheduler) Add added in v0.4.0

func (s *StoreScheduler) Add(agentID, cronExpr, input string, newSession bool) (*Schedule, error)

Add creates a new schedule (Runner interface; background context).

func (*StoreScheduler) AddContext added in v0.4.0

func (s *StoreScheduler) AddContext(ctx context.Context, agentID, cronExpr, input string, newSession bool) (*Schedule, error)

AddContext creates a new schedule.

func (*StoreScheduler) Get added in v0.4.0

func (s *StoreScheduler) Get(id string) (*Schedule, error)

Get returns a schedule by ID (Runner interface; background context).

func (*StoreScheduler) History added in v0.4.0

func (s *StoreScheduler) History(scheduleID string) []RunRecord

History returns run records for a schedule (Runner interface). A store error yields nil; use HistoryContext for error propagation.

func (*StoreScheduler) HistoryContext added in v0.4.0

func (s *StoreScheduler) HistoryContext(ctx context.Context, scheduleID string) ([]RunRecord, error)

HistoryContext returns run records for a schedule.

func (*StoreScheduler) List added in v0.4.0

func (s *StoreScheduler) List() []*Schedule

List returns all schedules (Runner interface). A store error yields an empty slice; use ListContext for error propagation.

func (*StoreScheduler) ListContext added in v0.4.0

func (s *StoreScheduler) ListContext(ctx context.Context) ([]*Schedule, error)

ListContext returns all schedules.

func (*StoreScheduler) Migrate added in v0.4.0

func (s *StoreScheduler) Migrate(ctx context.Context) error

Migrate creates the scheduler schema.

func (*StoreScheduler) Remove added in v0.4.0

func (s *StoreScheduler) Remove(id string) error

Remove deletes a schedule (Runner interface; background context).

func (*StoreScheduler) Start added in v0.4.0

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

Start begins the scheduler loop, migrating the schema first, then checking for due schedules every tick. It returns when ctx is canceled or Stop is called.

func (*StoreScheduler) Stop added in v0.4.0

func (s *StoreScheduler) Stop()

Stop halts the scheduler loop.

func (*StoreScheduler) Tick added in v0.4.0

func (s *StoreScheduler) Tick(ctx context.Context, now time.Time)

Tick claims and fires all schedules due at now. It is exported so tests (and external drivers) can drive a single deterministic scheduling cycle.

func (*StoreScheduler) WithTickInterval added in v0.4.0

func (s *StoreScheduler) WithTickInterval(d time.Duration) *StoreScheduler

WithTickInterval sets the polling interval (for testing). Default is 1 minute.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL