amadan
console: ship door — part-count cap, immutable 409, concurrency + time bounds
Review round on the multipart ship endpoint (apicli.go) found five gaps the byte cap alone didn't close: - apiShipMaxBytes bounds bytes, not part COUNT. r.MultipartReader() has no stdlib cap on part count (only ReadForm's 1000-part cap, which streaming opts out of), so a body of tiny validly-named parts could fit millions of them, blowing up the console's heap and then issuing one sequential store write per part until the ship context died, orphaning manifest-less bucket objects (Ship writes the manifest last). Fixed with a hard cap — 1 part for static, 8 for binary — enforced before io.ReadAll ever runs on the offending part. - Every release.Ship error answered 502 "try again", including its permanent refusal to overwrite an existing version with different bytes — a case a retry loop can never fix. That refusal has no sentinel or typed error (release.go), so it's matched by its literal wording and answered 409 instead; genuine store failures stay 502. Along the way, a GetManifest pre-check makes a genuinely IDENTICAL re-ship succeed (200, no-op) rather than tripping release.Ship's own manifest-level conflict — both this handler and the direct `carlos ship` path stamp a fresh BuiltAt on every call, so comparing manifests byte-for-byte (as Ship's putImmutable does) would treat identical content as a conflict for a reason no caller could avoid; comparing artifacts (content) instead of the whole manifest (content + metadata) is what makes the true idempotent case actually work. - The body read was unbounded in time: no server ReadTimeout, and the 2-minute context only starts after the body is already read. A trickled upload could hold a goroutine and buffer open for hours, since the rate limiter counts requests, not concurrency. Fixed with a 10-minute read deadline set via http.NewResponseController before touching the body, plus a small (cap 2) in-flight semaphore on Server — a third concurrent ship is refused with 429 rather than queuing. - Duplicate part names silently overwrote each other, charging the byte and part-count budgets twice for content only the last write kept. Refused with 400 instead. New tests: too-many-parts, duplicate-part, an expanded traversal-defense table (asserting the store is untouched after each refusal, not just the status code), the immutable-conflict/idempotent-reship pair, and a concurrency-limit test using io.Pipe-backed requests plus a shipStarted test hook (same shape as cliNotified) to synchronize on the semaphore actually being held rather than racing it.
3 files changed,
+414
−3
internal/console/apicli.go+171 −3internal/console/apicli_test.go+222 −0internal/console/server.go+21 −0
diff --git a/internal/console/apicli.go b/internal/console/apicli.go| index 948e66760..9ab77f6da 100644 |
| --- a/internal/console/apicli.go |
| +++ b/internal/console/apicli.go |
| @@ -343,6 +343,24 @@ func (s *Server) handleAPIRollback(w http.ResponseWriter, r *http.Request) { |
| // covers today's binaries and a static site tarball in one request. |
| var apiShipMaxBytes int64 = 256 << 20 |
| +// apiShipMaxPartsStatic and apiShipMaxPartsBinary bound the NUMBER of file |
| +// parts a ship request may carry: one for a static release (the site.tar.gz |
| +// name alone), eight for a binary release (comfortably above today's |
| +// platform count). r.MultipartReader() has no stdlib cap on part count at all — that |
| +// cap exists only on mime/multipart's ReadForm (1000 parts), which |
| +// streaming deliberately opts out of. Without a check of our own, a body |
| +// full of tiny, validly-named parts fits millions of them inside |
| +// apiShipMaxBytes: each one a map entry on the console's heap, and — had it |
| +// ever gotten that far — one sequential putImmutable call per part inside |
| +// release.Ship until the ship context timed out, orphaning however many |
| +// artifact objects had already landed (Ship writes the manifest LAST by |
| +// design, so a killed run leaves bytes with no manifest naming them). This |
| +// check IS the part-count cap; nothing upstream provides one. |
| +const ( |
| + apiShipMaxPartsStatic = 1 |
| + apiShipMaxPartsBinary = 8 |
| +) |
| + |
| // targetRe validates a binary artifact's part NAME before it is used to |
| // build a bucket key: release.ArtifactKey joins it straight into the object |
| // path, so this check is path-traversal defense, not tidiness — a part |
| @@ -362,6 +380,53 @@ func shipMaxBytesRefused(w http.ResponseWriter, err error) bool { |
| return false |
| } |
| +// shipImmutableConflict reports whether err is release.Ship's refusal to |
| +// overwrite an existing version's object with DIFFERENT bytes — a |
| +// permanent conflict a retry can never fix, unlike every other error Ship |
| +// returns (a store/IO failure, which a retry might actually resolve). |
| +// |
| +// release.Ship/putImmutable (internal/release/release.go) expose no |
| +// sentinel or typed error for this specific case — the only signal |
| +// available is the literal wording putImmutable formats ("... already |
| +// exists with different content — refusing to overwrite an immutable |
| +// release object"), so this is a STRING match against release.go's current |
| +// wording, not a durable error-type check (errors.Is/As have nothing to |
| +// grab onto here). If that wording ever changes in release.go, this |
| +// silently stops firing and the conflict falls back to the 502 below — |
| +// worth revisiting if release.go ever grows a proper sentinel for it. |
| +// |
| +// validateManifest's errors also return from Ship along this same |
| +// unwrapped path, but none of its messages ("invalid app...", "invalid |
| +// kind...", "manifest has no artifacts", etc) contain this wording, so they |
| +// can never match here. They are unreachable in this branch for an |
| +// unrelated reason: every field validateManifest checks (app, version, |
| +// kind, artifact file names, sha256, size) was already validated by |
| +// handleAPIShip itself, using the same rules, before Ship was ever called. |
| +func shipImmutableConflict(err error) bool { |
| + return strings.Contains(err.Error(), "already exists with different content") |
| +} |
| + |
| +// shipArtifactsMatch reports whether two artifact maps describe the exact |
| +// same shipped CONTENT — same file set, same bytes (by sha256+size) for |
| +// each. BuiltAt/BuiltBy are deliberately excluded: they are metadata, not |
| +// content, and both handleAPIShip and the direct `carlos ship` CLI path |
| +// stamp a fresh BuiltAt into every manifest they build, so comparing |
| +// manifests wholesale (as release.Ship's own putImmutable does, |
| +// byte-for-byte) would call a genuinely identical re-ship a conflict for a |
| +// reason no caller could ever avoid. |
| +func shipArtifactsMatch(a, b map[string]release.Artifact) bool { |
| + if len(a) != len(b) { |
| + return false |
| + } |
| + for k, av := range a { |
| + bv, ok := b[k] |
| + if !ok || av.File != bv.File || av.SHA256 != bv.SHA256 || av.Size != bv.Size { |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| // handleAPIShip answers `carlos ship`: a multipart upload of one release's |
| // artifacts. Hashes and sizes are computed HERE, from the bytes that |
| // actually arrived on the wire, and handed to release.Ship as |
| @@ -394,6 +459,40 @@ func (s *Server) handleAPIShip(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusTooManyRequests, "too many uploads — wait a while and try again") |
| return |
| } |
| + // The rate limiter bounds requests over TIME; it says nothing about how |
| + // many run AT ONCE. A ship holds a goroutine and a substantial buffer |
| + // open for as long as the client trickles bytes in, so a handful of |
| + // slow uploads in parallel is a resource-exhaustion path the counter |
| + // above does nothing to close. Non-blocking: a full semaphore means |
| + // "busy right now", answered immediately, rather than a request that |
| + // queues indefinitely behind someone else's slow upload. |
| + select { |
| + case s.shipSem <- struct{}{}: |
| + defer func() { <-s.shipSem }() |
| + default: |
| + jsonError(w, http.StatusTooManyRequests, "another ship is in progress — try again shortly") |
| + return |
| + } |
| + if s.shipStarted != nil { |
| + s.shipStarted <- struct{}{} |
| + } |
| + |
| + // Bounds the read side in TIME, not just in bytes: with no deadline, a |
| + // client sending one byte a second could hold this goroutine (and |
| + // whatever it has already buffered) open for hours — the semaphore |
| + // above only bounds how many can do that at once, not for how long. |
| + // http.ErrNotSupported is tolerated because some transports (test |
| + // servers among them, in principle) don't implement the underlying |
| + // deadline setter; the semaphore is still there as a backstop on those. |
| + rc := http.NewResponseController(w) |
| + if err := rc.SetReadDeadline(time.Now().Add(10 * time.Minute)); err != nil && !errors.Is(err, http.ErrNotSupported) { |
| + log.Printf("console: api ship set read deadline: %v", err) |
| + } |
| + defer func() { |
| + if err := rc.SetReadDeadline(time.Time{}); err != nil && !errors.Is(err, http.ErrNotSupported) { |
| + log.Printf("console: api ship clear read deadline: %v", err) |
| + } |
| + }() |
| r.Body = http.MaxBytesReader(w, r.Body, apiShipMaxBytes) |
| mr, err := r.MultipartReader() |
| @@ -431,6 +530,10 @@ func (s *Server) handleAPIShip(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusBadRequest, `invalid kind (want "binary" or "static")`) |
| return |
| } |
| + maxParts := apiShipMaxPartsBinary |
| + if meta.Kind == "static" { |
| + maxParts = apiShipMaxPartsStatic |
| + } |
| files := make(map[string][]byte) |
| artifacts := make(map[string]release.Artifact) |
| @@ -446,6 +549,16 @@ func (s *Server) handleAPIShip(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusBadRequest, "malformed multipart body") |
| return |
| } |
| + // Checked before anything else touches this part — including its |
| + // name, and certainly before io.ReadAll — because the cost this |
| + // guards against is per-PART, not per-byte: a body of nothing but |
| + // tiny, validly-named parts must be refused on the (maxParts+1)th |
| + // one, not after buffering it. |
| + if len(files) >= maxParts { |
| + jsonError(w, http.StatusBadRequest, |
| + fmt.Sprintf("too many file parts (max %d for a %s release)", maxParts, meta.Kind)) |
| + return |
| + } |
| name := part.FormName() |
| if meta.Kind == "static" { |
| if name != "site.tar.gz" { |
| @@ -456,6 +569,16 @@ func (s *Server) handleAPIShip(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusBadRequest, "invalid artifact name "+name) |
| return |
| } |
| + // A second part claiming a name already seen is refused rather than |
| + // silently overwriting the first: both would otherwise be charged |
| + // against apiShipMaxBytes and apiShipMaxPartsBinary/Static, but only |
| + // the last one's bytes would ever reach the manifest — a caller |
| + // deserves an error, not a release that quietly ships less than it |
| + // paid to upload. |
| + if _, dup := files[name]; dup { |
| + jsonError(w, http.StatusBadRequest, "duplicate artifact part "+name) |
| + return |
| + } |
| b, err := io.ReadAll(part) |
| if err != nil { |
| if shipMaxBytesRefused(w, err) { |
| @@ -491,11 +614,56 @@ func (s *Server) handleAPIShip(w http.ResponseWriter, r *http.Request) { |
| // short strings. |
| ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute) |
| defer cancel() |
| + |
| + // A version's objects are immutable once shipped, but BOTH callers that |
| + // build a manifest like this one (this handler and the direct `carlos |
| + // ship` CLI path) stamp BuiltAt with a fresh time.Now() on every |
| + // invocation. Comparing the freshly-marshaled manifest byte-for-byte |
| + // against whatever release.Ship's own putImmutable finds already |
| + // stored — which is exactly what it does — would make a genuinely |
| + // identical re-ship register as a "different content" conflict for a |
| + // reason no caller could ever avoid: its OWN timestamp. Comparing |
| + // ARTIFACTS here first — content, not metadata — is what makes |
| + // re-shipping the exact same bytes twice actually succeed instead of |
| + // permanently wedging on its own BuiltAt. |
| + if existing, gerr := release.GetManifest(ctx, s.cfg.Releases, acct.Sqid, app, ver); gerr == nil { |
| + if !shipArtifactsMatch(existing.Artifacts, artifacts) { |
| + jsonError(w, http.StatusConflict, |
| + "version "+ver+" was already shipped with different content — versions are immutable; ship under a new version") |
| + return |
| + } |
| + // Identical content already shipped: answer success without |
| + // writing anything — calling release.Ship again would only |
| + // rediscover this the slow way, and would in fact trip its OWN |
| + // manifest-level putImmutable conflict (see above), since BuiltAt |
| + // drifts every call even when the artifacts don't. 200, not 201: |
| + // nothing was created by this request. |
| + out := make(map[string]any, len(existing.Artifacts)) |
| + for name, a := range existing.Artifacts { |
| + out[name] = map[string]any{"sha256": a.SHA256, "size": a.Size} |
| + } |
| + writeJSON(w, http.StatusOK, map[string]any{ |
| + "result": "shipped", "version": ver, "artifacts": out, |
| + }) |
| + return |
| + } |
| + |
| if err := release.Ship(ctx, s.cfg.Releases, acct.Sqid, m, files); err != nil { |
| log.Printf("console: api ship %s %s: %v", app, ver, err) |
| - // The store failed, not the caller: a 502, matching the other doors' |
| - // convention that a bucket-side error is never described to the |
| - // caller in bucket-side terms. |
| + if shipImmutableConflict(err) { |
| + // A permanent conflict, not a transient one: reachable when this |
| + // raced against another ship of the same version (the |
| + // GetManifest check above is not atomic with the Ship call |
| + // below), not through the common path above. Retrying this |
| + // exact request will never succeed, so it gets its own status |
| + // rather than the retry-flavored 502 below. |
| + jsonError(w, http.StatusConflict, |
| + "version "+ver+" was already shipped with different content — versions are immutable; ship under a new version") |
| + return |
| + } |
| + // Anything else is the store failing, not the caller: a 502, |
| + // matching the other doors' convention that a bucket-side error is |
| + // never described to the caller in bucket-side terms. |
| jsonError(w, http.StatusBadGateway, "could not store the release — try again") |
| return |
| } |
diff --git a/internal/console/apicli_test.go b/internal/console/apicli_test.go| index bb758ce1e..60afc5b60 100644 |
| --- a/internal/console/apicli_test.go |
| +++ b/internal/console/apicli_test.go |
| @@ -15,10 +15,12 @@ import ( |
| "crypto/sha256" |
| "encoding/hex" |
| "encoding/json" |
| + "fmt" |
| "io" |
| "mime/multipart" |
| "net/http" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| @@ -910,3 +912,223 @@ func TestAPIShipMemberGate(t *testing.T) { |
| t.Fatalf("ship as non-member body = %q, want the stdlib http.NotFound output", body) |
| } |
| } |
| + |
| +// TestAPIShipTooManyParts: apiShipMaxBytes bounds bytes, not COUNT — a body |
| +// of nothing but tiny, validly-named parts would otherwise fit millions of |
| +// them in one small request. The ninth binary part (max is |
| +// apiShipMaxPartsBinary=8) is refused before it is ever read, and nothing |
| +// from this request reaches release.Ship: no manifest for this version |
| +// exists afterward. |
| +func TestAPIShipTooManyParts(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "titogo", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-ship-toomanyparts") |
| + |
| + parts := []shipPart{metaPart(t, "abc1234", "binary")} |
| + for i := 0; i < 9; i++ { |
| + parts = append(parts, shipPart{name: fmt.Sprintf("part-%d", i), body: []byte("x")}) |
| + } |
| + status, out := postShip(t, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", token, parts) |
| + if status != http.StatusBadRequest { |
| + t.Fatalf("9-part ship = %d %v, want 400", status, out) |
| + } |
| + if _, err := release.GetManifest(context.Background(), fx.rel, release.OpsAccount, "titogo", "abc1234"); err == nil { |
| + t.Fatalf("manifest exists after a refused too-many-parts ship, want none — nothing should reach release.Ship") |
| + } |
| +} |
| + |
| +// TestAPIShipDuplicatePart: two parts claiming the same artifact name are |
| +// refused rather than letting the second silently win — both would |
| +// otherwise be charged against the byte and part-count budgets while only |
| +// the last one's bytes ever reached the manifest. |
| +func TestAPIShipDuplicatePart(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "titogo", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-ship-duplicate") |
| + |
| + status, out := postShip(t, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", token, |
| + []shipPart{ |
| + metaPart(t, "abc1234", "binary"), |
| + {name: "linux-arm64", body: randomBytes(t, 32)}, |
| + {name: "linux-arm64", body: randomBytes(t, 32)}, |
| + }) |
| + if status != http.StatusBadRequest { |
| + t.Fatalf("duplicate-named ship = %d %v, want 400", status, out) |
| + } |
| + if _, err := release.GetManifest(context.Background(), fx.rel, release.OpsAccount, "titogo", "abc1234"); err == nil { |
| + t.Fatalf("manifest exists after a refused duplicate-part ship, want none") |
| + } |
| +} |
| + |
| +// TestAPIShipTraversalRefusals: every part name that reads like an attempt |
| +// to escape the artifact's bucket-key segment is refused with 400, and — the |
| +// point of this test, not just its status code — none of them leave any |
| +// trace in the release store. Part names become bucket key segments via the |
| +// manifest (release.ArtifactKey joins them straight into the object path), |
| +// so this is path-traversal defense, not tidiness. |
| +func TestAPIShipTraversalRefusals(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "titogo", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-ship-traversal") |
| + payload := randomBytes(t, 64) |
| + |
| + cases := []struct { |
| + name string |
| + version string |
| + kind string |
| + part string |
| + }{ |
| + {"dotdot slash etc passwd", "trv00001", "binary", "../../etc/passwd"}, |
| + {"nested traversal under a valid-looking prefix", "trv00002", "binary", "linux-arm64/../../x"}, |
| + {"bare dotdot", "trv00003", "binary", ".."}, |
| + {"empty part name", "trv00004", "binary", ""}, |
| + {"static traversal", "trv00005", "static", "../site.tar.gz"}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + status, out := postShip(t, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", token, |
| + []shipPart{metaPart(t, c.version, c.kind), {name: c.part, body: payload}}) |
| + if status != http.StatusBadRequest { |
| + t.Fatalf("%s = %d %v, want 400", c.name, status, out) |
| + } |
| + if _, err := release.GetManifest(context.Background(), fx.rel, release.OpsAccount, "titogo", c.version); err == nil { |
| + t.Fatalf("%s: manifest exists for version %s, want none — release.Ship must never have been reached", |
| + c.name, c.version) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// TestAPIShipImmutableConflict: re-shipping an already-shipped version with |
| +// DIFFERENT bytes is a permanent 409, not the generic 502 every other |
| +// release.Ship failure gets — a caller retrying a 502 would re-upload |
| +// forever and never learn this can't succeed. Re-shipping the exact same |
| +// bytes, by contrast, must stay a success: it is the one case release.Ship |
| +// itself calls idempotent, and a caller retrying a dropped connection needs |
| +// that retry to actually land. |
| +func TestAPIShipImmutableConflict(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "titogo", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-ship-immutable") |
| + payload := randomBytes(t, 256) |
| + |
| + status, out := postShip(t, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", token, |
| + []shipPart{metaPart(t, "abc1234", "binary"), {name: "linux-arm64", body: payload}}) |
| + if status != http.StatusCreated { |
| + t.Fatalf("first ship = %d %v", status, out) |
| + } |
| + |
| + different := randomBytes(t, 256) |
| + status, out = postShip(t, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", token, |
| + []shipPart{metaPart(t, "abc1234", "binary"), {name: "linux-arm64", body: different}}) |
| + if status != http.StatusConflict { |
| + t.Fatalf("re-ship with different bytes = %d %v, want 409", status, out) |
| + } |
| + if !strings.Contains(str(t, out, "error"), "immutable") { |
| + t.Fatalf("conflict error = %q, want it to name immutability", str(t, out, "error")) |
| + } |
| + |
| + status, out = postShip(t, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", token, |
| + []shipPart{metaPart(t, "abc1234", "binary"), {name: "linux-arm64", body: payload}}) |
| + if status != http.StatusOK { |
| + t.Fatalf("re-ship with identical bytes = %d %v, want 200 (idempotent, nothing created)", status, out) |
| + } |
| + if str(t, out, "result") != "shipped" || str(t, out, "version") != "abc1234" { |
| + t.Fatalf("idempotent re-ship body = %v", out) |
| + } |
| +} |
| + |
| +// TestAPIShipConcurrencyLimit: the rate limiter bounds requests over TIME, |
| +// but says nothing about how many run AT ONCE. Two ships already in flight |
| +// (each holding the server's ship semaphore while its body trickles in, |
| +// via an io.Pipe under the test's control) leave a third concurrent |
| +// request no room — it must be refused immediately with 429, not queued |
| +// behind either of the first two. |
| +func TestAPIShipConcurrencyLimit(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "titogo", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-ship-concurrency") |
| + |
| + // A hook (same shape as cliNotified): signalled once a request has |
| + // actually acquired shipSem, so this test waits on a real fact instead |
| + // of racing the two background goroutines below with a sleep. |
| + fx.srv.shipStarted = make(chan struct{}, 2) |
| + |
| + var wg sync.WaitGroup |
| + statuses := make([]int, 2) |
| + var closers []func() error |
| + for i := 0; i < 2; i++ { |
| + pr, pw := io.Pipe() |
| + closers = append(closers, pw.Close) |
| + req, err := http.NewRequest(http.MethodPost, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", pr) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + req.Header.Set("Content-Type", "multipart/form-data; boundary=xshipboundaryx") |
| + req.Header.Set("Authorization", "Bearer "+token) |
| + wg.Add(1) |
| + idx := i |
| + go func() { |
| + defer wg.Done() |
| + res, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + statuses[idx] = -1 |
| + return |
| + } |
| + defer res.Body.Close() |
| + io.Copy(io.Discard, res.Body) |
| + statuses[idx] = res.StatusCode |
| + }() |
| + } |
| + |
| + for i := 0; i < 2; i++ { |
| + select { |
| + case <-fx.srv.shipStarted: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("timed out waiting for the first two ships to start (acquire shipSem)") |
| + } |
| + } |
| + |
| + status, out := postShip(t, fx.ts.URL+"/api/cli/apps/bab/titogo/releases", token, |
| + []shipPart{metaPart(t, "abc1234", "binary"), {name: "linux-arm64", body: randomBytes(t, 64)}}) |
| + if status != http.StatusTooManyRequests { |
| + t.Fatalf("third concurrent ship = %d %v, want 429", status, out) |
| + } |
| + if !strings.Contains(str(t, out, "error"), "in progress") { |
| + t.Fatalf("third concurrent ship error = %q, want it to name the concurrency limit", str(t, out, "error")) |
| + } |
| + |
| + // Unblock the two held-open requests (each now sees an immediately-EOF |
| + // body and is refused as an empty upload) so the test doesn't leak |
| + // goroutines, and confirm they actually finish. |
| + for _, c := range closers { |
| + c() |
| + } |
| + done := make(chan struct{}) |
| + go func() { wg.Wait(); close(done) }() |
| + select { |
| + case <-done: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("timed out waiting for the held-open ships to finish after closing their bodies") |
| + } |
| + for i, st := range statuses { |
| + if st != http.StatusBadRequest { |
| + t.Fatalf("held-open ship %d finished with status %d, want 400 (empty upload, once unblocked)", i, st) |
| + } |
| + } |
| +} |
diff --git a/internal/console/server.go b/internal/console/server.go| index 53fb6adf1..abcf0babb 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -120,6 +120,22 @@ type Server struct { |
| // notification attempt. Tests only — the send is off the request |
| // goroutine, and a test that asserted on it without this would be racing. |
| cliNotified chan struct{} |
| + // shipSem bounds how many `carlos ship` uploads (apicli.go, |
| + // handleAPIShip) may run AT ONCE, as opposed to s.cli.ship, which |
| + // bounds them over TIME. A rate limiter is a counter; it says nothing |
| + // about concurrency, and a ship holds a goroutine plus a substantial |
| + // buffer open for as long as the client trickles bytes in — a handful |
| + // of slow uploads run in parallel is a resource-exhaustion path the |
| + // counter alone does nothing to close. Always initialized (never nil): |
| + // unlike cliNotified below this is load-bearing in production, not a |
| + // test hook. |
| + shipSem chan struct{} |
| + // shipStarted, when non-nil, is signalled after a ship request |
| + // acquires shipSem, before it does any (possibly slow, possibly |
| + // blocking) reading of the request body. Tests only, same reasoning as |
| + // cliNotified: without this a test racing to fill shipSem would have no |
| + // way to know the earlier requests actually hold it yet. |
| + shipStarted chan struct{} |
| } |
| // NewServer wires a server over an opened Store, loading (or minting) the |
| @@ -140,6 +156,11 @@ func NewServer(cfg Config, st *Store) (*Server, error) { |
| instances: newInstanceFetcher(cfg.AppClient, cfg.AppScheme), |
| secure: strings.HasPrefix(cfg.Origin, "https://"), |
| cli: newCLILimits(), |
| + // Two concurrent ships at a time: roomy for one operator's CI plus a |
| + // human running `carlos ship` by hand, tight enough that a fleet of |
| + // slow-loris uploads against a stolen or shared token can't pile up |
| + // goroutines and buffers without bound. |
| + shipSem: make(chan struct{}, 2), |
| } |
| s.mux.HandleFunc("GET /{$}", s.handleHome) |
| s.mux.HandleFunc("POST /auth/continue", s.handleContinue) |