Resume queued appends at their saved log ID

[onechat] / schedule.go

diff --git a/schedule.go b/schedule.go
index 3114ac7836b0843a4f3d2f99267003a5c5446645..5da024d39b0e4442235b3329695416cb9ac132eb 100644 (file)
--- a/schedule.go
+++ b/schedule.go
@@ -2,9 +2,11 @@ package main
 
 import (
        "cmp"
 
 import (
        "cmp"

+       "context"

        "encoding/json"
        "errors"
        "fmt"
        "encoding/json"
        "errors"
        "fmt"

+       "log/slog"

        "os"
        "slices"
        "strconv"
        "os"
        "slices"
        "strconv"


@@ -28,6 +30,14 @@ const (
        // tools and in the log.
        scheduleTimeLayout = "Monday 2006-01-02 15:04 -07:00"
 
        // tools and in the log.
        scheduleTimeLayout = "Monday 2006-01-02 15:04 -07:00"
 

+       // 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
+

        // scheduleLate is how long after its time a run is late.
        scheduleLate = time.Minute
 
        // scheduleLate is how long after its time a run is late.
        scheduleLate = time.Minute
 


@@ -39,11 +49,15 @@ const (
 )
 
 // openSchedules reads the schedules of the file at path. A missing file holds
 )
 
 // 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 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 *slog.Logger) (*Schedules, error) {

        s := &Schedules{
        s := &Schedules{

-               path:  path,
-               clock: clock,
+               path:    path,
+               clock:   clock,
+               logger:  logger,
+               check:   scheduleCheck,
+               changed: make(chan struct{}, 1),

        }
 
        data, err := os.ReadFile(path)
        }
 
        data, err := os.ReadFile(path)


@@ -77,16 +91,54 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
 // 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 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 the runs read
-// them.
+// 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 {
 type Schedules struct {

-       path  string
-       clock func() time.Time
+       path    string
+       clock   func() time.Time
+       logger  *slog.Logger
+       check   time.Duration
+       changed chan struct{}

 
        mutex sync.Mutex
        file  ScheduleFile
 }
 
 
        mutex sync.Mutex
        file  ScheduleFile
 }
 

+// 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.
+func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error {
+       for {
+               err := s.deliver(ctx, l)
+               if err != nil {
+                       return err
+               }
+
+               var (
+                       timer   *time.Timer
+                       timeout <-chan time.Time
+               )
+
+               wait, found := s.Wait(s.clock())
+               if found {
+                       timer = time.NewTimer(min(wait, s.check))
+                       timeout = timer.C
+               }
+
+               select {
+               case <-ctx.Done():
+                       return nil
+               case <-s.changed:
+               case <-timeout:
+               }
+
+               if timer != nil {
+                       timer.Stop()
+               }
+       }
+}
+

 // 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
 // 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


@@ -141,6 +193,8 @@ func (s *Schedules) Set(text string, at, cron *string) (string, error) {
                return "", err
        }
 
                return "", err
        }
 

+       s.change()
+

        return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
 }
 
        return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
 }
 


@@ -162,6 +216,14 @@ func (s *Schedules) List() string {
        return strings.Join(lines, "\n")
 }
 
        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()
 // Remove removes the schedule id and returns the answer of the tool.
 func (s *Schedules) Remove(id int) (string, error) {
        s.mutex.Lock()


@@ -181,12 +243,63 @@ func (s *Schedules) Remove(id int) (string, error) {
                return "", err
        }
 
                return "", err
        }
 

+       s.change()
+

        return fmt.Sprintf("Schedule %d removed.", id), nil
 }
 
        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
+// 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 {
+       now := s.clock()
+
+       var turns []Message
+
+       for _, schedule := range s.Due(now) {
+               message, err := l.Work(scheduleSource, schedule.Message(now), "")
+               if err != nil {
+                       return fmt.Errorf("log a schedule: %w", err)
+               }
+
+               // A stop of the daemon during the flush is no failure: the
+               // schedule stays due and runs at the next start.
+               err = l.Flush(ctx)
+               if ctx.Err() != nil {
+                       return nil
+               }
+
+               if err != nil {
+                       return fmt.Errorf("log a schedule: %w", err)
+               }
+
+               err = s.Done(schedule, 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)
+       }
+
+       l.Wake(turns...)
+
+       return nil
+}
+

 // Due returns the schedules whose time is not after now, by due time.
 func (s *Schedules) Due(now time.Time) []Schedule {
        var due []Schedule
 // 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)
        for _, schedule := range s.sorted() {
                if !schedule.Due.After(now) {
                        due = append(due, schedule)


@@ -246,9 +359,18 @@ func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
        return max(schedules[0].Due.Sub(now), 0), true
 }
 
        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:
+       }
+}
+

 // sorted returns a copy of the schedules by due time, then by ID.
 func (s *Schedules) sorted() []Schedule {
        s.mutex.Lock()
 // 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 := slices.Clone(s.file.Schedules)
        s.mutex.Unlock()
 


@@ -300,13 +422,19 @@ func (s Schedule) Message(now time.Time) string {
                s.Text,
        }
 
                s.Text,
        }
 

-       if now.Sub(s.Due) > scheduleLate {
+       if s.late(now) {

                lines = append(lines, scheduleLateLine)
        }
 
        return strings.Join(append(lines, scheduleEmptyReply), "\n")
 }
 
                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 {
 // repeats returns once for a one-time schedule, or else repeats and the cron
 // expression.
 func (s Schedule) repeats() string {