fix(webhooks): hold subscription checkpoint on non-2xx delivery

internal/coremsgs/en_error_messages.go

@@ -130,6 +130,7 @@ var (
 	MsgWebsocketsNoData                        = ffe("FF10244", "Websockets subscriptions do not support streaming the full data payload, just the references (withData must be false)", 400)
 	MsgWebhooksWithData                        = ffe("FF10245", "Webhook subscriptions require the full data payload (withData must be true)", 400)
 	MsgWebhooksReplyBadJSON                    = ffe("FF10257", "Failed to process reply from webhook as JSON")
+	MsgWebhooksDeliveryFailedStatus            = ffe("FF10486", "Webhook delivery was not successful (status=%d) - holding subscription checkpoint for redelivery", 502)
 	MsgRequestTimeout                          = ffe("FF10260", "The request with id '%s' timed out after %.2fms", 408)
 	MsgRequestReplyTagRequired                 = ffe("FF10261", "For request messages 'header.tag' must be set on the request message to route it to a suitable responder", 400)
 	MsgRequestCannotHaveCID                    = ffe("FF10262", "For request messages 'header.cid' must be unset", 400)

internal/events/webhooks/webhooks.go

@@ -437,7 +437,7 @@ func (wh *WebHooks) attemptRequest(ctx context.Context, sub *core.Subscription,
 	return req, res, nil
 }
 
-func (wh *WebHooks) doDelivery(ctx context.Context, connID string, reply bool, sub *core.Subscription, events []*core.CombinedEventDataDelivery, fastAck, batched bool) {
+func (wh *WebHooks) doDelivery(ctx context.Context, connID string, reply bool, sub *core.Subscription, events []*core.CombinedEventDataDelivery, fastAck, batched bool) error {
 	req, res, gwErr := wh.attemptRequest(ctx, sub, events, batched)
 	if gwErr != nil {
 		// Generate a bad-gateway error response - we always want to send something back,
@@ -457,6 +457,31 @@ func (wh *WebHooks) doDelivery(ctx context.Context, connID string, reply bool, s
 	b, _ := json.Marshal(&res)
 	log.L(ctx).Tracef("Webhook response: %s", string(b))
 
+	// Capture the event sequence range (checkpoint offsets) so the delivery outcome logs below
+	// show exactly which offsets are advancing or being held. Events are dispatched in order.
+	var firstOffset, lastOffset int64
+	if len(events) > 0 {
+		firstOffset = events[0].Event.Sequence
+		lastOffset = events[len(events)-1].Event.Sequence
+	}
+
+	// A delivery is only considered successful on a 2xx response. Any non-2xx (including the
+	// 502 synthesised above for a network-level failure) must NOT advance the subscription
+	// checkpoint - we return an error so the event dispatcher holds the offset and redelivers.
+	// NOTE: this does not apply to reply mode, where the webhook response (whatever its status)
+	// is intentionally packaged back to the caller as a reply message.
+	if !reply && (res.Status < 200 || res.Status >= 300) {
+		if fastAck {
+			// In fastack mode the event(s) were already acknowledged before this call, so the
+			// checkpoint cannot be held. Surface it loudly rather than silently advancing.
+			log.L(ctx).Warnf("Webhook delivery returned status=%d for %d event(s) (offsets %d-%d), but fastack already acknowledged them - checkpoint advanced past a failed delivery", res.Status, len(events), firstOffset, lastOffset)
+			return nil
+		}
+		log.L(ctx).Infof("Webhook delivery returned status=%d - holding subscription checkpoint below offset %d and redelivering %d event(s) (offsets %d-%d)", res.Status, firstOffset, len(events), firstOffset, lastOffset)
+		return i18n.NewError(ctx, coremsgs.MsgWebhooksDeliveryFailedStatus, res.Status)
+	}
+	log.L(ctx).Infof("Webhook delivery successful (status=%d) for %d event(s) - advancing subscription checkpoint to offset %d", res.Status, len(events), lastOffset)
+
 	// For each event emit a response
 	for _, combinedEvent := range events {
 		event := combinedEvent.Event
@@ -500,6 +525,7 @@ func (wh *WebHooks) doDelivery(ctx context.Context, connID string, reply bool, s
 		}
 	}
 
+	return nil
 }
 
 func (wh *WebHooks) DeliveryRequest(ctx context.Context, connID string, sub *core.Subscription, event *core.EventDelivery, data core.DataArray) error {
@@ -531,15 +557,17 @@ func (wh *WebHooks) DeliveryRequest(ctx context.Context, connID string, sub *cor
 				Subscription: event.Subscription,
 			})
 		}
-		go wh.doDelivery(ctx, connID, reply, sub, []*core.CombinedEventDataDelivery{{Event: event, Data: data}}, true, false)
+		go func() {
+			_ = wh.doDelivery(ctx, connID, reply, sub, []*core.CombinedEventDataDelivery{{Event: event, Data: data}}, true, false)
+		}()
 		return nil
 	}
 
 	// NOTE: We could check here for batching and accumulate but we can't return because this causes the offset to jump...
 
-	// TODO we don't look at the error here?
-	wh.doDelivery(ctx, connID, reply, sub, []*core.CombinedEventDataDelivery{{Event: event, Data: data}}, false, false)
-	return nil
+	// A non-nil error here causes the dispatcher to reject (nack) the event, holding the
+	// subscription checkpoint and redelivering, rather than advancing past a failed delivery.
+	return wh.doDelivery(ctx, connID, reply, sub, []*core.CombinedEventDataDelivery{{Event: event, Data: data}}, false, false)
 }
 
 func (wh *WebHooks) BatchDeliveryRequest(ctx context.Context, connID string, sub *core.Subscription, events []*core.CombinedEventDataDelivery) error {
@@ -584,12 +612,13 @@ func (wh *WebHooks) BatchDeliveryRequest(ctx context.Context, connID string, sub
 				})
 			}
 		}
-		go wh.doDelivery(ctx, connID, reply, sub, events, true, true)
+		go func() { _ = wh.doDelivery(ctx, connID, reply, sub, events, true, true) }()
 		return nil
 	}
 
-	wh.doDelivery(ctx, connID, reply, sub, events, false, true)
-	return nil
+	// A non-nil error here causes the dispatcher to reject (nack) the whole batch, holding the
+	// subscription checkpoint and redelivering, rather than advancing past a failed delivery.
+	return wh.doDelivery(ctx, connID, reply, sub, events, false, true)
 }
 
 func (wh *WebHooks) NamespaceRestarted(ns string, startTime time.Time) {

internal/events/webhooks/webhooks_test.go

@@ -783,6 +783,64 @@ func TestRequestNoBodyNoReply(t *testing.T) {
 	mcb.AssertExpectations(t)
 }
 
+func TestDeliveryRequestNon2xxHoldsCheckpoint(t *testing.T) {
+	wh, cancel := newTestWebHooks(t)
+	defer cancel()
+
+	msgID := fftypes.NewUUID()
+
+	called := false
+	r := mux.NewRouter()
+	r.HandleFunc("/myapi", func(res http.ResponseWriter, req *http.Request) {
+		res.WriteHeader(500)
+		_, _ = res.Write([]byte(`{"error":"boom"}`))
+		called = true
+	}).Methods(http.MethodPost)
+	server := httptest.NewServer(r)
+	defer server.Close()
+
+	dataID := fftypes.NewUUID()
+	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{
+		EnrichedEvent: core.EnrichedEvent{
+			Event: core.Event{
+				ID:       fftypes.NewUUID(),
+				Sequence: 12345,
+			},
+			Message: &core.Message{
+				Header: core.MessageHeader{
+					ID:   msgID,
+					Type: core.MessageTypeBroadcast,
+				},
+			},
+		},
+		Subscription: core.SubscriptionRef{
+			ID:        sub.ID,
+			Namespace: "ns1",
+		},
+	}
+	data := &core.Data{
+		ID:    dataID,
+		Value: fftypes.JSONAnyPtr(`{"foo":"bar"}`),
+	}
+
+	mcb := wh.callbacks.handlers["ns1"].(*eventsmocks.Callbacks)
+
+	// A non-2xx delivery (non-reply) must NOT acknowledge - the checkpoint is held by returning
+	// an error, so the dispatcher rejects/redelivers rather than advancing past the failure.
+	err := wh.DeliveryRequest(wh.ctx, mock.Anything, sub, event, core.DataArray{data})
+	assert.Regexp(t, "FF10486", err)
+	assert.True(t, called)
+
+	mcb.AssertNotCalled(t, "DeliveryResponse", mock.Anything, mock.Anything)
+}
+
 func TestRequestReplyEmptyData(t *testing.T) {
 	wh, cancel := newTestWebHooks(t)
 	defer cancel()