Name private helpers for their owning feature

[onechat] / schedule.go

diff --git a/schedule.go b/schedule.go
index 3114ac7836b0843a4f3d2f99267003a5c5446645..7eb747ff6d75205cf9a7b8c11b0ee4bb45dec370 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"
+
+       // 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
 
        // scheduleLate is how long after its time a run is late.
        scheduleLate = time.Minute
 


@@ -38,12 +48,51 @@ const (
        scheduleLateLine = "This run is late: the daemon was stopped at that time."
 )
 
        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
 // 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{
        s := &Schedules{

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

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


@@ -60,7 +109,18 @@ func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
                return nil, fmt.Errorf("%s: %w", path, err)
        }
 
                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 schedule.Cron == "" {
                        continue
                }


@@ -71,122 +131,133 @@ 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 the runs read
-// them.
-type Schedules struct {
-       path  string
-       clock func() time.Time
+       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

 }
 
 }
 

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

        }
 
        }
 

-       now := s.clock()
-       schedule := Schedule{
-               Text: text,
+       if err != nil {
+               return fmt.Errorf("save schedules: %w", err)

        }
 
        }
 

-       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)
+       return nil
+}
+
+// Run starts the runs of the due schedules in l until ctx ends. It waits until
+// 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)

                if err != nil {
                if err != nil {

-                       return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout))
+                       return err

                }
 
                }
 

-               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))
+               var (
+                       timer   *time.Timer
+                       timeout <-chan time.Time
+               )
+
+               wait, found := s.Wait(s.clock())
+               if found {
+                       timer = time.NewTimer(min(wait, scheduleClockWaitTimeout))
+                       timeout = timer.C

                }
                }

-       default:
-               parsed, err := scheduleParseCron(*cron)
-               if err != nil {
-                       return "", err
+
+               select {
+               case <-ctx.Done():
+                       return nil
+               case <-s.changed:
+               case <-timeout:

                }
 
                }
 

-               next, found := parsed.Next(now)
-               if !found {
-                       return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears)
+               if timer != nil {
+                       timer.Stop()

                }
                }

+       }
+}

 
 

-               schedule.Cron = parsed.String()
-               schedule.Due = next
+// 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 {
+       if l.agent.config.Name != s.owner {
+               return fmt.Errorf("schedule owner %q does not match agent %q", s.owner, l.agent.config.Name)

        }
 
        }
 

-       s.mutex.Lock()
-       defer s.mutex.Unlock()
+       now := s.clock()

 
 

-       s.file.Next++
-       schedule.ID = s.file.Next
-       s.file.Schedules = append(s.file.Schedules, schedule)
-       err := s.save()
-       if err != nil {
-               return "", err
-       }
+       var turns []Message

 
 

-       return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
-}
+       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)
+               }

 
 

-// 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."
-       }
+               // 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
+               }

 
 

-       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), " "))
-       }
+               if err != nil {
+                       return fmt.Errorf("log a schedule: %w", err)
+               }

 
 

-       return strings.Join(lines, "\n")
-}
+               err = s.Done(schedule, now)
+               if err != nil {
+                       return err
+               }

 
 

-// 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()
+               attrs := []any{"id", schedule.ID}
+               if schedule.late(now) {
+                       attrs = append(attrs, "late", now.Sub(schedule.Due))
+               }

 
 

-       index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool {
-               return schedule.ID == id
-       })
+               s.logger.Info("schedule ran", attrs...)

 
 

-       if index < 0 {
-               return "", fmt.Errorf("no schedule %d", id)
+               turns = append(turns, message)

        }
 
        }
 

-       s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
-       err := s.save()
-       if err != nil {
-               return "", err
+       l.Wake(turns...)
+
+       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 fmt.Sprintf("Schedule %d removed.", id), nil
+       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
 }
 
 // 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)


@@ -235,20 +306,10 @@ func (s *Schedules) Done(schedule Schedule, now time.Time) error {
        return s.save()
 }
 
        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
-       }
-
-       return max(schedules[0].Due.Sub(now), 0), true
-}
-

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


@@ -259,34 +320,122 @@ func (s *Schedules) sorted() []Schedule {
        return schedules
 }
 
        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 "", 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 {
        if err != nil {

-               return fmt.Errorf("save schedules: %w", err)
+               return "", 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 "", 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
 }
 
 // Message returns the text of a run at now for the log, with the times in the


@@ -300,13 +449,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 {