"encoding/json"
"errors"
"fmt"
+ "log/slog"
"os"
"slices"
"strconv"
// scheduleSource is the source of the work events of the runs.
scheduleSource = "schedule"
- // scheduleCheck is the longest wait of Run before it looks at the times
- // again, so that it sees a suspend of the machine or a change of the
- // clock.
- scheduleCheck = time.Minute
+ // scheduleClockWaitTimeout is the longest wait of Run before it looks
+ // at the times again, so that it sees a suspend of the machine or a
+ // change of the clock.
+ scheduleClockWaitTimeout = time.Minute
// scheduleLate is how long after its time a run is late.
scheduleLate = time.Minute
)
// openSchedules reads the schedules of the file at path. A missing file holds
-// none. The clock gives the time and the time zone of the schedules.
-func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
+// none. The owner names its agent. Schedules without an owner take this name. A
+// different stored owner is an error. The clock gives the time and the time
+// zone of the schedules, and logger gets one line for each run.
+func openSchedules(path, owner string, clock func() time.Time, logger *slog.Logger) (*Schedules, error) {
s := &Schedules{
path: path,
+ owner: owner,
clock: clock,
- check: scheduleCheck,
+ logger: logger,
changed: make(chan struct{}, 1),
}
return nil, fmt.Errorf("%s: %w", path, err)
}
- for _, schedule := range s.file.Schedules {
+ changed := false
+ for i := range s.file.Schedules {
+ schedule := &s.file.Schedules[i]
+ if schedule.Owner == "" {
+ schedule.Owner = owner
+ changed = true
+ }
+
+ if schedule.Owner != owner {
+ return nil, fmt.Errorf("schedule %d belongs to %q, not %q", schedule.ID, schedule.Owner, owner)
+ }
+
if schedule.Cron == "" {
continue
}
}
}
+ if changed {
+ err = s.save()
+ if err != nil {
+ return nil, fmt.Errorf("record schedule owner: %w", err)
+ }
+ }
+
return s, nil
}
// changed tells Run that a set or a remove changed the times.
type Schedules struct {
path string
+ owner string
clock func() time.Time
- check time.Duration
+ logger *slog.Logger
changed chan struct{}
mutex sync.Mutex
}
// Run starts the runs of the due schedules in l until ctx ends. It waits until
-// the first due time, at most check at a time, and also wakes at a set or a
-// remove. So a schedule that came due while the daemon was stopped runs at the
-// start. It returns the error of a write of the log or of the file.
+// the first due time, at most scheduleClockWaitTimeout at a time, and also
+// wakes at a set or a remove. So a schedule that came due while the daemon was
+// stopped runs at the start. It returns the error of a write of the log or of
+// the file.
func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error {
for {
err := s.deliver(ctx, l)
return err
}
- var timer *time.Timer
- var timeout <-chan time.Time
+ var (
+ timer *time.Timer
+ timeout <-chan time.Time
+ )
+
wait, found := s.Wait(s.clock())
if found {
- timer = time.NewTimer(min(wait, s.check))
+ timer = time.NewTimer(min(wait, scheduleClockWaitTimeout))
timeout = timer.C
}
now := s.clock()
schedule := Schedule{
- Text: text,
+ Text: text,
+ Owner: s.owner,
}
switch {
// deliver logs the message of each due schedule in l, in due order, as a work
// message of scheduleSource, and wakes the loop once with all of them. A
// schedule ends its run with Done after its message is durable in the log. A
-// crash between the two makes the run repeat once.
+// crash between the two makes the run repeat once. Each run logs one line with
+// the ID of the schedule and, for a late run, how late it is. The line holds no
+// text of the schedule.
func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error {
+ if l.agent.config.Name != s.owner {
+ return fmt.Errorf("schedule owner %q does not match agent %q", s.owner, l.agent.config.Name)
+ }
+
now := s.clock()
+
var turns []Message
+
for _, schedule := range s.Due(now) {
message, err := l.Work(scheduleSource, schedule.Message(now), "")
if err != nil {
return err
}
+ attrs := []any{"id", schedule.ID}
+ if schedule.late(now) {
+ attrs = append(attrs, "late", now.Sub(schedule.Due))
+ }
+
+ s.logger.Info("schedule ran", attrs...)
+
turns = append(turns, message)
}
// Due returns the schedules whose time is not after now, by due time.
func (s *Schedules) Due(now time.Time) []Schedule {
var due []Schedule
+
for _, schedule := range s.sorted() {
if !schedule.Due.After(now) {
due = append(due, schedule)
// sorted returns a copy of the schedules by due time, then by ID.
func (s *Schedules) sorted() []Schedule {
s.mutex.Lock()
+
schedules := slices.Clone(s.file.Schedules)
s.mutex.Unlock()
Schedules []Schedule `json:"schedules"`
}
-// Schedule is one schedule: its ID, the task for the agent, the cron expression
-// of a repeating schedule, empty for a one-time schedule, and its next time.
+// Schedule is one schedule: its owner, its ID, the task for the agent, the cron
+// expression of a repeating schedule, empty for a one-time schedule, and its
+// next time.
type Schedule struct {
- ID int `json:"id"`
- Text string `json:"text"`
- Cron string `json:"cron"`
- Due time.Time `json:"due"`
+ Owner string `json:"owner"`
+ ID int `json:"id"`
+ Text string `json:"text"`
+ Cron string `json:"cron"`
+ Due time.Time `json:"due"`
}
// Message returns the text of a run at now for the log, with the times in the
s.Text,
}
- if now.Sub(s.Due) > scheduleLate {
+ if s.late(now) {
lines = append(lines, scheduleLateLine)
}
return strings.Join(append(lines, scheduleEmptyReply), "\n")
}
+// late tells if a run at now is late: more than scheduleLate after the due
+// time.
+func (s Schedule) late(now time.Time) bool {
+ return now.Sub(s.Due) > scheduleLate
+}
+
// repeats returns once for a one-time schedule, or else repeats and the cron
// expression.
func (s Schedule) repeats() string {
projects / onechat / blobdiff