amadan

carlos / platform Public

Fence stale timer delivery and advertise safe rearming

Paul Campbell pushed by paul@keymail.dev 15477ea3bf8c692573c7ae12593318e28b7b3c47 parent 0310ae7
5 files changed, +221 −10
  • internal/registry/timers.go +1 −3
  • internal/schedule/control.go +12 −1
  • internal/schedule/control_test.go +63 −0
  • internal/schedule/runner.go +17 −3
  • internal/schedule/runner_test.go +128 −3

CI failed — run details

diff --git a/internal/registry/timers.go b/internal/registry/timers.go
index 4ea3281c5..cec61318c 100644
--- a/internal/registry/timers.go
+++ b/internal/registry/timers.go
@@ -42,9 +42,7 @@ func scanTimer(row interface{ Scan(...any) error }) (Timer, error) {
return t, err
}
-// UpsertTimer inserts t, or updates an existing row's spec/path — moving
-// next_at (and due_at with it) only when spec or path changed or t.NextAt
-// is non-zero — and never touches the runner's columns on update.
+// UpsertTimer inserts t, or updates an existing row's spec/path. A changed spec/path or non-zero NextAt starts a new registration and resets attempts; unchanged declarations preserve delivery state.
func (r *Registry) UpsertTimer(t Timer) error {
const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)`
res, err := r.sql.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,lower(hex(randomblob(16))))
diff --git a/internal/schedule/control.go b/internal/schedule/control.go
index 14e663f5f..acd30bd2f 100644
--- a/internal/schedule/control.go
+++ b/internal/schedule/control.go
@@ -56,6 +56,13 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri
}
}
+ mux.HandleFunc("GET /schedule-capabilities", authed(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(struct {
+ RegistrationFence bool `json:"registration_fence"`
+ }{RegistrationFence: true})
+ }))
+
mux.HandleFunc("PUT /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
if !ValidName(name) {
@@ -112,7 +119,11 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri
}
}
sp := Spec{Kind: KindOnce, At: atT.UTC()}
- t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: atT.Unix(), DueAt: atT.Unix(), CreatedAt: now().Unix()}
+ next := atT.Unix()
+ if next <= 0 {
+ next = max(int64(1), now().Unix())
+ }
+ t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: next, DueAt: atT.Unix(), CreatedAt: now().Unix()}
if err := store.UpsertTimer(t); err != nil {
http.Error(w, "registry error", http.StatusInternalServerError)
return
diff --git a/internal/schedule/control_test.go b/internal/schedule/control_test.go
index 23909e324..d58667c2b 100644
--- a/internal/schedule/control_test.go
+++ b/internal/schedule/control_test.go
@@ -193,3 +193,66 @@ func TestControlCapsOncePerHost(t *testing.T) {
}
func itoa(n int64) string { return strconv.FormatInt(n, 10) }
+
+func TestControlScheduleCapabilities(t *testing.T) {
+ _, mux := controlFixture(t)
+ for _, auth := range []string{"", "Bearer wrong", "Basic tok"} {
+ rec := do(mux, "GET", "/schedule-capabilities", auth, "")
+ if rec.Code != http.StatusUnauthorized || strings.Contains(rec.Body.String(), "registration_fence") {
+ t.Fatalf("unauthorized capability response: %d %s", rec.Code, rec.Body.String())
+ }
+ }
+ rec := do(mux, "GET", "/schedule-capabilities", "Bearer tok", "")
+ if rec.Code != http.StatusOK || rec.Header().Get("Content-Type") != "application/json" {
+ t.Fatalf("capability response: %d %s", rec.Code, rec.Body.String())
+ }
+ var got map[string]bool
+ if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 1 || !got["registration_fence"] {
+ t.Fatalf("capabilities: %v", got)
+ }
+ old := http.NewServeMux()
+ old.HandleFunc("GET /schedule", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
+ rec = do(old, "GET", "/schedule-capabilities", "Bearer tok", "")
+ if rec.Code != http.StatusNotFound || strings.Contains(rec.Body.String(), "registration_fence") {
+ t.Fatalf("old capability response: %d %s", rec.Code, rec.Body.String())
+ }
+}
+
+func TestControlNonpositiveOccurrenceArmsTimer(t *testing.T) {
+ for _, requested := range []int64{0, -1, -62135596800} {
+ for _, current := range []int64{1787490420, 0, -1} {
+ t.Run(strconv.FormatInt(requested, 10)+"/"+strconv.FormatInt(current, 10), func(t *testing.T) {
+ reg, _ := controlFixture(t)
+ mux := http.NewServeMux()
+ Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, func() time.Time { return time.Unix(current, 0) })
+ body := `{"at":"` + time.Unix(requested, 0).UTC().Format(time.RFC3339) + `","path":"/jobs/remind"}`
+ rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", body)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("PUT: %d %s", rec.Code, rec.Body.String())
+ }
+ row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != max(int64(1), current) || row.DueAt != requested || row.Spec != "at "+strconv.FormatInt(requested, 10) {
+ t.Fatalf("normalized timer: %+v", row)
+ }
+ first := row.Registration
+ rec = do(mux, "PUT", "/schedule/remind", "Bearer tok", body)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("repeat PUT: %d %s", rec.Code, rec.Body.String())
+ }
+ row, err = reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Registration == first {
+ t.Fatal("repeat PUT did not rearm")
+ }
+ })
+ }
+ }
+}
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go
index 9aa716af6..cfc6dfd46 100644
--- a/internal/schedule/runner.go
+++ b/internal/schedule/runner.go
@@ -308,7 +308,16 @@ func (r *Runner) slots() chan struct{} {
// sleeping. Nothing in the agent calls it.
func (r *Runner) wait() { r.inflight.Wait() }
+func (r *Runner) timerStillDue(t registry.Timer) bool {
+ current, err := r.Reg.GetTimer(t.Host, t.Name)
+ return err == nil && current.Registration == t.Registration && current.NextAt == t.NextAt && current.NextAt > 0 && current.NextAt <= r.now().Unix()
+}
+
func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) {
+ if !r.timerStillDue(t) {
+ return
+ }
+
sp, err := ParseSpec(t.Spec)
if err != nil {
log.Printf("schedule: %s/%s: bad spec %q: %v — dropping the row", rt.Host, t.Name, t.Spec, err)
@@ -328,6 +337,9 @@ func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer)
r.record(rt, t, sp, start, Outcome{Status: "bad path: " + err.Error()}, 0)
return
}
+ if !r.timerStillDue(t) {
+ return
+ }
dctx, cancel := context.WithTimeout(ctx, r.timeout())
req = req.WithContext(dctx)
status, derr := r.Deliver.Deliver(dctx, rt, req)
@@ -421,13 +433,15 @@ func classify(status int, err error, timeout time.Duration) Outcome {
// Classify is the exported policy for the host package's tests.
func Classify(status int, err error) Outcome { return classify(status, err, DeliveryTimeout) }
-// dueOf is the occurrence a row is delivering: DueAt, or NextAt for a row
-// written before due_at existed (whose NextAt, at attempt 0, IS the
-// occurrence — and once pinned by a retry it stays the occurrence).
func dueOf(t registry.Timer) int64 {
if t.DueAt != 0 {
return t.DueAt
}
+ if t.Kind == KindOnce {
+ if sp, err := ParseSpec(t.Spec); err == nil && sp.Kind == KindOnce && sp.At.Unix() == 0 {
+ return 0
+ }
+ }
return t.NextAt
}
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go
index 7ae3b4300..3602c996f 100644
--- a/internal/schedule/runner_test.go
+++ b/internal/schedule/runner_test.go
@@ -351,7 +351,8 @@ func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: "slow.bab.oncarlos.com", gate: make(chan struct{})}
reg, r := runnerFixture(t, d)
now := at("2026-08-23T13:07:00Z")
- r.Now = func() time.Time { return now }
+ clock := &testClock{t: now}
+ r.Now = clock.now
hosts := []string{"slow.bab.oncarlos.com", "b.bab.oncarlos.com", "c.bab.oncarlos.com", "d.bab.oncarlos.com", "e.bab.oncarlos.com", "f.bab.oncarlos.com", "g.bab.oncarlos.com"}
for i, h := range hosts {
if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
@@ -397,7 +398,7 @@ func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
}
// The parked host's row is still due; a second tick must skip it.
- now = now.Add(time.Minute)
+ clock.advance(time.Minute)
r.Tick(context.Background())
close(d.gate)
r.wait()
@@ -408,7 +409,7 @@ func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
t.Fatalf("slow row after release: %+v", tm)
}
// And once released, it is fair game again.
- now = at("2026-08-23T14:00:01Z")
+ clock.advance(at("2026-08-23T14:00:01Z").Sub(clock.now()))
r.tick()
if n := d.firedHosts()[d.slow]; n != 2 {
t.Fatalf("released host delivered %d times, want 2", n)
@@ -851,3 +852,127 @@ func TestInvalidTimerSnapshotCannotDeleteReplacement(t *testing.T) {
t.Fatalf("invalid snapshot changed replacement: got %+v, want %+v", after, replacement)
}
}
+
+func TestBatchSkipsTimerChangedByEarlierDelivery(t *testing.T) {
+ for _, mutation := range []string{"cancel", "rearm"} {
+ t.Run(mutation, func(t *testing.T) {
+ reg, runner := runnerFixture(t, nil)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ const host = "a.bab.oncarlos.com"
+ for _, name := range []string{"first", "later"} {
+ if err := reg.UpsertTimer(registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: "at 100", Path: "/jobs/" + name, NextAt: 100}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ calls := 0
+ runner.Deliver = timerMutationDeliverer{status: 204, mutate: func() {
+ calls++
+ if mutation == "cancel" {
+ if err := reg.DeleteTimer(host, "later"); err != nil {
+ t.Error(err)
+ }
+ } else {
+ if err := reg.FireTimerNow(host, "later", now.Add(time.Hour).Unix()); err != nil {
+ t.Error(err)
+ }
+ }
+ }}
+ runner.tick()
+ if calls != 1 {
+ t.Fatalf("delivered %d timers, want first only", calls)
+ }
+ })
+ }
+}
+
+func TestConcurrentTicksSkipSnapshotChangedDuringTokenLookup(t *testing.T) {
+ for _, mutation := range []string{"cancel", "rearm", "retry", "retry-still-due", "done"} {
+ t.Run(mutation, func(t *testing.T) {
+ deliver := &fakeDeliverer{status: 204}
+ reg, runner := runnerFixture(t, deliver)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 100", Path: "/jobs/remind", NextAt: 100}
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ before, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ entered, release := make(chan struct{}), make(chan struct{})
+ var tokenOnce sync.Once
+ runner.Token = func(string) (string, error) { tokenOnce.Do(func() { close(entered) }); <-release; return "tok", nil }
+ runner.Tick(context.Background())
+ <-entered
+ switch mutation {
+ case "cancel":
+ err = reg.DeleteTimer(tm.Host, tm.Name)
+ case "rearm":
+ err = reg.FireTimerNow(tm.Host, tm.Name, now.Add(time.Hour).Unix())
+ case "retry":
+ _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Add(time.Minute).Unix(), 100)
+ case "retry-still-due":
+ _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Unix()-1, 100)
+ case "done":
+ _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "204", 0, 0, 0, 0)
+ }
+ if err != nil {
+ close(release)
+ runner.wait()
+ t.Fatal(err)
+ }
+ runner.Tick(context.Background())
+ close(release)
+ runner.wait()
+ runner.fire(context.Background(), registry.Route{Host: tm.Host}, *before)
+ if len(deliver.calls) != 0 {
+ t.Fatalf("delivered stale snapshot %d times", len(deliver.calls))
+ }
+ })
+ }
+}
+
+func TestNonpositiveOccurrenceDeliveredAndRetriedWithOriginalTimestamp(t *testing.T) {
+ for _, requested := range []int64{0, -1, -62135596800} {
+ t.Run(fmt.Sprint(requested), func(t *testing.T) {
+ deliver := &fakeDeliverer{status: 503}
+ reg, runner := runnerFixture(t, deliver)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ mux := http.NewServeMux()
+ Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, runner.Now)
+ rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"`+time.Unix(requested, 0).UTC().Format(time.RFC3339)+`","path":"/jobs/remind"}`)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("PUT: %d %s", rec.Code, rec.Body.String())
+ }
+ runner.tick()
+ row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Attempts != 1 || row.NextAt != now.Add(time.Minute).Unix() || row.DueAt != requested {
+ t.Fatalf("retry: %+v", row)
+ }
+ now = now.Add(time.Minute)
+ deliver.status = 204
+ runner.tick()
+ if len(deliver.calls) != 2 {
+ t.Fatalf("deliveries: %d", len(deliver.calls))
+ }
+ for _, req := range deliver.calls {
+ if got := req.Header.Get("X-Carlos-Schedule-At"); got != fmt.Sprint(requested) {
+ t.Fatalf("occurrence: %q, want %d", got, requested)
+ }
+ }
+ row, err = reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != 0 || row.LastStatus != "204" {
+ t.Fatalf("completed: %+v", row)
+ }
+ })
+ }
+}