Fix RequestReply duplicate reply deadlock

Signed-off-by: rootp1 <arnav.iitr@gmail.com>

Author: rootp1

pkg/requestreply/ingress_handler.go

@@ -215,7 +215,9 @@ func (h *IngressHandler) addEvent(responseWriter http.ResponseWriter, event *clo
 	pr := &proxiedRequest{
 		received:       time.Now(),
 		responseWriter: responseWriter,
-		replyEvent:     make(chan *cloudevents.Event, 1),
+		// capacity must stay 1: handleReplyEvent relies on a full buffer to
+		// detect and drop duplicate replies without blocking.
+		replyEvent: make(chan *cloudevents.Event, 1),
 	}
 	if h.entries[rr.GetNamespacedName()] == nil {
 		h.entries[rr.GetNamespacedName()] = make(map[string]*proxiedRequest)
@@ -232,6 +234,14 @@ func (h *IngressHandler) deleteEvent(event *cloudevents.Event, rr *v1alpha1.Requ
 	delete(h.entries[rr.GetNamespacedName()], event.ID())
 }
 
+func (h *IngressHandler) getEvent(id string, rr *v1alpha1.RequestReply) (*proxiedRequest, bool) {
+	h.requestLock.RLock()
+	defer h.requestLock.RUnlock()
+
+	pr, ok := h.entries[rr.GetNamespacedName()][id]
+	return pr, ok
+}
+
 func (h *IngressHandler) handleNewEvent(ctx context.Context, responseWriter http.ResponseWriter, event *cloudevents.Event, rr *v1alpha1.RequestReply, headers http.Header) {
 	h.logger.Debug("handling new event")
 	pr := h.addEvent(responseWriter, event, rr)
@@ -289,9 +299,6 @@ func (h *IngressHandler) handleNewEvent(ctx context.Context, responseWriter http
 }
 
 func (h *IngressHandler) handleReplyEvent(responseWriter http.ResponseWriter, event *cloudevents.Event, rr *v1alpha1.RequestReply) {
-	h.requestLock.RLock()
-	defer h.requestLock.RUnlock()
-
 	h.logger.Debug("handling a response event")
 
 	// TODO: with OIDC enabled, we can skip validation of the key if we validate the identity of the trigger making the request
@@ -336,15 +343,17 @@ func (h *IngressHandler) handleReplyEvent(responseWriter http.ResponseWriter, ev
 		return
 	}
 
-	responseWriter.WriteHeader(http.StatusAccepted)
-
 	id := strings.Split(replyIdString, ":")[0]
-	pr, ok := h.entries[rr.GetNamespacedName()][id]
+	pr, ok := h.getEvent(id, rr)
 	if !ok {
 		h.logger.Warn("no event found matching the reply id, discarding event", zap.String("reply id", id))
-		return
+	} else {
+		select {
+		case pr.replyEvent <- event:
+		default:
+			h.logger.Warn("reply event already delivered or duplicate reply received, discarding event", zap.String("reply id", id))
+		}
 	}
 
-	// send the reply event back to the original response writer
-	pr.replyEvent <- event
+	responseWriter.WriteHeader(http.StatusAccepted)
 }

pkg/requestreply/ingress_handler_test.go

@@ -24,6 +24,7 @@ import (
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
@@ -210,6 +211,51 @@ func TestHandlerServeHttp(t *testing.T) {
 	}
 }
 
+func TestHandleReplyEventIgnoresDuplicateReplyWithoutBlocking(t *testing.T) {
+	t.Parallel()
+
+	ctx, _ := reconcilertesting.SetupFakeContext(t, setupInformerSelector)
+
+	rr := makeRequestReply("my-request-reply", "default")
+	inflight := cloudevents.NewEvent()
+	inflight.SetID("1234567890")
+
+	firstReply := inflight.Clone()
+	if err := SetCorrelationId(&firstReply, "replyid", exampleKey, 0); err != nil {
+		t.Fatalf("failed to create first reply id: %v", err)
+	}
+
+	duplicateReply := inflight.Clone()
+	if err := SetCorrelationId(&duplicateReply, "replyid", exampleKey, 0); err != nil {
+		t.Fatalf("failed to create duplicate reply id: %v", err)
+	}
+
+	keyStore := &AESKeyStore{}
+	keyStore.addAesKey(rr.GetNamespacedName(), "key", exampleKey)
+
+	handler := NewHandler(zap.NewNop(), requestreplyinformerfake.Get(ctx), configmapinformerfake.Get(ctx).Lister().ConfigMaps("ns"), keyStore, 0)
+
+	// Fill the reply channel once to simulate an already delivered reply.
+	pr := handler.addEvent(httptest.NewRecorder(), &inflight, rr)
+	pr.replyEvent <- &firstReply
+
+	recorder := httptest.NewRecorder()
+	done := make(chan struct{})
+
+	go func() {
+		handler.handleReplyEvent(recorder, &duplicateReply, rr)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleReplyEvent blocked on duplicate reply")
+	}
+
+	assert.Equal(t, http.StatusAccepted, recorder.Result().StatusCode)
+}
+
 type testServerHandler struct {
 	makeReplyEvent  func(e *cloudevents.Event) *cloudevents.Event
 	callbackHandler http.Handler