| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "mime/multipart" |
| 11 | "net/http" |
| 12 | "net/http/httptest" |
| 13 | "os" |
| 14 | "strings" |
| 15 | "sync" |
| 16 | "testing" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | "github.com/carlosframework/platform/internal/runs" |
| 22 | "github.com/carlosframework/platform/internal/seeds" |
| 23 | "github.com/carlosframework/platform/internal/svcrole" |
| 24 | ) |
| 25 | |
| 26 | const runTestHost = "acme.example.com" |
| 27 | |
| 28 | type runFx struct { |
| 29 | *consoleFixture |
| 30 | member string |
| 31 | } |
| 32 | |
| 33 | func runFixture(t *testing.T) *runFx { |
| 34 | t.Helper() |
| 35 | fx := newConsoleFixture(t) |
| 36 | now := time.Now() |
| 37 | if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", now); err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | if err := fx.st.AddMember(OpsAccountID, "alice@keymail.dev", RoleMember, now); err != nil { |
| 41 | t.Fatal(err) |
| 42 | } |
| 43 | member := cliToken(t, fx, "alice@keymail.dev", "code-run") |
| 44 | putShopRecord(t, fx, provision.Record{Host: runTestHost, Channel: "stable", CreatedAt: time.Now().UTC()}) |
| 45 | return &runFx{consoleFixture: fx, member: member} |
| 46 | } |
| 47 | |
| 48 | // putShopRecord writes an instance record for the shop app with the fields |
| 49 | // a hibernating exec instance has, overridden by whatever rec sets. |
| 50 | func putShopRecord(t *testing.T, fx *consoleFixture, rec provision.Record) { |
| 51 | t.Helper() |
| 52 | rec.App, rec.Account = "shop", release.OpsAccount |
| 53 | if rec.Hibernate == nil { |
| 54 | yes := true |
| 55 | rec.Hibernate = &yes |
| 56 | } |
| 57 | body, err := json.Marshal(rec) |
| 58 | if err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | if _, err := fx.rel.Put(context.Background(), provision.RecordKey(rec.Account, rec.App, rec.Host), body, ""); err != nil { |
| 62 | t.Fatal(err) |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | func (fx *runFx) runsURL(host string) string { |
| 67 | return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/instances/" + host + "/runs" |
| 68 | } |
| 69 | |
| 70 | // postRun sends the multipart create; parts lets a test send parts that |
| 71 | // disagree with meta. |
| 72 | func postRun(t *testing.T, u, bearer string, meta runs.Meta, parts []struct{ name, body string }) (int, string) { |
| 73 | t.Helper() |
| 74 | status, _, body := postRunH(t, u, bearer, meta, parts) |
| 75 | return status, body |
| 76 | } |
| 77 | |
| 78 | // postRunH is postRun that also returns the response headers. |
| 79 | func postRunH(t *testing.T, u, bearer string, meta runs.Meta, parts []struct{ name, body string }) (int, http.Header, string) { |
| 80 | t.Helper() |
| 81 | var buf bytes.Buffer |
| 82 | mw := multipart.NewWriter(&buf) |
| 83 | mp, err := mw.CreateFormField("meta") |
| 84 | if err != nil { |
| 85 | t.Fatal(err) |
| 86 | } |
| 87 | if err := json.NewEncoder(mp).Encode(meta); err != nil { |
| 88 | t.Fatal(err) |
| 89 | } |
| 90 | for _, p := range parts { |
| 91 | fp, err := mw.CreateFormFile("file", p.name) |
| 92 | if err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | io.WriteString(fp, p.body) |
| 96 | } |
| 97 | mw.Close() |
| 98 | req, err := http.NewRequest(http.MethodPost, u, &buf) |
| 99 | if err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | req.Header.Set("Content-Type", mw.FormDataContentType()) |
| 103 | if bearer != "" { |
| 104 | req.Header.Set("Authorization", "Bearer "+bearer) |
| 105 | } |
| 106 | res, err := http.DefaultClient.Do(req) |
| 107 | if err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | defer res.Body.Close() |
| 111 | out, _ := io.ReadAll(res.Body) |
| 112 | return res.StatusCode, res.Header, string(out) |
| 113 | } |
| 114 | |
| 115 | func okMeta() runs.Meta { return runs.Meta{Argv: []string{"import", "x.csv"}, TimeoutS: 60} } |
| 116 | |
| 117 | func errorOf(t *testing.T, body string) string { |
| 118 | t.Helper() |
| 119 | var e struct{ Error string } |
| 120 | if err := json.Unmarshal([]byte(body), &e); err != nil { |
| 121 | t.Fatalf("body %q: %v", body, err) |
| 122 | } |
| 123 | return e.Error |
| 124 | } |
| 125 | |
| 126 | // spoolDir points the door's spool files at a directory of the test's own |
| 127 | // and returns it, so a test can see that none is left behind. |
| 128 | func spoolDir(t *testing.T) string { |
| 129 | t.Helper() |
| 130 | dir := t.TempDir() |
| 131 | t.Setenv("TMPDIR", dir) |
| 132 | return dir |
| 133 | } |
| 134 | |
| 135 | func assertNoSpool(t *testing.T, dir, when string) { |
| 136 | t.Helper() |
| 137 | if left, _ := os.ReadDir(dir); len(left) != 0 { |
| 138 | t.Errorf("%s left %d spool file(s) behind", when, len(left)) |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | func TestAPIRunCreateAdmitsAndRecords(t *testing.T) { |
| 143 | fx := runFixture(t) |
| 144 | spool := spoolDir(t) |
| 145 | meta := okMeta() |
| 146 | meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}} |
| 147 | status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}}) |
| 148 | if status != http.StatusAccepted { |
| 149 | t.Fatalf("status %d %s", status, body) |
| 150 | } |
| 151 | assertNoSpool(t, spool, "an admitted run") |
| 152 | var c runs.Created |
| 153 | if err := json.Unmarshal([]byte(body), &c); err != nil || !runs.ValidID(c.ID) || c.State != provision.RunQueued { |
| 154 | t.Fatalf("created %+v, %v", c, err) |
| 155 | } |
| 156 | ctx := context.Background() |
| 157 | rec, err := provision.GetRecord(ctx, fx.rel, release.OpsAccount, "shop", runTestHost) |
| 158 | if err != nil || rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || rec.ActiveRun.By != "alice@keymail.dev" { |
| 159 | t.Fatalf("record run %+v, %v", rec.ActiveRun, err) |
| 160 | } |
| 161 | if d := time.Until(rec.ActiveRun.Deadline); d < 50*time.Second || d > 61*time.Second { |
| 162 | t.Errorf("pickup deadline %s away, want about 60 s", d) |
| 163 | } |
| 164 | req, err := runs.ReadRequest(ctx, fx.rel, release.OpsAccount, "shop", c.ID) |
| 165 | if err != nil || req.Host != runTestHost || len(req.Incarnation) != 64 || req.TimeoutS != 60 || len(req.Files) != 1 || len(req.Files[0].SHA256) != 64 { |
| 166 | t.Fatalf("request %+v, %v", req, err) |
| 167 | } |
| 168 | raw, _, err := fx.rel.Get(ctx, runs.InputKey(release.OpsAccount, "shop", c.ID, "x.csv")) |
| 169 | if err != nil || string(raw) != "a,b,c" { |
| 170 | t.Fatalf("input %q, %v", raw, err) |
| 171 | } |
| 172 | events, _ := fx.st.Events(OpsAccountID, "shop", 10) |
| 173 | found := false |
| 174 | for _, e := range events { |
| 175 | if e.Action == "run-start" && e.Target == runTestHost && e.Address == "alice@keymail.dev" && strings.Contains(e.Detail, c.ID) && strings.Contains(e.Detail, "x.csv") { |
| 176 | found = true |
| 177 | } |
| 178 | } |
| 179 | if !found { |
| 180 | t.Errorf("no run-start audit row in %+v", events) |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | // Each refusal in spec §5.1's order: every case is valid in every later |
| 185 | // respect, so the status and message show which check refused it. |
| 186 | func TestAPIRunCreateRefusesInOrder(t *testing.T) { |
| 187 | fx := runFixture(t) |
| 188 | ctx := context.Background() |
| 189 | svc := serviceToken(t, fx.st, OpsAccountID, "ci", "shop", svcrole.Instance) |
| 190 | stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-stranger") |
| 191 | |
| 192 | if status, _ := postRun(t, fx.runsURL(runTestHost), svc, okMeta(), nil); status != http.StatusForbidden { |
| 193 | t.Errorf("service credential: %d, want 403", status) |
| 194 | } |
| 195 | status, h, body := postRunH(t, fx.runsURL(runTestHost), stranger, okMeta(), nil) |
| 196 | assertNotFound(t, "stranger", canonicalNotFound(t), status, h, body) |
| 197 | if status, body := postRun(t, fx.runsURL("nobody.example.com"), fx.member, okMeta(), nil); status != http.StatusNotFound || errorOf(t, body) != "nobody.example.com is not an instance of shop" { |
| 198 | t.Errorf("no record: %d %s", status, body) |
| 199 | } |
| 200 | for host, c := range map[string]struct { |
| 201 | rec provision.Record |
| 202 | msg string |
| 203 | }{ |
| 204 | "static.example.com": {provision.Record{Kind: "static"}, "static.example.com is a static route. carlos run works on instances"}, |
| 205 | "addr.example.com": {provision.Record{Addr: "127.0.0.1:9"}, "addr.example.com is served from its own address, so the platform does not run its process"}, |
| 206 | "pool.example.com": {provision.Record{Pool: "shared"}, "pool.example.com is a pooled instance. carlos run works on instances with a process of their own"}, |
| 207 | "fleet.example.com": {provision.Record{Region: "fleet/pi"}, "fleet.example.com runs on your fleet pi. carlos run works on platform boxes only"}, |
| 208 | } { |
| 209 | c.rec.Host = host |
| 210 | putShopRecord(t, fx.consoleFixture, c.rec) |
| 211 | if status, body := postRun(t, fx.runsURL(host), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != c.msg { |
| 212 | t.Errorf("%s: %d %s", host, status, body) |
| 213 | } |
| 214 | } |
| 215 | no := false |
| 216 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: "awake.example.com", Hibernate: &no}) |
| 217 | if status, body := postRun(t, fx.runsURL("awake.example.com"), fx.member, okMeta(), nil); status != http.StatusConflict || !strings.Contains(errorOf(t, body), "does not sleep") { |
| 218 | t.Errorf("always-on: %d %s", status, body) |
| 219 | } |
| 220 | if _, err := provision.SetRecordMove(ctx, fx.rel, release.OpsAccount, "shop", runTestHost, provision.MoveRequest("uk", time.Now())); err != nil { |
| 221 | t.Fatal(err) |
| 222 | } |
| 223 | if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != runTestHost+" is moving. run the command once the move has finished" { |
| 224 | t.Errorf("moving: %d %s", status, body) |
| 225 | } |
| 226 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: "busy.example.com", ActiveRun: &provision.RunState{ID: "run_aaaaaaaaaaaaaaaaaaaaaaaaaa", State: provision.RunPicked, By: "bob@keymail.dev"}}) |
| 227 | if status, body := postRun(t, fx.runsURL("busy.example.com"), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != "busy.example.com already has a run open: run_aaaaaaaaaaaaaaaaaaaaaaaaaa, started by bob@keymail.dev" { |
| 228 | t.Errorf("open run: %d %s", status, body) |
| 229 | } |
| 230 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: "fresh.example.com"}) |
| 231 | bad := []struct { |
| 232 | meta runs.Meta |
| 233 | msg string |
| 234 | }{ |
| 235 | {runs.Meta{TimeoutS: 60}, "name the command to run after --"}, |
| 236 | {runs.Meta{Argv: make([]string, 257), TimeoutS: 60}, "a run takes at most 256 arguments"}, |
| 237 | {runs.Meta{Argv: []string{strings.Repeat("a", 64<<10+1)}, TimeoutS: 60}, "the arguments come to more than 64 KB"}, |
| 238 | {runs.Meta{Argv: []string{"a\x00"}, TimeoutS: 60}, "an argument holds a NUL byte"}, |
| 239 | {runs.Meta{Argv: []string{"a"}, TimeoutS: 6*3600 + 1}, "the timeout must be between 1 second and 6 hours"}, |
| 240 | {runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "a", Size: 65 << 20}}}, "the files come to more than 64 MB"}, |
| 241 | {runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "../x", Size: 1}}}, "../x is not a plain file name"}, |
| 242 | {runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "a", Size: 1}, {Name: "a", Size: 1}}}, "two files are both named a"}, |
| 243 | } |
| 244 | for i := range bad[1].meta.Argv { |
| 245 | bad[1].meta.Argv[i] = "x" |
| 246 | } |
| 247 | for _, c := range bad { |
| 248 | if status, body := postRun(t, fx.runsURL("fresh.example.com"), fx.member, c.meta, nil); status != http.StatusBadRequest || errorOf(t, body) != c.msg { |
| 249 | t.Errorf("%q: %d %s", c.msg, status, body) |
| 250 | } |
| 251 | } |
| 252 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: "import.example.com", Seed: &provision.SeedSource{Kind: seeds.KindImport, State: seeds.StatePlaced}}) |
| 253 | if status, body := postRun(t, fx.runsURL("import.example.com"), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != "import.example.com is an import that is not committed yet. finish or abort the import first" { |
| 254 | t.Errorf("import: %d %s", status, body) |
| 255 | } |
| 256 | } |
| 257 | |
| 258 | func TestAPIRunCreateSpendsTheInstanceBudget(t *testing.T) { |
| 259 | fx := runFixture(t) |
| 260 | // A window of one: the first hit in a window is always allowed |
| 261 | // (rateLimiter.allow, authcli.go:226), the second is over. The budget |
| 262 | // is refusal 3, so it comes before refusal 7 (the run the first opened). |
| 263 | fx.srv.cli.instances = newRateLimiter(1, cliRateWindow) |
| 264 | if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusAccepted { |
| 265 | t.Fatalf("first: %d %s", status, body) |
| 266 | } |
| 267 | if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusTooManyRequests || errorOf(t, body) != "too many run requests. wait a minute and try again" { |
| 268 | t.Fatalf("over budget: %d %s", status, body) |
| 269 | } |
| 270 | } |
| 271 | |
| 272 | func TestAPIRunCreateRefusesAPartThatDisagreesWithMeta(t *testing.T) { |
| 273 | fx := runFixture(t) |
| 274 | spool := spoolDir(t) |
| 275 | meta := okMeta() |
| 276 | meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}} |
| 277 | for name, parts := range map[string][]struct{ name, body string }{ |
| 278 | "short": {{"x.csv", "abc"}}, |
| 279 | "long": {{"x.csv", "abcdefg"}}, |
| 280 | "wrong name": {{"y.csv", "abcde"}}, |
| 281 | "missing": nil, |
| 282 | "extra": {{"x.csv", "abcde"}, {"y.csv", "abcde"}}, |
| 283 | } { |
| 284 | status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, parts) |
| 285 | if status != http.StatusBadRequest || errorOf(t, body) != "the run request is malformed" { |
| 286 | t.Errorf("%s: %d %s", name, status, body) |
| 287 | } |
| 288 | assertNoSpool(t, spool, name) |
| 289 | } |
| 290 | if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "shop")); len(left) != 0 { |
| 291 | t.Fatalf("a refused upload left %q", left) |
| 292 | } |
| 293 | if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost); rec.ActiveRun != nil { |
| 294 | t.Fatal("a refused upload admitted a run") |
| 295 | } |
| 296 | } |
| 297 | |
| 298 | // Two members at once: one run is admitted, the other is told who holds the |
| 299 | // instance, and its objects are gone (codex R1-1). |
| 300 | func TestAPIRunCreateRaceAdmitsOne(t *testing.T) { |
| 301 | fx := runFixture(t) |
| 302 | var wg sync.WaitGroup |
| 303 | statuses := make([]int, 2) |
| 304 | bodies := make([]string, 2) |
| 305 | for i := range 2 { |
| 306 | wg.Add(1) |
| 307 | go func() { |
| 308 | defer wg.Done() |
| 309 | statuses[i], bodies[i] = postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil) |
| 310 | }() |
| 311 | } |
| 312 | wg.Wait() |
| 313 | won, lost := 0, 0 |
| 314 | for i := range 2 { |
| 315 | switch statuses[i] { |
| 316 | case http.StatusAccepted: |
| 317 | won++ |
| 318 | case http.StatusConflict: |
| 319 | lost++ |
| 320 | if !strings.Contains(errorOf(t, bodies[i]), "already has a run open") { |
| 321 | t.Errorf("loser said %s", bodies[i]) |
| 322 | } |
| 323 | } |
| 324 | } |
| 325 | if won != 1 || lost != 1 { |
| 326 | t.Fatalf("statuses %v %v", statuses, bodies) |
| 327 | } |
| 328 | // DirStore keeps emptied directories, so count the runs that still own |
| 329 | // objects rather than the directories. |
| 330 | prefix := runs.RunsPrefix(release.OpsAccount, "shop") |
| 331 | keys, _ := fx.rel.List(context.Background(), prefix) |
| 332 | owners := map[string]bool{} |
| 333 | for _, k := range keys { |
| 334 | owners[strings.SplitN(strings.TrimPrefix(k, prefix), "/", 2)[0]] = true |
| 335 | } |
| 336 | if len(owners) != 1 { |
| 337 | t.Fatalf("runs with objects %v, want only the winner's", owners) |
| 338 | } |
| 339 | } |
| 340 | |
| 341 | // rivalStore admits another run on the instance record the moment the door |
| 342 | // writes request.json: after the door's own checks passed and its objects |
| 343 | // began to land, before its admission. That is the race, made certain. |
| 344 | type rivalStore struct { |
| 345 | release.Store |
| 346 | t *testing.T |
| 347 | once sync.Once |
| 348 | } |
| 349 | |
| 350 | func (r *rivalStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 351 | if strings.HasSuffix(key, "/request.json") { |
| 352 | r.once.Do(func() { |
| 353 | _, err := provision.SetRecordRun(ctx, r.Store, release.OpsAccount, "shop", runTestHost, |
| 354 | provision.RunAdmit(provision.RunAdmission{ID: "run_aaaaaaaaaaaaaaaaaaaaaaaaaa", By: "bob@keymail.dev", Deadline: time.Now().Add(time.Minute)})) |
| 355 | if err != nil { |
| 356 | r.t.Error(err) |
| 357 | } |
| 358 | }) |
| 359 | } |
| 360 | return r.Store.Put(ctx, key, body, ifMatch) |
| 361 | } |
| 362 | |
| 363 | // The loser of an admission that happens after its objects began to land is |
| 364 | // told who holds the instance, and nothing of its own is left in the bucket. |
| 365 | func TestAPIRunCreateLoserAfterUploadLeavesNothing(t *testing.T) { |
| 366 | fx := runFixture(t) |
| 367 | fx.srv.cfg.Releases = &rivalStore{Store: fx.rel, t: t} |
| 368 | meta := okMeta() |
| 369 | meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}} |
| 370 | status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}}) |
| 371 | if want := "already has a run open: run_aaaaaaaaaaaaaaaaaaaaaaaaaa, started by bob@keymail.dev"; status != http.StatusConflict || !strings.HasSuffix(errorOf(t, body), want) { |
| 372 | t.Fatalf("%d %s", status, body) |
| 373 | } |
| 374 | if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "shop")); len(left) != 0 { |
| 375 | t.Fatalf("the loser left %q", left) |
| 376 | } |
| 377 | } |
| 378 | |
| 379 | func TestAPIRunCreateOnAHostClaimConsole(t *testing.T) { |
| 380 | fx := runFixture(t) |
| 381 | fx.srv.cfg.HostClaim = true |
| 382 | if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg { |
| 383 | t.Fatalf("%d %s", status, body) |
| 384 | } |
| 385 | // RUN-C42 comes right after membership: a stranger still gets the |
| 386 | // byte-identical 404, and a member with no budget left gets RUN-C42. |
| 387 | stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-stranger") |
| 388 | status, h, body := postRunH(t, fx.runsURL(runTestHost), stranger, okMeta(), nil) |
| 389 | assertNotFound(t, "stranger on a claim console", canonicalNotFound(t), status, h, body) |
| 390 | fx.srv.cli.instances = newRateLimiter(1, cliRateWindow) |
| 391 | fx.srv.cli.instances.allow(hashToken(fx.member), time.Now()) |
| 392 | if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg { |
| 393 | t.Fatalf("over budget on a claim console: %d %s", status, body) |
| 394 | } |
| 395 | fx.srv.cfg.HostClaim = false |
| 396 | if status, _ := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusTooManyRequests { |
| 397 | t.Fatalf("the same member off a claim console: %d, want the 429 the claim console skipped", status) |
| 398 | } |
| 399 | } |
| 400 | |
| 401 | // The body cap (file bytes plus the meta allowance) ends a stream that grows |
| 402 | // past it with the 64 MB refusal: here a full-size declared file behind a |
| 403 | // meta part padded past the allowance. |
| 404 | func TestAPIRunCreateRefusesABodyPastTheCap(t *testing.T) { |
| 405 | fx := runFixture(t) |
| 406 | spool := spoolDir(t) |
| 407 | meta := okMeta() |
| 408 | meta.Files = []runs.MetaFile{{Name: "big", Size: runs.MaxFileBytes}} |
| 409 | var mb bytes.Buffer |
| 410 | json.NewEncoder(&mb).Encode(meta) |
| 411 | pad := strings.Repeat(" ", runs.MetaAllowance) // trailing white space: valid JSON, counted by the cap |
| 412 | var buf bytes.Buffer |
| 413 | mw := multipart.NewWriter(&buf) |
| 414 | mp, _ := mw.CreateFormField("meta") |
| 415 | mp.Write(mb.Bytes()) |
| 416 | mp.Write([]byte(pad)) |
| 417 | fp, _ := mw.CreateFormFile("file", "big") |
| 418 | fp.Write(make([]byte, runs.MaxFileBytes)) |
| 419 | mw.Close() |
| 420 | if buf.Len() <= runs.MaxFileBytes+runs.MetaAllowance { |
| 421 | t.Fatalf("body is %d bytes, under the cap; the test needs a longer pad", buf.Len()) |
| 422 | } |
| 423 | req, _ := http.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &buf) |
| 424 | req.Header.Set("Content-Type", mw.FormDataContentType()) |
| 425 | req.Header.Set("Authorization", "Bearer "+fx.member) |
| 426 | res, err := http.DefaultClient.Do(req) |
| 427 | if err != nil { |
| 428 | t.Fatal(err) |
| 429 | } |
| 430 | out, _ := io.ReadAll(res.Body) |
| 431 | res.Body.Close() |
| 432 | if res.StatusCode != http.StatusRequestEntityTooLarge || errorOf(t, string(out)) != "the files come to more than 64 MB" { |
| 433 | t.Fatalf("%d %s", res.StatusCode, out) |
| 434 | } |
| 435 | assertNoSpool(t, spool, "a body past the cap") |
| 436 | if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "shop")); len(left) != 0 { |
| 437 | t.Fatalf("left %q", left) |
| 438 | } |
| 439 | } |
| 440 | |
| 441 | // commitThenFailStore commits the first write to the instance record and then |
| 442 | // answers err, as a lost response or a cancelled request would: the record |
| 443 | // holds the run, the caller was told nothing. |
| 444 | type commitThenFailStore struct { |
| 445 | release.Store |
| 446 | recordKey string |
| 447 | err error |
| 448 | once sync.Once |
| 449 | } |
| 450 | |
| 451 | func (c *commitThenFailStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 452 | etag, err := c.Store.Put(ctx, key, body, ifMatch) |
| 453 | if err == nil && key == c.recordKey { |
| 454 | c.once.Do(func() { err = c.err }) |
| 455 | } |
| 456 | return etag, err |
| 457 | } |
| 458 | |
| 459 | // An admission whose answer was lost is still an admission: the door says |
| 460 | // 202 and keeps the run's objects, because the record holds the run and a |
| 461 | // box will read them. |
| 462 | func TestAPIRunCreateAdmissionWhoseAnswerWasLost(t *testing.T) { |
| 463 | for name, failure := range map[string]error{ |
| 464 | // The SDK re-sent a committed write: 412, then updateRecord's retry |
| 465 | // finds this very run queued. |
| 466 | "retried write": fmt.Errorf("s3store: record: %w", release.ErrCASConflict), |
| 467 | // A transport error, or the request's context cancelled, after the |
| 468 | // write landed: the door asks the record. |
| 469 | "transport error": errors.New("connection reset by peer"), |
| 470 | } { |
| 471 | t.Run(name, func(t *testing.T) { |
| 472 | fx := runFixture(t) |
| 473 | fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), err: failure} |
| 474 | meta := okMeta() |
| 475 | meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}} |
| 476 | status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}}) |
| 477 | if status != http.StatusAccepted { |
| 478 | t.Fatalf("%d %s", status, body) |
| 479 | } |
| 480 | var c runs.Created |
| 481 | if err := json.Unmarshal([]byte(body), &c); err != nil || c.State != provision.RunQueued { |
| 482 | t.Fatalf("created %+v, %v", c, err) |
| 483 | } |
| 484 | ctx := context.Background() |
| 485 | rec, err := provision.GetRecord(ctx, fx.rel, release.OpsAccount, "shop", runTestHost) |
| 486 | if err != nil || rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID { |
| 487 | t.Fatalf("record run %+v, %v", rec.ActiveRun, err) |
| 488 | } |
| 489 | keys, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", c.ID)) |
| 490 | if len(keys) != 2 { |
| 491 | t.Fatalf("the admitted run's objects are %q, want request.json and in/x.csv", keys) |
| 492 | } |
| 493 | }) |
| 494 | } |
| 495 | } |
| 496 | |
| 497 | // leavingStore answers every call whose context has ended with that |
| 498 | // context's error, as a real bucket client does, and ends the caller's |
| 499 | // request once the admission's record write has landed: the client went |
| 500 | // away right after its run was admitted. |
| 501 | type leavingStore struct { |
| 502 | release.Store |
| 503 | recordKey string |
| 504 | leave context.CancelFunc |
| 505 | } |
| 506 | |
| 507 | func (l *leavingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 508 | if err := ctx.Err(); err != nil { |
| 509 | return nil, "", err |
| 510 | } |
| 511 | return l.Store.Get(ctx, key) |
| 512 | } |
| 513 | |
| 514 | func (l *leavingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 515 | if err := ctx.Err(); err != nil { |
| 516 | return "", err |
| 517 | } |
| 518 | etag, err := l.Store.Put(ctx, key, body, ifMatch) |
| 519 | if err == nil && key == l.recordKey { |
| 520 | l.leave() |
| 521 | } |
| 522 | return etag, err |
| 523 | } |
| 524 | |
| 525 | // A run admitted for a client that then went away still bumps the serial, |
| 526 | // so a box picks it up at once, and its run-start row still names the |
| 527 | // version the instance follows: both run on the admission's own context. |
| 528 | func TestAPIRunCreateFinishesTheAdmissionForAClientThatLeft(t *testing.T) { |
| 529 | fx := runFixture(t) |
| 530 | shipVersion(t, fx.consoleFixture, "shop", "v1") |
| 531 | seedChannel(t, fx.consoleFixture, "shop", "stable", "v1") |
| 532 | before, _, _ := release.ReadSerial(context.Background(), fx.rel) |
| 533 | |
| 534 | var buf bytes.Buffer |
| 535 | mw := multipart.NewWriter(&buf) |
| 536 | mp, _ := mw.CreateFormField("meta") |
| 537 | if err := json.NewEncoder(mp).Encode(okMeta()); err != nil { |
| 538 | t.Fatal(err) |
| 539 | } |
| 540 | mw.Close() |
| 541 | ctx, leave := context.WithCancel(context.Background()) |
| 542 | defer leave() |
| 543 | fx.srv.cfg.Releases = &leavingStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), leave: leave} |
| 544 | req := httptest.NewRequestWithContext(ctx, http.MethodPost, fx.runsURL(runTestHost), &buf) |
| 545 | req.Header.Set("Content-Type", mw.FormDataContentType()) |
| 546 | req.Header.Set("Authorization", "Bearer "+fx.member) |
| 547 | w := httptest.NewRecorder() |
| 548 | fx.srv.ServeHTTP(w, req) |
| 549 | if w.Code != http.StatusAccepted { |
| 550 | t.Fatalf("%d %s", w.Code, w.Body.String()) |
| 551 | } |
| 552 | if ctx.Err() == nil { |
| 553 | t.Fatal("the client never left: the test proves nothing") |
| 554 | } |
| 555 | if after, _, err := release.ReadSerial(context.Background(), fx.rel); err != nil || after.Serial <= before.Serial { |
| 556 | t.Errorf("serial %d -> %d, %v: not bumped", before.Serial, after.Serial, err) |
| 557 | } |
| 558 | events, _ := fx.st.Events(OpsAccountID, "shop", 10) |
| 559 | found := false |
| 560 | for _, e := range events { |
| 561 | if e.Action == "run-start" && strings.Contains(e.Detail, `"version":"v1"`) { |
| 562 | found = true |
| 563 | } |
| 564 | } |
| 565 | if !found { |
| 566 | t.Errorf("no run-start row naming v1 in %+v", events) |
| 567 | } |
| 568 | } |
| 569 | |
amadan