req.Body not reset in doWithRetry

[embano1]

Scenario:

When using a net/http HTTP server as remote/receiver (instead of the receiver from this package), ceClient has issues when retrying a non-successful HTTP response.

Error: "msg":"Invalid result type, not HTTP Result: Post \"http://always-fail.default.127.0.0.1.sslip.io\": http: ContentLength=34 with Body length 0 (type: *protocol.Receipt)"}

This happens when the remote server returns a message in the response, e.g. „server down“ (not CE) but not when just returning a status code and empty response body.

The issue is caused by func (p *Protocol) doWithRetry() because it does not reset the req.Body during subsequent attempts.

Expected behavior:

Return encoding error, but correctly reset req.Body, e.g. return "failed to convert response into event: unknown Message encoding\n500: (3x)"

Client code:

/*
 Copyright 2021 The CloudEvents Authors
 SPDX-License-Identifier: Apache-2.0
*/

package main

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"github.com/cloudevents/sdk-go/observability/opencensus/v2/client"
	"go.uber.org/zap"
	"knative.dev/eventing/pkg/kncloudevents"
	"knative.dev/pkg/logging"

	cloudevents "github.com/cloudevents/sdk-go/v2"
	"github.com/cloudevents/sdk-go/v2/protocol"
	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
)

func main() {
	ctx := cloudevents.ContextWithTarget(context.Background(), "http://always-fail.default.127.0.0.1.sslip.io")

	topts := []cehttp.Option{cehttp.WithIsRetriableFunc(func(statusCode int) bool {
		retry, _ := kncloudevents.SelectiveRetry(ctx, &http.Response{StatusCode: statusCode}, nil)
		return retry
	})}

	ceClient, err := client.NewClientHTTP(topts, nil)
	if err != nil {
		panic(err)
	}

	e := cloudevents.NewEvent()
	e.SetType("com.cloudevents.sample.sent")
	e.SetSource("https://github.com/cloudevents/sdk-go/v2/samples/httpb/sender")
	_ = e.SetData(cloudevents.ApplicationJSON, map[string]interface{}{
		"id":      1,
		"message": "Hello, World!",
	})

	ctx = cloudevents.ContextWithRetriesLinearBackoff(ctx, 10*time.Millisecond, 3)
	response, result := ceClient.Request(ctx, e)

	if !isSuccess(ctx, result) {
		logging.FromContext(ctx).Errorw("Failed to deliver", zap.Error(result))
		return
	}

	fmt.Printf("got: %v", response)
}

func isSuccess(ctx context.Context, result protocol.Result) bool {
	var retriesResult *cehttp.RetriesResult
	if cloudevents.ResultAs(result, &retriesResult) {
		var httpResult *cehttp.Result
		if cloudevents.ResultAs(retriesResult.Result, &httpResult) {
			retry, _ := kncloudevents.SelectiveRetry(ctx, &http.Response{StatusCode: httpResult.StatusCode}, nil)
			return !retry
		}
		logging.FromContext(ctx).Warnf("Invalid result type, not HTTP Result: %v (type: %T)", retriesResult.Result, retriesResult.Result)
		return false
	}

	logging.FromContext(ctx).Warnf("Invalid result type, not RetriesResult")
	return false
}

HTTP server code (note: just using net/http)

package main

import (
	"context"
	"errors"
	"net/http"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/go-chi/chi/v5/middleware"
	"knative.dev/pkg/logging"
	"knative.dev/pkg/signals"
)

const (
	addr    = ":8080"
	timeout = 5 * time.Second
)

func main() {
	ctx := signals.NewContext()

	r := chi.NewRouter()
	r.Use(middleware.Logger)
	r.Post("/", func(w http.ResponseWriter, r *http.Request) {
		// w.WriteHeader(http.StatusInternalServerError) <- just doing this works!
		http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
	})

	srv := http.Server{
		Addr:         addr,
		Handler:      r,
		ReadTimeout:  timeout,
		WriteTimeout: timeout,
	}

	go func() {
		<-ctx.Done()
		logging.FromContext(ctx).Info("shutting down")
		if err := srv.Shutdown(context.Background()); err != nil {
			logging.FromContext(ctx).Warnf("could not shutdown: %v", err)
		}
	}()

	logging.FromContext(ctx).Infow("listening", "address", addr)
	if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
		logging.FromContext(ctx).Fatal(err)
	}
}

Patch

diff --git a/v2/protocol/http/protocol_retry.go b/v2/protocol/http/protocol_retry.go
index fb7bcd2..9050469 100644
--- a/v2/protocol/http/protocol_retry.go
+++ b/v2/protocol/http/protocol_retry.go
@@ -6,8 +6,10 @@
 package http
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"io"
 	"net/http"
 	"net/url"
 	"time"
@@ -53,6 +55,17 @@ func (p *Protocol) doWithRetry(ctx context.Context, params *cecontext.RetryParam
 	retry := 0
 	results := make([]protocol.Result, 0)
 
+	var (
+		body []byte
+		err  error
+	)
+	if req.Body != nil {
+		body, err = io.ReadAll(req.Body)
+		if err != nil {
+			panic(err)
+		}
+	}
+
 	for {
 		msg, result := p.doOnce(req)
 
@@ -90,6 +103,13 @@ func (p *Protocol) doWithRetry(ctx context.Context, params *cecontext.RetryParam
 		}
 
 	DoBackoff:
+		if req.Body != nil {
+			req.Body = io.NopCloser(bytes.NewReader(body))
+			req.GetBody = func() (io.ReadCloser, error) {
+				return io.NopCloser(bytes.NewReader(body)), nil
+			}
+		}
+
 		// Wait for the correct amount of backoff time.
 
 		// total tries = retry + 1

[n3wscott]

nice! Want to make a PR?

[embano1]

nice! Want to make a PR?

Sure, I just need to dig a little deeper as there might be another anomaly in the way requests are handled. Not sure I get the full picture yet.

[embano1]

Update:

I can reproduce the issue with net/http the following handlers:

func(w http.ResponseWriter, r *http.Request) {
		http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
}

func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusInternalServerError)
}

The second handler does not always show the http: ContentLength=34 with Body length 0 *url.Error, but I managed to catch it several times with a debugger.

I was also able to catch it once using the cloudevents receiver and a simple event handler:

func receive(ctx context.Context, event cloudevents.Event) error {
	fmt.Printf("got event: %v\n", event)
	return http.NewResult(nethttp.StatusInternalServerError, nethttp.StatusText(nethttp.StatusInternalServerError))
}

Still not sure where and how it gets triggered in all these (rare) scenarios, but perhaps there's a race which only shows during debugging - typical Heisenbug :D

Also, I was wondering why the unit tests were not exhibiting the problem: this is caused by only using a RoundTripper mock instead of a "full" HTTP server which reads/modifies the body (also the tests didn't use a body which was another reason we did not see this).