diff --git a/docs/architecture.md b/docs/architecture.md index 660d8102..3dd7cde9 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 84b869bd..e92a0acc 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. 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)). + ### 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 @@ -1250,7 +1255,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 b1dc7fc5..b3ca1c43 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -653,7 +653,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 @@ -917,6 +917,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) } @@ -1041,12 +1044,23 @@ func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch if err != nil { return nil, err } + // A caller handed the stream already has the bytes; opening the stored + // copy would cost a read it never uses, a GET on S3 or GCS. + h := handoffFrom(ctx) + handedOver := h != nil && h.handedOver() if unrecorded { f.unrecorded = &unrecordedObject{delete: func() { p.deleteUnrecorded(path) }} - f.unrecorded.refs.Add(1) // this caller 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 return p.serveUnrecorded(ctx, f.unrecorded, stored, path) } + if handedOver { + return nil, nil + } return p.openStoredArtifact(ctx, stored, path) } @@ -2480,7 +2494,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..fb8ae623 --- /dev/null +++ b/internal/handler/stream_miss.go @@ -0,0 +1,458 @@ +package handler + +import ( + "context" + "errors" + "fmt" + "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 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{} + +// 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 + // feed is the stream's source. Set and read only on the fetch's + // goroutine. + feed *missFeed +} + +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 +} + +// 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() + 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, 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) + if err != nil { + _ = artifact.Body.Close() + return artifacts.Artifact{}, "", err + } + + feed := newMissFeed(clientFeedLimit) + var client io.ReadCloser = feed + if hash != "" || declared != "" { + 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 { + _ = 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) { + _ = 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 { + 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 := copyToBoth(artifact.Body, storeW, feed) + _ = artifact.Body.Close() + _ = storeW.CloseWithError(readErr) + feed.end(readErr) + + out := <-storeDone + if feed.lagging() { + // 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() { + 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 +} + +// openStoredForFeed opens the stored copy for a client that fell behind. An +// 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 + } + 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 := true + for { + n, err := src.Read(buf) + if n > 0 { + if storeOK { + if _, werr := store.Write(buf[:n]); werr != nil { + storeOK = 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 + } + if err != nil { + 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 + // release drops the client's hold on an unrecorded stored copy. + release func() +} + +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() +} + +// 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 { + 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() + if m.closed { + m.mu.Unlock() + 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() + } + 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 +} + +// 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 new file mode 100644 index 00000000..172be1bc --- /dev/null +++ b/internal/handler/stream_miss_test.go @@ -0,0 +1,567 @@ +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") + } +} + +// 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..d87ddc92 100644 --- a/internal/handler/unrecorded_test.go +++ b/internal/handler/unrecorded_test.go @@ -340,3 +340,76 @@ func TestRecordedAt(t *testing.T) { t.Error("failed lookup taken as recorded") } } + +// TestUnrecordedStreamedMiss streams an unrecorded refetch to the caller +// 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) { + 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) + } +} 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)) })