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 "reflect"
7 "strings"
8 "testing"
9 "time"
10
11 "github.com/carlosframework/platform/internal/hostclaim"
12 "github.com/carlosframework/platform/internal/lease"
13 "github.com/carlosframework/platform/internal/lease/leasetest"
14)
15
16func sweepFixture(t *testing.T) (*fixture, hostclaim.Observed, View, []lease.ObjectVersion) {
17 f := newFixture(t)
18 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
19 obs, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
20 c.Obligations = nil
21 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
22 c.Holder.Generation = "g3"
23 c.Holder.Session = 3
24 c.Holder.Admissions = []hostclaim.AdmissionRecord{
25 {Session: 2, Generation: "g2", Predecessor: "g1", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
26 {Session: 3, Generation: "g3", Predecessor: "g2", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
27 }
28 return nil
29 })
30 if err != nil {
31 t.Fatal(err)
32 }
33 p := f.ob.Park.Location + "/"
34 for _, g := range []string{"g0", "g1", "g2", "g3"} {
35 f.b.Put("data", p+g+"/data", []byte(g))
36 }
37 f.b.Put("data", p+"generation.json", []byte("older"))
38 markerID := f.b.Put("data", p+"generation.json", []byte("g1"))
39 pinned := f.b.Put("data", p+"pinned/data", []byte("baseline"))
40 f.b.Put("data", p+"pinned/data", []byte("newer"))
41 classify := func(r Ref) (string, bool) {
42 s := strings.TrimPrefix(r.Key, p)
43 g, _, ok := strings.Cut(s, "/")
44 if r.Key == p+"generation.json" {
45 return "@marker", true
46 }
47 return g, ok
48 }
49 v := View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: "g1", Marker: Ref{p + "generation.json", markerID}, Pinned: map[Ref]bool{{p + "pinned/data", pinned}: true}, GenerationOf: classify}
50 vs, err := f.e.Discover(t.Context(), f.w.Host)
51 if err != nil {
52 t.Fatal(err)
53 }
54 return f, obs, v, vs
55}
56
57func TestSweepMovesNamespaceIsSeparateFromLiveInstance(t *testing.T) {
58 f := newFixture(t)
59 allowFixtureMoveSweep(f)
60 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
61 if err != nil {
62 t.Fatal(err)
63 }
64 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
65 c.Obligations = nil
66 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
67 return nil
68 })
69 if err != nil {
70 t.Fatal(err)
71 }
72 liveKey := f.ob.Park.Location + "/0000/live"
73 liveID := f.b.Put("data", liveKey, []byte("live"))
74 for _, op := range []string{"move-1", f.ob.Park.Incarnation} {
75 f.b.Put("data", "moves/"+f.w.Host+"/"+op+"/staging", []byte(op))
76 }
77 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
78 return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: hostclaim.RootGeneration, GenerationOf: func(Ref) (string, bool) { return hostclaim.RootGeneration, true }}, nil
79 }
80 d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit)
81 if err != nil {
82 t.Fatal(err)
83 }
84 want := []Candidate{{Location: "moves/" + f.w.Host, Incarnation: f.ob.Park.Incarnation}}
85 if !reflect.DeepEqual(d.Candidates, want) {
86 t.Fatalf("staging candidates=%+v, want %+v", d.Candidates, want)
87 }
88 vs, err := f.b.ListVersions(t.Context(), "data", "moves/"+f.w.Host+"/")
89 if err != nil || len(vs) != 0 {
90 t.Fatalf("staging residue: %+v %v", vs, err)
91 }
92 body, err := f.b.GetVersion(t.Context(), "data", liveKey, liveID)
93 if err != nil || string(body) != "live" {
94 t.Fatalf("live instance changed: %q %v", body, err)
95 }
96}
97
98func TestSweepJoinsRunningHolderlessOperation(t *testing.T) {
99 f := newFixture(t)
100 f.seed()
101 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
102 if err != nil {
103 t.Fatal(err)
104 }
105 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
106 c.Obligations = nil
107 return hostclaim.FreeAfterPark("op")(c)
108 })
109 if err != nil {
110 t.Fatal(err)
111 }
112 f.held.failed = true
113 if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit); !errors.Is(err, ErrChanged) {
114 t.Fatalf("first region queue: %v", err)
115 }
116 obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
117 if err != nil {
118 t.Fatal(err)
119 }
120 first := obs.Claim.Obligations[0]
121 second := f.e
122 second.Region = "uk"
123 b := leasetest.NewVersionedBucket()
124 b.AddBucket("data")
125 b.Put("data", f.ob.Park.Location+"/0000/data", []byte("second region"))
126 second.Store = b
127 second.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
128 return View{ClaimETag: o.ETag, Region: "uk"}, nil
129 }
130 second.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
131 if o.Op != first.Op || o.Region != "uk" {
132 return ErrUnsafe
133 }
134 return nil
135 }
136 if _, err := second.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit); !errors.Is(err, ErrChanged) {
137 t.Fatalf("second region did not join the running sweep: %v", err)
138 }
139 obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
140 if err != nil {
141 t.Fatal(err)
142 }
143 if obs.Claim.Holder != nil || obs.Claim.Op.ID != first.Op || len(obs.Claim.Obligations) != 2 || !reflect.DeepEqual(obs.Claim.Obligations[0], first) {
144 t.Fatalf("joined sweep: %+v", obs.Claim)
145 }
146 joined := obs.Claim.Obligations[1]
147 f.held.failed = false
148 if err := f.e.Run(t.Context(), Work{f.w.Host, first.ID, first.Op}); err != nil {
149 t.Fatal(err)
150 }
151 obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
152 if err != nil || obs.Claim.Free() || len(obs.Claim.Obligations) != 1 {
153 t.Fatalf("first completion freed an outstanding sweep: %+v %v", obs.Claim, err)
154 }
155 if err := second.Run(t.Context(), Work{f.w.Host, joined.ID, joined.Op}); err != nil {
156 t.Fatal(err)
157 }
158 obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
159 if err != nil || !obs.Claim.Free() {
160 t.Fatalf("last completion did not free the sweep: %+v %v", obs.Claim, err)
161 }
162 deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + f.ob.Park.RequestLimit)
163 for _, region := range []string{"ie", "uk"} {
164 if !obs.Claim.QuietAfter[region].Equal(deadline) {
165 t.Fatalf("quiet_after[%s]=%v, want %v", region, obs.Claim.QuietAfter[region], deadline)
166 }
167 }
168}
169func TestSweepProtectsUnfinishedRunAndPins(t *testing.T) {
170 f, obs, v, vs := sweepFixture(t)
171 d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
172 if err != nil || len(d.Alerts) != 0 || len(d.Candidates) != 1 || d.Candidates[0].Generation != "g0" {
173 t.Fatalf("%+v %v", d, err)
174 }
175 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
176 x := v
177 x.ClaimETag = o.ETag
178 return x, nil
179 }
180 if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); err != nil {
181 t.Fatal(err)
182 }
183 for _, x := range vs {
184 if strings.Contains(x.Key, "/g0/") {
185 continue
186 }
187 if _, err := f.b.GetVersion(t.Context(), "data", x.Key, x.VersionID); err != nil {
188 t.Fatalf("live version lost: %+v %v", x, err)
189 }
190 }
191}
192
193func TestSweepUnfinishedRunRegionalCoverage(t *testing.T) {
194 for _, scenario := range []string{"covered-foreign", "uncovered-foreign", "covered-local"} {
195 t.Run(scenario, func(t *testing.T) {
196 f, obs, v, _ := sweepFixture(t)
197 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
198 c.Holder.Admissions = []hostclaim.AdmissionRecord{
199 {Generation: "T", Predecessor: "g1", Region: "us", PredecessorRegion: "ie", Mode: "cold", Covered: scenario == "covered-local"},
200 {Generation: "U", Predecessor: "T", Region: "uk", PredecessorRegion: "us", Mode: "cold", Covered: scenario == "covered-foreign"},
201 {Generation: "g3", Predecessor: "U", Region: "ie", PredecessorRegion: "uk", Mode: "cold", Covered: scenario == "covered-foreign"},
202 }
203 return nil
204 })
205 if err != nil {
206 t.Fatal(err)
207 }
208 for _, g := range []string{"T", "U"} {
209 f.b.Put("data", f.ob.Park.Location+"/"+g+"/data", []byte(g))
210 }
211 before, err := f.e.Discover(t.Context(), f.w.Host)
212 if err != nil {
213 t.Fatal(err)
214 }
215 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
216 x := v
217 x.ClaimETag = o.ETag
218 return x, nil
219 }
220 snapshot := f.b.Snapshot()
221 d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second)
222 if err != nil {
223 t.Fatal(err)
224 }
225 if scenario == "covered-local" {
226 if len(d.Alerts) != 1 || len(d.Candidates) != 0 || f.b.Snapshot() != snapshot {
227 t.Fatalf("covered local baseline did not refuse unchanged: %+v", d)
228 }
229 return
230 }
231 var got []string
232 for _, c := range d.Candidates {
233 got = append(got, c.Generation)
234 }
235 if len(d.Alerts) != 0 || !reflect.DeepEqual(got, []string{"T", "U", "g0", "g2"}) {
236 t.Fatalf("foreign links blocked sweep or were protected: %+v", d)
237 }
238 for _, x := range before {
239 g, _ := v.GenerationOf(ref(x))
240 _, err := f.b.GetVersion(t.Context(), "data", x.Key, x.VersionID)
241 if g == "T" || g == "U" || g == "g0" || g == "g2" {
242 if !errors.Is(err, lease.ErrMirrorNotFound) {
243 t.Fatalf("dead version retained: %+v %v", x, err)
244 }
245 } else if err != nil {
246 t.Fatalf("live version lost: %+v %v", x, err)
247 }
248 }
249 completed, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
250 if err != nil || len(completed.Claim.Obligations) != 0 {
251 t.Fatalf("sweep did not complete: %+v %v", completed, err)
252 }
253 })
254 }
255}
256func TestSweepDisagreementAndBusyRefuse(t *testing.T) {
257 f, obs, v, vs := sweepFixture(t)
258 v.MarkerGeneration = "unrelated"
259 d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
260 if err != nil || len(d.Alerts) != 1 || len(d.Candidates) != 0 {
261 t.Fatalf("%+v %v", d, err)
262 }
263 obs.Claim.Op.Kind = hostclaim.KindMoving
264 obs.Claim.Op.ID = "moving"
265 d, err = Decide(obs, v, vs, f.ob.Park.Incarnation)
266 if err != nil || len(d.Candidates) != 0 {
267 t.Fatalf("%+v %v", d, err)
268 }
269}
270func TestSweepDiscoversOldNamespaceAndMoves(t *testing.T) {
271 f, obs, v, vs := sweepFixture(t)
272 old := strings.Repeat("a", 64)
273 other := strings.Repeat("b", 64)
274 for _, key := range []string{"instances/" + f.w.Host + "/" + old + "/0000/a", "moves/" + f.w.Host + "/" + other + "/g-op/a", "instances/" + f.w.Host + "/0000/legacy", "moves/" + f.w.Host + "/g-old/staging"} {
275 id := f.b.Put("data", key, []byte("old"))
276 vs = append(vs, lease.ObjectVersion{Key: key, VersionID: id})
277 }
278 d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
279 if err != nil || len(d.Candidates) != 4 {
280 t.Fatalf("%+v %v", d, err)
281 }
282 // Capped retired history is irrelevant to discovery.
283 obs.Claim.RetiredIncarnations = nil
284 v.Region = "us"
285 d, err = Decide(obs, v, vs, f.ob.Park.Incarnation)
286 if err != nil || len(d.Candidates) != 4 {
287 t.Fatalf("foreign location not residue: %+v %v", d, err)
288 }
289}
290func TestSweepFreeHostQueuesAndAcknowledges(t *testing.T) {
291 f := newFixture(t)
292 f.seed()
293 obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
294 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error { c.Obligations = nil; return hostclaim.FreeAfterPark("op")(c) })
295 if err != nil {
296 t.Fatal(err)
297 }
298 d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second)
299 if err != nil || len(d.Candidates) != 1 {
300 t.Fatalf("%+v %v", d, err)
301 }
302 obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
303 if err != nil || !obs.Claim.Free() || len(obs.Claim.Obligations) != 0 {
304 t.Fatalf("%+v %v", obs, err)
305 }
306}
307func TestSweepOperationOwnership(t *testing.T) {
308 for _, serving := range []bool{true, false} {
309 name := "holderless"
310 if serving {
311 name = "serving"
312 }
313 t.Run(name, func(t *testing.T) {
314 f, obs, v, _ := sweepFixture(t)
315 if !serving {
316 _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
317 c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "free"}
318 return hostclaim.FreeAfterPark("free")(c)
319 })
320 if err != nil {
321 t.Fatal(err)
322 }
323 }
324 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
325 x := v
326 x.ClaimETag = o.ETag
327 return x, nil
328 }
329 ids := []string{"generated-op", "chore", "retirement"}
330 wantOp, wantCalls := "generated-op", 3
331 if serving {
332 ids, wantOp, wantCalls = ids[1:], "session:3", 2
333 }
334 calls := 0
335 f.e.NewID = func() string {
336 if calls >= len(ids) {
337 t.Fatal("unexpected generated operation id")
338 }
339 id := ids[calls]
340 calls++
341 return id
342 }
343 authorized := false
344 f.e.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
345 authorized = true
346 if o.Op != wantOp {
347 return ErrUnsafe
348 }
349 return nil
350 }
351 // Stop after queueing, before Run can acknowledge the persisted chore.
352 f.held.failed = true
353 if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
354 t.Fatal(err)
355 }
356 if !authorized || calls != wantCalls {
357 t.Fatalf("authorized=%v ids=%d", authorized, calls)
358 }
359 queued, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
360 if err != nil {
361 t.Fatal(err)
362 }
363 if len(queued.Claim.Obligations) != 1 || queued.Claim.Obligations[0].Op != wantOp {
364 t.Fatalf("wrong chore owner: %+v", queued.Claim)
365 }
366 if serving {
367 if queued.Claim.Op.Kind != hostclaim.KindServing || queued.Claim.Holder.Session != 3 {
368 t.Fatal("serving state changed")
369 }
370 } else if queued.Claim.Holder != nil || queued.Claim.Op.ID != wantOp || queued.Claim.Op.Intent["park_sweep"] != "v1" {
371 t.Fatal("wrong holderless owner")
372 }
373 })
374 }
375}
376
377func TestSweepPageFailureAndStaleView(t *testing.T) {
378 f, obs, v, vs := sweepFixture(t)
379 before := f.b.Snapshot()
380 f.b.FailListAfterPages = 1
381 if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); err == nil {
382 t.Fatal("short sweep acted")
383 }
384 if before != f.b.Snapshot() {
385 t.Fatal("mutation")
386 }
387 v.ClaimETag = "old"
388 if _, err := Decide(obs, v, vs, f.ob.Park.Incarnation); !errors.Is(err, ErrUnsafe) {
389 t.Fatal(err)
390 }
391}
392