executus/schedule/runner.go

// Package schedule is the cron-runner battery: a generic ticker that, each
// interval, asks a store for the jobs whose next-run time has passed, runs each
// one, and stamps its next fire time. It is host-agnostic orchestration — the
// host wires the store (skill.SkillStore.ListDueScheduled /
// persona.Storage.ListScheduledAgents), the run (run.Executor), and the cron
// "next fire" function (a cron library, or skill's schedule parser). The
// battery owns no cron grammar of its own, so it never duplicates the parser.
package schedule

import (
	"context"
	"log/slog"
	"time"
)

// Due is one schedulable job: its id and its cron expression.
type Due struct {
	ID   string
	Cron string
}

// Runner periodically fires due jobs. Every func field is required except Now
// (defaults to time.Now) and Logger (defaults to slog.Default). Construct the
// struct directly and call Loop (or Tick for a single pass / tests).
type Runner struct {
	// Interval is how often Loop checks for due jobs. <= 0 defaults to 1m.
	Interval time.Duration
	// Due lists the jobs due at now.
	Due func(ctx context.Context, now time.Time) ([]Due, error)
	// Run executes one job by id.
	Run func(ctx context.Context, id string) error
	// Mark records that a job ran at ranAt and is next due at nextAt.
	Mark func(ctx context.Context, id string, ranAt, nextAt time.Time) error
	// Next computes a cron expression's next fire after a given time.
	Next func(cron string, after time.Time) (time.Time, error)

	Now    func() time.Time
	Logger *slog.Logger
}

func (r *Runner) now() time.Time {
	if r.Now != nil {
		return r.Now()
	}
	return time.Now()
}

func (r *Runner) log() *slog.Logger {
	if r.Logger != nil {
		return r.Logger
	}
	return slog.Default()
}

// Tick runs one pass: every currently-due job is run, then stamped with its
// next fire time. A job whose Run or Next errors is logged and skipped (its
// next-run time is left unchanged so it stays due and retries next tick) — one
// bad job never stalls the others. Returns the error from Due (the only
// pass-fatal step).
func (r *Runner) Tick(ctx context.Context) error {
	now := r.now()
	due, err := r.Due(ctx, now)
	if err != nil {
		return err
	}
	for _, j := range due {
		if err := r.Run(ctx, j.ID); err != nil {
			r.log().Warn("scheduled job failed; will retry next tick", "job", j.ID, "error", err)
			continue
		}
		next, err := r.Next(j.Cron, now)
		if err != nil {
			r.log().Warn("scheduled job has an unparseable cron; not rescheduling", "job", j.ID, "cron", j.Cron, "error", err)
			continue
		}
		if err := r.Mark(ctx, j.ID, now, next); err != nil {
			r.log().Warn("failed to stamp scheduled job's next run", "job", j.ID, "error", err)
		}
	}
	return nil
}

// Loop ticks every Interval until ctx is cancelled. A Tick error (the Due
// lister failing) is logged and the loop continues — a transient store hiccup
// shouldn't kill the scheduler.
func (r *Runner) Loop(ctx context.Context) {
	interval := r.Interval
	if interval <= 0 {
		interval = time.Minute
	}
	t := time.NewTicker(interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			if err := r.Tick(ctx); err != nil {
				r.log().Warn("schedule tick failed", "error", err)
			}
		}
	}
}