Move httprs into an internal package

It's not clear whether this source code is maintained anymore. This is
done in preparation to apply https://github.com/jfbus/httprs/pull/6,
which is 2 years old.

go.mod

@@ -15,14 +15,15 @@ require (
 	github.com/gorilla/websocket v1.4.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.0.0
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
-	github.com/jfbus/httprs v0.0.0-20190827093123-b0af8319bb15
 	github.com/johannesboyne/gofakes3 v0.0.0-20200510090907-02d71f533bec
 	github.com/jpillora/backoff v0.0.0-20170918002102-8eab2debe79d
+	github.com/mitchellh/copystructure v1.0.0
 	github.com/prometheus/client_golang v1.0.0
 	github.com/rafaeljusto/redigomock v0.0.0-20190202135759-257e089e14a1
 	github.com/sebest/xff v0.0.0-20160910043805-6c115e0ffa35
 	github.com/shabbyrobe/gocovmerge v0.0.0-20190829150210-3e036491d500 // indirect
 	github.com/sirupsen/logrus v1.3.0
+	github.com/smartystreets/goconvey v1.6.4
 	github.com/stretchr/testify v1.5.1
 	gitlab.com/gitlab-org/gitaly v1.74.0
 	gitlab.com/gitlab-org/labkit v0.0.0-20200520155818-96e583c57891

go.sum

@@ -175,8 +175,6 @@ github.com/iris-contrib/go.uuid v2.0.0+incompatible/go.mod h1:iz2lgM/1UnEf1kP0L/
 github.com/iris-contrib/httpexpect v0.0.0-20180314041918-ebe99fcebbce/go.mod h1:VER17o2JZqquOx41avolD/wMGQSFEFBKWmhag9/RQRY=
 github.com/iris-contrib/i18n v0.0.0-20171121225848-987a633949d0/go.mod h1:pMCz62A0xJL6I+umB2YTlFRwWXaDFA0jy+5HzGiJjqI=
 github.com/iris-contrib/schema v0.0.1/go.mod h1:urYA3uvUNG1TIIjOSCzHr9/LmbQo8LrOcOqfqxa4hXw=
-github.com/jfbus/httprs v0.0.0-20190827093123-b0af8319bb15 h1:HPqgCwRiChGXITjjipDuTJYVPkAUpM4lp0mfo7ONpjo=
-github.com/jfbus/httprs v0.0.0-20190827093123-b0af8319bb15/go.mod h1:hve3GCzwH1IcxgpZ3UN4XKAPSKoIqJhsYF2ZifruodQ=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmespath/go-jmespath v0.3.0 h1:OS12ieG61fsCg5+qLJ+SsW9NicxNkg3b25OyT2yCeUc=
 github.com/jmespath/go-jmespath v0.3.0/go.mod h1:9QtRXoHjLGCJ5IBSaohpXITPlowMeeYCZ7fLUTSywik=

internal/httprs/httprs.go

+/*
+Package httprs provides a ReadSeeker for http.Response.Body.
+
+Usage :
+
+	resp, err := http.Get(url)
+	rs := httprs.NewHttpReadSeeker(resp)
+	defer rs.Close()
+	io.ReadFull(rs, buf) // reads the first bytes from the response body
+	rs.Seek(1024, 0) // moves the position, but does no range request
+	io.ReadFull(rs, buf) // does a range request and reads from the response body
+
+If you want use a specific http.Client for additional range requests :
+	rs := httprs.NewHttpReadSeeker(resp, client)
+*/
+package httprs
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/mitchellh/copystructure"
+)
+
+const shortSeekBytes = 1024
+
+// A HttpReadSeeker reads from a http.Response.Body. It can Seek
+// by doing range requests.
+type HttpReadSeeker struct {
+	c       *http.Client
+	req     *http.Request
+	res     *http.Response
+	ctx     context.Context
+	r       io.ReadCloser
+	pos     int64
+	canSeek bool
+
+	Requests int
+}
+
+var _ io.ReadCloser = (*HttpReadSeeker)(nil)
+var _ io.Seeker = (*HttpReadSeeker)(nil)
+
+var (
+	// ErrNoContentLength is returned by Seek when the initial http response did not include a Content-Length header
+	ErrNoContentLength = errors.New("Content-Length was not set")
+	// ErrRangeRequestsNotSupported is returned by Seek and Read
+	// when the remote server does not allow range requests (Accept-Ranges was not set)
+	ErrRangeRequestsNotSupported = errors.New("Range requests are not supported by the remote server")
+	// ErrInvalidRange is returned by Read when trying to read past the end of the file
+	ErrInvalidRange = errors.New("Invalid range")
+	// ErrContentHasChanged is returned by Read when the content has changed since the first request
+	ErrContentHasChanged = errors.New("Content has changed since first request")
+)
+
+// NewHttpReadSeeker returns a HttpReadSeeker, using the http.Response and, optionaly, the http.Client
+// that needs to be used for future range requests. If no http.Client is given, http.DefaultClient will
+// be used.
+//
+// res.Request will be reused for range requests, headers may be added/removed
+func NewHttpReadSeeker(res *http.Response, client ...*http.Client) *HttpReadSeeker {
+	r := &HttpReadSeeker{
+		req:     res.Request,
+		ctx:     res.Request.Context(),
+		res:     res,
+		r:       res.Body,
+		canSeek: (res.Header.Get("Accept-Ranges") == "bytes"),
+	}
+	if len(client) > 0 {
+		r.c = client[0]
+	} else {
+		r.c = http.DefaultClient
+	}
+	return r
+}
+
+// Clone clones the reader to enable parallel downloads of ranges
+func (r *HttpReadSeeker) Clone() (*HttpReadSeeker, error) {
+	req, err := copystructure.Copy(r.req)
+	if err != nil {
+		return nil, err
+	}
+	return &HttpReadSeeker{
+		req:     req.(*http.Request),
+		res:     r.res,
+		r:       nil,
+		canSeek: r.canSeek,
+		c:       r.c,
+	}, nil
+}
+
+// Read reads from the response body. It does a range request if Seek was called before.
+//
+// May return ErrRangeRequestsNotSupported, ErrInvalidRange or ErrContentHasChanged
+func (r *HttpReadSeeker) Read(p []byte) (n int, err error) {
+	if r.r == nil {
+		err = r.rangeRequest()
+	}
+	if r.r != nil {
+		n, err = r.r.Read(p)
+		r.pos += int64(n)
+	}
+	return
+}
+
+// ReadAt reads from the response body starting at offset off.
+//
+// May return ErrRangeRequestsNotSupported, ErrInvalidRange or ErrContentHasChanged
+func (r *HttpReadSeeker) ReadAt(p []byte, off int64) (n int, err error) {
+	var nn int
+
+	r.Seek(off, 0)
+
+	for n < len(p) && err == nil {
+		nn, err = r.Read(p[n:])
+		n += nn
+	}
+	return
+}
+
+// Close closes the response body
+func (r *HttpReadSeeker) Close() error {
+	if r.r != nil {
+		return r.r.Close()
+	}
+	return nil
+}
+
+// Seek moves the reader position to a new offset.
+//
+// It does not send http requests, allowing for multiple seeks without overhead.
+// The http request will be sent by the next Read call.
+//
+// May return ErrNoContentLength or ErrRangeRequestsNotSupported
+func (r *HttpReadSeeker) Seek(offset int64, whence int) (int64, error) {
+	if !r.canSeek {
+		return 0, ErrRangeRequestsNotSupported
+	}
+	var err error
+	switch whence {
+	case 0:
+	case 1:
+		offset += r.pos
+	case 2:
+		if r.res.ContentLength <= 0 {
+			return 0, ErrNoContentLength
+		}
+		offset = r.res.ContentLength - offset
+	}
+	if r.r != nil {
+		// Try to read, which is cheaper than doing a request
+		if r.pos < offset && offset-r.pos <= shortSeekBytes {
+			_, err := io.CopyN(ioutil.Discard, r, offset-r.pos)
+			if err != nil {
+				return 0, err
+			}
+		}
+
+		if r.pos != offset {
+			err = r.r.Close()
+			r.r = nil
+		}
+	}
+	r.pos = offset
+	return r.pos, err
+}
+
+func cloneHeader(h http.Header) http.Header {
+	h2 := make(http.Header, len(h))
+	for k, vv := range h {
+		vv2 := make([]string, len(vv))
+		copy(vv2, vv)
+		h2[k] = vv2
+	}
+	return h2
+}
+
+func (r *HttpReadSeeker) newRequest() *http.Request {
+	newreq := r.req.WithContext(r.ctx) // includes shallow copies of maps, but okay
+	if r.req.ContentLength == 0 {
+		newreq.Body = nil // Issue 16036: nil Body for http.Transport retries
+	}
+	newreq.Header = cloneHeader(r.req.Header)
+	return newreq
+}
+
+func (r *HttpReadSeeker) rangeRequest() error {
+	r.req = r.newRequest()
+	r.req.Header.Set("Range", fmt.Sprintf("bytes=%d-", r.pos))
+	etag, last := r.res.Header.Get("ETag"), r.res.Header.Get("Last-Modified")
+	switch {
+	case last != "":
+		r.req.Header.Set("If-Range", last)
+	case etag != "":
+		r.req.Header.Set("If-Range", etag)
+	}
+
+	r.Requests++
+
+	res, err := r.c.Do(r.req)
+	if err != nil {
+		return err
+	}
+	switch res.StatusCode {
+	case http.StatusRequestedRangeNotSatisfiable:
+		return ErrInvalidRange
+	case http.StatusOK:
+		// some servers return 200 OK for bytes=0-
+		if r.pos > 0 ||
+			(etag != "" && etag != res.Header.Get("ETag")) {
+			return ErrContentHasChanged
+		}
+		fallthrough
+	case http.StatusPartialContent:
+		r.r = res.Body
+		return nil
+	}
+	return ErrRangeRequestsNotSupported
+}

internal/httprs/httprs_test.go

+package httprs
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+type fakeResponseWriter struct {
+	code int
+	h    http.Header
+	tmp  *os.File
+}
+
+func (f *fakeResponseWriter) Header() http.Header {
+	return f.h
+}
+
+func (f *fakeResponseWriter) Write(b []byte) (int, error) {
+	return f.tmp.Write(b)
+}
+
+func (f *fakeResponseWriter) Close(b []byte) error {
+	return f.tmp.Close()
+}
+
+func (f *fakeResponseWriter) WriteHeader(code int) {
+	f.code = code
+}
+
+func (f *fakeResponseWriter) Response() *http.Response {
+	f.tmp.Seek(0, os.SEEK_SET)
+	return &http.Response{Body: f.tmp, StatusCode: f.code, Header: f.h}
+}
+
+type fakeRoundTripper struct {
+	src                    *os.File
+	downgradeZeroToNoRange bool
+}
+
+func (f *fakeRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
+	fw := &fakeResponseWriter{h: http.Header{}}
+	var err error
+	fw.tmp, err = ioutil.TempFile(os.TempDir(), "httprs")
+	if err != nil {
+		return nil, err
+	}
+	if f.downgradeZeroToNoRange {
+		// There are implementations that downgrades bytes=0- to a normal un-ranged GET
+		if r.Header.Get("Range") == "bytes=0-" {
+			r.Header.Del("Range")
+		}
+	}
+	http.ServeContent(fw, r, "temp.txt", time.Now(), f.src)
+
+	return fw.Response(), nil
+}
+
+const SZ = 4096
+
+type RSFactory func() *HttpReadSeeker
+
+func newRSFactory(brokenServer bool) RSFactory {
+	return func() *HttpReadSeeker {
+		tmp, err := ioutil.TempFile(os.TempDir(), "httprs")
+		if err != nil {
+			return nil
+		}
+		for i := 0; i < SZ; i++ {
+			tmp.WriteString(fmt.Sprintf("%04d", i))
+		}
+
+		req, err := http.NewRequest("GET", "http://www.example.com", nil)
+		if err != nil {
+			return nil
+		}
+		res := &http.Response{
+			Header: http.Header{
+				"Accept-Ranges": []string{"bytes"},
+			},
+			Request:       req,
+			ContentLength: SZ * 4,
+		}
+		return NewHttpReadSeeker(res, &http.Client{Transport: &fakeRoundTripper{src: tmp, downgradeZeroToNoRange: brokenServer}})
+	}
+}
+
+func TestHttpWebServer(t *testing.T) {
+	Convey("Scenario: testing WebServer", t, func() {
+		dir, err := ioutil.TempDir("", "webserver")
+		So(err, ShouldBeNil)
+		defer os.RemoveAll(dir)
+
+		err = ioutil.WriteFile(filepath.Join(dir, "file"), make([]byte, 10000), 0755)
+		So(err, ShouldBeNil)
+
+		server := httptest.NewServer(http.FileServer(http.Dir(dir)))
+		So(server, ShouldNotBeNil)
+
+		Convey("When requesting /file", func() {
+			res, err := http.Get(server.URL + "/file")
+			So(err, ShouldBeNil)
+
+			stream := NewHttpReadSeeker(res)
+			So(stream, ShouldNotBeNil)
+
+			Convey("Can read 100 bytes from start of file", func() {
+				n, err := stream.Read(make([]byte, 100))
+				So(err, ShouldBeNil)
+				So(n, ShouldEqual, 100)
+
+				Convey("When seeking 4KiB forward", func() {
+					pos, err := stream.Seek(4096, io.SeekCurrent)
+					So(err, ShouldBeNil)
+					So(pos, ShouldEqual, 4096+100)
+
+					Convey("Can read 100 bytes", func() {
+						n, err := stream.Read(make([]byte, 100))
+						So(err, ShouldBeNil)
+						So(n, ShouldEqual, 100)
+					})
+				})
+			})
+		})
+	})
+}
+
+func TestHttpReaderSeeker(t *testing.T) {
+	tests := []struct {
+		name  string
+		newRS func() *HttpReadSeeker
+	}{
+		{name: "compliant", newRS: newRSFactory(false)},
+		{name: "broken", newRS: newRSFactory(true)},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			testHttpReaderSeeker(t, test.newRS)
+		})
+	}
+}
+
+func testHttpReaderSeeker(t *testing.T, newRS RSFactory) {
+	Convey("Scenario: testing HttpReaderSeeker", t, func() {
+
+		Convey("Read should start at the beginning", func() {
+			r := newRS()
+			So(r, ShouldNotBeNil)
+			defer r.Close()
+			buf := make([]byte, 4)
+			n, err := io.ReadFull(r, buf)
+			So(n, ShouldEqual, 4)
+			So(err, ShouldBeNil)
+			So(string(buf), ShouldEqual, "0000")
+		})
+
+		Convey("Seek w SEEK_SET should seek to right offset", func() {
+			r := newRS()
+			So(r, ShouldNotBeNil)
+			defer r.Close()
+			s, err := r.Seek(4*64, os.SEEK_SET)
+			So(s, ShouldEqual, 4*64)
+			So(err, ShouldBeNil)
+			buf := make([]byte, 4)
+			n, err := io.ReadFull(r, buf)
+			So(n, ShouldEqual, 4)
+			So(err, ShouldBeNil)
+			So(string(buf), ShouldEqual, "0064")
+		})
+
+		Convey("Read + Seek w SEEK_CUR should seek to right offset", func() {
+			r := newRS()
+			So(r, ShouldNotBeNil)
+			defer r.Close()
+			buf := make([]byte, 4)
+			io.ReadFull(r, buf)
+			s, err := r.Seek(4*64, os.SEEK_CUR)
+			So(s, ShouldEqual, 4*64+4)
+			So(err, ShouldBeNil)
+			n, err := io.ReadFull(r, buf)
+			So(n, ShouldEqual, 4)
+			So(err, ShouldBeNil)
+			So(string(buf), ShouldEqual, "0065")
+		})
+
+		Convey("Seek w SEEK_END should seek to right offset", func() {
+			r := newRS()
+			So(r, ShouldNotBeNil)
+			defer r.Close()
+			buf := make([]byte, 4)
+			io.ReadFull(r, buf)
+			s, err := r.Seek(4, os.SEEK_END)
+			So(s, ShouldEqual, SZ*4-4)
+			So(err, ShouldBeNil)
+			n, err := io.ReadFull(r, buf)
+			So(n, ShouldEqual, 4)
+			So(err, ShouldBeNil)
+			So(string(buf), ShouldEqual, fmt.Sprintf("%04d", SZ-1))
+		})
+
+		Convey("Short seek should consume existing request", func() {
+			r := newRS()
+			So(r, ShouldNotBeNil)
+			defer r.Close()
+			buf := make([]byte, 4)
+			So(r.Requests, ShouldEqual, 0)
+			io.ReadFull(r, buf)
+			So(r.Requests, ShouldEqual, 1)
+			s, err := r.Seek(shortSeekBytes, os.SEEK_CUR)
+			So(r.Requests, ShouldEqual, 1)
+			So(s, ShouldEqual, shortSeekBytes+4)
+			So(err, ShouldBeNil)
+			n, err := io.ReadFull(r, buf)
+			So(n, ShouldEqual, 4)
+			So(err, ShouldBeNil)
+			So(string(buf), ShouldEqual, "0257")
+			So(r.Requests, ShouldEqual, 1)
+		})
+
+		Convey("Long seek should do a new request", func() {
+			r := newRS()
+			So(r, ShouldNotBeNil)
+			defer r.Close()
+			buf := make([]byte, 4)
+			So(r.Requests, ShouldEqual, 0)
+			io.ReadFull(r, buf)
+			So(r.Requests, ShouldEqual, 1)
+			s, err := r.Seek(shortSeekBytes+1, os.SEEK_CUR)
+			So(r.Requests, ShouldEqual, 1)
+			So(s, ShouldEqual, shortSeekBytes+4+1)
+			So(err, ShouldBeNil)
+			n, err := io.ReadFull(r, buf)
+			So(n, ShouldEqual, 4)
+			So(err, ShouldBeNil)
+			So(string(buf), ShouldEqual, "2570")
+			So(r.Requests, ShouldEqual, 2)
+		})
+	})
+}

internal/zipartifacts/open_archive.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/jfbus/httprs"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/httprs"
 
 	"gitlab.com/gitlab-org/labkit/correlation"
 	"gitlab.com/gitlab-org/labkit/mask"