fshttp: add a fault injector for testing transient HTTP failures

Add fshttp.SetFaultInjector, which installs a function consulted by
every Transport before a request is sent. The injector can synthesise
an error status code or a transport error for chosen requests. The
request body is drained and closed as a real round trip would, but
nothing reaches the server.

This lets the integration tests check that backends cope with a
transient failure part way through an upload - in particular that a
retry re-sends the same data rather than an already consumed or
freed buffer - without needing a fake server for each backend.

fs/fshttp/fault.go

@@ -0,0 +1,67 @@
+package fshttp
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+)
+
+// FaultInjector decides whether to fail req instead of sending it.
+//
+// It returns a non-zero HTTP status code to synthesise an error
+// response, a non-nil error to fail the request at the transport, or
+// (0, nil) to send the request normally. When a fault is injected the
+// request body is drained and closed as it would be by a real round
+// trip, but nothing is sent to the server.
+type FaultInjector func(req *http.Request) (statusCode int, err error)
+
+var (
+	faultInjectorMu sync.RWMutex
+	faultInjector   FaultInjector
+)
+
+// SetFaultInjector installs f as the fault injector for every Transport,
+// or removes it if f is nil.
+//
+// This is intended for tests which need to check how callers cope with
+// transient HTTP failures, such as whether an upload is retried
+// correctly after a 5xx.
+func SetFaultInjector(f FaultInjector) {
+	faultInjectorMu.Lock()
+	defer faultInjectorMu.Unlock()
+	faultInjector = f
+}
+
+// injectFault consults the fault injector and returns the synthesised
+// response or error for req, or (nil, nil) if it should be sent.
+func injectFault(req *http.Request) (*http.Response, error) {
+	faultInjectorMu.RLock()
+	f := faultInjector
+	faultInjectorMu.RUnlock()
+	if f == nil {
+		return nil, nil
+	}
+	statusCode, err := f(req)
+	if statusCode == 0 && err == nil {
+		return nil, nil
+	}
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+		_ = req.Body.Close()
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &http.Response{
+		Status:        fmt.Sprintf("%d %s", statusCode, http.StatusText(statusCode)),
+		StatusCode:    statusCode,
+		Proto:         "HTTP/1.1",
+		ProtoMajor:    1,
+		ProtoMinor:    1,
+		Header:        make(http.Header),
+		Body:          http.NoBody,
+		ContentLength: 0,
+		Request:       req,
+	}, nil
+}

fs/fshttp/fault_test.go

@@ -0,0 +1,102 @@
+package fshttp
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// trackingBody records whether a request body was read to EOF and closed
+type trackingBody struct {
+	io.Reader
+	eof    bool
+	closed bool
+}
+
+func (b *trackingBody) Read(p []byte) (n int, err error) {
+	n, err = b.Reader.Read(p)
+	if err == io.EOF {
+		b.eof = true
+	}
+	return n, err
+}
+
+func (b *trackingBody) Close() error {
+	b.closed = true
+	return nil
+}
+
+func TestFaultInjector(t *testing.T) {
+	var hits atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits.Add(1)
+		_, _ = io.Copy(io.Discard, r.Body)
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer server.Close()
+
+	ctx := context.Background()
+	client := NewClientCustom(ctx, nil)
+	injectErr := errors.New("injected error")
+
+	// Fail the first request with a status code and the second with an
+	// error, then let everything through
+	var calls atomic.Int32
+	SetFaultInjector(func(req *http.Request) (int, error) {
+		switch calls.Add(1) {
+		case 1:
+			return http.StatusInternalServerError, nil
+		case 2:
+			return 0, injectErr
+		}
+		return 0, nil
+	})
+	defer SetFaultInjector(nil)
+
+	do := func() (*trackingBody, *http.Response, error) {
+		body := &trackingBody{Reader: strings.NewReader("hello")}
+		req, err := http.NewRequestWithContext(ctx, "PUT", server.URL, body)
+		require.NoError(t, err)
+		resp, err := client.Do(req)
+		if err == nil {
+			_ = resp.Body.Close()
+		}
+		return body, resp, err
+	}
+
+	body, resp, err := do()
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusInternalServerError, resp.StatusCode)
+	assert.Equal(t, "500 Internal Server Error", resp.Status)
+	assert.True(t, body.eof, "body should be drained")
+	assert.True(t, body.closed, "body should be closed")
+	assert.Equal(t, int32(0), hits.Load(), "server should not see a faulted request")
+
+	body, _, err = do()
+	require.ErrorIs(t, err, injectErr)
+	assert.True(t, body.eof, "body should be drained")
+	assert.True(t, body.closed, "body should be closed")
+	assert.Equal(t, int32(0), hits.Load(), "server should not see a faulted request")
+
+	_, resp, err = do()
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusNoContent, resp.StatusCode)
+	assert.Equal(t, int32(1), hits.Load())
+
+	// Removing the injector lets requests through
+	SetFaultInjector(nil)
+	_, resp, err = do()
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusNoContent, resp.StatusCode)
+	assert.Equal(t, int32(2), hits.Load())
+	assert.Equal(t, int32(3), calls.Load())
+
+}

fs/fshttp/http.go

@@ -671,8 +671,11 @@ func (t *Transport) RoundTrip(req *http.Request) (resp *http.Response, err error
 	if t.dump&fs.DumpTrace != 0 {
 		traceReq = req.WithContext(httptrace.WithClientTrace(req.Context(), newClientTrace(req)))
 	}
-	// Do round trip
-	resp, err = t.Transport.RoundTrip(traceReq)
+	// Do round trip unless a fault is being injected
+	resp, err = injectFault(traceReq)
+	if resp == nil && err == nil {
+		resp, err = t.Transport.RoundTrip(traceReq)
+	}
 	// Dump response, and the request too if we deferred it for --dump errors
 	if wantDump && (!onError || isRetryableResponse(resp, err)) {
 		logMutex.Lock()