amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package park
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "net/url"
8 "reflect"
9 "strings"
10 "testing"
11 "time"
12
13 "github.com/carlosframework/platform/internal/hostclaim"
14 "github.com/carlosframework/platform/internal/lease"
15 "github.com/carlosframework/platform/internal/lease/leasetest"
16 "github.com/carlosframework/platform/internal/release"
17)
18
19type testLease struct {
20 released int
21 failed bool
22}
23
24func (l *testLease) Check(context.Context) error {
25 if l.failed {
26 return ErrChanged
27 }
28 return nil
29}
30func (l *testLease) Release(context.Context) error { l.released++; return nil }
31func (*testLease) Close() {}
32
33type testLeases struct{ held *testLease }
34
35func (l testLeases) Acquire(context.Context, string, string) (HeldLease, error) { return l.held, nil }
36
37type fixture struct {
38 e Engine
39 b *leasetest.VersionedBucket
40 w Work
41 now time.Time
42 ob hostclaim.Obligation
43 held *testLease
44}
45
46func newFixture(t *testing.T) *fixture {
47 t.Helper()
48 ctx := t.Context()
49 now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
50 host := "shop.bab.example.com"
51 inc, err := hostclaim.Incarnation("bab", "jam", host, now)
52 if err != nil {
53 t.Fatal(err)
54 }
55 b := leasetest.NewVersionedBucket()
56 b.AddBucket("data")
57 b.PageSize = 1
58 f := &fixture{b: b, w: Work{host, "ob", "op"}, now: now.Add(16 * time.Minute), held: &testLease{}}
59 f.ob = hostclaim.Obligation{ID: "ob", Op: "op", Kind: "park-replica", Region: "ie", Retirement: "rt1", Owner: "region:ie", Park: &hostclaim.ParkScope{Location: "instances/" + host + "/" + inc, Incarnation: inc, EvidenceAt: now, RequestLimit: time.Second}}
60 st := release.NewDirStore(t.TempDir())
61 _, err = hostclaim.Update(ctx, st, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
62 c.Holder = &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Incarnation: inc, CreatedAt: now, Location: f.ob.Park.Location, Stamp: 1}
63 c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "op"}
64 c.Obligations = []hostclaim.Obligation{f.ob}
65 c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
66 return nil
67 })
68 if err != nil {
69 t.Fatal(err)
70 }
71 serial := 1
72 f.e = Engine{Store: b, Claims: st, Bucket: "data", Region: "ie", Leases: testLeases{f.held}, Now: func() time.Time { return f.now }, NewID: func() string { serial++; return fmt.Sprintf("rt%d", serial) }, Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }, View: func(_ context.Context, o hostclaim.Observed) (View, error) {
73 return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
74 }}
75 return f
76}
77func (f *fixture) seed() []lease.ObjectVersion {
78 p := f.ob.Park.Location + "/"
79 f.b.Put("data", p+"0000/a", []byte("old"))
80 f.b.Put("data", p+"0000/a", []byte("new"))
81 f.b.BareDelete("data", p+"0000/a")
82 f.b.Put("data", p+"lock.json", []byte("lease"))
83 vs, _ := f.e.list(context.Background(), f.w.Host, f.ob.Park.Location)
84 return vs
85}
86func (f *fixture) done(t *testing.T) {
87 t.Helper()
88 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
89 if err != nil {
90 t.Fatal(err)
91 }
92 if len(obs.Claim.Obligations) != 0 {
93 t.Fatal("obligation still open")
94 }
95 if got := obs.Claim.QuietAfter["ie"]; !got.Equal(f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)) {
96 t.Fatal(got)
97 }
98 vs, err := f.e.list(t.Context(), f.w.Host, f.ob.Park.Location)
99 if err != nil || len(vs) != 0 {
100 t.Fatalf("residue: %v %v", vs, err)
101 }
102}
103func TestRemoteCrashResume(t *testing.T) {
104 for _, tc := range []struct {
105 step string
106 n int
107 }{{"copy", 1}, {"copy", 2}, {"manifest", 1}, {"delete", 1}, {"delete", 2}, {"delete", 3}, {"verify", 1}, {"before-completion", 1}, {"completion", 1}} {
108 t.Run(fmt.Sprintf("%s-%d", tc.step, tc.n), func(t *testing.T) {
109 f := newFixture(t)
110 source := f.seed()
111 stop := errors.New("crash")
112 n := 0
113 f.e.After = func(s string) error {
114 if s == tc.step {
115 n++
116 if n == tc.n {
117 return stop
118 }
119 }
120 return nil
121 }
122 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
123 t.Fatalf("crash: %v", err)
124 }
125 if tc.step != "completion" && f.held.released != 0 {
126 t.Fatal("released before ack")
127 }
128 f.e.After = nil
129 if err := f.e.Run(t.Context(), f.w); err != nil {
130 t.Fatal(err)
131 }
132 f.done(t)
133 m, err := f.e.manifest(t.Context(), f.w, f.ob)
134 if err != nil || m == nil || len(m.Delete) != len(source) || len(m.Archive) != 2 {
135 t.Fatalf("manifest: %#v %v", m, err)
136 }
137 for _, a := range m.Archive {
138 body, err := f.b.GetVersion(t.Context(), "data", a.Destination.Key, a.Destination.VersionID)
139 if err != nil || int64(len(body)) != a.Source.Size {
140 t.Fatalf("copy: %q %v", body, err)
141 }
142 }
143 if f.b.LatestID("data", f.ob.Park.Location+"/lock.json") == "" {
144 t.Fatal("lock deleted")
145 }
146 })
147 }
148}
149func TestStorageFailuresKeepObligation(t *testing.T) {
150 for _, op := range []string{"list", "get", "copy", "put", "delete"} {
151 t.Run(op, func(t *testing.T) {
152 f := newFixture(t)
153 f.seed()
154 if op == "get" {
155 f.e.After = func(s string) error {
156 if s == "manifest" {
157 return ErrPending
158 }
159 return nil
160 }
161 _ = f.e.Run(t.Context(), f.w)
162 f.e.After = nil
163 }
164 failure := errors.New("injected")
165 f.b.FailOn = func(got, key string) error {
166 if got == op {
167 return failure
168 }
169 return nil
170 }
171 if err := f.e.Run(t.Context(), f.w); err == nil {
172 t.Fatal("failure acknowledged")
173 }
174 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
175 if err != nil || len(obs.Claim.Obligations) != 1 || len(obs.Claim.QuietAfter) != 0 {
176 t.Fatalf("claim: %v %v", obs, err)
177 }
178 f.b.FailOn = nil
179 if err := f.e.Run(t.Context(), f.w); err != nil {
180 t.Fatal(err)
181 }
182 f.done(t)
183 })
184 }
185}
186func TestPageFailureDoesNotAct(t *testing.T) {
187 f := newFixture(t)
188 f.seed()
189 before := f.b.Snapshot()
190 f.b.FailListAfterPages = 1
191 if err := f.e.Run(t.Context(), f.w); err == nil {
192 t.Fatal("short listing accepted")
193 }
194 if f.b.Snapshot() != before {
195 t.Fatal("mutated on listing failure")
196 }
197}
198func TestHorizonLatePut(t *testing.T) {
199 f := newFixture(t)
200 f.seed()
201 deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)
202 f.now = deadline
203 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
204 t.Fatal(err)
205 }
206 late := f.b.Put("data", f.ob.Park.Location+"/0000/late", []byte("late"))
207 f.now = deadline.Add(time.Nanosecond)
208 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
209 t.Fatal(err)
210 }
211 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
212 if obs.Claim.Obligations[0].Retirement == f.ob.Retirement {
213 t.Fatal("manifest reused")
214 }
215 if err := f.e.Run(t.Context(), f.w); err != nil {
216 t.Fatal(err)
217 }
218 f.done(t)
219 if _, err := f.b.GetVersion(t.Context(), "data", f.ob.Park.Location+"/0000/late", late); !errors.Is(err, lease.ErrMirrorNotFound) {
220 t.Fatal(err)
221 }
222}
223func TestVerificationListingStartsAfterHorizon(t *testing.T) {
224 f := newFixture(t)
225 f.seed()
226 deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)
227 f.now = deadline
228 f.e.After = func(s string) error {
229 if s == "verify" {
230 f.now = deadline.Add(time.Hour)
231 }
232 return nil
233 }
234 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
235 t.Fatal(err)
236 }
237}
238func TestManifestRetryDoesNotCopy(t *testing.T) {
239 f := newFixture(t)
240 f.seed()
241 f.e.After = func(s string) error {
242 if s == "delete" {
243 return ErrPending
244 }
245 return nil
246 }
247 _ = f.e.Run(t.Context(), f.w)
248 f.e.After = nil
249 f.b.FailOn = func(op, key string) error {
250 if op == "copy" {
251 t.Error("retry copied")
252 return ErrUnsafe
253 }
254 return nil
255 }
256 if err := f.e.Run(t.Context(), f.w); err != nil {
257 t.Fatal(err)
258 }
259}
260func TestCASConflictDoesNotAcknowledge(t *testing.T) {
261 f := newFixture(t)
262 f.seed()
263 f.e.After = func(s string) error {
264 if s == "before-completion" {
265 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
266 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error { c.Lost = append(c.Lost, "other-box"); return nil })
267 return err
268 }
269 return nil
270 }
271 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, hostclaim.ErrConflict) {
272 t.Fatal(err)
273 }
274 if f.held.released != 0 {
275 t.Fatal("released unacknowledged lease")
276 }
277 f.e.After = nil
278 if err := f.e.Run(t.Context(), f.w); err != nil {
279 t.Fatal(err)
280 }
281}
282func TestLiveGenerationCannotBeParked(t *testing.T) {
283 f := newFixture(t)
284 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
285 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
286 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
287 return nil
288 })
289 if err != nil {
290 t.Fatal(err)
291 }
292 f.seed()
293 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
294 return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: hostclaim.RootGeneration, GenerationOf: func(Ref) (string, bool) { return hostclaim.RootGeneration, true }}, nil
295 }
296 before := f.b.Snapshot()
297 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrUnsafe) {
298 t.Fatal(err)
299 }
300 if before != f.b.Snapshot() {
301 t.Fatal("live data changed")
302 }
303}
304
305func TestRemoteMarkerDisagreementIsLocationScoped(t *testing.T) {
306 for _, scenario := range []string{"old-incarnation", "live-location", "ancestor-scope", "ancestor-holder", "unknown-region", "unknown-marker", "unknown-generations", "unknown-pinned-generation", "view-error", "authorization-error"} {
307 t.Run(scenario, func(t *testing.T) {
308 f, obs, view, _ := sweepFixture(t)
309 view.MarkerGeneration = "unrelated"
310 f.ob.Op = "session:3"
311 f.w.Op = f.ob.Op
312 if scenario != "live-location" {
313 f.ob.Park.Incarnation = strings.Repeat("a", 64)
314 f.ob.Park.Location = "instances/" + f.w.Host + "/" + f.ob.Park.Incarnation
315 }
316 if scenario == "ancestor-scope" {
317 f.ob.Park.Location = "instances/" + f.w.Host
318 }
319 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
320 c.Obligations = []hostclaim.Obligation{f.ob}
321 if scenario == "ancestor-holder" {
322 c.Holder.Location = "instances/" + f.w.Host
323 }
324 return nil
325 })
326 if err != nil {
327 t.Fatal(err)
328 }
329 source := f.seed()
330 before := f.b.Snapshot()
331 liveBefore, err := f.e.list(t.Context(), f.w.Host, obs.Claim.Holder.Location)
332 if err != nil {
333 t.Fatal(err)
334 }
335 switch scenario {
336 case "unknown-region":
337 view.AuthoritativeRegion = ""
338 case "unknown-marker":
339 view.MarkerKnown = false
340 case "unknown-generations":
341 view.GenerationOf = nil
342 case "unknown-pinned-generation":
343 view.GenerationOf = func(Ref) (string, bool) { return "", false }
344 }
345 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
346 v := view
347 v.ClaimETag = o.ETag
348 if scenario == "view-error" {
349 return v, ErrUnsafe
350 }
351 return v, nil
352 }
353 authorized := false
354 f.e.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
355 authorized = true
356 if !reflect.DeepEqual(o, f.ob) || scenario == "authorization-error" {
357 return ErrUnsafe
358 }
359 return nil
360 }
361 err = f.e.Run(t.Context(), f.w)
362 if !authorized {
363 t.Fatal("scope was not authorized")
364 }
365 if scenario != "old-incarnation" {
366 if !errors.Is(err, ErrUnsafe) {
367 t.Fatalf("Run = %v, want refusal", err)
368 }
369 if f.b.Snapshot() != before {
370 t.Fatal("refused work changed storage")
371 }
372 pending, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
373 if err != nil {
374 t.Fatal(err)
375 }
376 if len(pending.Claim.Obligations) != 1 || len(pending.Claim.QuietAfter) != 0 || f.held.released != 0 {
377 t.Fatal("refused work acknowledged")
378 }
379 return
380 }
381 if err != nil {
382 t.Fatalf("old incarnation refused by live marker disagreement: %v", err)
383 }
384 f.done(t)
385 m, err := f.e.manifest(t.Context(), f.w, f.ob)
386 if err != nil || m == nil || !reflect.DeepEqual(m.Delete, source) || len(m.Archive) != 2 {
387 t.Fatalf("old incarnation manifest: %+v %v", m, err)
388 }
389 for _, copy := range m.Archive {
390 body, err := f.b.GetVersion(t.Context(), "data", copy.Destination.Key, copy.Destination.VersionID)
391 if err != nil || (string(body) != "old" && string(body) != "new") {
392 t.Fatalf("archive: %q %v", body, err)
393 }
394 }
395 liveAfter, err := f.e.list(t.Context(), f.w.Host, obs.Claim.Holder.Location)
396 if err != nil || !reflect.DeepEqual(liveBefore, liveAfter) {
397 t.Fatalf("live location changed: %v", err)
398 }
399 if f.held.released != 1 {
400 t.Fatal("completed work did not release its lease")
401 }
402 })
403 }
404}
405
406func TestBoundScopeAndQuietAfter(t *testing.T) {
407 f := newFixture(t)
408 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
409 later := f.now.Add(time.Hour)
410 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
411 c.Obligations[0].Park = nil
412 c.QuietAfter = map[string]time.Time{"ie": later}
413 return nil
414 })
415 if err != nil {
416 t.Fatal(err)
417 }
418 if err := f.e.Bind(t.Context(), f.w, *f.ob.Park); err != nil {
419 t.Fatal(err)
420 }
421 bad := *f.ob.Park
422 bad.EvidenceAt = bad.EvidenceAt.Add(time.Second)
423 if err := f.e.Bind(t.Context(), f.w, bad); !errors.Is(err, ErrChanged) {
424 t.Fatal(err)
425 }
426 if err := f.e.Run(t.Context(), f.w); err != nil {
427 t.Fatal(err)
428 }
429 obs, _ = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
430 if !obs.Claim.QuietAfter["ie"].Equal(later) {
431 t.Fatal("quiet_after regressed")
432 }
433}
434func TestBindSurvivesSerialization(t *testing.T) {
435 f := newFixture(t)
436 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
437 if err != nil {
438 t.Fatal(err)
439 }
440 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
441 c.Obligations[0].Park = nil
442 return nil
443 })
444 if err != nil {
445 t.Fatal(err)
446 }
447 scope := *f.ob.Park
448 scope.EvidenceAt = time.Now() // Carries a monotonic reading that JSON drops.
449 if err := f.e.Bind(t.Context(), f.w, scope); err != nil {
450 t.Fatal(err)
451 }
452 before, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
453 if err != nil {
454 t.Fatal(err)
455 }
456 for _, at := range []time.Time{scope.EvidenceAt, scope.EvidenceAt.UTC()} {
457 same := scope
458 same.EvidenceAt = at
459 if err := f.e.Bind(t.Context(), f.w, same); err != nil {
460 t.Fatal(err)
461 }
462 }
463 for _, change := range []func(*hostclaim.ParkScope){
464 func(p *hostclaim.ParkScope) { p.Location += "/other" },
465 func(p *hostclaim.ParkScope) { p.Incarnation = strings.Repeat("f", 64) },
466 func(p *hostclaim.ParkScope) { p.RequestLimit += time.Second },
467 func(p *hostclaim.ParkScope) { p.EvidenceAt = p.EvidenceAt.Add(time.Nanosecond) },
468 } {
469 bad := scope
470 change(&bad)
471 if err := f.e.Bind(t.Context(), f.w, bad); !errors.Is(err, ErrChanged) {
472 t.Fatal(err)
473 }
474 }
475 after, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
476 if err != nil || after.ETag != before.ETag {
477 t.Fatalf("bind changed persisted scope: %v", err)
478 }
479}
480
481func TestLegacyLocationDoesNotIncludeIncarnationChildren(t *testing.T) {
482 f := newFixture(t)
483 key := f.ob.Park.Location + "/0000/data"
484 f.b.Put("data", key, []byte("new incarnation"))
485 vs, err := f.e.list(t.Context(), f.w.Host, "instances/"+f.w.Host)
486 if err != nil || len(vs) != 0 {
487 t.Fatalf("%v %v", vs, err)
488 }
489 for _, v := range []string{"a/b", "a@b", "a+b", "a b", "a?b"} {
490 key := destination(f.w.Host, f.ob, lease.ObjectVersion{Key: key, VersionID: v})
491 if strings.HasSuffix(key, "@"+v) {
492 t.Errorf("version not escaped: %s", v)
493 }
494 }
495}
496
497func TestDeleteParksStoppedAuthoritativeLocation(t *testing.T) {
498 f := newFixture(t)
499 f.seed()
500 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
501 return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie"}, nil
502 }
503 if err := f.e.Run(t.Context(), f.w); err != nil {
504 t.Fatal(err)
505 }
506 f.done(t)
507}
508func TestForeignRootGenerationRetirement(t *testing.T) {
509 f := newFixture(t)
510 f.seed()
511 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
512 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
513 c.Obligations[0].Kind = "park-generation"
514 c.Obligations[0].Generation = hostclaim.RootGeneration
515 return nil
516 })
517 if err != nil {
518 t.Fatal(err)
519 }
520 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
521 return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk", GenerationOf: func(Ref) (string, bool) { return "", true }}, nil
522 }
523 if err := f.e.Run(t.Context(), f.w); err != nil {
524 t.Fatal(err)
525 }
526 f.done(t)
527}
528
529func TestLeaseLossAndAuthorizationRefuse(t *testing.T) {
530 for _, phase := range []string{"before-first-effect", "before-completion", "authorize"} {
531 t.Run(phase, func(t *testing.T) {
532 f := newFixture(t)
533 f.seed()
534 before := f.b.Snapshot()
535 switch phase {
536 case "before-first-effect":
537 f.held.failed = true
538 case "before-completion":
539 f.e.After = func(s string) error {
540 if s == phase {
541 f.held.failed = true
542 }
543 return nil
544 }
545 case "authorize":
546 f.e.Authorize = func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return ErrChanged }
547 }
548 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrChanged) {
549 t.Fatal(err)
550 }
551 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
552 if len(obs.Claim.Obligations) != 1 || f.held.released != 0 {
553 t.Fatal("acknowledged without authority")
554 }
555 if phase != "before-completion" && f.b.Snapshot() != before {
556 t.Fatal("acted without authority")
557 }
558 })
559 }
560}
561func TestMalformedManifestNeverDeletes(t *testing.T) {
562 f := newFixture(t)
563 f.seed()
564 _, err := f.b.PutIfAbsent(t.Context(), "data", archiveBase(f.w.Host, f.ob)+"manifest.json", []byte(`{"V":1,"Host":"other.example"}`))
565 if err != nil {
566 t.Fatal(err)
567 }
568 before := f.b.Snapshot()
569 if err := f.e.Run(t.Context(), f.w); err == nil {
570 t.Fatal("accepted malformed manifest")
571 }
572 if before != f.b.Snapshot() {
573 t.Fatal("deleted from bad manifest")
574 }
575}
576
577// Parked is what the landing's parking phase waits for: the manifest is
578// there and every version it lists is gone. It is not the acknowledgement,
579// which still waits for the horizon in Run.
580func TestParkedReportsManifestAndDeletions(t *testing.T) {
581 f := newFixture(t)
582 f.seed()
583 parked := func() bool {
584 t.Helper()
585 ok, err := f.e.Parked(t.Context(), f.w)
586 if err != nil {
587 t.Fatalf("Parked: %v", err)
588 }
589 return ok
590 }
591 if parked() {
592 t.Fatal("parked before any work")
593 }
594 stop := errors.New("crash")
595 crashAt := func(step string, n int) {
596 t.Helper()
597 seen := 0
598 f.e.After = func(s string) error {
599 if s == step {
600 seen++
601 if seen == n {
602 return stop
603 }
604 }
605 return nil
606 }
607 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
608 t.Fatalf("crash at %s: %v", step, err)
609 }
610 f.e.After = nil
611 }
612 crashAt("manifest", 1)
613 if parked() {
614 t.Fatal("parked with a manifest and nothing deleted")
615 }
616 crashAt("delete", 1)
617 if parked() {
618 t.Fatal("parked with versions left to delete")
619 }
620 // Resume through every delete, stopping at the verification: the
621 // manifest is there and everything in it is gone, but the obligation
622 // is not acknowledged.
623 crashAt("verify", 1)
624 if !parked() {
625 t.Fatal("not parked after every listed version is gone")
626 }
627 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
628 if err != nil || len(obs.Claim.Obligations) != 1 {
629 t.Fatalf("Parked must not acknowledge: %v %v", obs.Claim.Obligations, err)
630 }
631 // A version the manifest does not list is not in the deletion set: the
632 // retirement's own verification pass handles it, Parked only reports on
633 // what the manifest promised.
634 f.b.Put("data", f.ob.Park.Location+"/0000/late", []byte("late"))
635 if !parked() {
636 t.Fatal("an unlisted late version changed the manifest's answer")
637 }
638}
639
640// A key with many versions is listed once, however many of its versions are
641// in the deletion set.
642func TestParkedListsEachKeyOnce(t *testing.T) {
643 f := newFixture(t)
644 f.seed()
645 stop := errors.New("crash")
646 f.e.After = func(s string) error {
647 if s == "verify" {
648 return stop
649 }
650 return nil
651 }
652 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
653 t.Fatal(err)
654 }
655 f.e.After = nil
656 m, err := f.e.manifest(t.Context(), f.w, f.ob)
657 if err != nil || m == nil {
658 t.Fatal(m, err)
659 }
660 perKey := map[string]int{}
661 for _, d := range m.Delete {
662 perKey[d.Key]++
663 }
664 if len(m.Delete) < 3 || len(perKey) != 1 {
665 t.Fatalf("the fixture must put several versions under one key: %v", perKey)
666 }
667 // The manifest and chore reads list too; count only the deletion keys.
668 lists := 0
669 f.b.FailOn = func(op, key string) error {
670 if op == "list" && perKey[key] > 0 {
671 lists++
672 }
673 return nil
674 }
675 if ok, err := f.e.Parked(t.Context(), f.w); !ok || err != nil {
676 t.Fatalf("parked: %v %v", ok, err)
677 }
678 f.b.FailOn = nil
679 if lists != 1 {
680 t.Fatalf("a key with %d deleted versions was listed %d times", len(m.Delete), lists)
681 }
682}
683
684func TestParkedNeverGuesses(t *testing.T) {
685 f := newFixture(t)
686 f.seed()
687 stop := errors.New("crash")
688 f.e.After = func(s string) error {
689 if s == "verify" {
690 return stop
691 }
692 return nil
693 }
694 if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
695 t.Fatal(err)
696 }
697 f.e.After = nil
698 boom := errors.New("offline")
699 for _, fail := range []string{"list", "get"} {
700 f.b.FailOn = func(op, key string) error {
701 if op == fail {
702 return boom
703 }
704 return nil
705 }
706 ok, err := f.e.Parked(t.Context(), f.w)
707 if ok || err == nil {
708 t.Fatalf("%s failure answered %v, %v", fail, ok, err)
709 }
710 }
711 f.b.FailOn = nil
712 if ok, err := f.e.Parked(t.Context(), f.w); !ok || err != nil {
713 t.Fatalf("recovered read: %v %v", ok, err)
714 }
715 // A chore that is not in the claim cannot be verified from its manifest.
716 if ok, err := f.e.Parked(t.Context(), Work{f.w.Host, "nope", f.w.Op}); ok || !errors.Is(err, ErrChanged) {
717 t.Fatalf("unknown chore: %v %v", ok, err)
718 }
719 // A rotated retirement has no manifest yet: not parked, and not an error.
720 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
721 if err != nil {
722 t.Fatal(err)
723 }
724 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
725 c.Obligations[0].Retirement = "rt-other"
726 return nil
727 })
728 if err != nil {
729 t.Fatal(err)
730 }
731 if ok, err := f.e.Parked(t.Context(), f.w); ok || err != nil {
732 t.Fatalf("a retirement with no manifest answered %v, %v", ok, err)
733 }
734 if ok, err := (Engine{}).Parked(t.Context(), f.w); ok || !errors.Is(err, ErrUnsafe) {
735 t.Fatalf("unconfigured engine: %v %v", ok, err)
736 }
737}
738
739// A move's retirement goes through the same engine as any other: queued by
740// hostclaim.QueueLocationPark on the existing park-replica kind, archived
741// under parked/<incarnation>/<host>/<retirement>/, and the manifest's Op is
742// the move id (F3 reads it).
743func TestEngineArchivesAMoveRetirementUnderParked(t *testing.T) {
744 f := newFixture(t)
745 const move = "mv_abcdefghijklmnopqrstuvwxyz"
746 scope := *f.ob.Park
747 n := 0
748 newID := func() string { n++; return fmt.Sprintf("q%d", n) }
749 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
750 if err != nil {
751 t.Fatal(err)
752 }
753 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
754 // The source's region after the commit: the claim's landing/committed.
755 c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: move, Phase: "committed", Intent: map[string]string{"to": "uk"}}
756 c.Obligations = nil
757 return hostclaim.QueueLocationPark(move, "ie", "", scope, newID)(c)
758 })
759 if err != nil {
760 t.Fatal(err)
761 }
762 obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
763 if err != nil || len(obs.Claim.Obligations) != 1 {
764 t.Fatal(obs.Claim.Obligations, err)
765 }
766 ob := obs.Claim.Obligations[0]
767 f.ob, f.w = ob, Work{f.w.Host, ob.ID, move}
768 id := f.b.Put("data", scope.Location+"/0000/x.ltx", []byte("ltx"))
769 if err := f.e.Run(t.Context(), f.w); err != nil {
770 t.Fatal(err)
771 }
772 m, err := f.e.manifest(t.Context(), f.w, ob)
773 if err != nil || m == nil {
774 t.Fatalf("manifest: %v %v", m, err)
775 }
776 if m.Op != move || m.Retirement != ob.Retirement || m.Location != scope.Location {
777 t.Fatalf("manifest: %+v", m)
778 }
779 want := "parked/" + scope.Incarnation + "/" + f.w.Host + "/" + ob.Retirement + "/instances/0000/x.ltx@" + url.QueryEscape(id)
780 if len(m.Archive) != 1 || m.Archive[0].Destination.Key != want {
781 t.Fatalf("archive %+v, want %s", m.Archive, want)
782 }
783 f.done(t)
784}
785