diff --git a/router-tests/events/trigger_test.go b/router-tests/events/trigger_test.go new file mode 100644 index 0000000000..21df4cc290 --- /dev/null +++ b/router-tests/events/trigger_test.go @@ -0,0 +1,277 @@ +package events_test + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router-tests/testenv" +) + +// TestEDFSTriggerDeduplication verifies trigger ID generation for Cosmo Streams subscriptions. +// It tests that the trigger hash is based on the NATS subject and provider ID, not on +// the resolver input — meaning subscriptions to the same subject share one trigger +// even when they select different fields, have different headers or arguments. +func TestEDFSTriggerDeduplication(t *testing.T) { + t.Parallel() + + // Two subscriptions with the same employeeID but different selected fields should share + // a single NATS trigger because both resolve to the same subject ("employeeUpdated.3"). + t.Run("same subject different selected fields shares one trigger", func(t *testing.T) { + t.Parallel() + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, + EnableNats: true, + }, func(t *testing.T, xEnv *testenv.Environment) { + var done sync.WaitGroup + done.Add(2) + + go func() { + xEnv.WaitForSubscriptionCount(2, time.Second*10) + // Both subscriptions target the same NATS subject ("employeeUpdated.3") and + // provider ID, so they must resolve to the same trigger ID — wait for exactly + // one trigger to be initialized before asserting. + xEnv.WaitForTriggerCount(1, time.Second*10) + xEnv.RequireTriggerCount(1) + xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 2, time.Second*10) + }() + + // Subscription 1: selects only id. + go func() { + defer done.Done() + conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) + defer conn.Close() + + err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ + ID: "1", + Type: "subscribe", + Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), + }) + require.NoError(t, err) + + var msg testenv.WebSocketMessage + err = testenv.WSReadJSON(t, conn, &msg) + require.NoError(t, err) + require.Equal(t, "next", msg.Type) + require.Equal(t, "1", msg.ID) + + err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) + require.NoError(t, err) + + var complete testenv.WebSocketMessage + err = conn.SetReadDeadline(time.Now().Add(time.Second)) + require.NoError(t, err) + err = testenv.WSReadJSON(t, conn, &complete) + require.NoError(t, err) + require.Equal(t, "complete", complete.Type) + require.Equal(t, "1", complete.ID) + }() + + // Subscription 2: selects id and details — a different query shape over the same subject. + go func() { + defer done.Done() + conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) + defer conn.Close() + + err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ + ID: "1", + Type: "subscribe", + Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id details { forename surname } } }"}`), + }) + require.NoError(t, err) + + var msg testenv.WebSocketMessage + err = testenv.WSReadJSON(t, conn, &msg) + require.NoError(t, err) + require.Equal(t, "next", msg.Type) + require.Equal(t, "1", msg.ID) + + err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) + require.NoError(t, err) + + var complete testenv.WebSocketMessage + err = conn.SetReadDeadline(time.Now().Add(time.Second)) + require.NoError(t, err) + err = testenv.WSReadJSON(t, conn, &complete) + require.NoError(t, err) + require.Equal(t, "complete", complete.Type) + require.Equal(t, "1", complete.ID) + }() + + done.Wait() + xEnv.WaitForSubscriptionCount(0, time.Second*5) + }) + }) + + // Two subscriptions with the same query but different initial_payload (headers) should + // still share a single NATS trigger because the trigger ID is based on the NATS subject, + // not on connection-level metadata like headers. + t.Run("same subject different initial payload shares one trigger", func(t *testing.T) { + t.Parallel() + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, + EnableNats: true, + }, func(t *testing.T, xEnv *testenv.Environment) { + var done sync.WaitGroup + done.Add(2) + + go func() { + xEnv.WaitForSubscriptionCount(2, time.Second*10) + xEnv.WaitForTriggerCount(1, time.Second*10) + xEnv.RequireTriggerCount(1) + xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 2, time.Second*10) + }() + + // Subscription 1: sends Authorization header A in connection_init. + go func() { + defer done.Done() + conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, []byte(`{"headers":{"Authorization":"Bearer token-a"}}`)) + defer conn.Close() + + err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ + ID: "1", + Type: "subscribe", + Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), + }) + require.NoError(t, err) + + var msg testenv.WebSocketMessage + err = testenv.WSReadJSON(t, conn, &msg) + require.NoError(t, err) + require.Equal(t, "next", msg.Type) + require.Equal(t, "1", msg.ID) + + err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) + require.NoError(t, err) + + var complete testenv.WebSocketMessage + err = conn.SetReadDeadline(time.Now().Add(time.Second)) + require.NoError(t, err) + err = testenv.WSReadJSON(t, conn, &complete) + require.NoError(t, err) + require.Equal(t, "complete", complete.Type) + require.Equal(t, "1", complete.ID) + }() + + // Subscription 2: sends a different Authorization header in connection_init. + go func() { + defer done.Done() + conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, []byte(`{"headers":{"Authorization":"Bearer token-b"}}`)) + defer conn.Close() + + err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ + ID: "1", + Type: "subscribe", + Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), + }) + require.NoError(t, err) + + var msg testenv.WebSocketMessage + err = testenv.WSReadJSON(t, conn, &msg) + require.NoError(t, err) + require.Equal(t, "next", msg.Type) + require.Equal(t, "1", msg.ID) + + err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) + require.NoError(t, err) + + var complete testenv.WebSocketMessage + err = conn.SetReadDeadline(time.Now().Add(time.Second)) + require.NoError(t, err) + err = testenv.WSReadJSON(t, conn, &complete) + require.NoError(t, err) + require.Equal(t, "complete", complete.Type) + require.Equal(t, "1", complete.ID) + }() + + done.Wait() + xEnv.WaitForSubscriptionCount(0, time.Second*5) + }) + }) + + // Two subscriptions with different employeeIDs resolve to different NATS subjects and + // must therefore receive independent triggers. + t.Run("different subjects use separate triggers", func(t *testing.T) { + t.Parallel() + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, + EnableNats: true, + }, func(t *testing.T, xEnv *testenv.Environment) { + var done sync.WaitGroup + done.Add(2) + + go func() { + xEnv.WaitForSubscriptionCount(2, time.Second*10) + xEnv.WaitForTriggerCount(2, time.Second*10) + xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 1, time.Second*10) + xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.4"), []byte(`{"id":4,"__typename":"Employee"}`), 1, time.Second*10) + }() + + // Subscription 1: employeeID 3 → subject "employeeUpdated.3" + go func() { + defer done.Done() + conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) + defer conn.Close() + + err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ + ID: "1", + Type: "subscribe", + Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 3) { id } }"}`), + }) + require.NoError(t, err) + + var msg testenv.WebSocketMessage + err = testenv.WSReadJSON(t, conn, &msg) + require.NoError(t, err) + require.Equal(t, "next", msg.Type) + require.Equal(t, "1", msg.ID) + + err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) + require.NoError(t, err) + + var complete testenv.WebSocketMessage + err = conn.SetReadDeadline(time.Now().Add(time.Second)) + require.NoError(t, err) + err = testenv.WSReadJSON(t, conn, &complete) + require.NoError(t, err) + require.Equal(t, "complete", complete.Type) + require.Equal(t, "1", complete.ID) + }() + + // Subscription 2: employeeID 4 → subject "employeeUpdated.4" + go func() { + defer done.Done() + conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, nil) + defer conn.Close() + + err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ + ID: "1", + Type: "subscribe", + Payload: []byte(`{"query":"subscription { employeeUpdated(employeeID: 4) { id } }"}`), + }) + require.NoError(t, err) + + var msg testenv.WebSocketMessage + err = testenv.WSReadJSON(t, conn, &msg) + require.NoError(t, err) + require.Equal(t, "next", msg.Type) + require.Equal(t, "1", msg.ID) + + err = testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{ID: "1", Type: "complete"}) + require.NoError(t, err) + + var complete testenv.WebSocketMessage + err = conn.SetReadDeadline(time.Now().Add(time.Second)) + require.NoError(t, err) + err = testenv.WSReadJSON(t, conn, &complete) + require.NoError(t, err) + require.Equal(t, "complete", complete.Type) + require.Equal(t, "1", complete.ID) + }() + + done.Wait() + xEnv.WaitForSubscriptionCount(0, time.Second*5) + }) + }) +} diff --git a/router-tests/go.mod b/router-tests/go.mod index b6f0ee9511..24b9a70aeb 100644 --- a/router-tests/go.mod +++ b/router-tests/go.mod @@ -31,7 +31,7 @@ require ( github.com/wundergraph/cosmo/router v0.0.0-20260330183556-dc4388d100a4 github.com/wundergraph/cosmo/router-plugin v0.0.0-20250808194725-de123ba1c65e github.com/wundergraph/cosmo/speedtrap v0.0.0-00010101000000-000000000000 - github.com/wundergraph/graphql-go-tools/v2 v2.4.5 + github.com/wundergraph/graphql-go-tools/v2 v2.4.6 go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/sdk/metric v1.44.0 diff --git a/router-tests/go.sum b/router-tests/go.sum index cf0d6f2ae0..bb8aeb0de4 100644 --- a/router-tests/go.sum +++ b/router-tests/go.sum @@ -381,8 +381,8 @@ github.com/wundergraph/astjson v1.1.0 h1:xORDosrZ87zQFJwNGe/HIHXqzpdHOFmqWgykCLV github.com/wundergraph/astjson v1.1.0/go.mod h1:h12D/dxxnedtLzsKyBLK7/Oe4TAoGpRVC9nDpDrZSWw= github.com/wundergraph/go-arena v1.3.0 h1:n0ng5a1vbd8YGq1u3rMr0vPU5f6AZ1BXIiUhL1UIok8= github.com/wundergraph/go-arena v1.3.0/go.mod h1:ROOysEHWJjLQ8FSfNxZCziagb7Qw2nXY3/vgKRh7eWw= -github.com/wundergraph/graphql-go-tools/v2 v2.4.5 h1:hJsNYupt3i2zWD9X00ytcEkgU6Cr+//21oIKMDbydmE= -github.com/wundergraph/graphql-go-tools/v2 v2.4.5/go.mod h1:rGG9m74sUyucfvSZ83Mjuq/6qRJetl1CVP872f/dCok= +github.com/wundergraph/graphql-go-tools/v2 v2.4.6 h1:MqUDusuiOGlO0Oa8etTdcVA2ySmY0XehTDq/dILelEw= +github.com/wundergraph/graphql-go-tools/v2 v2.4.6/go.mod h1:rGG9m74sUyucfvSZ83Mjuq/6qRJetl1CVP872f/dCok= github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 h1:FnBeRrxr7OU4VvAzt5X7s6266i6cSVkkFPS0TuXWbIg= github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= diff --git a/router-tests/testenv/testenv.go b/router-tests/testenv/testenv.go index 338c4e79c1..94fd92fb95 100644 --- a/router-tests/testenv/testenv.go +++ b/router-tests/testenv/testenv.go @@ -2919,6 +2919,15 @@ func (e *Environment) WaitForTriggerCount(desiredCount uint64, timeout time.Dura } } +// RequireTriggerCount asserts that the current trigger count equals desiredCount exactly. +// Call this after WaitForSubscriptionCount has confirmed all subscriptions are active; +// trigger creation is synchronous with subscription registration so no additional waiting is needed. +func (e *Environment) RequireTriggerCount(desiredCount uint64) { + e.t.Helper() + report := e.syncReporter().GetReport() + require.Equal(e.t, desiredCount, report.Triggers, "expected exactly %d triggers, got %d", desiredCount, report.Triggers) +} + // NATSPublishUntilMinMessagesSent publishes a NATS message repeatedly until the // total MessagesSent count reaches minCount. This handles fan-out scenarios where // multiple subscriptions must all receive the message: if some consumers aren't diff --git a/router/go.mod b/router/go.mod index 936a84f2f0..9e8bba5f64 100644 --- a/router/go.mod +++ b/router/go.mod @@ -31,7 +31,7 @@ require ( github.com/tidwall/gjson v1.18.0 github.com/tidwall/sjson v1.2.5 github.com/twmb/franz-go v1.16.1 - github.com/wundergraph/graphql-go-tools/v2 v2.4.5 + github.com/wundergraph/graphql-go-tools/v2 v2.4.6 // Do not upgrade, it renames attributes we rely on go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 go.opentelemetry.io/contrib/propagators/b3 v1.44.0 diff --git a/router/go.sum b/router/go.sum index 84a549e2d5..9a28c67815 100644 --- a/router/go.sum +++ b/router/go.sum @@ -333,8 +333,8 @@ github.com/wundergraph/astjson v1.1.0 h1:xORDosrZ87zQFJwNGe/HIHXqzpdHOFmqWgykCLV github.com/wundergraph/astjson v1.1.0/go.mod h1:h12D/dxxnedtLzsKyBLK7/Oe4TAoGpRVC9nDpDrZSWw= github.com/wundergraph/go-arena v1.3.0 h1:n0ng5a1vbd8YGq1u3rMr0vPU5f6AZ1BXIiUhL1UIok8= github.com/wundergraph/go-arena v1.3.0/go.mod h1:ROOysEHWJjLQ8FSfNxZCziagb7Qw2nXY3/vgKRh7eWw= -github.com/wundergraph/graphql-go-tools/v2 v2.4.5 h1:hJsNYupt3i2zWD9X00ytcEkgU6Cr+//21oIKMDbydmE= -github.com/wundergraph/graphql-go-tools/v2 v2.4.5/go.mod h1:rGG9m74sUyucfvSZ83Mjuq/6qRJetl1CVP872f/dCok= +github.com/wundergraph/graphql-go-tools/v2 v2.4.6 h1:MqUDusuiOGlO0Oa8etTdcVA2ySmY0XehTDq/dILelEw= +github.com/wundergraph/graphql-go-tools/v2 v2.4.6/go.mod h1:rGG9m74sUyucfvSZ83Mjuq/6qRJetl1CVP872f/dCok= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= diff --git a/router/pkg/pubsub/datasource/datasource.go b/router/pkg/pubsub/datasource/datasource.go index 8c100c45f7..aa1600f056 100644 --- a/router/pkg/pubsub/datasource/datasource.go +++ b/router/pkg/pubsub/datasource/datasource.go @@ -3,12 +3,14 @@ package datasource import ( "net/http" + "github.com/cespare/xxhash/v2" "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve" ) type SubscriptionDataSource interface { SubscriptionEventConfiguration(input []byte) (SubscriptionEventConfiguration, error) Start(ctx *resolve.Context, header http.Header, input []byte, updater resolve.SubscriptionUpdater) error + HashTriggerInput(input []byte, xxh *xxhash.Digest) error SetHooks(hooks Hooks) } diff --git a/router/pkg/pubsub/datasource/subscription_datasource.go b/router/pkg/pubsub/datasource/subscription_datasource.go index 939c03f94d..6147b345db 100644 --- a/router/pkg/pubsub/datasource/subscription_datasource.go +++ b/router/pkg/pubsub/datasource/subscription_datasource.go @@ -12,18 +12,18 @@ import ( "go.uber.org/zap/zapcore" ) -type uniqueRequestIdFn func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error +type triggerHashInputFn func(input []byte, xxh *xxhash.Digest) error type EventBuilderFn func(data []byte) MutableStreamEvent // PubSubSubscriptionDataSource is a data source for handling subscriptions using a Pub/Sub mechanism. // It implements the SubscriptionDataSource interface and HookableSubscriptionDataSource type PubSubSubscriptionDataSource[C SubscriptionEventConfiguration] struct { - pubSub Adapter - uniqueRequestID uniqueRequestIdFn - hooks Hooks - logger *zap.Logger - eventBuilder EventBuilderFn + pubSub Adapter + triggerHashInput triggerHashInputFn + hooks Hooks + logger *zap.Logger + eventBuilder EventBuilderFn } func (s *PubSubSubscriptionDataSource[C]) SubscriptionEventConfiguration(input []byte) (SubscriptionEventConfiguration, error) { @@ -88,17 +88,21 @@ func (s *PubSubSubscriptionDataSource[C]) SetHooks(hooks Hooks) { s.hooks = hooks } +func (s *PubSubSubscriptionDataSource[C]) HashTriggerInput(input []byte, xxh *xxhash.Digest) error { + return s.triggerHashInput(input, xxh) +} + var _ SubscriptionDataSource = (*PubSubSubscriptionDataSource[SubscriptionEventConfiguration])(nil) var _ resolve.HookableSubscriptionDataSource = (*PubSubSubscriptionDataSource[SubscriptionEventConfiguration])(nil) -func NewPubSubSubscriptionDataSource[C SubscriptionEventConfiguration](pubSub Adapter, uniqueRequestIdFn uniqueRequestIdFn, logger *zap.Logger, eventBuilder EventBuilderFn) *PubSubSubscriptionDataSource[C] { +func NewPubSubSubscriptionDataSource[C SubscriptionEventConfiguration](pubSub Adapter, triggerHashInputFn triggerHashInputFn, logger *zap.Logger, eventBuilder EventBuilderFn) *PubSubSubscriptionDataSource[C] { if logger == nil { logger = zap.NewNop() } return &PubSubSubscriptionDataSource[C]{ - pubSub: pubSub, - uniqueRequestID: uniqueRequestIdFn, - logger: logger, - eventBuilder: eventBuilder, + pubSub: pubSub, + triggerHashInput: triggerHashInputFn, + logger: logger, + eventBuilder: eventBuilder, } } diff --git a/router/pkg/pubsub/datasource/subscription_datasource_test.go b/router/pkg/pubsub/datasource/subscription_datasource_test.go index 2662a2a2ca..923c9265df 100644 --- a/router/pkg/pubsub/datasource/subscription_datasource_test.go +++ b/router/pkg/pubsub/datasource/subscription_datasource_test.go @@ -39,7 +39,7 @@ func testSubscriptionDataSourceEventBuilder(data []byte) MutableStreamEvent { func TestPubSubSubscriptionDataSource_SubscriptionEventConfiguration_Success(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -64,7 +64,7 @@ func TestPubSubSubscriptionDataSource_SubscriptionEventConfiguration_Success(t * func TestPubSubSubscriptionDataSource_SubscriptionEventConfiguration_InvalidJSON(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -78,7 +78,7 @@ func TestPubSubSubscriptionDataSource_SubscriptionEventConfiguration_InvalidJSON func TestPubSubSubscriptionDataSource_Start_Success(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -103,7 +103,7 @@ func TestPubSubSubscriptionDataSource_Start_Success(t *testing.T) { func TestPubSubSubscriptionDataSource_Start_NoConfiguration(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -120,7 +120,7 @@ func TestPubSubSubscriptionDataSource_Start_NoConfiguration(t *testing.T) { func TestPubSubSubscriptionDataSource_Start_SubscribeError(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -147,7 +147,7 @@ func TestPubSubSubscriptionDataSource_Start_SubscribeError(t *testing.T) { func TestPubSubSubscriptionDataSource_SubscriptionOnStart_Success(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -171,7 +171,7 @@ func TestPubSubSubscriptionDataSource_SubscriptionOnStart_Success(t *testing.T) func TestPubSubSubscriptionDataSource_SubscriptionOnStart_WithHooks(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -227,7 +227,7 @@ func TestPubSubSubscriptionDataSource_SubscriptionOnStart_WithHooks(t *testing.T func TestPubSubSubscriptionDataSource_SubscriptionOnStart_HookReturnsClose(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -262,7 +262,7 @@ func TestPubSubSubscriptionDataSource_SubscriptionOnStart_HookReturnsClose(t *te func TestPubSubSubscriptionDataSource_SubscriptionOnStart_HookReturnsError(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -299,7 +299,7 @@ func TestPubSubSubscriptionDataSource_SubscriptionOnStart_HookReturnsError(t *te func TestPubSubSubscriptionDataSource_SetSubscriptionOnStartFns(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -333,7 +333,7 @@ func TestPubSubSubscriptionDataSource_SetSubscriptionOnStartFns(t *testing.T) { func TestNewPubSubSubscriptionDataSource(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -341,13 +341,13 @@ func TestNewPubSubSubscriptionDataSource(t *testing.T) { assert.NotNil(t, dataSource) assert.Equal(t, mockAdapter, dataSource.pubSub) - assert.NotNil(t, dataSource.uniqueRequestID) + assert.NotNil(t, dataSource.triggerHashInput) assert.Empty(t, dataSource.hooks.SubscriptionOnStart.Handlers) } func TestPubSubSubscriptionDataSource_InterfaceCompliance(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -362,7 +362,7 @@ func TestPubSubSubscriptionDataSource_InterfaceCompliance(t *testing.T) { func TestPubSubSubscriptionDataSource_SubscriptionOnStart_InvalidEventConfigInput(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } @@ -423,7 +423,7 @@ func TestPubSubSubscriptionDataSource_SubscriptionOnStart_PanicRecovery(t *testi for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { mockAdapter := NewMockProvider(t) - uniqueRequestIDFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + uniqueRequestIDFn := func(input []byte, xxh *xxhash.Digest) error { return nil } diff --git a/router/pkg/pubsub/kafka/engine_datasource_factory.go b/router/pkg/pubsub/kafka/engine_datasource_factory.go index 5beb74870e..dbae2855d2 100644 --- a/router/pkg/pubsub/kafka/engine_datasource_factory.go +++ b/router/pkg/pubsub/kafka/engine_datasource_factory.go @@ -64,7 +64,7 @@ func (c *EngineDataSourceFactory) ResolveDataSourceInput(eventData []byte) (stri } func (c *EngineDataSourceFactory) ResolveDataSourceSubscription() (datasource.SubscriptionDataSource, error) { - uniqueRequestIdFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + triggerHashInputFn := func(input []byte, xxh *xxhash.Digest) error { val, _, _, err := jsonparser.Get(input, "topics") if err != nil { return err @@ -89,7 +89,7 @@ func (c *EngineDataSourceFactory) ResolveDataSourceSubscription() (datasource.Su } return datasource.NewPubSubSubscriptionDataSource[*SubscriptionEventConfiguration]( - c.KafkaAdapter, uniqueRequestIdFn, c.logger, eventCreateFn, + c.KafkaAdapter, triggerHashInputFn, c.logger, eventCreateFn, ), nil } diff --git a/router/pkg/pubsub/nats/adapter.go b/router/pkg/pubsub/nats/adapter.go index 7ac8cbbc1d..6e62c225d6 100644 --- a/router/pkg/pubsub/nats/adapter.go +++ b/router/pkg/pubsub/nats/adapter.go @@ -176,6 +176,13 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, cfg datasource.Subscrip subscriptions[i] = subscription } + // Flush ensures the SUB commands are delivered to the NATS server before returning, + // so that publishers can immediately target these subjects without missing messages. + if err := p.client.Flush(); err != nil { + log.Error("flushing NATS connection after subscribe", zap.Error(err)) + return datasource.NewError("failed to flush NATS connection", err) + } + p.closeWg.Add(1) go func() { diff --git a/router/pkg/pubsub/nats/engine_datasource_factory.go b/router/pkg/pubsub/nats/engine_datasource_factory.go index c478148e81..d1ca384541 100644 --- a/router/pkg/pubsub/nats/engine_datasource_factory.go +++ b/router/pkg/pubsub/nats/engine_datasource_factory.go @@ -76,7 +76,7 @@ func (c *EngineDataSourceFactory) ResolveDataSourceInput(eventData []byte) (stri } func (c *EngineDataSourceFactory) ResolveDataSourceSubscription() (datasource.SubscriptionDataSource, error) { - uniqueRequestIdFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + triggerHashInputFn := func(input []byte, xxh *xxhash.Digest) error { val, _, _, err := jsonparser.Get(input, "subjects") if err != nil { return err @@ -102,7 +102,7 @@ func (c *EngineDataSourceFactory) ResolveDataSourceSubscription() (datasource.Su return datasource.NewPubSubSubscriptionDataSource[*SubscriptionEventConfiguration]( c.NatsAdapter, - uniqueRequestIdFn, c.logger, createEventFn), nil + triggerHashInputFn, c.logger, createEventFn), nil } func (c *EngineDataSourceFactory) ResolveDataSourceSubscriptionInput() (string, error) { diff --git a/router/pkg/pubsub/redis/engine_datasource_factory.go b/router/pkg/pubsub/redis/engine_datasource_factory.go index ac01322fbc..3a84be5c90 100644 --- a/router/pkg/pubsub/redis/engine_datasource_factory.go +++ b/router/pkg/pubsub/redis/engine_datasource_factory.go @@ -74,7 +74,7 @@ func (c *EngineDataSourceFactory) ResolveDataSourceInput(eventData []byte) (stri // ResolveDataSourceSubscription returns the subscription data source func (c *EngineDataSourceFactory) ResolveDataSourceSubscription() (datasource.SubscriptionDataSource, error) { - uniqueRequestIdFn := func(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error { + triggerHashInputFn := func(input []byte, xxh *xxhash.Digest) error { val, _, _, err := jsonparser.Get(input, "channels") if err != nil { return err @@ -99,7 +99,7 @@ func (c *EngineDataSourceFactory) ResolveDataSourceSubscription() (datasource.Su } return datasource.NewPubSubSubscriptionDataSource[*SubscriptionEventConfiguration]( - c.RedisAdapter, uniqueRequestIdFn, c.logger, eventCreateFn, + c.RedisAdapter, triggerHashInputFn, c.logger, eventCreateFn, ), nil }