amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package host
4
5import (
6 "context"
7 "crypto/sha256"
8 "encoding/hex"
9 "encoding/json"
10 "errors"
11 "fmt"
12 "net"
13 "os"
14 "path/filepath"
15 "reflect"
16 "strings"
17 "sync"
18 "testing"
19 "time"
20
21 "github.com/carlosframework/platform/internal/hostbox"
22 "github.com/carlosframework/platform/internal/hostclaim"
23 "github.com/carlosframework/platform/internal/hostguard"
24 "github.com/carlosframework/platform/internal/hostwrite"
25 "github.com/carlosframework/platform/internal/provision"
26 "golang.org/x/sys/unix"
27)
28
29// Disk persistence and lock exclusion are real. Root ownership, mount policy,
30// roster and authenticated process observations remain explicit test seams.
31type servingAuthorityDisk struct {
32 *rollbackDisk
33 mu sync.Mutex
34 locks map[string]chan struct{}
35 trace *[]string
36 prefix string
37}
38
39func (s *servingAuthorityDisk) Lock(ctx context.Context, name string) (func(), error) {
40 s.mu.Lock()
41 if s.locks == nil {
42 s.locks = make(map[string]chan struct{})
43 }
44 lock := s.locks[name]
45 if lock == nil {
46 lock = make(chan struct{}, 1)
47 s.locks[name] = lock
48 }
49 s.mu.Unlock()
50 select {
51 case lock <- struct{}{}:
52 if s.trace != nil {
53 *s.trace = append(*s.trace, s.prefix+name)
54 }
55 return func() { <-lock }, nil
56 case <-ctx.Done():
57 return nil, ctx.Err()
58 }
59}
60
61type servingRootFixture struct {
62 t *testing.T
63 f *recoveryFixture
64 p *servingParkRoot
65 disk *servingAuthorityDisk
66 rootDisk *servingAuthorityDisk
67 conn net.Conn
68 actor hostbox.Actor
69 current, prior servingArtifactIdentity
70 admitted hostclaim.Observed
71 predecessor hostclaim.Observed
72 receipt hostguard.FenceReceipt
73 db string
74 trace []string
75}
76
77// servingFixtureActor is the serving fixtures' root actor. Its binding is the
78// fixture shorthand "7"; newServingRootFixtureFor takes a production actor.
79var servingFixtureActor = hostbox.Actor{Box: "box-1", Region: "ie", Enrollment: "enrollment", Revision: 7, Binding: "7", Eligible: true}
80
81func newServingRootFixture(t *testing.T, launched bool) *servingRootFixture {
82 t.Helper()
83 return newServingRootFixtureFor(t, launched, servingFixtureActor)
84}
85
86// newServingRootFixtureFor stamps the installation, its stop receipt and the
87// root's inventory actor with actor (box-1 in ie).
88func newServingRootFixtureFor(t *testing.T, launched bool, actor hostbox.Actor) *servingRootFixture {
89 t.Helper()
90 f, _, _, in := servingRetirementFixture(t)
91 first := servingAdmit(t, f, &in, "G")
92 disk := &servingAuthorityDisk{rollbackDisk: &rollbackDisk{dir: t.TempDir()}, prefix: "park:"}
93 roots := &servingAuthorityDisk{rollbackDisk: &rollbackDisk{dir: t.TempDir()}, prefix: "host:"}
94 f.guard.Root = hostguard.NewRootAuthority(roots)
95 journals, err := hostbox.NewJournalService(f.journal)
96 if err != nil {
97 t.Fatal(err)
98 }
99 fx := &servingRootFixture{t: t, f: f, disk: disk, rootDisk: roots, db: in.DBPath, actor: actor}
100 fx.prior, err = servingWriterIdentity(first, "box-1", "ie")
101 if err != nil {
102 t.Fatal(err)
103 }
104 fs := newServingParkFS(f.guard.DataDir, disk)
105 fs.owner = uint32(os.Geteuid())
106 fs.mount = func(string) (bool, error) { return true, nil }
107 fx.p = &servingParkRoot{ledger: &servingParkLedger{state: disk, fs: fs, journal: f.journal, now: func() time.Time { return f.now }}}
108 fx.p.root = ClaimRootHTTP{Roots: map[string]string{"data": f.guard.DataDir}, Journals: journals,
109 Stop: ClaimStopHTTP{Coordinator: ClaimStopCoordinator{Guard: f.guard, Evidence: f.claims}, Authenticate: func(context.Context, net.Conn) (StopPeer, error) {
110 return StopPeer{UID: 1001, PID: 77, ProcessStart: "edge-start", BootID: "boot", ControllerID: "controller"}, nil
111 }},
112 InventoryActor: func(context.Context) (hostbox.Actor, error) { return fx.actor, nil }, RequireNewWork: func(context.Context) error { return nil }}
113 fx.install(first)
114 fx.tags(fx.prior.Stamp)
115 boot, err := fs.currentBoot()
116 if err != nil {
117 t.Fatal(err)
118 }
119 fx.receipt = servingStopFor(t, f, f.read(), fx.actor, boot)
120 target := fx.receipt.Target
121 if err := f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "edge-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/parent", Device: 1, Inode: 2}}}); err != nil {
122 t.Fatal(err)
123 }
124 if launched {
125 res := hostbox.LaunchReservation{ID: "attempt:main", Role: "main", InvocationID: "attempt", Inputs: "digest", Target: target, Group: hostbox.LaunchGroup{RelativePath: "system.slice/carlos-app@shop.service"}, Executable: hostbox.LaunchFile{Device: 1, Inode: 3}}
126 if err := f.journal.IntendUnit(t.Context(), res); err != nil {
127 t.Fatal(err)
128 }
129 if err := f.journal.StartedUnit(t.Context(), res, hostbox.LaunchGroup{RelativePath: res.Group.RelativePath, Device: 1, Inode: 4, InvocationID: "unit"}, 99, "tenant-start"); err != nil {
130 t.Fatal(err)
131 }
132 if err := f.journal.SettleUnit(t.Context(), target, res.ID); err != nil {
133 t.Fatal(err)
134 }
135 }
136 o := f.read()
137 fx.predecessor = o
138 fx.admitted, err = hostclaim.Update(t.Context(), f.claims, o, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation, Box: "box-1", Region: "ie", PredecessorRegion: "ie", Mode: "cold", NewGeneration: "G2"}))
139 if err != nil {
140 t.Fatal(err)
141 }
142 fx.current, err = servingWriterIdentity(fx.admitted, "box-1", "ie")
143 if err != nil {
144 t.Fatal(err)
145 }
146 for suffix, body := range map[string]string{"": "served bytes", ".gen": "G", "-wal": "WAL", "-shm": "SHM"} {
147 if err := os.WriteFile(fx.db+suffix, []byte(body), 0600); err != nil {
148 t.Fatal(err)
149 }
150 }
151 left, right := net.Pipe()
152 fx.conn = left
153 t.Cleanup(func() { left.Close(); right.Close() })
154 disk.trace, roots.trace = &fx.trace, &fx.trace
155 return fx
156}
157func (f *servingRootFixture) install(o hostclaim.Observed) {
158 f.t.Helper()
159 h := o.Claim.Holder
160 d, err := os.Open(filepath.Dir(f.db))
161 if err != nil {
162 f.t.Fatal(err)
163 }
164 id, err := servingParkDirID(d)
165 d.Close()
166 if err != nil {
167 f.t.Fatal(err)
168 }
169 owner := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", o.Host, h.Incarnation, h.Stamp, f.actor.Box)))
170 in := hostbox.StopInventory{Complete: true, Host: o.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, Box: f.actor.Box, Enrollment: f.actor.Enrollment, DirectoryRevision: f.actor.Binding, Entries: []hostbox.StopInventoryEntry{{Kind: "path", ManagedRoot: "data", RelativePath: "instances/" + provision.InstanceID(o.Host), ArtifactID: "state", Ownership: hex.EncodeToString(owner[:]), Materialization: "materialized", Device: id.Device, Inode: id.Inode, Type: unix.S_IFDIR}}}
171 inventory := f.p.root.Journals.Inventory
172 if err := inventory.WriteStopInventory(f.t.Context(), in); err != nil {
173 f.t.Fatal(err)
174 }
175 _, revision, err := inventory.Snapshot(f.t.Context(), o.Host, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp})
176 if err != nil {
177 f.t.Fatal(err)
178 }
179 root := hostguard.RootInstallation{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Migration: 1, InventoryRevision: revision, Account: h.Account, App: h.App, Record: h.Key, Box: f.actor.Box, Enrollment: f.actor.Enrollment}
180 if err := f.f.guard.EstablishRoot(f.t.Context(), root, func(*hostclaim.Claim) error { return nil }); err != nil {
181 f.t.Fatal(err)
182 }
183}
184func (f *servingRootFixture) tags(stamp uint64) {
185 f.t.Helper()
186 _, err := hostclaim.Update(f.t.Context(), f.f.claims, f.f.read(), func(c *hostclaim.Claim) error {
187 rel, _ := filepath.Rel(f.f.guard.DataDir, f.db)
188 c.Inventory = map[string]hostclaim.InventoryEntry{f.actor.Box: {Layout: "nested", Paths: map[string]uint64{rel: stamp, rel + ".gen": stamp}}}
189 return nil
190 })
191 if err != nil {
192 f.t.Fatal(err)
193 }
194}
195func (f *servingRootFixture) park() (servingParkAuthorization, error) {
196 return f.p.parkServingDatabase(f.t.Context(), f.conn, f.current, f.prior)
197}
198func (f *servingRootFixture) mutateRoot(edit func(map[string]any)) {
199 f.t.Helper()
200 name := f.current.Host + ".json"
201 b, err := f.rootDisk.Read(name)
202 if err != nil {
203 f.t.Fatal(err)
204 }
205 var m map[string]any
206 if err = json.Unmarshal(b, &m); err != nil {
207 f.t.Fatal(err)
208 }
209 edit(m)
210 b, err = json.Marshal(m)
211 if err != nil {
212 f.t.Fatal(err)
213 }
214 if err = f.rootDisk.Write(name, b); err != nil {
215 f.t.Fatal(err)
216 }
217}
218func TestServingParkRootGenerationAuthority(t *testing.T) {
219 for _, launched := range []bool{false, true} {
220 t.Run(fmt.Sprint(launched), func(t *testing.T) {
221 f := newServingRootFixture(t, launched)
222 a, err := f.park()
223 if err != nil {
224 t.Fatal(err)
225 }
226 want := []string{"park:namespace.lock", "host:" + f.current.Host + ".lock", "park:slot-" + servingParkKey(f.current.Box, f.current.Host) + ".lock", "park:retirement-" + a.ID + ".lock"}
227 if !reflect.DeepEqual(f.trace, want) {
228 t.Fatalf("lock order: %v", f.trace)
229 }
230 if err = f.p.completeServingDatabasePark(t.Context(), f.conn, a.ID, f.current); err == nil {
231 t.Fatal("completed before retag")
232 }
233 f.tags(f.current.Stamp)
234 if err = f.p.completeServingDatabasePark(t.Context(), f.conn, a.ID, f.current); err != nil {
235 t.Fatal(err)
236 }
237 r, err := f.p.ledger.read(a.ID)
238 if err != nil || !r.Complete {
239 t.Fatalf("record: %+v %v", r, err)
240 }
241 if (r.Disposition == servingParkRetain) != launched {
242 t.Fatalf("wrong disposition: %v", r.Disposition)
243 }
244 again, err := f.p.parkServingDatabase(t.Context(), f.conn, f.current, servingArtifactIdentity{})
245 if err != nil || again != a {
246 t.Fatalf("lost completion reply: %+v %v", again, err)
247 }
248 if err = os.WriteFile(f.db, []byte("replacement"), 0600); err != nil {
249 t.Fatal(err)
250 }
251 if _, err = f.p.parkServingDatabase(t.Context(), f.conn, f.current, servingArtifactIdentity{}); err == nil {
252 t.Fatal("accepted replacement source")
253 }
254 })
255 }
256}
257
258func TestServingParkRootReplaysOriginalSelectorAfterRetag(t *testing.T) {
259 for _, complete := range []bool{false, true} {
260 t.Run(fmt.Sprintf("complete=%v", complete), func(t *testing.T) {
261 f := newServingRootFixture(t, true)
262 a, err := f.park()
263 if err != nil {
264 t.Fatal(err)
265 }
266 f.tags(f.current.Stamp)
267 if complete {
268 if err := f.p.completeServingDatabasePark(t.Context(), f.conn, a.ID, f.current); err != nil {
269 t.Fatal(err)
270 }
271 }
272 before, err := f.p.ledger.read(a.ID)
273 if err != nil {
274 t.Fatal(err)
275 }
276 f.f.now = f.f.now.Add(time.Hour)
277 again, err := f.park()
278 if err != nil || again != a {
279 t.Fatalf("original selector after retag: %+v %v", again, err)
280 }
281 for _, field := range []string{"generation", "stamp", "session", "box", "incarnation"} {
282 bad := f.prior
283 switch field {
284 case "generation":
285 bad.Generation = "foreign"
286 case "stamp":
287 bad.Stamp = f.current.Stamp
288 case "session":
289 bad.Session = f.current.Session
290 case "box":
291 bad.Box = "foreign"
292 case "incarnation":
293 bad.Incarnation = "foreign"
294 }
295 if _, err := f.p.parkServingDatabase(t.Context(), f.conn, f.current, bad); err == nil {
296 t.Fatalf("accepted changed %s", field)
297 }
298 }
299 after, err := f.p.ledger.read(a.ID)
300 if err != nil || !reflect.DeepEqual(before, after) {
301 t.Fatalf("replay changed retirement: %+v %v", after, err)
302 }
303 if err := f.p.ledger.fs.CheckParked(&after); err != nil {
304 t.Fatal(err)
305 }
306 })
307 }
308}
309func TestServingParkRootRefusesBeforeEffects(t *testing.T) {
310 for _, attack := range []string{"peer", "peer-root", "census", "roster", "quarantine", "enrollment", "binding", "claim", "prior", "tags", "half-tags", "source", "inventory-revision", "receipt", "pending", "root-quarantine", "root-pending", "barrier", "canceled"} {
311 t.Run(attack, func(t *testing.T) {
312 f := newServingRootFixture(t, true)
313 ctx := t.Context()
314 switch attack {
315 case "peer":
316 f.p.root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return StopPeer{}, ErrClaimStopIdentity }
317 case "peer-root":
318 f.p.root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) {
319 return StopPeer{PID: 77, ProcessStart: "start", BootID: "boot", ControllerID: "controller"}, nil
320 }
321 case "census":
322 f.p.root.RequireNewWork = func(context.Context) error { return hostbox.ErrChanged }
323 case "roster":
324 f.actor.Eligible = false
325 case "quarantine":
326 f.actor.Quarantined = true
327 case "enrollment":
328 f.actor.Enrollment = "other"
329 case "binding":
330 f.actor.Binding = "8"
331 case "claim":
332 f.current.Generation = "foreign"
333 case "prior":
334 f.prior.Generation = "foreign"
335 case "tags":
336 f.tags(f.current.Stamp)
337 case "half-tags":
338 _, err := hostclaim.Update(t.Context(), f.f.claims, f.f.read(), func(c *hostclaim.Claim) error {
339 rel, _ := filepath.Rel(f.f.guard.DataDir, f.db)
340 delete(c.Inventory[f.actor.Box].Paths, rel+".gen")
341 return nil
342 })
343 if err != nil {
344 t.Fatal(err)
345 }
346 case "source":
347 if err := os.Rename(filepath.Dir(f.db), filepath.Dir(f.db)+".saved"); err != nil {
348 t.Fatal(err)
349 }
350 if err := os.Mkdir(filepath.Dir(f.db), 0700); err != nil {
351 t.Fatal(err)
352 }
353 case "inventory-revision":
354 in, _, err := f.p.root.Journals.Inventory.Snapshot(t.Context(), f.prior.Host, hostguard.Captured{Incarnation: f.prior.Incarnation, Stamp: f.prior.Stamp})
355 if err != nil {
356 t.Fatal(err)
357 }
358 if err = f.p.root.Journals.Inventory.WriteStopInventory(t.Context(), in); err != nil {
359 t.Fatal(err)
360 }
361 case "receipt":
362 f.mutateRoot(func(m map[string]any) {
363 s := m["State"].(map[string]any)
364 s["fence_journal"].([]any)[0].(map[string]any)["bytes"] = "Zm9yZ2Vk"
365 })
366 case "pending":
367 f.mutateRoot(func(m map[string]any) { m["State"].(map[string]any)["fence_journal"] = []any{} })
368 case "root-quarantine":
369 f.mutateRoot(func(m map[string]any) { m["Quarantined"] = true })
370 case "root-pending":
371 f.mutateRoot(func(m map[string]any) {
372 m["PendingInstallation"] = m["Installation"]
373 m["PendingInventory"] = "cGVuZGluZw=="
374 })
375 case "barrier":
376 f.mutateRoot(func(m map[string]any) { m["Barrier"] = m["State"].(map[string]any)["active_fence"] })
377 case "canceled":
378 var cancel context.CancelFunc
379 ctx, cancel = context.WithCancel(ctx)
380 cancel()
381 }
382 if _, err := f.p.parkServingDatabase(ctx, f.conn, f.current, f.prior); err == nil {
383 t.Fatal("unproved authority accepted")
384 }
385 names, err := f.disk.Names()
386 if err != nil {
387 t.Fatal(err)
388 }
389 if len(names) != 0 {
390 t.Fatalf("refusal wrote private state: %v", names)
391 }
392 db := f.db
393 if attack == "source" {
394 db = filepath.Join(filepath.Dir(db)+".saved", filepath.Base(db))
395 }
396 if b, err := os.ReadFile(db); err != nil || string(b) != "served bytes" {
397 t.Fatalf("source changed: %q %v", b, err)
398 }
399 })
400 }
401}
402
403type servingRootClientFixture struct{ f *servingRootFixture }
404
405func (c servingRootClientFixture) Park(ctx context.Context, current, prior servingArtifactIdentity) (servingParkAuthorization, error) {
406 return c.f.p.parkServingDatabase(ctx, c.f.conn, current, prior)
407}
408func (c servingRootClientFixture) ParkLegacy(ctx context.Context, current servingArtifactIdentity, prior servingLegacyPredecessor) (servingParkAuthorization, error) {
409 return c.f.p.parkLegacyServingDatabase(ctx, c.f.conn, current, prior)
410}
411func (c servingRootClientFixture) Complete(ctx context.Context, id string, current servingArtifactIdentity) error {
412 return c.f.p.completeServingDatabasePark(ctx, c.f.conn, id, current)
413}
414
415type servingRootWork struct {
416 g hostguard.Guard
417 o hostclaim.Observed
418}
419
420func (w servingRootWork) With(ctx context.Context, _ hostwrite.Identity, fn func() error) error {
421 return w.g.With(ctx, w.o.Host, hostguard.Captured{Incarnation: w.o.Claim.Holder.Incarnation, Stamp: w.o.Claim.Holder.Stamp}, func(c *hostclaim.Claim) error {
422 if !claimSameWriter(w.o, hostclaim.Observed{Host: w.o.Host, Claim: c}, nil) {
423 return hostbox.ErrChanged
424 }
425 return nil
426 }, fn)
427}
428func TestServingParkRootProductionRestoreSuccessor(t *testing.T) {
429 f := newServingRootFixture(t, true)
430 plan := plan12RestoreBaseline(t, f.f)
431 f.install(f.admitted)
432 r := &claimServingArtifactRetirement{Claims: f.f.claims, Guard: f.f.guard, DataDir: f.f.guard.DataDir, Box: f.actor.Box, Region: f.actor.Region, Root: servingRootClientFixture{f}}
433 in := ClaimLaunchInputs{DBPath: f.db}
434 work := servingRootWork{f.f.guard, f.admitted}
435 restore := hostbox.ColdRestore{Reader: f.f.data, Bucket: f.f.bucket, Generation: f.current.Generation, Work: work,
436 BeforeLegacyInstall: func(ctx context.Context, _, _ string) error { return r.BeforeLegacyInstall(ctx, f.admitted, in, work) },
437 BeforeInstallLocked: func(ctx context.Context, _, _ string) error { return r.EnrollLocked(ctx, f.admitted, in) },
438 RetireArtifacts: func(ctx context.Context, generation string) error { return r.Retire(ctx, f.admitted, in, generation) }}
439 if err := r.Recover(t.Context(), f.admitted, in); err != nil {
440 t.Fatal(err)
441 }
442 if err := restore.Install(t.Context(), f.db, plan); err != nil {
443 t.Fatal(err)
444 }
445 if got := mwReadDB(t, f.db); got != "selected recovery baseline" {
446 t.Fatal(got)
447 }
448 marker, err := hostbox.ReadGenerationMarker(f.db + ".gen")
449 if err != nil || marker != "G2" {
450 t.Fatalf("marker %q %v", marker, err)
451 }
452 names, err := f.disk.Names()
453 if err != nil {
454 t.Fatal(err)
455 }
456 for _, name := range names {
457 if strings.HasPrefix(name, "retirement-") && strings.HasSuffix(name, ".json") {
458 rec, err := f.p.ledger.read(strings.TrimSuffix(strings.TrimPrefix(name, "retirement-"), ".json"))
459 if err != nil || !rec.Complete {
460 t.Fatalf("incomplete: %v", err)
461 }
462 b, err := os.ReadFile(filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", rec.Authorization.ID, filepath.Base(f.db)))
463 if err != nil || string(b) != "served bytes" {
464 t.Fatalf("parked predecessor: %q %v", b, err)
465 }
466 }
467 }
468}
469func TestServingParkRootWaitReauthenticates(t *testing.T) {
470 f := newServingRootFixture(t, true)
471 unlock, err := f.disk.Lock(t.Context(), "namespace.lock")
472 if err != nil {
473 t.Fatal(err)
474 }
475 ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
476 defer cancel()
477 if _, err = f.p.parkServingDatabase(ctx, f.conn, f.current, f.prior); !errors.Is(err, context.DeadlineExceeded) {
478 t.Fatalf("lock bypass: %v", err)
479 }
480 unlock()
481 f.p.root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return StopPeer{}, ErrClaimStopIdentity }
482 if _, err = f.park(); !errors.Is(err, ErrClaimStopIdentity) {
483 t.Fatalf("stale authentication: %v", err)
484 }
485}
486
487func TestServingParkRootContinuesOriginalRetirement(t *testing.T) {
488 for _, point := range []string{"rename", "parked", "retag", "lost-continuation"} {
489 t.Run(point, func(t *testing.T) {
490 f := newServingRootFixture(t, true)
491 original := f.current
492 fs := f.p.ledger.fs.(*servingParkFS)
493 rename := fs.rename
494 if point == "rename" {
495 once := false
496 fs.rename = func(a int, b string, c int, d string, e uint) error {
497 err := rename(a, b, c, d, e)
498 if err == nil && !once {
499 once = true
500 return errors.New("lost rename reply")
501 }
502 return err
503 }
504 }
505 a, err := f.park()
506 if point == "rename" {
507 if err == nil {
508 t.Fatal("fault missed")
509 }
510 fs.rename = rename
511 a = servingParkAuthorization{ID: servingRetirementID(f.prior, original, false)}
512 } else if err != nil {
513 t.Fatal(err)
514 }
515 before, err := f.p.ledger.read(a.ID)
516 if err != nil {
517 t.Fatal(err)
518 }
519 if point == "retag" || point == "lost-continuation" {
520 f.tags(original.Stamp)
521 }
522 f.install(f.f.read())
523 servingStop(t, f.f, f.f.read())
524 o := f.f.read()
525 later, err := hostclaim.Update(t.Context(), f.f.claims, o, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation, Box: "box-1", Region: "ie", PredecessorRegion: "ie", Mode: "cold", NewGeneration: "G3"}))
526 if err != nil {
527 t.Fatal(err)
528 }
529 f.current, err = servingWriterIdentity(later, "box-1", "ie")
530 if err != nil {
531 t.Fatal(err)
532 }
533 f.f.now = f.f.now.Add(2 * time.Hour)
534 requested := f.prior
535 if point == "retag" || point == "lost-continuation" {
536 requested = original
537 }
538 if point == "lost-continuation" {
539 disk := &servingParkFaultStore{Store: f.disk, failAt: 1}
540 f.p.ledger.state = disk
541 if _, err = f.p.parkServingDatabase(t.Context(), f.conn, f.current, requested); err == nil {
542 t.Fatal("lost continuation reply missed")
543 }
544 f.p.ledger.state = f.disk
545 }
546 continued, err := f.p.parkServingDatabase(t.Context(), f.conn, f.current, requested)
547 if err != nil {
548 t.Fatal(err)
549 }
550 if continued.ID != a.ID || continued.OriginalSuccessor != original || continued.Successor != f.current || continued.Predecessor != f.prior {
551 t.Fatalf("relabelled retirement: %+v", continued)
552 }
553 f.tags(f.current.Stamp)
554 if err = f.p.completeServingDatabasePark(t.Context(), f.conn, a.ID, f.current); err != nil {
555 t.Fatal(err)
556 }
557 after, err := f.p.ledger.read(a.ID)
558 if err != nil {
559 t.Fatal(err)
560 }
561 if after.Authorization != before.Authorization || len(after.Continuations) != 1 || !after.Complete {
562 t.Fatalf("history: %+v", after)
563 }
564 if !before.ParkedAt.IsZero() && (!before.ParkedAt.Equal(after.ParkedAt) || !before.ReapAfter.Equal(after.ReapAfter)) {
565 t.Fatal("continuation reset retention")
566 }
567 if err = f.p.completeServingDatabasePark(t.Context(), f.conn, a.ID, original); err == nil {
568 t.Fatal("old successor completed continuation")
569 }
570 again, err := f.p.parkServingDatabase(t.Context(), f.conn, f.current, servingArtifactIdentity{})
571 if err != nil || again != continued {
572 t.Fatalf("replay: %+v %v", again, err)
573 }
574 again, err = f.p.parkServingDatabase(t.Context(), f.conn, f.current, requested)
575 if err != nil || again != continued {
576 t.Fatalf("original continuation selector: %+v %v", again, err)
577 }
578 })
579 }
580}
581
582func TestServingParkRootRechecksClaimAfterInputs(t *testing.T) {
583 f := newServingRootFixture(t, true)
584 original := f.p.root.Stop.Authenticate
585 f.p.root.Stop.Authenticate = func(ctx context.Context, conn net.Conn) (StopPeer, error) {
586 _, err := hostclaim.Update(ctx, f.f.claims, f.f.read(), func(c *hostclaim.Claim) error { c.Inventory = nil; return nil })
587 if err != nil {
588 return StopPeer{}, err
589 }
590 return original(ctx, conn)
591 }
592 if _, err := f.park(); err == nil {
593 t.Fatal("used claim captured before peer and inventory checks")
594 }
595 names, err := f.disk.Names()
596 if err != nil || len(names) != 0 {
597 t.Fatalf("effects: %v %v", names, err)
598 }
599}
600func TestServingParkRootContinuationNeedsStop(t *testing.T) {
601 f := newServingRootFixture(t, true)
602 a, err := f.park()
603 if err != nil {
604 t.Fatal(err)
605 }
606 f.install(f.admitted)
607 original := f.current
608 o := f.f.read()
609 next, err := hostclaim.Update(t.Context(), f.f.claims, o, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation, Box: "box-1", Region: "ie", PredecessorRegion: "ie", Mode: "cold", NewGeneration: "G3"}))
610 if err != nil {
611 t.Fatal(err)
612 }
613 f.current, err = servingWriterIdentity(next, "box-1", "ie")
614 if err != nil {
615 t.Fatal(err)
616 }
617 if _, err = f.park(); err == nil {
618 t.Fatal("continued without original successor stop")
619 }
620 rec, err := f.p.ledger.read(a.ID)
621 if err != nil || rec.activeAuthorization().Successor != original || len(rec.Continuations) != 0 {
622 t.Fatalf("changed record: %+v %v", rec, err)
623 }
624}
625