carlos / platform Public

registry: rows carry the lifetime pair; the hold, release and install setters write it with the status

Two columns, incarnation and serving_seq (INTEGER NOT NULL DEFAULT 0,
added idempotently), on Route as Incarnation and ServingSeq. A fresh
insert writes them; an Upsert over an existing row keeps them.
SetHoldIfSeq holds an unheld row of the same or an older lifetime with
the status, op and pair in one statement; ReleaseHoldSeq opens a row of
its own lifetime or a legacy row at 0, never a successor's;
InstallLifetimeIf moves an unheld row from one pair to the next. Both
fingerprints fold the pair in only when non-zero. The remote registry
dispatches the three setters. SetHold, SetHoldIf, HandHold, ReleaseHold
and DeleteIfHold are untouched.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev 2c824282f6304e0a74095162e41771923233b4ba parent f50d056
3 files changed, +472 −7
  • internal/registry/registry.go +141 −7
  • internal/registry/registry_test.go +323 −0
  • internal/registry/remote_dispatch.go +8 −0
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 4447cc982..77e476ba7 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -73,6 +73,20 @@ type Route struct {
// folded in only when non-empty so an unheld registry hashes exactly as it
// did before the column existed.
HoldOp string
+ // Incarnation and ServingSeq are the row's lifetime pair (self-serve
+ // moves plan 1 Task 7, design §6.4): the lifetime of the hostname the row
+ // serves, and the record's move_seq as the transition that produced the
+ // row's current state left it (the hold's for a held row, open's for a
+ // landed target, the abort's for a reopened source; the record's
+ // move_seq for a row minted outside a move). (0, 0) on every row written
+ // before the columns existed. A fresh insert writes them; an Upsert over
+ // an existing row never changes them; only SetHoldIfSeq, ReleaseHoldSeq
+ // and InstallLifetimeIf move them, each in the same statement as the
+ // status they belong with. IN Fingerprint() and SnapshotFingerprint,
+ // folded in only when non-zero, so a registry with no pair hashes as it
+ // did before the columns existed.
+ Incarnation uint64
+ ServingSeq uint64
// SeedSource is the record's seed as this row needs it
// (seeds.RowSeed's JSON, "" for none), written by the provision pass
// (seeded-instances spec §3.4): the activator's restore fallback reads
@@ -609,6 +623,8 @@ CREATE TABLE IF NOT EXISTS routes (
`ALTER TABLE routes ADD COLUMN build_state TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE routes ADD COLUMN seed_source TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE routes ADD COLUMN import_gate TEXT NOT NULL DEFAULT ''`,
+ `ALTER TABLE routes ADD COLUMN incarnation INTEGER NOT NULL DEFAULT 0`,
+ `ALTER TABLE routes ADD COLUMN serving_seq INTEGER NOT NULL DEFAULT 0`,
} {
if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
return err
@@ -897,10 +913,10 @@ 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, claim_incarnation, claim_stamp, claim_owner, claim_row_id, hold_op,
good_version, good_digest, good_demand, serving_version, update_settings, build_state,
-seed_source, import_gate`
+seed_source, import_gate, incarnation, serving_seq`
// routeColCount is the number of names in routeCols.
-const routeColCount = 55
+const routeColCount = 57
// placeholders is n comma-joined bind parameters.
func placeholders(n int) string {
@@ -918,7 +934,7 @@ func scanRoute(scan func(...any) error) (Route, error) {
&t.Addrs, &t.Health, &t.Steering, &t.Gate, &t.CreatedAt,
&t.Args, &t.IdleSeconds, &t.MaxRunSeconds, &t.Replicas, &rep, &t.ClaimIncarnation, &t.ClaimStamp, &t.ClaimOwner, &t.ClaimRowID, &t.HoldOp,
&t.GoodVersion, &t.GoodDigest, &t.GoodDemand, &t.ServingVersion, &t.UpdateSettings, &t.BuildState,
- &t.SeedSource, &t.ImportGate)
+ &t.SeedSource, &t.ImportGate, &t.Incarnation, &t.ServingSeq)
t.Hibernate = hib != 0
t.Canary = can != 0
t.StatusNamed = sn != 0
@@ -1133,6 +1149,12 @@ func (r *Registry) Delete(host string) error {
// purpose: an existing row keeps them whatever the incoming Route carries.
// They have one writer each (SetGoodBuild, SetServingVersion), and no
// full-row caller knows what an instance last served.
+// `incarnation` and `serving_seq` (self-serve moves plan 1 Task 7) are
+// named in the clause only to say so: an existing row keeps its pair. A
+// fresh INSERT writes the incoming pair (the mint's, provision.RouteFor);
+// after that only SetHoldIfSeq, ReleaseHoldSeq and InstallLifetimeIf move
+// it, each with the status it belongs with, so a stale full-row write can
+// never put a row back into an older lifetime.
// opsAccount is the ops account's sqid: account #1, pinned by migrate. It is
// release.OpsAccount (release cannot be imported here; a test in release
// pins the two together through EffectiveAccount).
@@ -1208,7 +1230,8 @@ ON CONFLICT(host) DO UPDATE SET
WHEN excluded.update_settings = '' THEN routes.update_settings ELSE excluded.update_settings END,
build_state=CASE WHEN `+tenantChangedSQL+` THEN excluded.build_state
WHEN excluded.build_state = '' THEN routes.build_state ELSE excluded.build_state END,
- seed_source=CASE WHEN excluded.seed_source = '' THEN routes.seed_source ELSE excluded.seed_source END
+ seed_source=CASE WHEN excluded.seed_source = '' THEN routes.seed_source ELSE excluded.seed_source END,
+ incarnation=routes.incarnation, serving_seq=routes.serving_seq
WHERE routes.claim_incarnation=excluded.claim_incarnation AND routes.claim_stamp=excluded.claim_stamp
AND routes.claim_owner=excluded.claim_owner AND routes.claim_row_id=excluded.claim_row_id`,
routeArgs(normalizeRoute(t))...)
@@ -1269,7 +1292,7 @@ func routeArgs(t Route) []any {
t.Steering, t.Gate, t.CreatedAt, t.Args, t.IdleSeconds, t.MaxRunSeconds,
t.Replicas, b2iDB(t.Replica), t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID, t.HoldOp,
t.GoodVersion, t.GoodDigest, t.GoodDemand, t.ServingVersion, t.UpdateSettings, t.BuildState,
- t.SeedSource, t.ImportGate,
+ t.SeedSource, t.ImportGate, t.Incarnation, t.ServingSeq,
}
}
@@ -1465,6 +1488,92 @@ func (r *Registry) DeleteIfHoldWithTimers(host, op string) (bool, error) {
return true, nil
}
+// Pair is a row's lifetime pair (Route.Incarnation, Route.ServingSeq):
+// which lifetime of the hostname the row serves and the move_seq of the
+// transition that produced its state. Edges order publications of one
+// hostname by it (design §6.4).
+type Pair struct {
+ Incarnation, ServingSeq uint64
+}
+
+// SetHoldIfSeq is SetHoldIf that writes the hold's lifetime pair in the same
+// statement as the status and the op (self-serve moves plan 1 Task 7). The
+// validation is SetHoldIf's, verbatim. The row is held only while its
+// hold_op is still expectOp and its incarnation is pair's or older: a hold of
+// a lifetime takes an unheld row of the same lifetime, a legacy row at 0, or
+// a predecessor's row left by a recreation, with the status, op and pair
+// together, so no ready row ever carries a hold's pair; a row of a newer
+// lifetime changes nothing. false with no error means nothing matched.
+//
+// As with SetHold, every caller touches the stage-requested marker after a
+// write that landed.
+func (r *Registry) SetHoldIfSeq(host, status, op, expectOp string, pair Pair) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "SetHoldIfSeq", Host: host, Status: status, Op: op, ExpectOp: expectOp, Pair: pair}, nil)
+ return out.Changed, err
+ }
+
+ if !IsHold(status) || op == "" {
+ return false, fmt.Errorf("registry: %q with op %q is not a hold", status, op)
+ }
+ if expectOp != "" && expectOp != op {
+ return false, fmt.Errorf("registry: hold %s by %q expecting %q: another op's hold is released by that op, not taken over", host, op, expectOp)
+ }
+ res, err := r.sql.Exec(`UPDATE routes SET status = ?, hold_op = ?, incarnation = ?, serving_seq = ? WHERE host = ? AND hold_op = ? AND incarnation <= ?`,
+ status, op, pair.Incarnation, pair.ServingSeq, host, expectOp, pair.Incarnation)
+ if err != nil {
+ return false, err
+ }
+ n, err := res.RowsAffected()
+ return n > 0, err
+}
+
+// ReleaseHoldSeq is ReleaseHold that writes the releasing transition's pair
+// with the ready status (self-serve moves plan 1 Task 7): an empty op is
+// ReleaseHold's no-op, and the row leaves the hold only under the op that
+// holds it and only while its incarnation is pair's or 0 (a legacy row held
+// before its record took a lifetime). A release names its lifetime, so a
+// stale release of a predecessor never moves a successor's row. false means
+// nothing matched.
+func (r *Registry) ReleaseHoldSeq(host, op string, pair Pair) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "ReleaseHoldSeq", Host: host, Op: op, Pair: pair}, nil)
+ return out.Changed, err
+ }
+
+ if op == "" {
+ return false, nil
+ }
+ res, err := r.sql.Exec(`UPDATE routes SET status = 'ready', hold_op = '', incarnation = ?, serving_seq = ? WHERE host = ? AND hold_op = ? AND (incarnation = ? OR incarnation = 0)`,
+ pair.Incarnation, pair.ServingSeq, host, op, pair.Incarnation)
+ if err != nil {
+ return false, err
+ }
+ n, err := res.RowsAffected()
+ return n > 0, err
+}
+
+// InstallLifetimeIf moves an unheld row from the pair expect to next in one
+// statement (self-serve moves plan 1 Task 7): reconcile's migration (a legacy
+// row takes its record's lifetime, keeping its own sequence) and recreation
+// (a predecessor's row takes the successor's lifetime and sequence). The
+// row changes only while it is still at expect and unheld; a held row is its
+// hold's. false with no error means nothing matched.
+func (r *Registry) InstallLifetimeIf(host string, expect, next Pair) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "InstallLifetimeIf", Host: host, ExpectPair: expect, Pair: next}, nil)
+ return out.Changed, err
+ }
+
+ res, err := r.sql.Exec(`UPDATE routes SET incarnation = ?, serving_seq = ? WHERE host = ? AND incarnation = ? AND serving_seq = ? AND hold_op = ''`,
+ next.Incarnation, next.ServingSeq, host, expect.Incarnation, expect.ServingSeq)
+ 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.
@@ -3069,7 +3178,7 @@ func (r *Registry) Fingerprint() (string, error) {
rows, err := db.Query(`
SELECT rowid, status, socket, addr, version, hibernate, channel, backing, tunnel,
db_path, region, fleet_box, addrs, health, gate, hold_op, import_gate,
- claim_incarnation, claim_stamp, claim_owner, claim_row_id
+ claim_incarnation, claim_stamp, claim_owner, claim_row_id, incarnation, serving_seq
FROM routes ORDER BY rowid`)
if err != nil {
return "", err
@@ -3079,11 +3188,12 @@ FROM routes ORDER BY rowid`)
n := 0
for rows.Next() {
var rowid, hibernate int64
+ var pair Pair
var claim ClaimOwnership
var status, socket, addr, version, channel, backing, tunnel, dbPath, region, fleetBox, addrs, health, gate, holdOp, importGate string
if err := rows.Scan(&rowid, &status, &socket, &addr, &version, &hibernate,
&channel, &backing, &tunnel, &dbPath, &region, &fleetBox, &addrs, &health, &gate, &holdOp, &importGate,
- &claim.ClaimIncarnation, &claim.ClaimStamp, &claim.ClaimOwner, &claim.ClaimRowID); err != nil {
+ &claim.ClaimIncarnation, &claim.ClaimStamp, &claim.ClaimOwner, &claim.ClaimRowID, &pair.Incarnation, &pair.ServingSeq); err != nil {
return "", err
}
fmt.Fprintf(h, "%d\x1f%s\x1f%s\x1f%s\x1f%s\x1f%d\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s\x1f%s",
@@ -3117,6 +3227,15 @@ FROM routes ORDER BY rowid`)
if claim.ClaimRowID != "" {
fmt.Fprintf(h, "\x1fclaim_row_id=%q", claim.ClaimRowID)
}
+ // The lifetime pair follows the same rule: (0, 0) is omitted, so a
+ // registry with no pair digests as it did before the columns
+ // existed, and the edge reloads when a row's pair moves.
+ if pair.Incarnation != 0 {
+ fmt.Fprintf(h, "\x1fincarnation=%d", pair.Incarnation)
+ }
+ if pair.ServingSeq != 0 {
+ fmt.Fprintf(h, "\x1fserving_seq=%d", pair.ServingSeq)
+ }
fmt.Fprint(h, "\x1e")
n++
}
@@ -3311,6 +3430,14 @@ var snapshotWhenSetCols = map[string]bool{
"update_settings": true,
}
+// snapshotPairCols are the lifetime pair's columns (self-serve moves plan 1
+// Task 7), restore state hashed only when non-zero, so a registry whose rows
+// are all at (0, 0) digests as it did before the columns existed.
+var snapshotPairCols = map[string]bool{
+ "incarnation": true,
+ "serving_seq": true,
+}
+
func (r *Registry) SnapshotFingerprint() (string, error) {
if r.remote != nil {
out, err := r.remote.call(nil, remoteRequest{Method: "SnapshotFingerprint"}, nil)
@@ -3388,6 +3515,13 @@ func (r *Registry) SnapshotFingerprint() (string, error) {
}
continue
}
+ // The lifetime pair: hashed only when non-zero, tagged.
+ if snapshotPairCols[cols[i]] {
+ if v.Valid && v.String != "" && v.String != "0" {
+ fmt.Fprintf(h, "%s=%q\x1f", cols[i], v.String)
+ }
+ continue
+ }
// NULL and "" must not hash alike: a column that has never been
// written is a different restore than one explicitly cleared.
if v.Valid {
diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go
index ab3b7b1fd..2a394de86 100644
--- a/internal/registry/registry_test.go
+++ b/internal/registry/registry_test.go
@@ -3,6 +3,7 @@ package registry
import (
"database/sql"
"errors"
+ "fmt"
"os"
"path/filepath"
"strings"
@@ -87,6 +88,7 @@ func TestUpsertRoundTripsAllColumns(t *testing.T) {
BuildState: `{"known":true,"state":{"schema":1,"last_executed":"bs345"}}`,
SeedSource: `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"ready"}`,
ImportGate: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+ Incarnation: 31, ServingSeq: 32,
}
if err := reg.Upsert(want); err != nil {
t.Fatal(err)
@@ -2949,3 +2951,324 @@ func TestGoodBuildMovesTheSnapshotFingerprint(t *testing.T) {
t.Fatal("a new good build did not move the snapshot fingerprint")
}
}
+
+// The lifetime pair (self-serve moves plan 1 Task 7): every row carries
+// (incarnation, serving_seq), written by a fresh insert and by the pair
+// setters, never by an Upsert over an existing row.
+
+const pairOp, pairOther = "mv_abcdefghijklmnopqrstuvwxyz", "mv_other0000000000000000000000"
+
+// pairRow inserts host at the pair, held by op when op is set.
+func pairRow(t *testing.T, r *Registry, host string, p Pair, op string) {
+ t.Helper()
+ rt := Route{Host: host, Kind: "instance", Socket: "/s", App: "shop", Account: "bab", Incarnation: p.Incarnation, ServingSeq: p.ServingSeq}
+ if op != "" {
+ rt.Status, rt.HoldOp = "moving", op
+ }
+ if err := r.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func pairOf(t *testing.T, r *Registry, host string) (Pair, Route) {
+ t.Helper()
+ rt, err := r.Get(host)
+ if err != nil || rt == nil {
+ t.Fatalf("Get(%s) = %+v, %v", host, rt, err)
+ }
+ return Pair{Incarnation: rt.Incarnation, ServingSeq: rt.ServingSeq}, *rt
+}
+
+// writeLog arms a trigger that logs every insert and update of a routes row,
+// so a test can count the writes a setter made and see each row state.
+func writeLog(t *testing.T, r *Registry) func() []string {
+ t.Helper()
+ for _, stmt := range []string{
+ `CREATE TABLE route_writes (n INTEGER PRIMARY KEY AUTOINCREMENT, host TEXT, status TEXT, hold_op TEXT, incarnation INTEGER, serving_seq INTEGER)`,
+ `CREATE TRIGGER route_writes_u AFTER UPDATE ON routes BEGIN INSERT INTO route_writes (host, status, hold_op, incarnation, serving_seq) VALUES (new.host, new.status, new.hold_op, new.incarnation, new.serving_seq); END`,
+ `CREATE TRIGGER route_writes_i AFTER INSERT ON routes BEGIN INSERT INTO route_writes (host, status, hold_op, incarnation, serving_seq) VALUES (new.host, new.status, new.hold_op, new.incarnation, new.serving_seq); END`,
+ } {
+ if _, err := r.sql.Exec(stmt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return func() []string {
+ rows, err := r.sql.Query(`SELECT host, status, hold_op, incarnation, serving_seq FROM route_writes ORDER BY n`)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer rows.Close()
+ var out []string
+ for rows.Next() {
+ var host, status, op string
+ var inc, seq int64
+ if err := rows.Scan(&host, &status, &op, &inc, &seq); err != nil {
+ t.Fatal(err)
+ }
+ out = append(out, fmt.Sprintf("%s %s %q (%d, %d)", host, status, op, inc, seq))
+ }
+ if _, err := r.sql.Exec(`DELETE FROM route_writes`); err != nil {
+ t.Fatal(err)
+ }
+ return out
+ }
+}
+
+func TestRegistryPairColumns(t *testing.T) {
+ r := openTestRegistry(t)
+ pairRow(t, r, "h", Pair{2, 3}, "")
+ if p, _ := pairOf(t, r, "h"); p != (Pair{2, 3}) {
+ t.Fatalf("a fresh insert wrote %+v, want (2, 3)", p)
+ }
+ // An Upsert over the row keeps the stored pair, whatever it carries.
+ if err := r.Upsert(Route{Host: "h", Kind: "instance", Socket: "/s2", App: "shop", Incarnation: 9, ServingSeq: 9}); err != nil {
+ t.Fatal(err)
+ }
+ if p, rt := pairOf(t, r, "h"); p != (Pair{2, 3}) || rt.Socket != "/s2" {
+ t.Fatalf("an Upsert over the row: pair %+v socket %q, want (2, 3) and the new socket", p, rt.Socket)
+ }
+
+ // Both fingerprints see every pair change, and a row at (0, 0) digests
+ // exactly as it did before the columns existed: the default is left
+ // out (TestFingerprintsUnchangedForUnheldRows holds the goldens).
+ fp := func() (string, string) {
+ f, err := r.Fingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ s, err := r.SnapshotFingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return f, s
+ }
+ if ok, err := r.InstallLifetimeIf("h", Pair{2, 3}, Pair{}); err != nil || !ok {
+ t.Fatalf("install (0, 0): %v %v", ok, err)
+ }
+ f0, s0 := fp()
+ f1, s1 := f0, s0
+ for _, next := range []Pair{{0, 4}, {3, 4}, {3, 0}} {
+ if ok, err := r.InstallLifetimeIf("h", mustPair(t, r, "h"), next); err != nil || !ok {
+ t.Fatalf("install %+v: %v %v", next, ok, err)
+ }
+ f2, s2 := fp()
+ if f2 == f1 || s2 == s1 {
+ t.Fatalf("the pair moved to %+v and a fingerprint did not (%v %v)", next, f2 == f1, s2 == s1)
+ }
+ f1, s1 = f2, s2
+ }
+ if ok, err := r.InstallLifetimeIf("h", Pair{3, 0}, Pair{}); err != nil || !ok {
+ t.Fatalf("install back to (0, 0): %v %v", ok, err)
+ }
+ if f, s := fp(); f != f0 || s != s0 {
+ t.Fatal("a row back at (0, 0) digests differently")
+ }
+}
+
+func mustPair(t *testing.T, r *Registry, host string) Pair {
+ t.Helper()
+ p, _ := pairOf(t, r, host)
+ return p
+}
+
+// A registry written by today's schema (no pair columns) opens with every
+// row at (0, 0), and opening it again changes nothing.
+func TestPairMigrationIdempotentOnLegacyDB(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", Status: "moving", HoldOp: pairOp}); err != nil {
+ t.Fatal(err)
+ }
+ for _, c := range []string{"incarnation", "serving_seq"} {
+ if _, err := r.sql.Exec(`ALTER TABLE routes DROP COLUMN ` + c); err != nil {
+ t.Fatal(err)
+ }
+ }
+ r.Close()
+ // The first open adds the columns at (0, 0) and the setters work on
+ // the migrated row; the second open finds them and keeps what the
+ // first left.
+ for i, want := range []Pair{{0, 0}, {1, 3}} {
+ 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.Incarnation != want.Incarnation || rt.ServingSeq != want.ServingSeq || rt.HoldOp != pairOp || rt.Status != "moving" {
+ t.Fatalf("open %d: %+v %v, want held at %+v", i, rt, err, want)
+ }
+ if ok, err := r.ReleaseHoldSeq("h", pairOp, Pair{1, 2}); err != nil || !ok {
+ t.Fatalf("open %d: release: %v %v", i, ok, err)
+ }
+ if ok, err := r.SetHoldIfSeq("h", "moving", pairOp, "", Pair{1, 3}); err != nil || !ok {
+ t.Fatalf("open %d: hold again: %v %v", i, ok, err)
+ }
+ r.Close()
+ }
+}
+
+// A hold of a lifetime takes an unheld row of the same or an older lifetime
+// with the status, the op and the pair in one write, and leaves a newer
+// lifetime's row and a row held by another op alone.
+func TestSetHoldIfSeqInstallsOnOlderUnheldRows(t *testing.T) {
+ r := openTestRegistry(t)
+ pairRow(t, r, "h0", Pair{0, 7}, "")
+ pairRow(t, r, "h1", Pair{1, 9}, "")
+ pairRow(t, r, "h2", Pair{2, 0}, "")
+ pairRow(t, r, "h3", Pair{3, 0}, "")
+ pairRow(t, r, "held", Pair{1, 1}, pairOther)
+ log := writeLog(t, r)
+ hold := Pair{2, 1}
+ for _, host := range []string{"h0", "h1", "h2"} {
+ ok, err := r.SetHoldIfSeq(host, "moving", pairOp, "", hold)
+ if err != nil || !ok {
+ t.Fatalf("%s: %v %v", host, ok, err)
+ }
+ p, rt := pairOf(t, r, host)
+ if p != hold || rt.Status != "moving" || rt.HoldOp != pairOp {
+ t.Fatalf("%s: %+v at %+v", host, rt, p)
+ }
+ if w := log(); len(w) != 1 || w[0] != fmt.Sprintf("%s moving %q (2, 1)", host, pairOp) {
+ t.Fatalf("%s: writes %v, want one held write at (2, 1)", host, w)
+ }
+ }
+ for _, c := range []struct {
+ host string
+ want Pair
+ op string
+ }{{"h3", Pair{3, 0}, ""}, {"held", Pair{1, 1}, pairOther}} {
+ ok, err := r.SetHoldIfSeq(c.host, "moving", pairOp, "", hold)
+ if err != nil || ok {
+ t.Fatalf("%s: %v %v, want no match", c.host, ok, err)
+ }
+ if p, rt := pairOf(t, r, c.host); p != c.want || rt.HoldOp != c.op {
+ t.Fatalf("%s changed: %+v at %+v", c.host, rt, p)
+ }
+ }
+ if w := log(); len(w) != 0 {
+ t.Fatalf("refused holds wrote %v", w)
+ }
+ // An op changes the status of its own hold, with its pair.
+ if ok, err := r.SetHoldIfSeq("h2", "landing", pairOp, pairOp, Pair{2, 2}); err != nil || !ok {
+ t.Fatalf("own hold: %v %v", ok, err)
+ }
+ if p, rt := pairOf(t, r, "h2"); p != (Pair{2, 2}) || rt.Status != "landing" {
+ t.Fatalf("own hold: %+v at %+v", rt, p)
+ }
+ // SetHoldIf's validation, verbatim.
+ for _, c := range []struct{ status, op, expect string }{
+ {"ready", pairOp, ""}, {"moving", "", ""}, {"moving", pairOp, pairOther},
+ } {
+ if ok, err := r.SetHoldIfSeq("h0", c.status, c.op, c.expect, hold); err == nil || ok {
+ t.Fatalf("SetHoldIfSeq(%q, %q, %q) = %v %v, want SetHoldIf's refusal", c.status, c.op, c.expect, ok, err)
+ }
+ _, want := r.SetHoldIf("h0", c.status, c.op, c.expect)
+ _, got := r.SetHoldIfSeq("h0", c.status, c.op, c.expect, hold)
+ if want == nil || got == nil || want.Error() != got.Error() {
+ t.Fatalf("refusal %v, want SetHoldIf's %v", got, want)
+ }
+ }
+ if ok, err := r.SetHoldIfSeq("absent", "moving", pairOp, "", hold); err != nil || ok {
+ t.Fatalf("no row: %v %v", ok, err)
+ }
+}
+
+// A release names its lifetime: it opens a row of that lifetime or a legacy
+// row at 0, and never a successor's.
+func TestReleaseHoldSeqNamesItsLifetime(t *testing.T) {
+ r := openTestRegistry(t)
+ pairRow(t, r, "same", Pair{1, 1}, pairOp)
+ pairRow(t, r, "legacy", Pair{0, 1}, pairOp)
+ pairRow(t, r, "successor", Pair{2, 1}, pairOp)
+ pairRow(t, r, "other", Pair{1, 1}, pairOther)
+ log := writeLog(t, r)
+ rel := Pair{1, 2}
+ for _, host := range []string{"same", "legacy"} {
+ ok, err := r.ReleaseHoldSeq(host, pairOp, rel)
+ if err != nil || !ok {
+ t.Fatalf("%s: %v %v", host, ok, err)
+ }
+ if p, rt := pairOf(t, r, host); p != rel || rt.Status != "ready" || rt.HoldOp != "" {
+ t.Fatalf("%s: %+v at %+v", host, rt, p)
+ }
+ if w := log(); len(w) != 1 || w[0] != host+` ready "" (1, 2)` {
+ t.Fatalf("%s: writes %v, want one ready write at (1, 2)", host, w)
+ }
+ }
+ for _, c := range []struct {
+ host, op string
+ want Pair
+ }{{"successor", pairOp, Pair{2, 1}}, {"other", pairOther, Pair{1, 1}}} {
+ ok, err := r.ReleaseHoldSeq(c.host, pairOp, rel)
+ if err != nil || ok {
+ t.Fatalf("%s: %v %v, want no match", c.host, ok, err)
+ }
+ if p, rt := pairOf(t, r, c.host); p != c.want || rt.HoldOp != c.op || rt.Status != "moving" {
+ t.Fatalf("%s changed: %+v at %+v", c.host, rt, p)
+ }
+ }
+ if ok, err := r.ReleaseHoldSeq("successor", "", rel); err != nil || ok {
+ t.Fatalf("empty op: %v %v", ok, err)
+ }
+ if w := log(); len(w) != 0 {
+ t.Fatalf("refused releases wrote %v", w)
+ }
+}
+
+// InstallLifetimeIf moves an unheld row from one pair to the next only while
+// the row is still at the expected pair.
+func TestInstallLifetimeIfIsGuarded(t *testing.T) {
+ r := openTestRegistry(t)
+ pairRow(t, r, "h", Pair{0, 5}, "")
+ pairRow(t, r, "held", Pair{0, 5}, pairOp)
+ if ok, err := r.InstallLifetimeIf("h", Pair{0, 5}, Pair{1, 5}); err != nil || !ok {
+ t.Fatalf("install: %v %v", ok, err)
+ }
+ if p, rt := pairOf(t, r, "h"); p != (Pair{1, 5}) || rt.Status != "ready" || rt.HoldOp != "" {
+ t.Fatalf("install: %+v at %+v", rt, p)
+ }
+ for _, c := range []struct {
+ host string
+ expect, next Pair
+ }{
+ {"h", Pair{0, 5}, Pair{1, 5}}, // the row moved on: the old expectation misses
+ {"h", Pair{1, 4}, Pair{2, 0}}, // the sequence is part of the expectation
+ {"held", Pair{0, 5}, Pair{1, 5}}, // a held row is the hold's
+ {"absent", Pair{}, Pair{1, 0}},
+ } {
+ if ok, err := r.InstallLifetimeIf(c.host, c.expect, c.next); err != nil || ok {
+ t.Fatalf("%s %+v -> %+v: %v %v, want no match", c.host, c.expect, c.next, ok, err)
+ }
+ }
+ if p, _ := pairOf(t, r, "h"); p != (Pair{1, 5}) {
+ t.Fatalf("h at %+v after refused installs", p)
+ }
+ if p, rt := pairOf(t, r, "held"); p != (Pair{0, 5}) || rt.HoldOp != pairOp {
+ t.Fatalf("held row changed: %+v at %+v", rt, p)
+ }
+}
+
+// The three pair setters go through the service-user child like every other
+// registry write.
+func TestRemotePairSetters(t *testing.T) {
+ r, _ := remoteTestRegistry(t)
+ pairRow(t, r, "h", Pair{0, 0}, "")
+ if ok, err := r.SetHoldIfSeq("h", "moving", pairOp, "", Pair{1, 1}); err != nil || !ok {
+ t.Fatalf("hold: %v %v", ok, err)
+ }
+ if ok, err := r.ReleaseHoldSeq("h", pairOp, Pair{1, 2}); err != nil || !ok {
+ t.Fatalf("release: %v %v", ok, err)
+ }
+ if ok, err := r.InstallLifetimeIf("h", Pair{1, 2}, Pair{2, 0}); err != nil || !ok {
+ t.Fatalf("install: %v %v", ok, err)
+ }
+ if _, err := r.SetHoldIfSeq("h", "ready", pairOp, "", Pair{}); err == nil {
+ t.Fatal("a refusal did not cross the transport")
+ }
+ if p, rt := pairOf(t, r, "h"); p != (Pair{2, 0}) || rt.Status != "ready" {
+ t.Fatalf("after the remote setters: %+v at %+v", rt, p)
+ }
+}
diff --git a/internal/registry/remote_dispatch.go b/internal/registry/remote_dispatch.go
index 23adccb5e..a4e78a7a5 100644
--- a/internal/registry/remote_dispatch.go
+++ b/internal/registry/remote_dispatch.go
@@ -38,6 +38,8 @@ type remoteRequest struct {
TValue Route
Op string
ExpectOp string
+ Pair Pair
+ ExpectPair Pair
ToOp string
FromOp string
Account string
@@ -211,6 +213,12 @@ func (r *Registry) dispatchRemote(ctx context.Context, q remoteRequest, callback
out.Changed, err = r.HandHold(q.Host, q.Status, q.ToOp, q.FromOp)
case "ReleaseHold":
out.Changed, err = r.ReleaseHold(q.Host, q.Op)
+ case "SetHoldIfSeq":
+ out.Changed, err = r.SetHoldIfSeq(q.Host, q.Status, q.Op, q.ExpectOp, q.Pair)
+ case "ReleaseHoldSeq":
+ out.Changed, err = r.ReleaseHoldSeq(q.Host, q.Op, q.Pair)
+ case "InstallLifetimeIf":
+ out.Changed, err = r.InstallLifetimeIf(q.Host, q.ExpectPair, q.Pair)
case "DeleteIfHold":
out.Changed, err = r.DeleteIfHold(q.Host, q.Op)
case "DeleteIfHoldWithTimers":