amadan

oficina / calendar Public

Merge branch 'fix/scheduled-reminder-delivery' into main

paul@keymail.dev 210cd32301376e1cb114d216003786b8d729d0df parent 9c44bc1parent 9c7f78b
18 files changed, +1693 −56
  • .gitignore +2 −0
  • cmd/calendar/main.go +0 −5
  • internal/calendar/app.go +23 −10
  • internal/calendar/handlers.go +1 −0
  • internal/calendar/handlers_jobs.go +15 −7
  • internal/calendar/routes.go +1 −1
  • internal/calendar/schedule_control.go +98 −0
  • internal/calendar/schedule_restart_test.go +172 −0
  • internal/calendar/schedule_restore.go +89 −0
  • internal/calendar/scheduled_recovery_test.go +256 −0
  • internal/calendar/scheduled_work.go +264 −0
  • internal/calendar/scheduled_work_test.go +297 −0
  • internal/calendar/store_reminder_next.go +11 −30
  • internal/calendartest/reminder_test.go +15 −3
  • internal/calendartest/schedule_control_test.go +116 −0
  • internal/calendartest/schedule_delivery_test.go +200 −0
  • internal/calendartest/schedule_progress_test.go +49 −0
  • internal/calendartest/schedule_recovery_test.go +84 −0

CI passed — run details

diff --git a/.gitignore b/.gitignore
index 1188c78..680725f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -41,3 +41,5 @@ Thumbs.db
# The JS test harness (Playwright) is installed, never vendored.
node_modules/
+
+writing/
diff --git a/cmd/calendar/main.go b/cmd/calendar/main.go
index abcafd7..bc2aa78 100644
--- a/cmd/calendar/main.go
+++ b/cmd/calendar/main.go
@@ -76,11 +76,6 @@ func main() {
opts.Mux = app.Handler()
opts.DBPath = ""
- // The scheduled-wake poll. The activator asks a running instance
- // this before hibernating it and wakes it at the answer, which is
- // what makes a reminder fire on a machine that is asleep almost all
- // the time. Without it a 09:00 reminder arrives whenever somebody
- // next happens to load a page.
opts.NextDue = app.NextDue
// The same page a handler's own error path renders, so a panic and
// a handled failure look identical to whoever hits one.
diff --git a/internal/calendar/app.go b/internal/calendar/app.go
index 86e14bf..62ab714 100644
--- a/internal/calendar/app.go
+++ b/internal/calendar/app.go
@@ -8,6 +8,7 @@ import (
"fmt"
"log/slog"
"net/http"
+ "sync"
"time"
rauth "github.com/carlosframework/rastrillo/auth"
@@ -42,7 +43,10 @@ type App struct {
auth *rauth.Auth
// h is held for the test harness; requests reach it through mux.
- h *handlers
+ h *handlers
+ scheduled *scheduledWork
+ closeOnce sync.Once
+ closeErr error
}
// New resolves every fail-closed setting, applies the schema, and wires
@@ -158,6 +162,14 @@ func New(cfg Config) (*App, error) {
}
h.homeEnabled = true
}
+ a.scheduled = newScheduledWork(store.nextDueAt, cfg.Env == "development", func(err error) {
+ logger.Error("calendar: scheduled work registration failed", "stage", "schedule.register", "err", scrubErr(err))
+ })
+ if err := a.scheduled.start(); err != nil {
+ d.Close()
+ return nil, err
+ }
+ h.scheduled = a.scheduled
a.mux = a.routes(h, authPlugin, list, origin)
return a, nil
}
@@ -165,12 +177,8 @@ func New(cfg Config) (*App, error) {
// NextDue answers the platform's scheduled-wake poll: the next moment
// this instance has work. rastrillo mounts it as GET /api/next-due,
// outside app middleware, and the activator asks before hibernating.
-//
-// Without it, a reminder due at 09:00 on an instance nobody visits
-// would arrive whenever somebody next loaded a page. With it, the
-// platform wakes the process at 09:00 to send it.
func (a *App) NextDue() time.Time {
- return a.store.NextDueAt(context.Background())
+ return a.scheduled.probe()
}
// Handler is the app's router, for rastrillo.Options.Mux and for
@@ -180,8 +188,13 @@ func (a *App) Handler() *http.ServeMux { return a.mux }
// Close releases the database. Safe to call on a partially built app.
func (a *App) Close() error {
- if a.db == nil {
- return nil
- }
- return a.db.Close()
+ a.closeOnce.Do(func() {
+ if a.scheduled != nil {
+ a.scheduled.close()
+ }
+ if a.db != nil {
+ a.closeErr = a.db.Close()
+ }
+ })
+ return a.closeErr
}
diff --git a/internal/calendar/handlers.go b/internal/calendar/handlers.go
index 871769d..8ff2dd6 100644
--- a/internal/calendar/handlers.go
+++ b/internal/calendar/handlers.go
@@ -10,6 +10,7 @@ import (
// handlers holds what every handler needs. One value, built in New.
type handlers struct {
+ scheduled *scheduledWork
homeEnabled bool
store *Store
logger *slog.Logger
diff --git a/internal/calendar/handlers_jobs.go b/internal/calendar/handlers_jobs.go
index 503db9c..0c22617 100644
--- a/internal/calendar/handlers_jobs.go
+++ b/internal/calendar/handlers_jobs.go
@@ -27,7 +27,12 @@ func (h *handlers) drainReminders(w http.ResponseWriter, r *http.Request) {
http.Error(w, "forbidden", http.StatusForbidden)
return
}
- sent, err := h.drain(r.Context(), time.Now())
+ started := time.Now()
+ sent, progress, err := h.drain(r.Context(), started)
+ registerErr := h.scheduled.finishDrain(started, progress, err)
+ if err == nil {
+ err = registerErr
+ }
if err != nil {
// 5xx so the platform retries. The occurrence key makes a
// retry safe.
@@ -49,18 +54,21 @@ const drainDeadline = 45 * time.Second
// the budget is gone, then booking mail with what is left (booking
// spec §8). Reminders get the first half so a queue of booking mail
// behind a slow relay cannot starve them.
-func (h *handlers) drain(ctx context.Context, now time.Time) (int, error) {
+func (h *handlers) drain(ctx context.Context, now time.Time) (int, int, error) {
+ progress := 0
deadline := now.Add(drainDeadline)
ctx, cancel := context.WithDeadline(ctx, deadline)
defer cancel()
if n, err := h.store.ExpireHolds(ctx, now, 200); err != nil {
- return 0, err
+ return 0, progress, err
} else if n > 0 {
+ progress += n
h.logger.Info("calendar: holds expired", "n", n)
}
if n, err := h.store.ForgetOldIPs(ctx, now, 500); err != nil {
- return 0, err
+ return 0, progress, err
} else if n > 0 {
+ progress += n
h.logger.Info("calendar: ip fingerprints forgotten", "n", n)
}
// Each phase's deadline is a CONTEXT deadline, not only a loop
@@ -73,14 +81,14 @@ func (h *handlers) drain(ctx context.Context, now time.Time) (int, error) {
sent, err := h.drainReminderQueue(rctx, now, half)
rcancel()
if err != nil && !errors.Is(err, context.DeadlineExceeded) {
- return sent, err
+ return sent, progress + sent, err
}
// The reminder phase running out is a handover, not a failure:
// booking mail still gets the second half.
mctx, mcancel := context.WithDeadline(ctx, deadline.Add(-keepReserve))
defer mcancel()
mailed, err := h.drainBookingMail(mctx, now, deadline.Add(-keepReserve))
- return sent + mailed, err
+ return sent + mailed, progress + sent + mailed, err
}
// drainReminderQueue sends every reminder that has come due, until the
@@ -109,7 +117,7 @@ func (h *handlers) drainReminderQueue(ctx context.Context, now, deadline time.Ti
// that says what it is beats an unbounded one that does not.
if seen[due.Queue.OccurrenceKey] {
h.logger.Error("calendar: the same occurrence was claimed twice in one drain; stopping the pass")
- return sent, nil
+ return sent, errors.New("calendar: the same occurrence was claimed twice in one drain; stopping the pass")
}
seen[due.Queue.OccurrenceKey] = true
diff --git a/internal/calendar/routes.go b/internal/calendar/routes.go
index 2c51ea0..019f81b 100644
--- a/internal/calendar/routes.go
+++ b/internal/calendar/routes.go
@@ -163,7 +163,7 @@ func (a *App) routes(h *handlers, authPlugin *rauth.Auth, list allowlist, origin
mux.Handle("GET /static/", assets.Handler())
mux.HandleFunc("GET /static/credits.css", serveCreditsCSS)
a.mountTick(mux, h)
- mux.Handle("/", r)
+ mux.Handle("/", a.scheduled.middleware(r))
return mux
}
diff --git a/internal/calendar/schedule_control.go b/internal/calendar/schedule_control.go
new file mode 100644
index 0000000..62882c5
--- /dev/null
+++ b/internal/calendar/schedule_control.go
@@ -0,0 +1,98 @@
+package calendar
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "net"
+ "net/http"
+ "os"
+
+ "github.com/carlosframework/rastrillo/carlos"
+)
+
+const scheduleUnsupportedMessage = "calendar: this platform does not support safe scheduled work; update the platform before starting Calendar"
+
+type unsupportedScheduling struct{}
+
+func (unsupportedScheduling) Error() string { return scheduleUnsupportedMessage }
+
+func scheduleControl(ctx context.Context, path string, value any) error {
+ socket := os.Getenv("CARLOS_CONTROL_SOCKET")
+ if socket == "" {
+ return carlos.ErrNotOnCarlos
+ }
+ token := os.Getenv("CARLOS_ADMIN_TOKEN")
+ if token == "" {
+ return carlos.ErrUnauthorized
+ }
+ transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", socket)
+ }}
+ defer transport.CloseIdleConnections()
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://carlos"+path, nil)
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Authorization", "Bearer "+token)
+ response, err := (&http.Client{Transport: transport}).Do(req)
+ if err != nil {
+ return err
+ }
+ defer response.Body.Close()
+ if response.StatusCode == http.StatusUnauthorized {
+ return carlos.ErrUnauthorized
+ }
+ if path == "/schedule-capabilities" && response.StatusCode == http.StatusNotFound {
+ return unsupportedScheduling{}
+ }
+ if response.StatusCode != http.StatusOK {
+ return &carlos.StatusError{Status: response.StatusCode, Name: scheduledWorkName}
+ }
+ return json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(value)
+}
+
+func scheduleCapability(ctx context.Context) error {
+ var capability struct {
+ RegistrationFence bool `json:"registration_fence"`
+ }
+ if err := scheduleControl(ctx, "/schedule-capabilities", &capability); err != nil {
+ return err
+ }
+ if !capability.RegistrationFence {
+ return unsupportedScheduling{}
+ }
+ return nil
+}
+
+type remoteSchedule struct {
+ Name string `json:"name"`
+ Kind string `json:"kind"`
+ Path string `json:"path"`
+ NextAt int64 `json:"next_at"`
+ DueAt int64 `json:"due_at"`
+ Attempts int `json:"attempts"`
+}
+
+func scheduleArmed(ctx context.Context) (remoteSchedule, error) {
+ var response struct {
+ Timers []remoteSchedule `json:"timers"`
+ }
+ if err := scheduleControl(ctx, "/schedule", &response); err != nil {
+ return remoteSchedule{}, err
+ }
+ for _, timer := range response.Timers {
+ if timer.Name == scheduledWorkName {
+ return timer, nil
+ }
+ }
+ return remoteSchedule{}, nil
+}
+
+func permanentScheduleError(err error) bool {
+ var unsupported unsupportedScheduling
+ var status *carlos.StatusError
+ return errors.As(err, &unsupported) || errors.Is(err, carlos.ErrNotOnCarlos) || errors.Is(err, carlos.ErrUnauthorized) ||
+ errors.Is(err, carlos.ErrDeclaredSchedule) || (errors.As(err, &status) && status.Status >= 400 && status.Status < 500 && status.Status != 408 && status.Status != 429)
+}
diff --git a/internal/calendar/schedule_restart_test.go b/internal/calendar/schedule_restart_test.go
new file mode 100644
index 0000000..14313aa
--- /dev/null
+++ b/internal/calendar/schedule_restart_test.go
@@ -0,0 +1,172 @@
+package calendar
+
+import (
+ "context"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+type remoteTimerFixture struct {
+ mu sync.Mutex
+ timer remoteSchedule
+ writes int
+}
+
+func restartRegistrar(t *testing.T, due time.Time, clock *atomic.Int64, remote *remoteTimerFixture) *scheduledWork {
+ t.Helper()
+ s := newScheduledWork(func(context.Context) (time.Time, error) { return due, nil }, false, nil)
+ s.now = func() time.Time { return time.Unix(clock.Load(), 0) }
+ s.capability = func(context.Context) error { return nil }
+ s.armed = func(context.Context) (remoteSchedule, error) {
+ remote.mu.Lock()
+ defer remote.mu.Unlock()
+ return remote.timer, nil
+ }
+ s.put = func(_ context.Context, _ string, at time.Time, _ string) error {
+ remote.mu.Lock()
+ defer remote.mu.Unlock()
+ remote.timer = remoteSchedule{NextAt: at.Unix(), DueAt: at.Unix(), Kind: "once", Path: "/jobs/reminders"}
+ remote.writes++
+ return nil
+ }
+ if err := s.start(); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(s.close)
+ return s
+}
+
+func TestDrainBackoffSurvivesRestartThroughPersistedLateness(t *testing.T) {
+ var clock atomic.Int64
+ clock.Store(time.Now().Unix())
+ due := time.Unix(clock.Load(), 0).Add(-4 * time.Minute)
+ remote := &remoteTimerFixture{}
+ first := restartRegistrar(t, due, &clock, remote)
+ if err := first.finishDrain(first.now(), 0, context.DeadlineExceeded); err != nil {
+ t.Fatal(err)
+ }
+ first.close()
+ original := remote.timer.NextAt
+ writes := remote.writes
+ clock.Add(30)
+ second := restartRegistrar(t, due, &clock, remote)
+ if remote.writes != writes || remote.timer.NextAt != original {
+ t.Fatal("boot replaced existing future backoff")
+ }
+ if got := second.probe(); got.Unix() != original {
+ t.Fatalf("boot probe bypassed backoff: %v", got)
+ }
+ second.close()
+ clock.Add(12 * 60)
+ third := restartRegistrar(t, due, &clock, remote)
+ if remote.writes != writes {
+ t.Fatal("boot replaced the delivery currently waking the process")
+ }
+ if err := third.finishDrain(third.now(), 0, context.DeadlineExceeded); err != nil {
+ t.Fatal(err)
+ }
+ if remote.timer.NextAt < clock.Load()+4*60 {
+ t.Fatalf("failure restarted at one minute: %d", remote.timer.NextAt-clock.Load())
+ }
+}
+
+func TestSuccessfulDrainWithoutProgressBacksOffButSweepProgressDoesNot(t *testing.T) {
+ due := time.Now().Add(-time.Minute)
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
+ now := s.now()
+ if err := s.finishDrain(now, 0, nil); err != nil {
+ t.Fatal(err)
+ }
+ calls := r.snapshot()
+ if calls[len(calls)-1].Before(now.Add(time.Minute)) {
+ t.Fatalf("no-progress success immediately rearmed: %v", calls)
+ }
+ if err := s.finishDrain(s.now(), 500, nil); err != nil {
+ t.Fatal(err)
+ }
+ calls = r.snapshot()
+ if calls[len(calls)-1].After(time.Now()) {
+ t.Fatal("real sweep progress delayed remaining backlog")
+ }
+}
+
+func TestLateCommittedReplacementIsRepairedWithoutResettingPlatformRetries(t *testing.T) {
+ var clock atomic.Int64
+ clock.Store(time.Now().Unix())
+ due := time.Unix(clock.Load(), 0).Add(time.Hour)
+ remote := &remoteTimerFixture{}
+ s := restartRegistrar(t, due, &clock, remote)
+ remote.mu.Lock()
+ remote.timer.NextAt = due.Add(time.Hour).Unix()
+ remote.timer.DueAt = remote.timer.NextAt
+ remote.mu.Unlock()
+ if _, err := s.synchronize(false, true); err != nil {
+ t.Fatal(err)
+ }
+ if remote.timer.NextAt != due.Unix() || remote.writes != 2 {
+ t.Fatalf("late stale write survived inspection: %+v", remote)
+ }
+ clock.Store(due.Unix() + 60)
+ remote.mu.Lock()
+ remote.timer.NextAt = clock.Load() + 8*60
+ remote.timer.Attempts = 4
+ remote.mu.Unlock()
+ if _, err := s.synchronize(false, true); err != nil {
+ t.Fatal(err)
+ }
+ if remote.writes != 2 {
+ t.Fatal("legitimate platform retry was reset")
+ }
+}
+
+func TestNextDueBusyRegistrationReturnsImmediately(t *testing.T) {
+ s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Now().Add(time.Hour), nil })
+ s.mu.Lock()
+ started := time.Now()
+ done := make(chan time.Time, 1)
+ go func() { done <- s.probe() }()
+ select {
+ case due := <-done:
+ if due.Before(started) || due.After(time.Now()) {
+ t.Fatal(due)
+ }
+ case <-time.After(100 * time.Millisecond):
+ s.mu.Unlock()
+ t.Fatal("probe waited for registration mutex")
+ }
+ s.mu.Unlock()
+}
+
+func TestNextDueHasOneOverallBudgetForSequentialControlCalls(t *testing.T) {
+ s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Now().Add(time.Hour), nil })
+ s.mu.Lock()
+ s.force = true
+ s.capability = func(ctx context.Context) error {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(1500 * time.Millisecond):
+ return nil
+ }
+ }
+ s.put = func(ctx context.Context, _ string, _ time.Time, _ string) error { <-ctx.Done(); return ctx.Err() }
+ s.query = func(ctx context.Context) (time.Time, error) {
+ select {
+ case <-ctx.Done():
+ return time.Time{}, ctx.Err()
+ case <-time.After(1500 * time.Millisecond):
+ return time.Now().Add(time.Hour), nil
+ }
+ }
+ s.mu.Unlock()
+ started := time.Now()
+ due := s.probe()
+ if elapsed := time.Since(started); elapsed >= 4500*time.Millisecond {
+ t.Fatalf("probe exceeded pre-stop budget: %v", elapsed)
+ }
+ if due.IsZero() || due.After(time.Now()) {
+ t.Fatalf("failed probe did not preserve wake: %v", due)
+ }
+}
diff --git a/internal/calendar/schedule_restore.go b/internal/calendar/schedule_restore.go
new file mode 100644
index 0000000..aaeee1b
--- /dev/null
+++ b/internal/calendar/schedule_restore.go
@@ -0,0 +1,89 @@
+package calendar
+
+import (
+ "context"
+ "errors"
+ "time"
+
+ "github.com/carlosframework/rastrillo/carlos"
+)
+
+func (s *scheduledWork) matches(remote remoteSchedule) bool {
+ if remote.NextAt <= 0 || remote.Kind != "once" || remote.Path != "/jobs/reminders" {
+ return false
+ }
+ occurrence := remote.DueAt
+ if occurrence == 0 {
+ occurrence = remote.NextAt
+ }
+ if occurrence != s.registeredAt.Unix() {
+ return false
+ }
+ return remote.NextAt == occurrence || (remote.Attempts > 0 && remote.NextAt <= s.now().Add(15*time.Minute).Unix())
+}
+
+func (s *scheduledWork) restore(parent context.Context, due time.Time) (bool, error) {
+ ctx, cancel := context.WithTimeout(parent, 2*time.Second)
+ err := s.capability(ctx)
+ cancel()
+ if errors.Is(err, carlos.ErrNotOnCarlos) && s.development {
+ return false, nil
+ }
+ if err != nil {
+ return false, err
+ }
+ ctx, cancel = context.WithTimeout(parent, 2*time.Second)
+ remote, err := s.armed(ctx)
+ cancel()
+ if err != nil {
+ return false, err
+ }
+ if due.IsZero() || remote.NextAt <= 0 || remote.Kind != "once" || remote.Path != "/jobs/reminders" {
+ return false, nil
+ }
+ occurrence := remote.DueAt
+ if occurrence == 0 {
+ occurrence = remote.NextAt
+ }
+ now := s.now()
+ if remote.NextAt != occurrence && (remote.Attempts <= 0 || remote.NextAt > now.Add(15*time.Minute).Unix()) {
+ return false, nil
+ }
+ if due.After(now) {
+ at := due
+ if latest := now.Add(carlos.MaxAhead - time.Minute); at.After(latest) {
+ at = latest
+ }
+ if occurrence != at.Unix() {
+ return false, nil
+ }
+ } else {
+ if occurrence < due.Unix() || remote.NextAt > now.Add(15*time.Minute).Unix() {
+ return false, nil
+ }
+ if remote.NextAt > now.Unix() {
+ s.notBefore = time.Unix(remote.NextAt, 0)
+ due = s.notBefore
+ }
+ }
+ s.known, s.force = true, false
+ s.registered, s.observed = due, due
+ s.registeredAt = time.Unix(occurrence, 0)
+ s.checked = now
+ return true, nil
+}
+
+func (s *scheduledWork) probe() time.Time {
+ if !s.mu.TryLock() {
+ return time.Now()
+ }
+ defer s.mu.Unlock()
+ ctx, cancel := context.WithTimeout(s.ctx, 4*time.Second)
+ defer cancel()
+ due, err := s.synchronizeLocked(ctx, false, true)
+ if err != nil {
+ s.request()
+ return s.now()
+ }
+ return due
+}
diff --git a/internal/calendar/scheduled_recovery_test.go b/internal/calendar/scheduled_recovery_test.go
new file mode 100644
index 0000000..2b546fd
--- /dev/null
+++ b/internal/calendar/scheduled_recovery_test.go
@@ -0,0 +1,256 @@
+package calendar
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func TestScheduledWorkTransientBootFailureServesAndRecovers(t *testing.T) {
+ for _, queryFailure := range []bool{false, true} {
+ t.Run(map[bool]string{false: "control", true: "query"}[queryFailure], func(t *testing.T) {
+ var fail atomic.Bool
+ fail.Store(true)
+ due := time.Now().Add(time.Hour)
+ s := newScheduledWork(func(context.Context) (time.Time, error) {
+ if queryFailure && fail.Load() {
+ return time.Time{}, context.DeadlineExceeded
+ }
+ return due, nil
+ }, false, nil)
+ s.retryDelay = time.Millisecond
+ s.capability = func(context.Context) error { return nil }
+ s.armed = func(context.Context) (remoteSchedule, error) { return remoteSchedule{}, nil }
+ written := make(chan struct{}, 1)
+ s.put = func(context.Context, string, time.Time, string) error {
+ if fail.Load() {
+ return context.DeadlineExceeded
+ }
+ select {
+ case written <- struct{}{}:
+ default:
+ }
+ return nil
+ }
+ if err := s.start(); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(s.close)
+ fail.Store(false)
+ select {
+ case <-written:
+ case <-time.After(time.Second):
+ t.Fatal("transient boot error never recovered")
+ }
+ })
+ }
+}
+
+func TestScheduledWorkCloseBeforeStartAndAfterPermanentFailure(t *testing.T) {
+ for _, start := range []bool{false, true} {
+ s := newScheduledWork(func(context.Context) (time.Time, error) { return time.Now(), nil }, false, nil)
+ s.capability = func(context.Context) error { return unsupportedScheduling{} }
+ if start {
+ if err := s.start(); err == nil {
+ t.Fatal("unsupported platform accepted")
+ }
+ }
+ done := make(chan struct{})
+ go func() { s.close(); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("Close hung")
+ }
+ }
+}
+
+func TestScheduledWorkDrainFailuresBackOffEveryRegistrationPath(t *testing.T) {
+ now := time.Now().Truncate(time.Second)
+ due := now.Add(-time.Minute)
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
+ s.mu.Lock()
+ s.now = func() time.Time { return now }
+ s.mu.Unlock()
+ for _, minutes := range []int{1, 2, 4, 8, 15, 15} {
+ if err := s.finishDrain(s.now(), 0, context.DeadlineExceeded); err != nil {
+ t.Fatal(err)
+ }
+ expected := now.Add(time.Duration(minutes) * time.Minute)
+ for _, inspect := range []bool{false, true} {
+ got, err := s.synchronize(false, inspect)
+ if err != nil || !got.Equal(expected) {
+ t.Fatalf("failure floor bypassed: %v %v, want %v", got, err, expected)
+ }
+ }
+ calls := r.snapshot()
+ if !calls[len(calls)-1].Equal(expected) {
+ t.Fatalf("backoff registration = %v, want %v", calls, expected)
+ }
+ }
+ if err := s.finishDrain(s.now(), 1, nil); err != nil {
+ t.Fatal(err)
+ }
+ calls := r.snapshot()
+ if !calls[len(calls)-1].Equal(now) {
+ t.Fatalf("success did not reset backoff: %v", calls)
+ }
+}
+
+func TestScheduledWorkEpochIsDueWorkNotCancellation(t *testing.T) {
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(0, 0), nil })
+ calls := r.snapshot()
+ if calls[0].Unix() <= 0 || calls[0].After(time.Now()) {
+ t.Fatalf("epoch disarmed the timer: %v", calls)
+ }
+ if _, err := s.sync(false); err != nil {
+ t.Fatal(err)
+ }
+ if len(r.snapshot()) != 1 {
+ t.Fatal("clamping defeated unchanged registration cache")
+ }
+}
+
+func TestScheduledWorkPublicRequestsDoNotWaitForControlSocket(t *testing.T) {
+ var due atomic.Int64
+ due.Store(time.Now().Add(time.Hour).Unix())
+ s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
+ entered := make(chan struct{}, 1)
+ s.mu.Lock()
+ s.put = func(ctx context.Context, _ string, _ time.Time, _ string) error {
+ select {
+ case entered <- struct{}{}:
+ default:
+ }
+ <-ctx.Done()
+ return ctx.Err()
+ }
+ s.mu.Unlock()
+ due.Add(1)
+ s.request()
+ select {
+ case <-entered:
+ case <-time.After(time.Second):
+ t.Fatal("worker did not start control request")
+ }
+ h := s.middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(204) }))
+ start := time.Now()
+ var wg sync.WaitGroup
+ for range 50 {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("POST", "/book", nil))
+ }()
+ }
+ done := make(chan struct{})
+ go func() { wg.Wait(); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(500 * time.Millisecond):
+ t.Fatal("public traffic queued behind registration")
+ }
+ if time.Since(start) > 500*time.Millisecond {
+ t.Fatal("public latency followed the control timeout")
+ }
+}
+
+func TestScheduledWorkReconcilesConsumedTimersWithoutResettingArmedRows(t *testing.T) {
+ var armed atomic.Bool
+ armed.Store(true)
+ due := time.Now().Add(-time.Minute)
+ s := newScheduledWork(func(context.Context) (time.Time, error) { return due, nil }, false, nil)
+ s.reconcileEvery = 5 * time.Millisecond
+ s.capability = func(context.Context) error { return nil }
+ s.armed = func(context.Context) (remoteSchedule, error) { return remoteSchedule{}, nil }
+ checks := make(chan struct{}, 20)
+ s.armed = func(context.Context) (remoteSchedule, error) {
+ select {
+ case checks <- struct{}{}:
+ default:
+ }
+ if !armed.Load() {
+ return remoteSchedule{}, nil
+ }
+ return remoteSchedule{NextAt: s.registeredAt.Unix(), DueAt: s.registeredAt.Unix(), Path: "/jobs/reminders", Kind: "once"}, nil
+ }
+ var writes atomic.Int64
+ s.put = func(context.Context, string, time.Time, string) error { writes.Add(1); armed.Store(true); return nil }
+ if err := s.start(); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(s.close)
+ for range 3 {
+ select {
+ case <-checks:
+ case <-time.After(time.Second):
+ t.Fatal("no periodic reconciliation")
+ }
+ }
+ if writes.Load() != 1 {
+ t.Fatal("armed or in-flight row was rearmed")
+ }
+ armed.Store(false)
+ deadline := time.After(time.Second)
+ for writes.Load() < 2 {
+ select {
+ case <-deadline:
+ t.Fatal("consumed row remained cached")
+ case <-time.After(time.Millisecond):
+ }
+ }
+ if got, err := s.synchronize(false, true); err != nil || !got.Equal(due) {
+ t.Fatalf("repaired due = %v %v", got, err)
+ }
+ if writes.Load() != 2 {
+ t.Fatal("healthy replacement registration was reset")
+ }
+}
+
+func TestScheduledWorkDetectsRollbackBeforeAnotherWrite(t *testing.T) {
+ var due atomic.Int64
+ due.Store(time.Now().Add(time.Hour).Unix())
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
+ s.mu.Lock()
+ s.capability = func(context.Context) error { return unsupportedScheduling{} }
+ s.mu.Unlock()
+ due.Add(60)
+ _, err := s.sync(false)
+ var unsupported unsupportedScheduling
+ if !errors.As(err, &unsupported) {
+ t.Fatalf("rollback was not distinguished: %v", err)
+ }
+ if len(r.snapshot()) != 1 {
+ t.Fatal("wrote unsafe rearm after platform rollback")
+ }
+}
+
+func TestScheduledWorkInspectionFailureDoesNotResetAnArmedRetry(t *testing.T) {
+ due := time.Now().Add(time.Hour)
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
+ var fail atomic.Bool
+ fail.Store(true)
+ s.mu.Lock()
+ s.armed = func(context.Context) (remoteSchedule, error) {
+ if fail.Load() {
+ return remoteSchedule{}, context.DeadlineExceeded
+ }
+ return remoteSchedule{NextAt: s.registeredAt.Unix(), DueAt: s.registeredAt.Unix(), Path: "/jobs/reminders", Kind: "once"}, nil
+ }
+ s.mu.Unlock()
+ if _, err := s.synchronize(false, true); err == nil {
+ t.Fatal("inspection failure swallowed")
+ }
+ fail.Store(false)
+ if _, err := s.synchronize(false, true); err != nil {
+ t.Fatal(err)
+ }
+ if len(r.snapshot()) != 1 {
+ t.Fatal("inspection retry replaced healthy armed row")
+ }
+}
diff --git a/internal/calendar/scheduled_work.go b/internal/calendar/scheduled_work.go
new file mode 100644
index 0000000..a8c6556
--- /dev/null
+++ b/internal/calendar/scheduled_work.go
@@ -0,0 +1,264 @@
+package calendar
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/rastrillo/carlos"
+)
+
+const scheduledWorkName = "calendar-next-due"
+
+type scheduledWork struct {
+ mu sync.Mutex
+ lifeMu sync.Mutex
+ started, closed bool
+ finishOnce sync.Once
+ ctx context.Context
+ cancel context.CancelFunc
+ done chan struct{}
+ wake chan struct{}
+ query func(context.Context) (time.Time, error)
+ put func(context.Context, string, time.Time, string) error
+ remove func(context.Context, string) error
+ capability func(context.Context) error
+ armed func(context.Context) (remoteSchedule, error)
+ now func() time.Time
+ report func(error)
+ development bool
+ known bool
+ registered time.Time
+ registeredAt time.Time
+ rawDue time.Time
+ observed time.Time
+ force bool
+ allowRestore bool
+ failures int
+ notBefore time.Time
+ checked time.Time
+ retryDelay time.Duration
+ reconcileEvery time.Duration
+}
+
+func newScheduledWork(query func(context.Context) (time.Time, error), development bool, report func(error)) *scheduledWork {
+ ctx, cancel := context.WithCancel(context.Background())
+ return &scheduledWork{ctx: ctx, cancel: cancel, done: make(chan struct{}), wake: make(chan struct{}, 1), query: query,
+ put: carlos.ScheduleAt, remove: carlos.ScheduleCancel, capability: scheduleCapability, armed: scheduleArmed, now: time.Now,
+ report: report, development: development, retryDelay: time.Second, reconcileEvery: time.Minute, allowRestore: true}
+}
+
+func (s *scheduledWork) start() error {
+ s.lifeMu.Lock()
+ defer s.lifeMu.Unlock()
+ if s.closed {
+ return context.Canceled
+ }
+ if s.started {
+ return nil
+ }
+ s.started = true
+ if _, err := s.sync(true); err != nil && permanentScheduleError(err) {
+ s.cancel()
+ s.finishOnce.Do(func() { close(s.done) })
+ return err
+ }
+ go s.retry()
+ return nil
+}
+
+func (s *scheduledWork) close() {
+ s.cancel()
+ s.lifeMu.Lock()
+ s.closed = true
+ if !s.started {
+ s.finishOnce.Do(func() { close(s.done) })
+ }
+ s.lifeMu.Unlock()
+ <-s.done
+ s.mu.Lock()
+ s.mu.Unlock()
+}
+
+func (s *scheduledWork) request() {
+ select {
+ case s.wake <- struct{}{}:
+ default:
+ }
+}
+
+func (s *scheduledWork) sync(force bool) (time.Time, error) {
+ due, err := s.synchronize(force, false)
+ if err != nil {
+ s.request()
+ }
+ return due, err
+}
+
+func (s *scheduledWork) finishDrain(start time.Time, progress int, drainErr error) error {
+ s.mu.Lock()
+ s.allowRestore = false
+ ctx, cancel := context.WithTimeout(s.ctx, 2*time.Second)
+ due, queryErr := s.query(ctx)
+ cancel()
+ if queryErr != nil {
+ due = s.rawDue
+ }
+ failed := drainErr != nil || queryErr != nil || (progress == 0 && !due.IsZero() && !due.After(start))
+ if !failed {
+ s.failures = 0
+ s.notBefore = time.Time{}
+ } else {
+ s.failures = min(s.failures+1, 5)
+ delay := min(time.Minute<<(s.failures-1), 15*time.Minute)
+ if !due.IsZero() {
+ delay = max(delay, min(max(s.now().Sub(due)/4, time.Minute), 15*time.Minute))
+ }
+ s.notBefore = s.now().Add(delay)
+ }
+ s.mu.Unlock()
+ _, err := s.sync(true)
+ if queryErr != nil {
+ return queryErr
+ }
+ return err
+}
+
+func (s *scheduledWork) synchronize(force, inspect bool) (due time.Time, err error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ ctx, cancel := context.WithTimeout(s.ctx, 4*time.Second)
+ defer cancel()
+ return s.synchronizeLocked(ctx, force, inspect)
+}
+
+func (s *scheduledWork) synchronizeLocked(ctx context.Context, force, inspect bool) (due time.Time, err error) {
+ if err := ctx.Err(); err != nil {
+ return s.observed, err
+ }
+ s.force = s.force || force
+ attempted := false
+ defer func() {
+ if err != nil {
+ s.force = s.force || attempted
+ if s.report != nil {
+ s.report(err)
+ }
+ }
+ }()
+ queryCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+ due, err = s.query(queryCtx)
+ cancel()
+ if err != nil {
+ return s.observed, err
+ }
+ s.rawDue = due
+ if !s.known && s.allowRestore {
+ if restored, restoreErr := s.restore(ctx, due); restoreErr != nil {
+ return due, restoreErr
+ } else if restored {
+ return s.observed, nil
+ }
+ }
+ if !s.notBefore.IsZero() && (due.IsZero() || due.Before(s.notBefore)) {
+ due = s.notBefore
+ }
+ s.observed = due
+ if s.known && !s.force && due.Equal(s.registered) {
+ if !inspect && s.now().Sub(s.checked) < s.reconcileEvery {
+ return due, nil
+ }
+ controlCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+ var armed remoteSchedule
+ armed, err = s.armed(controlCtx)
+ cancel()
+ if errors.Is(err, carlos.ErrNotOnCarlos) && s.development {
+ err = nil
+ armed = remoteSchedule{NextAt: s.registeredAt.Unix(), DueAt: s.registeredAt.Unix(), Path: "/jobs/reminders", Kind: "once"}
+ }
+ if err != nil {
+ return due, err
+ }
+ s.checked = s.now()
+ if due.IsZero() || s.matches(armed) {
+ return due, nil
+ }
+ s.force = true
+ }
+ capabilityCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+ err = s.capability(capabilityCtx)
+ cancel()
+ if errors.Is(err, carlos.ErrNotOnCarlos) && s.development {
+ err = nil
+ } else if err != nil {
+ return due, err
+ }
+ controlCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+ defer cancel()
+ attempted = true
+ if due.IsZero() {
+ err = s.remove(controlCtx, scheduledWorkName)
+ } else {
+ at := due
+ now := s.now()
+ if at.Before(now) || at.Unix() <= 0 {
+ at = now
+ }
+ if latest := now.Add(carlos.MaxAhead - time.Minute); at.After(latest) {
+ at = latest
+ }
+ err = s.put(controlCtx, scheduledWorkName, at, "/jobs/reminders")
+ if err == nil {
+ s.registeredAt = time.Unix(at.Unix(), 0)
+ }
+ }
+ if errors.Is(err, carlos.ErrNotOnCarlos) && s.development {
+ err = nil
+ }
+ if err == nil {
+ s.known, s.force = true, false
+ s.registered = due
+ s.checked = s.now()
+ }
+ return due, err
+}
+
+func (s *scheduledWork) retry() {
+ defer s.finishOnce.Do(func() { close(s.done) })
+ timer := time.NewTimer(s.reconcileEvery)
+ defer timer.Stop()
+ retrying := false
+ delay := s.retryDelay
+ for {
+ inspect := false
+ select {
+ case <-s.ctx.Done():
+ return
+ case <-s.wake:
+ if retrying {
+ continue
+ }
+ case <-timer.C:
+ inspect = !retrying
+ }
+ _, err := s.synchronize(false, inspect)
+ if err != nil {
+ retrying = true
+ timer.Reset(delay)
+ delay = min(delay*2, 30*time.Second)
+ } else {
+ retrying = false
+ delay = s.retryDelay
+ timer.Reset(s.reconcileEvery)
+ }
+ }
+}
+
+func (s *scheduledWork) middleware(next http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ defer s.request()
+ next.ServeHTTP(w, r)
+ })
+}
diff --git a/internal/calendar/scheduled_work_test.go b/internal/calendar/scheduled_work_test.go
new file mode 100644
index 0000000..96ae657
--- /dev/null
+++ b/internal/calendar/scheduled_work_test.go
@@ -0,0 +1,297 @@
+package calendar
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/rastrillo/carlos"
+)
+
+type scheduleRecorder struct {
+ mu sync.Mutex
+ calls []time.Time
+ fail bool
+}
+
+func (r *scheduleRecorder) write(ctx context.Context, at time.Time) error {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if r.fail {
+ return context.DeadlineExceeded
+ }
+ r.calls = append(r.calls, at)
+ return nil
+}
+
+func (r *scheduleRecorder) snapshot() []time.Time {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return append([]time.Time(nil), r.calls...)
+}
+
+func testScheduledWork(t *testing.T, query func(context.Context) (time.Time, error)) (*scheduledWork, *scheduleRecorder) {
+ t.Helper()
+ r := &scheduleRecorder{}
+ s := newScheduledWork(query, false, nil)
+ s.retryDelay = time.Millisecond
+ s.capability = func(context.Context) error { return nil }
+ s.armed = func(context.Context) (remoteSchedule, error) {
+ return remoteSchedule{NextAt: s.registeredAt.Unix(), DueAt: s.registeredAt.Unix(), Path: "/jobs/reminders", Kind: "once"}, nil
+ }
+ s.put = func(ctx context.Context, name string, at time.Time, path string) error {
+ if name != scheduledWorkName || path != "/jobs/reminders" {
+ t.Errorf("unexpected registration: %q %q", name, path)
+ }
+ return r.write(ctx, at)
+ }
+ s.remove = func(ctx context.Context, name string) error { return r.write(ctx, time.Time{}) }
+ if err := s.start(); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(s.close)
+ return s, r
+}
+
+func TestScheduledWorkKeepsUnchangedRequestsFromResettingRetries(t *testing.T) {
+ due := time.Now().Add(time.Hour).Truncate(time.Second)
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
+ for range 3 {
+ if _, err := s.sync(false); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if n := len(r.snapshot()); n != 1 {
+ t.Fatalf("ordinary requests wrote %d registrations, want 1", n)
+ }
+ if _, err := s.sync(true); err != nil {
+ t.Fatal(err)
+ }
+ if n := len(r.snapshot()); n != 2 {
+ t.Fatalf("completed delivery did not rearm the same due time: %d", n)
+ }
+}
+
+func TestScheduledWorkRetriesAForcedRearmWithoutAVisitor(t *testing.T) {
+ due := time.Now().Add(time.Hour)
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
+ r.mu.Lock()
+ r.fail = true
+ r.mu.Unlock()
+ if _, err := s.sync(true); err == nil {
+ t.Fatal("registration failure was swallowed")
+ }
+ r.mu.Lock()
+ r.fail = false
+ r.mu.Unlock()
+ deadline := time.After(time.Second)
+ for len(r.snapshot()) != 2 {
+ select {
+ case <-deadline:
+ t.Fatal("failed forced registration was never retried")
+ case <-time.After(time.Millisecond):
+ }
+ }
+ if _, err := s.sync(false); err != nil {
+ t.Fatal(err)
+ }
+ if n := len(r.snapshot()); n != 2 {
+ t.Fatalf("successful retry was repeated by ordinary request: %d", n)
+ }
+}
+
+func TestScheduledWorkReadsAfterOutOfOrderRequestCompletion(t *testing.T) {
+ var due atomic.Int64
+ due.Store(time.Now().Add(2 * time.Hour).Unix())
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
+ entered, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
+ first := s.middleware(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+ due.Store(time.Now().Add(time.Hour).Unix())
+ close(entered)
+ <-release
+ }))
+ go func() {
+ defer close(finished)
+ first.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("POST", "/first", nil))
+ }()
+ <-entered
+ earlier := time.Now().Add(time.Minute).Unix()
+ second := s.middleware(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { due.Store(earlier) }))
+ second.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("POST", "/second", nil))
+ close(release)
+ <-finished
+ if _, err := s.sync(false); err != nil {
+ t.Fatal(err)
+ }
+ calls := r.snapshot()
+ if len(calls) != 2 || calls[1].Unix() != earlier {
+ t.Fatalf("stale request replaced the earlier registration: %v", calls)
+ }
+}
+
+func TestScheduledWorkSerializesQueryWithRegistration(t *testing.T) {
+ var due atomic.Int64
+ due.Store(time.Now().Add(time.Hour).Unix())
+ entered, release := make(chan struct{}), make(chan struct{})
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
+ oldPut := s.put
+ var blocked atomic.Bool
+ s.put = func(ctx context.Context, name string, at time.Time, path string) error {
+ if blocked.CompareAndSwap(false, true) {
+ close(entered)
+ <-release
+ }
+ return oldPut(ctx, name, at, path)
+ }
+ firstDone := make(chan struct{})
+ go func() { defer close(firstDone); _, _ = s.sync(true) }()
+ <-entered
+ earlier := time.Now().Add(time.Minute).Unix()
+ due.Store(earlier)
+ secondDone := make(chan struct{})
+ go func() { defer close(secondDone); _, _ = s.sync(false) }()
+ // The second write must not overtake the first while its control call is held.
+ select {
+ case <-secondDone:
+ t.Fatal("query and registration were not serialized")
+ case <-time.After(10 * time.Millisecond):
+ }
+ close(release)
+ <-firstDone
+ <-secondDone
+ calls := r.snapshot()
+ if calls[len(calls)-1].Unix() != earlier {
+ t.Fatalf("last registration = %v", calls)
+ }
+}
+
+func TestScheduledWorkQueryFailureCannotCancelOrForgetWork(t *testing.T) {
+ var broken atomic.Bool
+ due := time.Now().Add(-time.Minute)
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) {
+ if broken.Load() {
+ return time.Time{}, context.DeadlineExceeded
+ }
+ return due, nil
+ })
+ broken.Store(true)
+ got, err := s.sync(false)
+ if err == nil || !got.Equal(due) {
+ t.Fatalf("failed query = %v, %v", got, err)
+ }
+ if len(r.snapshot()) != 1 {
+ t.Fatal("failed query cancelled the timer")
+ }
+}
+
+func TestScheduledWorkFarFutureAndCancellation(t *testing.T) {
+ var due atomic.Int64
+ due.Store(time.Now().Add(2 * carlos.MaxAhead).Unix())
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) {
+ if n := due.Load(); n != 0 {
+ return time.Unix(n, 0), nil
+ }
+ return time.Time{}, nil
+ })
+ got := r.snapshot()[0]
+ if got.After(time.Now().Add(carlos.MaxAhead)) || got.Before(time.Now().Add(carlos.MaxAhead-2*time.Minute)) {
+ t.Fatalf("far future recheck = %v", got)
+ }
+ if _, err := s.sync(false); err != nil {
+ t.Fatal(err)
+ }
+ if len(r.snapshot()) != 1 {
+ t.Fatal("far future registration drifted on an ordinary request")
+ }
+ due.Store(0)
+ if _, err := s.sync(false); err != nil {
+ t.Fatal(err)
+ }
+ if calls := r.snapshot(); len(calls) != 2 || !calls[1].IsZero() {
+ t.Fatalf("empty queue was not cancelled: %v", calls)
+ }
+}
+
+func TestScheduledWorkOffPlatformIsDevelopmentOnly(t *testing.T) {
+ t.Setenv("CARLOS_CONTROL_SOCKET", "")
+ for _, development := range []bool{false, true} {
+ s := newScheduledWork(func(context.Context) (time.Time, error) { return time.Time{}, nil }, development, nil)
+ err := s.start()
+ if development {
+ if err != nil {
+ t.Fatal(err)
+ }
+ s.close()
+ } else if !errors.Is(err, carlos.ErrNotOnCarlos) {
+ t.Fatalf("production startup = %v", err)
+ }
+ }
+}
+
+func TestScheduledWorkCloseCancelsAndJoinsRegistration(t *testing.T) {
+ s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Now().Add(time.Hour), nil })
+ entered := make(chan struct{})
+ s.mu.Lock()
+ s.put = func(ctx context.Context, name string, at time.Time, path string) error {
+ close(entered)
+ <-ctx.Done()
+ return ctx.Err()
+ }
+ s.mu.Unlock()
+ registered := make(chan struct{})
+ go func() { defer close(registered); _, _ = s.sync(true) }()
+ <-entered
+ closed := make(chan struct{})
+ go func() { s.close(); close(closed) }()
+ select {
+ case <-closed:
+ case <-time.After(time.Second):
+ t.Fatal("Close did not cancel registration")
+ }
+ <-registered
+ select {
+ case <-s.done:
+ default:
+ t.Fatal("retry worker outlived Close")
+ }
+ if _, err := s.sync(true); !errors.Is(err, context.Canceled) {
+ t.Fatalf("registration after Close = %v", err)
+ }
+}
+
+func TestScheduledWorkRepairsAnAmbiguousControlWrite(t *testing.T) {
+ var due atomic.Int64
+ original := time.Now().Add(time.Hour).Unix()
+ due.Store(original)
+ s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
+ oldPut := s.put
+ var fail atomic.Bool
+ fail.Store(true)
+ s.put = func(ctx context.Context, name string, at time.Time, path string) error {
+ if err := oldPut(ctx, name, at, path); err != nil {
+ return err
+ }
+ if fail.Load() {
+ return context.DeadlineExceeded
+ }
+ return nil
+ }
+ due.Store(original + 60)
+ if _, err := s.sync(false); err == nil {
+ t.Fatal("lost control response was ignored")
+ }
+ due.Store(original)
+ fail.Store(false)
+ if _, err := s.sync(false); err != nil {
+ t.Fatal(err)
+ }
+ calls := r.snapshot()
+ if len(calls) < 3 || calls[len(calls)-1].Unix() != original {
+ t.Fatalf("cached original hid ambiguous replacement: %v", calls)
+ }
+}
diff --git a/internal/calendar/store_reminder_next.go b/internal/calendar/store_reminder_next.go
index b8d6076..8030acc 100644
--- a/internal/calendar/store_reminder_next.go
+++ b/internal/calendar/store_reminder_next.go
@@ -44,43 +44,24 @@ func (s *Store) nextOccurrenceAfter(e Event, now int64, offsetMin int) (int64, s
// NextDueAt is what Options.NextDue answers: the earliest moment this
// instance has work.
//
-// The platform asks a running instance this before hibernating it, and
-// wakes it at the answer. That is what makes a reminder fire on a
-// machine that is asleep almost all the time — without it, a reminder
-// would arrive whenever the next person happened to load a page.
-//
// Zero means nothing is scheduled, and the platform hibernates
// indefinitely.
func (s *Store) NextDueAt(ctx context.Context) time.Time {
- var due time.Time
- consider := func(t time.Time) {
- if t.IsZero() {
- return
- }
- if due.IsZero() || t.Before(due) {
- due = t
- }
- }
- consider(s.nextClaimableReminder(ctx))
- consider(s.NextHoldExpiry(ctx))
- consider(s.NextBookingMailDue(ctx))
- consider(s.NextIPForget(ctx))
+ due, _ := s.nextDueAt(ctx)
return due
}
-// nextClaimableReminder is the earliest instant a reminder row may
-// actually be CLAIMED: its due time, or its retry time if deferred, or
-// the end of its lease if somebody holds it — whichever is latest for
-// that row, and then the earliest across rows. Advertising a bare
-// due_at woke the instance for a row nobody could take for five
-// minutes (booking spec §8, r4 A6).
-func (s *Store) nextClaimableReminder(ctx context.Context) time.Time {
+func (s *Store) nextDueAt(ctx context.Context) (time.Time, error) {
var at *int64
- err := s.g.WithContext(ctx).Raw(`
- SELECT MIN(MAX(due_at, COALESCE(retry_at, 0), COALESCE(claimed_at + ?, 0)))
- FROM reminder_queues WHERE fired_at IS NULL`, int64(claimLease.Seconds())).Scan(&at).Error
+ err := s.g.WithContext(ctx).Raw(`SELECT MIN(at) FROM (
+ SELECT MIN(MAX(due_at, COALESCE(retry_at, 0), COALESCE(claimed_at + ?, 0))) AS at
+ FROM reminder_queues WHERE fired_at IS NULL
+ UNION ALL SELECT MIN(hold_expires_at) FROM bookings WHERE state = ?
+ UNION ALL SELECT MIN(due_at) FROM booking_mails WHERE sent_at IS NULL
+ UNION ALL SELECT MIN(ip_forget_at) FROM bookings WHERE ip_fp IS NOT NULL
+ )`, int64(claimLease.Seconds()), BookingHeld).Scan(&at).Error
if err != nil || at == nil {
- return time.Time{}
+ return time.Time{}, err
}
- return time.Unix(*at, 0)
+ return time.Unix(*at, 0), nil
}
diff --git a/internal/calendartest/reminder_test.go b/internal/calendartest/reminder_test.go
index 2adf293..75d01b1 100644
--- a/internal/calendartest/reminder_test.go
+++ b/internal/calendartest/reminder_test.go
@@ -122,7 +122,7 @@ func TestNextDuePointsAtTheEarliestReminder(t *testing.T) {
app, session := signedIn(t, "someone@example.com")
h := app.Handler()
- if due := app.NextDue(); !due.IsZero() {
+ if due := settledNextDue(t, app); !due.IsZero() {
t.Errorf("NextDue on an empty calendar = %v, want the zero time", due)
}
@@ -136,7 +136,7 @@ func TestNextDuePointsAtTheEarliestReminder(t *testing.T) {
"repeat": {"none"},
})
- due := app.NextDue()
+ due := settledNextDue(t, app)
if due.IsZero() {
t.Fatal("NextDue is zero with a reminder queued; the platform would hibernate through it")
}
@@ -171,7 +171,7 @@ func TestARecurringReminderRequeuesItself(t *testing.T) {
// After firing, the rule's next occurrence must be queued —
// roughly a day later.
- due := app.NextDue()
+ due := settledNextDue(t, app)
if due.IsZero() {
t.Fatal("a repeating reminder fired and queued nothing; the series would remind once and stop")
}
@@ -181,3 +181,15 @@ func TestARecurringReminderRequeuesItself(t *testing.T) {
}
var _ = calendar.MaxReminderAttempts
+
+func settledNextDue(t *testing.T, app *calendar.App) time.Time {
+ t.Helper()
+ deadline := time.Now().Add(time.Second)
+ for {
+ due := app.NextDue()
+ if due.IsZero() || due.After(time.Now().Add(time.Second)) || time.Now().After(deadline) {
+ return due
+ }
+ time.Sleep(time.Millisecond)
+ }
+}
diff --git a/internal/calendartest/schedule_control_test.go b/internal/calendartest/schedule_control_test.go
new file mode 100644
index 0000000..7e5a9f4
--- /dev/null
+++ b/internal/calendartest/schedule_control_test.go
@@ -0,0 +1,116 @@
+package calendartest
+
+import (
+ "encoding/json"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+)
+
+type controlSchedules struct {
+ mu sync.Mutex
+ at time.Time
+ writes int
+ fail bool
+ unsupported bool
+}
+
+func controlForTest(t *testing.T) *controlSchedules {
+ t.Helper()
+ dir, err := os.MkdirTemp("/tmp", "cal-control-")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ socket := filepath.Join(dir, "control.sock")
+ listener, err := net.Listen("unix", socket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ state := &controlSchedules{}
+ server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ state.mu.Lock()
+ defer state.mu.Unlock()
+ if r.Header.Get("Authorization") != "Bearer test-token" {
+ http.Error(w, "forbidden", 403)
+ return
+ }
+ if state.fail {
+ http.Error(w, "retry", 503)
+ return
+ }
+ if r.URL.Path == "/schedule-capabilities" {
+ if state.unsupported {
+ w.WriteHeader(404)
+ return
+ }
+ json.NewEncoder(w).Encode(map[string]bool{"registration_fence": true})
+ return
+ }
+ if r.URL.Path == "/schedule" {
+ var timers []map[string]any
+ if !state.at.IsZero() {
+ timers = append(timers, map[string]any{"name": "calendar-next-due", "next_at": state.at.Unix(), "due_at": state.at.Unix(), "kind": "once", "path": "/jobs/reminders"})
+ }
+ json.NewEncoder(w).Encode(map[string]any{"timers": timers})
+ return
+ }
+ if r.URL.Path != "/schedule/calendar-next-due" {
+ w.WriteHeader(404)
+ return
+ }
+ switch r.Method {
+ case http.MethodPut:
+ var body struct {
+ At time.Time
+ Path string
+ }
+ if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+ t.Error(err)
+ w.WriteHeader(400)
+ return
+ }
+ if body.Path != "/jobs/reminders" {
+ t.Errorf("scheduled path = %q", body.Path)
+ }
+ state.at = body.At
+ case http.MethodDelete:
+ state.at = time.Time{}
+ default:
+ w.WriteHeader(405)
+ return
+ }
+ state.writes++
+ w.WriteHeader(204)
+ })}
+ go server.Serve(listener)
+ t.Cleanup(func() { server.Close() })
+ t.Setenv("CARLOS_CONTROL_SOCKET", socket)
+ t.Setenv("CARLOS_ADMIN_TOKEN", "test-token")
+ return state
+}
+
+func (s *controlSchedules) snapshot() (time.Time, int) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.at, s.writes
+}
+
+func (s *controlSchedules) waitArmed(t *testing.T) {
+ t.Helper()
+ deadline := time.After(2 * time.Second)
+ for {
+ if at, _ := s.snapshot(); !at.IsZero() {
+ return
+ }
+ select {
+ case <-deadline:
+ t.Fatal("timer was not armed")
+ case <-time.After(time.Millisecond):
+ }
+ }
+}
diff --git a/internal/calendartest/schedule_delivery_test.go b/internal/calendartest/schedule_delivery_test.go
new file mode 100644
index 0000000..cdbeba8
--- /dev/null
+++ b/internal/calendartest/schedule_delivery_test.go
@@ -0,0 +1,200 @@
+package calendartest
+
+import (
+ "context"
+ "errors"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ calendar "amadan.net/oficina/calendar/internal/calendar"
+ "github.com/carlosframework/rastrillo/carlos"
+)
+
+func TestPersistedWorkRegistersAtBootBeforeAnyRequest(t *testing.T) {
+ control := controlForTest(t)
+ dir := t.TempDir()
+ app := calendar.NewForTest(t, dir)
+ due := time.Now().Add(-time.Minute).Unix()
+ if _, err := app.DBForTest().Exec(`INSERT INTO reminder_queues (reminder_id, occurrence_key, due_at, attempts, late_by_sec) VALUES (1, 'boot', ?, 0, 0)`, due); err != nil {
+ t.Fatal(err)
+ }
+ if err := app.Close(); err != nil {
+ t.Fatal(err)
+ }
+ app = calendar.NewForTest(t, dir)
+ at, _ := control.snapshot()
+ if at.Unix() < due || at.After(time.Now()) || at.IsZero() {
+ t.Fatalf("boot did not register persisted work: got %v, want %d", at, due)
+ }
+ if body, _ := app.LastMailForTest(); body != "" {
+ t.Fatal("startup delivered mail outside a held request")
+ }
+}
+
+func TestScheduledHTTPDeliversWithoutAVisitorAndRearmsRecurrence(t *testing.T) {
+ control := controlForTest(t)
+ app, session := signedIn(t, "someone@example.com")
+ start := time.Now().UTC().Add(2 * time.Minute)
+ createEvent(t, app.Handler(), session, url.Values{
+ "title": {"Scheduled standup"}, "start": {start.Format("2006-01-02T15:04")}, "end": {start.Add(15 * time.Minute).Format("2006-01-02T15:04")},
+ "kind": {"meeting"}, "status": {"confirmed"}, "repeat": {"daily"},
+ })
+ expected := app.StoreForTest().NextDueAt(context.Background())
+ control.waitArmed(t)
+ before, _ := control.snapshot()
+ if before.Before(expected) || before.After(time.Now()) {
+ t.Fatalf("event POST failed to arm its overdue reminder: %v, expected %v", before, expected)
+ }
+ req := httptest.NewRequest(http.MethodPost, "/jobs/reminders", nil)
+ req.Header.Set("Authorization", "Bearer test-token")
+ req.Header.Set("X-Carlos-Schedule", "calendar-next-due")
+ rec := httptest.NewRecorder()
+ app.Handler().ServeHTTP(rec, req)
+ if rec.Code != http.StatusNoContent {
+ t.Fatalf("scheduled request = %d: %s", rec.Code, rec.Body.String())
+ }
+ body, _ := app.LastMailForTest()
+ if !strings.Contains(body, "Scheduled standup") {
+ t.Fatal("scheduled HTTP did not deliver the reminder")
+ }
+ next, _ := control.snapshot()
+ expected = app.StoreForTest().NextDueAt(context.Background())
+ if !next.Equal(expected) || !next.After(time.Now()) {
+ t.Fatalf("next recurrence not armed before completion: %v, expected %v", next, expected)
+ }
+}
+
+func TestPublicBookingPostRegistersItsHoldAndMail(t *testing.T) {
+ control := controlForTest(t)
+ app, _, path, _ := publishedPage(t)
+ _, starts := futureDaySlots(t, app.Handler(), path)
+ strangerPost(t, app.Handler(), path, holdForm(starts[0], "A", "a@example.com"))
+ expected := app.StoreForTest().NextDueAt(context.Background())
+ control.waitArmed(t)
+ at, _ := control.snapshot()
+ if expected.IsZero() || !at.Equal(expected) {
+ t.Fatalf("public booking did not register scheduled work: got %v, expected %v", at, expected)
+ }
+}
+
+func TestTickRegistrationFailureReturnsRetryAndRecoversWithoutAVisitor(t *testing.T) {
+ control := controlForTest(t)
+ app := calendar.NewForTest(t, t.TempDir())
+ due := time.Now().Add(time.Hour).Unix()
+ if _, err := app.DBForTest().Exec(`INSERT INTO reminder_queues (reminder_id, occurrence_key, due_at, attempts, late_by_sec) VALUES (1, 'retry', ?, 0, 0)`, due); err != nil {
+ t.Fatal(err)
+ }
+ control.mu.Lock()
+ control.fail = true
+ control.mu.Unlock()
+ rec := tick(t, app.Handler())
+ if rec.Code != http.StatusInternalServerError || strings.TrimSpace(rec.Body.String()) != "retry" {
+ t.Fatalf("registration failure acknowledged: %d %s", rec.Code, rec.Body.String())
+ }
+ control.mu.Lock()
+ control.fail = false
+ control.mu.Unlock()
+ deadline := time.After(4 * time.Second)
+ for {
+ at, _ := control.snapshot()
+ if at.Unix() == due {
+ break
+ }
+ select {
+ case <-deadline:
+ t.Fatal("registration did not recover without a visitor")
+ case <-time.After(10 * time.Millisecond):
+ }
+ }
+}
+
+func TestDatabaseFailureCannotCancelTheTimer(t *testing.T) {
+ control := controlForTest(t)
+ app := calendar.NewForTest(t, t.TempDir())
+ due := time.Now().Add(time.Hour).Unix()
+ if _, err := app.DBForTest().Exec(`INSERT INTO reminder_queues (reminder_id, occurrence_key, due_at, attempts, late_by_sec) VALUES (1, 'failure', ?, 0, 0)`, due); err != nil {
+ t.Fatal(err)
+ }
+ if got := app.NextDue(); got.Unix() != due {
+ t.Fatal(got)
+ }
+ if _, err := app.DBForTest().Exec(`DROP TABLE booking_mails`); err != nil {
+ t.Fatal(err)
+ }
+ got := app.NextDue()
+ at, _ := control.snapshot()
+ if got.IsZero() || got.After(time.Now()) || at.Unix() != due {
+ t.Fatalf("failed due query erased wake: probe %v, timer %v", got, at)
+ }
+}
+
+func TestProductionRefusesMissingControlSocket(t *testing.T) {
+ t.Setenv("CARLOS_CONTROL_SOCKET", "")
+ _, err := calendar.New(calendar.Config{DBPath: filepath.Join(t.TempDir(), "calendar.db"), Env: "production", Origin: calendar.TestOrigin,
+ InstanceKey: "01234567890123456789012345678901", DEK: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", AllowedAddresses: "person@example.com",
+ Mailer: &pickyMailer{}, Logger: slog.New(slog.NewTextHandler(io.Discard, nil))})
+ if !errors.Is(err, carlos.ErrNotOnCarlos) {
+ t.Fatalf("production startup error = %v, want missing platform scheduling", err)
+ }
+}
+
+func TestScheduledHTTPRearmsAFullSweepBacklog(t *testing.T) {
+ control := controlForTest(t)
+ app := calendar.NewForTest(t, t.TempDir())
+ due := time.Now().Add(-time.Minute).Unix()
+ _, err := app.DBForTest().Exec(`WITH RECURSIVE rows(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM rows WHERE n<501)
+ INSERT INTO bookings (page_id,event_id,token_hash,state,booker_address_fp,booker_zone,hold_expires_at,cancelled_by,ip_fp,ip_forget_at,reschedules,created_at,updated_at)
+ SELECT 0,0,CAST(n AS TEXT),'expired','','UTC',0,'','test',?,0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP FROM rows`, due)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec := tick(t, app.Handler()); rec.Code != http.StatusNoContent {
+ t.Fatalf("tick = %d", rec.Code)
+ }
+ var remaining int
+ if err := app.DBForTest().QueryRow(`SELECT COUNT(*) FROM bookings WHERE ip_fp IS NOT NULL`).Scan(&remaining); err != nil {
+ t.Fatal(err)
+ }
+ at, _ := control.snapshot()
+ if remaining != 1 || at.Unix() < due || at.After(time.Now()) || at.IsZero() {
+ t.Fatalf("backlog = %d, timer = %v", remaining, at)
+ }
+ if rec := tick(t, app.Handler()); rec.Code != http.StatusNoContent {
+ t.Fatalf("second tick = %d", rec.Code)
+ }
+ if at, _ := control.snapshot(); !at.IsZero() {
+ t.Fatalf("finished backlog remains armed at %v", at)
+ }
+}
+
+func TestScheduledHTTPKeepsRetryAndLeaseDeadlines(t *testing.T) {
+ control := controlForTest(t)
+ app := calendar.NewForTest(t, t.TempDir())
+ now := time.Now().Unix()
+ _, err := app.DBForTest().Exec(`INSERT INTO reminder_queues (reminder_id,occurrence_key,due_at,claimed_at,retry_at,attempts,late_by_sec) VALUES (1,'leased',?,?,?,0,0)`, now-60, now, now+600)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec := tick(t, app.Handler()); rec.Code != http.StatusNoContent {
+ t.Fatalf("tick = %d", rec.Code)
+ }
+ if at, _ := control.snapshot(); at.Unix() != now+600 {
+ t.Fatalf("retry time lost: %v", at)
+ }
+ if _, err := app.DBForTest().Exec(`UPDATE reminder_queues SET retry_at = NULL`); err != nil {
+ t.Fatal(err)
+ }
+ if rec := tick(t, app.Handler()); rec.Code != http.StatusNoContent {
+ t.Fatalf("second tick = %d", rec.Code)
+ }
+ if at, _ := control.snapshot(); at.Unix() != now+300 {
+ t.Fatalf("lease time lost: %v", at)
+ }
+}
diff --git a/internal/calendartest/schedule_progress_test.go b/internal/calendartest/schedule_progress_test.go
new file mode 100644
index 0000000..fe0811f
--- /dev/null
+++ b/internal/calendartest/schedule_progress_test.go
@@ -0,0 +1,49 @@
+package calendartest
+
+import (
+ "net/http"
+ "net/url"
+ "testing"
+ "time"
+
+ calendar "amadan.net/oficina/calendar/internal/calendar"
+)
+
+func TestSuccessfulClaimRaceWithoutProgressBacksOff(t *testing.T) {
+ control := controlForTest(t)
+ app := calendar.NewForTest(t, t.TempDir())
+ due := time.Now().Add(-time.Minute).Unix()
+ if _, err := app.DBForTest().Exec(`INSERT INTO reminder_queues (reminder_id,occurrence_key,due_at,attempts,late_by_sec) VALUES (1,'unclaimable',?,0,0)`, due); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := app.DBForTest().Exec(`CREATE TRIGGER decline_claim BEFORE UPDATE OF claimed_at ON reminder_queues BEGIN SELECT RAISE(IGNORE); END`); err != nil {
+ t.Fatal(err)
+ }
+ before := time.Now()
+ if rec := tick(t, app.Handler()); rec.Code != http.StatusNoContent {
+ t.Fatalf("claim race tick = %d", rec.Code)
+ }
+ if at, _ := control.snapshot(); at.Before(before.Add(time.Minute - time.Second)) {
+ t.Fatalf("no-progress successful drain rearmed immediately: %v", at)
+ }
+}
+
+func TestRepeatedOccurrenceGuardFailsEvenAfterASuccessfulSend(t *testing.T) {
+ control := controlForTest(t)
+ app, session := signedIn(t, "someone@example.com")
+ now := time.Now().UTC()
+ for _, offset := range []time.Duration{time.Hour, 2 * time.Hour} {
+ start := now.Add(offset)
+ createEvent(t, app.Handler(), session, url.Values{"title": {"Guarded reminder"}, "start": {start.Format("2006-01-02T15:04")}, "end": {start.Add(15 * time.Minute).Format("2006-01-02T15:04")}, "kind": {"meeting"}, "status": {"confirmed"}, "repeat": {"none"}})
+ }
+ if _, err := app.DBForTest().Exec(`UPDATE reminder_queues SET occurrence_key='same', due_at=?`, now.Add(-time.Minute).Unix()); err != nil {
+ t.Fatal(err)
+ }
+ before := time.Now()
+ if rec := tick(t, app.Handler()); rec.Code != http.StatusInternalServerError {
+ t.Fatalf("repeated occurrence was acknowledged despite a prior send: %d", rec.Code)
+ }
+ if at, _ := control.snapshot(); at.Before(before.Add(time.Minute - time.Second)) {
+ t.Fatalf("stalled occurrence did not back off: %v", at)
+ }
+}
diff --git a/internal/calendartest/schedule_recovery_test.go b/internal/calendartest/schedule_recovery_test.go
new file mode 100644
index 0000000..4c2699d
--- /dev/null
+++ b/internal/calendartest/schedule_recovery_test.go
@@ -0,0 +1,84 @@
+package calendartest
+
+import (
+ "context"
+ "net/http"
+ "testing"
+ "time"
+
+ calendar "amadan.net/oficina/calendar/internal/calendar"
+)
+
+func TestTransientControlFailureAtBootRecovers(t *testing.T) {
+ control := controlForTest(t)
+ control.mu.Lock()
+ control.fail = true
+ control.mu.Unlock()
+ app := calendar.NewForTest(t, t.TempDir())
+ due := time.Now().Add(time.Hour).Unix()
+ if _, err := app.DBForTest().Exec(`INSERT INTO reminder_queues (reminder_id,occurrence_key,due_at,attempts,late_by_sec) VALUES (1,'boot-retry',?,0,0)`, due); err != nil {
+ t.Fatal(err)
+ }
+ control.mu.Lock()
+ control.fail = false
+ control.mu.Unlock()
+ control.waitArmed(t)
+ if at, _ := control.snapshot(); at.Unix() != due {
+ t.Fatalf("recovered timer = %v", at)
+ }
+}
+
+func TestOldPlatformRefusesStartupBeforeWritingATimer(t *testing.T) {
+ control := controlForTest(t)
+ control.mu.Lock()
+ control.unsupported = true
+ control.mu.Unlock()
+ app, err := calendar.New(calendar.Config{Env: "development", DBPath: t.TempDir() + "/calendar.db", Origin: calendar.TestOrigin})
+ if err == nil {
+ app.Close()
+ t.Fatal("old platform accepted")
+ }
+ if _, writes := control.snapshot(); writes != 0 {
+ t.Fatalf("unsupported platform received %d writes", writes)
+ }
+}
+
+func TestReadOnlyDrainFailureBacksOffWithoutOrdinaryRequestsBypassingIt(t *testing.T) {
+ control := controlForTest(t)
+ app := calendar.NewForTest(t, t.TempDir())
+ _, err := app.DBForTest().Exec(`INSERT INTO bookings (page_id,event_id,token_hash,state,booker_address_fp,booker_zone,hold_expires_at,cancelled_by,ip_fp,ip_forget_at,reschedules,created_at,updated_at)
+ VALUES (0,0,'failure','expired','','UTC',0,'','test',?,0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)`, time.Now().Add(-time.Minute).Unix())
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = app.DBForTest().Exec(`CREATE TRIGGER prevent_forget BEFORE UPDATE OF ip_fp ON bookings BEGIN SELECT RAISE(FAIL,'test write failure'); END`)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, delay := range []time.Duration{time.Minute, 2 * time.Minute} {
+ before := time.Now()
+ rec := tick(t, app.Handler())
+ if rec.Code != http.StatusInternalServerError {
+ t.Fatalf("write failure status = %d", rec.Code)
+ }
+ due, _ := control.snapshot()
+ if due.Before(before.Add(delay-time.Second)) || due.After(time.Now().Add(delay)) {
+ t.Fatalf("drain failure did not back off %v: %v", delay, due)
+ }
+ if got := app.NextDue(); got.Before(due) {
+ t.Fatalf("NextDue bypassed failure floor: %v < %v", got, due)
+ }
+ if raw := app.StoreForTest().NextDueAt(context.Background()); raw.IsZero() || raw.After(time.Now()) {
+ t.Fatalf("fixture is not read-success/write-failure: %v", raw)
+ }
+ }
+ if _, err := app.DBForTest().Exec(`DROP TRIGGER prevent_forget`); err != nil {
+ t.Fatal(err)
+ }
+ if rec := tick(t, app.Handler()); rec.Code != http.StatusNoContent {
+ t.Fatalf("recovery = %d", rec.Code)
+ }
+ if at, _ := control.snapshot(); !at.IsZero() {
+ t.Fatalf("successful drain kept failure floor: %v", at)
+ }
+}