amadan

carlos / platform Public

zdd-5 task 2: canonicalise stored cold-handover bytes before the ownership comparison

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e8e1433ccc76f81a8995497bfe3230e3014b9d53 parent a44e05e
3 files changed, +222 −7
  • internal/registry/coldhandover.go +30 −2
  • internal/registry/coldhandover_test.go +181 −0
  • internal/registry/registry_test.go +11 −5
diff --git a/internal/registry/coldhandover.go b/internal/registry/coldhandover.go
index b41a2382d..c1770559e 100644
--- a/internal/registry/coldhandover.go
+++ b/internal/registry/coldhandover.go
@@ -14,6 +14,22 @@ type ColdRows struct {
Timers []Timer
}
+// canonicalJSON decodes stored bytes into a T and encodes them again with
+// this binary. Bytes written by an older binary lack any field added since,
+// so they decode with those fields empty and re-encode in today's shape: an
+// unchanged row then compares equal to a fresh encoding, while a row whose
+// newer fields have since been set still differs. This is the only
+// normalisation applied before the ownership comparisons below. An
+// undecodable body returns an error, and the caller keeps its fail-closed
+// ErrClaimOwnership.
+func canonicalJSON[T any](stored []byte) ([]byte, error) {
+ var v T
+ if err := json.Unmarshal(stored, &v); err != nil {
+ return nil, err
+ }
+ return json.Marshal(v)
+}
+
func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string, old ColdRows, next ColdRows) error {
if id == "" || retirement == "" || !old.Route.ClaimOwnership.Valid() || !next.Route.ClaimOwnership.Valid() || old.Route.Host != next.Route.Host || old.Route.ClaimIncarnation != next.Route.ClaimIncarnation || next.Route.ClaimStamp <= old.Route.ClaimStamp || len(old.Timers) != len(next.Timers) {
return ErrClaimOwnership
@@ -47,7 +63,11 @@ func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string,
var saved []byte
err = tx.QueryRowContext(ctx, `SELECT body FROM cold_handovers WHERE id=?`, id).Scan(&saved)
if err == nil {
- if !bytes.Equal(saved, encoded) {
+ // A body saved by a binary without today's Route fields is
+ // re-encoded first, or a handover started before an upgrade could
+ // never be replayed after it.
+ canonical, e := canonicalJSON[ColdRows](saved)
+ if e != nil || !bytes.Equal(canonical, encoded) {
return ErrClaimOwnership
}
current, e := scanRoute(tx.QueryRowContext(ctx, `SELECT `+routeCols+` FROM routes WHERE host=?`, next.Route.Host).Scan)
@@ -78,7 +98,15 @@ func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string,
if err != nil {
return err
}
- if !bytes.Equal(contents, archive.Contents) {
+ stored := []byte(archive.Contents)
+ if kind == "route" {
+ // An archive retired by a binary without today's Route
+ // fields is re-encoded before the comparison (see canonicalJSON).
+ if stored, err = canonicalJSON[Route](stored); err != nil {
+ return ErrClaimOwnership
+ }
+ }
+ if !bytes.Equal(contents, stored) {
return ErrClaimOwnership
}
return nil
diff --git a/internal/registry/coldhandover_test.go b/internal/registry/coldhandover_test.go
new file mode 100644
index 000000000..2657c74e4
--- /dev/null
+++ b/internal/registry/coldhandover_test.go
@@ -0,0 +1,181 @@
+package registry
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "path/filepath"
+ "testing"
+)
+
+// oldShape returns raw JSON with the four rollout fields deleted from the
+// route object, as a binary from before they existed would have written it.
+// route is the object holding them: the document itself, or the value under
+// the given key.
+func oldShape(t *testing.T, raw []byte, key string) []byte {
+ t.Helper()
+ var doc map[string]json.RawMessage
+ if err := json.Unmarshal(raw, &doc); err != nil {
+ t.Fatal(err)
+ }
+ target := doc
+ if key != "" {
+ target = map[string]json.RawMessage{}
+ if err := json.Unmarshal(doc[key], &target); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for _, f := range []string{"GoodVersion", "GoodDigest", "GoodDemand", "ServingVersion"} {
+ if _, ok := target[f]; !ok {
+ t.Fatalf("fixture has no %s to delete", f)
+ }
+ delete(target, f)
+ }
+ if key != "" {
+ b, err := json.Marshal(target)
+ if err != nil {
+ t.Fatal(err)
+ }
+ doc[key] = b
+ }
+ out, err := json.Marshal(doc)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return out
+}
+
+// A cold handover retired by a binary without the rollout fields must still
+// restore after the upgrade, at both comparison sites (the retirement archive
+// and the saved handover body), while a route whose fields differ stays
+// refused: the comparison is an ownership proof and is only canonicalised,
+// never loosened.
+func TestRestoreClaimRowsAcceptsPreUpgradeBytes(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
+ ctx := context.Background()
+ oldOwner := ClaimOwnership{"inc", 7, "installation-7", "route-7"}
+ newOwner := ClaimOwnership{"inc", 8, "installation-8", "route-8"}
+
+ // setup retires a route holding good (and whatever the route carries)
+ // and returns the registry plus the old and next rows.
+ setup := func(t *testing.T, good string) (*Registry, ColdRows, ColdRows) {
+ r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { r.Close() })
+ route := Route{Host: "shop.example", App: "app", ClaimOwnership: oldOwner}
+ if err = r.Upsert(route); err != nil {
+ t.Fatal(err)
+ }
+ if good != "" {
+ if err = r.SetGoodBuild(route.Host, GoodBuild{}, GoodBuild{Version: good, Digest: "aa", Demand: "swap"}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ current, err := r.Get(route.Host)
+ if err != nil || current == nil {
+ t.Fatalf("Get: %v", err)
+ }
+ old := ColdRows{Route: *current}
+ next := ColdRows{Route: *current}
+ next.Route.ClaimOwnership = newOwner
+ targets := []ClaimRow{{Kind: "route", Host: route.Host, Ownership: oldOwner}}
+ if err = r.RetireClaimRows(ctx, "ret", targets, func([]ClaimRowArchive) error { return nil }); err != nil {
+ t.Fatal(err)
+ }
+ return r, old, next
+ }
+ // rewriteArchive replaces the stored retirement archive's route contents
+ // with the pre-upgrade shape.
+ rewriteArchive := func(t *testing.T, r *Registry) {
+ var raw []byte
+ if err := r.sql.QueryRow(`SELECT archive FROM claim_row_retirements WHERE journal='ret'`).Scan(&raw); err != nil {
+ t.Fatal(err)
+ }
+ var a ClaimRowArchive
+ if err := json.Unmarshal(raw, &a); err != nil {
+ t.Fatal(err)
+ }
+ a.Contents = oldShape(t, a.Contents, "")
+ raw, err := json.Marshal(a)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = r.sql.Exec(`UPDATE claim_row_retirements SET archive=? WHERE journal='ret'`, raw); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ t.Run("archive site accepts the old shape", func(t *testing.T) {
+ r, old, next := setup(t, "")
+ rewriteArchive(t, r)
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); err != nil {
+ t.Fatalf("pre-upgrade archive refused: %v", err)
+ }
+ })
+ t.Run("archive site refuses a route whose rollout fields differ", func(t *testing.T) {
+ r, old, next := setup(t, "")
+ rewriteArchive(t, r)
+ old.Route.GoodVersion, next.Route.GoodVersion = "v9", "v9"
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatalf("err = %v, want ErrClaimOwnership", err)
+ }
+ })
+ t.Run("archive site still compares current-shape bytes exactly", func(t *testing.T) {
+ r, old, next := setup(t, "v1")
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); err != nil {
+ t.Fatalf("current-shape archive refused: %v", err)
+ }
+ r2, old2, next2 := setup(t, "v1")
+ old2.Route.GoodVersion, next2.Route.GoodVersion = "v2", "v2"
+ if err := r2.RestoreClaimRows(ctx, "h1", "ret", old2, next2); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatalf("err = %v, want ErrClaimOwnership", err)
+ }
+ })
+ t.Run("replay site accepts the old shape", func(t *testing.T) {
+ r, old, next := setup(t, "")
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); err != nil {
+ t.Fatal(err)
+ }
+ var body []byte
+ if err := r.sql.QueryRow(`SELECT body FROM cold_handovers WHERE id='h1'`).Scan(&body); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`UPDATE cold_handovers SET body=? WHERE id='h1'`, oldShape(t, body, "Route")); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); err != nil {
+ t.Fatalf("pre-upgrade handover body refused on replay: %v", err)
+ }
+ })
+ t.Run("replay site refuses a route whose rollout fields differ", func(t *testing.T) {
+ r, old, next := setup(t, "")
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); err != nil {
+ t.Fatal(err)
+ }
+ var body []byte
+ if err := r.sql.QueryRow(`SELECT body FROM cold_handovers WHERE id='h1'`).Scan(&body); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`UPDATE cold_handovers SET body=? WHERE id='h1'`, oldShape(t, body, "Route")); err != nil {
+ t.Fatal(err)
+ }
+ old.Route.GoodVersion, next.Route.GoodVersion = "v9", "v9"
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatalf("err = %v, want ErrClaimOwnership", err)
+ }
+ })
+ t.Run("replay site fails closed on an undecodable body", func(t *testing.T) {
+ r, old, next := setup(t, "")
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`UPDATE cold_handovers SET body=? WHERE id='h1'`, []byte("not json")); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.RestoreClaimRows(ctx, "h1", "ret", old, next); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatalf("err = %v, want ErrClaimOwnership", err)
+ }
+ })
+}
diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go
index baaa3a116..b162bcc8b 100644
--- a/internal/registry/registry_test.go
+++ b/internal/registry/registry_test.go
@@ -77,10 +77,12 @@ func TestUpsertRoundTripsAllColumns(t *testing.T) {
Environment: "demo", Caps: "host-role,port:25",
Hibernate: true, Canary: true, Replication: "batched",
NextDue: 4200, Backing: BackingUnit, Unit: "console.service",
- Steering: "latency",
- Replicas: "ie,uk,us",
- Gate: "$2a$10$distinct",
- CreatedAt: 1700000000,
+ Steering: "latency",
+ Replicas: "ie,uk,us",
+ Gate: "$2a$10$distinct",
+ CreatedAt: 1700000000,
+ GoodVersion: "good789", GoodDigest: "sha256:good", GoodDemand: "swap-demand",
+ ServingVersion: "serve012",
}
if err := reg.Upsert(want); err != nil {
t.Fatal(err)
@@ -2877,9 +2879,13 @@ func TestGoodBuildIsWrittenOnlyByItsOwnDoor(t *testing.T) {
if routes[0].Good() != g2 {
t.Fatalf("List reads %+v, want %+v", routes[0].Good(), g2)
}
- if err := reg.SetGoodBuild("nope.example.com", GoodBuild{}, g1); err == nil {
+ err = reg.SetGoodBuild("nope.example.com", GoodBuild{}, g1)
+ if err == nil {
t.Fatal("a missing row must be an error, not a silent no-op")
}
+ if errors.Is(err, ErrGoodBuildChanged) {
+ t.Fatalf("a missing row is not a stale expectation: %v", err)
+ }
}
// serving_version moves on every wake and swap, with no operator change