diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index 5b625631..47efdc77 100644 --- a/cmd/proxy/main.go +++ b/cmd/proxy/main.go @@ -521,7 +521,7 @@ func runMirror() { // Build proxy (reuses same pipeline as serve) fetcher := fetch.NewFetcher() resolver := fetch.NewResolver() - proxy := handler.NewProxy(db, store, fetcher, resolver, logger) + proxy := handler.NewProxy(db, store, handler.LimitArtifactSize(fetcher, cfg.ParseArtifactMaxSize()), resolver, logger) proxy.Denylist, err = denylist.New(cfg.Denylist.Packages) if err != nil { _ = db.Close() diff --git a/config.example.yaml b/config.example.yaml index 493035b0..52ebc26d 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -14,6 +14,10 @@ base_url: "http://localhost:8080" # Set to "0" to disable the timeout. Default: "30s". # http_timeout: "30s" +# Largest upstream artifact to fetch (e.g. "2GB"). Larger ones are refused +# and not cached. Default: unlimited. +# artifact_max_size: "" + # Memory for rewritten npm and Composer metadata, so each upstream document is # rewritten once rather than on every request. Set to "0" to rewrite on every # request. Default: "256MB". diff --git a/docs/configuration.md b/docs/configuration.md index 89e91411..b78474f4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -892,6 +892,16 @@ metadata_max_size: "100MB" # default Or via environment variable: `PROXY_METADATA_MAX_SIZE=250MB`. +### Artifact size limit + +Artifacts have no size limit by default. `artifact_max_size` refuses any artifact larger than the limit, so a client cannot fill storage or tie up the proxy by requesting huge downloads. An artifact whose upstream `Content-Length` is over the limit is refused before its body is read; one without a declared size fails once the download passes the limit. Either way nothing is cached and the client gets `502 Bad Gateway`, except when artifacts are streamed: one without a declared size has already started sending when it passes the limit, so the client sees the download cut off. The limit applies to `serve`, `mirror` and streamed artifacts. + +```yaml +artifact_max_size: "2GB" # default: unlimited ("" or "0") +``` + +Or via environment variable: `PROXY_ARTIFACT_MAX_SIZE=2GB`. + ### Rewritten metadata cache The npm and Composer handlers rewrite every metadata document they serve, so that download URLs point at the proxy. That means decoding the whole document and encoding it again, and Composer's minified documents are also expanded, which together cost milliseconds per typical package and far more for very large ones. The proxy keeps rewritten documents in memory, so each distinct upstream document is rewritten once and requests arriving while it is being rewritten wait for that rewrite. New bytes from upstream are rewritten again. With version cooldown enabled the cache is bypassed, because cooldown filtering depends on the current time. diff --git a/docs/swagger/docs.go b/docs/swagger/docs.go index dd50bf92..798b50ac 100644 --- a/docs/swagger/docs.go +++ b/docs/swagger/docs.go @@ -497,6 +497,12 @@ const docTemplate = `{ "schema": { "$ref": "#/definitions/server.ErrorResponse" } + }, + "503": { + "description": "Service Unavailable", + "schema": { + "$ref": "#/definitions/server.ErrorResponse" + } } } } diff --git a/docs/swagger/swagger.json b/docs/swagger/swagger.json index 50007b6e..fd175bf9 100644 --- a/docs/swagger/swagger.json +++ b/docs/swagger/swagger.json @@ -490,6 +490,12 @@ "schema": { "$ref": "#/definitions/server.ErrorResponse" } + }, + "503": { + "description": "Service Unavailable", + "schema": { + "$ref": "#/definitions/server.ErrorResponse" + } } } } diff --git a/internal/config/config.go b/internal/config/config.go index b79a2463..7522363c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -167,6 +167,11 @@ type Config struct { // size return ErrMetadataTooLarge. Default: "100MB". MetadataMaxSize string `json:"metadata_max_size" yaml:"metadata_max_size"` + // ArtifactMaxSize is the largest upstream artifact the proxy will fetch + // (e.g. "2GB"). Larger artifacts are refused and not cached. Default: "" + // (unlimited). "0" is also unlimited. + ArtifactMaxSize string `json:"artifact_max_size" yaml:"artifact_max_size"` + // MetadataRewriteCacheSize is how much rewritten npm and Composer // metadata to keep in memory, so a document is rewritten once rather than // on every request (e.g. "256MB", "1GB"). Default: "256MB". Set to "0" to @@ -1175,6 +1180,7 @@ func (c *Config) LoadFromEnv() { setEnvBool(&c.MirrorAPI, "PROXY_MIRROR_API") setEnvString(&c.MetadataTTL, "PROXY_METADATA_TTL") setEnvString(&c.MetadataMaxSize, "PROXY_METADATA_MAX_SIZE") + setEnvString(&c.ArtifactMaxSize, "PROXY_ARTIFACT_MAX_SIZE") setEnvString(&c.MetadataRewriteCacheSize, "PROXY_METADATA_REWRITE_CACHE_SIZE") setEnvString(&c.HTTPTimeout, "PROXY_HTTP_TIMEOUT") setEnvBool(&c.Gradle.BuildCache.ReadOnly, "PROXY_GRADLE_BUILD_CACHE_READ_ONLY") @@ -1272,7 +1278,7 @@ func (c *Config) Validate() error { } } - if err := validateMetadataMaxSize(c.MetadataMaxSize); err != nil { + if err := c.validateDownloadSizeLimits(); err != nil { return err } @@ -1444,6 +1450,27 @@ func validateMetadataMaxSize(s string) error { return nil } +func (c *Config) validateDownloadSizeLimits() error { + if err := validateMetadataMaxSize(c.MetadataMaxSize); err != nil { + return err + } + return validateArtifactMaxSize(c.ArtifactMaxSize) +} + +func validateArtifactMaxSize(s string) error { + if s == "" { + return nil + } + size, err := ParseSize(s) + if err != nil { + return fmt.Errorf("invalid artifact_max_size: %w", err) + } + if size < 0 { + return fmt.Errorf("invalid artifact_max_size %q: must not be negative", s) + } + return nil +} + func validateMetadataRewriteCacheSize(s string) error { if s == "" || s == "0" { return nil @@ -1487,6 +1514,19 @@ func (c *Config) ParseMetadataMaxSize() int64 { return size } +// ParseArtifactMaxSize returns the artifact size limit in bytes, or 0 for +// no limit. +func (c *Config) ParseArtifactMaxSize() int64 { + if c.ArtifactMaxSize == "" { + return 0 + } + size, err := ParseSize(c.ArtifactMaxSize) + if err != nil || size < 0 { + return 0 + } + return size +} + // ParseHTTPTimeout returns the upstream HTTP client timeout. // Returns 30s if unset, 0 (no timeout) if explicitly set to "0". func (c *Config) ParseHTTPTimeout() time.Duration { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 9bee2dca..55ea392b 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1669,3 +1669,52 @@ func TestLoadCacheArtifactsFromEnv(t *testing.T) { t.Error("Storage.CacheArtifacts should be false") } } + +func TestParseArtifactMaxSize(t *testing.T) { + tests := []struct { + name string + size string + want int64 + }{ + {"unset is unlimited", "", 0}, + {"zero is unlimited", "0", 0}, + {"explicit value", "2GB", 2 << 30}, + {"bytes", "1024", 1024}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := Default() + cfg.ArtifactMaxSize = tt.size + if got := cfg.ParseArtifactMaxSize(); got != tt.want { + t.Errorf("ParseArtifactMaxSize() = %d, want %d", got, tt.want) + } + }) + } +} + +func TestValidateArtifactMaxSize(t *testing.T) { + for _, invalid := range []string{"not-a-size", "-1"} { + cfg := Default() + cfg.ArtifactMaxSize = invalid + if err := cfg.Validate(); err == nil { + t.Errorf("expected validation error for artifact_max_size %q", invalid) + } + } + for _, valid := range []string{"", "0", "2GB"} { + cfg := Default() + cfg.ArtifactMaxSize = valid + if err := cfg.Validate(); err != nil { + t.Errorf("unexpected error for artifact_max_size %q: %v", valid, err) + } + } +} + +func TestLoadArtifactMaxSizeFromEnv(t *testing.T) { + cfg := Default() + t.Setenv("PROXY_ARTIFACT_MAX_SIZE", "2GB") + cfg.LoadFromEnv() + + if cfg.ArtifactMaxSize != "2GB" { + t.Errorf("ArtifactMaxSize = %q, want %q", cfg.ArtifactMaxSize, "2GB") + } +} diff --git a/internal/handler/artifact_size.go b/internal/handler/artifact_size.go new file mode 100644 index 00000000..d4c733b3 --- /dev/null +++ b/internal/handler/artifact_size.go @@ -0,0 +1,76 @@ +package handler + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + + "github.com/git-pkgs/registries/fetch" +) + +// ErrArtifactTooLarge is returned for an upstream artifact larger than the +// configured artifact size limit. +var ErrArtifactTooLarge = errors.New("artifact exceeds size limit") + +// LimitArtifactSize wraps fetcher so no artifact over maxSize bytes is +// fetched. An artifact whose declared size is over the limit is refused +// before its body is read; one of unknown size fails once reading passes the +// limit, so the store or stream consuming it fails and nothing is cached. +// A maxSize of zero or less returns fetcher unchanged. +func LimitArtifactSize(fetcher fetch.FetcherInterface, maxSize int64) fetch.FetcherInterface { //nolint:ireturn // returns fetcher itself when there is no limit + if maxSize <= 0 { + return fetcher + } + return &sizeLimitedFetcher{FetcherInterface: fetcher, maxSize: maxSize} +} + +type sizeLimitedFetcher struct { + fetch.FetcherInterface + maxSize int64 +} + +func (f *sizeLimitedFetcher) Fetch(ctx context.Context, url string) (*fetch.Artifact, error) { + return f.limit(f.FetcherInterface.Fetch(ctx, url)) +} + +func (f *sizeLimitedFetcher) FetchWithHeaders(ctx context.Context, url string, headers http.Header) (*fetch.Artifact, error) { + return f.limit(f.FetcherInterface.FetchWithHeaders(ctx, url, headers)) +} + +func (f *sizeLimitedFetcher) limit(artifact *fetch.Artifact, err error) (*fetch.Artifact, error) { + if err != nil { + return nil, err + } + if artifact.Size > f.maxSize { + _ = artifact.Body.Close() + return nil, fmt.Errorf("%w: upstream declared %d bytes, limit is %d", ErrArtifactTooLarge, artifact.Size, f.maxSize) + } + limited := *artifact + limited.Body = &sizeLimitedBody{ReadCloser: artifact.Body, remaining: f.maxSize} + return &limited, nil +} + +// sizeLimitedBody fails with ErrArtifactTooLarge once more than the limit has +// been read. It reads one byte past the limit to tell an artifact of exactly +// the limit from a larger one. +type sizeLimitedBody struct { + io.ReadCloser + remaining int64 +} + +func (b *sizeLimitedBody) Read(p []byte) (int, error) { + if b.remaining < 0 { + return 0, ErrArtifactTooLarge + } + if int64(len(p)) > b.remaining+1 { + p = p[:b.remaining+1] + } + n, err := b.ReadCloser.Read(p) + b.remaining -= int64(n) + if b.remaining < 0 { + return n + int(b.remaining), ErrArtifactTooLarge + } + return n, err +} diff --git a/internal/handler/artifact_size_test.go b/internal/handler/artifact_size_test.go new file mode 100644 index 00000000..f6fe9af1 --- /dev/null +++ b/internal/handler/artifact_size_test.go @@ -0,0 +1,115 @@ +package handler + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/git-pkgs/registries/fetch" +) + +type closeRecorder struct { + io.Reader + closed bool +} + +func (c *closeRecorder) Close() error { + c.closed = true + return nil +} + +func TestLimitArtifactSizeUnlimitedReturnsFetcher(t *testing.T) { + fetcher := &mockFetcher{} + if got := LimitArtifactSize(fetcher, 0); got != fetcher { + t.Errorf("LimitArtifactSize(f, 0) = %T, want the fetcher itself", got) + } +} + +func TestLimitArtifactSizeRefusesDeclaredOversize(t *testing.T) { + body := &closeRecorder{Reader: strings.NewReader("0123456789")} + limited := LimitArtifactSize(&mockFetcher{artifact: &fetch.Artifact{Body: body, Size: 10}}, 4) + + _, err := limited.Fetch(context.Background(), "https://registry.example/a.tgz") + if !errors.Is(err, ErrArtifactTooLarge) { + t.Fatalf("Fetch error = %v, want ErrArtifactTooLarge", err) + } + if !body.closed { + t.Error("refused body was not closed") + } +} + +func TestLimitArtifactSizeFailsUndeclaredOversizeWhileReading(t *testing.T) { + limited := LimitArtifactSize(&mockFetcher{artifact: &fetch.Artifact{ + Body: io.NopCloser(strings.NewReader("0123456789")), Size: -1, + }}, 4) + + artifact, err := limited.Fetch(context.Background(), "https://registry.example/a.tgz") + if err != nil { + t.Fatalf("Fetch failed: %v", err) + } + if _, err := io.ReadAll(artifact.Body); !errors.Is(err, ErrArtifactTooLarge) { + t.Errorf("reading body error = %v, want ErrArtifactTooLarge", err) + } +} + +func TestLimitArtifactSizeAllowsArtifactAtLimit(t *testing.T) { + limited := LimitArtifactSize(&mockFetcher{artifact: &fetch.Artifact{ + Body: io.NopCloser(strings.NewReader("0123")), Size: -1, + }}, 4) + + artifact, err := limited.FetchWithHeaders(context.Background(), "https://registry.example/a.tgz", nil) + if err != nil { + t.Fatalf("Fetch failed: %v", err) + } + data, err := io.ReadAll(artifact.Body) + if err != nil || string(data) != "0123" { + t.Errorf("body = %q, %v; want %q, nil", data, err, "0123") + } +} + +func TestOversizedArtifactIsNotCached(t *testing.T) { + proxy, db, store, fetcher := setupTestProxy(t) + fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("0123456789")), Size: -1} + proxy.Fetcher = LimitArtifactSize(fetcher, 4) + + _, err := proxy.GetOrFetchArtifactFromURL(context.Background(), "npm", "big", "1.0.0", "big-1.0.0.tgz", "https://registry.example/big-1.0.0.tgz") + if !errors.Is(err, ErrArtifactTooLarge) { + t.Fatalf("GetOrFetchArtifactFromURL error = %v, want ErrArtifactTooLarge", err) + } + if len(store.files) != 0 { + t.Errorf("storage holds %d objects after an oversized fetch, want 0", len(store.files)) + } + record, err := db.GetArtifact("pkg:npm/big@1.0.0", "big-1.0.0.tgz") + if err != nil { + t.Fatalf("GetArtifact failed: %v", err) + } + if record != nil { + t.Error("oversized artifact was recorded in the cache database") + } + + rec := httptest.NewRecorder() + proxy.serveArtifactError(rec, err, "failed to fetch package") + if rec.Code != http.StatusBadGateway { + t.Errorf("status = %d, want %d", rec.Code, http.StatusBadGateway) + } +} + +func TestOversizedStreamedArtifactFailsTheStream(t *testing.T) { + proxy, _, _, fetcher := setupTestProxy(t) + proxy.StreamArtifacts = true + fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("0123456789")), Size: -1} + proxy.Fetcher = LimitArtifactSize(fetcher, 4) + + result, err := proxy.GetOrFetchArtifactFromURL(context.Background(), "npm", "big", "1.0.0", "big-1.0.0.tgz", "https://registry.example/big-1.0.0.tgz") + if err != nil { + t.Fatalf("GetOrFetchArtifactFromURL failed: %v", err) + } + defer func() { _ = result.Reader.Close() }() + if _, err := io.ReadAll(result.Reader); !errors.Is(err, ErrArtifactTooLarge) { + t.Errorf("streamed body error = %v, want ErrArtifactTooLarge", err) + } +} diff --git a/internal/server/browse.go b/internal/server/browse.go index 26ee63ef..43cb855c 100644 --- a/internal/server/browse.go +++ b/internal/server/browse.go @@ -9,6 +9,7 @@ import ( "net/http" "path" "strings" + "sync" "github.com/git-pkgs/archives" "github.com/git-pkgs/archives/diff" @@ -31,6 +32,38 @@ const ( // memory exhaustion from a single request. const maxBrowseArchiveSize = 512 << 20 // 512 MB +// maxConcurrentCompares caps how many version diffs run at once. Each one +// buffers both archives whole, up to maxBrowseArchiveSize apiece, so without +// a cap parallel requests could exhaust memory. +const maxConcurrentCompares = 2 + +// compareRetryAfter is the Retry-After, in seconds, sent with a diff refused +// because maxConcurrentCompares are already running. +const compareRetryAfter = "5" + +// compareLimiter counts running version diffs. The zero value is ready to use. +type compareLimiter struct { + mu sync.Mutex + active int +} + +// tryAcquire takes a slot if fewer than maxConcurrentCompares are taken. +func (l *compareLimiter) tryAcquire() bool { + l.mu.Lock() + defer l.mu.Unlock() + if l.active >= maxConcurrentCompares { + return false + } + l.active++ + return true +} + +func (l *compareLimiter) release() { + l.mu.Lock() + defer l.mu.Unlock() + l.active-- +} + // firstBrowsableArtifact returns the first cached artifact that can be opened as // an archive, or nil if the version has none. // @@ -594,6 +627,7 @@ type BrowseSourceData struct { // @Success 200 {object} map[string]any // @Failure 404 {object} ErrorResponse // @Failure 500 {object} ErrorResponse +// @Failure 503 {object} ErrorResponse // @Router /ui/api/compare/{ecosystem}/{name}/{fromVersion}/{toVersion} [get] func (s *Server) compareDiff(w http.ResponseWriter, r *http.Request, ecosystem, name, fromVersion, toVersion string) { // Get artifacts for both versions @@ -621,6 +655,13 @@ func (s *Server) compareDiff(w http.ResponseWriter, r *http.Request, ecosystem, return } + if !s.compares.tryAcquire() { + w.Header().Set("Retry-After", compareRetryAfter) + writeError(w, http.StatusServiceUnavailable, ErrCodeBusy, "too many comparisons in progress, try again shortly") + return + } + defer s.compares.release() + // Open both archives fromReader, err := s.storage.Open(r.Context(), fromArtifact.StoragePath.String) if err != nil { diff --git a/internal/server/browse_test.go b/internal/server/browse_test.go index f4f2f9ab..c3352a42 100644 --- a/internal/server/browse_test.go +++ b/internal/server/browse_test.go @@ -471,6 +471,47 @@ func TestHandleCompareDiff(t *testing.T) { ts := newTestServer(t) defer ts.close() + seedCompareVersions(t, ts) + + // Test the compare endpoint + req := httptest.NewRequest("GET", "/ui/api/compare/npm/test-compare/1.0.0/2.0.0", nil) + w := httptest.NewRecorder() + ts.handler.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected status 200, got %d: %s", w.Code, w.Body.String()) + } + + // Parse response + var result map[string]interface{} + if err := json.NewDecoder(w.Body).Decode(&result); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + + // Check that we have files + files, ok := result["files"].([]interface{}) + if !ok { + t.Fatal("response should have files array") + } + + if len(files) == 0 { + t.Error("should have detected file changes") + } + + // Check counts exist + if _, ok := result["files_changed"]; !ok { + t.Error("response should have files_changed") + } + if _, ok := result["files_added"]; !ok { + t.Error("response should have files_added") + } +} + +// seedCompareVersions caches two versions of pkg:npm/test-compare with +// different archives for the compare endpoint. +func seedCompareVersions(t *testing.T, ts *testServer) { + t.Helper() + // Create two test archives with different content archive1Data := createArchiveWithContent(t, map[string]string{ "README.md": "# Version 1\n", @@ -540,38 +581,52 @@ func TestHandleCompareDiff(t *testing.T) { if err := ts.db.UpsertArtifact(artifact2); err != nil { t.Fatalf("failed to upsert artifact: %v", err) } +} - // Test the compare endpoint +func compareTestVersions(ts *testServer) *httptest.ResponseRecorder { req := httptest.NewRequest("GET", "/ui/api/compare/npm/test-compare/1.0.0/2.0.0", nil) w := httptest.NewRecorder() ts.handler.ServeHTTP(w, req) + return w +} - if w.Code != http.StatusOK { - t.Fatalf("expected status 200, got %d: %s", w.Code, w.Body.String()) - } +// Each comparison buffers both archives in memory, so only a few may run at +// once; the rest are refused instead of piling up until the process runs +// out of memory. +func TestCompareDiffRefusedWhileComparisonsAreRunning(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + seedCompareVersions(t, ts) - // Parse response - var result map[string]interface{} - if err := json.NewDecoder(w.Body).Decode(&result); err != nil { - t.Fatalf("failed to decode response: %v", err) + for range maxConcurrentCompares { + if !ts.server.compares.tryAcquire() { + t.Fatal("could not take a free comparison slot") + } } - // Check that we have files - files, ok := result["files"].([]interface{}) - if !ok { - t.Fatal("response should have files array") + w := compareTestVersions(ts) + if w.Code != http.StatusServiceUnavailable { + t.Fatalf("status with all slots taken = %d, want %d", w.Code, http.StatusServiceUnavailable) } - - if len(files) == 0 { - t.Error("should have detected file changes") + if w.Header().Get("Retry-After") == "" { + t.Error("refused comparison has no Retry-After header") } - // Check counts exist - if _, ok := result["files_changed"]; !ok { - t.Error("response should have files_changed") + ts.server.compares.release() + if w := compareTestVersions(ts); w.Code != http.StatusOK { + t.Fatalf("status after a slot was freed = %d, want %d: %s", w.Code, http.StatusOK, w.Body.String()) } - if _, ok := result["files_added"]; !ok { - t.Error("response should have files_added") +} + +func TestCompareDiffReleasesItsSlot(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + seedCompareVersions(t, ts) + + for i := range maxConcurrentCompares + 1 { + if w := compareTestVersions(ts); w.Code != http.StatusOK { + t.Fatalf("comparison %d: status = %d, want %d: %s", i, w.Code, http.StatusOK, w.Body.String()) + } } } diff --git a/internal/server/errors.go b/internal/server/errors.go index 474ecd75..d30bff92 100644 --- a/internal/server/errors.go +++ b/internal/server/errors.go @@ -12,6 +12,7 @@ const ( ErrCodeNotFound = "NOT_FOUND" ErrCodeUpstream = "UPSTREAM_ERROR" ErrCodeInternal = "INTERNAL_ERROR" + ErrCodeBusy = "BUSY" ) // ErrorResponse is the JSON body returned for API errors. diff --git a/internal/server/server.go b/internal/server/server.go index 8ee1c623..3bb51a61 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -122,6 +122,8 @@ type Server struct { // browseLimits overrides defaultBrowseLimits when set; tests use it. browseLimits archives.StreamOptions + + compares compareLimiter } // New creates a new Server with the given configuration. @@ -245,7 +247,7 @@ func (s *Server) serve(listener net.Listener) error { Ecosystems: s.cfg.Cooldown.Ecosystems, Packages: s.cfg.Cooldown.NormalizedPackages(), } - proxy := handler.NewProxy(s.db, s.storage, fetcher, resolver, s.logger) + proxy := handler.NewProxy(s.db, s.storage, handler.LimitArtifactSize(fetcher, s.cfg.ParseArtifactMaxSize()), resolver, s.logger) proxy.HTTPClient = &metadataClient proxy.AuthForURL = s.authForURL cooldownPolicy, err := cooldownpolicy.New(cd, s.cfg.Cooldown.PackagePatterns) diff --git a/internal/storage/blob.go b/internal/storage/blob.go index 4cadf8af..7a34d7de 100644 --- a/internal/storage/blob.go +++ b/internal/storage/blob.go @@ -207,14 +207,20 @@ func (b *Blob) Store(ctx context.Context, path string, r io.Reader) (int64, stri h := sha256.New() tee := io.TeeReader(r, h) + // gocloud commits whatever was written when the writer closes, unless + // its context is cancelled first, so a failed copy cancels it to abort. + writeCtx, abort := context.WithCancel(ctx) + defer abort() + opts := &blob.WriterOptions{} - w, err := b.bucket.NewWriter(ctx, path, opts) + w, err := b.bucket.NewWriter(writeCtx, path, opts) if err != nil { return 0, "", fmt.Errorf("creating writer: %w", err) } size, err := io.Copy(w, tee) if err != nil { + abort() _ = w.Close() return 0, "", fmt.Errorf("writing content: %w", err) } diff --git a/internal/storage/blob_test.go b/internal/storage/blob_test.go index efc1b765..5533174f 100644 --- a/internal/storage/blob_test.go +++ b/internal/storage/blob_test.go @@ -660,3 +660,41 @@ func TestStoreSucceedsWhenSidecarCannotBeRemoved(t *testing.T) { t.Errorf("Store failed because cleanup could not complete: %v", err) } } + +type failingReader struct { + data []byte + err error +} + +func (r *failingReader) Read(p []byte) (int, error) { + if len(r.data) == 0 { + return 0, r.err + } + n := copy(p, r.data) + r.data = r.data[n:] + return n, nil +} + +// A download that breaks off midway must not leave its first half behind as +// an object, or every failed fetch adds an orphan nothing ever deletes. +func TestBlobStoreFailedCopyLeavesNoObject(t *testing.T) { + b := createTestBlob(t) + ctx := context.Background() + readErr := errors.New("connection reset") + + _, _, err := b.Store(ctx, "npm/lodash/4.17.21/lodash.tgz", &failingReader{ + data: []byte(strings.Repeat("partial", 1024)), + err: readErr, + }) + if !errors.Is(err, readErr) { + t.Fatalf("Store error = %v, want %v", err, readErr) + } + + exists, err := b.Exists(ctx, "npm/lodash/4.17.21/lodash.tgz") + if err != nil { + t.Fatalf("Exists failed: %v", err) + } + if exists { + t.Error("failed Store left a partial object behind") + } +}