Fix Code Suggestions

event-gateway/gateway-runtime/go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/knadh/koanf/providers/env v1.1.0
 	github.com/knadh/koanf/providers/file v1.2.1
 	github.com/knadh/koanf/v2 v2.3.0
+	github.com/stretchr/testify v1.11.1
 	github.com/twmb/franz-go v1.18.1
 	github.com/twmb/franz-go/pkg/kadm v1.14.0
 	github.com/wso2/api-platform/common v0.0.0
@@ -26,6 +27,7 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.3.0 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
@@ -39,6 +41,7 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_golang v1.23.2 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect

event-gateway/gateway-runtime/internal/connectors/receiver/websub/delta.go

@@ -20,19 +20,28 @@ package websub
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 )
 
 // ApplyBindingDelta mutates the live receiver for channel add/remove changes
 // without recreating the receiver or the subscription sync topic.
 func (e *WebSubReceiver) ApplyBindingDelta(ctx context.Context, removedChannels map[string]string, addedChannels map[string]string) error {
+	var tombstoneErrs []error
+
 	for channelName := range removedChannels {
 		subscriptions := e.store.GetByTopic(channelName)
 		for _, sub := range subscriptions {
 			if e.syncProducer != nil {
 				if err := e.syncProducer.PublishTombstone(ctx, channelName, sub.CallbackURL); err != nil {
-					return fmt.Errorf("failed to tombstone subscription for removed channel %q: %w", channelName, err)
+					slog.Error("Failed to tombstone subscription for removed WebSub channel",
+						"api", e.channel.Name,
+						"channel", channelName,
+						"callback", sub.CallbackURL,
+						"error", err)
+					tombstoneErrs = append(tombstoneErrs,
+						fmt.Errorf("failed to tombstone subscription for removed channel %q callback %q: %w", channelName, sub.CallbackURL, err))
 				}
 			}
 		}
@@ -71,5 +80,9 @@ func (e *WebSubReceiver) ApplyBindingDelta(ctx context.Context, removedChannels
 		e.topics.Register(channelName)
 	}
 
+	if len(tombstoneErrs) > 0 {
+		return errors.Join(tombstoneErrs...)
+	}
+
 	return nil
 }

event-gateway/gateway-runtime/internal/connectors/receiver/websub/delta_test.go

@@ -0,0 +1,162 @@
+package websub
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/wso2/api-platform/event-gateway/gateway-runtime/internal/connectors"
+	"github.com/wso2/api-platform/event-gateway/gateway-runtime/internal/subscription"
+)
+
+type deltaTestBrokerDriver struct {
+	publishErrs map[string]error
+	published   []string
+}
+
+func (d *deltaTestBrokerDriver) Publish(_ context.Context, topic string, msg *connectors.Message) error {
+	key := string(msg.Key)
+	d.published = append(d.published, key)
+	if err, ok := d.publishErrs[key]; ok {
+		return err
+	}
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) Subscribe(string, []string, connectors.MessageHandler) (connectors.Receiver, error) {
+	return nil, nil
+}
+
+func (d *deltaTestBrokerDriver) SubscribeManual(string, []string, connectors.MessageHandler) (connectors.Receiver, error) {
+	return nil, nil
+}
+
+func (d *deltaTestBrokerDriver) Replay(context.Context, string, connectors.MessageHandler) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) TopicExists(context.Context, string) (bool, error) {
+	return false, nil
+}
+
+func (d *deltaTestBrokerDriver) EnsureTopics(context.Context, []string) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) EnsureCompactedTopic(context.Context, string) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) DeleteTopics(context.Context, []string) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) Close() error {
+	return nil
+}
+
+type deltaTestReceiver struct {
+	stopCalls int
+}
+
+func (r *deltaTestReceiver) Start(context.Context) error {
+	return nil
+}
+
+func (r *deltaTestReceiver) Stop(context.Context) error {
+	r.stopCalls++
+	return nil
+}
+
+func TestApplyBindingDeltaContinuesAfterTombstoneFailure(t *testing.T) {
+	store := subscription.NewInMemoryStore("runtime-1")
+	require.NoError(t, store.Add(&subscription.Subscription{
+		ID:          "sub-1",
+		Topic:       "removed-a",
+		CallbackURL: "https://callback-a.example",
+		State:       subscription.StateActive,
+	}))
+	require.NoError(t, store.Add(&subscription.Subscription{
+		ID:          "sub-2",
+		Topic:       "removed-b",
+		CallbackURL: "https://callback-b.example",
+		State:       subscription.StateActive,
+	}))
+
+	topics := NewTopicRegistry()
+	topics.Register("removed-a")
+	topics.Register("removed-b")
+
+	consumerA := &deltaTestReceiver{}
+	consumerB := &deltaTestReceiver{}
+	consumerMgr := &ConsumerManager{
+		consumers: map[string]*managedConsumer{
+			"https://callback-a.example": {
+				consumer: consumerA,
+				topics:   map[string]bool{"kafka-a": true},
+			},
+			"https://callback-b.example": {
+				consumer: consumerB,
+				topics:   map[string]bool{"kafka-b": true},
+			},
+		},
+	}
+
+	driver := &deltaTestBrokerDriver{
+		publishErrs: map[string]error{
+			"removed-a:https://callback-a.example": errors.New("boom"),
+		},
+	}
+
+	receiver := &WebSubReceiver{
+		topics:      topics,
+		store:       store,
+		consumerMgr: consumerMgr,
+		syncProducer: subscription.NewSyncProducer(
+			driver,
+			"runtime-1",
+			"internal-sub-topic",
+		),
+		channel: connectors.ChannelInfo{
+			Name: "test-api",
+			Channels: map[string]string{
+				"removed-a": "kafka-a",
+				"removed-b": "kafka-b",
+			},
+		},
+	}
+
+	err := receiver.ApplyBindingDelta(context.Background(),
+		map[string]string{
+			"removed-a": "kafka-a",
+			"removed-b": "kafka-b",
+		},
+		map[string]string{
+			"added-c": "kafka-c",
+		},
+	)
+
+	require.Error(t, err)
+	require.ErrorContains(t, err, `failed to tombstone subscription for removed channel "removed-a"`)
+
+	require.ElementsMatch(t, []string{
+		"removed-a:https://callback-a.example",
+		"removed-b:https://callback-b.example",
+	}, driver.published)
+
+	require.Nil(t, store.GetByTopic("removed-a"))
+	require.Nil(t, store.GetByTopic("removed-b"))
+
+	require.Equal(t, 1, consumerA.stopCalls)
+	require.Equal(t, 1, consumerB.stopCalls)
+	require.Empty(t, consumerMgr.consumers)
+
+	require.NotContains(t, receiver.channel.Channels, "removed-a")
+	require.NotContains(t, receiver.channel.Channels, "removed-b")
+	require.Equal(t, "kafka-c", receiver.channel.Channels["added-c"])
+
+	require.False(t, topics.IsRegistered("removed-a"))
+	require.False(t, topics.IsRegistered("removed-b"))
+	require.True(t, topics.IsRegistered("added-c"))
+}