amadan
registry: hold_op, SetHold and the conditional ReleaseHold (moves §5.2)
Adds Route.HoldOp (idempotent additive migration, routeCols, scanRoute, Upsert), HoldStatuses/IsHold, SetHold, ReleaseHold, HoldRows and DeleteIfHold. Upsert keeps an existing hold and its status when the incoming row carries no hold_op (F29).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+341
−7
internal/registry/hold_test.go+229 −0internal/registry/registry.go+112 −7
diff --git a/internal/registry/hold_test.go b/internal/registry/hold_test.go| new file mode 100644 |
| index 000000000..670844901 |
| --- /dev/null |
| +++ b/internal/registry/hold_test.go |
| @@ -0,0 +1,229 @@ |
| +package registry |
| + |
| +import ( |
| + "path/filepath" |
| + "testing" |
| +) |
| + |
| +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) |
| + } |
| +} |
| + |
| +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() |
| + } |
| +} |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index ba96979df..b7b330950 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -56,6 +56,13 @@ type Route struct { |
| // failing probe. |
| Health string |
| Status string // pending|creating|starting|issuing|ready|failed |
| + // HoldOp is the operation holding Status (moves spec §5.2); "" when not |
| + // held. A held row (Status in HoldStatuses) belongs to that operation: the |
| + // edge never serves it, ticks refuse it and no ordinary wake reaches it. |
| + // Only SetHold and ReleaseHold change a hold; Upsert keeps an existing one |
| + // when the incoming row carries none. Deliberately NOT in Fingerprint(): |
| + // every hold also moves Status, which is. |
| + HoldOp string |
| LastError string |
| DBPath string |
| @@ -508,6 +515,7 @@ CREATE TABLE IF NOT EXISTS routes ( |
| `ALTER TABLE routes ADD COLUMN max_run_seconds INTEGER NOT NULL DEFAULT 0`, |
| `ALTER TABLE routes ADD COLUMN replicas TEXT NOT NULL DEFAULT ''`, |
| `ALTER TABLE routes ADD COLUMN replica INTEGER NOT NULL DEFAULT 0`, |
| + `ALTER TABLE routes ADD COLUMN hold_op TEXT NOT NULL DEFAULT ''`, |
| } { |
| if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") { |
| return err |
| @@ -751,7 +759,7 @@ func (r *Registry) Accounts() ([]Account, error) { |
| } |
| // routeCols is the column list every route read and the one route write |
| -// share — 42 names, in this exact order. scanRoute's destinations, Upsert's |
| +// share — 43 names, in this exact order. scanRoute's destinations, Upsert's |
| // placeholder row and Upsert's argument list all count from here: adding a |
| // column means adding it in FOUR places, and a miscount is a silent column |
| // shift rather than an error (TestUpsertRoundTripsAllColumns is what catches |
| @@ -761,7 +769,7 @@ version, prev_version, app, pool, region, channel, hibernate, canary, |
| replication, next_due, backing, status_named, platform, dns_status, |
| dns_checked_at, cert_expires_at, account, environment, caps, tunnel, |
| provisioned, seed, unit, fleet_box, addrs, health, steering, gate, created_at, |
| -args, idle_seconds, max_run_seconds, replicas, replica` |
| +args, idle_seconds, max_run_seconds, replicas, replica, hold_op` |
| func scanRoute(scan func(...any) error) (Route, error) { |
| var t Route |
| @@ -772,7 +780,7 @@ func scanRoute(scan func(...any) error) (Route, error) { |
| &sn, &plat, &t.DNSStatus, &t.DNSCheckedAt, &t.CertExpiresAt, &t.Account, |
| &t.Environment, &t.Caps, &t.Tunnel, &prov, &t.Seed, &t.Unit, &t.FleetBox, |
| &t.Addrs, &t.Health, &t.Steering, &t.Gate, &t.CreatedAt, |
| - &t.Args, &t.IdleSeconds, &t.MaxRunSeconds, &t.Replicas, &rep) |
| + &t.Args, &t.IdleSeconds, &t.MaxRunSeconds, &t.Replicas, &rep, &t.HoldOp) |
| t.Hibernate = hib != 0 |
| t.Canary = can != 0 |
| t.StatusNamed = sn != 0 |
| @@ -834,7 +842,11 @@ func (r *Registry) Delete(host string) error { |
| return err |
| } |
| -// Upsert inserts or updates a route by host. Twelve columns are deliberately |
| +// Upsert inserts or updates a route by host. A held row (a non-empty hold_op) is |
| +// not released by it: when the incoming Route carries no HoldOp, the existing |
| +// row keeps its hold_op AND its status (only SetHold and ReleaseHold change a |
| +// hold; moves spec §5.2, defence in depth beside the reconciler's own skip). |
| +// Twelve columns are deliberately |
| // not plain overwrites — `environment`, `caps`, `tunnel`, `provisioned`, |
| // `seed`, `unit`, `fleet_box`, `replication`, `gate`, `args`, `idle_seconds` |
| // and `max_run_seconds` all keep the existing row's value when the incoming |
| @@ -987,10 +999,13 @@ func (r *Registry) Upsert(t Route) error { |
| } |
| _, err := r.sql.Exec(` |
| INSERT INTO routes (`+routeCols+`) |
| -VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| +VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| ON CONFLICT(host) DO UPDATE SET |
| slug=excluded.slug, kind=excluded.kind, socket=excluded.socket, |
| - addr=excluded.addr, status=excluded.status, last_error=excluded.last_error, |
| + addr=excluded.addr, |
| + status=CASE WHEN excluded.hold_op = '' AND routes.hold_op != '' THEN routes.status ELSE excluded.status END, |
| + hold_op=CASE WHEN excluded.hold_op = '' THEN routes.hold_op ELSE excluded.hold_op END, |
| + last_error=excluded.last_error, |
| db_path=excluded.db_path, version=excluded.version, |
| prev_version=excluded.prev_version, app=excluded.app, pool=excluded.pool, |
| region=excluded.region, channel=excluded.channel, |
| @@ -1021,10 +1036,100 @@ ON CONFLICT(host) DO UPDATE SET |
| t.CertExpiresAt, t.Account, t.Environment, t.Caps, t.Tunnel, |
| b2i(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.Addrs, t.Health, |
| t.Steering, t.Gate, t.CreatedAt, t.Args, t.IdleSeconds, t.MaxRunSeconds, |
| - t.Replicas, b2i(t.Replica)) |
| + t.Replicas, b2i(t.Replica), t.HoldOp) |
| return err |
| } |
| +// HoldStatuses are the row statuses the edge never serves, ticks refuse and |
| +// no ordinary wake reaches (moves spec §5.2). |
| +var HoldStatuses = map[string]bool{"moving": true, "landing": true, "restoring": true, "releasing": true} |
| + |
| +// IsHold reports whether status is one of HoldStatuses. |
| +func IsHold(status string) bool { return HoldStatuses[status] } |
| + |
| +// SetHold writes a hold status with the operation holding it. It refuses a |
| +// status that is not in HoldStatuses and an empty op, and errors when no row |
| +// matched. |
| +// |
| +// Every SetHold caller also touches the stage-requested marker (moves Task |
| +// 13's rollback hold rule relies on it); SetHold itself does not, because the |
| +// marker lives outside this table. |
| +func (r *Registry) SetHold(host, status, op string) error { |
| + if !IsHold(status) || op == "" { |
| + return fmt.Errorf("registry: %q with op %q is not a hold", status, op) |
| + } |
| + res, err := r.sql.Exec(`UPDATE routes SET status = ?, hold_op = ? WHERE host = ?`, status, op, host) |
| + if err != nil { |
| + return err |
| + } |
| + n, err := res.RowsAffected() |
| + if err != nil { |
| + return err |
| + } |
| + if n == 0 { |
| + return fmt.Errorf("no route for host %q", host) |
| + } |
| + return nil |
| +} |
| + |
| +// ReleaseHold is §5.2's one conditional statement: a row leaves a hold only |
| +// under the op that holds it. false means nothing matched (a late write from a |
| +// finished operation, an unheld row, or an empty op). |
| +func (r *Registry) ReleaseHold(host, op string) (bool, error) { |
| + if op == "" { |
| + return false, nil |
| + } |
| + res, err := r.sql.Exec(`UPDATE routes SET status = 'ready', hold_op = '' WHERE host = ? AND hold_op = ?`, host, op) |
| + if err != nil { |
| + return false, err |
| + } |
| + n, err := res.RowsAffected() |
| + return n > 0, err |
| +} |
| + |
| +// DeleteIfHold deletes host's row only while op holds it: the source's row |
| +// deletion at finalization (moves spec §7.9 step 3). A row held by another |
| +// op, or not held, is not deleted and the answer is false. |
| +func (r *Registry) DeleteIfHold(host, op string) (bool, error) { |
| + if op == "" { |
| + return false, nil |
| + } |
| + res, err := r.sql.Exec(`DELETE FROM routes WHERE host = ? AND hold_op = ?`, host, op) |
| + if err != nil { |
| + return false, err |
| + } |
| + n, err := res.RowsAffected() |
| + return n > 0, err |
| +} |
| + |
| +// HoldRows returns every row held by an operation (a non-empty hold_op), ordered by |
| +// host. Over a fresh connection, like List, so a reader in another process |
| +// sees the latest commit. |
| +func (r *Registry) HoldRows() ([]Route, error) { |
| + db, err := sql.Open("sqlite", r.path) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer db.Close() |
| + if _, err := db.Exec("PRAGMA busy_timeout=5000"); err != nil { |
| + return nil, err |
| + } |
| + rows, err := db.Query(`SELECT ` + routeCols + ` FROM routes WHERE hold_op != '' ORDER BY host`) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer rows.Close() |
| + var out []Route |
| + for rows.Next() { |
| + t, err := scanRoute(rows.Scan) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out = append(out, t) |
| + } |
| + return out, rows.Err() |
| +} |
| + |
| // SetAccount stamps a route's app→account linkage as a narrow UPDATE — the |
| // reconcile pass re-stamps this every tick from the bucket manifests, and a |
| // read-modify-write Upsert racing a deploy's SetVersions (or the activator's |