call finish on unused messages; tidy retry logic

taehwoi · taehwoi · commit bf156f1bd1a7 · 2023-07-14T16:43:54.000+09:00
Signed-off-by: taehwoi oum &lt;indiofish@gmail.com&gt;
diff --git a/v2/protocol/http/protocol_retry.go b/v2/protocol/http/protocol_retry.go
@@ -12,7 +12,6 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
-	"net/url"
 	"time"
 
 	"go.uber.org/zap"
@@ -52,7 +51,7 @@ func (p *Protocol) doOnce(req *http.Request) (binding.Message, protocol.Result)
 }
 
 func (p *Protocol) doWithRetry(ctx context.Context, params *cecontext.RetryParams, req *http.Request) (binding.Message, error) {
-	then := time.Now()
+	start := time.Now()
 	retry := 0
 	results := make([]protocol.Result, 0)
 
@@ -79,51 +78,34 @@ func (p *Protocol) doWithRetry(ctx context.Context, params *cecontext.RetryParam
 
 		// Fast track common case.
 		if protocol.IsACK(result) {
-			return msg, NewRetriesResult(result, retry, then, results)
+			return msg, NewRetriesResult(result, retry, start, results)
 		}
 
-		// Try again?
-		//
-		// Make sure the error was something we should retry.
-
-		{
-			var uErr *url.Error
-			if errors.As(result, &uErr) {
-				goto DoBackoff
+		var httpResult *Result
+		if errors.As(result, &httpResult) {
+			sc := httpResult.StatusCode
+			if !p.isRetriableFunc(sc) {
+				cecontext.LoggerFrom(ctx).Debugw("status code not retryable, will not try again",
+					zap.Error(httpResult),
+					zap.Int("statusCode", sc))
+				return msg, NewRetriesResult(result, retry, start, results)
 			}
 		}
 
-		{
-			var httpResult *Result
-			if errors.As(result, &httpResult) {
-				sc := httpResult.StatusCode
-				if p.isRetriableFunc(sc) {
-					// retry!
-					goto DoBackoff
-				} else {
-					// Permanent error
-					cecontext.LoggerFrom(ctx).Debugw("status code not retryable, will not try again",
-						zap.Error(httpResult),
-						zap.Int("statusCode", sc))
-					return msg, NewRetriesResult(result, retry, then, results)
-				}
-			}
-		}
-
-	DoBackoff:
-		resetBody(req, body)
-
-		// Wait for the correct amount of backoff time.
-
 		// total tries = retry + 1
-		if err := params.Backoff(ctx, retry+1); err != nil {
+		if err = params.Backoff(ctx, retry+1); err != nil {
 			// do not try again.
 			cecontext.LoggerFrom(ctx).Debugw("backoff error, will not try again", zap.Error(err))
-			return msg, NewRetriesResult(result, retry, then, results)
+			return msg, NewRetriesResult(result, retry, start, results)
 		}
 
 		retry++
+		resetBody(req, body)
 		results = append(results, result)
+		if msg != nil {
+			// avoid leak, forget message, ignore error
+			_ = msg.Finish(nil)
+		}
 	}
 }
 
diff --git a/v2/protocol/http/protocol_retry_test.go b/v2/protocol/http/protocol_retry_test.go
@@ -14,6 +14,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
 
 	"github.com/cloudevents/sdk-go/v2/binding"
 	cecontext "github.com/cloudevents/sdk-go/v2/context"
@@ -192,6 +193,7 @@ func TestRequestWithRetries_linear(t *testing.T) {
 			}
 
 			ctx := cecontext.WithTarget(context.Background(), srv.URL)
+			ctx = cecontext.WithLogger(ctx, zaptest.NewLogger(t).Sugar())
 			ctxWithRetries := cecontext.WithRetriesLinearBackoff(ctx, tc.delay, tc.retries)
 
 			dummyMsg := binding.ToMessage(&tc.event)
