| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "net" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "reflect" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/hostbox" |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | "github.com/carlosframework/platform/internal/hostguard" |
| 18 | "github.com/carlosframework/platform/internal/hostwrite" |
| 19 | "github.com/carlosframework/platform/internal/instancectl" |
| 20 | "github.com/carlosframework/platform/internal/provision" |
| 21 | "github.com/carlosframework/platform/internal/registry" |
| 22 | ) |
| 23 | |
| 24 | // Initial installation remains a fixture. Every later ownership transfer uses |
| 25 | // the production serving handover and retains the original source directory. |
| 26 | func servingHandoverRoot(t *testing.T, f *servingRootFixture) ClaimRootHTTP { |
| 27 | t.Helper() |
| 28 | inventory := f.p.root.Journals.Inventory |
| 29 | captured, err := f.f.guard.Capture(f.f.read().Host) |
| 30 | if err != nil { |
| 31 | t.Fatal(err) |
| 32 | } |
| 33 | in, _, err := inventory.Snapshot(t.Context(), f.f.read().Host, captured) |
| 34 | if err != nil { |
| 35 | t.Fatal(err) |
| 36 | } |
| 37 | if f.p.root.Reconcile.Registry == nil { |
| 38 | owner := registry.ClaimOwnership{ClaimIncarnation: in.Incarnation, ClaimStamp: in.InstallationStamp, ClaimOwner: in.Entries[0].Ownership, ClaimRowID: "prior-route"} |
| 39 | reg := restartTestRegistry(t) |
| 40 | holder := f.f.read().Claim.Holder |
| 41 | row := registry.Route{Host: in.Host, Account: holder.Account, App: holder.App, Kind: "instance", DBPath: f.db, Socket: provision.NestedSocket(in.Host), ClaimOwnership: owner} |
| 42 | if err := reg.Upsert(row); err != nil { |
| 43 | t.Fatal(err) |
| 44 | } |
| 45 | in.Entries = append(in.Entries, hostbox.StopInventoryEntry{Kind: "route", ArtifactID: owner.ClaimRowID, ManagedRoot: "registry", RelativePath: in.Host, RowIdentity: owner.ClaimRowID, Ownership: owner.ClaimOwner, Materialization: "materialized"}) |
| 46 | if err := inventory.WriteStopInventory(t.Context(), in); err != nil { |
| 47 | t.Fatal(err) |
| 48 | } |
| 49 | _, revision, err := inventory.Snapshot(t.Context(), in.Host, captured) |
| 50 | if err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | raw, err := f.rootDisk.Read(in.Host + ".json") |
| 54 | if err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | var saved map[string]any |
| 58 | if err = json.Unmarshal(raw, &saved); err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | saved["Installation"].(map[string]any)["InventoryRevision"] = revision |
| 62 | raw, err = json.Marshal(saved) |
| 63 | if err != nil { |
| 64 | t.Fatal(err) |
| 65 | } |
| 66 | if err = f.rootDisk.Write(in.Host+".json", raw); err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | record := provision.Record{Schema: 1, Host: in.Host, Account: holder.Account, App: holder.App, Region: f.actor.Region, CreatedAt: holder.CreatedAt, Channel: "edge"} |
| 70 | body, err := json.Marshal(record) |
| 71 | if err != nil { |
| 72 | t.Fatal(err) |
| 73 | } |
| 74 | if _, err = f.f.claims.Put(t.Context(), holder.Key, body, ""); err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | f.p.root.Reconcile.Registry = reg |
| 78 | f.p.root.Reconcile.Inventory = inventory |
| 79 | f.p.root.Reconcile.Box, f.p.root.Reconcile.Enrollment = f.actor.Box, f.actor.Enrollment |
| 80 | f.p.root.Reconcile.Retirement.JournalDir = filepath.Join(t.TempDir(), "handover") |
| 81 | f.p.root.Launches = f.f.journal |
| 82 | } |
| 83 | f.f.guard.Root = hostguard.NewRootAuthorityWithInventory(f.rootDisk, func(ctx context.Context, host string, c hostguard.Captured) (uint64, error) { |
| 84 | _, rev, err := inventory.Snapshot(ctx, host, c) |
| 85 | return rev, err |
| 86 | }) |
| 87 | f.p.root.Stop.Coordinator.Guard = f.f.guard |
| 88 | f.f.evidence = f.f.claims |
| 89 | f.p.root.Stop.Coordinator.Evidence = f.f.claims |
| 90 | return servingTransportRoot(t, f) |
| 91 | } |
| 92 | |
| 93 | func TestServingParkColdHandoverPreservesState(t *testing.T) { runServingColdHandoverBoundary(t, nil) } |
| 94 | |
| 95 | func TestServingHandoverThenPrelaunchParkReplay(t *testing.T) { |
| 96 | f := newServingRootFixture(t, true) |
| 97 | root := servingHandoverRoot(t, f) |
| 98 | socket := servingRootSocket(t, root) |
| 99 | hook := coldHandoverWith(handoverRecoveryIdentity(t, f.actor), socket, func(context.Context, hostclaim.Observed) (hostguard.FenceReceipt, error) { |
| 100 | return hostguard.FenceReceipt{Target: f.receipt.Target}, nil |
| 101 | }) |
| 102 | admitted := f.f.read() |
| 103 | for i := 0; i < 2; i++ { |
| 104 | if err := hook(t.Context(), f.predecessor, admitted); err != nil { |
| 105 | t.Fatal("handover", err) |
| 106 | } |
| 107 | } |
| 108 | installed, err := f.f.guard.Capture(f.current.Host) |
| 109 | if err != nil || installed != (hostguard.Captured{Incarnation: f.current.Incarnation, Stamp: f.current.Stamp}) { |
| 110 | t.Fatalf("installation: %+v %v", installed, err) |
| 111 | } |
| 112 | target := recoveryTuple(f.f.read(), claimStopLocal{Actor: f.actor, BootID: f.receipt.Target.BootID, ControllerID: "controller"}) |
| 113 | if err := f.f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "edge-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/next", Device: 1, Inode: 99}}}); err != nil { |
| 114 | t.Fatal(err) |
| 115 | } |
| 116 | client := hostbox.ServingParkClient{Call: rootJournalClient(socket)} |
| 117 | a, err := client.Park(t.Context(), f.current, f.prior) |
| 118 | if err != nil { |
| 119 | t.Fatal("park after prelaunch", err) |
| 120 | } |
| 121 | f.tags(f.current.Stamp) |
| 122 | if err := client.Complete(t.Context(), a.ID, f.current); err != nil { |
| 123 | t.Fatal(err) |
| 124 | } |
| 125 | before, err := f.p.ledger.read(a.ID) |
| 126 | if err != nil || !before.Complete { |
| 127 | t.Fatal("completion", err) |
| 128 | } |
| 129 | again, err := client.Park(t.Context(), f.current, f.prior) |
| 130 | if err != nil || again != a { |
| 131 | t.Fatalf("replay after handover and prelaunch: %+v %v", again, err) |
| 132 | } |
| 133 | after, err := f.p.ledger.read(a.ID) |
| 134 | if err != nil || !reflect.DeepEqual(before, after) { |
| 135 | t.Fatal("replay changed record", err) |
| 136 | } |
| 137 | if err := f.p.ledger.fs.CheckParked(&after); err != nil { |
| 138 | t.Fatal(err) |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | func runServingColdHandoverBoundary(t *testing.T, mount func(*testing.T, string), revisions ...uint64) { |
| 143 | t.Helper() |
| 144 | f := newServingRootFixture(t, true) |
| 145 | if mount != nil { |
| 146 | mount(t, f.f.guard.DataDir) |
| 147 | } |
| 148 | root := servingHandoverRoot(t, f) |
| 149 | if len(revisions) != 0 { |
| 150 | f.actor.Revision = revisions[0] |
| 151 | } |
| 152 | stateDir := filepath.Dir(f.db) |
| 153 | expected := map[string]string{filepath.Base(f.db): "served bytes", filepath.Base(f.db) + ".gen": "G", filepath.Base(f.db) + "-wal": "WAL", filepath.Base(f.db) + "-shm": "SHM", "files/upload": "tenant upload", "env": "tenant configuration"} |
| 154 | for _, name := range []string{"files/upload", "env"} { |
| 155 | p := filepath.Join(stateDir, name) |
| 156 | if err := os.MkdirAll(filepath.Dir(p), 0700); err != nil { |
| 157 | t.Fatal(err) |
| 158 | } |
| 159 | if err := os.WriteFile(p, []byte(expected[name]), 0600); err != nil { |
| 160 | t.Fatal(err) |
| 161 | } |
| 162 | } |
| 163 | before := f.f.read() |
| 164 | old, _, err := root.Reconcile.Inventory.Snapshot(t.Context(), before.Host, hostguard.Captured{Incarnation: f.prior.Incarnation, Stamp: f.prior.Stamp}) |
| 165 | if err != nil { |
| 166 | t.Fatal(err) |
| 167 | } |
| 168 | next := hostbox.ColdSuccessor{Host: f.current.Host, Incarnation: f.current.Incarnation, Box: f.actor.Box, Enrollment: f.actor.Enrollment, DirectoryRevision: f.actor.Binding, ETag: before.ETag, Generation: f.current.Generation, Stamp: f.current.Stamp, Session: f.current.Session} |
| 169 | handover := root.servingHandover() |
| 170 | for i := 0; i < 2; i++ { |
| 171 | if err = handover.Run(t.Context(), f.receipt.Target, next); err != nil { |
| 172 | t.Fatal("serving handover", err) |
| 173 | } |
| 174 | } |
| 175 | for name, want := range expected { |
| 176 | body, err := os.ReadFile(filepath.Join(stateDir, name)) |
| 177 | if err != nil || string(body) != want { |
| 178 | t.Fatalf("handover changed %s: %q %v", name, body, err) |
| 179 | } |
| 180 | } |
| 181 | retained, _, err := root.Reconcile.Inventory.Snapshot(t.Context(), before.Host, hostguard.Captured{Incarnation: f.prior.Incarnation, Stamp: f.prior.Stamp}) |
| 182 | if err != nil || !reflect.DeepEqual(old, retained) { |
| 183 | t.Fatal("predecessor inventory changed", err) |
| 184 | } |
| 185 | if f.f.read().ETag != before.ETag { |
| 186 | t.Fatal("handover rewrote claim") |
| 187 | } |
| 188 | a, err := f.park() |
| 189 | if err != nil { |
| 190 | t.Fatal("database retirement after handover", err) |
| 191 | } |
| 192 | f.tags(f.current.Stamp) |
| 193 | if err = f.p.completeServingDatabasePark(t.Context(), f.conn, a.ID, f.current); err != nil { |
| 194 | t.Fatal(err) |
| 195 | } |
| 196 | for _, name := range []string{"files/upload", "env"} { |
| 197 | body, err := os.ReadFile(filepath.Join(stateDir, name)) |
| 198 | if err != nil || string(body) != expected[name] { |
| 199 | t.Fatalf("retirement changed %s: %q %v", name, body, err) |
| 200 | } |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | func TestServingHandoverLegacy(t *testing.T) { |
| 205 | f, _, coordinator := newServingLegacyRootSetup(t, false) |
| 206 | root := servingHandoverRoot(t, f) |
| 207 | boot, err := servingParkBootID() |
| 208 | if err != nil { |
| 209 | t.Fatal(err) |
| 210 | } |
| 211 | f.f.peer.BootID = boot |
| 212 | o := f.f.read() |
| 213 | req := StopRequest{Region: f.actor.Region, Host: o.Host, Incarnation: o.Claim.Holder.Incarnation, InstallationStamp: o.Claim.Holder.Stamp, Box: f.actor.Box, Enrollment: f.actor.Enrollment, BootID: boot, ControllerID: "controller", ExpectedClaimETag: o.ETag, ExpectedDirectoryRevision: "7", RosterRevision: 7} |
| 214 | if _, err := coordinator.FenceAndStop(t.Context(), req); err != nil { |
| 215 | t.Fatal(err) |
| 216 | } |
| 217 | admitted, err := hostclaim.Update(t.Context(), f.f.claims, f.f.read(), hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: 0, Box: f.actor.Box, Region: f.actor.Region, PredecessorRegion: f.actor.Region, Mode: "cold", NewGeneration: "G"})) |
| 218 | if err != nil { |
| 219 | t.Fatal(err) |
| 220 | } |
| 221 | f.current, err = servingWriterIdentity(admitted, f.actor.Box, f.actor.Region) |
| 222 | if err != nil { |
| 223 | t.Fatal(err) |
| 224 | } |
| 225 | saved, err := f.p.ledger.legacyStop(f.current) |
| 226 | if err != nil { |
| 227 | t.Fatal(err) |
| 228 | } |
| 229 | f.actor.Revision = 8 |
| 230 | n := hostbox.ColdSuccessor{Host: f.current.Host, Incarnation: f.current.Incarnation, Box: f.actor.Box, Enrollment: f.actor.Enrollment, DirectoryRevision: f.actor.Binding, ETag: f.f.read().ETag, Generation: f.current.Generation, Stamp: f.current.Stamp, Session: f.current.Session} |
| 231 | if err := root.servingHandover().Run(t.Context(), saved.Result.Target, n); err != nil { |
| 232 | t.Fatalf("legacy transfer: %v prior=%+v next=%+v", err, saved.Result.Target, n) |
| 233 | } |
| 234 | if _, err := f.p.parkLegacyServingDatabase(t.Context(), f.conn, f.current, servingLegacyPredecessor{}); err != nil { |
| 235 | t.Fatal("legacy park after transfer", err) |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | func TestServingHandoverRefusesForeignAuthority(t *testing.T) { |
| 240 | for _, attack := range []string{"census", "actor", "successor", "directory-revision", "operation", "predecessor", "receipt", "root-revision", "registry-path", "registry-account", "source-replaced", "extra-timer", "prelaunch"} { |
| 241 | t.Run(attack, func(t *testing.T) { |
| 242 | f := newServingRootFixture(t, true) |
| 243 | root := servingHandoverRoot(t, f) |
| 244 | p := f.receipt.Target |
| 245 | n := hostbox.ColdSuccessor{Host: f.current.Host, Incarnation: f.current.Incarnation, Box: f.actor.Box, Enrollment: f.actor.Enrollment, DirectoryRevision: "7", ETag: f.f.read().ETag, Generation: f.current.Generation, Stamp: f.current.Stamp, Session: f.current.Session} |
| 246 | row, err := root.Reconcile.Registry.Get(n.Host) |
| 247 | if err != nil { |
| 248 | t.Fatal(err) |
| 249 | } |
| 250 | switch attack { |
| 251 | case "census": |
| 252 | root.RequireNewWork = func(context.Context) error { return hostbox.ErrChanged } |
| 253 | case "actor": |
| 254 | f.actor.Eligible = false |
| 255 | case "successor": |
| 256 | n.Generation = "foreign" |
| 257 | case "directory-revision": |
| 258 | n.DirectoryRevision = "8" |
| 259 | case "operation": |
| 260 | n.Op = "move" |
| 261 | case "predecessor": |
| 262 | p.Session++ |
| 263 | case "receipt": |
| 264 | p.FenceID = "unknown" |
| 265 | case "root-revision": |
| 266 | f.mutateRoot(func(m map[string]any) { m["Installation"].(map[string]any)["InventoryRevision"] = 999 }) |
| 267 | case "registry-path": |
| 268 | row.DBPath = filepath.Join(t.TempDir(), "victim.db") |
| 269 | if err = root.Reconcile.Registry.Upsert(*row); err != nil { |
| 270 | t.Fatal(err) |
| 271 | } |
| 272 | case "registry-account": |
| 273 | row.Account = "bac" |
| 274 | if err = root.Reconcile.Registry.Upsert(*row); err != nil { |
| 275 | t.Fatal(err) |
| 276 | } |
| 277 | case "source-replaced": |
| 278 | dir := filepath.Dir(f.db) |
| 279 | if err = os.Rename(dir, dir+"-original"); err != nil { |
| 280 | t.Fatal(err) |
| 281 | } |
| 282 | if err = os.Mkdir(dir, 0700); err != nil { |
| 283 | t.Fatal(err) |
| 284 | } |
| 285 | case "extra-timer": |
| 286 | if err = root.Reconcile.Registry.UpsertTimer(registry.Timer{Host: n.Host, Name: "foreign", Kind: "cron", Spec: "0 * * * *"}); err != nil { |
| 287 | t.Fatal(err) |
| 288 | } |
| 289 | case "prelaunch": |
| 290 | target := recoveryTuple(f.f.read(), claimStopLocal{Actor: f.actor, BootID: f.receipt.Target.BootID, ControllerID: "controller"}) |
| 291 | if err = f.f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "edge-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/next", Device: 1, Inode: 99}}}); err != nil { |
| 292 | t.Fatal(err) |
| 293 | } |
| 294 | } |
| 295 | before, err := root.Reconcile.Registry.Get(n.Host) |
| 296 | if err != nil { |
| 297 | t.Fatal(err) |
| 298 | } |
| 299 | if err = root.servingHandover().Run(t.Context(), p, n); err == nil { |
| 300 | t.Fatal("accepted", attack) |
| 301 | } |
| 302 | after, err := root.Reconcile.Registry.Get(n.Host) |
| 303 | if err != nil || !reflect.DeepEqual(before, after) { |
| 304 | t.Fatal("refusal changed row", err) |
| 305 | } |
| 306 | source := f.db |
| 307 | if attack == "source-replaced" { |
| 308 | source = filepath.Join(filepath.Dir(f.db)+"-original", filepath.Base(f.db)) |
| 309 | } |
| 310 | body, err := os.ReadFile(source) |
| 311 | if err != nil || string(body) != "served bytes" { |
| 312 | t.Fatal("refusal changed source", err) |
| 313 | } |
| 314 | }) |
| 315 | } |
| 316 | } |
| 317 | |
| 318 | // The stop coordinator writes the receipt used by the next handover. Process |
| 319 | // and authentication observations remain fixtures; the rootrunner covers them. |
| 320 | func TestServingStoppedWriterSecondHandover(t *testing.T) { |
| 321 | f := newServingRootFixture(t, true) |
| 322 | root := servingHandoverRoot(t, f) |
| 323 | f.f.peer = StopPeer{UID: 1001, PID: 77, ProcessStart: "edge-start", BootID: f.receipt.Target.BootID, ControllerID: "controller"} |
| 324 | coordinator := f.f.root(nil) |
| 325 | coordinator.Evidence, coordinator.Inventory = f.f.claims, root.Journals.Inventory |
| 326 | root.Stop.Coordinator = coordinator |
| 327 | root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return f.f.peer, nil } |
| 328 | f.p.root.Stop.Coordinator = coordinator |
| 329 | socket := servingRootSocket(t, root) |
| 330 | transfer := coldHandoverWith(handoverRecoveryIdentity(t, f.actor), socket, func(context.Context, hostclaim.Observed) (hostguard.FenceReceipt, error) { |
| 331 | return f.receipt, nil |
| 332 | }) |
| 333 | if err := transfer(t.Context(), f.predecessor, f.f.read()); err != nil { |
| 334 | t.Fatal(err) |
| 335 | } |
| 336 | target := recoveryTuple(f.f.read(), claimStopLocal{Actor: f.actor, BootID: f.receipt.Target.BootID, ControllerID: "controller"}) |
| 337 | if err := f.f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "edge-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/next", Device: 1, Inode: 99}}}); err != nil { |
| 338 | t.Fatal(err) |
| 339 | } |
| 340 | client := hostbox.ServingParkClient{Call: rootJournalClient(socket)} |
| 341 | first, err := client.Park(t.Context(), f.current, f.prior) |
| 342 | if err != nil { |
| 343 | t.Fatal(err) |
| 344 | } |
| 345 | f.tags(f.current.Stamp) |
| 346 | if err := client.Complete(t.Context(), first.ID, f.current); err != nil { |
| 347 | t.Fatal(err) |
| 348 | } |
| 349 | for suffix, bytes := range map[string]string{"": "second writer bytes", ".gen": f.current.Generation, "-wal": "second WAL", "-shm": "second SHM"} { |
| 350 | if err := os.WriteFile(f.db+suffix, []byte(bytes), 0600); err != nil { |
| 351 | t.Fatal(err) |
| 352 | } |
| 353 | } |
| 354 | r := hostbox.LaunchReservation{ID: "second:main", Role: "main", InvocationID: "second", Inputs: "digest", Target: target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(target.Host)}, Executable: hostbox.LaunchFile{Device: 1, Inode: 3}} |
| 355 | if err := f.f.journal.IntendUnit(t.Context(), r); err != nil { |
| 356 | t.Fatal(err) |
| 357 | } |
| 358 | if err := f.f.journal.StartedUnit(t.Context(), r, hostbox.LaunchGroup{RelativePath: r.Group.RelativePath, Device: 1, Inode: 4, InvocationID: "second-unit"}, 99, "tenant-start"); err != nil { |
| 359 | t.Fatal(err) |
| 360 | } |
| 361 | o := f.f.read() |
| 362 | req := StopRequest{Region: f.actor.Region, Host: o.Host, Incarnation: f.current.Incarnation, InstallationStamp: f.current.Stamp, WriterSession: f.current.Session, Box: f.actor.Box, Enrollment: f.actor.Enrollment, BootID: f.receipt.Target.BootID, ControllerID: "controller", ExpectedClaimETag: o.ETag, ExpectedDirectoryRevision: "7", RosterRevision: 7} |
| 363 | drain := stopDrainFunc(func(_ context.Context, req StopRequest, fence string) (DrainAcknowledgement, error) { |
| 364 | return DrainClaimAcknowledgement(fence, req.ControllerID, "edge-start", nil), nil |
| 365 | }) |
| 366 | if _, err := CollectClaimStop(t.Context(), socket, req, drain); err == nil { |
| 367 | t.Fatal("accepted an unsettled unit") |
| 368 | } |
| 369 | if err := f.f.journal.SettleUnit(t.Context(), target, r.ID); err != nil { |
| 370 | t.Fatal(err) |
| 371 | } |
| 372 | stopped, err := CollectClaimStop(t.Context(), socket, req, drain) |
| 373 | if err != nil { |
| 374 | t.Fatal("stop through transport", err) |
| 375 | } |
| 376 | f.receipt, err = f.f.guard.FenceReceiptFor(t.Context(), stopped.Target) |
| 377 | if err != nil || string(f.receipt.Bytes) != string(stopped.Bytes) { |
| 378 | t.Fatal("durable receipt", err) |
| 379 | } |
| 380 | previous := f.f.read() |
| 381 | h := previous.Claim.Holder |
| 382 | admitted, err := hostclaim.Update(t.Context(), f.f.claims, previous, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: f.actor.Box, Region: f.actor.Region, PredecessorRegion: f.actor.Region, Mode: "cold", NewGeneration: "G3"})) |
| 383 | if err != nil { |
| 384 | t.Fatal(err) |
| 385 | } |
| 386 | prior := f.current |
| 387 | f.current, err = servingWriterIdentity(admitted, f.actor.Box, f.actor.Region) |
| 388 | if err != nil { |
| 389 | t.Fatal(err) |
| 390 | } |
| 391 | if err := transfer(t.Context(), previous, admitted); err != nil { |
| 392 | t.Fatal("second handover", err) |
| 393 | } |
| 394 | parked, err := client.Park(t.Context(), f.current, prior) |
| 395 | if err != nil { |
| 396 | t.Fatal("second park", err) |
| 397 | } |
| 398 | f.tags(f.current.Stamp) |
| 399 | if err := client.Complete(t.Context(), parked.ID, f.current); err != nil { |
| 400 | t.Fatal(err) |
| 401 | } |
| 402 | record, err := f.p.ledger.read(parked.ID) |
| 403 | if err != nil || !record.Complete || record.Disposition != servingParkRetain || len(record.Receipts) != 1 || string(record.Receipts[0].Bytes) != string(stopped.Bytes) { |
| 404 | t.Fatal("second park did not consume exact transport stop receipt", err) |
| 405 | } |
| 406 | old, err := f.p.ledger.read(first.ID) |
| 407 | if err != nil || !old.Closed || old.Cause != "replacement" { |
| 408 | t.Fatal("previous park was not replaced", err) |
| 409 | } |
| 410 | for suffix, want := range map[string]string{"": "second writer bytes", ".gen": prior.Generation, "-wal": "second WAL", "-shm": "second SHM"} { |
| 411 | body, err := os.ReadFile(filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", parked.ID, filepath.Base(f.db)+suffix)) |
| 412 | if err != nil || string(body) != want { |
| 413 | t.Fatal("second park changed bytes", suffix, err) |
| 414 | } |
| 415 | } |
| 416 | if replay, err := client.Park(t.Context(), f.current, prior); err != nil || replay != parked { |
| 417 | t.Fatal("second park replay", err) |
| 418 | } |
| 419 | if after, err := f.p.ledger.read(parked.ID); err != nil || !reflect.DeepEqual(record, after) { |
| 420 | t.Fatal("second park replay changed record", err) |
| 421 | } |
| 422 | servingRestoreAfterStoppedWriter(t, f, client) |
| 423 | if after, err := f.p.ledger.read(parked.ID); err != nil || !reflect.DeepEqual(record, after) { |
| 424 | t.Fatal("restore changed retained park", err) |
| 425 | } |
| 426 | } |
| 427 | |
| 428 | // This connects production ColdRestore and retirement enrollment to the stopped |
| 429 | // writer's park. The exact-version LTX baseline and local policy are fixtures; |
| 430 | // full wake preparation and C8 remain separate acceptance gates. |
| 431 | func servingRestoreAfterStoppedWriter(t *testing.T, f *servingRootFixture, parker servingDatabaseParker) { |
| 432 | t.Helper() |
| 433 | plan := plan12RestoreBaseline(t, f.f) |
| 434 | before, err := f.f.data.GetVersion(t.Context(), f.f.bucket, plan.Files[0].Key, plan.Files[0].VersionID) |
| 435 | if err != nil { |
| 436 | t.Fatal(err) |
| 437 | } |
| 438 | admitted := f.f.read() |
| 439 | in := ClaimLaunchInputs{DBPath: f.db} |
| 440 | r := &claimServingArtifactRetirement{Claims: f.f.claims, Guard: f.f.guard, DataDir: f.f.guard.DataDir, Box: f.actor.Box, Region: f.actor.Region, Root: parker} |
| 441 | source := hostbox.WorkSource{Guard: f.f.guard, Box: f.actor.Box, Region: f.actor.Region, Now: func() time.Time { return f.f.now }, Policy: func(_ context.Context, c *hostclaim.Claim, selected hostbox.Selection, _ hostwrite.Identity) error { |
| 442 | return hostbox.Serving(c, selected) |
| 443 | }, Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }} |
| 444 | work, err := source.Select(t.Context(), admitted.Host, "restore") |
| 445 | if err != nil { |
| 446 | t.Fatal("select restore work", err) |
| 447 | } |
| 448 | if err := r.Recover(t.Context(), admitted, in); err != nil { |
| 449 | t.Fatal("recover completed retirement", err) |
| 450 | } |
| 451 | enrolled := 0 |
| 452 | restore := hostbox.ColdRestore{Reader: f.f.data, Bucket: f.f.bucket, Generation: f.current.Generation, Work: work, |
| 453 | BeforeLegacyInstall: func(ctx context.Context, _, _ string) error { return r.BeforeLegacyInstall(ctx, admitted, in, work) }, |
| 454 | BeforeInstallLocked: func(ctx context.Context, _, _ string) error { |
| 455 | if err := r.EnrollLocked(ctx, admitted, in); err != nil { |
| 456 | return err |
| 457 | } |
| 458 | enrolled++ |
| 459 | return nil |
| 460 | }, |
| 461 | RetireArtifacts: func(ctx context.Context, generation string) error { return r.Retire(ctx, admitted, in, generation) }, |
| 462 | } |
| 463 | if err := restore.Install(t.Context(), f.db, plan); err != nil { |
| 464 | t.Fatal("restore after stopped writer", err) |
| 465 | } |
| 466 | if enrolled != 1 || mwReadDB(t, f.db) != "selected recovery baseline" { |
| 467 | t.Fatal("restore missed enrollment or restored the wrong baseline") |
| 468 | } |
| 469 | if marker, err := hostbox.ReadGenerationMarker(f.db + ".gen"); err != nil || marker != f.current.Generation { |
| 470 | t.Fatal("restored generation", marker, err) |
| 471 | } |
| 472 | after, err := f.f.data.GetVersion(t.Context(), f.f.bucket, plan.Files[0].Key, plan.Files[0].VersionID) |
| 473 | if err != nil || string(after) != string(before) { |
| 474 | t.Fatal("restore changed remote baseline", err) |
| 475 | } |
| 476 | } |
| 477 | |
amadan