git.giacomolli.uk - onechat/blob - schedule.go

   1 package main
   2
   3 import (
   4         "cmp"
   5         "context"
   6         "encoding/json"
   7         "errors"
   8         "fmt"
   9         "log/slog"
  10         "os"
  11         "slices"
  12         "strconv"
  13         "strings"
  14         "sync"
  15         "time"
  16 )
  17
  18 const (
  19         // scheduleFileName is the file of the schedules in the directory of the
  20         // agent.
  21         scheduleFileName = "schedules.json"
  22
  23         // scheduleTextBytes is the most bytes of the text of a schedule.
  24         scheduleTextBytes = 4096
  25
  26         // scheduleAtLayout is the layout of the time of a one-time schedule.
  27         scheduleAtLayout = "2006-01-02 15:04"
  28
  29         // scheduleTimeLayout is the layout of the times in the answers of the
  30         // tools and in the log.
  31         scheduleTimeLayout = "Monday 2006-01-02 15:04 -07:00"
  32
  33         // scheduleSource is the source of the work events of the runs.
  34         scheduleSource = "schedule"
  35
  36         // scheduleClockWaitTimeout is the longest wait of Run before it looks
  37         // at the times again, so that it sees a suspend of the machine or a
  38         // change of the clock.
  39         scheduleClockWaitTimeout = time.Minute
  40
  41         // scheduleLate is how long after its time a run is late.
  42         scheduleLate = time.Minute
  43
  44         // scheduleEmptyReply is the last line of the message of a run.
  45         scheduleEmptyReply = "If nothing in this task needs the user, end your turn with an empty reply."
  46
  47         // scheduleLateLine tells the agent that a run is late.
  48         scheduleLateLine = "This run is late: the daemon was stopped at that time."
  49 )
  50
  51 // openSchedules reads the schedules of the file at path. A missing file holds
  52 // none. The owner names its agent. Schedules without an owner take this name. A
  53 // different stored owner is an error. The clock gives the time and the time
  54 // zone of the schedules, and logger gets one line for each run.
  55 func openSchedules(path, owner string, clock func() time.Time, logger *slog.Logger) (*Schedules, error) {
  56         s := &Schedules{
  57                 path:    path,
  58                 owner:   owner,
  59                 clock:   clock,
  60                 logger:  logger,
  61                 changed: make(chan struct{}, 1),
  62         }
  63
  64         data, err := os.ReadFile(path)
  65         if errors.Is(err, os.ErrNotExist) {
  66                 return s, nil
  67         }
  68
  69         if err != nil {
  70                 return nil, fmt.Errorf("read schedules: %w", err)
  71         }
  72
  73         err = json.Unmarshal(data, &s.file)
  74         if err != nil {
  75                 return nil, fmt.Errorf("%s: %w", path, err)
  76         }
  77
  78         changed := false
  79         for i := range s.file.Schedules {
  80                 schedule := &s.file.Schedules[i]
  81                 if schedule.Owner == "" {
  82                         schedule.Owner = owner
  83                         changed = true
  84                 }
  85
  86                 if schedule.Owner != owner {
  87                         return nil, fmt.Errorf("schedule %d belongs to %q, not %q", schedule.ID, schedule.Owner, owner)
  88                 }
  89
  90                 if schedule.Cron == "" {
  91                         continue
  92                 }
  93
  94                 _, err = scheduleParseCron(schedule.Cron)
  95                 if err != nil {
  96                         return nil, fmt.Errorf("%s: schedule %d: %w", path, schedule.ID, err)
  97                 }
  98         }
  99
 100         if changed {
 101                 err = s.save()
 102                 if err != nil {
 103                         return nil, fmt.Errorf("record schedule owner: %w", err)
 104                 }
 105         }
 106
 107         return s, nil
 108 }
 109
 110 // Schedules holds the schedules of the agent: tasks that start a turn at one
 111 // time or at the times of a cron expression. Each change writes the file whole
 112 // before it returns, so a schedule survives a restart. The mutex protects the
 113 // schedules, because the tools change them from a turn while Run reads them.
 114 // changed tells Run that a set or a remove changed the times.
 115 type Schedules struct {
 116         path    string
 117         owner   string
 118         clock   func() time.Time
 119         logger  *slog.Logger
 120         changed chan struct{}
 121
 122         mutex sync.Mutex
 123         file  ScheduleFile
 124 }
 125
 126 // Run starts the runs of the due schedules in l until ctx ends. It waits until
 127 // the first due time, at most scheduleClockWaitTimeout at a time, and also
 128 // wakes at a set or a remove. So a schedule that came due while the daemon was
 129 // stopped runs at the start. It returns the error of a write of the log or of
 130 // the file.
 131 func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error {
 132         for {
 133                 err := s.deliver(ctx, l)
 134                 if err != nil {
 135                         return err
 136                 }
 137
 138                 var (
 139                         timer   *time.Timer
 140                         timeout <-chan time.Time
 141                 )
 142
 143                 wait, found := s.Wait(s.clock())
 144                 if found {
 145                         timer = time.NewTimer(min(wait, scheduleClockWaitTimeout))
 146                         timeout = timer.C
 147                 }
 148
 149                 select {
 150                 case <-ctx.Done():
 151                         return nil
 152                 case <-s.changed:
 153                 case <-timeout:
 154                 }
 155
 156                 if timer != nil {
 157                         timer.Stop()
 158                 }
 159         }
 160 }
 161
 162 // Set adds a schedule with text and returns the answer of the tool. Exactly one
 163 // of at and cron is not nil. at is a local time in scheduleAtLayout after now,
 164 // and it goes through scheduleWallTime, so a time in the hour that the clock
 165 // repeats is its first occurrence. cron is an expression of scheduleParseCron.
 166 func (s *Schedules) Set(text string, at, cron *string) (string, error) {
 167         text = strings.TrimSpace(text)
 168         if text == "" || len(text) > scheduleTextBytes {
 169                 return "", fmt.Errorf("the text must have 1 to %d bytes", scheduleTextBytes)
 170         }
 171
 172         now := s.clock()
 173         schedule := Schedule{
 174                 Text:  text,
 175                 Owner: s.owner,
 176         }
 177
 178         switch {
 179         case (at == nil) == (cron == nil):
 180                 return "", errors.New("give either at or cron, and null for the other")
 181         case at != nil:
 182                 parsed, err := time.Parse(scheduleAtLayout, *at)
 183                 if err != nil {
 184                         return "", fmt.Errorf("at must be a local time such as %s", now.Format(scheduleAtLayout))
 185                 }
 186
 187                 schedule.Due = scheduleWallTime(parsed.Year(), parsed.Month(), parsed.Day(), parsed.Hour(), parsed.Minute(), now.Location())
 188                 if !schedule.Due.After(now) {
 189                         return "", fmt.Errorf("the time %s is not after now, %s", schedule.Due.Format(scheduleTimeLayout), now.Format(scheduleTimeLayout))
 190                 }
 191         default:
 192                 parsed, err := scheduleParseCron(*cron)
 193                 if err != nil {
 194                         return "", err
 195                 }
 196
 197                 next, found := parsed.Next(now)
 198                 if !found {
 199                         return "", fmt.Errorf("the cron expression %q matches no time in %d years", *cron, scheduleCronYears)
 200                 }
 201
 202                 schedule.Cron = parsed.String()
 203                 schedule.Due = next
 204         }
 205
 206         s.mutex.Lock()
 207         defer s.mutex.Unlock()
 208
 209         s.file.Next++
 210         schedule.ID = s.file.Next
 211         s.file.Schedules = append(s.file.Schedules, schedule)
 212         err := s.save()
 213         if err != nil {
 214                 return "", err
 215         }
 216
 217         s.change()
 218
 219         return fmt.Sprintf("Schedule %d set. Next: %s.", schedule.ID, schedule.Due.In(now.Location()).Format(scheduleTimeLayout)), nil
 220 }
 221
 222 // List returns the answer of the tool: one line for each schedule, by due time,
 223 // with its ID, its next time, its cron expression or once, and its text on one
 224 // line.
 225 func (s *Schedules) List() string {
 226         location := s.clock().Location()
 227         schedules := s.sorted()
 228         if len(schedules) == 0 {
 229                 return "No schedules."
 230         }
 231
 232         lines := make([]string, 0, len(schedules))
 233         for _, schedule := range schedules {
 234                 lines = append(lines, strconv.Itoa(schedule.ID)+" | "+schedule.Due.In(location).Format(scheduleTimeLayout)+" | "+cmp.Or(schedule.Cron, "once")+" | "+strings.Join(strings.Fields(schedule.Text), " "))
 235         }
 236
 237         return strings.Join(lines, "\n")
 238 }
 239
 240 // Len returns the number of schedules.
 241 func (s *Schedules) Len() int {
 242         s.mutex.Lock()
 243         defer s.mutex.Unlock()
 244
 245         return len(s.file.Schedules)
 246 }
 247
 248 // Remove removes the schedule id and returns the answer of the tool.
 249 func (s *Schedules) Remove(id int) (string, error) {
 250         s.mutex.Lock()
 251         defer s.mutex.Unlock()
 252
 253         index := slices.IndexFunc(s.file.Schedules, func(schedule Schedule) bool {
 254                 return schedule.ID == id
 255         })
 256
 257         if index < 0 {
 258                 return "", fmt.Errorf("no schedule %d", id)
 259         }
 260
 261         s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
 262         err := s.save()
 263         if err != nil {
 264                 return "", err
 265         }
 266
 267         s.change()
 268
 269         return fmt.Sprintf("Schedule %d removed.", id), nil
 270 }
 271
 272 // deliver logs the message of each due schedule in l, in due order, as a work
 273 // message of scheduleSource, and wakes the loop once with all of them. A
 274 // schedule ends its run with Done after its message is durable in the log. A
 275 // crash between the two makes the run repeat once. Each run logs one line with
 276 // the ID of the schedule and, for a late run, how late it is. The line holds no
 277 // text of the schedule.
 278 func (s *Schedules) deliver(ctx context.Context, l *TurnLoop) error {
 279         if l.agent.config.Name != s.owner {
 280                 return fmt.Errorf("schedule owner %q does not match agent %q", s.owner, l.agent.config.Name)
 281         }
 282
 283         now := s.clock()
 284
 285         var turns []Message
 286
 287         for _, schedule := range s.Due(now) {
 288                 message, err := l.Work(scheduleSource, schedule.Message(now), "")
 289                 if err != nil {
 290                         return fmt.Errorf("log a schedule: %w", err)
 291                 }
 292
 293                 // A stop of the daemon during the flush is no failure: the
 294                 // schedule stays due and runs at the next start.
 295                 err = l.Flush(ctx)
 296                 if ctx.Err() != nil {
 297                         return nil
 298                 }
 299
 300                 if err != nil {
 301                         return fmt.Errorf("log a schedule: %w", err)
 302                 }
 303
 304                 err = s.Done(schedule, now)
 305                 if err != nil {
 306                         return err
 307                 }
 308
 309                 attrs := []any{"id", schedule.ID}
 310                 if schedule.late(now) {
 311                         attrs = append(attrs, "late", now.Sub(schedule.Due))
 312                 }
 313
 314                 s.logger.Info("schedule ran", attrs...)
 315
 316                 turns = append(turns, message)
 317         }
 318
 319         l.Wake(turns...)
 320
 321         return nil
 322 }
 323
 324 // Due returns the schedules whose time is not after now, by due time.
 325 func (s *Schedules) Due(now time.Time) []Schedule {
 326         var due []Schedule
 327
 328         for _, schedule := range s.sorted() {
 329                 if !schedule.Due.After(now) {
 330                         due = append(due, schedule)
 331                 }
 332         }
 333
 334         return due
 335 }
 336
 337 // Done ends the run of schedule at now: a one-time schedule goes, and a
 338 // repeating schedule gets the next time after now. So a repeating schedule that
 339 // missed several times runs once. When a remove took the schedule first, Done
 340 // changes nothing.
 341 func (s *Schedules) Done(schedule Schedule, now time.Time) error {
 342         s.mutex.Lock()
 343         defer s.mutex.Unlock()
 344
 345         index := slices.IndexFunc(s.file.Schedules, func(held Schedule) bool {
 346                 return held.ID == schedule.ID
 347         })
 348
 349         if index < 0 {
 350                 return nil
 351         }
 352
 353         if schedule.Cron == "" {
 354                 s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
 355
 356                 return s.save()
 357         }
 358
 359         cron, err := scheduleParseCron(schedule.Cron)
 360         if err != nil {
 361                 return fmt.Errorf("schedule %d: %w", schedule.ID, err)
 362         }
 363
 364         next, found := cron.Next(now)
 365         if !found {
 366                 s.file.Schedules = slices.Delete(s.file.Schedules, index, index+1)
 367
 368                 return s.save()
 369         }
 370
 371         s.file.Schedules[index].Due = next
 372
 373         return s.save()
 374 }
 375
 376 // Wait returns how long after now the first schedule is due, or false when
 377 // there is no schedule.
 378 func (s *Schedules) Wait(now time.Time) (time.Duration, bool) {
 379         schedules := s.sorted()
 380         if len(schedules) == 0 {
 381                 return 0, false
 382         }
 383
 384         return max(schedules[0].Due.Sub(now), 0), true
 385 }
 386
 387 // change wakes Run after a set or a remove.
 388 func (s *Schedules) change() {
 389         select {
 390         case s.changed <- struct{}{}:
 391         default:
 392         }
 393 }
 394
 395 // sorted returns a copy of the schedules by due time, then by ID.
 396 func (s *Schedules) sorted() []Schedule {
 397         s.mutex.Lock()
 398
 399         schedules := slices.Clone(s.file.Schedules)
 400         s.mutex.Unlock()
 401
 402         slices.SortFunc(schedules, func(a, b Schedule) int {
 403                 return cmp.Or(a.Due.Compare(b.Due), cmp.Compare(a.ID, b.ID))
 404         })
 405
 406         return schedules
 407 }
 408
 409 // save writes the file whole, with mode 0600. The caller holds the mutex.
 410 func (s *Schedules) save() error {
 411         data, err := json.Marshal(s.file)
 412         if err == nil {
 413                 err = writeFileAtomic(s.path, append(data, '\n'))
 414         }
 415
 416         if err != nil {
 417                 return fmt.Errorf("save schedules: %w", err)
 418         }
 419
 420         return nil
 421 }
 422
 423 // ScheduleFile is the content of the file of the schedules: the ID of the last
 424 // schedule, so that an ID is never used again, and the schedules.
 425 type ScheduleFile struct {
 426         Next      int        `json:"next"`
 427         Schedules []Schedule `json:"schedules"`
 428 }
 429
 430 // Schedule is one schedule: its owner, its ID, the task for the agent, the cron
 431 // expression of a repeating schedule, empty for a one-time schedule, and its
 432 // next time.
 433 type Schedule struct {
 434         Owner string    `json:"owner"`
 435         ID    int       `json:"id"`
 436         Text  string    `json:"text"`
 437         Cron  string    `json:"cron"`
 438         Due   time.Time `json:"due"`
 439 }
 440
 441 // Message returns the text of a run at now for the log, with the times in the
 442 // location of now. The first line starts with [Schedule ID] and gives the due
 443 // time and once or the cron expression. Then come the text, the line about a
 444 // late run when now is more than scheduleLate after the due time, and the line
 445 // about the empty reply.
 446 func (s Schedule) Message(now time.Time) string {
 447         lines := []string{
 448                 "[Schedule " + strconv.Itoa(s.ID) + "] " + s.Due.In(now.Location()).Format(scheduleTimeLayout) + ", " + s.repeats(),
 449                 s.Text,
 450         }
 451
 452         if s.late(now) {
 453                 lines = append(lines, scheduleLateLine)
 454         }
 455
 456         return strings.Join(append(lines, scheduleEmptyReply), "\n")
 457 }
 458
 459 // late tells if a run at now is late: more than scheduleLate after the due
 460 // time.
 461 func (s Schedule) late(now time.Time) bool {
 462         return now.Sub(s.Due) > scheduleLate
 463 }
 464
 465 // repeats returns once for a one-time schedule, or else repeats and the cron
 466 // expression.
 467 func (s Schedule) repeats() string {
 468         if s.Cron == "" {
 469                 return "once"
 470         }
 471
 472         return "repeats " + s.Cron
 473 }