| 1 | //go:build unix |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "context" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strconv" |
| 13 | "strings" |
| 14 | "testing" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/hostbox" |
| 18 | "github.com/carlosframework/platform/internal/hostclaim" |
| 19 | "github.com/carlosframework/platform/internal/hostguard" |
| 20 | "github.com/carlosframework/platform/internal/hostwrite" |
| 21 | "github.com/carlosframework/platform/internal/park" |
| 22 | "github.com/carlosframework/platform/internal/registry" |
| 23 | "github.com/carlosframework/platform/internal/release" |
| 24 | ) |
| 25 | |
| 26 | const sourceHoldHost = "shop.example" |
| 27 | |
| 28 | // sourceHoldBox is the relocation's source "box" with its installation |
| 29 | // active in the host guard, one cold writer session admitted, a ready row |
| 30 | // owned by that installation, and the reconciler the claim assembly |
| 31 | // builds (the WorkSource selector, park.Local), its hooks set by |
| 32 | // relocationSourceRows under getenv and counted. |
| 33 | type sourceHoldBox struct { |
| 34 | t *testing.T |
| 35 | store release.Store |
| 36 | reg *registry.Registry |
| 37 | guard hostguard.Guard |
| 38 | r hostbox.Reconciler |
| 39 | selected hostbox.RecordSelection |
| 40 | owner registry.ClaimOwnership |
| 41 | asked int // FenceClosed calls |
| 42 | holds int // HoldRow calls |
| 43 | writes int // the activation path's row writes |
| 44 | } |
| 45 | |
| 46 | func newSourceHoldBox(t *testing.T, getenv func(string) string) *sourceHoldBox { |
| 47 | t.Helper() |
| 48 | ctx := t.Context() |
| 49 | b := &sourceHoldBox{t: t, store: release.NewDirStore(t.TempDir())} |
| 50 | dir := t.TempDir() |
| 51 | locks := filepath.Join(dir, "locks") |
| 52 | for _, p := range []string{locks, filepath.Join(dir, "hostguard")} { |
| 53 | if err := os.Mkdir(p, 0770); err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | } |
| 57 | obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost) |
| 58 | if err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | obs, err = hostclaim.Update(ctx, b.store, obs, hostclaim.Adopt(hostclaim.NewHolder{Family: "instance", Key: "apps/acct/app/instances/" + sourceHoldHost + ".json", Account: "acct", App: "app", Host: sourceHoldHost, Region: "ie"}, "box", nil, nil, "adopt", func() string { return "id" })) |
| 62 | if err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | b.guard = hostguard.Guard{DataDir: dir, LockDir: locks, Claims: hostguard.StoreReader{Store: b.store}} |
| 66 | if err := b.guard.Activate(ctx, sourceHoldHost, "box", nil); err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | obs, err = hostclaim.Update(ctx, b.store, obs, hostclaim.AdmitWriter(hostclaim.Admission{Box: "box", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true})) |
| 70 | if err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | captured, err := hostguard.Capture(dir, sourceHoldHost) |
| 74 | if err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | if b.reg, err = registry.Open(filepath.Join(t.TempDir(), "registry.db")); err != nil { |
| 78 | t.Fatal(err) |
| 79 | } |
| 80 | t.Cleanup(func() { b.reg.Close() }) |
| 81 | b.owner = registry.ClaimOwnership{ClaimIncarnation: captured.Incarnation, ClaimStamp: captured.Stamp, ClaimOwner: "owner-1", ClaimRowID: "row-1"} |
| 82 | if err := b.reg.Upsert(registry.Route{Host: sourceHoldHost, Status: "ready", Socket: "/run/shop.sock", ClaimOwnership: b.owner}); err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | now := func() time.Time { return time.Unix(100, 0) } |
| 86 | source := hostbox.WorkSource{Guard: b.guard, Box: "box", Region: "ie", Now: now, |
| 87 | Policy: hostbox.Policy{Held: func(context.Context, string) (string, error) { return "", nil }, Relocation: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil }}.Check, |
| 88 | Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }} |
| 89 | h := obs.Claim.Holder |
| 90 | b.selected = hostbox.RecordSelection{Host: sourceHoldHost, Key: h.Key, ETag: "record-etag", Incarnation: h.Incarnation, Family: h.Family, Layout: "flat", Paths: []string{"shop.example.db"}} |
| 91 | b.r = hostbox.Reconciler{Claims: b.store, Guard: b.guard, Local: park.Local{Guard: b.guard, Claims: b.store, DataDir: dir, Box: "box"}, Writes: hostwrite.Config{Enabled: true, Selector: source}, Box: "box", Region: "ie", Now: now, |
| 92 | CheckRecord: func(context.Context, hostbox.RecordSelection) error { return nil }, |
| 93 | CheckLocal: func(context.Context, hostclaim.Observed, hostbox.RecordSelection) error { return nil }, |
| 94 | CheckHeld: func(context.Context, hostclaim.Observed, hostbox.RecordSelection) error { return nil }} |
| 95 | relocationSourceRows(getenv, &b.r, b.guard, b.reg) |
| 96 | if fence := b.r.FenceClosed; fence != nil { |
| 97 | b.r.FenceClosed = func(ctx context.Context, o hostclaim.Observed) (bool, error) { b.asked++; return fence(ctx, o) } |
| 98 | } |
| 99 | if hold := b.r.HoldRow; hold != nil { |
| 100 | b.r.HoldRow = func(ctx context.Context, host, op string) error { b.holds++; return hold(ctx, host, op) } |
| 101 | } |
| 102 | return b |
| 103 | } |
| 104 | |
| 105 | func (b *sourceHoldBox) update(change func(*hostclaim.Claim)) hostclaim.Observed { |
| 106 | b.t.Helper() |
| 107 | ctx := b.t.Context() |
| 108 | obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost) |
| 109 | if err != nil { |
| 110 | b.t.Fatal(err) |
| 111 | } |
| 112 | obs, err = hostclaim.Update(ctx, b.store, obs, func(c *hostclaim.Claim) error { change(c); return nil }) |
| 113 | if err != nil { |
| 114 | b.t.Fatal(err) |
| 115 | } |
| 116 | return obs |
| 117 | } |
| 118 | |
| 119 | // releasing is a drain from "box" to "dst" in releasing, both |
| 120 | // acknowledged. |
| 121 | func (b *sourceHoldBox) releasing() hostclaim.Observed { |
| 122 | return b.update(func(c *hostclaim.Claim) { |
| 123 | c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseReleasing, Exec: "box", Participants: []string{"box", "dst"}, |
| 124 | Intent: map[string]string{"region": "ie", "reason": "drain", "source_box": "box", "target_box": "dst"}, Journal: map[string]string{"exec_ack.box": "{}", "exec_ack.dst": "{}"}} |
| 125 | }) |
| 126 | } |
| 127 | |
| 128 | // closeFence is root beginning the source's release stop: the durable |
| 129 | // admission fence in the guard record (hostguard.Guard.BeginFence). |
| 130 | func (b *sourceHoldBox) closeFence() { |
| 131 | b.t.Helper() |
| 132 | ctx := b.t.Context() |
| 133 | captured, err := hostguard.Capture(b.guard.DataDir, sourceHoldHost) |
| 134 | if err != nil { |
| 135 | b.t.Fatal(err) |
| 136 | } |
| 137 | obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost) |
| 138 | if err != nil { |
| 139 | b.t.Fatal(err) |
| 140 | } |
| 141 | target := hostguard.FenceTarget{FenceID: "fence-1", Host: sourceHoldHost, Incarnation: captured.Incarnation, Stamp: captured.Stamp, Session: obs.Claim.Holder.Session, |
| 142 | Op: "rel-1", Phase: hostclaim.PhaseReleasing, Box: "box", Enrollment: "enrollment-1", BootID: "boot", ControllerID: "ctl", ClaimETag: obs.ETag, DirectoryRevision: "4"} |
| 143 | if _, err := b.guard.BeginFence(ctx, target); err != nil { |
| 144 | b.t.Fatal(err) |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | func (b *sourceHoldBox) install(dry bool) error { |
| 149 | sel := b.selected |
| 150 | sel.DryRun = dry |
| 151 | return b.r.Install(b.t.Context(), sel, func() error { b.writes++; return nil }) |
| 152 | } |
| 153 | |
| 154 | func (b *sourceHoldBox) row() *registry.Route { |
| 155 | b.t.Helper() |
| 156 | row, err := b.reg.Get(sourceHoldHost) |
| 157 | if err != nil { |
| 158 | b.t.Fatal(err) |
| 159 | } |
| 160 | return row |
| 161 | } |
| 162 | |
| 163 | func (b *sourceHoldBox) guardRecord() []byte { |
| 164 | b.t.Helper() |
| 165 | raw, err := os.ReadFile(filepath.Join(b.guard.DataDir, "hostguard", sourceHoldHost+".json")) |
| 166 | if err != nil { |
| 167 | b.t.Fatal(err) |
| 168 | } |
| 169 | return raw |
| 170 | } |
| 171 | |
| 172 | // retireRow is the source's park-local retiring its row: the typed |
| 173 | // retirement's registry step (hostbox.RegistryRows, the one park-local |
| 174 | // row write), by the row's claim ownership, held or not. |
| 175 | func (b *sourceHoldBox) retireRow() { |
| 176 | b.t.Helper() |
| 177 | rows := hostbox.RegistryRows{Registry: b.reg, Inventory: hostbox.RetirementInventory{Host: sourceHoldHost, Captured: hostguard.Captured{Incarnation: b.owner.ClaimIncarnation, Stamp: b.owner.ClaimStamp}, Ownership: b.owner.ClaimOwner}} |
| 178 | route := hostbox.Artifact{Kind: hostbox.ArtifactRoute, Root: "registry", Name: sourceHoldHost, Owner: b.owner.ClaimOwner, RowID: b.owner.ClaimRowID} |
| 179 | if err := rows.RetireRowsJournal(b.t.Context(), "retire-1", b.owner.ClaimOwner, []hostbox.Artifact{route}, func(json.RawMessage) error { return nil }); err != nil { |
| 180 | b.t.Fatalf("the park-local's row retirement: %v", err) |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | func regionHA(on bool) func(string) string { |
| 185 | return func(k string) string { |
| 186 | if k == "CARLOS_REGION_HA" && on { |
| 187 | return "1" |
| 188 | } |
| 189 | return "" |
| 190 | } |
| 191 | } |
| 192 | |
| 193 | // complete is the target's probe passing and root's stop of the probe |
| 194 | // session, then the real serving CAS (hostclaim.CompleteRelocation), |
| 195 | // which queues the source's park-local unless a fence discharged it. |
| 196 | func (b *sourceHoldBox) complete() hostclaim.Observed { |
| 197 | b.t.Helper() |
| 198 | obs := b.completeOwing() |
| 199 | if !hostclaim.OwesLocal(obs.Claim, "box") { |
| 200 | b.t.Fatalf("after the completion: obligations %+v", obs.Claim.Obligations) |
| 201 | } |
| 202 | return obs |
| 203 | } |
| 204 | |
| 205 | // completeOwing is complete without asking which boxes owe a park-local. |
| 206 | func (b *sourceHoldBox) completeOwing() hostclaim.Observed { |
| 207 | b.t.Helper() |
| 208 | b.update(func(c *hostclaim.Claim) { |
| 209 | h := c.Holder |
| 210 | h.Stamp, h.Session, h.Generation, h.WriterBox = h.Stamp+5, h.Session+1, "P1", "dst" |
| 211 | h.Admissions = append(h.Admissions, hostclaim.AdmissionRecord{Session: h.Session, Box: "dst", Generation: "P1", Predecessor: "g1", Mode: "cold", Region: "ie"}) |
| 212 | probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: h.Session, At: time.Unix(90, 0).UTC()}) |
| 213 | stop, _ := json.Marshal(hostclaim.ProbeStop{Box: "dst", Session: h.Session, Receipt: "probe-fence-1", ReceiptSHA256: strings.Repeat("a", 64)}) |
| 214 | c.Op.Phase, c.Op.Exec = hostclaim.PhaseRelocatingProbing, "dst" |
| 215 | c.Op.Journal["probe"], c.Op.Journal["probe_stop"] = string(probe), string(stop) |
| 216 | }) |
| 217 | ctx := b.t.Context() |
| 218 | obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost) |
| 219 | if err != nil { |
| 220 | b.t.Fatal(err) |
| 221 | } |
| 222 | n := 0 |
| 223 | obs, err = hostclaim.Update(ctx, b.store, obs, hostclaim.CompleteRelocation("rel-1", func() string { n++; return "id-" + strconv.Itoa(n) })) |
| 224 | if err != nil { |
| 225 | b.t.Fatalf("the serving CAS: %v", err) |
| 226 | } |
| 227 | if obs.Claim.Op.Kind != hostclaim.KindServing || obs.Claim.Op.Exec != "dst" { |
| 228 | b.t.Fatalf("after the completion: op %+v obligations %+v", obs.Claim.Op, obs.Claim.Obligations) |
| 229 | } |
| 230 | return obs |
| 231 | } |
| 232 | |
| 233 | // leaves asserts reconcile, dry run or not, refuses with nothing done: |
| 234 | // the row exactly as before, the fence not asked, nothing held or |
| 235 | // written, and the guard record as it was. |
| 236 | func (b *sourceHoldBox) leaves(what string, before registry.Route, record []byte) { |
| 237 | b.t.Helper() |
| 238 | asked, holds, writes := b.asked, b.holds, b.writes |
| 239 | for _, dry := range []bool{true, false} { |
| 240 | if err := b.install(dry); !errors.Is(err, hostbox.ErrChanged) { |
| 241 | b.t.Fatalf("%s (dry run %v): %v", what, dry, err) |
| 242 | } |
| 243 | } |
| 244 | if got := b.row(); got == nil || *got != before || b.asked != asked || b.holds != holds || b.writes != writes || !bytes.Equal(b.guardRecord(), record) { |
| 245 | b.t.Fatalf("%s: row %+v (was %+v) fence asked %d holds %d writes %d", what, got, before, b.asked-asked, b.holds-holds, b.writes-writes) |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | // replays asserts reconcile, dry run or not, answers the catch-up hold's |
| 250 | // replay (L18 finding 3): nil, with the row exactly as before, the fence |
| 251 | // not asked, nothing activated or written, and the guard record as it was. |
| 252 | func (b *sourceHoldBox) replays(what string, before registry.Route, record []byte) { |
| 253 | b.t.Helper() |
| 254 | asked, writes := b.asked, b.writes |
| 255 | for _, dry := range []bool{true, false} { |
| 256 | if err := b.install(dry); err != nil { |
| 257 | b.t.Fatalf("%s (dry run %v): %v", what, dry, err) |
| 258 | } |
| 259 | } |
| 260 | if got := b.row(); got == nil || *got != before || b.asked != asked || b.writes != writes || !bytes.Equal(b.guardRecord(), record) { |
| 261 | b.t.Fatalf("%s: row %+v (was %+v) fence asked %d writes %d", what, got, before, b.asked-asked, b.writes-writes) |
| 262 | } |
| 263 | } |
| 264 | |
| 265 | // Task 10b test 3. The completion: the source's row, held at its fence's |
| 266 | // close, stays held through the real serving CAS. Reconcile never |
| 267 | // releases it: under the serving claim the source owes the completed op's |
| 268 | // park-local, so reconcile's catch-up hold (L18 finding 3) replays the |
| 269 | // hold and writes nothing. The park-local's retirement deletes the held |
| 270 | // row by its ownership: that ends the hold, and nothing is minted after. A failover |
| 271 | // from a fenced source: the source is on lost, so reconcile neither asks |
| 272 | // its fence nor holds its row, whether the row was ready or already held |
| 273 | // when the box was fenced, both while the failover runs and after its |
| 274 | // completion queues the park-local; the row is left exactly as it was, |
| 275 | // and the park-local's retirement removes it. |
| 276 | func TestRelocationSourceHoldEndsWithTheCompletionOrTheParkLocal(t *testing.T) { |
| 277 | t.Parallel() |
| 278 | t.Run("completion", func(t *testing.T) { |
| 279 | b := newSourceHoldBox(t, regionHA(true)) |
| 280 | b.releasing() |
| 281 | b.closeFence() |
| 282 | if err := b.install(false); err != nil { |
| 283 | t.Fatalf("the fenced source's hold: %v", err) |
| 284 | } |
| 285 | held := *b.row() |
| 286 | if held.Status != "relocating" || held.HoldOp != "rel-1" || b.holds != 1 { |
| 287 | t.Fatalf("held row %+v holds %d", held, b.holds) |
| 288 | } |
| 289 | b.complete() |
| 290 | b.replays("the source after the completion", held, b.guardRecord()) |
| 291 | b.retireRow() |
| 292 | if err := b.install(false); !errors.Is(err, hostbox.ErrChanged) || b.row() != nil || b.writes != 0 { |
| 293 | t.Fatalf("after the park-local: %v row %+v writes %d", err, b.row(), b.writes) |
| 294 | } |
| 295 | }) |
| 296 | for _, fencedHeld := range []bool{false, true} { |
| 297 | name := "failover, row ready when fenced" |
| 298 | if fencedHeld { |
| 299 | name = "failover, row already held when fenced" |
| 300 | } |
| 301 | t.Run(name, func(t *testing.T) { |
| 302 | b := newSourceHoldBox(t, regionHA(true)) |
| 303 | b.releasing() |
| 304 | if fencedHeld { |
| 305 | b.closeFence() |
| 306 | if err := b.install(false); err != nil { |
| 307 | t.Fatalf("the fenced source's hold: %v", err) |
| 308 | } |
| 309 | } |
| 310 | before, record := *b.row(), b.guardRecord() |
| 311 | // hostclaim.FenceSource on a source terminated before its |
| 312 | // release is journaled: a failover in pinning, the source off |
| 313 | // the participants and on lost, its fence journaled. |
| 314 | b.update(func(c *hostclaim.Claim) { |
| 315 | c.Op.Phase, c.Op.Exec, c.Op.Participants = hostclaim.PhasePinning, "none", []string{"dst"} |
| 316 | c.Op.Journal["reason"], c.Op.Journal["fence.box"] = hostclaim.ReasonFailover, `{"key":"fences/box.json"}` |
| 317 | c.Lost = append(c.Lost, "box") |
| 318 | }) |
| 319 | b.leaves("the fenced source during the failover", before, record) |
| 320 | // No qualified discharge: the completion queues its park-local. |
| 321 | b.complete() |
| 322 | b.leaves("the fenced source after the failover's completion", before, record) |
| 323 | b.retireRow() |
| 324 | if got := b.row(); got != nil { |
| 325 | t.Fatalf("the park-local left the row: %+v", got) |
| 326 | } |
| 327 | }) |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | // Task 10b test 4. With CARLOS_REGION_HA off, or any value that is not |
| 332 | // true, relocationSourceRows sets neither hook and reconcile is main's: |
| 333 | // the serving executor writes its row through the activation path, and a |
| 334 | // relocating claim is refused by Serving with the row, the claim and the |
| 335 | // guard record unchanged, even once the source's fence has closed. With |
| 336 | // it on, the same box holds the row at the fence's close. |
| 337 | func TestRelocationSourceHoldFollowsTheFlag(t *testing.T) { |
| 338 | t.Parallel() |
| 339 | for _, v := range []string{"", "0", "false", "off", "yes please", "1", "true"} { |
| 340 | on := v == "1" || v == "true" |
| 341 | t.Run("CARLOS_REGION_HA="+v, func(t *testing.T) { |
| 342 | b := newSourceHoldBox(t, func(k string) string { |
| 343 | if k == "CARLOS_REGION_HA" { |
| 344 | return v |
| 345 | } |
| 346 | return "" |
| 347 | }) |
| 348 | if (b.r.FenceClosed != nil) != on || (b.r.HoldRow != nil) != on { |
| 349 | t.Fatalf("hooks set %v %v with the flag %q", b.r.FenceClosed != nil, b.r.HoldRow != nil, v) |
| 350 | } |
| 351 | if err := b.install(false); err != nil || b.writes != 1 { |
| 352 | t.Fatalf("the serving executor: %v writes %d", err, b.writes) |
| 353 | } |
| 354 | b.releasing() |
| 355 | b.closeFence() |
| 356 | before, record := *b.row(), b.guardRecord() |
| 357 | obs, err := hostclaim.Read(t.Context(), b.store, sourceHoldHost) |
| 358 | if err != nil { |
| 359 | t.Fatal(err) |
| 360 | } |
| 361 | if on { |
| 362 | if err := b.install(false); err != nil { |
| 363 | t.Fatalf("region HA on: %v", err) |
| 364 | } |
| 365 | if got := b.row(); got.Status != "relocating" || got.HoldOp != "rel-1" { |
| 366 | t.Fatalf("region HA on: row %+v", got) |
| 367 | } |
| 368 | return |
| 369 | } |
| 370 | for _, dry := range []bool{true, false} { |
| 371 | if err := b.install(dry); !errors.Is(err, hostbox.ErrChanged) { |
| 372 | t.Fatalf("region HA off (dry run %v): %v", dry, err) |
| 373 | } |
| 374 | } |
| 375 | now, err := hostclaim.Read(t.Context(), b.store, sourceHoldHost) |
| 376 | if err != nil { |
| 377 | t.Fatal(err) |
| 378 | } |
| 379 | if got := b.row(); *got != before || b.writes != 1 || now.ETag != obs.ETag || !bytes.Equal(b.guardRecord(), record) { |
| 380 | t.Fatalf("region HA off changed something: row %+v (was %+v) writes %d claim %v", got, before, b.writes, now.ETag != obs.ETag) |
| 381 | } |
| 382 | }) |
| 383 | } |
| 384 | } |
| 385 | |
| 386 | // L18 joint round, finding 3: the source releases and its fence closes, |
| 387 | // its reconcile is delayed, the target completes, and only then does the |
| 388 | // source reconcile. The claim serves on another box, so Task 10b's |
| 389 | // relocating scope no longer reaches it; the catch-up hold does. Through |
| 390 | // the production hooks (relocationSourceRows) and the real reconcile: the |
| 391 | // row ends held for the completed op, never ready, its ownership and |
| 392 | // every other column as they were, the fence not asked, nothing activated |
| 393 | // and the claim and the guard record unchanged; a replay changes nothing; |
| 394 | // the park-local's retirement then ends it and nothing is minted. |
| 395 | // |
| 396 | // The guards: a row another op holds is never taken; with region HA off |
| 397 | // reconcile is main's (refused, the row ready and unchanged); a box that |
| 398 | // was not the relocation's source, or owes a park-local for another op |
| 399 | // only, does nothing; a box on lost is left to its park-local (Task 10b |
| 400 | // decision 3, the failover test above); a completion record that cannot |
| 401 | // be read is its error, never "no completion", with the row unchanged. |
| 402 | func TestRelocationSourceCatchUpHoldAfterTheCompletion(t *testing.T) { |
| 403 | t.Parallel() |
| 404 | t.Run("delayed source reconcile", func(t *testing.T) { |
| 405 | b := newSourceHoldBox(t, regionHA(true)) |
| 406 | minted := *b.row() |
| 407 | b.releasing() |
| 408 | b.closeFence() |
| 409 | done := b.complete() |
| 410 | record := b.guardRecord() |
| 411 | if err := b.install(false); err != nil { |
| 412 | t.Fatalf("the source's first reconcile after the completion: %v", err) |
| 413 | } |
| 414 | got := *b.row() |
| 415 | if got.Status != "relocating" || got.HoldOp != "rel-1" || b.holds != 1 || b.asked != 0 || b.writes != 0 { |
| 416 | t.Fatalf("the source's row %+v holds %d fence asked %d writes %d", got, b.holds, b.asked, b.writes) |
| 417 | } |
| 418 | held := got |
| 419 | held.Status, held.HoldOp = minted.Status, minted.HoldOp |
| 420 | if held != minted { |
| 421 | t.Fatalf("the hold changed more than status and hold_op: %+v (minted %+v)", got, minted) |
| 422 | } |
| 423 | if now, err := hostclaim.Read(t.Context(), b.store, sourceHoldHost); err != nil || now.ETag != done.ETag || !bytes.Equal(b.guardRecord(), record) { |
| 424 | t.Fatalf("the catch-up hold wrote the claim or the guard record: %v", err) |
| 425 | } |
| 426 | b.replays("the replay", got, record) |
| 427 | b.retireRow() |
| 428 | if err := b.install(false); !errors.Is(err, hostbox.ErrChanged) || b.row() != nil || b.writes != 0 { |
| 429 | t.Fatalf("after the park-local: %v row %+v writes %d", err, b.row(), b.writes) |
| 430 | } |
| 431 | }) |
| 432 | t.Run("another op's row is never taken", func(t *testing.T) { |
| 433 | b := newSourceHoldBox(t, regionHA(true)) |
| 434 | b.releasing() |
| 435 | b.closeFence() |
| 436 | if ok, err := b.reg.SetHoldIf(sourceHoldHost, "moving", "mv-9", ""); err != nil || !ok { |
| 437 | t.Fatalf("holding the row for another op: %v %v", ok, err) |
| 438 | } |
| 439 | before := *b.row() |
| 440 | b.complete() |
| 441 | for _, dry := range []bool{true, false} { |
| 442 | if err := b.install(dry); !dry && !errors.Is(err, hostbox.ErrChanged) { |
| 443 | t.Fatalf("another op's row: %v", err) |
| 444 | } |
| 445 | } |
| 446 | if got := b.row(); *got != before || b.writes != 0 { |
| 447 | t.Fatalf("another op's row was taken: %+v (was %+v) writes %d", got, before, b.writes) |
| 448 | } |
| 449 | }) |
| 450 | t.Run("region HA off", func(t *testing.T) { |
| 451 | b := newSourceHoldBox(t, regionHA(false)) |
| 452 | b.releasing() |
| 453 | b.closeFence() |
| 454 | b.complete() |
| 455 | b.leaves("region HA off after the completion", *b.row(), b.guardRecord()) |
| 456 | if got := b.row(); got.Status != "ready" || got.HoldOp != "" { |
| 457 | t.Fatalf("region HA off: row %+v", got) |
| 458 | } |
| 459 | }) |
| 460 | t.Run("not the source", func(t *testing.T) { |
| 461 | b := newSourceHoldBox(t, regionHA(true)) |
| 462 | before, record := *b.row(), b.guardRecord() |
| 463 | // A drain of the host from "other" to "dst": this box holds a |
| 464 | // stale local record and row, but was never a participant. |
| 465 | b.update(func(c *hostclaim.Claim) { |
| 466 | c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseReleasing, Exec: "other", Participants: []string{"other", "dst"}, |
| 467 | Intent: map[string]string{"region": "ie", "reason": "drain", "source_box": "other", "target_box": "dst"}, Journal: map[string]string{"exec_ack.other": "{}", "exec_ack.dst": "{}"}} |
| 468 | }) |
| 469 | done := b.completeOwing() |
| 470 | if hostclaim.OwesLocal(done.Claim, "box") || !hostclaim.OwesLocal(done.Claim, "other") { |
| 471 | t.Fatalf("the completion's park-locals: %+v", done.Claim.Obligations) |
| 472 | } |
| 473 | b.leaves("a box that was not the source", before, record) |
| 474 | }) |
| 475 | t.Run("a park-local for another op only", func(t *testing.T) { |
| 476 | b := newSourceHoldBox(t, regionHA(true)) |
| 477 | b.releasing() |
| 478 | b.closeFence() |
| 479 | b.complete() |
| 480 | b.update(func(c *hostclaim.Claim) { |
| 481 | for i := range c.Obligations { |
| 482 | if c.Obligations[i].Kind == "park-local" && c.Obligations[i].Box == "box" { |
| 483 | c.Obligations[i].Op = "rel-0" |
| 484 | } |
| 485 | } |
| 486 | }) |
| 487 | b.leaves("a park-local for another op", *b.row(), b.guardRecord()) |
| 488 | }) |
| 489 | t.Run("an unreadable completion record", func(t *testing.T) { |
| 490 | b := newSourceHoldBox(t, regionHA(true)) |
| 491 | b.releasing() |
| 492 | b.closeFence() |
| 493 | b.complete() |
| 494 | b.update(func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalCompleted] = "{" }) |
| 495 | before, record := *b.row(), b.guardRecord() |
| 496 | for _, dry := range []bool{true, false} { |
| 497 | if err := b.install(dry); err == nil || errors.Is(err, hostbox.ErrChanged) || !errors.Is(err, hostclaim.ErrRelocation) { |
| 498 | t.Fatalf("an unreadable completion record (dry run %v): %v (want the read's error)", dry, err) |
| 499 | } |
| 500 | } |
| 501 | if got := b.row(); *got != before || b.holds != 0 || b.writes != 0 || !bytes.Equal(b.guardRecord(), record) { |
| 502 | t.Fatalf("an unreadable completion record changed something: row %+v holds %d writes %d", got, b.holds, b.writes) |
| 503 | } |
| 504 | }) |
| 505 | } |
| 506 | |
amadan