Fix ack for non-fastack, non-reply case

Signed-off-by: Peter Broadhurst <[email protected]>

Author: peterbroadhurst

internal/events/webhooks/webhooks.go

@@ -271,7 +271,7 @@ func (wh *WebHooks) attemptRequest(sub *core.Subscription, event *core.EventDeli
 	return req, res, nil
 }
 
-func (wh *WebHooks) doDelivery(connID string, reply bool, sub *core.Subscription, event *core.EventDelivery, data core.DataArray) error {
+func (wh *WebHooks) doDelivery(connID string, reply bool, sub *core.Subscription, event *core.EventDelivery, data core.DataArray, fastAck bool) error {
 	req, res, gwErr := wh.attemptRequest(sub, event, data)
 	if gwErr != nil {
 		// Generate a bad-gateway error response - we always want to send something back,
@@ -298,7 +298,7 @@ func (wh *WebHooks) doDelivery(connID string, reply bool, sub *core.Subscription
 			txType = fftypes.FFEnum(strings.ToLower(req.replyTx))
 		}
 		if cb, ok := wh.callbacks[sub.Namespace]; ok {
-			log.L(wh.ctx).Tracef("Sending reply message in response to webhook message %s", event.Message.Header.ID)
+			log.L(wh.ctx).Debugf("Sending reply message for %s CID=%s", event.ID, event.Message.Header.ID)
 			cb.DeliveryResponse(connID, &core.EventDeliveryResponse{
 				ID:           event.ID,
 				Rejected:     false,
@@ -320,6 +320,14 @@ func (wh *WebHooks) doDelivery(connID string, reply bool, sub *core.Subscription
 				},
 			})
 		}
+	} else if !fastAck {
+		if cb, ok := wh.callbacks[sub.Namespace]; ok {
+			cb.DeliveryResponse(connID, &core.EventDeliveryResponse{
+				ID:           event.ID,
+				Rejected:     false,
+				Subscription: event.Subscription,
+			})
+		}
 	}
 	return nil
 }
@@ -347,13 +355,22 @@ func (wh *WebHooks) DeliveryRequest(connID string, sub *core.Subscription, event
 	}
 
 	// In fastack mode we drive calls in parallel to the backend, immediately acknowledging the event
-	if sub.Options.TransportOptions().GetBool("fastack") {
+	// NOTE: We cannot use this with reply mode, as when we're sending a reply the `DeliveryResponse`
+	//       callback must include the reply in-line.
+	if !reply && sub.Options.TransportOptions().GetBool("fastack") {
+		if cb, ok := wh.callbacks[sub.Namespace]; ok {
+			cb.DeliveryResponse(connID, &core.EventDeliveryResponse{
+				ID:           event.ID,
+				Rejected:     false,
+				Subscription: event.Subscription,
+			})
+		}
 		go func() {
-			err := wh.doDelivery(connID, reply, sub, event, data)
+			err := wh.doDelivery(connID, reply, sub, event, data, true)
 			log.L(wh.ctx).Warnf("Webhook delivery failed in fastack mode for event '%s': %s", event.ID, err)
 		}()
 		return nil
 	}
 
-	return wh.doDelivery(connID, reply, sub, event, data)
+	return wh.doDelivery(connID, reply, sub, event, data, false)
 }

internal/events/webhooks/webhooks_test.go

@@ -360,7 +360,11 @@ func TestRequestNoBodyNoReply(t *testing.T) {
 
 	dataID := fftypes.NewUUID()
 	groupHash := fftypes.NewRandB32()
-	sub := &core.Subscription{}
+	sub := &core.Subscription{
+		SubscriptionRef: core.SubscriptionRef{
+			Namespace: "ns1",
+		},
+	}
 	to := sub.Options.TransportOptions()
 	to["url"] = fmt.Sprintf("http://%s/myapi", server.Listener.Addr())
 	event := &core.EventDelivery{
@@ -380,7 +384,8 @@ func TestRequestNoBodyNoReply(t *testing.T) {
 			},
 		},
 		Subscription: core.SubscriptionRef{
-			ID: sub.ID,
+			ID:        sub.ID,
+			Namespace: "ns1",
 		},
 	}
 	data := &core.Data{
@@ -390,9 +395,16 @@ func TestRequestNoBodyNoReply(t *testing.T) {
 		}`),
 	}
 
+	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
+	mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
+		return !response.Rejected
+	})).Return(nil)
+
 	err := wh.DeliveryRequest(mock.Anything, sub, event, core.DataArray{data})
 	assert.NoError(t, err)
 	assert.True(t, called)
+
+	mcb.AssertExpectations(t)
 }
 
 func TestRequestReplyEmptyData(t *testing.T) {
@@ -638,7 +650,7 @@ func TestRequestReplyDataArrayError(t *testing.T) {
 	mcb.AssertExpectations(t)
 }
 
-func TestRequestReplyBuildRequestFailFastAsk(t *testing.T) {
+func TestWebhookFailFastAsk(t *testing.T) {
 	wh, cancel := newTestWebHooks(t)
 	defer cancel()
 
@@ -652,7 +664,6 @@ func TestRequestReplyBuildRequestFailFastAsk(t *testing.T) {
 			Namespace: "ns1",
 		},
 	}
-	sub.Options.TransportOptions()["reply"] = true
 	sub.Options.TransportOptions()["fastack"] = true
 	event := &core.EventDelivery{
 		EnrichedEvent: core.EnrichedEvent{
@@ -673,17 +684,11 @@ func TestRequestReplyBuildRequestFailFastAsk(t *testing.T) {
 
 	waiter := make(chan struct{})
 	mcb := wh.callbacks["ns1"].(*eventsmocks.Callbacks)
-	dr := mcb.On("DeliveryResponse", mock.Anything, mock.MatchedBy(func(response *core.EventDeliveryResponse) bool {
-		assert.Equal(t, *msgID, *response.Reply.Message.Header.CID)
-		assert.Nil(t, response.Reply.Message.Header.Group)
-		assert.Equal(t, core.MessageTypeBroadcast, response.Reply.Message.Header.Type)
-		assert.Equal(t, float64(502), response.Reply.InlineData[0].Value.JSONObject()["status"])
-		assert.Regexp(t, "FF10242", response.Reply.InlineData[0].Value.JSONObject().GetObject("body")["error"])
-		return true
-	})).Return(nil)
-	dr.RunFn = func(a mock.Arguments) {
-		close(waiter)
-	}
+	mcb.On("DeliveryResponse", mock.Anything, mock.Anything).
+		Return(nil).
+		Run(func(a mock.Arguments) {
+			close(waiter)
+		})
 
 	err := wh.DeliveryRequest(mock.Anything, sub, event, core.DataArray{
 		{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"value1"`)},