X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/7088b82414bdf947a6f88918c0a7b2c7fb0c53ef..48d7bce28b18ecef4c52c98821675605464cecb9:/schedule.go diff --git a/schedule.go b/schedule.go index 3b4a4ee..a1eba1c 100644 --- a/schedule.go +++ b/schedule.go @@ -6,7 +6,7 @@ import ( "encoding/json" "errors" "fmt" - "log" + "log/slog" "os" "slices" "strconv" @@ -49,14 +49,15 @@ const ( ) // 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), } @@ -74,7 +75,18 @@ func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*Sc 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 } @@ -85,6 +97,13 @@ func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*Sc } } + if changed { + err = s.save() + if err != nil { + return nil, fmt.Errorf("record schedule owner: %w", err) + } + } + return s, nil } @@ -95,9 +114,9 @@ func openSchedules(path string, clock func() time.Time, logger *log.Logger) (*Sc // changed tells Run that a set or a remove changed the times. type Schedules struct { path string + owner string clock func() time.Time - logger *log.Logger - check time.Duration + logger *slog.Logger changed chan struct{} mutex sync.Mutex @@ -105,9 +124,9 @@ type Schedules struct { } // 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 scheduleCheck 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) @@ -115,11 +134,14 @@ func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error { 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, scheduleCheck)) timeout = timer.C } @@ -148,7 +170,8 @@ func (s *Schedules) Set(text string, at, cron *string) (string, error) { now := s.clock() schedule := Schedule{ - Text: text, + Text: text, + Owner: s.owner, } switch { @@ -252,8 +275,14 @@ func (s *Schedules) Remove(id int) (string, error) { // 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 { @@ -276,12 +305,13 @@ func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error { return 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) } @@ -293,6 +323,7 @@ func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error { // 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) @@ -363,6 +394,7 @@ func (s *Schedules) change() { // 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() @@ -394,13 +426,15 @@ type ScheduleFile struct { 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