amadan

carlos / platform Public

Preserve timer replacements across in-flight delivery completion

Co-Authored-By: Codex <noreply@openai.com>
Paul Campbell pushed by paul@keymail.dev 0310ae769acf789f24bc5a8d01a5736640df73c8 parent 5abc9ec
7 files changed, +520 −30
  • internal/host/schedule_rearm_wake_test.go +119 −0
  • internal/registry/registry.go +3 −0
  • internal/registry/timers.go +34 −21
  • internal/registry/timers_test.go +169 −3
  • internal/schedule/converge_test.go +12 −3
  • internal/schedule/runner.go +4 −3
  • internal/schedule/runner_test.go +179 −0

CI passed — run details

diff --git a/internal/host/schedule_rearm_wake_test.go b/internal/host/schedule_rearm_wake_test.go
new file mode 100644
index 000000000..560861d41
--- /dev/null
+++ b/internal/host/schedule_rearm_wake_test.go
@@ -0,0 +1,119 @@
+package host
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+func TestSleepingInstanceRearmsItsNextDeliveryBeforeReturning(t *testing.T) {
+ const host = "rearm.bab.oncarlos.com"
+ const idle = 200 * time.Millisecond
+ entered := make(chan struct{}, 1)
+ release := make(chan struct{})
+ events := make(chan string, 4)
+ var calls atomic.Int32
+ var reg *registry.Registry
+ now := time.Now().UTC()
+ var clock atomic.Int64
+ clock.Store(now.UnixNano())
+ next := now.Add(time.Hour).Unix()
+ var firstError atomic.Pointer[error]
+ var instReg *registry.Registry
+ instReg, inst, svc, dd := wakeFixture(t, host, idle, func(w http.ResponseWriter, req *http.Request) {
+ if req.Method != http.MethodPost || req.URL.Path != "/jobs/reminders" || !strings.HasPrefix(req.Header.Get("Authorization"), "Bearer ") {
+ w.WriteHeader(http.StatusForbidden)
+ return
+ }
+ if calls.Add(1) == 1 {
+ if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "next-due", Kind: schedule.KindOnce,
+ Spec: "at " + strconv.FormatInt(next, 10), Path: "/jobs/reminders", NextAt: next, DueAt: next}); err != nil {
+ firstError.Store(&err)
+ }
+ entered <- struct{}{}
+ <-release
+ }
+ w.WriteHeader(http.StatusNoContent)
+ })
+ reg = instReg
+ t.Cleanup(func() {
+ select {
+ case <-release:
+ default:
+ close(release)
+ }
+ })
+ runner := &schedule.Runner{Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg),
+ Now: func() time.Time { return time.Unix(0, clock.Load()) }, Timeout: 10 * time.Second,
+ Event: func(_ registry.Route, _, msg string) { events <- msg },
+ }
+ due := now.Unix() - 1
+ if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "next-due", Kind: schedule.KindOnce,
+ Spec: fmt.Sprintf("at %d", due), Path: "/jobs/reminders", NextAt: due, DueAt: due}); err != nil {
+ t.Fatal(err)
+ }
+ runner.Tick(context.Background())
+ select {
+ case <-entered:
+ case <-time.After(10 * time.Second):
+ t.Fatal("sleeping instance never received its due delivery")
+ }
+ if err := firstError.Load(); err != nil {
+ t.Fatal(*err)
+ }
+ time.Sleep(3 * idle)
+ if !inst.running() || !svc.Owns(host) {
+ t.Fatal("instance slept while its scheduled handler was still running")
+ }
+ close(release)
+ select {
+ case <-events:
+ case <-time.After(10 * time.Second):
+ t.Fatal("first delivery did not finish")
+ }
+ timer, err := reg.GetTimer(host, "next-due")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if timer.NextAt != next {
+ t.Fatalf("completion erased the replacement delivery: next_at=%d, want %d", timer.NextAt, next)
+ }
+ deadline := time.Now().Add(10 * time.Second)
+ for inst.running() || svc.Owns(host) {
+ if time.Now().After(deadline) {
+ t.Fatal("instance did not hibernate between deliveries")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ clock.Store(time.Unix(next, 0).UnixNano())
+ for calls.Load() < 2 {
+ runner.Tick(context.Background())
+ if time.Now().After(deadline) {
+ t.Fatal("replacement delivery never woke the instance")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ select {
+ case <-events:
+ case <-time.After(10 * time.Second):
+ t.Fatal("second delivery did not finish")
+ }
+ timer, err = reg.GetTimer(host, "next-due")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if timer.NextAt != 0 || timer.LastStatus != "204" {
+ t.Fatalf("second delivery not completed: %+v", timer)
+ }
+ if inst.startCount() != 2 {
+ t.Fatalf("got %d wakes, want 2", inst.startCount())
+ }
+}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 7dfc050cd..2d1f935aa 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -550,6 +550,9 @@ CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`);
if _, err := r.sql.Exec(`ALTER TABLE timers ADD COLUMN due_at INTEGER NOT NULL DEFAULT 0`); err != nil && !strings.Contains(err.Error(), "duplicate column") {
return err
}
+ if _, err := r.sql.Exec(`ALTER TABLE timers ADD COLUMN registration TEXT NOT NULL DEFAULT ''`); err != nil && !strings.Contains(err.Error(), "duplicate column") {
+ return err
+ }
if err := r.migrateAccounts(); err != nil {
return err
}
diff --git a/internal/registry/timers.go b/internal/registry/timers.go
index 12cdde238..4ea3281c5 100644
--- a/internal/registry/timers.go
+++ b/internal/registry/timers.go
@@ -28,16 +28,17 @@ type Timer struct {
// when the row advances to a new occurrence (MarkTimerFired with a new
// due, FireTimerNow, or an upsert that moves next_at). Zero on rows
// written before the column existed; readers fall back to NextAt.
- DueAt int64
+ DueAt int64
+ Registration string `json:"-"`
}
var ErrTimerKind = errors.New("registry: timer exists with a different kind")
-const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at`
+const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at, registration`
func scanTimer(row interface{ Scan(...any) error }) (Timer, error) {
var t Timer
- err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt, &t.DueAt)
+ err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt, &t.DueAt, &t.Registration)
return t, err
}
@@ -45,24 +46,27 @@ func scanTimer(row interface{ Scan(...any) error }) (Timer, error) {
// 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.
func (r *Registry) UpsertTimer(t Timer) error {
- existing, err := r.GetTimer(t.Host, t.Name)
- switch {
- case errors.Is(err, os.ErrNotExist):
- _, err := r.sql.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`,
- t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt)
+ 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))))
+ON CONFLICT(host, name) DO UPDATE SET
+ spec=excluded.spec, path=excluded.path,
+ next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END,
+ due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END,
+ attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END,
+ registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END
+WHERE timers.kind=excluded.kind`,
+ t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt)
+ if err != nil {
return err
- case err != nil:
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
return err
}
- if existing.Kind != t.Kind {
+ if n == 0 {
return ErrTimerKind
}
- next, due := existing.NextAt, existing.DueAt
- if existing.Spec != t.Spec || existing.Path != t.Path || t.NextAt != 0 {
- next, due = t.NextAt, t.DueAt
- }
- _, err = r.sql.Exec(`UPDATE timers SET spec=?, path=?, next_at=?, due_at=? WHERE host=? AND name=?`, t.Spec, t.Path, next, due, t.Host, t.Name)
- return err
+ return nil
}
func (r *Registry) GetTimer(host, name string) (*Timer, error) {
@@ -81,6 +85,11 @@ func (r *Registry) DeleteTimer(host, name string) error {
return err
}
+func (r *Registry) DeleteTimerRegistration(host, name, registration string) error {
+ _, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=? AND registration=?`, host, name, registration)
+ return err
+}
+
func (r *Registry) queryTimers(q string, args ...any) ([]Timer, error) {
rows, err := r.sql.Query(q, args...)
if err != nil {
@@ -130,7 +139,7 @@ func (r *Registry) DueTimers(now int64, limit int, excludeHosts []string) ([]Tim
// delivered (a `carlos schedule run` is its own occurrence, not an early
// delivery of the next natural one).
func (r *Registry) FireTimerNow(host, name string, now int64) error {
- _, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=? WHERE host=? AND name=?`, now, now, host, name)
+ _, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=?, attempts=0, registration=lower(hex(randomblob(16))) WHERE host=? AND name=?`, now, now, host, name)
return err
}
@@ -138,10 +147,14 @@ func (r *Registry) FireTimerNow(host, name string, now int64) error {
// due is the occurrence THAT next delivery is for: the same value as next
// when the row advances, and the row's current DueAt when next is a retry
// of the occurrence just attempted — the runner decides which.
-func (r *Registry) MarkTimerFired(host, name string, at int64, status string, ms int64, attempts int, next, due int64) error {
- _, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=?`,
- at, status, ms, attempts, next, due, host, name)
- return err
+func (r *Registry) MarkTimerFired(host, name, registration string, at int64, status string, ms int64, attempts int, next, due int64) (bool, error) {
+ res, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=? AND registration=?`,
+ at, status, ms, attempts, next, due, host, name, registration)
+ if err != nil {
+ return false, err
+ }
+ n, err := res.RowsAffected()
+ return n != 0, err
}
// SetTimerStatus writes last_status alone — converge's one word on a row
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go
index 57b2eb85c..004cada04 100644
--- a/internal/registry/timers_test.go
+++ b/internal/registry/timers_test.go
@@ -24,7 +24,7 @@ func TestTimersUpsertPreservesRunnerFields(t *testing.T) {
if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 100, CreatedAt: 1}); err != nil {
t.Fatal(err)
}
- if err := r.MarkTimerFired("h", "sync", 100, "204", 300, 0, 21700, 21700); err != nil {
+ if _, err := r.MarkTimerFired("h", "sync", timerRegistration(t, r, "h", "sync"), 100, "204", 300, 0, 21700, 21700); err != nil {
t.Fatal(err)
}
// Converge sees the same declaration again: spec/path unchanged, next_at
@@ -135,7 +135,7 @@ func TestTimersDueAtFollowsTheOccurrence(t *testing.T) {
t.Fatal(err)
}
// A retry: next moves to the retry instant, due stays the occurrence.
- if err := r.MarkTimerFired("h", "s", 3600, "503", 1, 1, 3660, 3600); err != nil {
+ if _, err := r.MarkTimerFired("h", "s", timerRegistration(t, r, "h", "s"), 3600, "503", 1, 1, 3660, 3600); err != nil {
t.Fatal(err)
}
got, _ := r.GetTimer("h", "s")
@@ -143,7 +143,7 @@ func TestTimersDueAtFollowsTheOccurrence(t *testing.T) {
t.Fatalf("after retry: %+v", got)
}
// An advance: both move together.
- if err := r.MarkTimerFired("h", "s", 3660, "204", 1, 0, 7200, 7200); err != nil {
+ if _, err := r.MarkTimerFired("h", "s", timerRegistration(t, r, "h", "s"), 3660, "204", 1, 0, 7200, 7200); err != nil {
t.Fatal(err)
}
if got, _ = r.GetTimer("h", "s"); got.NextAt != 7200 || got.DueAt != 7200 {
@@ -288,3 +288,169 @@ func TestEarliestTimersEmpty(t *testing.T) {
t.Errorf("EarliestTimers on an empty table = %v, want empty", got)
}
}
+
+func timerRegistration(t *testing.T, r *Registry, host, name string) string {
+ t.Helper()
+ row, err := r.GetTimer(host, name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return row.Registration
+}
+
+func TestTimerRegistrationProtectsMutations(t *testing.T) {
+ for _, mutation := range []string{"identical", "spec", "path", "manual", "recreate"} {
+ t.Run(mutation, func(t *testing.T) {
+ r := openTimersReg(t)
+ tm := Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/jobs/sync", NextAt: 100, DueAt: 100, Attempts: 4, LastAt: 80, LastStatus: "503", LastMs: 1}
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ before, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ switch mutation {
+ case "spec":
+ tm.Spec, tm.NextAt, tm.DueAt = "every 2h0m0s", 0, 0
+ case "path":
+ tm.Path, tm.NextAt, tm.DueAt = "/jobs/other", 0, 0
+ case "recreate":
+ if err := r.DeleteTimer(tm.Host, tm.Name); err != nil {
+ t.Fatal(err)
+ }
+ tm.Attempts = 0
+ }
+ if mutation == "manual" {
+ err = r.FireTimerNow(tm.Host, tm.Name, 100)
+ } else {
+ err = r.UpsertTimer(tm)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ replacement, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if replacement.Registration == "" || replacement.Registration == before.Registration || replacement.Attempts != 0 {
+ t.Fatalf("new registration: %+v", replacement)
+ }
+ if _, err := r.MarkTimerFired(tm.Host, tm.Name, before.Registration, 100, "204", 3, 0, 3700, 3700); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.DeleteTimerRegistration(tm.Host, tm.Name, before.Registration); err != nil {
+ t.Fatal(err)
+ }
+ after, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if *after != *replacement {
+ t.Fatalf("stale mutation changed registration: %+v -> %+v", replacement, after)
+ }
+ if err := r.DeleteTimerRegistration(tm.Host, tm.Name, after.Registration); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.GetTimer(tm.Host, tm.Name); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("current deletion: %v", err)
+ }
+ })
+ }
+}
+
+func TestTimerRegistrationLegacyMigration(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "registry.db")
+ r, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`ALTER TABLE timers DROP COLUMN registration`); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`INSERT INTO timers (host,name,kind,spec,path,next_at,created_at) VALUES ('h','s','once','at 100','/jobs/sync',100,1)`); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < 2; i++ {
+ r, err = Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ row, err := r.GetTimer("h", "s")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Registration != "" {
+ t.Fatalf("legacy registration: %q", row.Registration)
+ }
+ if _, err := r.MarkTimerFired("h", "s", row.Registration, 100, "503", 3, 1, 160, 100); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ }
+ r, err = Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ row, err := r.GetTimer("h", "s")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != 160 || row.DueAt != 100 || row.Attempts != 1 {
+ t.Fatalf("legacy completion: %+v", row)
+ }
+ if err := r.FireTimerNow("h", "s", 100); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.MarkTimerFired("h", "s", "", 100, "204", 3, 0, 0, 0); err != nil {
+ t.Fatal(err)
+ }
+ row, err = r.GetTimer("h", "s")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Registration == "" || row.NextAt != 100 || row.Attempts != 0 {
+ t.Fatalf("rearmed legacy: %+v", row)
+ }
+}
+
+func TestUnchangedTimerUpsertConcurrentWithCompletion(t *testing.T) {
+ r := openTimersReg(t)
+ tm := Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/jobs/sync", NextAt: 100, DueAt: 100}
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ registration := timerRegistration(t, r, tm.Host, tm.Name)
+ tm.NextAt, tm.DueAt = 0, 0
+ for i := int64(1); i <= 100; i++ {
+ start := make(chan struct{})
+ errs := make(chan error, 9)
+ for n := 0; n < 8; n++ {
+ go func() { <-start; errs <- r.UpsertTimer(tm) }()
+ }
+ go func() {
+ <-start
+ _, err := r.MarkTimerFired(tm.Host, tm.Name, registration, i, "503", 1, 1, 100+i, 100)
+ errs <- err
+ }()
+ close(start)
+ for n := 0; n < 9; n++ {
+ if err := <-errs; err != nil {
+ t.Fatal(err)
+ }
+ }
+ row, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != 100+i || row.DueAt != 100 || row.LastAt != i || row.Attempts != 1 || row.Registration != registration {
+ t.Fatalf("completion overwritten at iteration %d: %+v", i, row)
+ }
+ }
+}
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go
index c8375bf39..95603d9e4 100644
--- a/internal/schedule/converge_test.go
+++ b/internal/schedule/converge_test.go
@@ -131,7 +131,7 @@ func TestConvergeMaterialisesDeclarationPerExecInstance(t *testing.T) {
// A second pass with nothing changed: no new reports, next_at untouched.
reports = nil
- reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 10, 0, 999, 999)
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 10, 0, 999, 999)
if err := c.Pass(ctx); err != nil {
t.Fatal(err)
}
@@ -190,7 +190,7 @@ func TestConvergeRunSignalFiresOnceAcrossInstances(t *testing.T) {
t.Fatalf("%s not fired: %+v", h, tm)
}
}
- reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777)
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
c2.Pass(ctx) // same etag: must not re-fire
if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
t.Fatal("run signal replayed on an unchanged etag")
@@ -511,7 +511,7 @@ func TestConvergeRunSignalGoingAbsentIsNotAFire(t *testing.T) {
if err := c.Pass(ctx); err != nil {
t.Fatal(err)
}
- reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777)
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
t.Fatalf("setup: run signal did not fire: %+v", tm)
}
@@ -597,3 +597,12 @@ func TestConvergeParksAScheduleThatNeverFires(t *testing.T) {
t.Fatalf("a parked row is due: %+v", due)
}
}
+
+func scheduleTimerRegistration(t *testing.T, reg *registry.Registry, host, name string) string {
+ t.Helper()
+ row, err := reg.GetTimer(host, name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return row.Registration
+}
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go
index 888a5a784..9aa716af6 100644
--- a/internal/schedule/runner.go
+++ b/internal/schedule/runner.go
@@ -312,7 +312,7 @@ func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer)
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)
- r.Reg.DeleteTimer(rt.Host, t.Name)
+ r.Reg.DeleteTimerRegistration(rt.Host, t.Name, t.Registration)
return
}
start := r.now()
@@ -379,12 +379,13 @@ func (r *Runner) record(rt registry.Route, t registry.Timer, sp Spec, at time.Ti
} else if !isSuccess(o.Status) {
level = "warn"
}
- if err := r.Reg.MarkTimerFired(rt.Host, t.Name, at.Unix(), status, took.Milliseconds(), attempts, next, due); err != nil {
+ applied, err := r.Reg.MarkTimerFired(rt.Host, t.Name, t.Registration, at.Unix(), status, took.Milliseconds(), attempts, next, due)
+ if err != nil {
log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err)
}
if r.Event != nil {
msg := fmt.Sprintf("schedule %s fired POST %s → %s in %s", t.Name, t.Path, status, took.Truncate(time.Millisecond))
- if attempts > 0 {
+ if attempts > 0 && applied {
msg += fmt.Sprintf(" (attempt %d of %d; retrying at %s)", attempts, MaxAttempts, time.Unix(next, 0).UTC().Format(time.RFC3339))
}
r.Event(rt, level, msg)
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go
index a6880a9f9..7ae3b4300 100644
--- a/internal/schedule/runner_test.go
+++ b/internal/schedule/runner_test.go
@@ -5,6 +5,8 @@ import (
"errors"
"fmt"
"net/http"
+ "net/http/httptest"
+ "os"
"path/filepath"
"strings"
"sync"
@@ -672,3 +674,180 @@ func TestDeliveriesAreParallelToTheBoundAndAllComplete(t *testing.T) {
t.Fatalf("delivered to %d hosts, want %d", len(fired), hosts)
}
}
+
+type timerMutationDeliverer struct {
+ mutate func()
+ status int
+}
+
+func (d timerMutationDeliverer) Deliver(context.Context, registry.Route, *http.Request) (int, error) {
+ d.mutate()
+ return d.status, nil
+}
+
+func TestDeliveryCompletionPreservesReplacementRegistration(t *testing.T) {
+ for _, mutation := range []string{"identical", "timing", "path", "cancel", "recreate", "manual"} {
+ for _, status := range []int{204, 503} {
+ t.Run(fmt.Sprintf("%s/%d", mutation, status), 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"
+ mux := http.NewServeMux()
+ Mount(mux, host, reg, func() (string, error) { return "tok", nil }, runner.Now)
+ put := func(when time.Time, path string) {
+ req := httptest.NewRequest(http.MethodPut, "/schedule/remind", strings.NewReader(fmt.Sprintf(`{"at":%q,"path":%q}`, when.Format(time.RFC3339), path)))
+ req.Header.Set("Authorization", "Bearer tok")
+ w := httptest.NewRecorder()
+ mux.ServeHTTP(w, req)
+ if w.Code != http.StatusOK {
+ t.Errorf("PUT: %d %s", w.Code, w.Body.String())
+ }
+ }
+ put(now, "/jobs/remind")
+ before, err := reg.GetTimer(host, "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ var replacement *registry.Timer
+ var events []string
+ runner.Event = func(_ registry.Route, _, msg string) { events = append(events, msg) }
+ runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
+ switch mutation {
+ case "identical":
+ put(now, "/jobs/remind")
+ case "timing":
+ put(now.Add(time.Hour), "/jobs/remind")
+ case "path":
+ put(now, "/jobs/changed")
+ case "cancel", "recreate":
+ req := httptest.NewRequest(http.MethodDelete, "/schedule/remind", nil)
+ req.Header.Set("Authorization", "Bearer tok")
+ w := httptest.NewRecorder()
+ mux.ServeHTTP(w, req)
+ if w.Code != http.StatusNoContent {
+ t.Errorf("DELETE: %d", w.Code)
+ }
+ if mutation == "recreate" {
+ put(now, "/jobs/remind")
+ }
+ case "manual":
+ if err := reg.FireTimerNow(host, "remind", now.Unix()); err != nil {
+ t.Error(err)
+ }
+ }
+ replacement, err = reg.GetTimer(host, "remind")
+ if mutation != "cancel" && err != nil {
+ t.Error(err)
+ }
+ }}
+ runner.tick()
+ if len(events) != 1 || strings.Contains(events[0], "retrying at") {
+ t.Fatalf("superseded delivery event: %v", events)
+ }
+ after, err := reg.GetTimer(host, "remind")
+ if mutation == "cancel" {
+ if !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("cancelled timer: %+v, %v", after, err)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if replacement == nil {
+ t.Fatal("replacement missing")
+ }
+ if after.Registration == before.Registration || after.Registration == "" {
+ t.Fatalf("registration did not change: %+v", after)
+ }
+ if *after != *replacement {
+ t.Fatalf("completion changed replacement: got %+v, want %+v", after, replacement)
+ }
+ runner.Deliver = &fakeDeliverer{status: 204}
+ now = time.Unix(after.NextAt, 0)
+ runner.tick()
+ finished, err := reg.GetTimer(host, "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if finished.NextAt != 0 || finished.LastStatus != "204" {
+ t.Fatalf("replacement did not fire: %+v", finished)
+ }
+ })
+ }
+ }
+}
+
+func TestDeliveryCompletionSurvivesUnchangedDeclaration(t *testing.T) {
+ for _, status := range []int{204, 503} {
+ t.Run(fmt.Sprint(status), func(t *testing.T) {
+ reg, runner := runnerFixture(t, nil)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix(), DueAt: now.Unix()}
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ before, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tm.NextAt, tm.DueAt = 0, 0
+ runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Error(err)
+ }
+ }}
+ runner.tick()
+ after, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if after.Registration != before.Registration || after.LastStatus != fmt.Sprint(status) {
+ t.Fatalf("completion missing: %+v", after)
+ }
+ if status == 503 && (after.Attempts != 1 || after.DueAt != before.DueAt || after.NextAt != now.Add(time.Minute).Unix()) {
+ t.Fatalf("retry changed: %+v", after)
+ }
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ converged, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if *converged != *after {
+ t.Fatalf("converge changed completed state: got %+v, want %+v", converged, after)
+ }
+ })
+ }
+}
+
+func TestInvalidTimerSnapshotCannotDeleteReplacement(t *testing.T) {
+ reg, runner := runnerFixture(t, &fakeDeliverer{status: 204})
+ tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "invalid", Path: "/jobs/remind", NextAt: 100}
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ old, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tm.Spec = "at 100"
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ replacement, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ runner.fire(context.Background(), registry.Route{Host: tm.Host}, *old)
+ after, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if *after != *replacement {
+ t.Fatalf("invalid snapshot changed replacement: got %+v, want %+v", after, replacement)
+ }
+}