delay PostUploadPack response until request is fully read

Git and Gitaly stream upload-pack response data as soon as it is
available. Some HTTP clients do not support the response being streamed
back while the server is still reading from the request body. To avoid
this, the request was buffered in to a temporary file before passing it
to Gitaly for handling. The buffer had a maximum size of 10MB, which is
now reached by requests to large repositories. These requests were then
truncated, causing Git to fail.

This commit fixes the problem by removing the request buffering and the
maximum size. Instead, the response is buffered in to a temporary file
until the request body is fully read, thus avoiding the problem of
streaming the request and the response simultaneously.

gitaly_test.go

@@ -13,6 +13,7 @@ import (
 	"os/exec"
 	"path"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -190,15 +191,17 @@ func TestPostReceivePackProxiedToGitalySuccessfully(t *testing.T) {
 
 	gitProtocol := "fake Git protocol"
 	resource := "/gitlab-org/gitlab-test.git/git-receive-pack"
-	resp, body := httpPost(
+	resp := httpPost(
 		t,
 		ws.URL+resource,
 		map[string]string{
 			"Content-Type": "application/x-git-receive-pack-request",
 			"Git-Protocol": gitProtocol,
 		},
-		testhelper.GitalyReceivePackResponseMock,
+		bytes.NewReader(testhelper.GitalyReceivePackResponseMock),
 	)
+	defer resp.Body.Close()
+	body := string(testhelper.ReadAll(t, resp.Body))
 
 	split := strings.SplitN(body, "\000", 2)
 	require.Len(t, split, 2)
@@ -257,6 +260,11 @@ func TestPostReceivePackProxiedToGitalyInterrupted(t *testing.T) {
 	}
 }
 
+// ReaderFunc is an adapter to turn a conforming function into an io.Reader.
+type ReaderFunc func(b []byte) (int, error)
+
+func (r ReaderFunc) Read(b []byte) (int, error) { return r(b) }
+
 func TestPostUploadPackProxiedToGitalySuccessfully(t *testing.T) {
 	for i, tc := range []struct {
 		showAllRefs bool
@@ -283,19 +291,39 @@ func TestPostUploadPackProxiedToGitalySuccessfully(t *testing.T) {
 
 			gitProtocol := "fake git protocol"
 			resource := "/gitlab-org/gitlab-test.git/git-upload-pack"
-			resp, body := httpPost(
+
+			requestReader := bytes.NewReader(testhelper.GitalyUploadPackResponseMock)
+			var m sync.Mutex
+			requestReadFinished := false
+			resp := httpPost(
 				t,
 				ws.URL+resource,
 				map[string]string{
 					"Content-Type": "application/x-git-upload-pack-request",
 					"Git-Protocol": gitProtocol,
 				},
-				testhelper.GitalyUploadPackResponseMock,
+				ReaderFunc(func(b []byte) (int, error) {
+					n, err := requestReader.Read(b)
+					if err != nil {
+						m.Lock()
+						requestReadFinished = true
+						m.Unlock()
+					}
+					return n, err
+				}),
 			)
-
+			defer resp.Body.Close()
 			require.Equal(t, 200, resp.StatusCode, "POST %q", resource)
 			testhelper.AssertResponseHeader(t, resp, "Content-Type", "application/x-git-upload-pack-result")
 
+			m.Lock()
+			requestFinished := requestReadFinished
+			m.Unlock()
+			if !requestFinished {
+				t.Fatalf("response written before request was fully read")
+			}
+
+			body := string(testhelper.ReadAll(t, resp.Body))
 			bodySplit := strings.SplitN(body, "\000", 2)
 			require.Len(t, bodySplit, 2)
 

internal/git/upload-pack.go

@@ -21,13 +21,8 @@ var (
 func handleUploadPack(w *HttpResponseWriter, r *http.Request, a *api.Response) error {
 	ctx := r.Context()
 
-	// The body will consist almost entirely of 'have XXX' and 'want XXX'
-	// lines; these are about 50 bytes long. With a size limit of 10MiB, the
-	// client can send over 200,000 have/want lines.
-	sizeLimited := io.LimitReader(r.Body, 10*1024*1024)
-
 	// Prevent the client from holding the connection open indefinitely. A
-	// transfer rate of 17KiB/sec is sufficient to fill the 10MiB buffer in
+	// transfer rate of 17KiB/sec is sufficient to send 10MiB of data in
 	// ten minutes, which seems adequate. Most requests will be much smaller.
 	// This mitigates a use-after-check issue.
 	//
@@ -36,21 +31,16 @@ func handleUploadPack(w *HttpResponseWriter, r *http.Request, a *api.Response) e
 	readerCtx, cancel := context.WithTimeout(ctx, uploadPackTimeout)
 	defer cancel()
 
-	limited := helper.NewContextReader(readerCtx, sizeLimited)
-	buffer, err := helper.ReadAllTempfile(limited)
-
-	if err != nil {
-		return fmt.Errorf("ReadAllTempfile: %v", err)
-	}
-	defer buffer.Close()
-	r.Body.Close()
+	limited := helper.NewContextReader(readerCtx, r.Body)
+	cr, cw := helper.NewWriteAfterReader(limited, w)
+	defer cw.Flush()
 
 	action := getService(r)
 	writePostRPCHeader(w, action)
 
 	gitProtocol := r.Header.Get("Git-Protocol")
 
-	return handleUploadPackWithGitaly(ctx, a, buffer, w, gitProtocol)
+	return handleUploadPackWithGitaly(ctx, a, cr, cw, gitProtocol)
 }
 
 func handleUploadPackWithGitaly(ctx context.Context, a *api.Response, clientRequest io.Reader, clientResponse io.Writer, gitProtocol string) error {

internal/git/upload-pack_test.go

 package git
 
 import (
+	"fmt"
+	"io/ioutil"
+	"net"
 	"net/http/httptest"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
+	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 
 	"gitlab.com/gitlab-org/gitlab-workhorse/internal/api"
+	"gitlab.com/gitlab-org/gitlab-workhorse/internal/gitaly"
 )
 
 var (
@@ -23,17 +33,54 @@ func (f *fakeReader) Read(b []byte) (int, error) {
 	return f.n, f.err
 }
 
+type smartHTTPServiceServer struct {
+	gitalypb.UnimplementedSmartHTTPServiceServer
+	PostUploadPackFunc func(gitalypb.SmartHTTPService_PostUploadPackServer) error
+}
+
+func (srv *smartHTTPServiceServer) PostUploadPack(s gitalypb.SmartHTTPService_PostUploadPackServer) error {
+	return srv.PostUploadPackFunc(s)
+}
+
 func TestUploadPackTimesOut(t *testing.T) {
 	uploadPackTimeout = time.Millisecond
 	defer func() { uploadPackTimeout = originalUploadPackTimeout }()
 
+	addr, cleanUp := startSmartHTTPServer(t, &smartHTTPServiceServer{
+		PostUploadPackFunc: func(stream gitalypb.SmartHTTPService_PostUploadPackServer) error {
+			_, err := stream.Recv() // trigger a read on the client request body
+			require.NoError(t, err)
+			return nil
+		},
+	})
+	defer cleanUp()
+
 	body := &fakeReader{n: 0, err: nil}
 
 	w := httptest.NewRecorder()
 	r := httptest.NewRequest("GET", "/", body)
-	a := &api.Response{}
+	a := &api.Response{GitalyServer: gitaly.Server{Address: addr}}
 
 	err := handleUploadPack(NewHttpResponseWriter(w), r, a)
-	require.EqualError(t, err, "ReadAllTempfile: context deadline exceeded")
+	require.EqualError(t, err, "smarthttp.UploadPack: busyReader: context deadline exceeded")
+}
+
+func startSmartHTTPServer(t testing.TB, s gitalypb.SmartHTTPServiceServer) (string, func()) {
+	tmp, err := ioutil.TempDir("", "")
+	require.NoError(t, err)
+
+	socket := filepath.Join(tmp, "gitaly.sock")
+	ln, err := net.Listen("unix", socket)
+	require.NoError(t, err)
+
+	srv := grpc.NewServer()
+	gitalypb.RegisterSmartHTTPServiceServer(srv, s)
+	go func() {
+		require.NoError(t, srv.Serve(ln))
+	}()
 
+	return fmt.Sprintf("%s://%s", ln.Addr().Network(), ln.Addr().String()), func() {
+		srv.Stop()
+		assert.NoError(t, os.RemoveAll(tmp), "error removing temp dir %q", tmp)
+	}
 }

internal/testhelper/gitaly.go

@@ -195,11 +195,16 @@ func (s *GitalyTestServer) PostUploadPack(stream gitalypb.SmartHTTPService_PostU
 		return err
 	}
 
-	data := []byte(strings.Join([]string{
-		jsonString,
-	}, "\000") + "\000")
+	if err := stream.Send(&gitalypb.PostUploadPackResponse{
+		Data: []byte(strings.Join([]string{jsonString}, "\000") + "\000"),
+	}); err != nil {
+		return err
+	}
 
-	// The body of the request starts in the second message
+	nSends := 0
+	// The body of the request starts in the second message. Gitaly streams PostUploadPack responses
+	// as soon as possible without reading the request completely first. We stream messages here
+	// directly back to the client to simulate the streaming of the actual implementation.
 	for {
 		req, err := stream.Recv()
 		if err != nil {
@@ -209,12 +214,12 @@ func (s *GitalyTestServer) PostUploadPack(stream gitalypb.SmartHTTPService_PostU
 			break
 		}
 
-		data = append(data, req.GetData()...)
-	}
+		if err := stream.Send(&gitalypb.PostUploadPackResponse{Data: req.GetData()}); err != nil {
+			return err
+		}
 
-	nSends, _ := sendBytes(data, 100, func(p []byte) error {
-		return stream.Send(&gitalypb.PostUploadPackResponse{Data: p})
-	})
+		nSends++
+	}
 
 	if nSends <= 1 {
 		panic("should have sent more than one message")

internal/testhelper/testhelper.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
@@ -16,6 +17,7 @@ import (
 	"testing"
 
 	"github.com/dgrijalva/jwt-go"
+	"github.com/stretchr/testify/require"
 
 	"gitlab.com/gitlab-org/labkit/log"
 
@@ -177,6 +179,14 @@ func LoadFile(t *testing.T, filePath string) string {
 	return string(content)
 }
 
+func ReadAll(t *testing.T, r io.Reader) []byte {
+	t.Helper()
+
+	b, err := ioutil.ReadAll(r)
+	require.NoError(t, err)
+	return b
+}
+
 func ParseJWT(token *jwt.Token) (interface{}, error) {
 	// Don't forget to validate the alg is what you expect:
 	if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {

main_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
@@ -593,8 +594,8 @@ func httpGet(t *testing.T, url string, headers map[string]string) (*http.Respons
 	return resp, string(b)
 }
 
-func httpPost(t *testing.T, url string, headers map[string]string, reqBody []byte) (*http.Response, string) {
-	req, err := http.NewRequest("POST", url, bytes.NewReader(reqBody))
+func httpPost(t *testing.T, url string, headers map[string]string, reqBody io.Reader) *http.Response {
+	req, err := http.NewRequest("POST", url, reqBody)
 	require.NoError(t, err)
 
 	for k, v := range headers {
@@ -603,12 +604,8 @@ func httpPost(t *testing.T, url string, headers map[string]string, reqBody []byt
 
 	resp, err := http.DefaultClient.Do(req)
 	require.NoError(t, err)
-	defer resp.Body.Close()
 
-	b, err := ioutil.ReadAll(resp.Body)
-	require.NoError(t, err)
-
-	return resp, string(b)
+	return resp
 }
 
 func assertNginxResponseBuffering(t *testing.T, expected string, resp *http.Response, msgAndArgs ...interface{}) {