Send the compaction view in block messages

[onechat] / schedule.go

diff --git a/schedule.go b/schedule.go
index 3114ac7836b0843a4f3d2f99267003a5c5446645..d27274bfe416f589b58ff5e0b047e3df129f1246 100644 (file)
--- a/schedule.go
+++ b/schedule.go
@@ -2,6 +2,7 @@ package main
 
 import (
        "cmp"
 
 import (
        "cmp"

+       "context"

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


@@ -28,6 +29,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
 


@@ -42,8 +51,10 @@ const (
 // none. The clock gives the time and the time zone of the schedules.
 func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
        s := &Schedules{
 // none. The clock gives the time and the time zone of the schedules.
 func openSchedules(path string, clock func() time.Time) (*Schedules, error) {
        s := &Schedules{

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

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


@@ -77,16 +88,50 @@ 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
+       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
+               var 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 +186,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
 }
 


@@ -181,9 +228,48 @@ 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.
+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
+               }
+
+               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


@@ -246,6 +332,14 @@ 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()