amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3// A keyed run create (service runs spec § 3): admitted once, settled by the
4// instance record, and a retry with the same key gets the same run.
5
6import (
7 "bytes"
8 "context"
9 "crypto/sha256"
10 "encoding/hex"
11 "encoding/json"
12 "errors"
13 "fmt"
14 "io"
15 "mime/multipart"
16 "net/http"
17 "net/http/httptest"
18 "net/url"
19 "slices"
20 "strings"
21 "sync"
22 "testing"
23 "time"
24
25 "github.com/carlosframework/platform/internal/provision"
26 "github.com/carlosframework/platform/internal/release"
27 "github.com/carlosframework/platform/internal/runs"
28 "github.com/carlosframework/platform/internal/svcrole"
29)
30
31const testKey = "fleet:import:acct-42"
32
33// keyedFixture is runFixture with keyed creates turned on.
34func keyedFixture(t *testing.T) *runFx {
35 t.Helper()
36 fx := runFixture(t)
37 fx.srv.cfg.RunKeysEnabled = true
38 return fx
39}
40
41func sha(s string) string {
42 h := sha256.Sum256([]byte(s))
43 return hex.EncodeToString(h[:])
44}
45
46// fileMeta is okMeta with one declared file, x.csv holding body.
47func fileMeta(body string) runs.Meta {
48 m := okMeta()
49 m.Files = []runs.MetaFile{{Name: "x.csv", Size: int64(len(body)), SHA256: sha(body)}}
50 return m
51}
52
53func xcsv(body string) []struct{ name, body string } {
54 return []struct{ name, body string }{{"x.csv", body}}
55}
56
57// postKeyed is postRun with an Idempotency-Key header ("" sends none).
58func postKeyed(t *testing.T, u, bearer, key string, meta runs.Meta, parts []struct{ name, body string }) (int, string) {
59 t.Helper()
60 var buf bytes.Buffer
61 mw := multipart.NewWriter(&buf)
62 mp, err := mw.CreateFormField("meta")
63 if err != nil {
64 t.Fatal(err)
65 }
66 if err := json.NewEncoder(mp).Encode(meta); err != nil {
67 t.Fatal(err)
68 }
69 for _, p := range parts {
70 fp, err := mw.CreateFormFile("file", p.name)
71 if err != nil {
72 t.Fatal(err)
73 }
74 io.WriteString(fp, p.body)
75 }
76 mw.Close()
77 req, err := http.NewRequest(http.MethodPost, u, &buf)
78 if err != nil {
79 t.Fatal(err)
80 }
81 req.Header.Set("Content-Type", mw.FormDataContentType())
82 req.Header.Set("Authorization", "Bearer "+bearer)
83 if key != "" {
84 req.Header.Set("Idempotency-Key", key)
85 }
86 res, err := http.DefaultClient.Do(req)
87 if err != nil {
88 t.Fatal(err)
89 }
90 defer res.Body.Close()
91 out, _ := io.ReadAll(res.Body)
92 return res.StatusCode, string(out)
93}
94
95// keyedCreated is the 202 body, with replayed.
96type keyedCreated struct {
97 runs.Created
98 Replayed *bool `json:"replayed"`
99}
100
101func createdOf(t *testing.T, status int, body string) keyedCreated {
102 t.Helper()
103 if status != http.StatusAccepted {
104 t.Fatalf("create: %d %s", status, body)
105 }
106 var c keyedCreated
107 if err := json.Unmarshal([]byte(body), &c); err != nil || !runs.ValidID(c.ID) {
108 t.Fatalf("created %s: %v", body, err)
109 }
110 return c
111}
112
113func (c keyedCreated) replayed() bool { return c.Replayed != nil && *c.Replayed }
114
115// createKeyed is a keyed create that must answer 202.
116func (fx *runFx) createKeyed(t *testing.T, bearer, key string, meta runs.Meta, parts []struct{ name, body string }) keyedCreated {
117 t.Helper()
118 status, body := postKeyed(t, fx.runsURL(runTestHost), bearer, key, meta, parts)
119 return createdOf(t, status, body)
120}
121
122// runIDsWithObjects is every run id with objects under the shop app.
123func (fx *runFx) runIDsWithObjects(t *testing.T) []string {
124 t.Helper()
125 prefix := runs.RunsPrefix(release.OpsAccount, "shop")
126 keys, err := fx.rel.List(context.Background(), prefix)
127 if err != nil {
128 t.Fatal(err)
129 }
130 var ids []string
131 for _, k := range keys {
132 id := strings.SplitN(strings.TrimPrefix(k, prefix), "/", 2)[0]
133 if !slices.Contains(ids, id) {
134 ids = append(ids, id)
135 }
136 }
137 return ids
138}
139
140func (fx *runFx) instRecord(t *testing.T) *provision.Record {
141 t.Helper()
142 rec, err := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost)
143 if err != nil {
144 t.Fatal(err)
145 }
146 return rec
147}
148
149func (fx *runFx) runStarts(t *testing.T) int {
150 t.Helper()
151 events, _ := fx.st.Events(OpsAccountID, "shop", 100)
152 n := 0
153 for _, e := range events {
154 if e.Action == "run-start" {
155 n++
156 }
157 }
158 return n
159}
160
161// finishAndAck takes id from queued to acknowledged through the doors a box
162// and the CLI use.
163func (fx *runFx) finishAndAck(t *testing.T, bearer, id string) {
164 t.Helper()
165 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
166 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0, LastSeq: 0}); err != nil {
167 t.Fatal(err)
168 }
169 fx.transition(t, provision.RunFinish(id))
170 if status, body := fx.followAs(t, bearer, "ack", id); status != http.StatusOK {
171 t.Fatalf("ack: %d %s", status, body)
172 }
173}
174
175func TestRunKeyedCreateThenRetryReturnsTheSameRun(t *testing.T) {
176 fx := keyedFixture(t)
177 first := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
178 if first.replayed() || first.State != provision.RunQueued {
179 t.Fatalf("a fresh admission: %+v", first)
180 }
181 again := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
182 if again.ID != first.ID || !again.replayed() || again.Host != runTestHost {
183 t.Fatalf("the retry: %+v, want %s replayed", again, first.ID)
184 }
185 if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != first.ID {
186 t.Fatalf("runs with objects %v, want only %s", ids, first.ID)
187 }
188 if n := fx.runStarts(t); n != 1 {
189 t.Errorf("%d run-start rows, want 1", n)
190 }
191 rec := fx.instRecord(t)
192 if len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != first.ID || rec.RunKeys[0].Key != runs.ScopedKey("user:alice@keymail.dev", testKey) {
193 t.Fatalf("run_keys %+v", rec.RunKeys)
194 }
195 req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", first.ID)
196 if err != nil || len(req.Files) != 1 || req.Files[0].SHA256 != sha("a,b,c") {
197 t.Fatalf("request %+v, %v", req, err)
198 }
199 // request.json links the run to its key by the scoped hash, never the
200 // caller's own words.
201 if req.IdempotencyKey != runs.ScopedKey("user:alice@keymail.dev", testKey) {
202 t.Fatalf("request.json idempotency_key %q, want the scoped hash", req.IdempotencyKey)
203 }
204 raw, _, err := fx.rel.Get(context.Background(), runs.RequestKey(release.OpsAccount, "shop", first.ID))
205 if err != nil || strings.Contains(string(raw), testKey) {
206 t.Fatalf("request.json holds the raw key: %s, %v", raw, err)
207 }
208}
209
210func TestRunKeyedRetryWhileOpenReturnsTheRun(t *testing.T) {
211 fx := keyedFixture(t)
212 runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
213 first := fx.createKeyed(t, runner, testKey, serviceMeta(), nil)
214 fx.transition(t, provision.RunPick(first.ID, "box-1", "boot-1", time.Now().Add(time.Hour)))
215 again := fx.createKeyed(t, runner, testKey, serviceMeta(), nil)
216 if again.ID != first.ID || !again.replayed() || again.State != provision.RunPicked {
217 t.Fatalf("the retry while open: %+v, want %s picked and replayed", again, first.ID)
218 }
219 if n := fx.runStarts(t); n != 1 {
220 t.Errorf("%d run-start rows, want 1", n)
221 }
222}
223
224func TestRunKeyedRetryAfterAckStillReturnsTheRun(t *testing.T) {
225 fx := keyedFixture(t)
226 first := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
227 fx.finishAndAck(t, fx.member, first.ID)
228 again := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
229 if again.ID != first.ID || !again.replayed() || again.State != "" {
230 t.Fatalf("the retry after ack: %+v, want %s replayed, with no record state", again, first.ID)
231 }
232 // A later unkeyed run takes the record; the key still answers.
233 other := fx.startRun(t)
234 again = fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
235 if again.ID != first.ID || !again.replayed() {
236 t.Fatalf("the retry after %s replaced it: %+v", other, again)
237 }
238 if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != other {
239 t.Fatalf("runs with objects %v, want only %s", ids, other)
240 }
241}
242
243func TestRunKeyReusedWithADifferentRequest(t *testing.T) {
244 fx := keyedFixture(t)
245 first := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
246 before := fx.instRecord(t)
247 other := okMeta()
248 other.Argv = []string{"import", "y.csv"}
249 status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, other, nil)
250 if status != http.StatusConflict || codeOf(t, body) != runCodeKeyReused || errorOf(t, body) == "" {
251 t.Fatalf("the same key, another request: %d %s, want 409 %s", status, body, runCodeKeyReused)
252 }
253 if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != first.ID {
254 t.Fatalf("runs with objects %v, want only %s", ids, first.ID)
255 }
256 if after := fx.instRecord(t); after.RecordSeq != before.RecordSeq {
257 t.Fatalf("the refusal wrote the record: seq %d -> %d", before.RecordSeq, after.RecordSeq)
258 }
259}
260
261// barrierStore holds every admission's record write until want request.json
262// writes have landed, so that concurrent creates are all past the door's key
263// lookup and have all written their objects before any is admitted.
264type barrierStore struct {
265 release.Store
266 recordKey string
267 want int
268 mu sync.Mutex
269 requests int
270 ready chan struct{}
271}
272
273func (b *barrierStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
274 if key == b.recordKey {
275 select {
276 case <-b.ready:
277 case <-time.After(10 * time.Second):
278 return "", errors.New("barrier: the other create never wrote its objects")
279 }
280 }
281 etag, err := b.Store.Put(ctx, key, body, ifMatch)
282 if err == nil && strings.HasSuffix(key, "/request.json") {
283 b.mu.Lock()
284 b.requests++
285 if b.requests == b.want {
286 close(b.ready)
287 }
288 b.mu.Unlock()
289 }
290 return etag, err
291}
292
293func TestRunConcurrentIdenticalKeyedCreatesMakeOneRun(t *testing.T) {
294 fx := keyedFixture(t)
295 runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
296 fx.srv.cfg.Releases = &barrierStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), want: 2, ready: make(chan struct{})}
297 meta := serviceMeta()
298 meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5, SHA256: sha("a,b,c")}}
299 var wg sync.WaitGroup
300 statuses, bodies := make([]int, 2), make([]string, 2)
301 for i := range 2 {
302 wg.Add(1)
303 go func() {
304 defer wg.Done()
305 statuses[i], bodies[i] = postKeyed(t, fx.runsURL(runTestHost), runner, testKey, meta, xcsv("a,b,c"))
306 }()
307 }
308 wg.Wait()
309 a, b := createdOf(t, statuses[0], bodies[0]), createdOf(t, statuses[1], bodies[1])
310 if a.ID != b.ID {
311 t.Fatalf("two runs for one key: %s and %s", a.ID, b.ID)
312 }
313 if a.replayed() == b.replayed() {
314 t.Fatalf("want one fresh admission and one replay: %s / %s", bodies[0], bodies[1])
315 }
316 if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != a.ID {
317 t.Fatalf("runs with objects %v, want only the winner %s: the loser's objects must be gone", ids, a.ID)
318 }
319 if n := fx.runStarts(t); n != 1 {
320 t.Errorf("%d run-start rows, want 1", n)
321 }
322 if rec := fx.instRecord(t); rec.ActiveRun == nil || rec.ActiveRun.ID != a.ID || len(rec.RunKeys) != 1 {
323 t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys)
324 }
325}
326
327// lookupKey is GET …/runs?key=<key>.
328func (fx *runFx) lookupKey(t *testing.T, bearer, key string) (int, string) {
329 t.Helper()
330 status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+url.QueryEscape(key), bearer, "")
331 return status, body
332}
333
334func TestRunKeyOfAnotherCallerIsInvisible(t *testing.T) {
335 fx := keyedFixture(t)
336 runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
337 person := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
338 if status, body := fx.lookupKey(t, runner, testKey); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
339 t.Fatalf("the service looking up the person's key: %d %s, want 404", status, body)
340 }
341 fx.finishAndAck(t, fx.member, person.ID)
342 svc := fx.createKeyed(t, runner, testKey, serviceMeta(), nil)
343 if svc.ID == person.ID || svc.replayed() {
344 t.Fatalf("the service's create with the person's key: %+v, want a fresh run", svc)
345 }
346 for _, c := range []struct {
347 bearer, want string
348 }{{fx.member, person.ID}, {runner, svc.ID}} {
349 status, body := fx.lookupKey(t, c.bearer, testKey)
350 var got struct {
351 RunID string `json:"run_id"`
352 }
353 if status != http.StatusOK || json.Unmarshal([]byte(body), &got) != nil || got.RunID != c.want {
354 t.Errorf("lookup: %d %s, want %s", status, body, c.want)
355 }
356 }
357}
358
359func TestRunKeyedHashMismatchWritesNothing(t *testing.T) {
360 fx := keyedFixture(t)
361 dir := spoolDir(t)
362 status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,d"))
363 if status != http.StatusBadRequest || codeOf(t, body) != runCodeFileHashMismatch {
364 t.Fatalf("a file that is not its declared sha256: %d %s, want 400 %s", status, body, runCodeFileHashMismatch)
365 }
366 fx.noRunObjects(t, "a hash mismatch")
367 assertNoSpool(t, dir, "a hash mismatch")
368 if rec := fx.instRecord(t); len(rec.RunKeys) != 0 {
369 t.Fatalf("run_keys %+v", rec.RunKeys)
370 }
371
372 // A keyed create must declare every file's sha256.
373 undeclared := fileMeta("a,b,c")
374 undeclared.Files[0].SHA256 = ""
375 status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, undeclared, xcsv("a,b,c"))
376 if status != http.StatusBadRequest || codeOf(t, body) != runCodeInvalidRequest {
377 t.Fatalf("a keyed file with no sha256: %d %s, want 400 %s", status, body, runCodeInvalidRequest)
378 }
379 // A declared sha256 is checked whether or not the create is keyed.
380 status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, "", fileMeta("a,b,c"), xcsv("a,b,d"))
381 if status != http.StatusBadRequest || codeOf(t, body) != runCodeFileHashMismatch {
382 t.Fatalf("an unkeyed mismatch: %d %s", status, body)
383 }
384 badShape := fileMeta("a,b,c")
385 badShape.Files[0].SHA256 = "ABC"
386 status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, "", badShape, xcsv("a,b,c"))
387 if status != http.StatusBadRequest || codeOf(t, body) != runCodeInvalidRequest {
388 t.Fatalf("a malformed sha256: %d %s", status, body)
389 }
390 fx.noRunObjects(t, "the hash refusals")
391}
392
393// failAfter is a request body that serves head and then fails every read.
394type failAfter struct {
395 head io.Reader
396 hit bool
397}
398
399func (f *failAfter) Read(p []byte) (int, error) {
400 if n, err := f.head.Read(p); n > 0 || err != io.EOF {
401 return n, err
402 }
403 f.hit = true
404 return 0, errors.New("failAfter: the file part must not be read")
405}
406
407// serveBrokenBody sends a create whose body is the meta part and the start
408// of a file part, after which every read fails, straight to the handler.
409func (fx *runFx) serveBrokenBody(t *testing.T, bearer, key string, meta runs.Meta) *httptest.ResponseRecorder {
410 t.Helper()
411 var buf bytes.Buffer
412 mw := multipart.NewWriter(&buf)
413 mp, _ := mw.CreateFormField("meta")
414 if err := json.NewEncoder(mp).Encode(meta); err != nil {
415 t.Fatal(err)
416 }
417 fp, _ := mw.CreateFormFile("file", "x.csv")
418 io.WriteString(fp, "a")
419 req := httptest.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &failAfter{head: &buf})
420 req.Header.Set("Content-Type", mw.FormDataContentType())
421 req.Header.Set("Authorization", "Bearer "+bearer)
422 if key != "" {
423 req.Header.Set("Idempotency-Key", key)
424 }
425 w := httptest.NewRecorder()
426 fx.srv.ServeHTTP(w, req)
427 return w
428}
429
430func TestRunKnownKeyAnsweredWithoutReadingFiles(t *testing.T) {
431 fx := keyedFixture(t)
432 first := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
433 w := fx.serveBrokenBody(t, fx.member, testKey, fileMeta("a,b,c"))
434 again := createdOf(t, w.Code, w.Body.String())
435 if again.ID != first.ID || !again.replayed() {
436 t.Fatalf("the retry: %+v, want %s replayed", again, first.ID)
437 }
438 // The same broken body with a fresh key reaches the file part and fails.
439 fx.finishAndAck(t, fx.member, first.ID)
440 w = fx.serveBrokenBody(t, fx.member, "another-key", fileMeta("a,b,c"))
441 if w.Code != http.StatusBadRequest || codeOf(t, w.Body.String()) != runCodeMalformed {
442 t.Fatalf("a fresh key with a broken file part: %d %s, want 400 malformed", w.Code, w.Body.String())
443 }
444}
445
446func TestRunRefusedKeyIsReevaluated(t *testing.T) {
447 fx := keyedFixture(t)
448 open := fx.startRun(t)
449 status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, okMeta(), nil)
450 if status != http.StatusConflict || codeOf(t, body) != runCodeRunOpen {
451 t.Fatalf("a keyed create over an open run: %d %s, want 409 run_open", status, body)
452 }
453 fx.finishAndAck(t, fx.member, open)
454 c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
455 if c.replayed() || c.ID == open {
456 t.Fatalf("the retry once the other run was acked: %+v, want a fresh run", c)
457 }
458}
459
460func TestRunKeyLookup(t *testing.T) {
461 fx := keyedFixture(t)
462 if status, body := fx.lookupKey(t, fx.member, testKey); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
463 t.Fatalf("an unknown key: %d %s, want 404", status, body)
464 }
465 c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
466 status, body := fx.lookupKey(t, fx.member, testKey)
467 if status != http.StatusOK || body != fmt.Sprintf("{\"run_id\":%q}\n", c.ID) {
468 t.Fatalf("a known key: %d %s, want %s", status, body, c.ID)
469 }
470 for _, bad := range []string{"", "a b", strings.Repeat("k", 129)} {
471 status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+url.QueryEscape(bad), fx.member, "")
472 if status != http.StatusBadRequest || codeOf(t, body) != runCodeKeyInvalid {
473 t.Errorf("key %q: %d %s, want 400 %s", bad, status, body, runCodeKeyInvalid)
474 }
475 }
476 status, _, body = rawReq(t, http.MethodGet, fx.ts.URL+"/api/cli/apps/"+release.OpsAccount+"/shop/instances/nobody.example.com/runs?key="+testKey, fx.member, "")
477 if status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
478 t.Fatalf("an unknown host: %d %s", status, body)
479 }
480 // The shared gates hold here as at every run door.
481 stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-key-stranger")
482 status, h, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+testKey, stranger, "")
483 assertNotFound(t, "a stranger's lookup", canonicalNotFound(t), status, h, body)
484 fx.srv.cfg.HostClaim = true
485 if status, body := fx.lookupKey(t, fx.member, testKey); status != http.StatusConflict || codeOf(t, body) != runCodeHostClaim {
486 t.Fatalf("a lookup on a host-claim console: %d %s", status, body)
487 }
488}
489
490// The key lookup is a run door: the run role reaches it, every other role
491// gets the role gate's 403, Admin included.
492func TestRunKeyLookupRoleMatrix(t *testing.T) {
493 fx := keyedFixture(t)
494 runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
495 u := fx.runsURL(runTestHost) + "?key=" + testKey
496 if status, _, body := rawReq(t, http.MethodGet, u, runner, ""); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
497 t.Fatalf("a run service: %d %s, want the lookup's own 404", status, body)
498 }
499 for _, r := range svcrole.All() {
500 if r == svcrole.Run {
501 continue
502 }
503 app, rungs := "shop", []string(nil)
504 if r == svcrole.Admin {
505 app = ""
506 }
507 if r == svcrole.Deploy {
508 rungs = []string{"edge"}
509 }
510 tok, err := fx.st.NewServicePrincipalScoped(OpsAccountID, "role-"+string(r), app, r, rungs, nil, "paul@keymail.dev", time.Now())
511 if err != nil {
512 t.Fatalf("mint %s: %v", r, err)
513 }
514 raw, _ := json.Marshal(map[string]string{"error": fmt.Sprintf("this credential holds role %q; this door needs one of [run]", r)})
515 if status, _, body := rawReq(t, http.MethodGet, u, tok, ""); status != http.StatusForbidden || body != string(raw)+"\n" {
516 t.Errorf("role %s: %d %q, want 403 %s", r, status, body, raw)
517 }
518 }
519}
520
521// TestRunUnkeyedOrderUnchanged: an unkeyed create over an open run is
522// refused before its body is read, as before keys. A keyed one reads its
523// meta first, so the same broken body is malformed.
524func TestRunUnkeyedOrderUnchanged(t *testing.T) {
525 fx := keyedFixture(t)
526 fx.startRun(t)
527 broken := func(key string) *httptest.ResponseRecorder {
528 req := httptest.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &failAfter{head: strings.NewReader("")})
529 req.Header.Set("Content-Type", "multipart/form-data; boundary=x")
530 req.Header.Set("Authorization", "Bearer "+fx.member)
531 if key != "" {
532 req.Header.Set("Idempotency-Key", key)
533 }
534 w := httptest.NewRecorder()
535 fx.srv.ServeHTTP(w, req)
536 return w
537 }
538 if w := broken(""); w.Code != http.StatusConflict || codeOf(t, w.Body.String()) != runCodeRunOpen {
539 t.Fatalf("unkeyed over an open run: %d %s, want 409 run_open before the body", w.Code, w.Body.String())
540 }
541 if w := broken(testKey); w.Code != http.StatusBadRequest || codeOf(t, w.Body.String()) != runCodeMalformed {
542 t.Fatalf("keyed over an open run with a broken body: %d %s, want 400 malformed", w.Code, w.Body.String())
543 }
544}
545
546// The keyed form of TestAPIRunCreateAdmissionWhoseAnswerWasLost: the retry
547// inside updateRecord finds this request's own key, which is a fresh
548// admission, and the run's objects stay.
549func TestAPIRunCreateAdmissionWhoseAnswerWasLostKeyed(t *testing.T) {
550 for name, failure := range map[string]error{
551 "retried write": fmt.Errorf("s3store: record: %w", release.ErrCASConflict),
552 "transport error": errors.New("connection reset by peer"),
553 } {
554 t.Run(name, func(t *testing.T) {
555 fx := keyedFixture(t)
556 fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), err: failure}
557 c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
558 if c.replayed() || c.State != provision.RunQueued {
559 t.Fatalf("created %+v, want a fresh admission", c)
560 }
561 rec := fx.instRecord(t)
562 if rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != c.ID {
563 t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys)
564 }
565 keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID))
566 if len(keys) != 2 {
567 t.Fatalf("the admitted run's objects are %q, want request.json and in/x.csv", keys)
568 }
569 if n := fx.runStarts(t); n != 1 {
570 t.Errorf("%d run-start rows, want 1", n)
571 }
572 })
573 }
574}
575
576// admitRun itself answers the retry inside updateRecord that finds this
577// request's own key: admitted, with the record's state, and no error for the
578// handler to settle.
579func TestAdmitRunOwnKeyReplayIsAdmitted(t *testing.T) {
580 fx := keyedFixture(t)
581 fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost),
582 err: fmt.Errorf("s3store: record: %w", release.ErrCASConflict)}
583 id, err := runs.MintID()
584 if err != nil {
585 t.Fatal(err)
586 }
587 state, err := fx.srv.admitRun(context.Background(), release.OpsAccount, "shop", runTestHost, id, "alice@keymail.dev", "K", "D", "C")
588 if err != nil || state != provision.RunQueued {
589 t.Fatalf("admitRun = %q, %v; want queued and no error", state, err)
590 }
591}
592
593func TestKeyedCreateRefusedUntilKeysAreOn(t *testing.T) {
594 fx := runFixture(t)
595 status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
596 if status != http.StatusServiceUnavailable || codeOf(t, body) != runCodeKeysNotReady || errorOf(t, body) == "" {
597 t.Fatalf("a keyed create with keys off: %d %s, want 503 %s", status, body, runCodeKeysNotReady)
598 }
599 fx.noRunObjects(t, "a keyed create with keys off")
600 if rec := fx.instRecord(t); len(rec.RunKeys) != 0 {
601 t.Fatalf("run_keys %+v", rec.RunKeys)
602 }
603 // Even an invalid key is refused as not ready, never run unkeyed.
604 if status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, "a b", okMeta(), nil); status != http.StatusServiceUnavailable {
605 t.Fatalf("an invalid key with keys off: %d %s", status, body)
606 }
607 c := fx.createKeyed(t, fx.member, "", okMeta(), nil)
608 if c.replayed() {
609 t.Fatalf("an unkeyed create: %+v", c)
610 }
611 if req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", c.ID); err != nil || req.IdempotencyKey != "" {
612 t.Fatalf("an unkeyed request: %+v, %v", req, err)
613 }
614 if rec := fx.instRecord(t); len(rec.RunKeys) != 0 {
615 t.Fatalf("an unkeyed create recorded a key: %+v", rec.RunKeys)
616 }
617 // With keys on, an invalid key is refused by its shape.
618 fx.srv.cfg.RunKeysEnabled = true
619 if status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, "a/b", okMeta(), nil); status != http.StatusBadRequest || codeOf(t, body) != runCodeKeyInvalid {
620 t.Fatalf("an invalid key: %d %s, want 400 %s", status, body, runCodeKeyInvalid)
621 }
622}
623
624// The codes this task adds are in the one list.
625func TestRunKeyCodesAreListed(t *testing.T) {
626 for _, c := range []string{runCodeKeyReused, runCodeKeyInvalid, runCodeKeysNotReady, runCodeFileHashMismatch} {
627 if !slices.Contains(runRefusalCodes, c) {
628 t.Errorf("%s is not in runRefusalCodes", c)
629 }
630 }
631 for c, want := range map[string]string{
632 runCodeKeyReused: "idempotency_key_reused", runCodeKeyInvalid: "idempotency_key_invalid",
633 runCodeKeysNotReady: "keys_not_ready", runCodeFileHashMismatch: "file_hash_mismatch",
634 } {
635 if c != want {
636 t.Errorf("code %s, want %s", c, want)
637 }
638 }
639}
640
641// movedOnStore commits the admission's record write, then has the run picked,
642// finished and cleared (a box and an ack, faster than the answer), and
643// answers err: the record's ActiveRun no longer names the run when the door
644// settles the lost answer.
645type movedOnStore struct {
646 release.Store
647 recordKey string
648 err error
649 once sync.Once
650 t *testing.T
651}
652
653func (m *movedOnStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
654 etag, err := m.Store.Put(ctx, key, body, ifMatch)
655 if err != nil || key != m.recordKey {
656 return etag, err
657 }
658 m.once.Do(func() {
659 rec, gerr := provision.DecodeRecord(body)
660 if gerr != nil || rec.ActiveRun == nil {
661 m.t.Errorf("movedOnStore: %v %+v", gerr, rec.ActiveRun)
662 return
663 }
664 id := rec.ActiveRun.ID
665 for _, mut := range []func(*provision.Record) error{
666 provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)),
667 provision.RunFinish(id),
668 provision.RunClear(id),
669 } {
670 if _, e := provision.SetRecordRun(context.Background(), m.Store, release.OpsAccount, "shop", runTestHost, mut); e != nil {
671 m.t.Errorf("movedOnStore: %v", e)
672 }
673 }
674 err = m.err
675 })
676 return etag, err
677}
678
679// A keyed admission whose answer was lost is settled by its key, not by
680// ActiveRun: the record has moved on, the key still names this run, so the
681// door answers it admitted and keeps its objects.
682func TestRunKeyedLostAdmissionSettledByItsKey(t *testing.T) {
683 fx := keyedFixture(t)
684 fx.srv.cfg.Releases = &movedOnStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost),
685 err: errors.New("connection reset by peer"), t: t}
686 c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
687 if c.replayed() || c.State != "" {
688 t.Fatalf("created %+v, want a fresh admission whose run the record no longer names", c)
689 }
690 rec := fx.instRecord(t)
691 if rec.ActiveRun != nil || len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != c.ID {
692 t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys)
693 }
694 if keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID)); len(keys) != 2 {
695 t.Fatalf("the committed run's objects are %q, want request.json and in/x.csv", keys)
696 }
697 if n := fx.runStarts(t); n != 1 {
698 t.Errorf("%d run-start rows, want 1", n)
699 }
700}
701
702// keysDroppedStore commits the admission's record write, then has a record
703// writer drop run_keys (a box rolled back to code that does not know them)
704// while the record still holds the run, and answers err.
705type keysDroppedStore struct {
706 release.Store
707 recordKey string
708 err error
709 once sync.Once
710 t *testing.T
711}
712
713func (m *keysDroppedStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
714 etag, err := m.Store.Put(ctx, key, body, ifMatch)
715 if err != nil || key != m.recordKey {
716 return etag, err
717 }
718 m.once.Do(func() {
719 drop := func(r *provision.Record) error { r.RunKeys = nil; return nil }
720 if _, e := provision.SetRecordRun(context.Background(), m.Store, release.OpsAccount, "shop", runTestHost, drop); e != nil {
721 m.t.Errorf("keysDroppedStore: %v", e)
722 }
723 })
724 return etag, m.err
725}
726
727// A keyed admission whose answer was lost, and whose key is gone from the
728// record while the record still names the run, is admitted: the objects stay
729// and the answer names the run, as the unkeyed settle would have.
730func TestRunKeyedSettleKeepsARunTheRecordHoldsWhenItsKeyIsGone(t *testing.T) {
731 fx := keyedFixture(t)
732 fx.srv.cfg.Releases = &keysDroppedStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost),
733 err: errors.New("connection reset by peer"), t: t}
734 c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
735 rec := fx.instRecord(t)
736 if rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || len(rec.RunKeys) != 0 {
737 t.Fatalf("answer %+v; record run %+v keys %+v, want the run held with no key", c, rec.ActiveRun, rec.RunKeys)
738 }
739 if c.State != provision.RunQueued {
740 t.Errorf("answered %+v, want the record's queued state", c)
741 }
742 if keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID)); len(keys) != 2 {
743 t.Fatalf("the held run's objects are %q, want request.json and in/x.csv", keys)
744 }
745}
746
747// twinOnRetryStore loses both of updateRecord's attempts: the first record
748// write is refused as a conflict without landing, and before the second a
749// twin's keyed admission lands, so the second conflicts for real.
750type twinOnRetryStore struct {
751 release.Store
752 recordKey string
753 twin provision.RunAdmission
754 puts int
755 t *testing.T
756}
757
758func (s *twinOnRetryStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
759 if key == s.recordKey {
760 s.puts++
761 switch s.puts {
762 case 1:
763 return "", fmt.Errorf("s3store: record: %w", release.ErrCASConflict)
764 case 2:
765 if _, err := provision.SetRecordRun(context.Background(), s.Store, release.OpsAccount, "shop", runTestHost, provision.RunAdmit(s.twin)); err != nil {
766 s.t.Errorf("twin admission: %v", err)
767 }
768 }
769 }
770 return s.Store.Put(ctx, key, body, ifMatch)
771}
772
773// A keyed create that loses both CAS attempts to a twin answers with the
774// twin's run, replayed, and deletes only its own objects: not a 502.
775func TestRunKeyedCreateLosingBothAttemptsAnswersTheTwin(t *testing.T) {
776 fx := keyedFixture(t)
777 scope := "user:alice@keymail.dev"
778 twinID, err := runs.MintID()
779 if err != nil {
780 t.Fatal(err)
781 }
782 st := &twinOnRetryStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), t: t,
783 twin: provision.RunAdmission{ID: twinID, By: "alice@keymail.dev", Deadline: time.Now().Add(time.Minute),
784 Key: runs.ScopedKey(scope, testKey), Digest: runs.MetaDigest(okMeta()), Caller: runs.CallerOf(scope)}}
785 fx.srv.cfg.Releases = st
786 c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
787 if st.puts != 2 {
788 t.Fatalf("%d record writes, want the two lost attempts", st.puts)
789 }
790 if c.ID != twinID || !c.replayed() || c.State != provision.RunQueued {
791 t.Fatalf("created %+v, want the twin %s, replayed and queued", c, twinID)
792 }
793 if ids := fx.runIDsWithObjects(t); len(ids) != 0 {
794 t.Fatalf("runs with objects %v: the loser's objects must be gone", ids)
795 }
796 if n := fx.runStarts(t); n != 0 {
797 t.Errorf("%d run-start rows, want none from the loser", n)
798 }
799}
800
801// A declared sha256 may be in either case: it is lowercased before it is
802// compared or digested, so an uppercase declaration is not refused and a
803// retry in the other case is the same request.
804func TestRunDeclaredHashIsCaseInsensitive(t *testing.T) {
805 fx := keyedFixture(t)
806 upper := fileMeta("a,b,c")
807 upper.Files[0].SHA256 = strings.ToUpper(upper.Files[0].SHA256)
808 plain := fx.createKeyed(t, fx.member, "", upper, xcsv("a,b,c"))
809 fx.finishAndAck(t, fx.member, plain.ID)
810 first := fx.createKeyed(t, fx.member, testKey, upper, xcsv("a,b,c"))
811 again := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
812 if again.ID != first.ID || !again.replayed() {
813 t.Fatalf("a lowercase retry of an uppercase create: %+v, want %s replayed", again, first.ID)
814 }
815 req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", first.ID)
816 if err != nil || req.Files[0].SHA256 != sha("a,b,c") {
817 t.Fatalf("request %+v, %v", req, err)
818 }
819}
820