X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/d008b46cbfc6348255e9b9a1fc009f0e4bd6aa2a..f5c601f1801aaa276f8c335fcb51c373edc0b50c:/schedule.go diff --git a/schedule.go b/schedule.go index d27274b..5fa35da 100644 --- a/schedule.go +++ b/schedule.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "os" "slices" "strconv" @@ -32,10 +33,10 @@ const ( // 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 @@ -47,13 +48,50 @@ const ( 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. -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), } @@ -71,7 +109,18 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) { 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 } @@ -82,28 +131,21 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) { } } - 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 - check time.Duration - changed chan struct{} + if changed { + err = s.save() + if err != nil { + return nil, fmt.Errorf("record schedule owner: %w", err) + } + } - mutex sync.Mutex - file ScheduleFile + return s, 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) @@ -111,11 +153,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, scheduleClockWaitTimeout)) timeout = timer.C } @@ -144,7 +189,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 { @@ -209,6 +255,14 @@ func (s *Schedules) List() string { 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() @@ -236,10 +290,18 @@ func (s *Schedules) Remove(id int) (string, error) { // 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 { @@ -262,6 +324,13 @@ func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error { 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) } @@ -273,6 +342,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) @@ -343,6 +413,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() @@ -367,22 +438,6 @@ func (s *Schedules) save() error { return 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"` -} - -// 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"` -} - // Message returns the text of a run at now for the log, with the times in the // location of now. The first line starts with [Schedule ID] and gives the due // time and once or the cron expression. Then come the text, the line about a @@ -394,13 +449,19 @@ func (s Schedule) Message(now time.Time) string { 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 {