Fail metadata reads when upstream cuts a document off

internal/handler/handler.go

@@ -157,7 +157,9 @@ func (p *Proxy) ReadMetadata(r io.Reader) ([]byte, error) {
 // Reading into a buffer of that size avoids io.ReadAll growing its buffer
 // step by step, which peaks at more than twice the document size. size is
 // only a hint: a negative one, or one above the limit, reads as ReadMetadata
-// does, and a body of a different length is still read correctly.
+// does, and a body of a different length is still read correctly. A body
+// read through upstreamBodyReader that upstream cuts off fails, even though
+// it ends like a body shorter than size.
 func (p *Proxy) ReadMetadataSized(r io.Reader, size int64) ([]byte, error) {
 	limit := p.MetadataMaxSize
 	if limit <= 0 {
@@ -170,7 +172,10 @@ func (p *Proxy) ReadMetadataSized(r io.Reader, size int64) ([]byte, error) {
 	// the buffer and a longer one is noticed.
 	buf := make([]byte, size+1)
 	n, err := io.ReadFull(r, buf)
+	var cut *upstreamBodyError
 	switch {
+	case errors.As(err, &cut):
+		return nil, err
 	case errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF):
 		if int64(n) < size/2 {
 			// The hint was far off; don't keep the oversized buffer alive.
@@ -2010,7 +2015,7 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e
 
 	// ContentLength is -1 when the transport decompresses the body, and
 	// otherwise the length of the bytes read here.
-	body, err := p.ReadMetadataSized(resp.Body, resp.ContentLength)
+	body, err := p.ReadMetadataSized(upstreamBodyReader{resp.Body}, resp.ContentLength)
 	if err != nil {
 		return nil, fmt.Errorf("reading response: %w", err)
 	}

internal/handler/metadata_truncated_test.go

@@ -0,0 +1,121 @@
+package handler
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/git-pkgs/cooldown"
+)
+
+// cutMetadataUpstream serves one npm packument and, when cut is set, closes
+// the connection halfway through the document it announced.
+type cutMetadataUpstream struct {
+	*httptest.Server
+	mu   sync.Mutex
+	body string
+	cut  bool
+}
+
+func newCutMetadataUpstream(t *testing.T, body string) *cutMetadataUpstream {
+	t.Helper()
+	u := &cutMetadataUpstream{body: body}
+	u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		u.mu.Lock()
+		body, cut := u.body, u.cut
+		u.mu.Unlock()
+		if r.URL.Path != "/pkg" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.Header().Set("Content-Length", fmt.Sprint(len(body)))
+		if cut {
+			_, _ = io.WriteString(w, body[:len(body)/2])
+			return
+		}
+		_, _ = io.WriteString(w, body)
+	}))
+	t.Cleanup(u.Close)
+	return u
+}
+
+func (u *cutMetadataUpstream) set(body string, cut bool) {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	u.body, u.cut = body, cut
+}
+
+// truncPackument is an npm packument with n versions published long ago.
+func truncPackument(n int) string {
+	var versions, times []string
+	for i := range n {
+		v := fmt.Sprintf("1.0.%d", i)
+		versions = append(versions, fmt.Sprintf(`%q:{"name":"pkg","version":%q,"dist":{"tarball":"https://registry.npmjs.org/pkg/-/pkg-%s.tgz"}}`, v, v, v))
+		times = append(times, fmt.Sprintf(`%q:"2020-01-01T00:00:00Z"`, v))
+	}
+	return `{"name":"pkg","dist-tags":{"latest":"1.0.0"},"versions":{` + strings.Join(versions, ",") +
+		`},"time":{` + strings.Join(times, ",") + `}}`
+}
+
+// TestMetadataTruncatedUpstream checks that a document upstream cuts off
+// before its Content-Length is neither served nor cached: the cached copy is
+// served instead, with and without cooldown.
+func TestMetadataTruncatedUpstream(t *testing.T) {
+	for _, withCooldown := range []bool{false, true} {
+		t.Run(map[bool]string{false: "no cooldown", true: "cooldown"}[withCooldown], func(t *testing.T) {
+			u := newCutMetadataUpstream(t, truncPackument(50))
+			proxy, db, _, _ := setupTestProxy(t)
+			proxy.HTTPClient = u.Client()
+			proxy.CacheMetadata = true
+			if withCooldown {
+				proxy.Cooldown = &cooldown.Config{Default: "3d"}
+			}
+			handler := NewNPMHandler(proxy, "http://proxy.example", u.URL).Routes()
+
+			first := serve(handler, "/pkg", nil)
+			if first.Code != http.StatusOK {
+				t.Fatalf("first request: status %d", first.Code)
+			}
+			before, err := db.GetMetadataCache("npm", "pkg")
+			if err != nil || before == nil {
+				t.Fatalf("no metadata row after the first request: %v", err)
+			}
+
+			u.set(truncPackument(51), true)
+			expireMetadata(t, db, "npm", "pkg")
+			proxy.SetMetadataRewriteCacheSize(0)
+
+			rec := serve(handler, "/pkg", nil)
+			if rec.Code != http.StatusOK || rec.Body.String() != first.Body.String() {
+				t.Errorf("status %d, want 200 with the cached document", rec.Code)
+			}
+			after, err := db.GetMetadataCache("npm", "pkg")
+			if err != nil || after == nil || after.ContentDigest != before.ContentDigest {
+				t.Error("the cut-off document replaced the cached copy")
+			}
+		})
+	}
+}
+
+// TestMetadataTruncatedUpstreamUncached checks that a cut-off document with
+// nothing cached fails the request and is not cached.
+func TestMetadataTruncatedUpstreamUncached(t *testing.T) {
+	u := newCutMetadataUpstream(t, truncPackument(50))
+	u.set(truncPackument(50), true)
+	proxy, db, _, _ := setupTestProxy(t)
+	proxy.HTTPClient = u.Client()
+	proxy.CacheMetadata = true
+	handler := NewNPMHandler(proxy, "http://proxy.example", u.URL).Routes()
+
+	if rec := serve(handler, "/pkg", nil); rec.Code != http.StatusBadGateway {
+		t.Errorf("status %d, want 502", rec.Code)
+	}
+	if entry, _ := db.GetMetadataCache("npm", "pkg"); entry != nil {
+		t.Error("cached the cut-off document")
+	}
+}

internal/handler/read_metadata_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"runtime"
+	"strings"
 	"testing"
 	"testing/iotest"
 )
@@ -131,3 +132,18 @@ func TestReadMetadataSizedAllocatesOnce(t *testing.T) {
 		t.Errorf("allocated %d bytes reading %d", allocated, size)
 	}
 }
+
+// TestReadMetadataSizedTruncatedUpstream checks that an upstream body cut
+// off before its Content-Length fails, while a body shorter than a size
+// hint is still read.
+func TestReadMetadataSizedTruncatedUpstream(t *testing.T) {
+	p := &Proxy{}
+	cut := io.MultiReader(strings.NewReader(`{"name":`), iotest.ErrReader(io.ErrUnexpectedEOF))
+	if _, err := p.ReadMetadataSized(upstreamBodyReader{cut}, 100); !errors.Is(err, io.ErrUnexpectedEOF) {
+		t.Errorf("cut off body: error %v, want io.ErrUnexpectedEOF", err)
+	}
+	got, err := p.ReadMetadataSized(strings.NewReader(`{"name":"x"}`), 100)
+	if err != nil || string(got) != `{"name":"x"}` {
+		t.Errorf("short body = %q, %v", got, err)
+	}
+}

internal/handler/upstream_body.go

@@ -0,0 +1,26 @@
+package handler
+
+import (
+	"errors"
+	"io"
+)
+
+// upstreamBodyReader marks a failure to read a document from upstream, so a
+// body upstream cuts off is told apart from one that ends, and from a
+// failure to store it.
+type upstreamBodyReader struct{ r io.Reader }
+
+func (u upstreamBodyReader) Read(b []byte) (int, error) {
+	n, err := u.r.Read(b)
+	if err != nil && !errors.Is(err, io.EOF) {
+		err = &upstreamBodyError{err}
+	}
+	return n, err
+}
+
+// upstreamBodyError is a failure to read a document from upstream. It is
+// handled as an upstream failure: the stored copy, if any, is served.
+type upstreamBodyError struct{ err error }
+
+func (e *upstreamBodyError) Error() string { return "reading upstream metadata: " + e.err.Error() }
+func (e *upstreamBodyError) Unwrap() error { return e.err }