amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package provision
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "go/ast"
9 "go/parser"
10 "go/token"
11 "os"
12 "path/filepath"
13 "strings"
14 "testing"
15 "time"
16
17 "github.com/carlosframework/platform/internal/release"
18 "github.com/carlosframework/platform/internal/runs"
19 "github.com/carlosframework/platform/internal/seeds"
20)
21
22// keyed is caller A's admission that carries key k and digest d.
23func keyed(id, k, d string) RunAdmission { return keyedBy("A", id, k, d) }
24
25// keyedBy is keyed for caller.
26func keyedBy(caller, id, k, d string) RunAdmission {
27 return RunAdmission{ID: id, By: "svc", Deadline: time.Now().Add(time.Minute), Key: k, Digest: d, Caller: caller}
28}
29
30func wantReplay(t *testing.T, what string, err error, id string) {
31 t.Helper()
32 var replay *RunKeyReplay
33 if !errors.As(err, &replay) || replay.RunID != id {
34 t.Fatalf("%s: %v, want a RunKeyReplay of %s", what, err, id)
35 }
36}
37
38func TestRunAdmitRecordsAndReplaysAKey(t *testing.T) {
39 s := release.NewDirStore(t.TempDir())
40 rec := runTestRecord(t, s)
41 got, err := setRun(t, s, rec, RunAdmit(keyed("run_a", "K", "D")))
42 if err != nil || got.ActiveRun.ID != "run_a" {
43 t.Fatalf("keyed admit: %+v, %v", got, err)
44 }
45 if len(got.RunKeys) != 1 || got.RunKeys[0].Key != "K" || got.RunKeys[0].RunID != "run_a" || got.RunKeys[0].Digest != "D" || got.RunKeys[0].At.IsZero() {
46 t.Fatalf("the admission must record its key: %+v", got.RunKeys)
47 }
48
49 // While run_a is still open, the replay is answered before the open-run
50 // refusal.
51 _, err = setRun(t, s, rec, RunAdmit(keyed("run_b", "K", "D")))
52 wantReplay(t, "the same key and digest while open", err, "run_a")
53 if _, err := setRun(t, s, rec, RunAdmit(keyed("run_b", "K", "D2"))); !errors.Is(err, ErrRunKeyReused) {
54 t.Fatalf("the same key with another digest: %v, want ErrRunKeyReused", err)
55 }
56 // Another key still meets the open run.
57 if _, err := setRun(t, s, rec, RunAdmit(keyed("run_b", "K2", "D"))); !errors.Is(err, ErrRunOpen) {
58 t.Fatalf("another key while open: %v, want ErrRunOpen", err)
59 }
60
61 // Picked, finished and cleared (the ack): the key still replays.
62 for _, step := range []func(*Record) error{
63 RunPick("run_a", "box-1", "boot-1", time.Now().Add(time.Hour)),
64 RunFinish("run_a"),
65 RunClear("run_a"),
66 } {
67 if _, err := setRun(t, s, rec, step); err != nil {
68 t.Fatal(err)
69 }
70 _, err = setRun(t, s, rec, RunAdmit(keyed("run_b", "K", "D")))
71 wantReplay(t, "after a run transition", err, "run_a")
72 }
73
74 // A later, unkeyed run replaces it in the record; the key still replays.
75 if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_c", Deadline: time.Now().Add(time.Minute)})); err != nil {
76 t.Fatal(err)
77 }
78 if _, err := setRun(t, s, rec, RunExpire("run_c")); err != nil {
79 t.Fatal(err)
80 }
81 _, err = setRun(t, s, rec, RunAdmit(keyed("run_d", "K", "D")))
82 wantReplay(t, "after a later run replaced it", err, "run_a")
83
84 // A refused replay wrote nothing.
85 cur, err := GetRecord(context.Background(), s, rec.Account, rec.App, rec.Host)
86 if err != nil || cur.ActiveRun.ID != "run_c" || len(cur.RunKeys) != 1 {
87 t.Fatalf("record after the replays: run %+v keys %+v, %v", cur.ActiveRun, cur.RunKeys, err)
88 }
89}
90
91// putRunKeys installs keys on the record directly, as an earlier console's
92// admissions would have left them.
93func putRunKeys(t *testing.T, s release.Store, rec Record, keys []RunKey) {
94 t.Helper()
95 if _, err := updateRecord(context.Background(), s, rec.Account, rec.App, rec.Host, func(r *Record) error {
96 r.RunKeys = keys
97 return nil
98 }); err != nil {
99 t.Fatal(err)
100 }
101}
102
103func TestRunKeysPrunedByAgeAndCount(t *testing.T) {
104 s := release.NewDirStore(t.TempDir())
105 rec := runTestRecord(t, s)
106 now := time.Now().UTC()
107 var keys []RunKey
108 keys = append(keys, RunKey{Key: "old", RunID: "run_old", Digest: "D", Caller: "A", At: now.Add(-runs.KeyTTL - time.Minute)})
109 for i := 0; i < runs.MaxRunKeysPerCaller+2; i++ {
110 keys = append(keys, RunKey{Key: fmt.Sprintf("k%03d", i), RunID: fmt.Sprintf("run_%03d", i), Digest: "D", Caller: "A", At: now.Add(-time.Hour + time.Duration(i)*time.Second)})
111 }
112 putRunKeys(t, s, rec, keys)
113
114 // An expired key is not a replay: it admits a fresh run.
115 got, err := setRun(t, s, rec, RunAdmit(keyed("run_new", "old", "D")))
116 if err != nil || got.ActiveRun.ID != "run_new" {
117 t.Fatalf("an expired key must admit afresh: %+v, %v", got, err)
118 }
119 if len(got.RunKeys) != runs.MaxRunKeysPerCaller {
120 t.Fatalf("%d keys kept, want %d", len(got.RunKeys), runs.MaxRunKeysPerCaller)
121 }
122 kept := map[string]bool{}
123 for _, k := range got.RunKeys {
124 kept[k.Key+"/"+k.RunID] = true
125 }
126 if kept["old/run_old"] {
127 t.Error("the entry older than KeyTTL survived")
128 }
129 if !kept["old/run_new"] {
130 t.Error("the newest entry was not kept")
131 }
132 // 34 fresh + 1 new, the caller's newest 32 kept: k000, k001 and k002 go.
133 for i := 0; i < runs.MaxRunKeysPerCaller+2; i++ {
134 name := fmt.Sprintf("k%03d/run_%03d", i, i)
135 if want := i >= 3; kept[name] != want {
136 t.Errorf("%s kept %v, want %v", name, kept[name], want)
137 }
138 }
139
140 // Under the count, age alone prunes.
141 s2 := release.NewDirStore(t.TempDir())
142 rec2 := runTestRecord(t, s2)
143 putRunKeys(t, s2, rec2, []RunKey{
144 {Key: "old", RunID: "run_old", Digest: "D", Caller: "A", At: now.Add(-runs.KeyTTL - time.Minute)},
145 {Key: "a", RunID: "run_a", Digest: "D", Caller: "A", At: now.Add(-time.Hour)},
146 })
147 got, err = setRun(t, s2, rec2, RunAdmit(keyed("run_b", "b", "D")))
148 if err != nil || len(got.RunKeys) != 2 || got.RunKeys[0].Key != "a" || got.RunKeys[1].Key != "b" {
149 t.Fatalf("an expired entry under the count: %+v, %v; want a then b", got.RunKeys, err)
150 }
151
152 // The record's lookup ignores an expired entry too.
153 r := Record{RunKeys: []RunKey{{Key: "x", RunID: "run_x", Digest: "D", At: now.Add(-runs.KeyTTL - time.Second)}}}
154 if _, ok := r.RunKeyFor("x", now); ok {
155 t.Error("RunKeyFor answered an expired key")
156 }
157 r.RunKeys[0].At = now.Add(-time.Minute)
158 if k, ok := r.RunKeyFor("x", now); !ok || k.RunID != "run_x" {
159 t.Errorf("RunKeyFor = %+v, %v", k, ok)
160 }
161}
162
163// TestRunKeysCappedPerCaller: the count cap is each caller's own, so caller
164// B's admissions never evict caller A's key, and A's own 33rd admission
165// evicts A's oldest.
166func TestRunKeysCappedPerCaller(t *testing.T) {
167 s := release.NewDirStore(t.TempDir())
168 rec := runTestRecord(t, s)
169 admit := func(a RunAdmission) {
170 t.Helper()
171 if _, err := setRun(t, s, rec, RunAdmit(a)); err != nil {
172 t.Fatalf("admit %s: %v", a.ID, err)
173 }
174 if _, err := setRun(t, s, rec, RunExpire(a.ID)); err != nil {
175 t.Fatal(err)
176 }
177 }
178 admit(keyedBy("A", "run_a0", "KA", "D"))
179 for i := range 40 {
180 admit(keyedBy("B", fmt.Sprintf("run_b%02d", i), fmt.Sprintf("KB%02d", i), "D"))
181 }
182 _, err := setRun(t, s, rec, RunAdmit(keyedBy("A", "run_x", "KA", "D")))
183 wantReplay(t, "A's key after B's 40 admissions", err, "run_a0")
184 cur, err := GetRecord(context.Background(), s, rec.Account, rec.App, rec.Host)
185 if err != nil {
186 t.Fatal(err)
187 }
188 per := map[string]int{}
189 for _, k := range cur.RunKeys {
190 per[k.Caller]++
191 }
192 if per["A"] != 1 || per["B"] != runs.MaxRunKeysPerCaller {
193 t.Fatalf("entries per caller %v, want A 1 and B %d", per, runs.MaxRunKeysPerCaller)
194 }
195 for i := 1; i < runs.MaxRunKeysPerCaller; i++ {
196 admit(keyedBy("A", fmt.Sprintf("run_a%02d", i), fmt.Sprintf("KA%02d", i), "D"))
197 }
198 _, err = setRun(t, s, rec, RunAdmit(keyedBy("A", "run_y", "KA", "D")))
199 wantReplay(t, "A's key with 32 of A's own", err, "run_a0")
200 admit(keyedBy("A", "run_a32", "KA32", "D")) // A's 33rd
201 got, err := setRun(t, s, rec, RunAdmit(keyedBy("A", "run_z", "KA", "D")))
202 if err != nil || got.ActiveRun.ID != "run_z" {
203 t.Fatalf("A's oldest key after A's 33rd admission: %+v, %v; want it evicted and a fresh run", got, err)
204 }
205}
206
207func TestRunAdmitRefusesAMalformedKeyedAdmission(t *testing.T) {
208 s := release.NewDirStore(t.TempDir())
209 rec := runTestRecord(t, s)
210 if _, err := setRun(t, s, rec, RunAdmit(keyed("run_a", "K", ""))); !errors.Is(err, ErrRunConflict) {
211 t.Fatalf("a key without a digest: %v, want ErrRunConflict", err)
212 }
213 if _, err := setRun(t, s, rec, RunAdmit(keyedBy("", "run_a", "K", "D"))); !errors.Is(err, ErrRunConflict) {
214 t.Fatalf("a key without a caller: %v, want ErrRunConflict", err)
215 }
216}
217
218// keysJSON is the record's run_keys as stored, byte for byte.
219func keysJSON(t *testing.T, s release.Store, rec Record) string {
220 t.Helper()
221 raw, _, err := s.Get(context.Background(), RecordKey(rec.Account, rec.App, rec.Host))
222 if err != nil {
223 t.Fatal(err)
224 }
225 var m map[string]json.RawMessage
226 if err := json.Unmarshal(raw, &m); err != nil {
227 t.Fatal(err)
228 }
229 return string(m["run_keys"])
230}
231
232// TestRunKeysSurviveEveryRecordMutation: every run transition, the
233// ordinary writers, the move transitions and the import transitions carry
234// run_keys through unchanged. Only a keyed admission adds to them.
235func TestRunKeysSurviveEveryRecordMutation(t *testing.T) {
236 ctx := context.Background()
237 at := time.Now().UTC().Truncate(time.Second)
238 seeded := []RunKey{{Key: strings.Repeat("a", 64), RunID: "run_k", Digest: strings.Repeat("d", 64), At: at}}
239 fresh := func(t *testing.T, r Record) (release.Store, Record, string) {
240 t.Helper()
241 s := release.NewDirStore(t.TempDir())
242 if err := CreateRecord(ctx, s, r); err != nil {
243 t.Fatal(err)
244 }
245 putRunKeys(t, s, r, seeded)
246 want := keysJSON(t, s, r)
247 if !strings.Contains(want, "run_k") {
248 t.Fatalf("run_keys = %s", want)
249 }
250 return s, r, want
251 }
252 base := func() Record {
253 return Record{Schema: 1, Host: "a.go.example.com", Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
254 }
255 type step struct {
256 name string
257 apply func(release.Store, Record) error
258 }
259 run := func(mut func(*Record) error) func(release.Store, Record) error {
260 return func(s release.Store, r Record) error {
261 _, err := SetRecordRun(ctx, s, r.Account, r.App, r.Host, mut)
262 return err
263 }
264 }
265 move := func(mut func(*Record) error) func(release.Store, Record) error {
266 return func(s release.Store, r Record) error {
267 _, err := SetRecordMove(ctx, s, r.Account, r.App, r.Host, mut)
268 return err
269 }
270 }
271 seed := func(mut func(*Record) error) func(release.Store, Record) error {
272 return func(s release.Store, r Record) error {
273 _, err := SetRecordSeed(ctx, s, r.Account, r.App, r.Host, mut)
274 return err
275 }
276 }
277 deadline := at.Add(time.Hour)
278 moveID := "mv_" + strings.Repeat("a", 26)
279 seedID, err := seeds.NewID()
280 if err != nil {
281 t.Fatal(err)
282 }
283 importRecord := base()
284 importRecord.Seed = &SeedSource{ID: seedID, Kind: seeds.KindImport, State: seeds.StateAwaitingUpload}
285 abortRecord := base()
286 abortRecord.Seed = &SeedSource{ID: seedID, Kind: seeds.KindImport, State: seeds.StateAwaitingUpload}
287
288 for _, path := range []struct {
289 name string
290 rec Record
291 steps []step
292 }{
293 {"the six run transitions", base(), []step{
294 {"RunAdmit, unkeyed", run(RunAdmit(RunAdmission{ID: "run_a", Deadline: deadline}))},
295 {"RunPick", run(RunPick("run_a", "box-1", "boot-1", deadline))},
296 {"RunCancelTransition, picked", run(RunCancelTransition("run_a"))},
297 {"RunFinish", run(RunFinish("run_a"))},
298 {"RunClear", run(RunClear("run_a"))},
299 {"RunAdmit, unkeyed, again", run(RunAdmit(RunAdmission{ID: "run_b", Deadline: deadline}))},
300 {"RunExpire", run(RunExpire("run_b"))},
301 {"RunAdmit replacing a terminal run", run(RunAdmit(RunAdmission{ID: "run_c", Deadline: deadline}))},
302 {"RunCancelTransition, queued", run(RunCancelTransition("run_c"))},
303 {"RunClear, cancelled", run(RunClear("run_c"))},
304 }},
305 {"the ordinary writers", base(), []step{
306 {"SetRecordRegion", func(s release.Store, r Record) error { return SetRecordRegion(ctx, s, r.Account, r.App, r.Host, "uk") }},
307 {"UpdateRecord", func(s release.Store, r Record) error {
308 cur, etag, err := GetRecordETag(ctx, s, r.Account, r.App, r.Host)
309 if err != nil {
310 return err
311 }
312 cur.Channel = "beta"
313 return UpdateRecord(ctx, s, *cur, etag)
314 }},
315 }},
316 {"a legacy move to its end", base(), []step{
317 {"MoveRequest", move(MoveRequest("uk", at))},
318 {"MovePromoteReady", move(MovePromoteReady(at))},
319 {"MoveMarkCutting", move(MoveMarkCutting())},
320 {"MoveComplete", move(MoveComplete())},
321 }},
322 {"a legacy move aborted", base(), []step{
323 {"MoveRequest", move(MoveRequest("uk", at))},
324 {"MoveAbort", move(MoveAbort())},
325 {"MoveAbortComplete", move(MoveAbortComplete())},
326 }},
327 {"a phase-1 move start", base(), []step{
328 {"MoveStartTransition", move(MoveStartTransition(MoveStart{ID: moveID, Incarnation: strings.Repeat("a", 64), From: "ie", To: "uk", Cut: "auto", Now: at}))},
329 }},
330 {"an offline move aborted", base(), []step{
331 {"OfflineStart", move(OfflineStart(moveID, "ie", "uk", "paul", at))},
332 {"OfflineAbort", move(OfflineAbort(moveID))},
333 {"OfflineFinish", move(OfflineFinish(moveID, at))},
334 }},
335 {"an import to its commit", importRecord, []step{
336 {"SeedUploaded", seed(SeedUploaded(seedID))},
337 {"SeedStopped", seed(SeedStopped(seedID))},
338 {"SeedSwapped", seed(SeedSwapped(seedID))},
339 {"SeedPlace", seed(SeedPlace(seedID))},
340 {"SeedCommit", seed(SeedCommit(seedID))},
341 }},
342 {"an import aborted", abortRecord, []step{
343 {"SeedAbort", seed(SeedAbort(seedID))},
344 }},
345 } {
346 t.Run(path.name, func(t *testing.T) {
347 s, r, want := fresh(t, path.rec)
348 for _, st := range path.steps {
349 if err := st.apply(s, r); err != nil {
350 t.Fatalf("%s: %v", st.name, err)
351 }
352 if got := keysJSON(t, s, r); got != want {
353 t.Fatalf("%s: run_keys %s, want %s", st.name, got, want)
354 }
355 }
356 })
357 }
358}
359
360// TestOnlyRunAdmitWritesRunKeys reads the package's source: RunAdmit is the
361// one function that assigns RunKeys, and no function overwrites a whole
362// record through a pointer, which would drop run_keys with everything else
363// it did not copy. Every other writer mutates the record it read in place,
364// which is what TestRunKeysSurviveEveryRecordMutation relies on.
365func TestOnlyRunAdmitWritesRunKeys(t *testing.T) {
366 files, err := filepath.Glob("*.go")
367 if err != nil {
368 t.Fatal(err)
369 }
370 fset := token.NewFileSet()
371 for _, file := range files {
372 if strings.HasSuffix(file, "_test.go") {
373 continue
374 }
375 src, err := os.ReadFile(file)
376 if err != nil {
377 t.Fatal(err)
378 }
379 f, err := parser.ParseFile(fset, file, src, 0)
380 if err != nil {
381 t.Fatal(err)
382 }
383 for _, decl := range f.Decls {
384 fn, ok := decl.(*ast.FuncDecl)
385 if !ok || fn.Body == nil {
386 continue
387 }
388 ast.Inspect(fn.Body, func(n ast.Node) bool {
389 as, ok := n.(*ast.AssignStmt)
390 if !ok {
391 return true
392 }
393 for _, lhs := range as.Lhs {
394 switch l := lhs.(type) {
395 case *ast.SelectorExpr:
396 if l.Sel.Name == "RunKeys" && !(file == "run.go" && fn.Name.Name == "RunAdmit") {
397 t.Errorf("%s: %s assigns RunKeys; only RunAdmit may", fset.Position(l.Pos()), fn.Name.Name)
398 }
399 case *ast.StarExpr:
400 if id, ok := l.X.(*ast.Ident); ok && (id.Name == "rec" || id.Name == "r") {
401 t.Errorf("%s: %s overwrites a whole record through *%s", fset.Position(l.Pos()), fn.Name.Name, id.Name)
402 }
403 }
404 }
405 return true
406 })
407 }
408 }
409}
410