Had a little play with how the http.StatusContinue behavior works in a potential Knative setup.

continue.go

package main

import (
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net/http"
	"net/http/httptest"
	"net/http/httputil"
	"net/url"
	"strings"
	"testing"
)

const bodySize = 1024 * 1024 * 10

func TestContinueBuffersNormally(t *testing.T) {
	app := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprint(w, "success")
	}))

	proxyToApp := httputil.NewSingleHostReverseProxy(parseURL(t, app.URL))
	queue := httptest.NewServer(queueHandler(proxyToApp, false))

	proxyToQueue := httputil.NewSingleHostReverseProxy(parseURL(t, queue.URL))
	activatorBuffer := &recordingReader{}
	activator := httptest.NewServer(activatorHandler(proxyToQueue, activatorBuffer))

	status, body := send(t, activator.URL, randomString(bodySize))
	if status != http.StatusOK || body != "success" {
		t.Fatalf("Excpected a successful request, got: %d %s", status, body)
	}

	if activatorBuffer.read == 0 {
		t.Fatal("Expected body to be read at least a bit, but it wasn't")
	}
}

func TestContinueReadsFully(t *testing.T) {
	app := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		io.Copy(ioutil.Discard, r.Body)
		fmt.Fprint(w, "success")
	}))

	proxyToApp := httputil.NewSingleHostReverseProxy(parseURL(t, app.URL))
	queue := httptest.NewServer(queueHandler(proxyToApp, false))

	proxyToQueue := httputil.NewSingleHostReverseProxy(parseURL(t, queue.URL))
	activatorBuffer := &recordingReader{}
	activator := httptest.NewServer(activatorHandler(proxyToQueue, activatorBuffer))

	status, body := send(t, activator.URL, randomString(bodySize))
	if status != http.StatusOK || body != "success" {
		t.Fatalf("Excpected a successful request, got: %d %s", status, body)
	}

	if activatorBuffer.read != bodySize {
		t.Fatalf("Expected body to be read fully, only read %d", activatorBuffer.read)
	}
}

func TestContinueReadsNone(t *testing.T) {
	app := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		io.Copy(ioutil.Discard, r.Body)
		defer r.Body.Close()
		fmt.Fprint(w, "success")
	}))

	proxyToApp := httputil.NewSingleHostReverseProxy(parseURL(t, app.URL))
	queue := httptest.NewServer(queueHandler(proxyToApp, true))

	proxyToQueue := httputil.NewSingleHostReverseProxy(parseURL(t, queue.URL))
	activatorBuffer := &recordingReader{}
	activator := httptest.NewServer(activatorHandler(proxyToQueue, activatorBuffer))

	status, body := send(t, activator.URL, randomString(bodySize))
	if status != http.StatusBadGateway {
		t.Fatalf("Excpected a bad request, got: %d %s", status, body)
	}

	if activatorBuffer.read != 0 {
		t.Fatalf("Expected no body to be read from the activator, read %d", activatorBuffer.read)
	}
}

// queueHandler replicates the behavior of the queue-proxy.
// It removes the "Expect" header so the application doesn't see it.
// It also needs to make sure the entire body is read if it proxies at all.
func queueHandler(next http.Handler, fail bool) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if fail {
			w.WriteHeader(http.StatusBadGateway)
			return
		}

		consumer := &consumingReader{readCloser: r.Body}
		r.Body = consumer

		r.Header.Del("Expect")
		next.ServeHTTP(w, r)
	}
}

// activatorHandler replicates the behavior of the activator.
// It adds the "Expect" header and proxies to the queue-proxy.
func activatorHandler(next http.Handler, recorder *recordingReader) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		r.Header.Add("Expect", "100-continue")

		recorder.readCloser = r.Body
		r.Body = recorder

		next.ServeHTTP(w, r)
	}
}

// consumingReader makes sure the entire request is consumed before closing it. This
// prevents connection failures and allows the connection to be reused should the app
// not consume the entire body.
type consumingReader struct {
	readCloser io.ReadCloser
	sawEOF     bool
}

func (c *consumingReader) Read(p []byte) (int, error) {
	n, err := c.readCloser.Read(p)
	if err == io.EOF {
		c.sawEOF = true
	}

	return n, err
}

func (c *consumingReader) Close() error {
	if !c.sawEOF {
		io.Copy(ioutil.Discard, c.readCloser)
	}
	return c.readCloser.Close()

}

/*
TEST HELPERS
*/

func send(t *testing.T, url, body string) (int, string) {
	r, err := http.NewRequest("POST", url, strings.NewReader(body))
	if err != nil {
		t.Fatal("Failed to create request")
	}

	client := http.Client{}
	resp, err := client.Do(r)
	if err != nil {
		t.Fatal("Failed to send request", err)
	}
	defer resp.Body.Close()

	gotBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		t.Fatal("Failed to read body", err)
	}

	return resp.StatusCode, string(gotBody)
}

type recordingReader struct {
	read       int
	readCloser io.ReadCloser
}

func (r *recordingReader) Read(p []byte) (int, error) {
	n, err := r.readCloser.Read(p)
	r.read += n
	return n, err
}

func (r *recordingReader) Close() error {
	return r.readCloser.Close()
}

func randomString(n int) string {
	var letter = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")

	b := make([]rune, n)
	for i := range b {
		b[i] = letter[rand.Intn(len(letter))]
	}
	return string(b)
}

func parseURL(t *testing.T, str string) *url.URL {
	t.Helper()

	parsed, err := url.Parse(str)
	if err != nil {
		t.Fatal("Failed to parse URL", err)
	}
	return parsed
}