Streaming: convert HTTP service invocation handler to net/http (dapr#6594)

* Streaming: convert HTTP service invocation handler to net/http

Signed-off-by: ItalyPaleAle <43508+ItalyPaleAle@users.noreply.github.com>

* Remove extra slashes in service invocation when using dapr-app-id header

Signed-off-by: ItalyPaleAle <43508+ItalyPaleAle@users.noreply.github.com>

* Re-enable test

Signed-off-by: ItalyPaleAle <43508+ItalyPaleAle@users.noreply.github.com>

---------

Signed-off-by: ItalyPaleAle <43508+ItalyPaleAle@users.noreply.github.com>
Signed-off-by: Alessandro (Ale) Segala <43508+ItalyPaleAle@users.noreply.github.com>
Co-authored-by: Dapr Bot <56698301+dapr-bot@users.noreply.github.com>

Author: ItalyPaleAle

pkg/grpc/api.go

@@ -249,7 +249,7 @@ var (
 // Deprecated: Use proxy mode service invocation instead.
 func (a *api) InvokeService(ctx context.Context, in *runtimev1pb.InvokeServiceRequest) (*commonv1pb.InvokeResponse, error) {
 	if a.directMessaging == nil {
-		return nil, status.Errorf(codes.Internal, messages.ErrDirectInvokeNotReady)
+		return nil, messages.ErrDirectInvokeNotReady
 	}
 
 	if invokeServiceDeprecationNoticeShown.CompareAndSwap(false, true) {
@@ -285,7 +285,7 @@ func (a *api) InvokeService(ctx context.Context, in *runtimev1pb.InvokeServiceRe
 			}
 		}
 		if rErr != nil {
-			return rResp, status.Errorf(codes.Internal, messages.ErrDirectInvoke, in.Id, rErr)
+			return rResp, messages.ErrDirectInvoke.WithFormat(in.Id, rErr)
 		}
 
 		rResp.headers = invokev1.InternalMetadataToGrpcMetadata(ctx, imr.Headers(), true)

pkg/http/api.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	nethttp "net/http"
 	"net/url"
+	"path"
 	"strconv"
 	"strings"
 	"time"
@@ -57,6 +58,7 @@ import (
 	"github.com/dapr/dapr/pkg/runtime/compstore"
 	runtimePubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
 	"github.com/dapr/dapr/utils"
+	"github.com/dapr/dapr/utils/responsewriter"
 )
 
 // API returns a list of HTTP endpoints for Dapr.
@@ -278,7 +280,7 @@ func (a *api) constructDirectMessagingEndpoints() []Endpoint {
 			IsFallback:            true,
 			Version:               apiVersionV1,
 			KeepWildcardUnescaped: true,
-			FastHTTPHandler:       a.onDirectMessage,
+			Handler:               a.onDirectMessage,
 		},
 	}
 }
@@ -1077,11 +1079,11 @@ func (a *api) getStateStoreName(reqCtx *fasthttp.RequestCtx) string {
 
 type invokeError struct {
 	statusCode int
-	msg        ErrorResponse
+	msg        []byte
 }
 
 func (ie invokeError) Error() string {
-	return fmt.Sprintf("invokeError (statusCode='%d') msg.errorCode='%s' msg.message='%s'", ie.statusCode, ie.msg.ErrorCode, ie.msg.Message)
+	return fmt.Sprintf("invokeError (statusCode='%d') msg='%v'", ie.statusCode, string(ie.msg))
 }
 
 func (a *api) isHTTPEndpoint(appID string) bool {
@@ -1099,22 +1101,21 @@ func (a *api) getBaseURL(targetAppID string) string {
 	return ""
 }
 
-func (a *api) onDirectMessage(reqCtx *fasthttp.RequestCtx) {
-	targetID, invokeMethodName := findTargetIDAndMethod(string(reqCtx.URI().PathOriginal()), reqCtx.Request.Header.Peek)
+func (a *api) onDirectMessage(w nethttp.ResponseWriter, r *nethttp.Request) {
+	targetID, invokeMethodName := findTargetIDAndMethod(r.URL.String(), r.Header)
 	if targetID == "" {
-		msg := NewErrorResponse("ERR_DIRECT_INVOKE", messages.ErrDirectInvokeNoAppID)
-		fasthttpRespond(reqCtx, fasthttpResponseWithError(nethttp.StatusNotFound, msg))
+		respondWithError(w, messages.ErrDirectInvokeNoAppID)
 		return
 	}
 
-	// Store target and method as user values so they can be picked up by the tracing library
-	reqCtx.SetUserValue("id", targetID)
-	reqCtx.SetUserValue("method", invokeMethodName)
+	// Store target and method as values in the context so they can be picked up by the tracing library
+	rw := responsewriter.EnsureResponseWriter(w)
+	rw.SetUserValue("id", targetID)
+	rw.SetUserValue("method", invokeMethodName)
 
-	verb := strings.ToUpper(string(reqCtx.Method()))
+	verb := strings.ToUpper(r.Method)
 	if a.directMessaging == nil {
-		msg := NewErrorResponse("ERR_DIRECT_INVOKE", messages.ErrDirectInvokeNotReady)
-		fasthttpRespond(reqCtx, fasthttpResponseWithError(nethttp.StatusInternalServerError, msg))
+		respondWithError(w, messages.ErrDirectInvokeNotReady)
 		return
 	}
 
@@ -1134,18 +1135,18 @@ func (a *api) onDirectMessage(reqCtx *fasthttp.RequestCtx) {
 	}
 
 	req := invokev1.NewInvokeMethodRequest(invokeMethodName).
-		WithHTTPExtension(verb, reqCtx.QueryArgs().String()).
-		WithRawDataBytes(reqCtx.Request.Body()).
-		WithContentType(string(reqCtx.Request.Header.ContentType())).
+		WithHTTPExtension(verb, r.URL.RawQuery).
+		WithRawData(r.Body).
+		WithContentType(r.Header.Get("content-type")).
 		// Save headers to internal metadata
-		WithFastHTTPHeaders(&reqCtx.Request.Header)
+		WithHTTPHeaders(r.Header)
 	if policyDef != nil {
 		req.WithReplay(policyDef.HasRetries())
 	}
 	defer req.Close()
 
 	policyRunner := resiliency.NewRunnerWithOptions(
-		reqCtx, policyDef,
+		r.Context(), policyDef,
 		resiliency.RunnerOpts[*invokev1.InvokeMethodResponse]{
 			Disposer: resiliency.DisposerCloser[*invokev1.InvokeMethodResponse],
 		},
@@ -1156,9 +1157,10 @@ func (a *api) onDirectMessage(reqCtx *fasthttp.RequestCtx) {
 		if rErr != nil {
 			// Allowlist policies that are applied on the callee side can return a Permission Denied error.
 			// For everything else, treat it as a gRPC transport error
+			apiErr := messages.ErrDirectInvoke.WithFormat(targetID, rErr)
 			invokeErr := invokeError{
-				statusCode: nethttp.StatusInternalServerError,
-				msg:        NewErrorResponse("ERR_DIRECT_INVOKE", fmt.Sprintf(messages.ErrDirectInvoke, targetID, rErr)),
+				statusCode: apiErr.HTTPCode(),
+				msg:        apiErr.JSONErrorValue(),
 			}
 
 			if status.Code(rErr) == codes.PermissionDenied {
@@ -1181,7 +1183,7 @@ func (a *api) onDirectMessage(reqCtx *fasthttp.RequestCtx) {
 				if rErr != nil {
 					return rResp, invokeError{
 						statusCode: nethttp.StatusInternalServerError,
-						msg:        NewErrorResponse("ERR_MALFORMED_RESPONSE", rErr.Error()),
+						msg:        NewErrorResponse("ERR_MALFORMED_RESPONSE", rErr.Error()).JSONErrorValue(),
 					}
 				}
 			} else {
@@ -1197,67 +1199,70 @@ func (a *api) onDirectMessage(reqCtx *fasthttp.RequestCtx) {
 
 	// Special case for timeouts/circuit breakers since they won't go through the rest of the logic.
 	if errors.Is(err, context.DeadlineExceeded) || breaker.IsErrorPermanent(err) {
-		fasthttpRespond(reqCtx, fasthttpResponseWithError(nethttp.StatusInternalServerError, NewErrorResponse("ERR_DIRECT_INVOKE", err.Error())))
+		respondWithError(w, messages.ErrDirectInvoke.WithFormat(targetID, err))
 		return
 	}
 
 	if resp != nil {
 		headers := resp.Headers()
 		if len(headers) > 0 {
-			invokev1.InternalMetadataToHTTPHeader(reqCtx, headers, reqCtx.Response.Header.Add)
+			invokev1.InternalMetadataToHTTPHeader(r.Context(), headers, w.Header().Add)
 		}
 	}
 
 	invokeErr := invokeError{}
 	if errors.As(err, &invokeErr) {
-		fasthttpRespond(reqCtx, fasthttpResponseWithError(invokeErr.statusCode, invokeErr.msg))
+		respondWithData(w, invokeErr.statusCode, invokeErr.msg)
 		if resp != nil {
 			_ = resp.Close()
 		}
 		return
 	}
 
 	if resp == nil {
-		fasthttpRespond(reqCtx, fasthttpResponseWithError(nethttp.StatusInternalServerError, NewErrorResponse("ERR_DIRECT_INVOKE", fmt.Sprintf(messages.ErrDirectInvoke, targetID, "response object is nil"))))
+		respondWithError(w, messages.ErrDirectInvoke.WithFormat(targetID, "response object is nil"))
 		return
 	}
 	defer resp.Close()
 
 	statusCode := int(resp.Status().Code)
 
-	body, err := resp.RawDataFull()
+	if ct := resp.ContentType(); ct != "" {
+		w.Header().Set("content-type", ct)
+	}
+
+	w.WriteHeader(statusCode)
+
+	_, err = io.Copy(w, resp.RawData())
 	if err != nil {
-		fasthttpRespond(reqCtx, fasthttpResponseWithError(nethttp.StatusInternalServerError, NewErrorResponse("ERR_DIRECT_INVOKE", fmt.Sprintf(messages.ErrDirectInvoke, targetID, err))))
+		respondWithError(w, messages.ErrDirectInvoke.WithFormat(targetID, err))
 		return
 	}
-
-	reqCtx.Response.Header.SetContentType(resp.ContentType())
-	fasthttpRespond(reqCtx, fasthttpResponseWith(statusCode, body))
 }
 
 // findTargetIDAndMethod finds ID of the target service and method from the following three places:
 // 1. HTTP header 'dapr-app-id' (path is method)
 // 2. Basic auth header: `http://dapr-app-id:<service-id>@localhost:3500/<method>`
 // 3. URL parameter: `http://localhost:3500/v1.0/invoke/<app-id>/method/<method>`
-func findTargetIDAndMethod(path string, peekHeader func(string) []byte) (targetID string, method string) {
-	if appID := peekHeader(daprAppID); len(appID) != 0 {
-		return string(appID), strings.TrimPrefix(path, "/")
+func findTargetIDAndMethod(reqPath string, headers nethttp.Header) (targetID string, method string) {
+	if appID := headers.Get(daprAppID); appID != "" {
+		return appID, strings.TrimPrefix(path.Clean(reqPath), "/")
 	}
 
-	if auth := string(peekHeader(fasthttp.HeaderAuthorization)); strings.HasPrefix(auth, "Basic ") {
+	if auth := headers.Get(fasthttp.HeaderAuthorization); strings.HasPrefix(auth, "Basic ") {
 		if s, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(auth, "Basic ")); err == nil {
 			pair := strings.Split(string(s), ":")
 			if len(pair) == 2 && pair[0] == daprAppID {
-				return pair[1], strings.TrimPrefix(path, "/")
+				return pair[1], strings.TrimPrefix(path.Clean(reqPath), "/")
 			}
 		}
 	}
 
 	// If we're here, the handler was probably invoked with /v1.0/invoke/ (or the invocation is invalid, missing the app id provided as header or Basic auth)
 	// However, we are not relying on wildcardParam because the URL may have been sanitized to remove `//``, so `http://` would have been turned into `http:/`
 	// First, check to make sure that the path has the prefix
-	if idx := pathHasPrefix(path, apiVersionV1, "invoke"); idx > 0 {
-		path = path[idx:]
+	if idx := pathHasPrefix(reqPath, apiVersionV1, "invoke"); idx > 0 {
+		reqPath = reqPath[idx:]
 
 		// Scan to find app ID and method
 		// Matches `<appid>/method/<method>`.
@@ -1266,9 +1271,9 @@ func findTargetIDAndMethod(path string, peekHeader func(string) []byte) (targetI
 		// - `http://example.com/method/mymethod`
 		// - `https://example.com/method/mymethod`
 		// - `http%3A%2F%2Fexample.com/method/mymethod`
-		if idx = strings.Index(path, "/method/"); idx > 0 {
-			targetID = path[:idx]
-			method = path[(idx + len("/method/")):]
+		if idx = strings.Index(reqPath, "/method/"); idx > 0 {
+			targetID = reqPath[:idx]
+			method = reqPath[(idx + len("/method/")):]
 			if t, _ := url.QueryUnescape(targetID); t != "" {
 				targetID = t
 			}

pkg/http/api_test.go

@@ -65,6 +65,7 @@ import (
 	"github.com/dapr/dapr/pkg/messages"
 	invokev1 "github.com/dapr/dapr/pkg/messaging/v1"
 	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	commonv1 "github.com/dapr/dapr/pkg/proto/common/v1"
 	"github.com/dapr/dapr/pkg/resiliency"
 	"github.com/dapr/dapr/pkg/runtime/compstore"
 	runtimePubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
@@ -1091,7 +1092,16 @@ func TestV1DirectMessagingEndpoints(t *testing.T) {
 				mock.MatchedBy(func(b string) bool {
 					return b == "http://123.45.67.89:3000"
 				}),
-				mock.AnythingOfType("*v1.InvokeMethodRequest"),
+				mock.MatchedBy(func(req *invokev1.InvokeMethodRequest) bool {
+					msg := req.Message()
+					if msg.Method != "fakeMethod" {
+						return false
+					}
+					if msg.HttpExtension.Verb != commonv1.HTTPExtension_POST {
+						return false
+					}
+					return true
+				}),
 			).
 			Return(fakeDirectMessageResponse, nil).
 			Once()
@@ -1101,12 +1111,12 @@ func TestV1DirectMessagingEndpoints(t *testing.T) {
 
 		// assert
 		mockDirectMessaging.AssertNumberOfCalls(t, "Invoke", 1)
-		assert.Equal(t, 200, resp.StatusCode)
+		assert.Equal(t, gohttp.StatusOK, resp.StatusCode)
 		assert.Equal(t, "fakeDirectMessageResponse", string(resp.RawBody))
 	})
 
 	t.Run("Invoke direct messaging without querystring - 201 Created", func(t *testing.T) {
-		fakeDirectMessageResponse := getFakeDirectMessageResponseWithStatusCode(fasthttp.StatusCreated)
+		fakeDirectMessageResponse := getFakeDirectMessageResponseWithStatusCode(gohttp.StatusCreated)
 		defer fakeDirectMessageResponse.Close()
 
 		apiPath := "v1.0/invoke/fakeAppID/method/fakeMethod"
@@ -1364,7 +1374,7 @@ func TestV1DirectMessagingEndpoints(t *testing.T) {
 		// assert
 		mockDirectMessaging.AssertNumberOfCalls(t, "Invoke", 1)
 		assert.Equal(t, 500, resp.StatusCode)
-		assert.True(t, strings.HasPrefix(string(resp.RawBody), "{\"errorCode\":\"ERR_MALFORMED_RESPONSE\",\"message\":\""))
+		assert.Truef(t, strings.HasPrefix(string(resp.RawBody), "{\"errorCode\":\"ERR_MALFORMED_RESPONSE\",\"message\":\""), "code not found in response: %v", string(resp.RawBody))
 	})
 
 	t.Run("Invoke direct messaging with malformed status response for external invocation", func(t *testing.T) {
@@ -5610,33 +5620,27 @@ func TestFindTargetIDAndMethod(t *testing.T) {
 	tests := []struct {
 		name         string
 		path         string
-		headers      map[string]string
+		headers      gohttp.Header
 		wantTargetID string
 		wantMethod   string
 	}{
-		{name: "dapr-app-id header", path: "/foo/bar", headers: map[string]string{"dapr-app-id": "myapp"}, wantTargetID: "myapp", wantMethod: "foo/bar"},
-		{name: "basic auth", path: "/foo/bar", headers: map[string]string{"Authorization": "Basic ZGFwci1hcHAtaWQ6YXV0aA=="}, wantTargetID: "auth", wantMethod: "foo/bar"},
-		{name: "dapr-app-id header has priority over basic auth", path: "/foo/bar", headers: map[string]string{"dapr-app-id": "myapp", "Authorization": "Basic ZGFwci1hcHAtaWQ6YXV0aA=="}, wantTargetID: "myapp", wantMethod: "foo/bar"},
+		{name: "dapr-app-id header", path: "/foo/bar", headers: gohttp.Header{"Dapr-App-Id": []string{"myapp"}}, wantTargetID: "myapp", wantMethod: "foo/bar"},
+		{name: "basic auth", path: "/foo/bar", headers: gohttp.Header{"Authorization": []string{"Basic ZGFwci1hcHAtaWQ6YXV0aA=="}}, wantTargetID: "auth", wantMethod: "foo/bar"},
+		{name: "dapr-app-id header has priority over basic auth", path: "/foo/bar", headers: gohttp.Header{"Dapr-App-Id": []string{"myapp"}, "Authorization": []string{"Basic ZGFwci1hcHAtaWQ6YXV0aA=="}}, wantTargetID: "myapp", wantMethod: "foo/bar"},
 		{name: "path with internal target", path: "/v1.0/invoke/myapp/method/foo", wantTargetID: "myapp", wantMethod: "foo"},
-		{name: "basic auth has priority over path", path: "/v1.0/invoke/myapp/method/foo", headers: map[string]string{"Authorization": "Basic ZGFwci1hcHAtaWQ6YXV0aA=="}, wantTargetID: "auth", wantMethod: "v1.0/invoke/myapp/method/foo"},
+		{name: "basic auth has priority over path", path: "/v1.0/invoke/myapp/method/foo", headers: gohttp.Header{"Authorization": []string{"Basic ZGFwci1hcHAtaWQ6YXV0aA=="}}, wantTargetID: "auth", wantMethod: "v1.0/invoke/myapp/method/foo"},
 		{name: "path with '/' method", path: "/v1.0/invoke/myapp/method/", wantTargetID: "myapp", wantMethod: ""},
 		{name: "path with missing method", path: "/v1.0/invoke/myapp/method", wantTargetID: "", wantMethod: ""},
 		{name: "path with http target unescaped", path: "/v1.0/invoke/http://example.com/method/foo", wantTargetID: "http://example.com", wantMethod: "foo"},
 		{name: "path with https target unescaped", path: "/v1.0/invoke/https://example.com/method/foo", wantTargetID: "https://example.com", wantMethod: "foo"},
 		{name: "path with http target escaped", path: "/v1.0/invoke/http%3A%2F%2Fexample.com/method/foo", wantTargetID: "http://example.com", wantMethod: "foo"},
 		{name: "path with https target escaped", path: "/v1.0/invoke/https%3A%2F%2Fexample.com/method/foo", wantTargetID: "https://example.com", wantMethod: "foo"},
 		{name: "path with https target partly escaped", path: "/v1.0/invoke/https%3A/%2Fexample.com/method/foo", wantTargetID: "https://example.com", wantMethod: "foo"},
+		{name: "extra slashes are removed", path: "///foo//bar", headers: gohttp.Header{"Dapr-App-Id": []string{"myapp"}}, wantTargetID: "myapp", wantMethod: "foo/bar"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			peekHeader := func(k string) []byte {
-				if len(tt.headers) == 0 {
-					return nil
-				}
-				return []byte(tt.headers[k])
-			}
-
-			gotTargetID, gotMethod := findTargetIDAndMethod(tt.path, peekHeader)
+			gotTargetID, gotMethod := findTargetIDAndMethod(tt.path, tt.headers)
 			if gotTargetID != tt.wantTargetID {
 				t.Errorf("findTargetIDAndMethod() gotTargetID = %v, want %v", gotTargetID, tt.wantTargetID)
 			}

pkg/messages/predefined.go

@@ -72,11 +72,6 @@ const (
 	ErrActorStateGet             = "error getting actor state: %s"
 	ErrActorStateTransactionSave = "error saving actor transaction state: %s"
 
-	// DirectMessaging.
-	ErrDirectInvoke         = "fail to invoke, id: %s, err: %s"
-	ErrDirectInvokeNoAppID  = "failed getting app id either from the URL path or the header dapr-app-id"
-	ErrDirectInvokeNotReady = "invoke API is not ready"
-
 	// Configuration.
 	ErrConfigurationStoresNotConfigured = "configuration stores not configured"
 	ErrConfigurationStoreNotFound       = "configuration store %s not found"
@@ -94,6 +89,11 @@ var (
 	ErrBodyRead         = APIError{"failed to read request body: %v", "ERR_BODY_READ", http.StatusBadRequest, grpcCodes.InvalidArgument}
 	ErrMalformedRequest = APIError{"failed deserializing HTTP body: %v", "ERR_MALFORMED_REQUEST", http.StatusBadRequest, grpcCodes.InvalidArgument}
 
+	// DirectMessaging.
+	ErrDirectInvoke         = APIError{"fail to invoke, id: %s, err: %v", "ERR_DIRECT_INVOKE", http.StatusInternalServerError, grpcCodes.Internal}
+	ErrDirectInvokeNoAppID  = APIError{"failed getting app id either from the URL path or the header dapr-app-id", "ERR_DIRECT_INVOKE", http.StatusNotFound, grpcCodes.NotFound}
+	ErrDirectInvokeNotReady = APIError{"invoke API is not ready", "ERR_DIRECT_INVOKE", http.StatusInternalServerError, grpcCodes.Internal}
+
 	// Healthz.
 	ErrHealthNotReady         = APIError{"dapr is not ready", "ERR_HEALTH_NOT_READY", http.StatusInternalServerError, grpcCodes.Internal}
 	ErrOutboundHealthNotReady = APIError{"dapr outbound is not ready", "ERR_OUTBOUND_HEALTH_NOT_READY", http.StatusInternalServerError, grpcCodes.Internal}

pkg/messaging/v1/invoke_method_request_test.go

@@ -17,6 +17,7 @@ package v1
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -344,7 +345,7 @@ func TestRequestProto(t *testing.T) {
 		ir, err := InternalInvokeRequest(&pb)
 		assert.NoError(t, err)
 		defer ir.Close()
-		ir.data = io.NopCloser(strings.NewReader("test"))
+		ir.data = newReaderCloser(strings.NewReader("test"))
 		req2 := ir.Proto()
 
 		assert.Equal(t, "application/json", req2.GetMessage().ContentType)
@@ -391,7 +392,7 @@ func TestRequestProtoWithData(t *testing.T) {
 		ir, err := InternalInvokeRequest(&pb)
 		assert.NoError(t, err)
 		defer ir.Close()
-		ir.data = io.NopCloser(strings.NewReader("test"))
+		ir.data = newReaderCloser(strings.NewReader("test"))
 		req2, err := ir.ProtoWithData()
 		assert.NoError(t, err)
 
@@ -499,7 +500,7 @@ func TestRequestReplayable(t *testing.T) {
 	const message = "Nel mezzo del cammin di nostra vita mi ritrovai per una selva oscura, che' la diritta via era smarrita."
 	newReplayable := func() *InvokeMethodRequest {
 		return NewInvokeMethodRequest("test_method").
-			WithRawDataString(message).
+			WithRawData(newReaderCloser(strings.NewReader(message))).
 			WithReplay(true)
 	}
 
@@ -519,7 +520,7 @@ func TestRequestReplayable(t *testing.T) {
 			buf := make([]byte, 9)
 			n, err := io.ReadFull(req.data, buf)
 			assert.Equal(t, 0, n)
-			assert.ErrorIs(t, err, io.EOF)
+			assert.Truef(t, errors.Is(err, io.EOF) || errors.Is(err, http.ErrBodyReadAfterClose), "unexpected error: %v", err)
 		})
 
 		t.Run("replay buffer is full", func(t *testing.T) {
@@ -551,7 +552,7 @@ func TestRequestReplayable(t *testing.T) {
 			buf := make([]byte, 9)
 			n, err := io.ReadFull(req.data, buf)
 			assert.Equal(t, 0, n)
-			assert.ErrorIs(t, err, io.EOF)
+			assert.Truef(t, errors.Is(err, io.EOF) || errors.Is(err, http.ErrBodyReadAfterClose), "unexpected error: %v", err)
 		})
 
 		t.Run("replay buffer is full", func(t *testing.T) {
@@ -652,7 +653,7 @@ func TestRequestReplayable(t *testing.T) {
 			buf := make([]byte, 9)
 			n, err := io.ReadFull(req.data, buf)
 			assert.Equal(t, 0, n)
-			assert.ErrorIs(t, err, io.EOF)
+			assert.Truef(t, errors.Is(err, io.EOF) || errors.Is(err, http.ErrBodyReadAfterClose), "unexpected error: %v", err)
 		})
 
 		t.Run("replay buffer is full", func(t *testing.T) {

pkg/messaging/v1/replayable_request_test.go

@@ -16,7 +16,9 @@ package v1
 import (
 	"bytes"
 	"crypto/rand"
+	"errors"
 	"io"
+	"net/http"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -31,7 +33,7 @@ func TestReplayableRequest(t *testing.T) {
 
 	newReplayable := func() *replayableRequest {
 		rr := &replayableRequest{}
-		rr.WithRawData(bytes.NewReader(message))
+		rr.WithRawData(newReaderCloser(bytes.NewReader(message)))
 		rr.SetReplay(true)
 		return rr
 	}
@@ -50,7 +52,7 @@ func TestReplayableRequest(t *testing.T) {
 			buf := make([]byte, 9)
 			n, err := io.ReadFull(rr.data, buf)
 			assert.Equal(t, 0, n)
-			assert.ErrorIs(t, err, io.EOF)
+			assert.Truef(t, errors.Is(err, io.EOF) || errors.Is(err, http.ErrBodyReadAfterClose), "unexpected error: %v", err)
 		})
 
 		t.Run("replay buffer is full", func(t *testing.T) {
@@ -82,7 +84,7 @@ func TestReplayableRequest(t *testing.T) {
 			buf := make([]byte, 9)
 			n, err := io.ReadFull(rr.data, buf)
 			assert.Equal(t, 0, n)
-			assert.ErrorIs(t, err, io.EOF)
+			assert.Truef(t, errors.Is(err, io.EOF) || errors.Is(err, http.ErrBodyReadAfterClose), "unexpected error: %v", err)
 		})
 
 		t.Run("replay buffer is full", func(t *testing.T) {
@@ -191,7 +193,7 @@ func TestReplayableRequest(t *testing.T) {
 			buf := make([]byte, partial)
 			n, err := io.ReadFull(rr.data, buf)
 			assert.Equal(t, 0, n)
-			assert.ErrorIs(t, err, io.EOF)
+			assert.Truef(t, errors.Is(err, io.EOF) || errors.Is(err, http.ErrBodyReadAfterClose), "unexpected error: %v", err)
 		})
 
 		t.Run("replay buffer is full", func(t *testing.T) {
@@ -215,3 +217,29 @@ func TestReplayableRequest(t *testing.T) {
 		})
 	})
 }
+
+// readerCloser is a io.Reader that can be closed. Once the stream is closed, reading from it returns an error.
+type readerCloser struct {
+	r      io.Reader
+	closed bool
+}
+
+func newReaderCloser(r io.Reader) *readerCloser {
+	return &readerCloser{
+		r:      r,
+		closed: false,
+	}
+}
+
+func (b *readerCloser) Read(p []byte) (n int, err error) {
+	if b.closed {
+		// Use http.ErrBodyReadAfterClose which is the error returned by http.Response.Body
+		return 0, http.ErrBodyReadAfterClose
+	}
+	return b.r.Read(p)
+}
+
+func (b *readerCloser) Close() error {
+	b.closed = true
+	return nil
+}

tests/integration/suite/daprd/serviceinvocation/http/basic.go

@@ -130,10 +130,7 @@ func (b *basic) Run(t *testing.T, ctx context.Context) {
 			{url: fmt.Sprintf("http://localhost:%d/v1.0/invoke/%s/method/foo", b.daprd2.HTTPPort(), b.daprd1.AppID())},
 			{url: fmt.Sprintf("http://localhost:%d/v1.0////invoke/%s/method/foo", b.daprd2.HTTPPort(), b.daprd1.AppID())},
 			{url: fmt.Sprintf("http://localhost:%d/v1.0//invoke//%s/method//foo", b.daprd1.HTTPPort(), b.daprd2.AppID())},
-			// We cannot use `///foo` here because the test app uses the standard Go mux which responds with a 301 status code if the invocation is for `///foo`
-			// This makes Dapr retry with a GET request in all cases, as per specs
-			// See: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/301
-			{url: fmt.Sprintf("http://localhost:%d/foo", b.daprd1.HTTPPort()), headers: map[string]string{
+			{url: fmt.Sprintf("http://localhost:%d///foo", b.daprd1.HTTPPort()), headers: map[string]string{
 				"foo":         "bar",
 				"dapr-app-id": b.daprd2.AppID(),
 			}},

utils/streams/multireadercloser.go

@@ -13,7 +13,11 @@ limitations under the License.
 
 package streams
 
-import "io"
+import (
+	"errors"
+	"io"
+	"net/http"
+)
 
 // NewMultiReaderCloser returns a stream that is like io.MultiReader but that can be closed.
 // When the returned stream is closed, it closes the readable streams too, if they implement io.Closer.
@@ -41,7 +45,13 @@ func (mr *MultiReaderCloser) Read(p []byte) (n int, err error) {
 	for len(mr.readers) > 0 {
 		r := mr.readers[0]
 		n, err = r.Read(p)
-		if err == io.EOF {
+
+		// When reading from a http.Response Body, we may get ErrBodyReadAfterClose if we already read it all
+		// We consider that the same as io.EOF
+		if errors.Is(err, http.ErrBodyReadAfterClose) {
+			err = io.EOF
+			mr.readers = mr.readers[1:]
+		} else if err == io.EOF {
 			if rc, ok := r.(io.Closer); ok {
 				_ = rc.Close()
 			}