Stream npm packuments through storage when rewriting them

cmd/proxy/main.go

@@ -554,6 +554,7 @@ func runMirror() {
 	proxy.CacheMetadata = true // mirror always caches metadata
 	proxy.MetadataTTL = cfg.ParseMetadataTTL()
 	proxy.MetadataMaxSize = cfg.ParseMetadataMaxSize()
+	proxy.MetadataStreamMaxSize = cfg.ParseMetadataStreamMaxSize()
 	// As serve does, resolve Swift source repositories in the denylist through
 	// the upstream, and refuse to run when one can't be.
 	if err := server.ConfigureDenylist(cfg, proxy); err != nil {

docs/architecture.md

@@ -316,7 +316,7 @@ HTTP protocol handlers for each registry type.
 
 ### `internal/metadata/npm`
 
-The npm metadata rewrite, separated from the handler so it can run where the proxy's infrastructure can't, such as WASI. It takes a packument's bytes and everything else as explicit inputs (proxy URL, cooldown policy, the time cooldown is evaluated at, denied versions, pushed versions) and returns the rewritten document with the versions it kept. It filters versions by cooldown and the denylist, moves `latest` off a withheld version, lists pushed versions in place of upstream ones, and points each tarball URL at the proxy, copying everything else byte for byte. It does no I/O and never reads the clock; callbacks report what it did, which the handler logs. It imports no database, storage, metrics, HTTP or logging packages, which a test checks.
+The npm metadata rewrite, separated from the handler so it can run where the proxy's infrastructure can't, such as WASI. It takes a packument's bytes and everything else as explicit inputs (proxy URL, cooldown policy, the time cooldown is evaluated at, denied versions, pushed versions) and returns the rewritten document with the versions it kept. It filters versions by cooldown and the denylist, moves `latest` off a withheld version, lists pushed versions in place of upstream ones, and points each tarball URL at the proxy, copying everything else byte for byte. It does no I/O and never reads the clock; callbacks report what it did, which the handler logs. `ScanPackument` and `RewriteStream` do the same rewrite over a reader in two passes, the first collecting the time map, dist-tags and version names and the second copying each kept version, so the handler can rewrite a packument from storage without holding it in memory. It imports no database, storage, metrics, HTTP or logging packages, which a test checks.
 
 ### `internal/jsonscan`
 

docs/configuration.md

@@ -1411,13 +1411,14 @@ When upstream is unreachable and the cached entry is past its TTL, the proxy ser
 
 ### Metadata size limit
 
-Upstream metadata responses are buffered in memory before being rewritten and served. `metadata_max_size` caps that buffer to protect against OOM from a misbehaving upstream. Some npm packages with thousands of versions (for example `renovate`) exceed the 100 MB default, so raise this if you see `metadata response exceeds size limit` in the logs. Conda `repodata.json` with cooldown enabled is filtered as it streams and isn't held in memory decompressed. With `cache_metadata` the limit applies to the compressed document as it does without cooldown; one over it is filtered as it streams from the upstream and not cached.
+Upstream metadata responses are buffered in memory before being rewritten and served. `metadata_max_size` caps that buffer to protect against OOM from a misbehaving upstream. Some npm packages with thousands of versions (for example `renovate`) exceed the 100 MB default, so raise this if you see `metadata response exceeds size limit` in the logs. Conda `repodata.json` with cooldown enabled is filtered as it streams and isn't held in memory decompressed. With `cache_metadata` the limit applies to the compressed document as it does without cooldown; one over it is filtered as it streams from the upstream and not cached. With `cache_metadata` and the metadata rewrite cache on, an npm packument is streamed from the upstream into storage and rewritten from there in two passes, so it isn't held in memory and the limit doesn't apply to it; the rewritten copy is kept gzip-compressed. Requests for a single version (`/<package>/<version>`) and tarball downloads read the version from the stored copy the same way. `metadata_stream_max_size` (1 GB by default) caps a document streamed this way; one over it isn't stored, and the stored copy, if any, is served instead.
 
 ```yaml
-metadata_max_size: "100MB"   # default
+metadata_max_size: "100MB"          # default
+metadata_stream_max_size: "1GB"     # default
 ```
 
-Or via environment variable: `PROXY_METADATA_MAX_SIZE=250MB`.
+Or via environment variable: `PROXY_METADATA_MAX_SIZE=250MB`, and `PROXY_METADATA_STREAM_MAX_SIZE=2GB` for `metadata_stream_max_size`.
 
 ### Artifact size limit
 

internal/config/config.go

@@ -170,6 +170,11 @@ type Config struct {
 	// size return ErrMetadataTooLarge. Default: "100MB".
 	MetadataMaxSize string `json:"metadata_max_size" yaml:"metadata_max_size"`
 
+	// MetadataStreamMaxSize is the largest metadata document the proxy
+	// streams from upstream into storage instead of buffering it, which
+	// metadata_max_size doesn't cover (e.g. "1GB"). Default: "1GB".
+	MetadataStreamMaxSize string `json:"metadata_stream_max_size" yaml:"metadata_stream_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.
@@ -1481,6 +1486,7 @@ func (c *Config) LoadFromEnv() {
 	setEnvStringSlice(&c.Hosted.ComposerVendors, "PROXY_HOSTED_COMPOSER_VENDORS")
 	setEnvString(&c.MetadataTTL, "PROXY_METADATA_TTL")
 	setEnvString(&c.MetadataMaxSize, "PROXY_METADATA_MAX_SIZE")
+	setEnvString(&c.MetadataStreamMaxSize, "PROXY_METADATA_STREAM_MAX_SIZE")
 	setEnvString(&c.ArtifactMaxSize, "PROXY_ARTIFACT_MAX_SIZE")
 	setEnvString(&c.MetadataRewriteCacheSize, "PROXY_METADATA_REWRITE_CACHE_SIZE")
 	setEnvString(&c.HTTPTimeout, "PROXY_HTTP_TIMEOUT")
@@ -1727,6 +1733,7 @@ const (
 	defaultHTTPTimeout                   = 30 * time.Second //nolint:mnd // sensible default
 	defaultHitFlushInterval              = time.Second
 	defaultMetadataMaxSize               = 100 << 20
+	defaultMetadataStreamMaxSize         = 1 << 30
 	defaultMetadataRewriteCacheSize      = 256 << 20
 	defaultGradleBuildCacheMaxUploadSize = 100 << 20
 	defaultGradleBuildCacheSweepInterval = 10 * time.Minute
@@ -1780,22 +1787,25 @@ func validateRequestTimeout(field, s string) error {
 	return nil
 }
 
-func validateMetadataMaxSize(s string) error {
+func validateMetadataMaxSize(name, s string) error {
 	if s == "" {
 		return nil
 	}
 	size, err := ParseSize(s)
 	if err != nil {
-		return fmt.Errorf("invalid metadata_max_size: %w", err)
+		return fmt.Errorf("invalid %s: %w", name, err)
 	}
 	if size <= 0 {
-		return fmt.Errorf("invalid metadata_max_size %q: must be positive", s)
+		return fmt.Errorf("invalid %s %q: must be positive", name, s)
 	}
 	return nil
 }
 
 func (c *Config) validateDownloadSizeLimits() error {
-	if err := validateMetadataMaxSize(c.MetadataMaxSize); err != nil {
+	if err := validateMetadataMaxSize("metadata_max_size", c.MetadataMaxSize); err != nil {
+		return err
+	}
+	if err := validateMetadataMaxSize("metadata_stream_max_size", c.MetadataStreamMaxSize); err != nil {
 		return err
 	}
 	return validateArtifactMaxSize(c.ArtifactMaxSize)
@@ -1858,6 +1868,19 @@ func (c *Config) ParseMetadataMaxSize() int64 {
 	return size
 }
 
+// ParseMetadataStreamMaxSize returns the size limit for metadata streamed
+// into storage in bytes. Returns 1GB if unset or invalid.
+func (c *Config) ParseMetadataStreamMaxSize() int64 {
+	if c.MetadataStreamMaxSize == "" {
+		return defaultMetadataStreamMaxSize
+	}
+	size, err := ParseSize(c.MetadataStreamMaxSize)
+	if err != nil || size <= 0 {
+		return defaultMetadataStreamMaxSize
+	}
+	return size
+}
+
 // ParseArtifactMaxSize returns the artifact size limit in bytes, or 0 for
 // no limit.
 func (c *Config) ParseArtifactMaxSize() int64 {

internal/config/config_test.go

@@ -785,6 +785,29 @@ func TestParseMetadataMaxSize(t *testing.T) {
 	}
 }
 
+func TestParseMetadataStreamMaxSize(t *testing.T) {
+	cfg := Default()
+	if got := cfg.ParseMetadataStreamMaxSize(); got != defaultMetadataStreamMaxSize {
+		t.Errorf("unset: %d, want %d", got, defaultMetadataStreamMaxSize)
+	}
+	cfg.MetadataStreamMaxSize = "2GB"
+	if got := cfg.ParseMetadataStreamMaxSize(); got != 2<<30 {
+		t.Errorf("2GB: %d", got)
+	}
+	t.Setenv("PROXY_METADATA_STREAM_MAX_SIZE", "300MB")
+	cfg = Default()
+	cfg.LoadFromEnv()
+	if got := cfg.ParseMetadataStreamMaxSize(); got != 300<<20 {
+		t.Errorf("from env: %d", got)
+	}
+	for _, bad := range []string{"0", "lots"} {
+		cfg.MetadataStreamMaxSize = bad
+		if err := cfg.Validate(); err == nil {
+			t.Errorf("metadata_stream_max_size %q validated", bad)
+		}
+	}
+}
+
 func TestValidateMetadataMaxSize(t *testing.T) {
 	cfg := Default()
 	cfg.MetadataMaxSize = "not-a-size"

internal/handler/conda.go

@@ -409,7 +409,8 @@ func (h *CondaHandler) rememberRepodata(cacheKey, digest string, filter *repodat
 	h.repodataStates[cacheKey] = repodataState{digest: digest, etag: etag, until: filter.until}
 }
 
-// countingWriter records whether anything has reached the response yet.
+// countingWriter counts the bytes written through it, such as whether
+// anything has reached the response yet.
 type countingWriter struct {
 	w io.Writer
 	n int64

internal/handler/handler.go

@@ -142,6 +142,10 @@ func ifNoneMatchHits(header, etag string) bool {
 // defaultMetadataMaxSize is used when Proxy.MetadataMaxSize is unset.
 const defaultMetadataMaxSize = 100 << 20
 
+// defaultMetadataStreamMaxSize is used when Proxy.MetadataStreamMaxSize is
+// unset.
+const defaultMetadataStreamMaxSize = 1 << 30
+
 // ErrMetadataTooLarge is returned when upstream metadata exceeds the configured limit.
 var ErrMetadataTooLarge = errors.New("metadata response exceeds size limit")
 
@@ -215,14 +219,17 @@ type Proxy struct {
 	Cooldown CooldownPolicy
 	// ShortestCooldown is the shortest cooldown configured anywhere. It
 	// bounds how long a rewrite filtered by cooldown is cached.
-	ShortestCooldown    time.Duration
-	Denylist            *denylist.Policy
-	CacheMetadata       bool
-	MetadataTTL         time.Duration
-	MetadataMaxSize     int64
-	GradleReadOnly      bool
-	HostedEnabled       bool
-	GradleMaxUploadSize int64
+	ShortestCooldown time.Duration
+	Denylist         *denylist.Policy
+	CacheMetadata    bool
+	MetadataTTL      time.Duration
+	MetadataMaxSize  int64
+	// MetadataStreamMaxSize caps a metadata document streamed from upstream
+	// into storage, which metadata_max_size doesn't cover. Default 1 GB.
+	MetadataStreamMaxSize int64
+	GradleReadOnly        bool
+	HostedEnabled         bool
+	GradleMaxUploadSize   int64
 	// NPMFullMetadata requests full npm packuments from upstream even when
 	// cooldown is disabled, so served metadata carries publish times.
 	NPMFullMetadata bool
@@ -1816,6 +1823,10 @@ type metadataResult struct {
 	// fromCache is set when the body is a stored copy within its TTL, read
 	// without asking upstream.
 	fromCache bool
+	// entry is the cache row of the stored document, for a lookup that
+	// leaves the document in storage (see storedMetadataDocument). body is
+	// nil then.
+	entry *database.MetadataCacheEntry
 }
 
 // describedBy returns res with the digest that entry, the cache row its body
@@ -1955,24 +1966,9 @@ type upstreamMetadata struct {
 // alongside for the caller to store and replay. An empty acceptEncoding leaves Go to negotiate and
 // decompress transparently.
 func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, entry *database.MetadataCacheEntry, accept, acceptEncoding string) (*upstreamMetadata, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil)
-	if err != nil {
-		return nil, fmt.Errorf("creating request: %w", err)
-	}
-	req.Header.Set("Accept", accept)
-	if acceptEncoding != "" {
-		req.Header.Set(headerAcceptEncoding, acceptEncoding)
-	}
-
-	if entry != nil && entry.ETag.Valid {
-		req.Header.Set("If-None-Match", entry.ETag.String)
-	} else if entry != nil && entry.LastModified.Valid {
-		req.Header.Set("If-Modified-Since", entry.LastModified.Time.UTC().Format(http.TimeFormat))
-	}
-
-	resp, err := p.HTTPClient.Do(req)
+	resp, err := p.requestUpstreamMetadata(ctx, upstreamURL, entry, accept, acceptEncoding)
 	if err != nil {
-		return nil, fmt.Errorf("fetching metadata: %w", err)
+		return nil, err
 	}
 	defer func() { _ = resp.Body.Close() }()
 
@@ -2000,12 +1996,8 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e
 		return meta, nil
 	}
 
-	if resp.StatusCode == http.StatusNotFound {
-		body, _ := io.ReadAll(io.LimitReader(resp.Body, upstreamNotFoundBodyLimit))
-		return nil, &upstreamNotFoundError{body: body}
-	}
-	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("upstream returned %d", resp.StatusCode)
+	if err := upstreamMetadataStatus(resp); err != nil {
+		return nil, err
 	}
 
 	// ContentLength is -1 when the transport decompresses the body, and
@@ -2030,6 +2022,44 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e
 	return meta, nil
 }
 
+// requestUpstreamMetadata sends the request for a metadata document,
+// revalidating entry when it is given.
+func (p *Proxy) requestUpstreamMetadata(ctx context.Context, upstreamURL string, entry *database.MetadataCacheEntry, accept, acceptEncoding string) (*http.Response, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil)
+	if err != nil {
+		return nil, fmt.Errorf("creating request: %w", err)
+	}
+	req.Header.Set("Accept", accept)
+	if acceptEncoding != "" {
+		req.Header.Set(headerAcceptEncoding, acceptEncoding)
+	}
+
+	if entry != nil && entry.ETag.Valid {
+		req.Header.Set("If-None-Match", entry.ETag.String)
+	} else if entry != nil && entry.LastModified.Valid {
+		req.Header.Set("If-Modified-Since", entry.LastModified.Time.UTC().Format(http.TimeFormat))
+	}
+
+	resp, err := p.HTTPClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("fetching metadata: %w", err)
+	}
+	return resp, nil
+}
+
+// upstreamMetadataStatus returns the error for a metadata response other
+// than 200. Callers handle a 304 they asked for before.
+func upstreamMetadataStatus(resp *http.Response) error {
+	switch resp.StatusCode {
+	case http.StatusOK:
+		return nil
+	case http.StatusNotFound:
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, upstreamNotFoundBodyLimit))
+		return &upstreamNotFoundError{body: body}
+	}
+	return fmt.Errorf("upstream returned %d", resp.StatusCode)
+}
+
 // refreshMetadataBlob records that upstream answered 304 for the cached
 // document meta was read from, by moving the row's fetched_at on. The bytes in
 // storage are unchanged, so they are neither written nor hashed again. It
@@ -2057,14 +2087,22 @@ func (p *Proxy) refreshMetadataBlob(ecosystem, cacheKey string, meta *upstreamMe
 // database. It returns the digest it recorded for the bytes, or "" when they
 // were not cached.
 func (p *Proxy) cacheMetadataBlob(ctx context.Context, ecosystem, cacheKey, storagePath string, meta *upstreamMetadata) string {
+	digest, _ := p.storeMetadataBlob(ctx, ecosystem, cacheKey, storagePath, bytes.NewReader(meta.body), meta)
+	return digest
+}
+
+// storeMetadataBlob is cacheMetadataBlob for a document read from body,
+// which meta describes. It also returns the error that kept it from caching
+// the document.
+func (p *Proxy) storeMetadataBlob(ctx context.Context, ecosystem, cacheKey, storagePath string, body io.Reader, meta *upstreamMetadata) (string, error) {
 	if p.DB == nil || p.Storage == nil {
-		return ""
+		return "", errMetadataNotStored
 	}
 
-	size, hash, err := p.Storage.Store(ctx, storagePath, bytes.NewReader(meta.body))
+	size, hash, err := p.Storage.Store(ctx, storagePath, body)
 	if err != nil {
 		p.Logger.Warn("failed to cache metadata", "ecosystem", ecosystem, "key", cacheKey, "error", err)
-		return ""
+		return "", err
 	}
 	var digest string
 	if hash != "" {
@@ -2094,9 +2132,9 @@ func (p *Proxy) cacheMetadataBlob(ctx context.Context, ecosystem, cacheKey, stor
 		if delErr := p.Storage.Delete(ctx, storagePath); delErr != nil {
 			p.Logger.Warn("failed to discard metadata blob", "ecosystem", ecosystem, "key", cacheKey, "error", delErr)
 		}
-		return ""
+		return "", err
 	}
-	return digest
+	return digest, nil
 }
 
 // currentMetadataEntry re-reads the metadata cache row and returns it, or