amadan

carlos / platform Public

moves: the timers snapshot and the canonical digest (§9.4)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b797a1d0208fbb035f7a42c18b4f522dbb9e8ec1 parent 336e14e
2 files changed, +262 −0
  • internal/moves/timers.go +99 −0
  • internal/moves/timers_test.go +163 −0
diff --git a/internal/moves/timers.go b/internal/moves/timers.go
new file mode 100644
index 000000000..eb21b050f
--- /dev/null
+++ b/internal/moves/timers.go
@@ -0,0 +1,99 @@
+package moves
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// timerWire is registry.Timer with the registration on the wire: the
+// snapshot carries it so ReplaceTimers can preserve it (§9.4).
+type timerWire struct {
+ Host, Name, Kind, Spec, Path string
+ NextAt, LastAt int64
+ LastStatus string
+ LastMs int64
+ Attempts int
+ CreatedAt, DueAt int64
+ Registration string
+}
+
+func toWire(t registry.Timer) timerWire {
+ return timerWire{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}
+}
+
+func fromWire(w timerWire) registry.Timer {
+ return registry.Timer{Host: w.Host, Name: w.Name, Kind: w.Kind, Spec: w.Spec, Path: w.Path, NextAt: w.NextAt, LastAt: w.LastAt, LastStatus: w.LastStatus, LastMs: w.LastMs, Attempts: w.Attempts, CreatedAt: w.CreatedAt, DueAt: w.DueAt, Registration: w.Registration}
+}
+
+// TimersSnapshot is the timers table for one host as carried (§6.2, §9.4).
+type TimersSnapshot struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Rows []registry.Timer `json:"-"`
+ TakenAt time.Time `json:"taken_at"`
+}
+
+type timersWire struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Rows []timerWire `json:"rows"`
+ TakenAt time.Time `json:"taken_at"`
+}
+
+func (s TimersSnapshot) Encode() ([]byte, error) {
+ w := timersWire{V: 1, Host: s.Host, TakenAt: s.TakenAt}
+ for _, r := range s.Rows {
+ w.Rows = append(w.Rows, toWire(r))
+ }
+ return json.Marshal(w)
+}
+
+func DecodeTimers(b []byte) (TimersSnapshot, error) {
+ var w timersWire
+ if err := json.Unmarshal(b, &w); err != nil {
+ return TimersSnapshot{}, fmt.Errorf("moves: timers snapshot: %w", err)
+ }
+ s := TimersSnapshot{V: w.V, Host: w.Host, TakenAt: w.TakenAt}
+ for _, r := range w.Rows {
+ s.Rows = append(s.Rows, fromWire(r))
+ }
+ return s, nil
+}
+
+// ActiveTimers drops disposable history: a once-timer that has fired and
+// has no next occurrence. Everything else is acknowledged state (INV-2).
+func ActiveTimers(rows []registry.Timer) []registry.Timer {
+ var out []registry.Timer
+ for _, t := range rows {
+ if t.Kind == "once" && t.NextAt == 0 && t.LastStatus != "" {
+ continue
+ }
+ out = append(out, t)
+ }
+ return out
+}
+
+// TimersDigest is the canonical digest a clean recovery compares (§9.4):
+// active rows, sorted by name then kind, every bookkeeping column.
+func TimersDigest(rows []registry.Timer) string {
+ active := ActiveTimers(rows)
+ sort.Slice(active, func(i, j int) bool {
+ if active[i].Name != active[j].Name {
+ return active[i].Name < active[j].Name
+ }
+ return active[i].Kind < active[j].Kind
+ })
+ h := sha256.New()
+ for _, t := range active {
+ b, _ := json.Marshal(toWire(t))
+ h.Write(b)
+ h.Write([]byte{'\n'})
+ }
+ return "sha256:" + hex.EncodeToString(h.Sum(nil))
+}
diff --git a/internal/moves/timers_test.go b/internal/moves/timers_test.go
new file mode 100644
index 000000000..2a83cf59d
--- /dev/null
+++ b/internal/moves/timers_test.go
@@ -0,0 +1,163 @@
+package moves
+
+import (
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// §9.4: the digest covers active rows and their bookkeeping and excludes
+// disposable history, so the runner's daily prune of completed
+// once-timers never changes what a clean recovery compares (joint N14).
+func TestTimersDigestIgnoresCompletedOnceTimers(t *testing.T) {
+ sched := registry.Timer{Host: "h", Name: "nightly", Kind: "schedule", Spec: "0 2 * * *", Path: "/n", NextAt: 100, DueAt: 100, Attempts: 1, Registration: "r1"}
+ pending := registry.Timer{Host: "h", Name: "send-later", Kind: "once", Path: "/s", NextAt: 200, DueAt: 200, Registration: "r2"}
+ done := registry.Timer{Host: "h", Name: "sent", Kind: "once", Path: "/s", NextAt: 0, LastAt: 50, LastStatus: "200", Registration: "r3"}
+ d1 := TimersDigest([]registry.Timer{sched, pending, done})
+ d2 := TimersDigest([]registry.Timer{pending, sched})
+ if d1 != d2 {
+ t.Fatal("history or order changed the digest")
+ }
+ moved := pending
+ moved.Attempts = 1
+ if TimersDigest([]registry.Timer{sched, moved}) == d1 {
+ t.Fatal("bookkeeping on an active row is part of the digest")
+ }
+ if n := len(ActiveTimers([]registry.Timer{sched, pending, done})); n != 2 {
+ t.Fatalf("active = %d", n)
+ }
+ snap := TimersSnapshot{V: 1, Host: "h", Rows: []registry.Timer{sched, pending, done}}
+ b, err := snap.Encode()
+ if err != nil {
+ t.Fatal(err)
+ }
+ back, err := DecodeTimers(b)
+ if err != nil || len(back.Rows) != 3 || back.Rows[0].Registration != "r1" {
+ t.Fatalf("round trip lost the registration: %+v %v", back, err)
+ }
+}
+
+// Which rows are active: every schedule row (even one with no next
+// occurrence and a last status), a once-timer that has never fired, and a
+// once-timer that fired but is retrying or rescheduled. Only a fired
+// once-timer with nothing next is history.
+func TestActiveTimersSelection(t *testing.T) {
+ rows := []registry.Timer{
+ {Name: "sched-idle", Kind: "schedule", NextAt: 0, LastStatus: "200"},
+ {Name: "sched-next", Kind: "schedule", NextAt: 10, LastStatus: "200"},
+ {Name: "once-new", Kind: "once", NextAt: 0, LastStatus: ""},
+ {Name: "once-retry", Kind: "once", NextAt: 20, LastStatus: "503"},
+ {Name: "once-done", Kind: "once", NextAt: 0, LastStatus: "200"},
+ {Name: "once-failed", Kind: "once", NextAt: 0, LastStatus: "500"},
+ }
+ var got []string
+ for _, r := range ActiveTimers(rows) {
+ got = append(got, r.Name)
+ }
+ want := "sched-idle,sched-next,once-new,once-retry"
+ if strings.Join(got, ",") != want {
+ t.Fatalf("active = %v, want %s", got, want)
+ }
+ if n := len(ActiveTimers(nil)); n != 0 {
+ t.Fatalf("active of nothing = %d", n)
+ }
+}
+
+// The digest must not reorder or drop rows in the caller's slice.
+func TestTimersDigestLeavesInputAlone(t *testing.T) {
+ a := registry.Timer{Host: "h", Name: "b", Kind: "once", NextAt: 1}
+ b := registry.Timer{Host: "h", Name: "a", Kind: "once", NextAt: 2}
+ done := registry.Timer{Host: "h", Name: "c", Kind: "once", LastStatus: "200"}
+ in := []registry.Timer{a, b, done}
+ TimersDigest(in)
+ if in[0].Name != "b" || in[1].Name != "a" || in[2].Name != "c" || len(in) != 3 {
+ t.Fatalf("input changed: %+v", in)
+ }
+}
+
+// Every column of an active row moves the digest, including the
+// registration, which registry.Timer hides from JSON.
+func TestTimersDigestCoversEveryColumn(t *testing.T) {
+ base := registry.Timer{Host: "h", Name: "n", Kind: "schedule", Spec: "* * * * *", Path: "/p", NextAt: 5, LastAt: 4, LastStatus: "200", LastMs: 7, Attempts: 1, CreatedAt: 3, DueAt: 5, Registration: "r"}
+ want := TimersDigest([]registry.Timer{base})
+ muts := map[string]func(*registry.Timer){
+ "Host": func(x *registry.Timer) { x.Host = "h2" },
+ "Spec": func(x *registry.Timer) { x.Spec = "0 * * * *" },
+ "Path": func(x *registry.Timer) { x.Path = "/q" },
+ "NextAt": func(x *registry.Timer) { x.NextAt++ },
+ "LastAt": func(x *registry.Timer) { x.LastAt++ },
+ "LastStatus": func(x *registry.Timer) { x.LastStatus = "500" },
+ "LastMs": func(x *registry.Timer) { x.LastMs++ },
+ "Attempts": func(x *registry.Timer) { x.Attempts++ },
+ "CreatedAt": func(x *registry.Timer) { x.CreatedAt++ },
+ "DueAt": func(x *registry.Timer) { x.DueAt++ },
+ "Registration": func(x *registry.Timer) { x.Registration = "r2" },
+ }
+ for name, mut := range muts {
+ x := base
+ mut(&x)
+ if TimersDigest([]registry.Timer{x}) == want {
+ t.Errorf("%s is not part of the digest", name)
+ }
+ }
+ if !strings.HasPrefix(want, "sha256:") || len(want) != len("sha256:")+64 {
+ t.Fatalf("digest form = %q", want)
+ }
+}
+
+// Rows with the same name and different kinds sort by kind, so the digest
+// does not depend on input order for them either.
+func TestTimersDigestOrdersByNameThenKind(t *testing.T) {
+ a := registry.Timer{Host: "h", Name: "x", Kind: "once", NextAt: 1}
+ b := registry.Timer{Host: "h", Name: "x", Kind: "schedule", Spec: "* * * * *", NextAt: 2}
+ if TimersDigest([]registry.Timer{a, b}) != TimersDigest([]registry.Timer{b, a}) {
+ t.Fatal("order of same-named rows changed the digest")
+ }
+}
+
+// The snapshot carries every column through the wire, the registration
+// included, and the version, host and time.
+func TestTimersSnapshotRoundTrip(t *testing.T) {
+ at := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)
+ row := registry.Timer{Host: "h", Name: "n", Kind: "schedule", Spec: "0 2 * * *", Path: "/n", NextAt: 1, LastAt: 2, LastStatus: "200", LastMs: 3, Attempts: 4, CreatedAt: 5, DueAt: 6, Registration: "reg"}
+ b, err := TimersSnapshot{V: 1, Host: "h", Rows: []registry.Timer{row}, TakenAt: at}.Encode()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !strings.Contains(string(b), `"reg"`) {
+ t.Fatalf("registration not on the wire: %s", b)
+ }
+ back, err := DecodeTimers(b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if back.V != 1 || back.Host != "h" || !back.TakenAt.Equal(at) || len(back.Rows) != 1 || back.Rows[0] != row {
+ t.Fatalf("round trip = %+v", back)
+ }
+ if TimersDigest(back.Rows) != TimersDigest([]registry.Timer{row}) {
+ t.Fatal("digest changed across the wire")
+ }
+}
+
+func TestDecodeTimersRejectsGarbage(t *testing.T) {
+ if _, err := DecodeTimers([]byte("not json")); err == nil {
+ t.Fatal("garbage decoded")
+ }
+}
+
+// An empty snapshot round-trips and its digest is stable.
+func TestTimersSnapshotEmpty(t *testing.T) {
+ b, err := TimersSnapshot{V: 1, Host: "h"}.Encode()
+ if err != nil {
+ t.Fatal(err)
+ }
+ back, err := DecodeTimers(b)
+ if err != nil || len(back.Rows) != 0 {
+ t.Fatalf("empty = %+v %v", back, err)
+ }
+ if TimersDigest(nil) != TimersDigest(back.Rows) {
+ t.Fatal("empty digests differ")
+ }
+}