amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "os"
11 "path/filepath"
12 "runtime"
13 "strings"
14 "sync"
15 "testing"
16 "time"
17
18 "github.com/carlosframework/platform/internal/activator"
19 "github.com/carlosframework/platform/internal/hostclaim"
20 "github.com/carlosframework/platform/internal/hostwrite"
21 "github.com/carlosframework/platform/internal/registry"
22 "github.com/carlosframework/platform/internal/release"
23)
24
25// storesOf is a storeFor over fixed stores, one per bucket name.
26func storesOf(m map[string]release.Store) func(context.Context, string) (release.Store, error) {
27 return func(_ context.Context, bucket string) (release.Store, error) {
28 if s, ok := m[bucket]; ok {
29 return s, nil
30 }
31 return nil, fmt.Errorf("no store for bucket %q", bucket)
32 }
33}
34
35func testBuildStates(t *testing.T) (*buildStates, release.Store, *registry.Registry) {
36 t.Helper()
37 reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
38 if err != nil {
39 t.Fatal(err)
40 }
41 if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/carlos/a.sock"}); err != nil {
42 t.Fatal(err)
43 }
44 s := release.NewDirStore(t.TempDir())
45 return &buildStates{hibBucket: "hib", prefix: func(h string) string { return "inst/" + h },
46 storeFor: storesOf(map[string]release.Store{"hib": s}), reg: reg, optDir: t.TempDir(), bakeDir: t.TempDir(), now: time.Now}, s, reg
47}
48
49func copyOf(t *testing.T, reg *registry.Registry) release.BuildStateCopy {
50 t.Helper()
51 rt, _ := reg.Get("a.example.com")
52 c, err := release.ParseBuildStateCopy(rt.BuildState)
53 if err != nil {
54 t.Fatal(err)
55 }
56 return c
57}
58
59func TestRefreshRecordsNoStateAndNoReplica(t *testing.T) {
60 b, _, reg := testBuildStates(t)
61 if err := b.refresh(context.Background(), "a.example.com"); err != nil {
62 t.Fatal(err)
63 }
64 if c := copyOf(t, reg); !c.Known || c.Replica || c.State != nil || c.ReadErr != "" {
65 t.Fatalf("copy = %+v", c)
66 }
67}
68
69func TestRefreshSeesAReplicaAndAState(t *testing.T) {
70 b, s, reg := testBuildStates(t)
71 ctx := context.Background()
72 if _, err := s.Put(ctx, "inst/a.example.com/generations/0000/x.ltx", []byte("x"), ""); err != nil {
73 t.Fatal(err)
74 }
75 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1", Completed: 2}); err != nil {
76 t.Fatal(err)
77 }
78 if err := b.refresh(ctx, "a.example.com"); err != nil {
79 t.Fatal(err)
80 }
81 c := copyOf(t, reg)
82 if !c.Known || !c.Replica || c.State == nil || c.State.LastExecuted != "v1" || c.State.Completed != 2 {
83 t.Fatalf("copy = %+v", c)
84 }
85}
86
87// Astra I1 #6: a failed read keeps the last state and is recorded, and the
88// next good read clears the record.
89func TestAFailedRefreshIsRecorded(t *testing.T) {
90 b, s, reg := testBuildStates(t)
91 ctx := context.Background()
92 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
93 t.Fatal(err)
94 }
95 if err := b.refresh(ctx, "a.example.com"); err != nil {
96 t.Fatal(err)
97 }
98 key := "inst/a.example.com/build-state.json"
99 if _, err := s.Put(ctx, key, []byte("{"), mustEtag(t, s, key)); err != nil {
100 t.Fatal(err)
101 }
102 if err := b.refresh(ctx, "a.example.com"); err == nil {
103 t.Fatal("an unparseable build state refreshed cleanly")
104 }
105 c := copyOf(t, reg)
106 if c.ReadErr == "" || !c.Known || c.State == nil || c.State.LastExecuted != "v1" {
107 t.Fatalf("copy = %+v; want the last state kept and the failure recorded", c)
108 }
109 // update never writes over an object it cannot read: the merge would
110 // drop what that object recorded.
111 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err == nil {
112 t.Fatal("an update wrote over an unparseable build state")
113 }
114 if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v1" {
115 t.Fatalf("copy = %+v; a refused update changed it", c)
116 }
117 // Repaired by hand, the next good read clears the failure.
118 if _, err := s.Put(ctx, key, []byte(`{"schema":1,"host":"a.example.com","last_executed":"v2"}`), mustEtag(t, s, key)); err != nil {
119 t.Fatal(err)
120 }
121 if err := b.refresh(ctx, "a.example.com"); err != nil {
122 t.Fatal(err)
123 }
124 if c := copyOf(t, reg); c.ReadErr != "" || c.State == nil || c.State.LastExecuted != "v2" {
125 t.Fatalf("copy = %+v; want a good read to clear the failure", c)
126 }
127}
128
129// A host whose copy was never read stays unknown when its first read fails.
130func TestAFailedFirstRefreshStaysUnknown(t *testing.T) {
131 b, s, reg := testBuildStates(t)
132 ctx := context.Background()
133 if _, err := s.Put(ctx, "inst/a.example.com/build-state.json", []byte("{"), ""); err != nil {
134 t.Fatal(err)
135 }
136 if err := b.refresh(ctx, "a.example.com"); err == nil {
137 t.Fatal("an unparseable build state refreshed cleanly")
138 }
139 if c := copyOf(t, reg); c.Known || c.ReadErr == "" || c.State != nil {
140 t.Fatalf("copy = %+v; want unknown with the failure recorded", c)
141 }
142}
143
144func mustEtag(t *testing.T, s release.Store, key string) string {
145 t.Helper()
146 _, etag, err := s.Get(context.Background(), key)
147 if err != nil {
148 t.Fatal(err)
149 }
150 return etag
151}
152
153func TestALockAloneIsNotAReplica(t *testing.T) {
154 t.Parallel()
155 b, s, reg := testBuildStates(t)
156 ctx := context.Background()
157 if _, err := s.Put(ctx, "inst/a.example.com/lock.json", []byte("{}"), ""); err != nil {
158 t.Fatal(err)
159 }
160 if err := b.refresh(ctx, "a.example.com"); err != nil {
161 t.Fatal(err)
162 }
163 if copyOf(t, reg).Replica {
164 t.Fatal("a lease's lock was taken for a replica")
165 }
166}
167
168// A state naming another host is refused, not adopted.
169func TestAStateForAnotherHostIsAReadError(t *testing.T) {
170 b, s, _ := testBuildStates(t)
171 ctx := context.Background()
172 if _, err := s.Put(ctx, "inst/a.example.com/build-state.json", []byte(`{"schema":1,"host":"b.example.com","last_executed":"v1"}`), ""); err != nil {
173 t.Fatal(err)
174 }
175 if _, _, err := b.read(ctx, "a.example.com"); err == nil {
176 t.Fatal("a state naming another host was read as this one's")
177 }
178}
179
180// refreshAll covers the activator-managed exec instances and the standalone
181// workers (ruling R-WORKER-ROWS): every row whose spawns run under a lease
182// in the replica bucket.
183func TestRefreshAllCoversExecInstancesAndWorkers(t *testing.T) {
184 b, _, reg := testBuildStates(t)
185 for _, rt := range []registry.Route{
186 {Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
187 {Host: "always.example.com", Kind: "sidecar", App: "km"},
188 {Host: "unit.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/u.sock", Unit: "km-unit"},
189 {Host: "backed.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/b.sock", Backing: "unit"},
190 {Host: "backedw.example.com", Kind: "sidecar", App: "km", Hibernate: true, Backing: "unit"},
191 {Host: "awake.example.com", Kind: "instance", App: "km", Socket: "/run/w.sock"},
192 {Host: "plat.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/p.sock", Platform: true},
193 {Host: "svc.example.com", Kind: "service", App: "km", Hibernate: true, Socket: "/run/s.sock"},
194 } {
195 if err := reg.Upsert(rt); err != nil {
196 t.Fatal(err)
197 }
198 }
199 if skipped := b.refreshAll(context.Background()); skipped != 0 {
200 t.Fatalf("skipped %d", skipped)
201 }
202 for _, h := range []string{"a.example.com", "w.example.com", "always.example.com"} {
203 rt, err := reg.Get(h)
204 if err != nil || rt == nil {
205 t.Fatalf("%s: %v", h, err)
206 }
207 if c, _ := release.ParseBuildStateCopy(rt.BuildState); !c.Known {
208 t.Fatalf("%s was not refreshed: %+v", h, c)
209 }
210 }
211 for _, h := range []string{"unit.example.com", "backed.example.com", "backedw.example.com", "awake.example.com", "plat.example.com", "svc.example.com"} {
212 rt, err := reg.Get(h)
213 if err != nil || rt == nil {
214 t.Fatalf("%s: %v", h, err)
215 }
216 if rt.BuildState != "" {
217 t.Fatalf("%s got a build state copy: %s", h, rt.BuildState)
218 }
219 }
220}
221
222// With host claims on (settings plan 4 Task 5d, carried from Task 4) a
223// worker row never leases: the claim runtime refuses sidecar rows before
224// any lease (claimwakeassembly.go Prerequisites). refreshAll skips them, so
225// no claim is read for them and no copy is written; the claimed instance is
226// still refreshed through its claim.
227func TestRefreshAllSkipsWorkerRowsWithClaimsOn(t *testing.T) {
228 b, hib, reg := testBuildStates(t)
229 data := release.NewDirStore(t.TempDir())
230 b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": data})
231 b.claimBucket = "data"
232 var read []string
233 b.claims = func(_ context.Context, host string) (hostclaim.Observed, error) {
234 read = append(read, host)
235 return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: "instances/" + host, Generation: "7", Session: 3}}}, nil
236 }
237 for _, rt := range []registry.Route{
238 {Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
239 {Host: "always.example.com", Kind: "sidecar", App: "km"},
240 } {
241 if err := reg.Upsert(rt); err != nil {
242 t.Fatal(err)
243 }
244 }
245 if skipped := b.refreshAll(context.Background()); skipped != 0 {
246 t.Fatalf("skipped %d", skipped)
247 }
248 if c := copyOf(t, reg); !c.Known || c.ReadErr != "" {
249 t.Fatalf("the claimed instance was not refreshed: %+v", c)
250 }
251 if len(read) != 1 || read[0] != "a.example.com" {
252 t.Fatalf("claims read for %v, want the instance alone", read)
253 }
254 for _, h := range []string{"w.example.com", "always.example.com"} {
255 if rt, err := reg.Get(h); err != nil || rt == nil || rt.BuildState != "" {
256 t.Fatalf("%s: a worker row got a build state copy with claims on: %+v %v", h, rt, err)
257 }
258 }
259 // With claims off the same rows are refreshed (ruling R-WORKER-ROWS).
260 b.claims = nil
261 b.refreshAll(context.Background())
262 if rt, _ := reg.Get("w.example.com"); rt.BuildState == "" {
263 t.Fatal("claims off: the worker row was not refreshed")
264 }
265}
266
267// Ruling R-BS-REFRESH: refreshAll stops at its budget; the hosts it did not
268// reach keep their last copy and are counted.
269func TestRefreshAllStopsAtItsBudget(t *testing.T) {
270 b, _, reg := testBuildStates(t)
271 ctx := context.Background()
272 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
273 t.Fatal(err)
274 }
275 before := copyOf(t, reg)
276 if err := reg.Upsert(registry.Route{Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true}); err != nil {
277 t.Fatal(err)
278 }
279 b.budget = time.Nanosecond
280 if skipped := b.refreshAll(ctx); skipped != 2 {
281 t.Fatalf("skipped %d, want 2", skipped)
282 }
283 if after := copyOf(t, reg); after.Column() != before.Column() {
284 t.Fatalf("a host not reached changed its copy: %+v -> %+v", before, after)
285 }
286 if rt, _ := reg.Get("w.example.com"); rt.BuildState != "" {
287 t.Fatalf("a host not reached got a copy: %s", rt.BuildState)
288 }
289}
290
291// listCounter records how the replica was listed; hidePages hides the
292// bounded listing so the fallback is exercised.
293type listCounter struct {
294 release.Store
295 pages []int
296 full int
297}
298
299func (l *listCounter) List(ctx context.Context, prefix string) ([]string, error) {
300 l.full++
301 return l.Store.List(ctx, prefix)
302}
303
304type pagedCounter struct{ *listCounter }
305
306func (p pagedCounter) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
307 p.pages = append(p.pages, limit)
308 return p.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit)
309}
310
311// Ruling R-BS-REFRESH: three keys decide the replica, so a store with a
312// bounded listing is asked for three, never for the whole prefix; one
313// without it still answers through List.
314func TestTheReplicaIsDecidedByABoundedListing(t *testing.T) {
315 ctx := context.Background()
316 dir := release.NewDirStore(t.TempDir())
317 for _, k := range []string{"inst/a/build-state.json", "inst/a/lock.json", "inst/a/generations/0000/a.ltx", "inst/a/generations/0000/b.ltx", "inst/a/generations/0001/c.ltx"} {
318 if _, err := dir.Put(ctx, k, []byte("x"), ""); err != nil {
319 t.Fatal(err)
320 }
321 }
322 lc := &listCounter{Store: dir}
323 paged := pagedCounter{lc}
324 if ok, err := hasReplica(ctx, paged, "inst/a"); err != nil || !ok {
325 t.Fatalf("paged: %v, %v", ok, err)
326 }
327 if lc.full != 0 || len(lc.pages) != 1 || lc.pages[0] != 3 {
328 t.Fatalf("the whole prefix was listed %d times; pages asked %v, want [3]", lc.full, lc.pages)
329 }
330 if ok, err := hasReplica(ctx, lc, "inst/a"); err != nil || !ok || lc.full != 1 {
331 t.Fatalf("fallback: %v, %v, %d lists", ok, err, lc.full)
332 }
333 lockOnly := release.NewDirStore(t.TempDir())
334 for _, k := range []string{"inst/a/build-state.json", "inst/a/lock.json"} {
335 if _, err := lockOnly.Put(ctx, k, []byte("x"), ""); err != nil {
336 t.Fatal(err)
337 }
338 }
339 if ok, err := hasReplica(ctx, lockOnly, "inst/a"); err != nil || ok {
340 t.Fatalf("lock and state alone: %v, %v", ok, err)
341 }
342}
343
344// --- where the object lives (ruling R-CLAIMPREFIX) ---
345
346// Host claims off: the hibernate bucket, at the hibernate prefix.
347func TestClaimsOffUsesTheHibernatePrefix(t *testing.T) {
348 b, s, _ := testBuildStates(t)
349 ctx := context.Background()
350 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
351 t.Fatal(err)
352 }
353 if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); err != nil {
354 t.Fatalf("the build state is not beside the hibernate lease: %v", err)
355 }
356}
357
358// A location on the context (the wake's own lease) wins over both modes.
359func TestAContextLocationWins(t *testing.T) {
360 b, hib, reg := testBuildStates(t)
361 other := release.NewDirStore(t.TempDir())
362 b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": other})
363 b.claimBucket = "data"
364 b.claims = func(context.Context, string) (hostclaim.Observed, error) {
365 return hostclaim.Observed{}, errors.New("the claim must not be read when the wake names its lease")
366 }
367 ctx := activator.WithBuildStateLocation(context.Background(), activator.BuildStateLocation{Bucket: "data", Prefix: "instances/a.example.com/inc1"})
368 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
369 t.Fatal(err)
370 }
371 if _, _, err := other.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
372 t.Fatalf("the build state is not at the context's location: %v", err)
373 }
374 if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
375 t.Fatalf("the hibernate prefix was written too: %v", err)
376 }
377 if err := b.refresh(ctx, "a.example.com"); err != nil {
378 t.Fatal(err)
379 }
380 if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v1" {
381 t.Fatalf("copy = %+v", c)
382 }
383}
384
385func claimedAt(location string) func(context.Context, string) (hostclaim.Observed, error) {
386 return func(_ context.Context, host string) (hostclaim.Observed, error) {
387 return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: location, Generation: "7", Session: 3}}}, nil
388 }
389}
390
391// Host claims on: the claim actor's bucket, at the holder's Location, never
392// under the session's g-<gen>/ prefix; the generation's objects are the
393// replica.
394func TestClaimsOnUsesTheHolderLocationInTheActorBucket(t *testing.T) {
395 b, hib, reg := testBuildStates(t)
396 data := release.NewDirStore(t.TempDir())
397 b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": data})
398 b.claimBucket = "data"
399 b.claims = claimedAt("instances/a.example.com/inc1")
400 ctx := context.Background()
401 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
402 t.Fatal(err)
403 }
404 if _, _, err := data.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
405 t.Fatalf("the build state is not beside the claimed lease: %v", err)
406 }
407 if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
408 t.Fatalf("a claimed host's build state went to the hibernate prefix: %v", err)
409 }
410 if err := b.refresh(ctx, "a.example.com"); err != nil {
411 t.Fatal(err)
412 }
413 if c := copyOf(t, reg); !c.Known || c.Replica || c.State == nil || c.State.LastExecuted != "v1" {
414 t.Fatalf("copy = %+v", c)
415 }
416 if _, err := data.Put(ctx, "instances/a.example.com/inc1/g-7/0000.ltx", []byte("x"), ""); err != nil {
417 t.Fatal(err)
418 }
419 if err := b.refresh(ctx, "a.example.com"); err != nil {
420 t.Fatal(err)
421 }
422 if c := copyOf(t, reg); !c.Replica {
423 t.Fatalf("copy = %+v; want the generation's objects seen as the replica", c)
424 }
425}
426
427// A claimed host whose claim cannot be read (or has no holder) records the
428// failure and keeps its last state; it never falls back to the hibernate
429// prefix, even when an object sits there.
430func TestAnUnreadableClaimIsAReadError(t *testing.T) {
431 t.Parallel()
432 for name, read := range map[string]func(context.Context, string) (hostclaim.Observed, error){
433 "unreadable": func(context.Context, string) (hostclaim.Observed, error) {
434 return hostclaim.Observed{}, errors.New("bucket blip")
435 },
436 "absent": func(_ context.Context, h string) (hostclaim.Observed, error) {
437 return hostclaim.Observed{Host: h, Absent: true}, nil
438 },
439 "no holder": func(_ context.Context, h string) (hostclaim.Observed, error) {
440 return hostclaim.Observed{Host: h, Claim: &hostclaim.Claim{Host: h}}, nil
441 },
442 } {
443 t.Run(name, func(t *testing.T) {
444 b, hib, reg := testBuildStates(t)
445 data := release.NewDirStore(t.TempDir())
446 b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": data})
447 b.claimBucket = "data"
448 b.claims = claimedAt("instances/a.example.com")
449 ctx := context.Background()
450 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
451 t.Fatal(err)
452 }
453 if err := b.refresh(ctx, "a.example.com"); err != nil {
454 t.Fatal(err)
455 }
456 if _, err := hib.Put(ctx, "inst/a.example.com/build-state.json", []byte(`{"schema":1,"host":"a.example.com","last_executed":"hib"}`), ""); err != nil {
457 t.Fatal(err)
458 }
459 b.claims = read
460 if err := b.refresh(ctx, "a.example.com"); err == nil {
461 t.Fatal("a claimed host refreshed without its claim")
462 }
463 c := copyOf(t, reg)
464 if c.ReadErr == "" || !c.Known || c.State == nil || c.State.LastExecuted != "v1" {
465 t.Fatalf("copy = %+v; want the last state kept and the failure recorded", c)
466 }
467 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err == nil {
468 t.Fatal("a claimed host's build state was written without its claim")
469 }
470 })
471 }
472}
473
474// The production store opener: one store per known bucket, opened once,
475// each with its lease's region; an unknown bucket is refused.
476func TestStoresAreOpenedPerBucketAndCached(t *testing.T) {
477 opened := map[string]int{}
478 open := func(_ context.Context, bucket, region string) (release.Store, error) {
479 opened[bucket+"@"+region]++
480 return release.NewDirStore(t.TempDir()), nil
481 }
482 storeFor := cachedStores(map[string]string{"hib": "eu-west-1", "data": "us-east-1"}, open)
483 ctx := context.Background()
484 for _, bucket := range []string{"hib", "data", "hib", "data"} {
485 if _, err := storeFor(ctx, bucket); err != nil {
486 t.Fatal(err)
487 }
488 }
489 if len(opened) != 2 || opened["hib@eu-west-1"] != 1 || opened["data@us-east-1"] != 1 {
490 t.Fatalf("opened = %v", opened)
491 }
492 if _, err := storeFor(ctx, "elsewhere"); err == nil {
493 t.Fatal("a bucket the box does not lease in was opened")
494 }
495}
496
497func put(t *testing.T, b *buildStates, ctx context.Context, host string, s release.BuildState) error {
498 t.Helper()
499 _, err := b.update(ctx, host, func(*release.BuildState) release.BuildState { return s })
500 return err
501}
502
503// --- update (ruling R-BS-UPDATE) ---
504
505// bump is a merge that keeps what it read and raises Completed by one.
506func bump(cur *release.BuildState) release.BuildState {
507 next := release.BuildState{}
508 if cur != nil {
509 next = *cur
510 }
511 next.Completed++
512 return next
513}
514
515// Concurrent updates for one host (an instance spawn and its worker or
516// sidecar spawns, under the same lease) never lose a merge.
517func TestConcurrentUpdatesKeepEveryMerge(t *testing.T) {
518 b, _, reg := testBuildStates(t)
519 ctx := context.Background()
520 const n = 8
521 errs := make(chan error, n)
522 for i := 0; i < n; i++ {
523 go func() {
524 _, err := b.update(ctx, "a.example.com", bump)
525 errs <- err
526 }()
527 }
528 for i := 0; i < n; i++ {
529 if err := <-errs; err != nil {
530 t.Fatal(err)
531 }
532 }
533 st, _, err := b.read(ctx, "a.example.com")
534 if err != nil || st == nil || st.Completed != n {
535 t.Fatalf("state = %+v, %v; want Completed %d", st, err, n)
536 }
537 if c := copyOf(t, reg); !c.Known || !c.Replica || c.State == nil || c.State.Completed != n {
538 t.Fatalf("copy = %+v", c)
539 }
540}
541
542// A write that lands between update's read and its Put (another process)
543// is not overwritten: update re-reads and merges again.
544func TestAnUpdateRacedBetweenReadAndPutMergesAgain(t *testing.T) {
545 b, _, reg := testBuildStates(t)
546 ctx := context.Background()
547 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1", Completed: 1}); err != nil {
548 t.Fatal(err)
549 }
550 raced := false
551 b.beforePut = func(ctx context.Context, s release.Store, key string) {
552 if raced {
553 return
554 }
555 raced = true
556 other := []byte(`{"schema":1,"host":"a.example.com","last_executed":"v1","resolved":{"id":7,"target":"v3"},"completed":6}`)
557 if _, err := s.Put(ctx, key, other, mustEtag(t, s, key)); err != nil {
558 t.Fatal(err)
559 }
560 }
561 merges := 0
562 got, err := b.update(ctx, "a.example.com", func(cur *release.BuildState) release.BuildState {
563 merges++
564 next := *cur
565 next.LastExecuted = "v2"
566 return next
567 })
568 if err != nil {
569 t.Fatal(err)
570 }
571 if merges != 2 || got.LastExecuted != "v2" || got.Completed != 6 || got.Resolved == nil || got.Resolved.ID != 7 {
572 t.Fatalf("merges %d, got %+v; want the raced write's Completed and Resolved kept", merges, got)
573 }
574 st, _, err := b.read(ctx, "a.example.com")
575 if err != nil || st == nil || st.Completed != 6 || st.LastExecuted != "v2" {
576 t.Fatalf("stored %+v, %v", st, err)
577 }
578 if c := copyOf(t, reg); c.State == nil || c.State.Completed != 6 || c.State.LastExecuted != "v2" {
579 t.Fatalf("copy = %+v", c)
580 }
581}
582
583// A conflict on every try is an error, and the copy keeps what it had.
584func TestAPersistentConflictIsAnError(t *testing.T) {
585 b, _, reg := testBuildStates(t)
586 ctx := context.Background()
587 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
588 t.Fatal(err)
589 }
590 before := copyOf(t, reg)
591 n := 0
592 b.beforePut = func(ctx context.Context, s release.Store, key string) {
593 n++
594 other := []byte(fmt.Sprintf(`{"schema":1,"host":"a.example.com","last_executed":"v1","completed":%d}`, n))
595 if _, err := s.Put(ctx, key, other, mustEtag(t, s, key)); err != nil {
596 t.Fatal(err)
597 }
598 }
599 if _, err := b.update(ctx, "a.example.com", bump); !errors.Is(err, release.ErrCASConflict) {
600 t.Fatalf("err = %v; want a CAS conflict", err)
601 }
602 // The put above left this agent a copy of the object, so the first try
603 // goes on that copy (buildStates.write) and every fresh read still gets
604 // its own: updateTries of them after it.
605 if n != updateTries+1 {
606 t.Fatalf("tried %d times, want %d (the copy's try and %d fresh ones)", n, updateTries+1, updateTries)
607 }
608 if after := copyOf(t, reg); after.Column() != before.Column() {
609 t.Fatalf("a lost update changed the copy: %+v -> %+v", before, after)
610 }
611}
612
613// The first update on a host with no object is a create-only write.
614func TestTheFirstUpdateCreates(t *testing.T) {
615 b, s, _ := testBuildStates(t)
616 ctx := context.Background()
617 b.beforePut = func(ctx context.Context, st release.Store, key string) {
618 if _, _, err := st.Get(ctx, key); err == nil {
619 return
620 }
621 if _, err := st.Put(ctx, key, []byte(`{"schema":1,"host":"a.example.com","completed":4}`), ""); err != nil {
622 t.Fatal(err)
623 }
624 }
625 got, err := b.update(ctx, "a.example.com", bump)
626 if err != nil || got.Completed != 5 {
627 t.Fatalf("got %+v, %v; want the created object merged (Completed 5)", got, err)
628 }
629 if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); err != nil {
630 t.Fatal(err)
631 }
632}
633
634// Minor (fix round 1): with host claims on only the claim bucket is
635// registered, so a hibernate bucket that is the same bucket with no region
636// set cannot fail the box.
637func TestClaimsOnRegistersOnlyTheClaimBucket(t *testing.T) {
638 read := func(context.Context, string) (hostclaim.Observed, error) { return hostclaim.Observed{}, nil }
639 cfg := &activator.Config{Bucket: "data", Region: "", Prefix: "hib",
640 HostWrites: hostwrite.Config{Enabled: true},
641 Claims: &activator.ClaimRuntime{Bucket: "data", LeaseRegion: "eu-west-1", Read: read}}
642 regions, err := buildStateRegions(cfg)
643 if err != nil {
644 t.Fatal(err)
645 }
646 if len(regions) != 1 || regions["data"] != "eu-west-1" {
647 t.Fatalf("regions = %v", regions)
648 }
649 b, err := newBuildStates(cfg, nil, "")
650 if err != nil || b == nil || b.claimBucket != "data" || b.claims == nil {
651 t.Fatalf("newBuildStates = %+v, %v", b, err)
652 }
653 cfg.Claims.Bucket = "other"
654 if regions, err := buildStateRegions(cfg); err != nil || len(regions) != 1 || regions["other"] != "eu-west-1" {
655 t.Fatalf("regions = %v, %v; want only the claim bucket", regions, err)
656 }
657 cfg.HostWrites.Enabled = false
658 if regions, err := buildStateRegions(cfg); err != nil || len(regions) != 1 || regions["data"] != "" {
659 t.Fatalf("claims off: regions = %v, %v", regions, err)
660 }
661 cfg.HostWrites.Enabled, cfg.Claims = true, nil
662 if _, err := buildStateRegions(cfg); err == nil {
663 t.Fatal("claims on with no claim runtime was accepted")
664 }
665}
666
667// --- the pre-exec write (Service.BeforeExec, settings spec §4.2) ---
668
669// writeConsumed writes root's consumed file for the test host. An untagged
670// c is tagged with the route's tenant, as root writes every file since
671// ruling R22. Tests of an untagged or foreign file use writeConsumedAsIs.
672func writeConsumed(t *testing.T, b *buildStates, c release.Consumed) {
673 t.Helper()
674 c.Host = "a.example.com"
675 if rt, err := b.reg.Get(c.Host); err == nil && rt != nil && c.Account == "" && c.App == "" {
676 c.Account, c.App = release.RouteAccount(*rt), rt.App
677 }
678 if err := release.WriteConsumed(b.optDir, c); err != nil {
679 t.Fatal(err)
680 }
681}
682
683// The write before the exec names the build, carries root's resolution
684// from the consumed file, and completes the request when the build is its
685// target. The registry copy follows.
686func TestBeforeExecCompletesAResolvedRequestWhenItsTargetRuns(t *testing.T) {
687 b, _, reg := testBuildStates(t)
688 ctx := context.Background()
689 writeConsumed(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 7, Target: "v2"}, Completed: 3})
690 if err := b.beforeExec(ctx, "a.example.com", "v1"); err != nil {
691 t.Fatal(err)
692 }
693 st, _, err := b.read(ctx, "a.example.com")
694 if err != nil {
695 t.Fatal(err)
696 }
697 if st.LastExecuted != "v1" || st.Resolved == nil || st.Resolved.ID != 7 || st.Resolved.Target != "v2" || st.Completed != 3 {
698 t.Fatalf("state after v1 = %+v, want the resolution carried and R pending", st)
699 }
700 if err := b.beforeExec(ctx, "a.example.com", "v2"); err != nil {
701 t.Fatal(err)
702 }
703 st, _, _ = b.read(ctx, "a.example.com")
704 if st.LastExecuted != "v2" || st.Completed != 7 || st.CompletedFailed {
705 t.Fatalf("state after T = %+v, want R completed", st)
706 }
707 if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v2" || c.State.Completed != 7 || !c.Known {
708 t.Fatalf("registry copy = %+v", c)
709 }
710}
711
712// Fail closed: a consumed file the agent cannot read refuses the exec, and
713// nothing is written. So does a build with no name.
714func TestBeforeExecRefusesAnUnreadableConsumedFile(t *testing.T) {
715 b, s, _ := testBuildStates(t)
716 ctx := context.Background()
717 if err := os.MkdirAll(filepath.Join(b.optDir, "consumed"), 0o755); err != nil {
718 t.Fatal(err)
719 }
720 if err := os.WriteFile(release.ConsumedPath(b.optDir, "a.example.com"), []byte("{not json"), 0o644); err != nil {
721 t.Fatal(err)
722 }
723 if err := b.beforeExec(ctx, "a.example.com", "v1"); err == nil {
724 t.Fatal("an unreadable consumed file let the exec through")
725 }
726 if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
727 t.Fatalf("a build state was written without root's resolution: %v", err)
728 }
729 if err := os.Remove(release.ConsumedPath(b.optDir, "a.example.com")); err != nil {
730 t.Fatal(err)
731 }
732 if err := b.beforeExec(ctx, "a.example.com", ""); err == nil {
733 t.Fatal("a build with no name was recorded")
734 }
735}
736
737// The write lands at the location the wake's context names (its lease).
738func TestBeforeExecWritesAtTheContextLocation(t *testing.T) {
739 b, hib, _ := testBuildStates(t)
740 other := release.NewDirStore(t.TempDir())
741 b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": other})
742 ctx := activator.WithBuildStateLocation(context.Background(), activator.BuildStateLocation{Bucket: "data", Prefix: "instances/a.example.com/inc1"})
743 if err := b.beforeExec(ctx, "a.example.com", "v1"); err != nil {
744 t.Fatal(err)
745 }
746 if _, _, err := other.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
747 t.Fatalf("the build state is not at the context's location: %v", err)
748 }
749 if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
750 t.Fatalf("the hibernate prefix was written too: %v", err)
751 }
752}
753
754// Ruling R-BS-CORRUPT: an object that does not parse is never overwritten;
755// the refusal names the key and says what to do.
756func TestBeforeExecRefusesACorruptStateAndNamesIt(t *testing.T) {
757 b, s, _ := testBuildStates(t)
758 ctx := context.Background()
759 key := "inst/a.example.com/build-state.json"
760 if _, err := s.Put(ctx, key, []byte("{torn"), ""); err != nil {
761 t.Fatal(err)
762 }
763 err := b.beforeExec(ctx, "a.example.com", "v1")
764 if err == nil || !strings.Contains(err.Error(), key) || !strings.Contains(err.Error(), "repair or delete") {
765 t.Fatalf("err = %v, want a refusal naming %s and saying to repair or delete it", err, key)
766 }
767 if raw, _, _ := s.Get(ctx, key); string(raw) != "{torn" {
768 t.Fatalf("the corrupt object was overwritten: %q", raw)
769 }
770}
771
772// Astra I3 #1: an initialization failure never leaves the nil hook.
773func TestAFailedBuildStateStartRefusesEveryExec(t *testing.T) {
774 hook := buildStateHook(nil, errors.New("s3: no credentials"))
775 if hook == nil {
776 t.Fatal("an initialization failure installed the nil hook: execs would run unrecorded")
777 }
778 if err := hook(context.Background(), "a.example.com", "v1"); err == nil {
779 t.Fatal("the refusing hook let an exec through")
780 }
781 if buildStateHook(nil, nil) != nil {
782 t.Fatal("a box with no replica bucket must keep today's nil hook")
783 }
784 b, _, _ := testBuildStates(t)
785 if hook := buildStateHook(b, nil); hook == nil || hook(context.Background(), "a.example.com", "v1") != nil {
786 t.Fatal("a started store's hook does not record")
787 }
788}
789
790// --- the held wake check's answer (Service.HeldTarget, settings spec §4.3
791// rule 3) ---
792
793func setSettings(t *testing.T, reg *registry.Registry, u release.UpdateSettings) {
794 t.Helper()
795 if err := reg.SetUpdateSettings("a.example.com", u.Column()); err != nil {
796 t.Fatal(err)
797 }
798}
799
800func setCopy(t *testing.T, reg *registry.Registry, c release.BuildStateCopy) {
801 t.Helper()
802 if err := reg.SetBuildState("a.example.com", c.Column()); err != nil {
803 t.Fatal(err)
804 }
805}
806
807// stageWithManifest stages body as acct/app at ver, with its manifest
808// sibling naming body's sha256 for this platform, as root's staging leaves
809// a verified copy.
810func stageWithManifest(t *testing.T, optDir, acct, app, ver string, body []byte) {
811 t.Helper()
812 if err := release.StageArtifact(optDir, acct, ver, app, body); err != nil {
813 t.Fatal(err)
814 }
815 sum := sha256.Sum256(body)
816 m := release.Manifest{Schema: 1, App: app, Version: ver, Kind: "binary",
817 Artifacts: map[string]release.Artifact{runtime.GOOS + "-" + runtime.GOARCH: {File: app, SHA256: hex.EncodeToString(sum[:]), Size: int64(len(body))}}}
818 raw, err := json.Marshal(m)
819 if err != nil {
820 t.Fatal(err)
821 }
822 if err := release.StageManifest(optDir, acct, ver, app, raw); err != nil {
823 t.Fatal(err)
824 }
825}
826
827func fileSHA(t *testing.T, path string) string {
828 t.Helper()
829 sum, err := release.FileSHA256(path)
830 if err != nil {
831 t.Fatal(err)
832 }
833 return sum
834}
835
836// Astra I1 #8, the agent's half: root reports a major hold with a request
837// baking beside it (Consumed.Baking); the wake check still treats the
838// instance as held, on last_executed.
839func TestAMajorHoldStaysHeldWhileARequestBakes(t *testing.T) {
840 b, _, reg := testBuildStates(t)
841 two := 2
842 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Major: &two, InstallRequest: 1})
843 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
844 decided(t, b, release.Consumed{Reason: string(release.ReasonMajor), Desired: "v2", Baking: "v3"})
845 if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v2" {
846 t.Fatalf("heldTarget = %+v; want v2, held", h)
847 }
848}
849
850// Astra I2 #7: a pin set after a request resolved outranks it, as in root's
851// Target. The wake targets the pin P, never the request's T.
852func TestAPinOutranksAPendingRequestAtTheWake(t *testing.T) {
853 b, _, reg := testBuildStates(t)
854 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", Pin: "v5", InstallRequest: 2})
855 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2", Completed: 1}})
856 writeConsumed(t, b, release.Consumed{Completed: 1, Resolved: &release.ResolvedRequest{ID: 2, Target: "v4"}})
857 if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v5" {
858 t.Fatalf("heldTarget = %+v; want the pin v5", h)
859 }
860 // Without the pin, the pending request's T outranks last_executed.
861 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", InstallRequest: 2})
862 if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v4" {
863 t.Fatalf("heldTarget = %+v; want the request's v4", h)
864 }
865}
866
867// Ruling R-RESOLUTION-STATE-WINS: when root's file and the build state
868// resolve the same request to different builds (a box that resolved R, lost
869// the instance to a move before any wake, and got it back after another box
870// resolved R again), the wake targets the build state's T, as root does.
871func TestAHeldWakeFollowsTheBuildStatesResolutionOverTheFile(t *testing.T) {
872 b, _, reg := testBuildStates(t)
873 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", InstallRequest: 1})
874 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true,
875 State: &release.BuildState{LastExecuted: "v1", Resolved: &release.ResolvedRequest{ID: 1, Target: "v4"}}})
876 writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonRequest), Desired: "v1", Resolved: &release.ResolvedRequest{ID: 1, Target: "v1"}})
877 if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v4" {
878 t.Fatalf("heldTarget = %+v; want the build state's v4, held", h)
879 }
880}
881
882// Task 2 review: with the build state unread (Known false) root does no
883// request bookkeeping and will not install T, so neither does the wake
884// follow it. Held settings then read root's unknown; auto ones are not held
885// (root's decisions here are for the current column, ruling R-HELD-PURE). A
886// copy whose last refresh failed is refused as unread, whatever the settings
887// (Astra round 1 finding 1), and never follows T either.
888func TestAnUnreadBuildStateDoesNotFollowAPendingRequest(t *testing.T) {
889 b, _, reg := testBuildStates(t)
890 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", InstallRequest: 2})
891 setCopy(t, reg, release.BuildStateCopy{Known: false})
892 decided(t, b, release.Consumed{Reason: string(release.ReasonUnknown), Resolved: &release.ResolvedRequest{ID: 2, Target: "v4"}})
893 if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown || h.Target != "" {
894 t.Fatalf("heldTarget = %+v; want held, unknown, never the request's v4", h)
895 }
896 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", InstallRequest: 2})
897 decided(t, b, release.Consumed{Reason: string(release.ReasonAdopt), Desired: "v5", Resolved: &release.ResolvedRequest{ID: 2, Target: "v4"}})
898 if h := b.heldTarget("a.example.com"); h.Held {
899 t.Fatalf("heldTarget = %+v; an auto instance with an unread state follows root's adoption", h)
900 }
901 setCopy(t, reg, release.BuildStateCopy{Known: false, ReadErr: "s3: timeout"})
902 if h := b.heldTarget("a.example.com"); !h.Held || !h.Unread || h.Target != "" {
903 t.Fatalf("heldTarget = %+v; a failed refresh is refused as unread, never the request's v4", h)
904 }
905}
906
907// Ruling R-SETTINGS-UNREADABLE (Task 6): a column that does not parse even
908// leniently holds the wake as unknown when root says unknown for it. With no
909// word from root for that column the wake holds and asks root (ruling
910// R-HELD-PURE: never "not held"). A Control name this binary does not know
911// is dropped, not read as unreadable, so the hold stays.
912func TestAnUnreadableColumnHoldsOnlyAsRootDoes(t *testing.T) {
913 b, _, reg := testBuildStates(t)
914 if err := reg.SetUpdateSettings("a.example.com", `{"updates":"sometimes"}`); err != nil {
915 t.Fatal(err)
916 }
917 if h := b.heldTarget("a.example.com"); !h.Held || !h.AskRoot || h.Unknown {
918 t.Fatalf("heldTarget = %+v with no word from root; want held, asking root", h)
919 }
920 decided(t, b, release.Consumed{Reason: string(release.ReasonUnknown), Detail: "update settings: updates \"sometimes\""})
921 if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown {
922 t.Fatalf("heldTarget = %+v; want held as unknown, as root holds it", h)
923 }
924 if err := reg.SetUpdateSettings("a.example.com", `{"updates":"manual","channel":"stable","control":["updates","colour"]}`); err != nil {
925 t.Fatal(err)
926 }
927 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
928 decided(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
929 if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v2" {
930 t.Fatalf("heldTarget = %+v; want the manual hold on v2 through an unknown Control name", h)
931 }
932}
933
934// Astra I3 #2: root's failed decision, or its refusal record, marks the
935// target failed, so the wake's own rule refuses it with no planner. Only a
936// not-up refusal is a build failure (ruling R-REFUSALKIND).
937func TestARootFailedTargetIsReportedFailed(t *testing.T) {
938 b, _, reg := testBuildStates(t)
939 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Pin: "v5"})
940 writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonFailed), Detail: "v5"})
941 stageWithManifest(t, b.optDir, "bab", "km", "v5", []byte("five"))
942 if h := b.heldTarget("a.example.com"); !h.Held || !h.Failed || h.Target != "v5" || h.Staged != nil {
943 t.Fatalf("heldTarget = %+v; want v5 held and failed, no retained copy offered", h)
944 }
945 writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonPin), Detail: "v5"})
946 for _, kind := range []string{release.RefusalWrongProgram, release.RefusalNotProven} {
947 if err := release.WriteRefusal(b.bakeDir, release.Refusal{Kind: kind, Host: "a.example.com", App: "km", Version: "v5", Reason: "refused"}); err != nil {
948 t.Fatal(err)
949 }
950 if h := b.heldTarget("a.example.com"); h.Failed {
951 t.Fatalf("a %s refusal of the pin = %+v; it is not a build failure", kind, h)
952 }
953 }
954 if err := release.WriteRefusal(b.bakeDir, release.Refusal{Host: "a.example.com", App: "km", Version: "v5", Reason: "not up"}); err != nil {
955 t.Fatal(err)
956 }
957 if h := b.heldTarget("a.example.com"); !h.Failed || h.Staged != nil {
958 t.Fatalf("a not-up refusal record for the pin = %+v; want failed", h)
959 }
960}
961
962// Astra I3 #5: a retained copy whose bytes match its staged manifest is
963// offered; one whose bytes differ is not.
964func TestARetainedCopyIsOfferedOnlyWhenVerified(t *testing.T) {
965 b, _, reg := testBuildStates(t)
966 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"})
967 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
968 decided(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
969 stageWithManifest(t, b.optDir, "bab", "km", "v2", []byte("two"))
970 h := b.heldTarget("a.example.com")
971 if !h.Held || h.Target != "v2" || h.Staged == nil || h.Staged.Version != "v2" || h.Staged.StagedPath != release.VersionPath(b.optDir, "bab", "v2", "km") || h.Staged.Digest != fileSHA(t, h.Staged.StagedPath) {
972 t.Fatalf("heldTarget = %+v; want the verified retained v2", h)
973 }
974 if err := os.WriteFile(release.VersionPath(b.optDir, "bab", "v2", "km"), []byte("tampered"), 0o755); err != nil {
975 t.Fatal(err)
976 }
977 if h := b.heldTarget("a.example.com"); h.Staged != nil {
978 t.Fatal("a staged copy whose bytes differ from its manifest was offered")
979 }
980}
981
982// Default settings are today's wake: not held, and nothing is read for it.
983// Held settings refresh the copy first (§4.2: "refreshed ... before every
984// wake"), so a last_executed the reconcile has not copied yet still counts.
985func TestTheWakeTargetRefreshesOnlyAHeldRoute(t *testing.T) {
986 b, s, reg := testBuildStates(t)
987 ctx := context.Background()
988 if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err != nil {
989 t.Fatal(err)
990 }
991 setCopy(t, reg, release.BuildStateCopy{}) // what the last reconcile saw: nothing yet
992 gets := &getCounter{Store: s}
993 b.storeFor = storesOf(map[string]release.Store{"hib": gets})
994 if h := b.wakeTarget(ctx, "a.example.com"); h.Held || gets.n() != 0 {
995 t.Fatalf("default settings: %+v after %d reads; want today's wake, nothing read", h, gets.n())
996 }
997 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"})
998 writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
999 if h := b.wakeTarget(ctx, "a.example.com"); !h.Held || h.Target != "v2" || gets.n() == 0 {
1000 t.Fatalf("held: %+v after %d reads; want the refreshed last_executed v2", h, gets.n())
1001 }
1002}
1003
1004// Ruling R-HELD-SCOPE (Task 12a): settings hold only the rows root applies
1005// them to (Kind instance, no Unit; adopt.go's settings condition). A named
1006// unit row and an exec worker row under an app-policy manual wake as today:
1007// not held, and nothing is read for them. Before, the unit row read
1008// "build unknown" on every wake with nothing to clear it, and the worker was
1009// held on a build root never targets.
1010func TestOnlyRowsRootHoldsAreHeldAtTheWake(t *testing.T) {
1011 b, s, reg := testBuildStates(t)
1012 ctx := context.Background()
1013 manual := release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"}
1014 for _, rt := range []registry.Route{
1015 {Host: "unit.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/u.sock", Unit: "km-unit"},
1016 {Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
1017 } {
1018 if err := reg.Upsert(rt); err != nil {
1019 t.Fatal(err)
1020 }
1021 if err := reg.SetUpdateSettings(rt.Host, manual.Column()); err != nil {
1022 t.Fatal(err)
1023 }
1024 }
1025 // The worker keeps a build state (its lease's), read and copied.
1026 if err := put(t, b, ctx, "w.example.com", release.BuildState{LastExecuted: "v2"}); err != nil {
1027 t.Fatal(err)
1028 }
1029 if err := reg.SetBuildState("w.example.com", release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}}.Column()); err != nil {
1030 t.Fatal(err)
1031 }
1032 gets := &getCounter{Store: s}
1033 b.storeFor = storesOf(map[string]release.Store{"hib": gets})
1034 for _, host := range []string{"unit.example.com", "w.example.com"} {
1035 if h := b.heldTarget(host); h.Held || h.Unknown || h.Target != "" {
1036 t.Errorf("%s: heldTarget = %+v; want today's wake", host, h)
1037 }
1038 if h := b.wakeTarget(ctx, host); h.Held || h.Unknown || h.Target != "" {
1039 t.Errorf("%s: wakeTarget = %+v; want today's wake", host, h)
1040 }
1041 }
1042 if n := gets.n(); n != 0 {
1043 t.Errorf("%d bucket reads for rows root never holds; want none", n)
1044 }
1045 // The instance row under the same settings is still held.
1046 setSettings(t, reg, manual)
1047 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v3"}})
1048 if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v3" {
1049 t.Fatalf("the instance: heldTarget = %+v; want held on v3", h)
1050 }
1051}
1052
1053type getCounter struct {
1054 release.Store
1055 mu sync.Mutex
1056 gets int
1057}
1058
1059func (g *getCounter) Get(ctx context.Context, key string) ([]byte, string, error) {
1060 g.mu.Lock()
1061 g.gets++
1062 g.mu.Unlock()
1063 return g.Store.Get(ctx, key)
1064}
1065
1066func (g *getCounter) n() int {
1067 g.mu.Lock()
1068 defer g.mu.Unlock()
1069 return g.gets
1070}
1071
1072// Ruling R-HELD-FAILED: root's rule 3 rewrites a major hold whose
1073// last_executed this box holds as failed to Reason failed, Detail that
1074// build. The wake still reads the major hold: held on it, and failed, so
1075// with no planner the wake is refused rather than run the live link.
1076func TestAMajorHoldOnAFailedBuildStaysHeldAndFailed(t *testing.T) {
1077 b, _, reg := testBuildStates(t)
1078 two := 2
1079 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Major: &two})
1080 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
1081 decided(t, b, release.Consumed{Reason: string(release.ReasonFailed), Detail: "v2"})
1082 if h := b.heldTarget("a.example.com"); !h.Held || !h.Failed || h.Target != "v2" {
1083 t.Fatalf("heldTarget = %+v; want v2 held and failed", h)
1084 }
1085 // Root's failed head under rule 4 (Detail is the head, not
1086 // last_executed) is not a hold.
1087 decided(t, b, release.Consumed{Reason: string(release.ReasonFailed), Detail: "v3"})
1088 if h := b.heldTarget("a.example.com"); h.Held {
1089 t.Fatalf("heldTarget = %+v; a failed head under adoption is not a hold", h)
1090 }
1091}
1092
1093// Ruling R-HELD-UNKNOWN: rule 3 as root's Target decides it. An unread copy
1094// is unknown whatever root last wrote; a read copy with last_executed
1095// targets it over a stale unknown; root's unknown stands for a read copy
1096// with no last_executed.
1097func TestAHeldWakeReadsTheCopyAsRootsTargetDoes(t *testing.T) {
1098 b, _, reg := testBuildStates(t)
1099 setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"})
1100 setCopy(t, reg, release.BuildStateCopy{}) // never read
1101 decided(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
1102 if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown {
1103 t.Fatalf("an unread copy under manual = %+v; want held, unknown", h)
1104 }
1105 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
1106 decided(t, b, release.Consumed{Reason: string(release.ReasonUnknown)}) // written while the copy was unread
1107 if h := b.heldTarget("a.example.com"); !h.Held || h.Unknown || h.Target != "v2" {
1108 t.Fatalf("a read copy with last_executed under root's stale unknown = %+v; want held on v2", h)
1109 }
1110 setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true})
1111 if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown {
1112 t.Fatalf("a read copy with no last_executed and root's unknown = %+v; want unknown", h)
1113 }
1114}
1115
1116// budgetEater records which host's build state each read asks for and
1117// holds the read until the caller's context ends: the host it is reading
1118// spends the whole refresh budget.
1119type budgetEater struct {
1120 release.Store
1121 mu sync.Mutex
1122 hosts []string
1123}
1124
1125func (e *budgetEater) Get(ctx context.Context, key string) ([]byte, string, error) {
1126 if host, ok := strings.CutSuffix(strings.TrimPrefix(key, "inst/"), "/build-state.json"); ok {
1127 e.mu.Lock()
1128 e.hosts = append(e.hosts, host)
1129 e.mu.Unlock()
1130 <-ctx.Done()
1131 return nil, "", ctx.Err()
1132 }
1133 return e.Store.Get(ctx, key)
1134}
1135
1136// Ruling R-REFRESH-ROTATE: a box whose refresh budget runs out does not
1137// starve the same tail every pass. Each refreshAll starts at the first host
1138// the last one did not reach, so every host is reached in turn.
1139func TestRefreshAllRotatesPastWhereTheLastOneStopped(t *testing.T) {
1140 t.Parallel()
1141 b, s, reg := testBuildStates(t)
1142 for _, h := range []string{"b.example.com", "c.example.com"} {
1143 if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", App: "km", Hibernate: true, Socket: "/run/carlos/" + h + ".sock"}); err != nil {
1144 t.Fatal(err)
1145 }
1146 }
1147 eater := &budgetEater{Store: s}
1148 b.storeFor = storesOf(map[string]release.Store{"hib": eater})
1149 b.budget = 200 * time.Millisecond
1150 for i := 0; i < 4; i++ {
1151 if skipped := b.refreshAll(context.Background()); skipped != 2 {
1152 t.Fatalf("pass %d skipped %d, want 2 (one host eats the budget)", i, skipped)
1153 }
1154 }
1155 want := []string{"a.example.com", "b.example.com", "c.example.com", "a.example.com"}
1156 eater.mu.Lock()
1157 defer eater.mu.Unlock()
1158 if strings.Join(eater.hosts, " ") != strings.Join(want, " ") {
1159 t.Fatalf("passes read %v, want %v: each pass starts where the last stopped", eater.hosts, want)
1160 }
1161}
1162