amadan

oficina / calendar Public

Recover scheduled work safely across failures and hibernation

Paul Campbell pushed by paul@keymail.dev 9c7f78bd4a8a67b8614deb8478063000897f8212 parent 4af5cb6
13 files changed, +1084 −147
  • internal/calendar/app.go +4 −6
  • internal/calendar/handlers_jobs.go +12 −8
  • 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 +177 −55
  • internal/calendar/scheduled_work_test.go +7 −0
  • internal/calendartest/reminder_test.go +15 −3
  • internal/calendartest/schedule_control_test.go +116 −0
  • internal/calendartest/schedule_delivery_test.go +5 −75
  • internal/calendartest/schedule_progress_test.go +49 −0
  • internal/calendartest/schedule_recovery_test.go +84 −0

CI passed — run details

diff --git a/internal/calendar/app.go b/internal/calendar/app.go
index 649f937..62ab714 100644
--- a/internal/calendar/app.go
+++ b/internal/calendar/app.go
@@ -162,7 +162,9 @@ func New(cfg Config) (*App, error) {
}
h.homeEnabled = true
}
- a.scheduled = newScheduledWork(store.nextDueAt, cfg.Env == "development", nil)
+ 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
@@ -176,11 +178,7 @@ func New(cfg Config) (*App, error) {
// this instance has work. rastrillo mounts it as GET /api/next-due,
// outside app middleware, and the activator asks before hibernating.
func (a *App) NextDue() time.Time {
- due, err := a.scheduled.sync(false)
- if err != nil {
- return time.Now()
- }
- return due
+ return a.scheduled.probe()
}
// Handler is the app's router, for rastrillo.Options.Mux and for
diff --git a/internal/calendar/handlers_jobs.go b/internal/calendar/handlers_jobs.go
index 907da9a..0c22617 100644
--- a/internal/calendar/handlers_jobs.go
+++ b/internal/calendar/handlers_jobs.go
@@ -27,8 +27,9 @@ 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())
- _, registerErr := h.scheduled.sync(true)
+ started := time.Now()
+ sent, progress, err := h.drain(r.Context(), started)
+ registerErr := h.scheduled.finishDrain(started, progress, err)
if err == nil {
err = registerErr
}
@@ -53,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
@@ -77,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
@@ -113,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/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
index 2e52dfa..a8c6556 100644
--- a/internal/calendar/scheduled_work.go
+++ b/internal/calendar/scheduled_work.go
@@ -13,33 +13,56 @@ import (
const scheduledWorkName = "calendar-next-due"
type scheduledWork struct {
- mu sync.Mutex
- 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
- report func(error)
- development bool
- known bool
- registered time.Time
- observed time.Time
- dirty bool
- force bool
- retryDelay time.Duration
+ 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, report: report, development: development, retryDelay: time.Second}
+ 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 {
- if _, err := s.sync(true); err != nil {
+ 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()
@@ -48,95 +71,194 @@ func (s *scheduledWork) start() error {
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) sync(force bool) (due time.Time, err error) {
+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()
- if err := s.ctx.Err(); err != nil {
+ 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.dirty = true
- s.force = true
- select {
- case s.wake <- struct{}{}:
- default:
- }
+ s.force = s.force || attempted
if s.report != nil {
s.report(err)
}
}
}()
- ctx, cancel := context.WithTimeout(s.ctx, 2*time.Second)
- defer cancel()
- due, err = s.query(ctx)
+ 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) {
- s.dirty = false
- return due, nil
+ 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(ctx, scheduledWorkName)
+ err = s.remove(controlCtx, scheduledWorkName)
} else {
at := due
- if latest := time.Now().Add(carlos.MaxAhead - time.Minute); at.After(latest) {
+ 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(ctx, scheduledWorkName, at, "/jobs/reminders")
+ 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.dirty, s.force = true, false, false
+ s.known, s.force = true, false
s.registered = due
+ s.checked = s.now()
}
return due, err
}
func (s *scheduledWork) retry() {
- defer close(s.done)
+ 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:
- }
- delay := s.retryDelay
- for {
- timer := time.NewTimer(delay)
- select {
- case <-s.ctx.Done():
- timer.Stop()
- return
- case <-timer.C:
- }
- s.mu.Lock()
- dirty := s.dirty
- s.mu.Unlock()
- if !dirty {
- break
- }
- if _, err := s.sync(false); err == nil {
- break
+ 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 func() { _, _ = s.sync(false) }()
+ defer s.request()
next.ServeHTTP(w, r)
})
}
diff --git a/internal/calendar/scheduled_work_test.go b/internal/calendar/scheduled_work_test.go
index f686f5a..96ae657 100644
--- a/internal/calendar/scheduled_work_test.go
+++ b/internal/calendar/scheduled_work_test.go
@@ -40,6 +40,10 @@ func testScheduledWork(t *testing.T, query func(context.Context) (time.Time, err
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)
@@ -121,6 +125,9 @@ func TestScheduledWorkReadsAfterOutOfOrderRequestCompletion(t *testing.T) {
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)
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
index 478d95b..cdbeba8 100644
--- a/internal/calendartest/schedule_delivery_test.go
+++ b/internal/calendartest/schedule_delivery_test.go
@@ -2,18 +2,14 @@ package calendartest
import (
"context"
- "encoding/json"
"errors"
"io"
"log/slog"
- "net"
"net/http"
"net/http/httptest"
"net/url"
- "os"
"path/filepath"
"strings"
- "sync"
"testing"
"time"
@@ -21,74 +17,6 @@ import (
"github.com/carlosframework/rastrillo/carlos"
)
-type controlSchedules struct {
- mu sync.Mutex
- at time.Time
- writes int
- fail 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" || r.URL.Path != "/schedule/calendar-next-due" {
- http.Error(w, "forbidden", 403)
- return
- }
- if state.fail {
- http.Error(w, "retry", 503)
- 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 TestPersistedWorkRegistersAtBootBeforeAnyRequest(t *testing.T) {
control := controlForTest(t)
dir := t.TempDir()
@@ -102,7 +30,7 @@ func TestPersistedWorkRegistersAtBootBeforeAnyRequest(t *testing.T) {
}
app = calendar.NewForTest(t, dir)
at, _ := control.snapshot()
- if at.Unix() != due {
+ 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 != "" {
@@ -119,8 +47,9 @@ func TestScheduledHTTPDeliversWithoutAVisitorAndRearmsRecurrence(t *testing.T) {
"kind": {"meeting"}, "status": {"confirmed"}, "repeat": {"daily"},
})
expected := app.StoreForTest().NextDueAt(context.Background())
+ control.waitArmed(t)
before, _ := control.snapshot()
- if !before.Equal(expected) || before.After(time.Now()) {
+ 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)
@@ -148,6 +77,7 @@ func TestPublicBookingPostRegistersItsHoldAndMail(t *testing.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)
@@ -233,7 +163,7 @@ func TestScheduledHTTPRearmsAFullSweepBacklog(t *testing.T) {
t.Fatal(err)
}
at, _ := control.snapshot()
- if remaining != 1 || at.Unix() != due {
+ 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 {
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)
+ }
+}