diff --git a/internal/coremsgs/en_error_messages.go b/internal/coremsgs/en_error_messages.go index 237025599..96350b1dd 100644 --- a/internal/coremsgs/en_error_messages.go +++ b/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) diff --git a/internal/events/webhooks/webhooks.go b/internal/events/webhooks/webhooks.go index 781083cea..53cb2ea0f 100644 --- a/internal/events/webhooks/webhooks.go +++ b/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) { diff --git a/internal/events/webhooks/webhooks_test.go b/internal/events/webhooks/webhooks_test.go index d2d931265..ade956915 100644 --- a/internal/events/webhooks/webhooks_test.go +++ b/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()