package registry import ( "errors" "path/filepath" "testing" "time" ) const ( holdOpA = "mv_abcdefghijklmnopqrstuvwxyz" holdOpB = "mv_other0000000000000000000000" ) func holdFixture(t *testing.T, host string) *Registry { t.Helper() r := openTestRegistry(t) if err := r.Upsert(Route{Host: host, Kind: "instance", Socket: "/s", App: "a", Hibernate: true}); err != nil { t.Fatal(err) } return r } func TestHoldIsConditional(t *testing.T) { r := holdFixture(t, "h") if err := r.SetHold("h", "moving", holdOpA); err != nil { t.Fatal(err) } rt, _ := r.Get("h") if rt.Status != "moving" || rt.HoldOp != holdOpA || !IsHold(rt.Status) { t.Fatalf("held row: %+v", rt) } if err := r.SetHold("h", "ready", "x"); err == nil { t.Fatal("ready is not a hold") } // §5.2: a late write from a finished operation matches nothing. if changed, err := r.ReleaseHold("h", holdOpB); err != nil || changed { t.Fatalf("release under another op: %v %v", changed, err) } if changed, err := r.ReleaseHold("h", holdOpA); err != nil || !changed { t.Fatalf("release: %v %v", changed, err) } rt, _ = r.Get("h") if rt.Status != "ready" || rt.HoldOp != "" { t.Fatalf("after release: %+v", rt) } // Upsert preserves hold_op through routeCols. if err := r.SetHold("h", "landing", holdOpA); err != nil { t.Fatal(err) } rt, _ = r.Get("h") if err := r.Upsert(*rt); err != nil { t.Fatal(err) } rt, _ = r.Get("h") if rt.HoldOp == "" { t.Fatal("Upsert dropped hold_op") } } func TestSetHoldRefusals(t *testing.T) { r := holdFixture(t, "h") for _, status := range []string{"moving", "landing", "restoring", "releasing"} { if err := r.SetHold("h", status, holdOpA); err != nil { t.Fatalf("%s: %v", status, err) } if rt, _ := r.Get("h"); rt.Status != status || rt.HoldOp != holdOpA { t.Fatalf("%s: %+v", status, rt) } } for _, status := range []string{"", "ready", "pending", "failed", "Moving"} { if err := r.SetHold("h", status, holdOpA); err == nil { t.Fatalf("status %q accepted as a hold", status) } } if err := r.SetHold("h", "moving", ""); err == nil { t.Fatal("a hold with no op was accepted") } if err := r.SetHold("nope", "moving", holdOpA); err == nil { t.Fatal("SetHold on an unknown host did not error") } if IsHold("ready") || !IsHold("releasing") { t.Fatal("IsHold disagrees with HoldStatuses") } } // An Upsert from a writer that knows nothing of holds (carlos add, the // reconciler's mint paths) must not release one: the incoming row has an empty // hold_op and a ready status, and the held row keeps both (F29). func TestUpsertDoesNotClearHold(t *testing.T) { r := holdFixture(t, "h") if err := r.SetHold("h", "moving", holdOpA); err != nil { t.Fatal(err) } if err := r.Upsert(Route{Host: "h", Kind: "instance", Socket: "/s2", App: "a", Hibernate: true}); err != nil { t.Fatal(err) } rt, _ := r.Get("h") if rt.Status != "moving" || rt.HoldOp != holdOpA { t.Fatalf("Upsert released the hold: %+v", rt) } if rt.Socket != "/s2" { t.Fatalf("the rest of the row did not update: %+v", rt) } // Only ReleaseHold ends it, after which Upsert is an ordinary write again. if ok, err := r.ReleaseHold("h", holdOpA); err != nil || !ok { t.Fatalf("release: %v %v", ok, err) } if err := r.Upsert(Route{Host: "h", Kind: "instance", Socket: "/s3", Status: "starting"}); err != nil { t.Fatal(err) } if rt, _ = r.Get("h"); rt.Status != "starting" || rt.HoldOp != "" { t.Fatalf("Upsert on an unheld row: %+v", rt) } } func TestHoldRows(t *testing.T) { r := openTestRegistry(t) if rows, err := r.HoldRows(); err != nil || len(rows) != 0 { t.Fatalf("empty registry: %v %v", rows, err) } for _, h := range []string{"a", "b", "c"} { if err := r.Upsert(Route{Host: h, Kind: "instance", Socket: "/s"}); err != nil { t.Fatal(err) } } if err := r.SetHold("c", "restoring", holdOpA); err != nil { t.Fatal(err) } if err := r.SetHold("a", "landing", holdOpB); err != nil { t.Fatal(err) } rows, err := r.HoldRows() if err != nil { t.Fatal(err) } if len(rows) != 2 || rows[0].Host != "a" || rows[0].HoldOp != holdOpB || rows[0].Status != "landing" || rows[1].Host != "c" || rows[1].HoldOp != holdOpA || rows[1].Status != "restoring" { t.Fatalf("HoldRows = %+v", rows) } // A bare "moving" status written by SetStatus has no op, so it is not a // held row in this sense. if err := r.SetStatus("b", "moving"); err != nil { t.Fatal(err) } if rows, _ = r.HoldRows(); len(rows) != 2 { t.Fatalf("a status without an op counted as a hold: %+v", rows) } } func TestDeleteIfHold(t *testing.T) { r := holdFixture(t, "h") // Not held: nothing is deleted. if ok, err := r.DeleteIfHold("h", holdOpA); err != nil || ok { t.Fatalf("unheld row: %v %v", ok, err) } if err := r.SetHold("h", "releasing", holdOpA); err != nil { t.Fatal(err) } // Held by another op: nothing is deleted. if ok, err := r.DeleteIfHold("h", holdOpB); err != nil || ok { t.Fatalf("row held by another op: %v %v", ok, err) } // An empty op must not match unheld rows' empty hold_op. if err := r.Upsert(Route{Host: "plain", Kind: "instance", Socket: "/s"}); err != nil { t.Fatal(err) } if ok, err := r.DeleteIfHold("plain", ""); err != nil || ok { t.Fatalf("empty op deleted an unheld row: %v %v", ok, err) } if rt, _ := r.Get("plain"); rt == nil { t.Fatal("unheld row was deleted") } if rt, _ := r.Get("h"); rt == nil || rt.HoldOp != holdOpA { t.Fatalf("held row changed by the refusals: %+v", rt) } if ok, err := r.DeleteIfHold("h", holdOpA); err != nil || !ok { t.Fatalf("delete under the holding op: %v %v", ok, err) } if rt, _ := r.Get("h"); rt != nil { t.Fatalf("row survived: %+v", rt) } if ok, err := r.DeleteIfHold("h", holdOpA); err != nil || ok { t.Fatalf("second delete: %v %v", ok, err) } } // The source's finalization (moves spec §7.9 step 3, ruling H5(d)): the row // and the host's timer rows go in one transaction, and only while the op // holds the row. Any refusal leaves both, and the neighbour's timers stay. func TestDeleteIfHoldWithTimers(t *testing.T) { r := holdFixture(t, "h") if err := r.Upsert(Route{Host: "n", Kind: "instance", Socket: "/n", App: "a", Hibernate: true}); err != nil { t.Fatal(err) } for _, tm := range []Timer{{Host: "h", Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}, {Host: "h", Name: "hourly", Kind: "cron", Spec: "0 * * * *", NextAt: 2000}, {Host: "n", Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}} { if err := r.UpsertTimer(tm); err != nil { t.Fatal(err) } } timers := func(host string) int { t.Helper() got, err := r.TimersFor(host) if err != nil { t.Fatal(err) } return len(got) } // Not held, held by another op, an empty op: nothing goes. for _, c := range []struct{ name, hold, op string }{{"unheld", "", holdOpA}, {"another op", holdOpB, holdOpA}, {"empty op", holdOpA, ""}} { if c.hold != "" { if err := r.SetHold("h", "moving", c.hold); err != nil { t.Fatal(err) } } if ok, err := r.DeleteIfHoldWithTimers("h", c.op); err != nil || ok { t.Fatalf("%s: %v %v", c.name, ok, err) } if rt, _ := r.Get("h"); rt == nil || timers("h") != 2 { t.Fatalf("%s: the row or its timers changed: %+v, %d timers", c.name, rt, timers("h")) } if c.hold != "" { if ok, err := r.ReleaseHold("h", c.hold); err != nil || !ok { t.Fatal(ok, err) } } } if err := r.SetHold("h", "moving", holdOpA); err != nil { t.Fatal(err) } if ok, err := r.DeleteIfHoldWithTimers("h", holdOpA); err != nil || !ok { t.Fatalf("delete under the holding op: %v %v", ok, err) } if rt, _ := r.Get("h"); rt != nil || timers("h") != 0 { t.Fatalf("after the delete: %+v, %d timers", rt, timers("h")) } if rt, _ := r.Get("n"); rt == nil || timers("n") != 1 { t.Fatalf("the neighbour changed: %+v, %d timers", rt, timers("n")) } // A replay finds no row held by the op and deletes nothing, timers // written since included. if err := r.UpsertTimer(Timer{Host: "h", Name: "late", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}); err != nil { t.Fatal(err) } if ok, err := r.DeleteIfHoldWithTimers("h", holdOpA); err != nil || ok || timers("h") != 1 { t.Fatalf("replay: %v %v, %d timers", ok, err, timers("h")) } } func TestReleaseHoldEmptyOpMatchesNothing(t *testing.T) { r := holdFixture(t, "h") if ok, err := r.ReleaseHold("h", ""); err != nil || ok { t.Fatalf("release with empty op on an unheld row: %v %v", ok, err) } if err := r.SetStatus("h", "starting"); err != nil { t.Fatal(err) } if ok, _ := r.ReleaseHold("h", ""); ok { t.Fatal("an empty op released an unheld row and reset its status") } if rt, _ := r.Get("h"); rt.Status != "starting" { t.Fatalf("status moved: %+v", rt) } } // The migration is additive and idempotent: opening a registry that predates // the column adds it, and opening again is a no-op. func TestHoldOpMigrationIdempotentOnLegacyDB(t *testing.T) { path := filepath.Join(t.TempDir(), "registry.db") r, err := Open(path) if err != nil { t.Fatal(err) } if err := r.Upsert(Route{Host: "h", Kind: "instance", Socket: "/s"}); err != nil { t.Fatal(err) } // Strip the column to look like a pre-hold registry. if _, err := r.sql.Exec(`ALTER TABLE routes DROP COLUMN hold_op`); err != nil { t.Fatal(err) } r.Close() for i := 0; i < 2; i++ { r, err = Open(path) if err != nil { t.Fatalf("open %d: %v", i, err) } rt, err := r.Get("h") if err != nil || rt == nil || rt.HoldOp != "" || rt.Status != "ready" { t.Fatalf("open %d: %+v %v", i, rt, err) } r.Close() } } // Fix round 1: on conflict Upsert never changes hold_op, and a held row keeps // its status whatever the incoming row carries. Only SetHold, ReleaseHold and // DeleteIfHold change a hold. func TestUpsertStaleOpCannotTakeAHoldBack(t *testing.T) { r := holdFixture(t, "h") if err := r.SetHold("h", "moving", holdOpA); err != nil { t.Fatal(err) } stale, _ := r.Get("h") // a caller reads op A's row ... if ok, err := r.ReleaseHold("h", holdOpA); err != nil || !ok { t.Fatal(ok, err) } if err := r.SetHold("h", "landing", holdOpB); err != nil { // ... B takes the row ... t.Fatal(err) } if err := r.Upsert(*stale); err != nil { // ... and the stale write arrives. t.Fatal(err) } rt, _ := r.Get("h") if rt.HoldOp != holdOpB || rt.Status != "landing" { t.Fatalf("stale Upsert changed B's hold: %+v", rt) } if ok, err := r.ReleaseHold("h", holdOpA); err != nil || ok { t.Fatalf("ReleaseHold under the stale op: %v %v", ok, err) } if rt, _ = r.Get("h"); rt.HoldOp != holdOpB { t.Fatalf("hold lost: %+v", rt) } } func TestUpsertSameOpWithReadyStatusKeepsHeldStatus(t *testing.T) { r := holdFixture(t, "h") if err := r.SetHold("h", "restoring", holdOpA); err != nil { t.Fatal(err) } for _, status := range []string{"ready", "failed", "", "landing"} { if err := r.Upsert(Route{Host: "h", Kind: "instance", Socket: "/s", Status: status, HoldOp: holdOpA}); err != nil { t.Fatal(err) } rt, _ := r.Get("h") if rt.Status != "restoring" || rt.HoldOp != holdOpA { t.Fatalf("incoming status %q: %+v", status, rt) } } } func TestUpsertCannotHoldAnUnheldRowOnConflict(t *testing.T) { r := holdFixture(t, "h") if err := r.Upsert(Route{Host: "h", Kind: "instance", Socket: "/s", Status: "moving", HoldOp: holdOpA}); err != nil { t.Fatal(err) } rt, _ := r.Get("h") if rt.HoldOp != "" { t.Fatalf("Upsert took a hold on conflict: %+v", rt) } if rows, _ := r.HoldRows(); len(rows) != 0 { t.Fatalf("HoldRows = %+v", rows) } if err := r.SetHold("h", "moving", holdOpA); err != nil { t.Fatal(err) } } // A held row can be materialised: a fresh INSERT writes the incoming hold. func TestUpsertInsertMaterialisesAHeldRow(t *testing.T) { r := openTestRegistry(t) if err := r.Upsert(Route{Host: "new", Kind: "instance", Socket: "/s", Status: "landing", HoldOp: holdOpA}); err != nil { t.Fatal(err) } rt, _ := r.Get("new") if rt.Status != "landing" || rt.HoldOp != holdOpA { t.Fatalf("%+v", rt) } } // The edge reloads only when Fingerprint moves: a hold handed from one op to // another with the status unchanged must move it, and an unheld registry must // hash exactly as it did before the column existed. func TestFingerprintSeesHoldOpChangeAndIgnoresUnheld(t *testing.T) { r := holdFixture(t, "h") fp := func() (string, string) { a, err := r.Fingerprint() if err != nil { t.Fatal(err) } b, err := r.SnapshotFingerprint() if err != nil { t.Fatal(err) } return a, b } if err := r.SetHold("h", "moving", holdOpA); err != nil { t.Fatal(err) } f1, s1 := fp() if ok, _ := r.ReleaseHold("h", holdOpA); !ok { t.Fatal("release") } if err := r.SetHold("h", "moving", holdOpB); err != nil { // same status, other op t.Fatal(err) } f2, s2 := fp() if f1 == f2 { t.Fatal("Fingerprint did not move when only hold_op changed") } if s1 == s2 { t.Fatal("SnapshotFingerprint did not move when only hold_op changed") } // Direct column change with the status untouched. if _, err := r.sql.Exec(`UPDATE routes SET hold_op = ? WHERE host = 'h'`, holdOpA); err != nil { t.Fatal(err) } if f3, s3 := fp(); f3 == f2 || s3 == s2 { t.Fatal("a bare hold_op change did not move both fingerprints") } } // Switch-off contract: an unheld registry hashes byte-for-byte as it did // before hold_op existed. The goldens were produced by the same two rows on // the parent commit's registry.go (8437004c^). func TestFingerprintsUnchangedForUnheldRows(t *testing.T) { u := openTestRegistry(t) if err := u.Upsert(Route{Host: "u", Kind: "instance", Socket: "/s", App: "a", CreatedAt: 1700000000}); err != nil { t.Fatal(err) } if err := u.Upsert(Route{Host: "v", Kind: "static", Status: "starting", Hibernate: true, CreatedAt: 1700000001}); err != nil { t.Fatal(err) } const wantF = "2:f1013971d496e9401cd6a9928a1f4e18669deda40089d2c2a8b84263b32f34c6" const wantS = "2:149e8f668771c7493bdb7d4245fdabe03e957b0909c4daa1d037484b11064b80" if got, _ := u.Fingerprint(); got != wantF { t.Fatalf("Fingerprint = %s, want %s", got, wantF) } if got, _ := u.SnapshotFingerprint(); got != wantS { t.Fatalf("SnapshotFingerprint = %s, want %s", got, wantS) } } // WithHoldsFrozen is root's activation guard (moves Task 13): while fn runs, // no other handle can write the registry, so a SetHold from the agent waits // and the held rows fn saw are still the held rows when it returns. func TestWithHoldsFrozenHoldsOffASetHold(t *testing.T) { r := holdFixture(t, "h") other, err := Open(r.Path()) if err != nil { t.Fatal(err) } defer other.Close() done := make(chan error, 1) err = r.WithHoldsFrozen(func(held []Route) error { if len(held) != 0 { t.Errorf("held = %+v", held) } go func() { done <- other.SetHold("h", "moving", holdOpA) }() select { case err := <-done: t.Errorf("SetHold finished while the holds were frozen: %v", err) done <- err // for the read below case <-time.After(300 * time.Millisecond): } return nil }) if err != nil { t.Fatal(err) } if err := <-done; err != nil { t.Fatalf("SetHold after the freeze: %v", err) } var seen []Route if err := r.WithHoldsFrozen(func(held []Route) error { seen = held; return nil }); err != nil { t.Fatal(err) } if len(seen) != 1 || seen[0].Host != "h" || seen[0].HoldOp != holdOpA { t.Fatalf("held after SetHold = %+v", seen) } // fn's error is the call's, and nothing fn did not do is written. sentinel := errors.New("refused") if err := r.WithHoldsFrozen(func([]Route) error { return sentinel }); !errors.Is(err, sentinel) { t.Fatalf("err = %v", err) } } // Moves Task 6 fix round 1: SetHoldIf writes a hold only while the row's // hold_op is what the caller expects, in one statement, so a hold decided // from an earlier read cannot land on a row another op has taken since. // expectOp is "" (the row was unheld) or op (the op changes its own hold's // status); taking over another op's hold is never one statement. func TestSetHoldIfIsConditionalOnTheCurrentOp(t *testing.T) { r := holdFixture(t, "h") if ok, err := r.SetHoldIf("h", "moving", holdOpA, ""); err != nil || !ok { t.Fatalf("unheld row, expecting unheld: %v %v", ok, err) } if rt, _ := r.Get("h"); rt.Status != "moving" || rt.HoldOp != holdOpA { t.Fatalf("after the first hold: %+v", rt) } // Held by A now; a writer that still thinks the row is unheld is refused // and changes nothing. if ok, err := r.SetHoldIf("h", "landing", holdOpB, ""); err != nil || ok { t.Fatalf("stale write over A's hold: %v %v", ok, err) } if rt, _ := r.Get("h"); rt.Status != "moving" || rt.HoldOp != holdOpA { t.Fatalf("the refused write changed the row: %+v", rt) } // A may change its own hold's status. if ok, err := r.SetHoldIf("h", "landing", holdOpA, holdOpA); err != nil || !ok { t.Fatalf("A changing its own hold: %v %v", ok, err) } if rt, _ := r.Get("h"); rt.Status != "landing" || rt.HoldOp != holdOpA { t.Fatalf("after A's status change: %+v", rt) } // B expecting A's hold: not a takeover this statement allows. if _, err := r.SetHoldIf("h", "moving", holdOpB, holdOpA); err == nil { t.Fatal("SetHoldIf accepted expectOp of another op") } if rt, _ := r.Get("h"); rt.HoldOp != holdOpA { t.Fatalf("the refused takeover changed the row: %+v", rt) } // SetHold's own refusals hold. if _, err := r.SetHoldIf("h", "ready", holdOpA, holdOpA); err == nil { t.Fatal("ready accepted as a hold") } if _, err := r.SetHoldIf("h", "moving", "", ""); err == nil { t.Fatal("a hold with no op was accepted") } // No such row: nothing matched, not an error (the caller re-judges // next pass either way). if ok, err := r.SetHoldIf("nope", "moving", holdOpA, ""); err != nil || ok { t.Fatalf("unknown host: %v %v", ok, err) } // After a release the row is unheld again and "" matches once more. if ok, _ := r.ReleaseHold("h", holdOpA); !ok { t.Fatal("release") } if ok, err := r.SetHoldIf("h", "restoring", holdOpB, ""); err != nil || !ok { t.Fatalf("hold after release: %v %v", ok, err) } } // Ruling J3: a rollback takes over the move's hold in one statement, so the // row is never unheld between the two ops. func TestHandHold(t *testing.T) { const rb = "rb_abcdefghijklmnopqrstuvwxyz" r := holdFixture(t, "h") if err := r.SetHold("h", "landing", holdOpA); err != nil { t.Fatal(err) } // Held by another op: nothing moves. if ok, err := r.HandHold("h", "releasing", rb, holdOpB); err != nil || ok { t.Fatalf("hand-over from an op that does not hold the row: %v %v", ok, err) } if rt, _ := r.Get("h"); rt.Status != "landing" || rt.HoldOp != holdOpA { t.Fatalf("row changed by the refusal: %+v", rt) } // The hand-over: one write, the new op and status. if ok, err := r.HandHold("h", "releasing", rb, holdOpA); err != nil || !ok { t.Fatalf("hand-over: %v %v", ok, err) } if rt, _ := r.Get("h"); rt.Status != "releasing" || rt.HoldOp != rb { t.Fatalf("after the hand-over: %+v", rt) } // A replay finds the row already handed over and matches nothing. if ok, err := r.HandHold("h", "releasing", rb, holdOpA); err != nil || ok { t.Fatalf("replay: %v %v", ok, err) } // An unheld row is not taken: that is SetHoldIf's, with expectOp "". if err := r.Upsert(Route{Host: "plain", Kind: "instance", Socket: "/s"}); err != nil { t.Fatal(err) } if ok, err := r.HandHold("plain", "restoring", rb, holdOpA); err != nil || ok { t.Fatalf("unheld row: %v %v", ok, err) } if rt, _ := r.Get("plain"); rt.HoldOp != "" || IsHold(rt.Status) { t.Fatalf("unheld row taken: %+v", rt) } // An absent row matches nothing. if ok, err := r.HandHold("absent", "restoring", rb, holdOpA); err != nil || ok { t.Fatalf("absent row: %v %v", ok, err) } // Malformed: not a hold status, an empty op on either side, the same op. for name, c := range map[string][3]string{ "ready": {"ready", rb, rb}, "no to": {"restoring", "", holdOpA}, "no from": {"restoring", rb, ""}, "same op": {"restoring", rb, rb}, } { if ok, err := r.HandHold("h", c[0], c[1], c[2]); err == nil || ok { t.Fatalf("%s: %v %v", name, ok, err) } } if rt, _ := r.Get("h"); rt.Status != "releasing" || rt.HoldOp != rb { t.Fatalf("row changed by a malformed hand-over: %+v", rt) } }