diff --git a/pkg/requestreply/ingress_handler.go b/pkg/requestreply/ingress_handler.go index 5439c3e2dc5..ca763a64b61 100644 --- a/pkg/requestreply/ingress_handler.go +++ b/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) } diff --git a/pkg/requestreply/ingress_handler_test.go b/pkg/requestreply/ingress_handler_test.go index 5e2553b7853..8536a85f5c1 100644 --- a/pkg/requestreply/ingress_handler_test.go +++ b/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