amadan

carlos / platform Public

hostfiles: a host's timers, next wake and replication class travel with its files

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 94b80df8fa6d5e6883476fe92ed67a823e5a8939 parent 2501d73
4 files changed, +413 −0
  • internal/hostfiles/wake.go +129 −0
  • internal/hostfiles/wake_test.go +176 −0
  • internal/registry/timers.go +31 −0
  • internal/registry/timers_test.go +77 −0
diff --git a/internal/hostfiles/wake.go b/internal/hostfiles/wake.go
new file mode 100644
index 000000000..5421a3e22
--- /dev/null
+++ b/internal/hostfiles/wake.go
@@ -0,0 +1,129 @@
+package hostfiles
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// WakeItem is the wake state's name among the manifest's platform items.
+const WakeItem = "wake.json"
+
+// WakeTimer is registry.Timer on the wire. It exists because
+// registry.Timer hides Registration from JSON, and an import must keep it:
+// the runner's delivery bookkeeping is keyed by it.
+type WakeTimer struct {
+ Name string `json:"name"`
+ Kind string `json:"kind"`
+ Spec string `json:"spec"`
+ Path string `json:"path"`
+ NextAt int64 `json:"next_at"`
+ LastAt int64 `json:"last_at"`
+ LastStatus string `json:"last_status"`
+ LastMs int64 `json:"last_ms"`
+ Attempts int `json:"attempts"`
+ CreatedAt int64 `json:"created_at"`
+ DueAt int64 `json:"due_at"`
+ Registration string `json:"registration"`
+}
+
+// WakeState is everything that decides when a host next runs without a
+// request: its timers, the app's own next wake, and its replication class
+// (region HA spec 1 §5.7).
+type WakeState struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Timers []WakeTimer `json:"timers"`
+ NextDue int64 `json:"next_due"`
+ Replication string `json:"replication"`
+}
+
+// ExportWake reads host's wake state. A finished once-timer (next_at 0) is
+// history: it is left out, so the runner's daily prune of such rows never
+// changes what is exported.
+func ExportWake(reg *registry.Registry, host string) (WakeState, error) {
+ rt, err := reg.Get(host)
+ if err != nil {
+ return WakeState{}, err
+ }
+ if rt == nil {
+ return WakeState{}, fmt.Errorf("hostfiles: wake state for %s: no route", host)
+ }
+ rows, err := reg.TimersFor(host)
+ if err != nil {
+ return WakeState{}, err
+ }
+ ws := WakeState{V: 1, Host: host, Timers: []WakeTimer{}, NextDue: rt.NextDue, Replication: rt.Replication}
+ for _, t := range rows {
+ if t.Kind == "once" && t.NextAt == 0 {
+ continue
+ }
+ ws.Timers = append(ws.Timers, WakeTimer{Name: t.Name, Kind: t.Kind, Spec: t.Spec, Path: t.Path,
+ NextAt: t.NextAt, LastAt: t.LastAt, LastStatus: t.LastStatus, LastMs: t.LastMs, Attempts: t.Attempts,
+ CreatedAt: t.CreatedAt, DueAt: t.DueAt, Registration: t.Registration})
+ }
+ return ws, nil
+}
+
+// ImportWake makes host's wake state exactly ws. It validates everything
+// before it writes anything.
+func ImportWake(reg *registry.Registry, host string, ws WakeState) error {
+ if ws.V != 1 {
+ return fmt.Errorf("hostfiles: wake state version %d; this binary reads 1", ws.V)
+ }
+ if ws.Host != host {
+ return fmt.Errorf("hostfiles: wake state is for %q, not %q", ws.Host, host)
+ }
+ if !registry.ValidReplication(ws.Replication) {
+ return fmt.Errorf("hostfiles: wake state for %s names replication class %q", host, ws.Replication)
+ }
+ rows := make([]registry.Timer, 0, len(ws.Timers))
+ for _, t := range ws.Timers {
+ rows = append(rows, registry.Timer{Host: host, Name: t.Name, Kind: t.Kind, Spec: t.Spec, Path: t.Path,
+ NextAt: t.NextAt, LastAt: t.LastAt, LastStatus: t.LastStatus, LastMs: t.LastMs, Attempts: t.Attempts,
+ CreatedAt: t.CreatedAt, DueAt: t.DueAt, Registration: t.Registration})
+ }
+ if err := reg.ReplaceTimers(host, rows); err != nil {
+ return err
+ }
+ if err := reg.SetNextDue(host, ws.NextDue); err != nil {
+ return err
+ }
+ // Always, the empty class included: "" is the default rung, and a
+ // destination left on its previous class would not be the state that
+ // was exported.
+ return reg.SetReplication(host, ws.Replication)
+}
+
+// StageWake stages the wake state as a platform item. Unchanged state
+// stages nothing.
+func StageWake(ctx context.Context, p *Publisher, ws WakeState) error {
+ body, err := json.Marshal(ws)
+ if err != nil {
+ return fmt.Errorf("hostfiles: marshal wake state: %w", err)
+ }
+ return p.StagePlatform(ctx, WakeItem, body)
+}
+
+// LoadWake reads the wake state a manifest carries. The bool is false only
+// when the manifest has no wake entry at all. A manifest that names wake
+// state whose object is missing or damaged is an error: that is lost
+// timers, not a host with none, and the caller must not install as if it
+// were.
+func LoadWake(ctx context.Context, s release.Store, prefix string, m Manifest) (WakeState, bool, error) {
+ if _, ok := m.Platform[WakeItem]; !ok {
+ return WakeState{}, false, nil
+ }
+ body, err := ReadPlatform(ctx, s, prefix, m, WakeItem)
+ if err != nil {
+ return WakeState{}, false, fmt.Errorf("hostfiles: the manifest names wake state that cannot be read: %w", err)
+ }
+ var ws WakeState
+ if err := json.Unmarshal(body, &ws); err != nil {
+ return WakeState{}, false, fmt.Errorf("hostfiles: wake state: %w", err)
+ }
+ return ws, true, nil
+}
diff --git a/internal/hostfiles/wake_test.go b/internal/hostfiles/wake_test.go
new file mode 100644
index 000000000..36b01d234
--- /dev/null
+++ b/internal/hostfiles/wake_test.go
@@ -0,0 +1,176 @@
+package hostfiles
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func wakeReg(t *testing.T) *registry.Registry {
+ t.Helper()
+ r, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { r.Close() })
+ return r
+}
+
+func seedHost(t *testing.T, r *registry.Registry, host string) {
+ t.Helper()
+ if err := r.Upsert(registry.Route{Host: host, Kind: "service", Addr: "10.0.0.1:3000", Status: "ready", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestWakeStateRoundTripsThroughTheManifest(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ src, dst := wakeReg(t), wakeReg(t)
+ seedHost(t, src, "shop.test")
+ seedHost(t, dst, "shop.test")
+ if err := src.ReplaceTimers("shop.test", []registry.Timer{
+ {Host: "shop.test", Name: "nightly", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, LastAt: 4000,
+ LastStatus: "ok", Attempts: 1, CreatedAt: 10, DueAt: 4900, Registration: "reg-n"},
+ {Host: "shop.test", Name: "remind", Kind: "once", Spec: "2026-10-06T00:00:00Z", Path: "/r", NextAt: 6000, CreatedAt: 11, DueAt: 6000, Registration: "reg-r"},
+ {Host: "shop.test", Name: "done", Kind: "once", Spec: "2026-10-01T00:00:00Z", Path: "/d", NextAt: 0, LastAt: 3000, CreatedAt: 9, Registration: "reg-d"},
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if err := src.SetNextDue("shop.test", 7777); err != nil {
+ t.Fatal(err)
+ }
+ if err := src.SetReplication("shop.test", registry.ReplicationBatched); err != nil {
+ t.Fatal(err)
+ }
+
+ ws, err := ExportWake(src, "shop.test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(ws.Timers) != 2 {
+ t.Fatalf("exported %d timers, want 2: a finished once-timer is history, not wake state", len(ws.Timers))
+ }
+ p := open(t, s)
+ if err := StageWake(ctx, p, ws); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+
+ m, _, err := Load(ctx, s, pfx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, found, err := LoadWake(ctx, s, pfx, m)
+ if err != nil || !found {
+ t.Fatalf("LoadWake found=%v err=%v", found, err)
+ }
+ if err := ImportWake(dst, "shop.test", got); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ := dst.TimersFor("shop.test")
+ if len(rows) != 2 || rows[0].Name != "nightly" || rows[0].Registration != "reg-n" || rows[0].NextAt != 5000 || rows[0].Attempts != 1 {
+ t.Fatalf("imported timers: %+v", rows)
+ }
+ rt, err := dst.Get("shop.test")
+ if err != nil || rt == nil {
+ t.Fatalf("route: %v %v", rt, err)
+ }
+ if rt.NextDue != 7777 || rt.Replication != registry.ReplicationBatched {
+ t.Fatalf("NextDue=%d Replication=%q", rt.NextDue, rt.Replication)
+ }
+}
+
+func TestStageWakeOfUnchangedStateStagesNothing(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ ws := WakeState{V: 1, Host: "shop.test", NextDue: 5}
+ if err := StageWake(ctx, p, ws); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ if err := StageWake(ctx, p, ws); err != nil {
+ t.Fatal(err)
+ }
+ if p.Dirty() {
+ t.Fatal("unchanged wake state was staged again")
+ }
+}
+
+func TestLoadWakeWithNoneReportsNotFound(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ m, _, _ := Load(ctx, s, pfx)
+ if _, found, err := LoadWake(ctx, s, pfx, m); found || err != nil {
+ t.Fatalf("found=%v err=%v", found, err)
+ }
+}
+
+// A manifest that names wake state whose object has gone is lost data,
+// not a host without timers.
+func TestLoadWakeWithItsObjectMissingIsAnError(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ if err := StageWake(ctx, p, WakeState{V: 1, Host: "shop.test", NextDue: 5}); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ m, _, err := Load(ctx, s, pfx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := s.Delete(ctx, pfx+m.Platform[WakeItem].Object); err != nil {
+ t.Fatal(err)
+ }
+ if _, found, err := LoadWake(ctx, s, pfx, m); err == nil || found {
+ t.Fatalf("found = %v, err = %v; a missing wake object must be an error", found, err)
+ }
+}
+
+func TestImportWakeRefusesAnotherHostsState(t *testing.T) {
+ r := wakeReg(t)
+ seedHost(t, r, "shop.test")
+ if err := ImportWake(r, "shop.test", WakeState{V: 1, Host: "other.test"}); err == nil {
+ t.Fatal("imported another host's wake state")
+ }
+}
+
+func TestImportWakeRefusesAnUnknownReplicationClassAndChangesNothing(t *testing.T) {
+ r := wakeReg(t)
+ seedHost(t, r, "shop.test")
+ if err := r.ReplaceTimers("shop.test", []registry.Timer{{Host: "shop.test", Name: "keep", Kind: "schedule", Spec: "@every 1m", Path: "/k", NextAt: 1, CreatedAt: 1}}); err != nil {
+ t.Fatal(err)
+ }
+ err := ImportWake(r, "shop.test", WakeState{V: 1, Host: "shop.test", Replication: "hourly"})
+ if err == nil {
+ t.Fatal("accepted replication class \"hourly\"")
+ }
+ if rows, _ := r.TimersFor("shop.test"); len(rows) != 1 {
+ t.Fatalf("a refused import changed the timers: %+v", rows)
+ }
+}
+
+func TestExportWakeOfAnUnknownHostIsAnError(t *testing.T) {
+ if _, err := ExportWake(wakeReg(t), "nobody.test"); err == nil {
+ t.Fatal("exported wake state for a host with no route")
+ }
+}
+
+func TestImportWakeOfTheDefaultClassReplacesAnotherClass(t *testing.T) {
+ r := wakeReg(t)
+ seedHost(t, r, "shop.test")
+ if err := r.SetReplication("shop.test", registry.ReplicationDaily); err != nil {
+ t.Fatal(err)
+ }
+ if err := ImportWake(r, "shop.test", WakeState{V: 1, Host: "shop.test", Replication: ""}); err != nil {
+ t.Fatal(err)
+ }
+ rt, err := r.Get("shop.test")
+ if err != nil || rt == nil {
+ t.Fatalf("route: %v %v", rt, err)
+ }
+ if rt.Replication != "" {
+ t.Fatalf("Replication = %q, want the default", rt.Replication)
+ }
+}
diff --git a/internal/registry/timers.go b/internal/registry/timers.go
index cec61318c..89f65b2e0 100644
--- a/internal/registry/timers.go
+++ b/internal/registry/timers.go
@@ -10,6 +10,7 @@ package registry
import (
"database/sql"
"errors"
+ "fmt"
"os"
"strings"
)
@@ -67,6 +68,36 @@ WHERE timers.kind=excluded.kind`,
return nil
}
+// ReplaceTimers makes host's timer rows exactly rows, in one transaction,
+// keeping every column as given: next_at, due_at, attempts and
+// registration included. It is how an installation imports the wake state
+// it was handed (region HA spec 1 §5.7, moves spec §9.4).
+//
+// UpsertTimer is the wrong tool for that: it mints a fresh registration
+// and resets the runner's bookkeeping on a rearm.
+func (r *Registry) ReplaceTimers(host string, rows []Timer) error {
+ for _, t := range rows {
+ if t.Host != host {
+ return fmt.Errorf("registry: timer %q is for host %q, not %q", t.Name, t.Host, host)
+ }
+ }
+ tx, err := r.sql.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil {
+ return err
+ }
+ for _, t := range rows {
+ if _, err := tx.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, t.Registration); err != nil {
+ return fmt.Errorf("registry: insert timer %q: %w", t.Name, err)
+ }
+ }
+ return tx.Commit()
+}
+
func (r *Registry) GetTimer(host, name string) (*Timer, error) {
t, err := scanTimer(r.sql.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, host, name))
if errors.Is(err, sql.ErrNoRows) {
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go
index 004cada04..2210ab0e8 100644
--- a/internal/registry/timers_test.go
+++ b/internal/registry/timers_test.go
@@ -454,3 +454,80 @@ func TestUnchangedTimerUpsertConcurrentWithCompletion(t *testing.T) {
}
}
}
+
+func TestReplaceTimersSwapsAHostsRowsExactly(t *testing.T) {
+ r := openTimersReg(t)
+ for _, tm := range []Timer{
+ {Host: "a.test", Name: "old", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1},
+ {Host: "b.test", Name: "other", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1},
+ } {
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ rows := []Timer{
+ {Host: "a.test", Name: "nightly", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, LastAt: 4000,
+ LastStatus: "ok", LastMs: 12, Attempts: 2, CreatedAt: 10, DueAt: 4900, Registration: "reg-nightly"},
+ {Host: "a.test", Name: "once-1", Kind: "once", Spec: "2026-10-06T00:00:00Z", Path: "/o", NextAt: 6000, CreatedAt: 11, DueAt: 6000, Registration: "reg-once"},
+ }
+ if err := r.ReplaceTimers("a.test", rows); err != nil {
+ t.Fatal(err)
+ }
+ got, err := r.TimersFor("a.test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 2 {
+ t.Fatalf("a.test has %d timers, want 2: %+v", len(got), got)
+ }
+ n := got[0] // ordered by name: nightly, once-1
+ if n.Name != "nightly" || n.NextAt != 5000 || n.LastAt != 4000 || n.LastStatus != "ok" || n.LastMs != 12 ||
+ n.Attempts != 2 || n.CreatedAt != 10 || n.DueAt != 4900 || n.Registration != "reg-nightly" {
+ t.Fatalf("bookkeeping not preserved: %+v", n)
+ }
+ if other, _ := r.TimersFor("b.test"); len(other) != 1 {
+ t.Fatalf("b.test has %d timers, want 1: another host's rows were touched", len(other))
+ }
+}
+
+func TestReplaceTimersIsAllOrNothing(t *testing.T) {
+ r := openTimersReg(t)
+ if err := r.UpsertTimer(Timer{Host: "a.test", Name: "keep", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1}); err != nil {
+ t.Fatal(err)
+ }
+ // Two rows with one name: the delete and the first insert succeed, the
+ // second insert breaks the primary key. Without a transaction the host
+ // would be left with "dup" and without "keep".
+ dup := []Timer{
+ {Host: "a.test", Name: "dup", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 1, CreatedAt: 1},
+ {Host: "a.test", Name: "dup", Kind: "schedule", Spec: "@every 2m", Path: "/t", NextAt: 1, CreatedAt: 1},
+ }
+ if err := r.ReplaceTimers("a.test", dup); err == nil {
+ t.Fatal("two rows with one name were accepted")
+ }
+ got, _ := r.TimersFor("a.test")
+ if len(got) != 1 || got[0].Name != "keep" {
+ t.Fatalf("a failed replace changed the rows: %+v", got)
+ }
+}
+
+func TestReplaceTimersRefusesARowForAnotherHost(t *testing.T) {
+ r := openTimersReg(t)
+ rows := []Timer{{Host: "elsewhere.test", Name: "y", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 1, CreatedAt: 1}}
+ if err := r.ReplaceTimers("a.test", rows); err == nil {
+ t.Fatal("a row for another host was accepted")
+ }
+}
+
+func TestReplaceTimersWithNoRowsClearsTheHost(t *testing.T) {
+ r := openTimersReg(t)
+ if err := r.UpsertTimer(Timer{Host: "a.test", Name: "gone", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1}); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.ReplaceTimers("a.test", nil); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := r.TimersFor("a.test"); len(got) != 0 {
+ t.Fatalf("rows left: %+v", got)
+ }
+}