schedule

type Calendar ¶

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

Calendar constructs ordinary declarations in one configured timezone. It owns no worker or registry; execution reuses the existing scheduler and DST policy.

func NewCalendar(name temporal.ZoneName) (Calendar, error)

func (c Calendar) Cron(id ID, expression string, handler Handler) (Declaration, error)

func (c Calendar) Daily(id ID, handler Handler) (Declaration, error)

func (c Calendar) DailyAt(id ID, text string, handler Handler) (Declaration, error)

func (c Calendar) EveryFifteenMinutes(id ID, handler Handler) (Declaration, error)

func (c Calendar) EveryFiveMinutes(id ID, handler Handler) (Declaration, error)

func (c Calendar) EveryMinute(id ID, handler Handler) (Declaration, error)

func (c Calendar) EveryTenMinutes(id ID, handler Handler) (Declaration, error)

func (c Calendar) EveryThirtyMinutes(id ID, handler Handler) (Declaration, error)

func (c Calendar) Hourly(id ID, handler Handler) (Declaration, error)

func (c Calendar) In(name temporal.ZoneName) (Calendar, error)

func (c Calendar) LastDayOfMonthAt(id ID, text string, handler Handler) (Declaration, error)

func (c Calendar) Monthly(id ID, handler Handler) (Declaration, error)

func (c Calendar) Weekly(id ID, handler Handler) (Declaration, error)

type CatchUp ¶

type CatchUp struct {
	Window time.Duration
	Max    int
}

CatchUp is opt-in bounded replay on startup, leadership acquisition and missed ticks. Zero skips missed occurrences. At most Max occurrences from Window are considered per schedule per tick; excess backlog is skipped, never accumulated.

func (c CatchUp) Validate() error

type Config ¶

type Config struct {
	Group         Group
	Clock         clock.Clock
	Concurrency   int
	MaxHistory    int
	MaxPerTick    int
	PollInterval  time.Duration
	LeadershipTTL time.Duration
	// Grace bounds lateness for schedules without catch-up. Older occurrences are
	// skipped. Process startup itself skips all missed occurrences by default.
	Grace time.Duration
	// CapacityWait lets a due occurrence wait for a free execution slot when
	// Concurrency is exhausted, before it is skipped as CapacityReached.
	CapacityWait time.Duration
	// DrainTimeout is the graceful-shutdown budget. When Run's context ends or
	// Drain is called, admission stops while running tasks keep leadership and
	// overlap leases for up to DrainTimeout; they are then cancelled. Zero
	// cancels immediately. Keep it within the application shutdown timeout.
	DrainTimeout time.Duration
	// Logger receives skipped occurrences and failures with redacted
	// diagnostics. Module uses the application logger when it is nil.
	Logger *slog.Logger
	// Wake optionally replaces the polling ticker for controlled application time.
	// The caller owns the channel; closing it stops Run with an error. Leadership
	// and overlap heartbeats still use real safety time independent of this clock.
	Wake <-chan struct{}
}

func DefaultConfig(group Group) Config

func (c Config) Validate() error

func (c Config) ValidateShutdown(timeout time.Duration) error

type CursorBackend ¶

type CursorBackend interface {
	ScheduleCursor(context.Context, lease.Key) (time.Time, bool, error)
	AdvanceScheduleCursor(context.Context, lease.Key, time.Time, time.Duration) error
}

CursorBackend is optional on the scheduler's lease backend. It persists, per schedule, the latest occurrence that completed or was deliberately skipped, so a restart or leadership change resumes catch-up after it instead of replaying completed occurrences. Advance must keep the maximum instant and may expire the entry after ttl. The Redis coordination client implements it; without it the cursor lives only in this scheduler process.

type Declaration ¶

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

Declaration freezes parsed timing and domain callbacks. No I/O or background activity occurs during declaration/registration. Services are constructor-injected.

func Cron(id ID, expression string, zone *time.Location, handler Handler) (Declaration, error)

func Daily(id ID, zone *time.Location, handler Handler) (Declaration, error)

func DailyAt(id ID, text string, zone *time.Location, handler Handler) (Declaration, error)

func Define(id ID, spec Spec, handler Handler, options Options) (Declaration, error)

func Every(id ID, every time.Duration, handler Handler) (Declaration, error)

func EveryFifteenMinutes(id ID, zone *time.Location, handler Handler) (Declaration, error)

func EveryFiveMinutes(id ID, zone *time.Location, handler Handler) (Declaration, error)

func EveryMinute(id ID, zone *time.Location, handler Handler) (Declaration, error)

func EveryTenMinutes(id ID, zone *time.Location, handler Handler) (Declaration, error)

func EveryThirtyMinutes(id ID, zone *time.Location, handler Handler) (Declaration, error)

func Hourly(id ID, zone *time.Location, handler Handler) (Declaration, error)

func LastDayOfMonthAt(id ID, text string, zone *time.Location, handler Handler) (Declaration, error)

func Monthly(id ID, zone *time.Location, handler Handler) (Declaration, error)

func Weekly(id ID, zone *time.Location, handler Handler) (Declaration, error)

func (d Declaration) ID() ID

func (d Declaration) Options() Options

func (d Declaration) Spec() Spec

func (d Declaration) Validate() error

func (d Declaration) With(options Options) (Declaration, error)

type Description ¶

type Description struct {
	ID             ID            `json:"id"`
	Expression     string        `json:"expression"`
	TimeZone       string        `json:"time_zone"`
	Interval       time.Duration `json:"interval,omitzero"`
	Anchor         time.Time     `json:"anchor,omitzero"`
	Timeout        time.Duration `json:"timeout"`
	WithoutOverlap bool          `json:"without_overlap"`
	OverlapTTL     time.Duration `json:"overlap_ttl"`
	Environments   []string      `json:"environments"`
	CatchUp        CatchUp       `json:"catch_up"`
}

Description contains timing and execution policy without callback references, coordination ownership or task history. Interval anchors are explicit UTC.

type Execution ¶

type Execution struct{}

type ExecutionID ¶

type ExecutionID = model.ID[Execution]

type Group ¶

type Group string

Group isolates independent scheduler elections in one namespace. All replicas of a group must deploy the same registry, timing and environment policy.

type Handler ¶

type Handler func(context.Context, Invocation) error

func ConnectionJobTarget[P any](connection *jobs.Connection, definition jobs.Definition[P], queue jobs.Queue, build func(context.Context, Invocation) (P, error)) (Handler, error)

func JobTarget[P any](dispatcher *jobs.Dispatcher, definition jobs.Definition[P], build func(context.Context, Invocation) (P, error)) (Handler, error)

type ID ¶

type ID string

type Invocation ¶

type Invocation struct {
	Schedule   ID
	Occurrence OccurrenceID
	IntendedAt time.Time
	Origin     attribution.Origin
}

Invocation carries stable occurrence identity, not a live service container. The same namespace/group/schedule/UTC instant has the same ID after failover. It is an idempotency key, not proof of exactly-once business effects.

func Current(ctx context.Context) (Invocation, bool)

type Occurrence ¶

type Occurrence struct{}

type OccurrenceID ¶

type OccurrenceID = model.ID[Occurrence]

type Options ¶

type Options struct {
	Timeout               time.Duration
	WithoutOverlap        bool
	OverlapTTL            time.Duration
	Environments          []string
	CatchUp               CatchUp
	Days                  []time.Weekday
	Between               Window
	LastDayOfMonth        bool
	When                  Predicate
	EvenInMaintenanceMode bool
	Before                Handler
	After                 Handler
	Failed                func(context.Context, Invocation, error) error
}

Options configure one declaration. Days, Between and LastDayOfMonth narrow the spec in its zone without callbacks: a filtered occurrence advances the schedule silently. When runs as an owned callback before the Before hook; a false result records the occurrence as Skipped with reason Filtered (not logged as a failure), and an error fails it as a hook failure. EvenInMaintenanceMode keeps admitting this schedule while the application's maintenance gate is paused or draining.

func DefaultOptions() Options

func (o Options) Validate() error

type Predicate ¶

type Predicate func(context.Context, Invocation) (bool, error)

Predicate decides at execution time whether an occurrence runs. It is an owned callback: it receives the invocation context and must honor it.

type Reason ¶

type Reason string

const (
	NoReason           Reason = ""
	HandlerFailed      Reason = "handler_failed"
	HookFailed         Reason = "hook_failed"
	Panicked           Reason = "panicked"
	TimedOut           Reason = "timed_out"
	Stopped            Reason = "stopped"
	LeadershipLost     Reason = "leadership_lost"
	OverlapLost        Reason = "overlap_lost"
	OverlapBusy        Reason = "overlap_busy"
	CoordinationFailed Reason = "coordination_failed"
	CapacityReached    Reason = "capacity_reached"
	Missed             Reason = "missed"
	BacklogLimited     Reason = "backlog_limited"
	ClockFailed        Reason = "clock_failed"
	// Filtered records an occurrence whose Options.When predicate declined it.
	Filtered Reason = "filtered"
)

type Record ¶

type Record struct {
	Execution  ExecutionID
	Invocation Invocation
	State      State
	Reason     Reason
	StartedAt  time.Time
	FinishedAt time.Time
}

Record is bounded process-local operational history, not a durable ledger. It deliberately contains classifications instead of arbitrary callback errors.

type Registry ¶

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

func NewRegistry(declarations ...Declaration) (*Registry, error)

func (r *Registry) Describe() []Description

type ScheduleStatus ¶

type ScheduleStatus struct {
	ID      ID
	Enabled bool
	Next    time.Time
	Active  int
}

type Scheduler ¶

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

Scheduler borrows the shared lease manager. Run is single-use. Cancelling Run's context, or Drain, stops admission and lets running tasks finish within DrainTimeout while leadership is kept; Stop cancels tasks immediately. Done waits for their actual exit. Keep the manager/backend alive until Done; Module supplies foundation dependency order. A Scheduler must not be copied.

func New(manager *lease.Manager, registry *Registry, config Config) (*Scheduler, error)

func (s *Scheduler) Describe() []Description

func (s *Scheduler) Done() <-chan struct{}

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

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

func (s *Scheduler) RunNow(ctx context.Context, id ID) (Record, error)

func (s *Scheduler) Snapshot() Snapshot

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

type Snapshot ¶

type Snapshot struct {
	Running                bool
	Stopping               bool
	Leader                 bool
	Active                 int
	LeadershipAcquisitions uint64
	LeadershipLosses       uint64
	CoordinationFailures   uint64
	Schedules              []ScheduleStatus
	History                []Record
}

type Spec ¶

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

Spec is an immutable parsed cron or anchored interval. Calendar schedules use local wall time: nonexistent DST times are skipped, repeated times each denote a separate UTC occurrence. Intervals measure elapsed time and ignore DST.

func Interval(every time.Duration) (Spec, error)

func IntervalFrom(every time.Duration, anchor time.Time) (Spec, error)

func ParseCron(source string, zone *time.Location) (Spec, error)

func (s Spec) Next(after time.Time) (time.Time, error)

func (s Spec) String() string

func (s Spec) TimeZone() string

func (s Spec) Validate() error

type State ¶

type State string

const (
	Running   State = "running"
	Succeeded State = "succeeded"
	Failed    State = "failed"
	Cancelled State = "cancelled"
	Skipped   State = "skipped"
)

type Window ¶

type Window struct {
	From, To temporal.Time
	Unless   bool
}

Window limits occurrences to local wall-clock times in [From, To) in the spec's zone (UTC for intervals). From after To wraps midnight, for example 22:00 to 06:00. Unless inverts the window: occurrences inside it are skipped. The zero Window allows every time.

func Between(from, to temporal.Time) Window

func UnlessBetween(from, to temporal.Time) Window

func (w Window) Validate() error