From 2fed8901aeb07a033e5f2aa310dbbbb372cbe517 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Sun, 25 Oct 2026 20:29:03 +0100 Subject: [PATCH 1/3] Send artifacts to the client while a cache miss is stored On a cache miss the proxy stored the whole artifact before sending any of it, so clients with an idle timeout gave up on large artifacts: git-lfs after 30 seconds, pacman after 10. The caller that starts a shared fetch now gets the bytes as they arrive while they are stored; requests joining the fetch, HEAD and range requests still wait for the stored copy. The client's copy is checked against the upstream's digest as it is read, and a mismatch aborts the response before it completes. A client going away leaves the copy to storage running. A storage error leaves the client's copy running: the client gets the artifact, and the failure is logged at error level and counted in proxy_artifacts_served_uncached_total. A block-mode scanner has to see the whole artifact first, so with one configured for the ecosystem a miss is handled as before. --- docs/architecture.md | 9 +- docs/configuration.md | 21 +- internal/handler/handler.go | 12 +- internal/handler/stream_miss.go | 255 +++++++++++++++ internal/handler/stream_miss_test.go | 444 +++++++++++++++++++++++++++ internal/metrics/metrics.go | 17 + internal/scanner/group.go | 14 + internal/scanner/group_test.go | 24 ++ internal/server/middleware.go | 5 +- 9 files changed, 787 insertions(+), 14 deletions(-) create mode 100644 internal/handler/stream_miss.go create mode 100644 internal/handler/stream_miss_test.go diff --git a/docs/architecture.md b/docs/architecture.md index e7ce8c8f..b2f31c6c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -58,11 +58,14 @@ Metadata is not cached - always fetched fresh. This ensures clients see new vers **Cache Miss:** - Resolve download URL using Resolver - Fetch artifact from upstream using Fetcher + - Copy the body to Storage and, for a client GET without a `Range` + header, to the client at once (`storeWhileServing`); the client's copy + is checked against the upstream's digest as it is read, and a mismatch + aborts it. A block-mode scanner, `HEAD` and range requests, and + requests joining a fetch already running wait for the stored file. - Store artifact in Storage (returns size, hash) - Create/update database records (package, version, artifact) - - Open stored file - - Return reader to handler - - Handler streams file to client + - Requests that waited open the stored file and stream it to their client Artifact handlers honor single byte ranges when the storage reader supports efficient seeking (currently the local filesystem reader). They advertise diff --git a/docs/configuration.md b/docs/configuration.md index 269d5bdb..f52601b5 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -72,6 +72,14 @@ storage: `storage.max_size` counts cached artifacts and cached metadata documents (`cache_metadata`). When the total goes over the limit, the least recently served entries of either kind are evicted first; an evicted metadata document is fetched from upstream again on the next request for it. Pushed artifacts are not counted or evicted. An artifact replaced by a refetch stays in storage for at least an hour, or `storage.direct_serve_ttl` if longer, so requests already reading it can finish, and storage use can exceed the limit by what was replaced in that time. +### Cache misses + +On a cache miss the proxy sends the artifact to the client while it stores it, so the client gets its first bytes as soon as the upstream sends them, rather than after the whole artifact is stored. Clients that give up after a while without data, such as git-lfs (30 seconds) or pacman (10 seconds), no longer time out on large artifacts. The response goes out chunked, without `Content-Length`, as with `cache_artifacts: false`; once stored, the artifact is served with its size. + +The bytes are checked against the digest the upstream declares, where it declares one, as they are sent. On a mismatch the connection is aborted before the response completes, so the client never receives a tampered artifact as a good one, and nothing is cached. If the client goes away, the artifact is still stored. If storing it fails, for example because the disk is full, the client still gets the whole artifact; the failure is logged at error level and counted in `proxy_artifacts_served_uncached_total`, so a cache that has stopped filling doesn't go unnoticed while every request succeeds. + +Requests for the same artifact that arrive while it is being stored wait for the stored copy, as do `HEAD` requests and requests with a `Range` header, which are answered from it with its size. With a `block`-mode scanner configured for the ecosystem, a miss is stored and scanned before anything is sent, as before (see [Artifact Scanning](#artifact-scanning)). + ### Serving artifacts without storing them With `storage.cache_artifacts: false` the proxy streams every artifact download from upstream to the client and stores no artifacts. Metadata is still filtered and cached, so cooldown and the denylist apply as usual, and the database still holds the publish times cooldown needs. Use it when another caching layer, such as an Artifactory remote repository, sits in front of the proxy and caching artifacts twice only costs storage. @@ -854,13 +862,10 @@ repository, such as a fork, it is copied from there instead of downloaded, once the upstream has confirmed this repository has it; the copy takes its own space. -Large LFS objects may need longer timeouts. The proxy downloads an object it -doesn't have before sending any of it, within `artifact_timeout`, and the -whole response, download included, must fit within `write_timeout`; both -default to 5 minutes. git-lfs gives up on a response after 30 seconds without -data (`lfs.activitytimeout`) and retries. A retry waits for the download still -running, so the object is served once it is cached, but a download that -outlasts git-lfs's retries fails that clone or pull until it finishes. +Large LFS objects may need longer timeouts. An object the proxy doesn't +have is sent to git-lfs as it is downloaded (see [Cache misses](#cache-misses)), +but the download must finish within `artifact_timeout` and the whole response +within `write_timeout`; both default to 5 minutes. Download URLs are often signed; only the URL without its query is logged and recorded with the object. An object is only cached once its content matches its sha256 object ID. Objects are cached per repository, so one @@ -1248,7 +1253,7 @@ scanning: ### How caching defers to a scan verdict -The proxy never uploads artifact bytes to a scanner. When an artifact is fetched from upstream, it's stored to the configured storage backend first, exactly as without scanning. If scanning is enabled for the artifact's ecosystem, the proxy then notifies each applicable scanner with package metadata and a short-lived, HMAC-signed URL pointing at the internal `/_internal/scan-fetch` route; each scanner GETs that URL itself to pull the exact bytes staged in storage and runs its own scan against them. +The proxy never uploads artifact bytes to a scanner. When an artifact is fetched from upstream, it's stored to the configured storage backend first, exactly as without scanning. If scanning is enabled for the artifact's ecosystem, the proxy then notifies each applicable scanner with package metadata and a short-lived, HMAC-signed URL pointing at the internal `/_internal/scan-fetch` route; each scanner GETs that URL itself to pull the exact bytes staged in storage and runs its own scan against them. With a `block`-mode scanner for the ecosystem, the client gets nothing before the verdict; with only `monitor`-mode scanners, a cache miss is sent to the client while it is stored, and the scan runs after (see [Cache misses](#cache-misses)). Scanners configured for the same ecosystem all run concurrently, never sequentially. The moment any `block`-mode scanner reports a not-allowed verdict (or errors, unless `fail_open` is set), the proxy cancels the in-flight calls to the other scanners and deletes the staged artifact — it's never committed to the cache database, so it was never visible to a client. If nothing blocks, the proxy waits for every `block`-mode scanner to finish before caching the artifact and serving it. A `monitor`-mode scanner's findings are logged and never gate the wait or the caching decision, even when it reports not-allowed. diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 363314e8..c96358c1 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -652,7 +652,7 @@ func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, fil } metrics.RecordUpstreamFetch(ecosystem, fetchDuration) - return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, info.URL, "", "", artifact) + return p.storeFetched(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, info.URL, "", "", artifact) } // storeArtifact writes a fetched artifact to storage, verifies it against @@ -848,6 +848,9 @@ func (p *Proxy) coalesceFetch(ctx context.Context, key string, recheck func() (a p.fetchMu.Unlock() if !joined { + if streamsCacheMiss(ctx) { + return p.runSharedFetchStreaming(ctx, key, f, recheck, commit) + } return p.runSharedFetch(ctx, key, f, recheck, commit) } @@ -940,6 +943,11 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch if err != nil { return nil, err } + if h := handoffFrom(ctx); h != nil && h.handedOver() { + // The caller already has the bytes; opening the stored copy would + // cost a read it never uses, a GET on S3 or GCS. + return nil, nil + } return p.openStoredArtifact(ctx, stored, path) } @@ -2339,7 +2347,7 @@ func (p *Proxy) fetchAndCacheFromURL(ctx context.Context, ecosystem, name, versi return artifacts.Artifact{}, "", fmt.Errorf("fetching from upstream: %w", err) } - return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, upstreamHash, declared, artifact) + return p.storeFetched(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, upstreamHash, declared, artifact) } type upstreamLastModifiedKey struct{} diff --git a/internal/handler/stream_miss.go b/internal/handler/stream_miss.go new file mode 100644 index 00000000..9ddcc206 --- /dev/null +++ b/internal/handler/stream_miss.go @@ -0,0 +1,255 @@ +package handler + +import ( + "context" + "errors" + "io" + "strings" + "sync" + + "github.com/git-pkgs/artifacts" + "github.com/git-pkgs/proxy/internal/metrics" + "github.com/git-pkgs/registries/fetch" + "github.com/opencontainers/go-digest" +) + +// On a cache miss the proxy used to store the whole artifact before sending +// any of it, so a client with an idle timeout (git-lfs gives up after 30 +// seconds without data, pacman after 10) could give up on a large one. A +// marked request instead gets the bytes as they arrive from upstream while +// they are stored: the caller that starts the shared fetch is handed a +// stream, and callers joining it still wait for the stored artifact. + +type streamCacheMissKey struct{} + +// StreamCacheMiss marks ctx so that on an artifact cache miss the bytes are +// sent to the client while they are being stored. The server sets it on GET +// requests without a Range header; a range of a stream that is still +// arriving can't be served. +func StreamCacheMiss(ctx context.Context) context.Context { + return context.WithValue(ctx, streamCacheMissKey{}, true) +} + +func streamsCacheMiss(ctx context.Context) bool { + stream, _ := ctx.Value(streamCacheMissKey{}).(bool) + return stream && !servingHead(ctx) +} + +// missHandoff passes the client's stream from the shared fetch, which runs on +// its own goroutine, to the caller that started it. The caller can leave +// before a stream is offered, and the fetch must then not write to a stream +// nobody reads. +type missHandoff struct { + mu sync.Mutex + ready chan struct{} + stream *CacheResult + taken bool + abandoned bool +} + +type missHandoffKey struct{} + +func handoffFrom(ctx context.Context) *missHandoff { + h, _ := ctx.Value(missHandoffKey{}).(*missHandoff) + return h +} + +// offer hands stream to the caller and reports whether it took it. +func (h *missHandoff) offer(stream *CacheResult) bool { + h.mu.Lock() + defer h.mu.Unlock() + if h.abandoned { + return false + } + h.stream, h.taken = stream, true + close(h.ready) + return true +} + +// handedOver reports whether the caller took a stream. +func (h *missHandoff) handedOver() bool { + h.mu.Lock() + defer h.mu.Unlock() + return h.taken +} + +// runSharedFetchStreaming runs the shared fetch on its own goroutine and +// returns as soon as it offers a stream, or with its result if it finishes +// without one: on a recheck hit, an artifact another instance stored, an +// error, or an artifact a block-mode scanner has to see first. Waiters and +// the fetch lease still see the fetch end only once the artifact is stored. +func (p *Proxy) runSharedFetchStreaming(ctx context.Context, key string, f *inflightFetch, recheck func() (artifacts.Artifact, string, bool), commit func(context.Context) (artifacts.Artifact, string, error)) (*CacheResult, error) { + h := &missHandoff{ready: make(chan struct{})} + type outcome struct { + res *CacheResult + err error + } + done := make(chan outcome, 1) + go func() { + res, err := p.runSharedFetch(context.WithValue(ctx, missHandoffKey{}, h), key, f, recheck, commit) + h.mu.Lock() + defer h.mu.Unlock() + if h.taken || h.abandoned { + // The caller has the stream, or has gone; nobody reads this. + if res != nil && res.Reader != nil { + _ = res.Reader.Close() + } + return + } + done <- outcome{res, err} + }() + + select { + case <-h.ready: + return h.stream, nil + case o := <-done: + return o.res, o.err + case <-ctx.Done(): + h.mu.Lock() + defer h.mu.Unlock() + if h.taken { + return h.stream, nil + } + h.abandoned = true + select { + case o := <-done: + if o.res != nil && o.res.Reader != nil { + _ = o.res.Reader.Close() + } + default: + } + return nil, ctx.Err() + } +} + +// storeFetched stores an artifact fetched from upstream. When the caller +// that started the fetch is waiting for a stream, it is sent the bytes as +// they are stored; see storeWhileServing. +func (p *Proxy) storeFetched(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared string, artifact *fetch.Artifact) (artifacts.Artifact, string, error) { + h := handoffFrom(ctx) + if h == nil || p.Scanners.Blocks(ecosystem) { + return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared, artifact) + } + return p.storeWhileServing(ctx, h, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared, artifact) +} + +// errNoReaderLeft stops reading upstream once neither storage nor the +// client takes the bytes any more. +var errNoReaderLeft = errors.New("neither storage nor the client is reading the artifact") + +// storeWhileServing copies the upstream body to storage and to the client at +// once. The client's copy is checked against upstreamHash and declared as it +// is read, and a mismatch fails it before its last bytes, so the response is +// aborted and the client discards it; storeArtifact checks the stored copy +// the same way and doesn't cache it. A client that goes away leaves the copy +// to storage running. A storage error leaves the client's copy running, so +// the client still gets the artifact; that is logged and counted, since every +// request would otherwise succeed while nothing is cached. The client's +// stream ends once the last byte is checked, before storeArtifact's scan and +// database update. +func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared string, artifact *fetch.Artifact) (artifacts.Artifact, string, error) { + hash := strings.ToLower(upstreamHash) + checks, err := newIntegrityChecks(hash, declared) + if err != nil { + _ = artifact.Body.Close() + return artifacts.Artifact{}, "", err + } + + clientR, clientW := io.Pipe() + var client io.ReadCloser = clientR + if hash != "" || declared != "" { + if client, err = checks.wrapFailOnMismatch(clientR, func(reason string) { + p.Logger.Warn("artifact sent on a cache miss failed its integrity check; aborting the response", + "purl", versionPURL, "filename", filename, "reason", reason) + }); err != nil { + _ = artifact.Body.Close() + return artifacts.Artifact{}, "", err + } + } + stream := &CacheResult{ + Reader: client, + Artifact: artifacts.Artifact{ + PURL: versionPURL, + Filename: filename, + MediaType: artifact.ContentType, + // Unknown until stored, so the response goes out chunked and + // can still be aborted. + Size: -1, + }, + } + if hash != "" { + stream.Artifact.Digest = digest.Digest("sha256:" + hash) + } + if !h.offer(stream) { + _ = clientR.Close() + return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared, artifact) + } + + storeR, storeW := io.Pipe() + type stored struct { + artifact artifacts.Artifact + path string + err error + } + storeDone := make(chan stored, 1) + go func() { + // Not cancelled with ctx: the client's request ends once it has the + // last byte, while storage may still be finishing the upload and the + // database write. Reading the upstream still stops with ctx, which + // ends the copy to storage too. + a, path, err := p.storeArtifact(context.WithoutCancel(ctx), ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared, + &fetch.Artifact{Body: storeR, Size: artifact.Size, ContentType: artifact.ContentType, ETag: artifact.ETag}) + // storeArtifact closes storeR; this covers a return before it read + // anything, so the copy below never blocks on it. + _ = storeR.Close() + storeDone <- stored{a, path, err} + }() + + readErr, clientOK := copyToBoth(artifact.Body, storeW, clientW) + _ = artifact.Body.Close() + _ = storeW.CloseWithError(readErr) + _ = clientW.CloseWithError(readErr) + + out := <-storeDone + if out.err != nil && readErr == nil && clientOK { + p.Logger.Error("artifact sent to the client but not cached", + "purl", versionPURL, "filename", filename, "url", upstreamURL, "error", out.err) + metrics.RecordArtifactServedUncached(ecosystem, metrics.FailureDetail{ + PURL: versionPURL, Filename: filename, URL: upstreamURL, Err: out.err}) + } + return out.artifact, out.path, out.err +} + +// copyToBoth copies src to store and client until src ends, dropping either +// destination once a write to it fails. It returns the error that ended the +// copy, nil at the end of src, and whether the client took every byte. +func copyToBoth(src io.Reader, store, client io.Writer) (error, bool) { + buffer := artifactCopyBufferPool.Get().(*[]byte) + defer artifactCopyBufferPool.Put(buffer) + buf := *buffer + storeOK, clientOK := true, true + for { + n, err := src.Read(buf) + if n > 0 { + if storeOK { + if _, werr := store.Write(buf[:n]); werr != nil { + storeOK = false + } + } + if clientOK { + if _, werr := client.Write(buf[:n]); werr != nil { + clientOK = false + } + } + if !storeOK && !clientOK { + return errNoReaderLeft, false + } + } + if errors.Is(err, io.EOF) { + return nil, clientOK + } + if err != nil { + return err, false + } + } +} diff --git a/internal/handler/stream_miss_test.go b/internal/handler/stream_miss_test.go new file mode 100644 index 00000000..e3b449b2 --- /dev/null +++ b/internal/handler/stream_miss_test.go @@ -0,0 +1,444 @@ +package handler + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/config" + "github.com/git-pkgs/proxy/internal/metrics" + "github.com/git-pkgs/proxy/internal/scanner" + "github.com/git-pkgs/registries/fetch" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// lockedBuffer is a log destination safe for the goroutine that stores a +// streamed artifact to write to while a test reads it. +type lockedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// streamMissSetup is a Git LFS upstream whose object store sends the first +// half of an object, then waits for release before sending the rest, and a +// proxy serving it to clients the way the server marks their requests. +type streamMissSetup struct { + proxy *Proxy + store *mockStorage + client *httptest.Server + git *httptest.Server + objects *httptest.Server + release chan struct{} + content string // what the object store sends + oid string // the object ID clients ask for + path string + downloads atomic.Int32 + logs lockedBuffer +} + +// streamMissContent spans many copy buffers: a stream checked against its +// digest holds back its latest chunk until the next one arrives. +var streamMissContent = strings.Repeat("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ\n", 400<<10/63) + +// streamMissPrefix is how much a client must get while the upstream still +// holds back the second half. +const streamMissPrefix = 64 << 10 + +func newStreamMiss(t *testing.T, served string) *streamMissSetup { + t.Helper() + return newStreamMissMarked(t, served, true) +} + +// newStreamMissMarked is newStreamMiss, with requests marked to detach +// their upstream fetch as the server does, or not, as when the server has +// no detached fetch to spare. +func newStreamMissMarked(t *testing.T, served string, detached bool) *streamMissSetup { + t.Helper() + sum := sha256.Sum256([]byte(streamMissContent)) + s := &streamMissSetup{content: served, oid: hex.EncodeToString(sum[:]), release: make(chan struct{})} + s.objects = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.downloads.Add(1) + half := len(s.content) / 2 + _, _ = io.WriteString(w, s.content[:half]) + w.(http.Flusher).Flush() + select { + case <-s.release: + case <-r.Context().Done(): + return + } + _, _ = io.WriteString(w, s.content[half:]) + })) + s.git = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req lfsBatchRequest + _ = json.NewDecoder(r.Body).Decode(&req) + resp := lfsBatchResponse{Transfer: "basic"} + for _, o := range req.Objects { + resp.Objects = append(resp.Objects, lfsObjectResponse{OID: o.OID, Size: o.Size, + Actions: map[string]lfsAction{"download": {Href: s.objects.URL + "/objects/" + o.OID}}}) + } + w.Header().Set("Content-Type", gitLFSMediaType) + _ = json.NewEncoder(w).Encode(resp) + })) + + proxy, _, store, _ := setupTestProxy(t) + store.seekable = true + proxy.Logger = slog.New(slog.NewTextHandler(&s.logs, nil)) + proxy.HTTPClient = s.git.Client() + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(s.objects.Client()), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + s.proxy, s.store = proxy, store + h := NewGitHandler(proxy, nil, config.UpstreamConfig{Git: map[string]config.GitUpstream{ + "up": {URL: s.git.URL}, + }}).WithBaseURL("http://proxy.test/") + routes := h.Routes() + // As the server marks client requests. + s.client = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ctx := StreamCacheMiss(r.Context()) + if detached { + ctx = DetachUpstreamFetch(ctx) + } + routes.ServeHTTP(w, r.WithContext(ctx)) + })) + s.path = fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", s.oid, len(streamMissContent)) + t.Cleanup(func() { + s.releaseUpstream() + s.client.Close() + s.git.Close() + s.objects.Close() + _ = fetcher.Close() + }) + return s +} + +func (s *streamMissSetup) releaseUpstream() { + select { + case <-s.release: + default: + close(s.release) + } +} + +// get requests the object. The response headers must arrive in time even +// while the upstream holds back the second half. +func (s *streamMissSetup) get(t *testing.T) *http.Response { + t.Helper() + type result struct { + resp *http.Response + err error + } + got := make(chan result, 1) + go func() { + resp, err := s.client.Client().Get(s.client.URL + s.path) + got <- result{resp, err} + }() + select { + case r := <-got: + if r.err != nil { + t.Fatal(r.err) + } + return r.resp + case <-time.After(5 * time.Second): + // Once the upstream is released the response arrives; close it. + t.Cleanup(func() { + s.releaseUpstream() + if r := <-got; r.resp != nil { + _ = r.resp.Body.Close() + } + }) + t.Fatal("no response while the upstream was still sending") + return nil + } +} + +// cached waits for the stored copy, which is written after the client's +// stream ends. +func (s *streamMissSetup) cached(t *testing.T) bool { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + res, err := s.proxy.GetCachedArtifact(t.Context(), gitEcosystem, "up/acme/widget", s.oid, s.oid) + if err == nil && res != nil { + _ = res.Reader.Close() + return true + } + if time.Now().After(deadline) { + return false + } + time.Sleep(10 * time.Millisecond) + } +} + +// readFirst reads the first streamMissPrefix bytes of body, failing if they +// don't arrive in time. +func readFirst(t *testing.T, body io.Reader) string { + t.Helper() + got := make(chan string, 1) + go func() { + buf := make([]byte, streamMissPrefix) + _, _ = io.ReadFull(body, buf) + got <- string(buf) + }() + select { + case s := <-got: + return s + case <-time.After(5 * time.Second): + t.Fatal("no bytes reached the client while the upstream was still sending") + return "" + } +} + +func TestCacheMissStreamsBeforeUpstreamFinishes(t *testing.T) { + s := newStreamMiss(t, streamMissContent) + resp := s.get(t) + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK || resp.ContentLength != -1 { + t.Fatalf("status %d, Content-Length %d; want 200, chunked", resp.StatusCode, resp.ContentLength) + } + if first := readFirst(t, resp.Body); first != streamMissContent[:streamMissPrefix] { + t.Fatal("first bytes differ from the artifact's") + } + + s.releaseUpstream() + rest, err := io.ReadAll(resp.Body) + if err != nil || string(rest) != streamMissContent[streamMissPrefix:] { + t.Fatalf("rest: %d bytes, %v", len(rest), err) + } + if !s.cached(t) { + t.Fatal("streamed artifact was not cached") + } + + // Served from the cache now, with its size. + s.objects.Close() + again := s.get(t) + body, _ := io.ReadAll(again.Body) + _ = again.Body.Close() + if again.StatusCode != http.StatusOK || string(body) != streamMissContent || again.ContentLength != int64(len(streamMissContent)) { + t.Errorf("cached GET: status %d, length %d, %d bytes", again.StatusCode, again.ContentLength, len(body)) + } + if s.downloads.Load() != 1 { + t.Errorf("downloads = %d, want 1", s.downloads.Load()) + } +} + +func TestCacheMissStreamAbortedOnMismatch(t *testing.T) { + tampered := "X" + streamMissContent[1:] + s := newStreamMiss(t, tampered) + s.releaseUpstream() + resp := s.get(t) + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err == nil { + t.Fatalf("read %d bytes without error; want the response aborted", len(body)) + } + if string(body) == tampered { + t.Error("client got every byte of an artifact that doesn't match its object ID") + } + // Nothing to wait for: storeArtifact has refused it once the stream ends. + time.Sleep(100 * time.Millisecond) + if res, _ := s.proxy.GetCachedArtifact(t.Context(), gitEcosystem, "up/acme/widget", s.oid, s.oid); res != nil { + _ = res.Reader.Close() + t.Error("mismatching artifact was cached") + } +} + +func TestCacheMissStreamFinishesWhenStorageFails(t *testing.T) { + s := newStreamMiss(t, streamMissContent) + s.store.mu.Lock() + s.store.storeErr = errors.New("no space left on device") + s.store.mu.Unlock() + before := testutil.ToFloat64(metrics.ArtifactsServedUncached.WithLabelValues(gitEcosystem)) + s.releaseUpstream() + + resp := s.get(t) + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil || resp.StatusCode != http.StatusOK || string(body) != streamMissContent { + t.Fatalf("status %d, %d bytes, %v; want the whole artifact", resp.StatusCode, len(body), err) + } + deadline := time.Now().Add(5 * time.Second) + for testutil.ToFloat64(metrics.ArtifactsServedUncached.WithLabelValues(gitEcosystem)) == before && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := testutil.ToFloat64(metrics.ArtifactsServedUncached.WithLabelValues(gitEcosystem)) - before; got != 1 { + t.Errorf("served-uncached count went up by %v, want 1", got) + } + if !strings.Contains(s.logs.String(), "level=ERROR") || !strings.Contains(s.logs.String(), "not cached") { + t.Errorf("storage failure not logged at error level:\n%s", s.logs.String()) + } +} + +func TestCacheMissStoredAfterClientLeaves(t *testing.T) { + s := newStreamMiss(t, streamMissContent) + resp := s.get(t) + readFirst(t, resp.Body) + _ = resp.Body.Close() + s.client.CloseClientConnections() + + s.releaseUpstream() + if !s.cached(t) { + t.Fatal("artifact not cached after the client that started the fetch left") + } +} + +func TestCacheMissWaiterGetsStoredArtifact(t *testing.T) { + s := newStreamMiss(t, streamMissContent) + first := s.get(t) + defer func() { _ = first.Body.Close() }() + readFirst(t, first.Body) + + second := make(chan string, 1) + go func() { + resp, err := s.client.Client().Get(s.client.URL + s.path) + if err != nil { + second <- err.Error() + return + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + second <- fmt.Sprintf("%d %d %s", resp.StatusCode, resp.ContentLength, body) + }() + select { + case got := <-second: + t.Fatalf("second request answered before the artifact was stored: %s", got) + case <-time.After(200 * time.Millisecond): + } + + s.releaseUpstream() + if _, err := io.ReadAll(first.Body); err != nil { + t.Fatal(err) + } + want := fmt.Sprintf("200 %d %s", len(streamMissContent), streamMissContent) + if got := <-second; got != want { + t.Errorf("second request = %.60q..., want the stored artifact with its length", got) + } + if s.downloads.Load() != 1 { + t.Errorf("downloads = %d, want 1", s.downloads.Load()) + } +} + +func newStreamMissScanGroup(t *testing.T, mode string) *scanner.Group { + t.Helper() + g, err := scanner.NewGroup(config.ScanningConfig{ + Enabled: true, + Timeout: "15s", + SigningKey: "test-signing-key", + Scanners: []config.ScannerConfig{ + {Name: "test-scanner", URL: newTestScanServer(t, true, "").URL, Mode: mode}, + }, + }, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatal(err) + } + return g +} + +// A block-mode scanner must see the whole artifact before any of it is +// served, so a miss isn't streamed. +func TestCacheMissNotStreamedWithBlockingScanner(t *testing.T) { + s := newStreamMiss(t, streamMissContent) + s.proxy.ScanSigningKey = []byte("test-signing-key") + s.proxy.Scanners = newStreamMissScanGroup(t, "block") + + got := make(chan *http.Response, 1) + go func() { + resp, err := s.client.Client().Get(s.client.URL + s.path) + if err != nil { + t.Error(err) + } + got <- resp + }() + select { + case resp := <-got: + _ = resp.Body.Close() + t.Fatal("response started before the upstream finished, with a block-mode scanner") + case <-time.After(300 * time.Millisecond): + } + s.releaseUpstream() + resp := <-got + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK || string(body) != streamMissContent || resp.ContentLength != int64(len(streamMissContent)) { + t.Errorf("status %d, length %d, %d bytes; want the scanned, stored artifact", resp.StatusCode, resp.ContentLength, len(body)) + } +} + +// A monitor-mode scanner doesn't gate the artifact, so the miss is streamed +// and the scan runs after the client's stream ends. +func TestCacheMissStreamedWithMonitoringScanner(t *testing.T) { + s := newStreamMiss(t, streamMissContent) + s.proxy.ScanSigningKey = []byte("test-signing-key") + s.proxy.Scanners = newStreamMissScanGroup(t, "monitor") + + resp := s.get(t) + defer func() { _ = resp.Body.Close() }() + readFirst(t, resp.Body) + s.releaseUpstream() + if _, err := io.ReadAll(resp.Body); err != nil { + t.Fatal(err) + } + if !s.cached(t) { + t.Error("artifact not cached") + } +} + +// slowStorage stores like mockStorage, but takes a while to finish once it +// has read the last byte, as an upload to S3 does. +type slowStorage struct { + *mockStorage + delay time.Duration +} + +func (s *slowStorage) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) { + data, err := io.ReadAll(r) + if err != nil { + return 0, "", err + } + select { + case <-time.After(s.delay): + case <-ctx.Done(): + return 0, "", ctx.Err() + } + return s.mockStorage.Store(ctx, path, bytes.NewReader(data)) +} + +// Without a detached fetch, storing still finishes after the client's +// request has ended with the last byte. +func TestCacheMissStoredAfterResponseWithoutDetach(t *testing.T) { + s := newStreamMissMarked(t, streamMissContent, false) + s.proxy.Storage = &slowStorage{mockStorage: s.store, delay: 300 * time.Millisecond} + s.releaseUpstream() + resp := s.get(t) + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil || string(body) != streamMissContent { + t.Fatalf("%d bytes, %v", len(body), err) + } + if !s.cached(t) { + t.Fatal("artifact not cached once the response ended") + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 708b5b30..2acde79d 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -163,6 +163,14 @@ var ( }, ) + ArtifactsServedUncached = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "proxy_artifacts_served_uncached_total", + Help: "Artifacts sent to the client on a cache miss that could not be cached, such as when storage is full", + }, + []string{"ecosystem"}, + ) + IntegrityFailures = prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "proxy_integrity_failures_total", @@ -319,6 +327,7 @@ func init() { UpstreamRateLimitHolds, StorageOperationDuration, StorageErrors, + ArtifactsServedUncached, ActiveRequests, IntegrityFailures, HealthProbeFailures, @@ -465,6 +474,14 @@ func RecordStorageError(operation string, detail FailureDetail) { recordFailure("proxy_storage_errors_total", map[string]string{"operation": operation}, detail) } +// RecordArtifactServedUncached counts an artifact a client got on a cache +// miss although storing it failed, and keeps the detail for /ui/analytics. +func RecordArtifactServedUncached(ecosystem string, detail FailureDetail) { + ecosystem = purl.NormalizeEcosystem(ecosystem) + ArtifactsServedUncached.WithLabelValues(ecosystem).Inc() + recordFailure("proxy_artifacts_served_uncached_total", map[string]string{"ecosystem": ecosystem}, detail) +} + // RecordScanResult tracks a completed pre-cache scan call. func RecordScanResult(ecosystem, scannerName string, allowed bool, duration time.Duration) { ecosystem = purl.NormalizeEcosystem(ecosystem) diff --git a/internal/scanner/group.go b/internal/scanner/group.go index 227a0370..cf5930fb 100644 --- a/internal/scanner/group.go +++ b/internal/scanner/group.go @@ -91,6 +91,20 @@ func (g *Group) Covers(ecosystem string) bool { return g.Enabled() && len(g.applicable(ecosystem)) > 0 } +// Blocks reports whether a block-mode scanner applies to ecosystem, so an +// artifact must be scanned whole before any of it is served. +func (g *Group) Blocks(ecosystem string) bool { + if !g.Enabled() { + return false + } + for _, e := range g.applicable(ecosystem) { + if e.mode == modeBlock { + return true + } + } + return false +} + // Timeout returns the per-scan-call timeout used to bound the signed fetch // URL's validity. func (g *Group) Timeout() time.Duration { diff --git a/internal/scanner/group_test.go b/internal/scanner/group_test.go index 48cbebca..cbbaa01c 100644 --- a/internal/scanner/group_test.go +++ b/internal/scanner/group_test.go @@ -270,3 +270,27 @@ func TestGroup_Scan_ErrorFailOpen(t *testing.T) { t.Error("Allowed = false, want true — FailOpen must treat scanner errors as allowed") } } + +func TestGroupBlocks(t *testing.T) { + g, err := NewGroup(config.ScanningConfig{ + Enabled: true, + SigningKey: "key", + Scanners: []config.ScannerConfig{ + {Name: "watch", URL: "http://watch.invalid", Mode: "monitor"}, + {Name: "gate", URL: "http://gate.invalid", Mode: "block", Ecosystems: []string{"npm"}}, + }, + }, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatal(err) + } + if !g.Blocks("npm") { + t.Error("Blocks(npm) = false with a block-mode scanner for npm") + } + if g.Blocks("pypi") { + t.Error("Blocks(pypi) = true with only a monitor-mode scanner for it") + } + var none *Group + if none.Blocks("npm") { + t.Error("nil group blocks") + } +} diff --git a/internal/server/middleware.go b/internal/server/middleware.go index f0dd831c..7d2bd983 100644 --- a/internal/server/middleware.go +++ b/internal/server/middleware.go @@ -44,8 +44,11 @@ func GetRequestID(ctx context.Context) string { func markRequestContext(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ctx := handler.DetachUpstreamFetch(r.Context()) - if r.Method == http.MethodHead { + switch { + case r.Method == http.MethodHead: ctx = handler.ServingHead(ctx) + case r.Method == http.MethodGet && r.Header.Get("Range") == "": + ctx = handler.StreamCacheMiss(ctx) } next.ServeHTTP(w, r.WithContext(ctx)) }) From 9a5e5a2c9a85db5cb9de165726af4384c678c87b Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Mon, 26 Oct 2026 16:44:11 +0100 Subject: [PATCH 2/3] Keep a slow client from holding back a streamed miss The upstream read fed the client through a pipe, so it ran at the client's speed. The client is now fed from a bounded buffer and, once it falls behind, continues from the stored copy where it left off. A caller handed the stream takes no hold on an unrecorded object. --- docs/configuration.md | 2 +- internal/handler/stream_miss.go | 240 +++++++++++++++++++++++---- internal/handler/stream_miss_test.go | 123 ++++++++++++++ internal/handler/unrecorded_test.go | 56 +++++++ 4 files changed, 384 insertions(+), 37 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 936be4d9..e92a0acc 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -76,7 +76,7 @@ storage: On a cache miss the proxy sends the artifact to the client while it stores it, so the client gets its first bytes as soon as the upstream sends them, rather than after the whole artifact is stored. Clients that give up after a while without data, such as git-lfs (30 seconds) or pacman (10 seconds), no longer time out on large artifacts. The response goes out chunked, without `Content-Length`, as with `cache_artifacts: false`; once stored, the artifact is served with its size. -The bytes are checked against the digest the upstream declares, where it declares one, as they are sent. On a mismatch the connection is aborted before the response completes, so the client never receives a tampered artifact as a good one, and nothing is cached. If the client goes away, the artifact is still stored. If storing it fails, for example because the disk is full, the client still gets the whole artifact; the failure is logged at error level and counted in `proxy_artifacts_served_uncached_total`, so a cache that has stopped filling doesn't go unnoticed while every request succeeds. +The bytes are checked against the digest the upstream declares, where it declares one, as they are sent. On a mismatch the connection is aborted before the response completes, so the client never receives a tampered artifact as a good one, and nothing is cached. The upstream is read at its own speed whatever the client's: a client more than 8 MB behind continues from the stored copy once it is stored, so a slow client doesn't hold back the download, other requests waiting for it, or `artifact_timeout`. If the client goes away, the artifact is still stored. If storing it fails, for example because the disk is full, the client still gets the whole artifact; the failure is logged at error level and counted in `proxy_artifacts_served_uncached_total`, so a cache that has stopped filling doesn't go unnoticed while every request succeeds. Requests for the same artifact that arrive while it is being stored wait for the stored copy, as do `HEAD` requests and requests with a `Range` header, which are answered from it with its size. With a `block`-mode scanner configured for the ecosystem, a miss is stored and scanned before anything is sent, as before (see [Artifact Scanning](#artifact-scanning)). diff --git a/internal/handler/stream_miss.go b/internal/handler/stream_miss.go index 9ddcc206..bc401b29 100644 --- a/internal/handler/stream_miss.go +++ b/internal/handler/stream_miss.go @@ -3,6 +3,7 @@ package handler import ( "context" "errors" + "fmt" "io" "strings" "sync" @@ -13,12 +14,12 @@ import ( "github.com/opencontainers/go-digest" ) -// On a cache miss the proxy used to store the whole artifact before sending -// any of it, so a client with an idle timeout (git-lfs gives up after 30 -// seconds without data, pacman after 10) could give up on a large one. A -// marked request instead gets the bytes as they arrive from upstream while -// they are stored: the caller that starts the shared fetch is handed a -// stream, and callers joining it still wait for the stored artifact. +// On a cache miss a marked request is sent the bytes as they arrive from +// upstream while they are stored, rather than after the whole artifact is +// stored. Clients with idle timeouts need this for large artifacts: git-lfs +// gives up after 30 seconds without data, pacman after 10. The caller that +// starts the shared fetch is handed the stream; callers joining it still wait +// for the stored artifact. type streamCacheMissKey struct{} @@ -138,15 +139,9 @@ func (p *Proxy) storeFetched(ctx context.Context, ecosystem, name, version, file var errNoReaderLeft = errors.New("neither storage nor the client is reading the artifact") // storeWhileServing copies the upstream body to storage and to the client at -// once. The client's copy is checked against upstreamHash and declared as it -// is read, and a mismatch fails it before its last bytes, so the response is -// aborted and the client discards it; storeArtifact checks the stored copy -// the same way and doesn't cache it. A client that goes away leaves the copy -// to storage running. A storage error leaves the client's copy running, so -// the client still gets the artifact; that is logged and counted, since every -// request would otherwise succeed while nothing is cached. The client's -// stream ends once the last byte is checked, before storeArtifact's scan and -// database update. +// once, at the upstream's speed; see missFeed for a client that can't keep up. +// The client's copy is checked before its last bytes, so a mismatch aborts +// the response. A storage failure leaves the client's copy running. func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared string, artifact *fetch.Artifact) (artifacts.Artifact, string, error) { hash := strings.ToLower(upstreamHash) checks, err := newIntegrityChecks(hash, declared) @@ -155,10 +150,10 @@ func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem return artifacts.Artifact{}, "", err } - clientR, clientW := io.Pipe() - var client io.ReadCloser = clientR + feed := newMissFeed(clientFeedLimit) + var client io.ReadCloser = feed if hash != "" || declared != "" { - if client, err = checks.wrapFailOnMismatch(clientR, func(reason string) { + if client, err = checks.wrapFailOnMismatch(feed, func(reason string) { p.Logger.Warn("artifact sent on a cache miss failed its integrity check; aborting the response", "purl", versionPURL, "filename", filename, "reason", reason) }); err != nil { @@ -181,7 +176,7 @@ func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem stream.Artifact.Digest = digest.Digest("sha256:" + hash) } if !h.offer(stream) { - _ = clientR.Close() + _ = feed.Close() return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared, artifact) } @@ -205,13 +200,17 @@ func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem storeDone <- stored{a, path, err} }() - readErr, clientOK := copyToBoth(artifact.Body, storeW, clientW) + readErr := copyToBoth(artifact.Body, storeW, feed) _ = artifact.Body.Close() _ = storeW.CloseWithError(readErr) - _ = clientW.CloseWithError(readErr) + feed.end(readErr) out := <-storeDone - if out.err != nil && readErr == nil && clientOK { + if feed.lagging() { + // Opened before returning, so an unrecorded object is still there. + feed.resolve(p.openStoredForFeed(ctx, out.artifact, out.path, out.err)) + } + if out.err != nil && readErr == nil && feed.fedFully() { p.Logger.Error("artifact sent to the client but not cached", "purl", versionPURL, "filename", filename, "url", upstreamURL, "error", out.err) metrics.RecordArtifactServedUncached(ecosystem, metrics.FailureDetail{ @@ -220,14 +219,27 @@ func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem return out.artifact, out.path, out.err } -// copyToBoth copies src to store and client until src ends, dropping either -// destination once a write to it fails. It returns the error that ended the -// copy, nil at the end of src, and whether the client took every byte. -func copyToBoth(src io.Reader, store, client io.Writer) (error, bool) { +// openStoredForFeed opens the stored copy for a client that fell behind. An +// unrecorded object is opened too: it is deleted only after this returns. +func (p *Proxy) openStoredForFeed(ctx context.Context, a artifacts.Artifact, path string, storeErr error) (io.ReadCloser, error) { + if storeErr != nil && !errors.Is(storeErr, errUnrecorded) { + return nil, storeErr + } + res, err := p.openStoredArtifact(ctx, a, path) + if err != nil { + return nil, err + } + return res.Reader, nil +} + +// copyToBoth copies src to store and to the client's feed until src ends, +// dropping the store once a write to it fails. It returns the error that +// ended the copy, nil at the end of src. Feeding the client never blocks. +func copyToBoth(src io.Reader, store io.Writer, client *missFeed) error { buffer := artifactCopyBufferPool.Get().(*[]byte) defer artifactCopyBufferPool.Put(buffer) buf := *buffer - storeOK, clientOK := true, true + storeOK := true for { n, err := src.Read(buf) if n > 0 { @@ -236,20 +248,176 @@ func copyToBoth(src io.Reader, store, client io.Writer) (error, bool) { storeOK = false } } - if clientOK { - if _, werr := client.Write(buf[:n]); werr != nil { - clientOK = false - } - } - if !storeOK && !clientOK { - return errNoReaderLeft, false + // A client that has gone, or that waits for the stored copy, + // has no use for the rest once storing has failed. + if !client.push(buf[:n]) && !storeOK { + return errNoReaderLeft } } if errors.Is(err, io.EOF) { - return nil, clientOK + return nil } if err != nil { - return err, false + return err } } } + +// clientFeedLimit bounds the bytes held for a client that reads a streamed +// miss slower than they arrive. A variable so tests can lower it. +var clientFeedLimit = 8 << 20 + +// missFeed is the client's side of a streamed miss. The upstream read pushes +// into it without blocking, so a slow client can't hold back the copy to +// storage, the fetch lease or the callers waiting on the fetch. Once the +// client is more than the limit behind, it stops being fed: it reads what +// is held, then continues from the stored copy where it left off. +type missFeed struct { + mu sync.Mutex + cond *sync.Cond + chunks [][]byte + held int + limit int + // fed counts the bytes pushed before the client fell behind, where it + // continues in the stored copy. + fed int64 + ended bool + endErr error + behind bool + closed bool + early bool // closed before reading everything it was fed + + stored io.ReadCloser + storedErr error + resolved bool + skipped bool +} + +func newMissFeed(limit int) *missFeed { + m := &missFeed{limit: limit} + m.cond = sync.NewCond(&m.mu) + return m +} + +// push queues a copy of b and reports whether the client is still fed. +func (m *missFeed) push(b []byte) bool { + m.mu.Lock() + defer m.mu.Unlock() + if m.closed || m.behind { + return false + } + if m.held+len(b) > m.limit { + m.behind = true + return false + } + m.chunks = append(m.chunks, append([]byte(nil), b...)) + m.held += len(b) + m.fed += int64(len(b)) + m.cond.Broadcast() + return true +} + +// end records how the upstream read ended. +func (m *missFeed) end(err error) { + m.mu.Lock() + defer m.mu.Unlock() + m.ended, m.endErr = true, err + m.cond.Broadcast() +} + +func (m *missFeed) lagging() bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.behind && !m.closed +} + +// fedFully reports whether the client was sent every byte from upstream. +func (m *missFeed) fedFully() bool { + m.mu.Lock() + defer m.mu.Unlock() + return !m.behind && !m.early +} + +// resolve gives a client that fell behind the stored copy, or the error +// that left none. +func (m *missFeed) resolve(r io.ReadCloser, err error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.closed { + if r != nil { + _ = r.Close() + } + return + } + m.stored, m.storedErr, m.resolved = r, err, true + m.cond.Broadcast() +} + +func (m *missFeed) Read(p []byte) (int, error) { + m.mu.Lock() + for { + switch { + case m.closed: + m.mu.Unlock() + return 0, io.ErrClosedPipe + case len(m.chunks) > 0: + n := copy(p, m.chunks[0]) + if m.chunks[0] = m.chunks[0][n:]; len(m.chunks[0]) == 0 { + m.chunks[0] = nil + m.chunks = m.chunks[1:] + } + m.held -= n + m.mu.Unlock() + return n, nil + case m.behind && m.resolved: + r, err, skipped := m.stored, m.storedErr, m.skipped + m.skipped = true + m.mu.Unlock() + if err != nil { + return 0, err + } + if !skipped { + if err := skipTo(r, m.fed); err != nil { + return 0, err + } + } + return r.Read(p) + case !m.behind && m.ended: + err := m.endErr + m.mu.Unlock() + if err == nil { + err = io.EOF + } + return 0, err + } + m.cond.Wait() + } +} + +func (m *missFeed) Close() error { + m.mu.Lock() + defer m.mu.Unlock() + if m.closed { + return nil + } + m.closed = true + m.early = !m.ended || len(m.chunks) > 0 + m.chunks, m.held = nil, 0 + if m.stored != nil { + _ = m.stored.Close() + } + m.cond.Broadcast() + return nil +} + +// skipTo moves r past the first n bytes. +func skipTo(r io.Reader, n int64) error { + if s, ok := r.(io.Seeker); ok { + _, err := s.Seek(n, io.SeekStart) + return err + } + if _, err := io.CopyN(io.Discard, r, n); err != nil { + return fmt.Errorf("skipping to byte %d of the stored artifact: %w", n, err) + } + return nil +} diff --git a/internal/handler/stream_miss_test.go b/internal/handler/stream_miss_test.go index e3b449b2..172be1bc 100644 --- a/internal/handler/stream_miss_test.go +++ b/internal/handler/stream_miss_test.go @@ -442,3 +442,126 @@ func TestCacheMissStoredAfterResponseWithoutDetach(t *testing.T) { t.Fatal("artifact not cached once the response ended") } } + +// lowerFeedLimit makes a client fall behind after n bytes. +func lowerFeedLimit(t *testing.T, n int) { + t.Helper() + old := clientFeedLimit + clientFeedLimit = n + t.Cleanup(func() { clientFeedLimit = old }) +} + +// TestCacheMissSlowClientDoesNotHoldBackFetch has the caller that starts a +// streamed miss read nothing until a second caller, waiting on the same +// fetch, has the stored artifact. The slow client then gets every byte, the +// ones past the limit from the stored copy. +func TestCacheMissSlowClientDoesNotHoldBackFetch(t *testing.T) { + for _, seekable := range []bool{false, true} { + t.Run(fmt.Sprintf("seekable=%v", seekable), func(t *testing.T) { testSlowClient(t, seekable) }) + } +} + +func testSlowClient(t *testing.T, seekable bool) { + t.Helper() + lowerFeedLimit(t, 64<<10) + proxy, _, store, _ := setupTestProxy(t) + store.seekable = seekable + fetcher := newGatedFetcher(streamMissContent) + proxy.Fetcher = fetcher + get := func(ctx context.Context) (*CacheResult, error) { + return proxy.GetOrFetchArtifactFromURL(ctx, "npm", "big", "1.0.0", "big-1.0.0.tgz", "https://example.com/big.tgz") + } + + slow := make(chan *CacheResult, 1) + go func() { + res, err := get(StreamCacheMiss(context.Background())) + if err != nil { + t.Errorf("slow client: %v", err) + } + slow <- res + }() + <-fetcher.entered + waiter := make(chan *CacheResult, 1) + go func() { + res, err := get(context.Background()) + if err != nil { + t.Errorf("waiter: %v", err) + } + waiter <- res + }() + waitForWaiters(t, proxy, 1) + close(fetcher.release) + + lead := <-slow + if lead == nil { + t.FailNow() + } + defer func() { _ = lead.Reader.Close() }() + if lead.Artifact.Size != -1 { + t.Fatalf("slow client got size %d, want a stream", lead.Artifact.Size) + } + select { + case res := <-waiter: + if res == nil { + t.FailNow() + } + body, err := io.ReadAll(res.Reader) + _ = res.Reader.Close() + if err != nil || string(body) != streamMissContent { + t.Fatalf("waiter: %d bytes, %v", len(body), err) + } + case <-time.After(5 * time.Second): + t.Fatal("waiter held back by a client that reads nothing") + } + + body, err := io.ReadAll(lead.Reader) + if err != nil || string(body) != streamMissContent { + t.Fatalf("slow client: %d bytes, %v; want the whole artifact", len(body), err) + } + if fetcher.calls.Load() != 1 { + t.Errorf("upstream fetches = %d, want 1", fetcher.calls.Load()) + } +} + +// TestCacheMissSlowClientFailsWithoutStoredCopy: a client that fell behind +// has nothing to continue from when storing fails, so its response fails +// rather than ending short. +func TestCacheMissSlowClientFailsWithoutStoredCopy(t *testing.T) { + lowerFeedLimit(t, 64<<10) + proxy, _, store, _ := setupTestProxy(t) + store.storeErr = errors.New("no space left on device") + proxy.Fetcher = &countingFetcher{content: streamMissContent} + + res, err := proxy.GetOrFetchArtifactFromURL(StreamCacheMiss(context.Background()), "npm", "big", "1.0.0", "big-1.0.0.tgz", "https://example.com/big.tgz") + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(res.Reader) + _ = res.Reader.Close() + if err == nil { + t.Fatalf("read %d bytes without error; want the response to fail", len(body)) + } +} + +func TestMissFeedCloseWhileWaiting(t *testing.T) { + m := newMissFeed(4) + done := make(chan error, 1) + go func() { + _, err := m.Read(make([]byte, 8)) + done <- err + }() + time.Sleep(10 * time.Millisecond) + _ = m.Close() + select { + case err := <-done: + if !errors.Is(err, io.ErrClosedPipe) { + t.Errorf("Read after Close = %v", err) + } + case <-time.After(time.Second): + t.Fatal("Read still blocked after Close") + } + if m.push([]byte("ab")) { + t.Error("push to a closed feed reported the client fed") + } + m.resolve(io.NopCloser(strings.NewReader("x")), nil) +} diff --git a/internal/handler/unrecorded_test.go b/internal/handler/unrecorded_test.go index 3533a10e..75325f25 100644 --- a/internal/handler/unrecorded_test.go +++ b/internal/handler/unrecorded_test.go @@ -340,3 +340,59 @@ func TestRecordedAt(t *testing.T) { t.Error("failed lookup taken as recorded") } } + +// TestUnrecordedStreamedMiss streams an unrecorded refetch to the caller +// that started it: it holds no reader on the stored object, so the object is +// deleted once the waiters are done, and the waiters still get it. +func TestUnrecordedStreamedMiss(t *testing.T) { + for _, behind := range []bool{false, true} { + t.Run(map[bool]string{false: "fed", true: "behind"}[behind], func(t *testing.T) { + if behind { + old := clientFeedLimit + clientFeedLimit = 4 + t.Cleanup(func() { clientFeedLimit = old }) + } + proxy, store := setupUnrecorded(t) + fetcher := newGatedFetcher(unrecordedContent) + proxy.Fetcher = fetcher + + leader := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(StreamCacheMiss(context.Background()), proxy) + if err != nil { + t.Errorf("leader: %v", err) + } + leader <- res + }() + <-fetcher.entered + waiter := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("waiter: %v", err) + } + waiter <- res + }() + waitForWaiters(t, proxy, 1) + close(fetcher.release) + + lead := <-leader + if lead == nil || lead.Artifact.Size != -1 { + t.Fatalf("leader = %+v, want a stream", lead) + } + w := <-waiter + if w == nil { + t.FailNow() + } + readAndClose(t, w) + deadline := time.Now().Add(5 * time.Second) + for len(storedFiles(store)) != 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after the waiter closed = %v, want none", got) + } + readAndClose(t, lead) + }) + } +} From fb346ae3559130fbdc48f52d2f7c9b3911fbbd1a Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 27 Oct 2026 12:30:12 +0100 Subject: [PATCH 3/3] Hold an unrecorded object for a client reading its stored copy A client that fell behind is switched to the stored copy, which for an unrecorded refetch is deleted once the waiters close. It now takes its own hold, released when it closes. --- internal/handler/handler.go | 1 + internal/handler/stream_miss.go | 41 +++++++++- internal/handler/unrecorded_test.go | 113 ++++++++++++++++------------ 3 files changed, 104 insertions(+), 51 deletions(-) diff --git a/internal/handler/handler.go b/internal/handler/handler.go index a2cd8119..b3ca1c43 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -1052,6 +1052,7 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch f.unrecorded = &unrecordedObject{delete: func() { p.deleteUnrecorded(path) }} f.unrecorded.refs.Add(1) // until the waiters are counted if handedOver { + h.holdUnrecorded(f.unrecorded) return nil, nil } f.unrecorded.refs.Add(1) // this caller diff --git a/internal/handler/stream_miss.go b/internal/handler/stream_miss.go index bc401b29..fb8ae623 100644 --- a/internal/handler/stream_miss.go +++ b/internal/handler/stream_miss.go @@ -46,6 +46,9 @@ type missHandoff struct { stream *CacheResult taken bool abandoned bool + // feed is the stream's source. Set and read only on the fetch's + // goroutine. + feed *missFeed } type missHandoffKey struct{} @@ -67,6 +70,14 @@ func (h *missHandoff) offer(stream *CacheResult) bool { return true } +// holdUnrecorded gives a client that fell behind, and reads the stored copy +// of u, its own hold on u until it closes, so u isn't deleted under it. +func (h *missHandoff) holdUnrecorded(u *unrecordedObject) { + if h.feed != nil { + h.feed.holdUntilClose(u) + } +} + // handedOver reports whether the caller took a stream. func (h *missHandoff) handedOver() bool { h.mu.Lock() @@ -179,6 +190,7 @@ func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem _ = feed.Close() return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared, artifact) } + h.feed = feed storeR, storeW := io.Pipe() type stored struct { @@ -207,7 +219,8 @@ func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem out := <-storeDone if feed.lagging() { - // Opened before returning, so an unrecorded object is still there. + // Opened before returning, so an unrecorded object is still there; + // runSharedFetch then gives the client a hold on it. feed.resolve(p.openStoredForFeed(ctx, out.artifact, out.path, out.err)) } if out.err != nil && readErr == nil && feed.fedFully() { @@ -220,7 +233,8 @@ func (p *Proxy) storeWhileServing(ctx context.Context, h *missHandoff, ecosystem } // openStoredForFeed opens the stored copy for a client that fell behind. An -// unrecorded object is opened too: it is deleted only after this returns. +// unrecorded object is opened too: it is deleted only after this returns, and +// the client holds it from then on (missHandoff.holdUnrecorded). func (p *Proxy) openStoredForFeed(ctx context.Context, a artifacts.Artifact, path string, storeErr error) (io.ReadCloser, error) { if storeErr != nil && !errors.Is(storeErr, errUnrecorded) { return nil, storeErr @@ -291,6 +305,8 @@ type missFeed struct { storedErr error resolved bool skipped bool + // release drops the client's hold on an unrecorded stored copy. + release func() } func newMissFeed(limit int) *missFeed { @@ -353,6 +369,18 @@ func (m *missFeed) resolve(r io.ReadCloser, err error) { m.cond.Broadcast() } +// holdUntilClose takes a hold on u for a client reading its stored copy and +// releases it when the client closes. +func (m *missFeed) holdUntilClose(u *unrecordedObject) { + m.mu.Lock() + defer m.mu.Unlock() + if m.closed || m.stored == nil { + return + } + u.refs.Add(1) + m.release = u.release +} + func (m *missFeed) Read(p []byte) (int, error) { m.mu.Lock() for { @@ -396,8 +424,8 @@ func (m *missFeed) Read(p []byte) (int, error) { func (m *missFeed) Close() error { m.mu.Lock() - defer m.mu.Unlock() if m.closed { + m.mu.Unlock() return nil } m.closed = true @@ -406,7 +434,14 @@ func (m *missFeed) Close() error { if m.stored != nil { _ = m.stored.Close() } + release := m.release + m.release = nil m.cond.Broadcast() + m.mu.Unlock() + if release != nil { + // Outside the lock: the last release deletes the object. + release() + } return nil } diff --git a/internal/handler/unrecorded_test.go b/internal/handler/unrecorded_test.go index 75325f25..d87ddc92 100644 --- a/internal/handler/unrecorded_test.go +++ b/internal/handler/unrecorded_test.go @@ -342,57 +342,74 @@ func TestRecordedAt(t *testing.T) { } // TestUnrecordedStreamedMiss streams an unrecorded refetch to the caller -// that started it: it holds no reader on the stored object, so the object is -// deleted once the waiters are done, and the waiters still get it. +// that started it, and the waiters still get the stored object. A caller fed +// from upstream holds no reader on it, so it is deleted once the waiters are +// done; one that fell behind reads it, so it is deleted once that caller is +// done too. func TestUnrecordedStreamedMiss(t *testing.T) { for _, behind := range []bool{false, true} { t.Run(map[bool]string{false: "fed", true: "behind"}[behind], func(t *testing.T) { - if behind { - old := clientFeedLimit - clientFeedLimit = 4 - t.Cleanup(func() { clientFeedLimit = old }) - } - proxy, store := setupUnrecorded(t) - fetcher := newGatedFetcher(unrecordedContent) - proxy.Fetcher = fetcher - - leader := make(chan *CacheResult, 1) - go func() { - res, err := fetchGone(StreamCacheMiss(context.Background()), proxy) - if err != nil { - t.Errorf("leader: %v", err) - } - leader <- res - }() - <-fetcher.entered - waiter := make(chan *CacheResult, 1) - go func() { - res, err := fetchGone(context.Background(), proxy) - if err != nil { - t.Errorf("waiter: %v", err) - } - waiter <- res - }() - waitForWaiters(t, proxy, 1) - close(fetcher.release) - - lead := <-leader - if lead == nil || lead.Artifact.Size != -1 { - t.Fatalf("leader = %+v, want a stream", lead) - } - w := <-waiter - if w == nil { - t.FailNow() - } - readAndClose(t, w) - deadline := time.Now().Add(5 * time.Second) - for len(storedFiles(store)) != 0 && time.Now().Before(deadline) { - time.Sleep(time.Millisecond) - } - if got := storedFiles(store); len(got) != 0 { - t.Errorf("stored files after the waiter closed = %v, want none", got) - } - readAndClose(t, lead) + testUnrecordedStreamedMiss(t, behind) }) } } + +func testUnrecordedStreamedMiss(t *testing.T, behind bool) { + t.Helper() + if behind { + old := clientFeedLimit + clientFeedLimit = 4 + t.Cleanup(func() { clientFeedLimit = old }) + } + proxy, store := setupUnrecorded(t) + fetcher := newGatedFetcher(unrecordedContent) + proxy.Fetcher = fetcher + + leader := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(StreamCacheMiss(context.Background()), proxy) + if err != nil { + t.Errorf("leader: %v", err) + } + leader <- res + }() + <-fetcher.entered + waiter := make(chan *CacheResult, 1) + go func() { + res, err := fetchGone(context.Background(), proxy) + if err != nil { + t.Errorf("waiter: %v", err) + } + waiter <- res + }() + waitForWaiters(t, proxy, 1) + close(fetcher.release) + + lead := <-leader + if lead == nil || lead.Artifact.Size != -1 { + t.Fatalf("leader = %+v, want a stream", lead) + } + w := <-waiter + if w == nil { + t.FailNow() + } + readAndClose(t, w) + if behind { + // The leader reads the stored copy, so it must outlive the + // waiter. + if len(storedFiles(store)) == 0 { + t.Fatal("stored copy deleted while the leader still reads it") + } + readAndClose(t, lead) + } + deadline := time.Now().Add(5 * time.Second) + for len(storedFiles(store)) != 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := storedFiles(store); len(got) != 0 { + t.Errorf("stored files after every reader closed = %v, want none", got) + } + if !behind { + readAndClose(t, lead) + } +}