Delivery failure

pkg/dispatcher/dispatcher.go

@@ -18,6 +18,7 @@ package dispatcher
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"sync"
 	"time"
@@ -101,7 +102,7 @@ func (d *Dispatcher) ConsumeFromQueue(ctx context.Context, conn rabbit.RabbitMQC
 		go func() {
 			defer wg.Done()
 			for msg := range workerQueue {
-				d.dispatch(ctx, msg, ceClient)
+				d.dispatch(ctx, msg, ceClient, channel)
 			}
 		}()
 	}
@@ -150,15 +151,22 @@ func getStatus(ctx context.Context, result protocol.Result) (int, bool) {
 	return -1, false
 }
 
-func (d *Dispatcher) dispatch(ctx context.Context, msg amqp.Delivery, ceClient cloudevents.Client) {
+func (d *Dispatcher) dispatch(ctx context.Context, msg amqp.Delivery, ceClient cloudevents.Client, channel rabbit.RabbitMQChannelInterface) {
 	start := time.Now()
+	if _, ok := msg.Headers["knativeerrordest"]; ok {
+		if err := msg.Ack(false); err != nil {
+			logging.FromContext(ctx).Warn("failed to Nack event: ", err)
+		}
+		return
+	}
+
 	msgBinding := rabbit.NewMessageFromDelivery(ComponentName, "", "", &msg)
 	event, err := binding.ToEvent(cloudevents.WithEncodingBinary(ctx), msgBinding)
 	if err != nil {
-		logging.FromContext(ctx).Warn("failed creating event from delivery, err (NACK-ing and not re-queueing): ", err)
-		err = msg.Nack(false, false)
-		if err != nil {
-			logging.FromContext(ctx).Warn("failed to NACK event: ", err)
+		logging.FromContext(ctx).Warn("failed parsing event, setting knativeerrordest header and NACK-ing and re-queueing: ", err)
+
+		if err = msg.Nack(false, false); err != nil {
+			logging.FromContext(ctx).Warn("failed to Nack event: ", err)
 		}
 		return
 	}
@@ -180,35 +188,56 @@ func (d *Dispatcher) dispatch(ctx context.Context, msg amqp.Delivery, ceClient c
 	statusCode, isSuccess := getStatus(ctx, result)
 	if statusCode != -1 {
 		args := &dispatcher.ReportArgs{EventType: event.Type()}
-		if err := d.Reporter.ReportEventCount(args, statusCode); err != nil {
+		if err = d.Reporter.ReportEventCount(args, statusCode); err != nil {
 			logging.FromContext(ctx).Errorf("Something happened: %v", err)
 		}
 	}
 
 	if !isSuccess {
 		logging.FromContext(ctx).Warnf("Failed to deliver to %q", d.SubscriberURL)
-		if err := msg.Nack(false, false); err != nil {
-			logging.FromContext(ctx).Warn("failed to NACK event: ", err)
+
+		// We need to ack the original message.
+		if err = msg.Ack(false); err != nil {
+			logging.FromContext(ctx).Warn("failed to Ack event: ", err)
+		}
+
+		// Add headers as described here: https://knative.dev/docs/eventing/event-delivery/#configuring-channel-event-delivery
+		event.SetExtension("knativeerrordest", d.SubscriberURL)
+		event.SetExtension("knativeerrorcode", statusCode)
+
+		// Queue the event into DLQ with the correct headers.
+		if err = sendToRabbitMQ(channel, msg.Exchange, span, event); err != nil {
+			logging.FromContext(ctx).Warn("failed to send event: ", err)
 		}
 		return
 	} else if response != nil {
 		logging.FromContext(ctx).Infof("Sending an event: %+v", response)
 		ctx = cloudevents.ContextWithTarget(ctx, d.BrokerIngressURL)
-		result := ceClient.Send(ctx, *response)
-		_, isSuccess := getStatus(ctx, result)
+		result = ceClient.Send(ctx, *response)
+		_, isSuccess = getStatus(ctx, result)
 		if !isSuccess {
 			logging.FromContext(ctx).Warnf("Failed to deliver to %q", d.BrokerIngressURL)
-			err = msg.Nack(false, false) // not multiple
-			if err != nil {
-				logging.FromContext(ctx).Warn("failed to NACK event: ", err)
+
+			// We need to ack the original message.
+			if err = msg.Ack(false); err != nil {
+				logging.FromContext(ctx).Warn("failed to Ack event: ", err)
+			}
+
+			// Add headers as described here: https://knative.dev/docs/eventing/event-delivery/#configuring-channel-event-delivery
+			event.SetExtension("knativeerrordest", d.SubscriberURL)
+			event.SetExtension("knativeerrorcode", statusCode)
+			event.SetExtension("knativeerrordata", result)
+
+			// Queue the event into DLQ with the correct headers.
+			if err = sendToRabbitMQ(channel, msg.Exchange, span, event); err != nil {
+				logging.FromContext(ctx).Warn("failed to send event: ", err)
 			}
 			return
 		}
 	}
 
-	err = msg.Ack(false)
-	if err != nil {
-		logging.FromContext(ctx).Warn("failed to ACK event: ", err)
+	if err = msg.Ack(false); err != nil {
+		logging.FromContext(ctx).Warn("failed to Ack event: ", err)
 	}
 	if statusCode != -1 {
 		args := &dispatcher.ReportArgs{EventType: event.Type()}
@@ -217,6 +246,24 @@ func (d *Dispatcher) dispatch(ctx context.Context, msg amqp.Delivery, ceClient c
 	}
 }
 
+func sendToRabbitMQ(channel rabbit.RabbitMQChannelInterface, exchangeName string, span *trace.Span, event *cloudevents.Event) error {
+	tp, ts := (&tracecontext.HTTPFormat{}).SpanContextToHeaders(span.SpanContext())
+	dc, err := channel.PublishWithDeferredConfirm(
+		exchangeName,
+		"",    // routing key
+		false, // mandatory
+		false, // immediate
+		*rabbit.CloudEventToRabbitMQMessage(event, tp, ts))
+	if err != nil {
+		return fmt.Errorf("failed to publish message: %w", err)
+	}
+
+	if ack := dc.Wait(); !ack {
+		return errors.New("failed to publish message: nacked")
+	}
+	return nil
+}
+
 func readSpan(ctx context.Context, msg amqp.Delivery) (context.Context, *trace.Span) {
 	traceparent, ok := msg.Headers["traceparent"].(string)
 	if !ok {