"encoding/json"
"errors"
"fmt"
- "log"
+ "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
scheduleLateLine = "This run is late: the daemon was stopped at that time."
)
+// Schedules holds the schedules of the agent: tasks that start a turn at one
+// time or at the times of a cron expression. Each change writes the file whole
+// before it returns, so a schedule survives a restart. The mutex protects the
+// schedules, because the tools change them from a turn while Run reads them.
+// changed tells Run that a set or a remove changed the times.
+type Schedules struct {
+ path string
+ owner string
+ clock func() time.Time
+ logger *slog.Logger
+ changed chan struct{}
+
+ mutex sync.Mutex
+ file ScheduleFile
+}
+
+// ScheduleFile is the content of the file of the schedules: the ID of the last
+// schedule, so that an ID is never used again, and the schedules.
+type ScheduleFile struct {
+ Next int `json:"next"`
+ Schedules []Schedule `json:"schedules"`
+}
+
+// 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 {
+ Owner string `json:"owner"`
+ ID int `json:"id"`
+ Text string `json:"text"`
+ Cron string `json:"cron"`
+ Due time.Time `json:"due"`
+}
+
// 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, and logger
-// gets one line for each run.
-func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*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,
logger: logger,
- check: scheduleCheck,
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
}
-// Schedules holds the schedules of the agent: tasks that start a turn at one
-// time or at the times of a cron expression. Each change writes the file whole
-// before it returns, so a schedule survives a restart. The mutex protects the
-// schedules, because the tools change them from a turn while Run reads them.
-// changed tells Run that a set or a remove changed the times.
-type Schedules struct {
- path string
- clock func() time.Time
- logger *log.Logger
- check time.Duration
- changed chan struct{}
+// save writes the file whole, with mode 0600. The caller holds the mutex.
+func (s *Schedules) save() error {
+ data, err := json.Marshal(s.file)
+ if err == nil {
+ err = writeFileAtomic(s.path, append(data, '\n'))
+ }
- mutex sync.Mutex
- file ScheduleFile
+ if err != nil {
+ return fmt.Errorf("save schedules: %w", err)
+ }
+
+ return nil
}
// 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
}
}
}
-// Set adds a schedule with text and returns the answer of the tool. Exactly one
-// of at and cron is not nil. at is a local time in scheduleAtLayout after now,
-// and it goes through scheduleWallTime, so a time in the hour that the clock
-// repeats is its first occurrence. cron is an expression of scheduleParseCron.
-func (s *Schedules) Set(text string, at, cron *string) (string, error) {
- text = strings.TrimSpace(text)
- if text == "" || len(text) > scheduleTextBytes {
- return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes)
- }
-
- now := s.clock()
- schedule := Schedule{
- Text: text,
- }
-
- switch {
- case (at == nil) == (cron == nil):
- return "", errors.New("give either at or cron, and null for the other")
- case at != nil:
- parsed, err := time.Parse(scheduleAtLayout, *at)
- if err != nil {
- return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout))
- }
-
- schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location())
- if !schedule.Due.After(now) {
- return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout))
- }
- default:
- parsed, err := scheduleParseCron(*cron)
- if err != nil {
- return "", err
- }
-
- next, found := parsed.Next(now)
- if !found {
- return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears)
- }
-
- schedule.Cron = parsed.String()
- schedule.Due = next
- }
-
- s.mutex.Lock()
- defer s.mutex.Unlock()
-
- s.file.Next++
- schedule.ID = s.file.Next
- s.file.Schedules = append(s.file.Schedules, schedule)
- err := s.save()
- if err != nil {
- return "", err
- }
-
- s.change()
-
- return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
-}
-
-// List returns the answer of the tool: one line for each schedule, by due time,
-// with its ID, its next time, its cron expression or once, and its text on one
-// line.
-func (s *Schedules) List() string {
- location := s.clock().Location()
- schedules := s.sorted()
- if len(schedules) == 0 {
- return "No schedules."
- }
-
- lines := make([]string, 0, len(schedules))
- for _, schedule := range schedules {
- lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " "))
- }
-
- return strings.Join(lines, "\n")
-}
-
-// Len returns the number of schedules.
-func (s *Schedules) Len() int {
- s.mutex.Lock()
- defer s.mutex.Unlock()
-
- return len(s.file.Schedules)
-}
-
-// Remove removes the schedule id and returns the answer of the tool.
-func (s *Schedules) Remove(id int) (string, error) {
- s.mutex.Lock()
- defer s.mutex.Unlock()
-
- index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool {
- return schedule.ID == id
- })
-
- if index < 0 {
- return "", fmt.Errorf("no schedule %d", id)
- }
-
- s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
- err := s.save()
- if err != nil {
- return "", err
- }
-
- s.change()
-
- return fmt.Sprintf("Schedule %d removed.", id), nil
-}
-
// 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
// 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 {
err = s.Done(schedule, now)
if err != nil {
- return err
+ return fmt.Errorf("acknowledge schedule run: %w", err)
}
+ attrs := []any{"id", schedule.ID}
if schedule.late(now) {
- s.logger.Printf("schedule %d ran %s late", schedule.ID, durationText(now.Sub(schedule.Due)))
- } else {
- s.logger.Printf("schedule %d ran", schedule.ID)
+ attrs = append(attrs, "late", now.Sub(schedule.Due))
}
+ s.logger.Info("schedule ran", attrs...)
+
turns = append(turns, message)
}
return nil
}
+// Wait returns how long after now the first schedule is due, or false when
+// there is no schedule.
+func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
+ schedules := s.sorted()
+ if len(schedules) == 0 {
+ return 0, false
+ }
+
+ return max(schedules[0].Due.Sub(now), 0), true
+}
+
// 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)
if schedule.Cron == "" {
s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
- return s.save()
+ err := s.save()
+ if err != nil {
+ return fmt.Errorf("save completed one-time schedule: %w", err)
+ }
+
+ return nil
}
cron, err := scheduleParseCron(schedule.Cron)
if !found {
s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
- return s.save()
+ err = s.save()
+ if err != nil {
+ return fmt.Errorf("save exhausted schedule: %w", err)
+ }
+
+ return nil
}
s.file.Schedules[index].Due = next
- return s.save()
-}
-
-// Wait returns how long after now the first schedule is due, or false when
-// there is no schedule.
-func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
- schedules := s.sorted()
- if len(schedules) == 0 {
- return 0, false
+ err = s.save()
+ if err != nil {
+ return fmt.Errorf("save next schedule run: %w", err)
}
- return max(schedules[0].Due.Sub(now), 0), true
-}
-
-// change wakes Run after a set or a remove.
-func (s *Schedules) change() {
- select {
- case s.changed <- struct{}{}:
- default:
- }
+ return nil
}
// 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()
return schedules
}
-// save writes the file whole, with mode 0600. The caller holds the mutex.
-func (s *Schedules) save() error {
- data, err := json.Marshal(s.file)
- if err == nil {
- err = writeFileAtomic(s.path, append(data, '\n'))
+// Set adds a schedule with text and returns the answer of the tool. Exactly one
+// of at and cron is not nil. at is a local time in scheduleAtLayout after now,
+// and it goes through scheduleWallTime, so a time in the hour that the clock
+// repeats is its first occurrence. cron is an expression of scheduleParseCron.
+func (s *Schedules) Set(text string, at, cron *string) (string, error) {
+ text = strings.TrimSpace(text)
+ if text == "" || len(text) > scheduleTextBytes {
+ return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes)
+ }
+
+ now := s.clock()
+ schedule := Schedule{
+ Text: text,
+ Owner: s.owner,
+ }
+
+ switch {
+ case (at == nil) == (cron == nil):
+ return "", errors.New("give either at or cron, and null for the other")
+ case at != nil:
+ parsed, err := time.Parse(scheduleAtLayout, *at)
+ if err != nil {
+ return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout))
+ }
+
+ schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location())
+ if !schedule.Due.After(now) {
+ return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout))
+ }
+ default:
+ parsed, err := scheduleParseCron(*cron)
+ if err != nil {
+ return "", fmt.Errorf("parse schedule expression: %w", err)
+ }
+
+ next, found := parsed.Next(now)
+ if !found {
+ return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears)
+ }
+
+ schedule.Cron = parsed.String()
+ schedule.Due = next
}
+ s.mutex.Lock()
+ defer s.mutex.Unlock()
+
+ s.file.Next++
+ schedule.ID = s.file.Next
+ s.file.Schedules = append(s.file.Schedules, schedule)
+ err := s.save()
if err != nil {
- return fmt.Errorf("save schedules: %w", err)
+ return "", fmt.Errorf("save new schedule: %w", err)
}
- return nil
+ s.change()
+
+ return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
}
-// ScheduleFile is the content of the file of the schedules: the ID of the last
-// schedule, so that an ID is never used again, and the schedules.
-type ScheduleFile struct {
- Next int `json:"next"`
- Schedules []Schedule `json:"schedules"`
+// change wakes Run after a set or a remove.
+func (s *Schedules) change() {
+ select {
+ case s.changed <- struct{}{}:
+ default:
+ }
}
-// 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.
-type Schedule struct {
- ID int `json:"id"`
- Text string `json:"text"`
- Cron string `json:"cron"`
- Due time.Time `json:"due"`
+// List returns the answer of the tool: one line for each schedule, by due time,
+// with its ID, its next time, its cron expression or once, and its text on one
+// line.
+func (s *Schedules) List() string {
+ location := s.clock().Location()
+ schedules := s.sorted()
+ if len(schedules) == 0 {
+ return "No schedules."
+ }
+
+ lines := make([]string, 0, len(schedules))
+ for _, schedule := range schedules {
+ lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " "))
+ }
+
+ return strings.Join(lines, "\n")
+}
+
+// Len returns the number of schedules.
+func (s *Schedules) Len() int {
+ s.mutex.Lock()
+ defer s.mutex.Unlock()
+
+ return len(s.file.Schedules)
+}
+
+// Remove removes the schedule id and returns the answer of the tool.
+func (s *Schedules) Remove(id int) (string, error) {
+ s.mutex.Lock()
+ defer s.mutex.Unlock()
+
+ index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool {
+ return schedule.ID == id
+ })
+
+ if index < 0 {
+ return "", fmt.Errorf("no schedule %d", id)
+ }
+
+ s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
+ err := s.save()
+ if err != nil {
+ return "", fmt.Errorf("save removed schedule: %w", err)
+ }
+
+ s.change()
+
+ return fmt.Sprintf("Schedule %d removed.", id), nil
}
// Message returns the text of a run at now for the log, with the times in the
projects / onechat / blobdiff