Fix RequestReply duplicate reply deadlock

Author: rootp1

pkg/requestreply/ingress_handler.go

@@ -232,6 +232,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 +297,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 +341,19 @@ 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))
+		responseWriter.WriteHeader(http.StatusAccepted)
 		return
 	}
 
-	// send the reply event back to the original response writer
-	pr.replyEvent <- event
+	select {
+	case pr.replyEvent <- event:
+		responseWriter.WriteHeader(http.StatusAccepted)
+	default:
+		h.logger.Warn("reply event already delivered or duplicate reply received, discarding event", zap.String("reply id", id))
+		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,63 @@ func TestHandlerServeHttp(t *testing.T) {
 	}
 }
 
+func TestHandleReplyEventIgnoresDuplicateReplyWithoutBlocking(t *testing.T) {
+	t.Parallel()
+
+	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 := &IngressHandler{
+		logger:   zap.NewNop(),
+		keyStore: keyStore,
+		entries: map[types.NamespacedName]map[string]*proxiedRequest{
+			rr.GetNamespacedName(): {
+				inflight.ID(): {
+					responseWriter: httptest.NewRecorder(),
+					replyEvent:     make(chan *cloudevents.Event, 1),
+				},
+			},
+		},
+	}
+
+	// Fill the reply channel once to simulate an already delivered reply.
+	pr, ok := handler.getEvent(inflight.ID(), rr)
+	if !ok {
+		t.Fatal("expected inflight request to exist")
+	}
+	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