amadan

oficina / calendar Public

Register durable delivery for Calendar scheduled work

Co-Authored-By: Codex <noreply@openai.com>
Paul Campbell pushed by paul@keymail.dev 4af5cb62c78bde1c7a88ec9e82137d3136aaea3a parent 9c44bc1
10 files changed, +746 −46
  • .gitignore +2 −0
  • cmd/calendar/main.go +0 −5
  • internal/calendar/app.go +25 −10
  • internal/calendar/handlers.go +1 −0
  • internal/calendar/handlers_jobs.go +4 −0
  • internal/calendar/routes.go +1 −1
  • internal/calendar/scheduled_work.go +142 −0
  • internal/calendar/scheduled_work_test.go +290 −0
  • internal/calendar/store_reminder_next.go +11 −30
  • internal/calendartest/schedule_delivery_test.go +270 −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..649f937 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,12 @@ func New(cfg Config) (*App, error) {
}
h.homeEnabled = true
}
+ a.scheduled = newScheduledWork(store.nextDueAt, cfg.Env == "development", nil)
+ 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 +175,12 @@ 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())
+ due, err := a.scheduled.sync(false)
+ if err != nil {
+ return time.Now()
+ }
+ return due
}
// Handler is the app's router, for rastrillo.Options.Mux and for
@@ -180,8 +190,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..907da9a 100644
--- a/internal/calendar/handlers_jobs.go
+++ b/internal/calendar/handlers_jobs.go
@@ -28,6 +28,10 @@ func (h *handlers) drainReminders(w http.ResponseWriter, r *http.Request) {
return
}
sent, err := h.drain(r.Context(), time.Now())
+ _, registerErr := h.scheduled.sync(true)
+ if err == nil {
+ err = registerErr
+ }
if err != nil {
// 5xx so the platform retries. The occurrence key makes a
// retry safe.
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/scheduled_work.go b/internal/calendar/scheduled_work.go
new file mode 100644
index 0000000..2e52dfa
--- /dev/null
+++ b/internal/calendar/scheduled_work.go
@@ -0,0 +1,142 @@
+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
+ 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
+}
+
+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}
+}
+
+func (s *scheduledWork) start() error {
+ if _, err := s.sync(true); err != nil {
+ s.cancel()
+ return err
+ }
+ go s.retry()
+ return nil
+}
+
+func (s *scheduledWork) close() {
+ s.cancel()
+ <-s.done
+ s.mu.Lock()
+ s.mu.Unlock()
+}
+
+func (s *scheduledWork) sync(force bool) (due time.Time, err error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if err := s.ctx.Err(); err != nil {
+ return s.observed, err
+ }
+ s.force = s.force || force
+ defer func() {
+ if err != nil {
+ s.dirty = true
+ s.force = true
+ select {
+ case s.wake <- struct{}{}:
+ default:
+ }
+ if s.report != nil {
+ s.report(err)
+ }
+ }
+ }()
+ ctx, cancel := context.WithTimeout(s.ctx, 2*time.Second)
+ defer cancel()
+ due, err = s.query(ctx)
+ if err != nil {
+ return s.observed, err
+ }
+ s.observed = due
+ if s.known && !s.force && due.Equal(s.registered) {
+ s.dirty = false
+ return due, nil
+ }
+ if due.IsZero() {
+ err = s.remove(ctx, scheduledWorkName)
+ } else {
+ at := due
+ if latest := time.Now().Add(carlos.MaxAhead - time.Minute); at.After(latest) {
+ at = latest
+ }
+ err = s.put(ctx, scheduledWorkName, at, "/jobs/reminders")
+ }
+ if errors.Is(err, carlos.ErrNotOnCarlos) && s.development {
+ err = nil
+ }
+ if err == nil {
+ s.known, s.dirty, s.force = true, false, false
+ s.registered = due
+ }
+ return due, err
+}
+
+func (s *scheduledWork) retry() {
+ defer close(s.done)
+ for {
+ 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
+ }
+ delay = min(delay*2, 30*time.Second)
+ }
+ }
+}
+
+func (s *scheduledWork) middleware(next http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ defer func() { _, _ = s.sync(false) }()
+ 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..f686f5a
--- /dev/null
+++ b/internal/calendar/scheduled_work_test.go
@@ -0,0 +1,290 @@
+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.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
+ 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/schedule_delivery_test.go b/internal/calendartest/schedule_delivery_test.go
new file mode 100644
index 0000000..478d95b
--- /dev/null
+++ b/internal/calendartest/schedule_delivery_test.go
@@ -0,0 +1,270 @@
+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"
+
+ calendar "amadan.net/oficina/calendar/internal/calendar"
+ "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()
+ 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 {
+ 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())
+ before, _ := control.snapshot()
+ if !before.Equal(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())
+ 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 {
+ 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)
+ }
+}