From cf4fb6631163891dbc0a85e66504e024dd4a6e68 Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Sat, 20 Jun 2026 19:24:59 +0200 Subject: [PATCH 1/7] feat: allow router to start even if some events providers are not defined --- docs-website/router/configuration.mdx | 7 + docs-website/router/cosmo-streams.mdx | 15 ++ router/pkg/config/config.go | 10 +- router/pkg/config/config.schema.json | 5 + router/pkg/config/config_test.go | 50 +++++ router/pkg/config/fixtures/full.yaml | 1 + .../pkg/config/testdata/config_defaults.json | 1 + router/pkg/config/testdata/config_full.json | 1 + router/pkg/pubsub/pubsub.go | 43 +++- router/pkg/pubsub/pubsub_test.go | 207 +++++++++++++++++- 10 files changed, 326 insertions(+), 14 deletions(-) diff --git a/docs-website/router/configuration.mdx b/docs-website/router/configuration.mdx index 176228aa19..346a7e3266 100644 --- a/docs-website/router/configuration.mdx +++ b/docs-website/router/configuration.mdx @@ -1814,6 +1814,7 @@ We support NATS, Kafka and Redis as event bus provider. version: "1" events: + skip_missing_providers: false providers: nats: - id: default @@ -1843,6 +1844,12 @@ events: - "localhost:9092" ``` +### Options + +| Environment Variable | YAML | Required | Description | Default Value | +| ----------------------------- | ---------------------- | ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------- | +| EVENTS_SKIP_MISSING_PROVIDERS | skip_missing_providers | | Start the router even when the execution config references an event provider that is not defined under `providers`. The router logs an error and disables only the affected fields instead of failing to start. By default a missing provider prevents startup. | false | + ### Provider | Environment Variable | YAML | Required | Description | Default Value | diff --git a/docs-website/router/cosmo-streams.mdx b/docs-website/router/cosmo-streams.mdx index 034bce8056..926c19c388 100644 --- a/docs-website/router/cosmo-streams.mdx +++ b/docs-website/router/cosmo-streams.mdx @@ -186,6 +186,21 @@ We've intentionally moved this part of the configuration out of the Schema to ke If you run `make edfs-demo` in the Cosmo Monorepo, you'll automatically get a NATS instance running on the default port (4222) using Docker. +### Missing providers + +If the execution config references a provider ID that is not defined under `events.providers`, the router fails to start with an error such as `redis provider with ID my-provider is not defined`. This stops a router from coming up with a partially-configured event setup. + +Set `skip_missing_providers: true` (or the `EVENTS_SKIP_MISSING_PROVIDERS` environment variable) to start the router anyway. The router logs the error and disables only the fields backed by the missing provider. The rest of the graph keeps serving traffic, and requests to the affected fields fail at query time. Set up an alert on the logged error to detect the missing configuration. + +```yaml config.yaml +events: + skip_missing_providers: true + providers: + nats: + - id: default + url: "nats://localhost:4222" +``` + ## Example Configuration Below, you'll find an example Schema that use the NATS provider directives to connect the `@edfs__natsRequest` directive to a Query root field (employeeFromEvent), a Mutation root field (updateEmployee) that's connected to another topic using `@edfs__natsPublish` and a Subscription root field (employeeUpdated) that's connected via `@edfs__natsSubscribe`. Each of these fields is completely independent. Important to notice is that you can't implement this subgraph because the engine will implement the resolvers based on the router configuration. diff --git a/router/pkg/config/config.go b/router/pkg/config/config.go index 61e053dd63..6c066096ab 100644 --- a/router/pkg/config/config.go +++ b/router/pkg/config/config.go @@ -800,8 +800,14 @@ type EventProviders struct { } type EventsConfiguration struct { - Providers EventProviders `yaml:"providers,omitempty"` - Handlers StreamsHandlerConfiguration `yaml:"handlers,omitempty"` + Providers EventProviders `yaml:"providers,omitempty"` + // SkipMissingProviders allows the router to start even when the execution config + // references an event provider that is not defined in the router configuration. + // When enabled, the router logs an error and skips the data sources that depend on + // the missing provider instead of failing to start. Only the affected fields become + // unavailable; the rest of the graph keeps serving traffic. + SkipMissingProviders bool `yaml:"skip_missing_providers" envDefault:"false" env:"EVENTS_SKIP_MISSING_PROVIDERS"` + Handlers StreamsHandlerConfiguration `yaml:"handlers,omitempty"` } type StreamsHandlerConfiguration struct { diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json index 264006fd6c..3929c60da3 100644 --- a/router/pkg/config/config.schema.json +++ b/router/pkg/config/config.schema.json @@ -2891,6 +2891,11 @@ "description": "The configuration for EDFS. See https://cosmo-docs.wundergraph.com/router/event-driven-federated-subscriptions-edfs for more information.", "additionalProperties": false, "properties": { + "skip_missing_providers": { + "type": "boolean", + "default": false, + "description": "Skip the data sources that reference an event provider not defined in this configuration, instead of aborting startup. When enabled, the router logs an error and starts anyway, leaving only the affected fields unavailable. Use this when a missing provider should not prevent the router from serving the rest of the graph. Defaults to false, in which case a missing provider prevents the router from starting." + }, "providers": { "type": "object", "description": "The provider configuration. The provider configuration is used to configure the event-driven federated subscriptions.", diff --git a/router/pkg/config/config_test.go b/router/pkg/config/config_test.go index 00cb69de18..e0ca9d62e8 100644 --- a/router/pkg/config/config_test.go +++ b/router/pkg/config/config_test.go @@ -177,6 +177,56 @@ engine: }) } +func TestEventsSkipMissingProvidersConfigLoading(t *testing.T) { + t.Run("defaults to false", func(t *testing.T) { + t.Parallel() + + f := createTempFileFromFixture(t, ` +version: "1" + +graph: + token: "token" +`) + + cfg, err := LoadConfig([]string{f}) + require.NoError(t, err) + require.False(t, cfg.Config.Events.SkipMissingProviders) + }) + + t.Run("can be enabled from yaml", func(t *testing.T) { + t.Parallel() + + f := createTempFileFromFixture(t, ` +version: "1" + +graph: + token: "token" + +events: + skip_missing_providers: true +`) + + cfg, err := LoadConfig([]string{f}) + require.NoError(t, err) + require.True(t, cfg.Config.Events.SkipMissingProviders) + }) + + t.Run("can be enabled from env", func(t *testing.T) { + t.Setenv("EVENTS_SKIP_MISSING_PROVIDERS", "true") + + f := createTempFileFromFixture(t, ` +version: "1" + +graph: + token: "token" +`) + + cfg, err := LoadConfig([]string{f}) + require.NoError(t, err) + require.True(t, cfg.Config.Events.SkipMissingProviders) + }) +} + // Confirms https://github.com/caarlos0/env/issues/354 is fixed func TestConfigSlicesHaveDefaults(t *testing.T) { t.Parallel() diff --git a/router/pkg/config/fixtures/full.yaml b/router/pkg/config/fixtures/full.yaml index 68db894fc7..06976b88fb 100644 --- a/router/pkg/config/fixtures/full.yaml +++ b/router/pkg/config/fixtures/full.yaml @@ -358,6 +358,7 @@ cdn: cache_size: 100MB events: + skip_missing_providers: true providers: nats: - id: default diff --git a/router/pkg/config/testdata/config_defaults.json b/router/pkg/config/testdata/config_defaults.json index b251de5410..6ed523cb17 100644 --- a/router/pkg/config/testdata/config_defaults.json +++ b/router/pkg/config/testdata/config_defaults.json @@ -373,6 +373,7 @@ "Kafka": null, "Redis": null }, + "SkipMissingProviders": false, "Handlers": { "OnReceiveEvents": { "MaxConcurrentHandlers": 100, diff --git a/router/pkg/config/testdata/config_full.json b/router/pkg/config/testdata/config_full.json index 26a5835849..6759f9b85d 100644 --- a/router/pkg/config/testdata/config_full.json +++ b/router/pkg/config/testdata/config_full.json @@ -783,6 +783,7 @@ } ] }, + "SkipMissingProviders": true, "Handlers": { "OnReceiveEvents": { "MaxConcurrentHandlers": 100, diff --git a/router/pkg/pubsub/pubsub.go b/router/pkg/pubsub/pubsub.go index 19c908712e..7360ff030e 100644 --- a/router/pkg/pubsub/pubsub.go +++ b/router/pkg/pubsub/pubsub.go @@ -66,6 +66,13 @@ func BuildProvidersAndDataSources( if store == nil { store = metric.NewNoopStreamMetricStore() } + if logger == nil { + logger = zap.NewNop() + } + + if config.SkipMissingProviders { + logger.Warn("EDFS lenient mode is enabled (events.skip_missing_providers=true): the router will start even if an event provider referenced by the execution config is not defined, disabling only the affected fields") + } var pubSubProviders []pubsub_datasource.Provider var outs []plan.DataSource @@ -79,7 +86,7 @@ func BuildProvidersAndDataSources( events: dsConf.Configuration.GetCustomEvents().GetKafka(), }) } - kafkaPubSubProviders, kafkaOuts, err := build(ctx, kafkaBuilder, config.Providers.Kafka, kafkaDsConfsWithEvents, store, hooks) + kafkaPubSubProviders, kafkaOuts, err := build(ctx, kafkaBuilder, config.Providers.Kafka, kafkaDsConfsWithEvents, store, hooks, logger, config.SkipMissingProviders) if err != nil { return nil, nil, err } @@ -97,7 +104,7 @@ func BuildProvidersAndDataSources( events: dsConf.Configuration.GetCustomEvents().GetNats(), }) } - natsPubSubProviders, natsOuts, err := build(ctx, natsBuilder, config.Providers.Nats, natsDsConfsWithEvents, store, hooks) + natsPubSubProviders, natsOuts, err := build(ctx, natsBuilder, config.Providers.Nats, natsDsConfsWithEvents, store, hooks, logger, config.SkipMissingProviders) if err != nil { return nil, nil, err } @@ -115,7 +122,7 @@ func BuildProvidersAndDataSources( events: dsConf.Configuration.GetCustomEvents().GetRedis(), }) } - redisPubSubProviders, redisOuts, err := build(ctx, redisBuilder, config.Providers.Redis, redisDsConfsWithEvents, store, hooks) + redisPubSubProviders, redisOuts, err := build(ctx, redisBuilder, config.Providers.Redis, redisDsConfsWithEvents, store, hooks, logger, config.SkipMissingProviders) if err != nil { return nil, nil, err } @@ -133,6 +140,8 @@ func build[P GetID, E GetEngineEventConfiguration]( providersData []P, dsConfs []dsConfAndEvents[E], store metric.StreamMetricStore, hooks pubsub_datasource.Hooks, + logger *zap.Logger, + skipMissingProviders bool, ) (map[string]pubsub_datasource.Provider, []plan.DataSource, error) { pubSubProviders := make(map[string]pubsub_datasource.Provider) var outs []plan.DataSource @@ -163,18 +172,36 @@ func build[P GetID, E GetEngineEventConfiguration]( } // check if all used providers are initialized + missingProviderIds := make(map[string]struct{}) for _, providerId := range usedProviderIds { - if _, ok := pubSubProviders[providerId]; !ok { - return pubSubProviders, nil, &ProviderNotDefinedError{ - ProviderID: providerId, - ProviderTypeID: builder.TypeID(), - } + if _, ok := pubSubProviders[providerId]; ok { + continue + } + err := &ProviderNotDefinedError{ + ProviderID: providerId, + ProviderTypeID: builder.TypeID(), } + if !skipMissingProviders { + return pubSubProviders, nil, err + } + // Lenient mode: do not prevent the router from starting. Log the error so it + // surfaces in alerting, record the provider as missing, and skip the data + // sources that depend on it. Only the affected fields become unavailable. + logger.Error("Event provider referenced by the execution config is not defined; skipping affected data sources, the corresponding fields will be unavailable", + zap.String("provider_id", providerId), + zap.String("provider_type", builder.TypeID()), + ) + missingProviderIds[providerId] = struct{}{} } // build data sources for each event for _, dsConf := range dsConfs { for i, event := range dsConf.events { + // Skip events that reference a provider which could not be initialized + // (only possible when skipMissingProviders is enabled). + if _, ok := missingProviderIds[event.GetEngineEventConfiguration().GetProviderId()]; ok { + continue + } plannerConfig := pubsub_datasource.NewPlannerConfig( builder, event, diff --git a/router/pkg/pubsub/pubsub_test.go b/router/pkg/pubsub/pubsub_test.go index e89c94e885..4b90833790 100644 --- a/router/pkg/pubsub/pubsub_test.go +++ b/router/pkg/pubsub/pubsub_test.go @@ -72,7 +72,7 @@ func TestBuild_OK(t *testing.T) { // ctx, kafkaBuilder, config.Providers.Kafka, kafkaDsConfsWithEvents // Execute the function - providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}) + providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}, zap.NewNop(), false) // Assertions assert.NoError(t, err) @@ -128,7 +128,7 @@ func TestBuild_ProviderError(t *testing.T) { mockBuilder.On("BuildProvider", natsEventSources[0], mock.Anything).Return(nil, errors.New("provider error")) // Execute the function - providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}) + providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}, zap.NewNop(), false) // Assertions assert.Error(t, err) @@ -183,7 +183,7 @@ func TestBuild_ShouldGetAnErrorIfProviderIsNotDefined(t *testing.T) { mockBuilder.On("TypeID").Return("nats") // Execute the function - providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}) + providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}, zap.NewNop(), false) // Assertions assert.Error(t, err) @@ -195,6 +195,141 @@ func TestBuild_ShouldGetAnErrorIfProviderIsNotDefined(t *testing.T) { require.Len(t, dataSources, 0) } +func TestBuild_ShouldSkipDataSourcesForMissingProviderWhenSkipEnabled(t *testing.T) { + ctx := context.Background() + mockBuilder := datasource.NewMockProviderBuilder[config.NatsEventSource, *nodev1.NatsEventConfiguration](t) + mockPubSubProvider := datasource.NewMockProvider(t) + + dsMeta := &plan.DataSourceMetadata{ + RootNodes: []plan.TypeField{ + { + TypeName: "Type1", + FieldNames: []string{"Field1", "Field2"}, + }, + }, + } + + // One event references a defined provider, the other a provider that is not defined. + definedEvent := &nodev1.NatsEventConfiguration{ + EngineEventConfiguration: &nodev1.EngineEventConfiguration{ + ProviderId: "provider-1", + TypeName: "Type1", + FieldName: "Field1", + Type: nodev1.EventType_PUBLISH, + }, + } + missingEvent := &nodev1.NatsEventConfiguration{ + EngineEventConfiguration: &nodev1.EngineEventConfiguration{ + ProviderId: "provider-2", + TypeName: "Type1", + FieldName: "Field2", + Type: nodev1.EventType_PUBLISH, + }, + } + dsConf := DataSourceConfigurationWithMetadata{ + Configuration: &nodev1.DataSourceConfiguration{ + Id: "test-id", + CustomEvents: &nodev1.DataSourceCustomEvents{ + Nats: []*nodev1.NatsEventConfiguration{definedEvent, missingEvent}, + }, + }, + Metadata: dsMeta, + } + dsConfs := []dsConfAndEvents[*nodev1.NatsEventConfiguration]{ + { + dsConf: &dsConf, + events: dsConf.Configuration.GetCustomEvents().GetNats(), + }, + } + natsEventSources := []config.NatsEventSource{ + {ID: "provider-1"}, + } + + mockPubSubProvider.On("ID").Return("provider-1") + mockPubSubProvider.On("SetHooks", mock.Anything) + + mockBuilder.On("TypeID").Return("nats") + mockBuilder.On("BuildProvider", natsEventSources[0], mock.Anything). + Return(mockPubSubProvider, nil) + + // Execute the function with ignoreMissingProviders enabled + providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}, zap.NewNop(), true) + + // Assertions: the router does not fail, only the data source for the missing provider is skipped + assert.NoError(t, err) + require.Len(t, providers, 1) + require.Len(t, dataSources, 1) + assert.True(t, dataSources[0].HasRootNode("Type1", "Field1")) + assert.False(t, dataSources[0].HasRootNode("Type1", "Field2")) +} + +func TestBuild_ShouldSkipDataSourcesWhenMissingProviderIsFirst(t *testing.T) { + ctx := context.Background() + mockBuilder := datasource.NewMockProviderBuilder[config.NatsEventSource, *nodev1.NatsEventConfiguration](t) + mockPubSubProvider := datasource.NewMockProvider(t) + + dsMeta := &plan.DataSourceMetadata{ + RootNodes: []plan.TypeField{ + { + TypeName: "Type1", + FieldNames: []string{"Field1", "Field2"}, + }, + }, + } + + // The missing-provider event comes first so the surviving data source is built at a + // non-zero index, exercising the index gap in the generated data-source ID. + missingEvent := &nodev1.NatsEventConfiguration{ + EngineEventConfiguration: &nodev1.EngineEventConfiguration{ + ProviderId: "provider-2", + TypeName: "Type1", + FieldName: "Field1", + Type: nodev1.EventType_PUBLISH, + }, + } + definedEvent := &nodev1.NatsEventConfiguration{ + EngineEventConfiguration: &nodev1.EngineEventConfiguration{ + ProviderId: "provider-1", + TypeName: "Type1", + FieldName: "Field2", + Type: nodev1.EventType_PUBLISH, + }, + } + dsConf := DataSourceConfigurationWithMetadata{ + Configuration: &nodev1.DataSourceConfiguration{ + Id: "test-id", + CustomEvents: &nodev1.DataSourceCustomEvents{ + Nats: []*nodev1.NatsEventConfiguration{missingEvent, definedEvent}, + }, + }, + Metadata: dsMeta, + } + dsConfs := []dsConfAndEvents[*nodev1.NatsEventConfiguration]{ + { + dsConf: &dsConf, + events: dsConf.Configuration.GetCustomEvents().GetNats(), + }, + } + natsEventSources := []config.NatsEventSource{ + {ID: "provider-1"}, + } + + mockPubSubProvider.On("ID").Return("provider-1") + mockPubSubProvider.On("SetHooks", mock.Anything) + + mockBuilder.On("TypeID").Return("nats") + mockBuilder.On("BuildProvider", natsEventSources[0], mock.Anything). + Return(mockPubSubProvider, nil) + + providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}, zap.NewNop(), true) + + assert.NoError(t, err) + require.Len(t, providers, 1) + require.Len(t, dataSources, 1) + assert.True(t, dataSources[0].HasRootNode("Type1", "Field2")) + assert.False(t, dataSources[0].HasRootNode("Type1", "Field1")) +} + func TestBuild_ShouldNotInitializeProviderIfNotUsed(t *testing.T) { ctx := context.Background() mockBuilder := datasource.NewMockProviderBuilder[config.NatsEventSource, *nodev1.NatsEventConfiguration](t) @@ -251,7 +386,7 @@ func TestBuild_ShouldNotInitializeProviderIfNotUsed(t *testing.T) { Return(mockPubSubUsedProvider, nil) // Execute the function - providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}) + providers, dataSources, err := build(ctx, mockBuilder, natsEventSources, dsConfs, rmetric.NewNoopStreamMetricStore(), datasource.Hooks{}, zap.NewNop(), false) // Assertions assert.NoError(t, err) @@ -314,6 +449,70 @@ func TestBuildProvidersAndDataSources_Nats_OK(t *testing.T) { assert.False(t, dataSources[0].HasRootNode("Type1", "Field2")) } +func TestBuildProvidersAndDataSources_SkipMissingProviders(t *testing.T) { + ctx := context.Background() + + dsMeta := &plan.DataSourceMetadata{ + RootNodes: []plan.TypeField{ + { + TypeName: "Type1", + FieldNames: []string{"Field1", "Field2"}, + }, + }, + } + + // One event uses the configured provider, the other references a provider that is not defined. + definedEvent := &nodev1.NatsEventConfiguration{ + EngineEventConfiguration: &nodev1.EngineEventConfiguration{ + ProviderId: "provider-1", + TypeName: "Type1", + FieldName: "Field1", + Type: nodev1.EventType_PUBLISH, + }, + } + missingEvent := &nodev1.NatsEventConfiguration{ + EngineEventConfiguration: &nodev1.EngineEventConfiguration{ + ProviderId: "missing-provider", + TypeName: "Type1", + FieldName: "Field2", + Type: nodev1.EventType_PUBLISH, + }, + } + dsConf := DataSourceConfigurationWithMetadata{ + Configuration: &nodev1.DataSourceConfiguration{ + Id: "test-id", + CustomEvents: &nodev1.DataSourceCustomEvents{ + Nats: []*nodev1.NatsEventConfiguration{definedEvent, missingEvent}, + }, + }, + Metadata: dsMeta, + } + dsConfs := []DataSourceConfigurationWithMetadata{dsConf} + + eventsConfig := config.EventsConfiguration{ + Providers: config.EventProviders{ + Nats: []config.NatsEventSource{ + {ID: "provider-1"}, + }, + }, + } + + // Without the flag, a missing provider prevents the router from starting. + _, _, err := BuildProvidersAndDataSources(ctx, eventsConfig, nil, zap.NewNop(), dsConfs, "host", "addr", datasource.Hooks{}) + require.Error(t, err) + assert.IsType(t, &ProviderNotDefinedError{}, err) + + // With the flag enabled, the router starts and only the affected data source is skipped. + eventsConfig.SkipMissingProviders = true + providers, dataSources, err := BuildProvidersAndDataSources(ctx, eventsConfig, nil, zap.NewNop(), dsConfs, "host", "addr", datasource.Hooks{}) + require.NoError(t, err) + require.Len(t, providers, 1) + require.Equal(t, "provider-1", providers[0].ID()) + require.Len(t, dataSources, 1) + assert.True(t, dataSources[0].HasRootNode("Type1", "Field1")) + assert.False(t, dataSources[0].HasRootNode("Type1", "Field2")) +} + func TestBuildProvidersAndDataSources_Kafka_OK(t *testing.T) { ctx := context.Background() From 3df4b042b8097cec72eaf99380b93e584f330709 Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Tue, 23 Jun 2026 23:46:20 +0200 Subject: [PATCH 2/7] feat: allow router to start even with unavailable provider --- docs-website/router/configuration.mdx | 10 +- docs-website/router/cosmo-streams.mdx | 13 +- router/core/graph_server.go | 52 ++++++-- router/core/graph_server_test.go | 113 ++++++++++++++++++ router/demo.config.yaml | 2 +- router/pkg/config/config.go | 16 +-- router/pkg/config/config.schema.json | 4 +- router/pkg/config/config_test.go | 12 +- router/pkg/config/fixtures/full.yaml | 2 +- .../pkg/config/testdata/config_defaults.json | 2 +- router/pkg/config/testdata/config_full.json | 2 +- .../pkg/pubsub/datasource/pubsubprovider.go | 37 ++++++ .../pubsub/datasource/pubsubprovider_test.go | 47 ++++++++ router/pkg/pubsub/nats/engine_datasource.go | 6 + .../pkg/pubsub/nats/engine_datasource_test.go | 18 +++ router/pkg/pubsub/pubsub.go | 16 +-- router/pkg/pubsub/pubsub_test.go | 4 +- router/pkg/pubsub/redis/adapter.go | 8 ++ router/pkg/pubsub/redis/adapter_test.go | 20 ++++ 19 files changed, 338 insertions(+), 46 deletions(-) diff --git a/docs-website/router/configuration.mdx b/docs-website/router/configuration.mdx index 346a7e3266..6263e3b3e6 100644 --- a/docs-website/router/configuration.mdx +++ b/docs-website/router/configuration.mdx @@ -1814,7 +1814,7 @@ We support NATS, Kafka and Redis as event bus provider. version: "1" events: - skip_missing_providers: false + skip_unavailable_providers: false providers: nats: - id: default @@ -1841,14 +1841,14 @@ events: redis: - id: my-redis urls: - - "localhost:9092" + - "redis://localhost:6379" ``` ### Options -| Environment Variable | YAML | Required | Description | Default Value | -| ----------------------------- | ---------------------- | ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------- | -| EVENTS_SKIP_MISSING_PROVIDERS | skip_missing_providers | | Start the router even when the execution config references an event provider that is not defined under `providers`. The router logs an error and disables only the affected fields instead of failing to start. By default a missing provider prevents startup. | false | +| Environment Variable | YAML | Required | Description | Default Value | +| --------------------------------- | -------------------------- | ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------- | +| EVENTS_SKIP_UNAVAILABLE_PROVIDERS | skip_unavailable_providers | | Start the router even when an event provider referenced by the execution config is unavailable: either not defined under `providers`, or defined but unreachable at startup (e.g. the broker is down). The router logs an error and disables only the affected fields instead of failing to start. By default this prevents startup. | false | ### Provider diff --git a/docs-website/router/cosmo-streams.mdx b/docs-website/router/cosmo-streams.mdx index 926c19c388..bedb84a80b 100644 --- a/docs-website/router/cosmo-streams.mdx +++ b/docs-website/router/cosmo-streams.mdx @@ -186,15 +186,20 @@ We've intentionally moved this part of the configuration out of the Schema to ke If you run `make edfs-demo` in the Cosmo Monorepo, you'll automatically get a NATS instance running on the default port (4222) using Docker. -### Missing providers +### Unavailable providers -If the execution config references a provider ID that is not defined under `events.providers`, the router fails to start with an error such as `redis provider with ID my-provider is not defined`. This stops a router from coming up with a partially-configured event setup. +By default the router fails to start if an event provider referenced by the execution config is unavailable. This happens in two cases: -Set `skip_missing_providers: true` (or the `EVENTS_SKIP_MISSING_PROVIDERS` environment variable) to start the router anyway. The router logs the error and disables only the fields backed by the missing provider. The rest of the graph keeps serving traffic, and requests to the affected fields fail at query time. Set up an alert on the logged error to detect the missing configuration. +- The provider is not defined under `events.providers`. The router reports an error such as `redis provider with ID my-provider is not defined`. +- The provider is defined but cannot be reached at startup (e.g. the broker is down). + +This stops a router from coming up with a broken event setup. + +Set `skip_unavailable_providers: true` (or the `EVENTS_SKIP_UNAVAILABLE_PROVIDERS` environment variable) to start the router anyway. The router logs the error and disables only the fields backed by the unavailable provider. The rest of the graph keeps serving traffic. A field whose provider is not defined is dropped (it cannot be queried); a field whose provider failed to connect returns an error at query time. Set up an alert on the logged error to detect the misconfiguration or outage. ```yaml config.yaml events: - skip_missing_providers: true + skip_unavailable_providers: true providers: nats: - id: default diff --git a/router/core/graph_server.go b/router/core/graph_server.go index af206997be..dc6ad9e8e1 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -2202,9 +2202,12 @@ func (s *graphServer) startupPubSubProviders(ctx context.Context) error { // Default timeout for pubsub provider startup const defaultStartupTimeout = 5 * time.Second + // When skip_unavailable_providers is enabled, a provider that fails to start (e.g. the + // broker is unreachable) must not prevent the router from starting. The provider is + // logged and marked unavailable so requests to its fields fail gracefully instead. return s.providersActionWithTimeout(ctx, func(ctx context.Context, provider datasource.Provider) error { return provider.Startup(ctx) - }, defaultStartupTimeout, "pubsub provider startup timed out") + }, defaultStartupTimeout, "pubsub provider startup timed out", s.eventsConfig.SkipUnavailableProviders) } // shutdownPubSubProviders shuts down all pubsub providers @@ -2216,29 +2219,62 @@ func (s *graphServer) shutdownPubSubProviders(ctx context.Context) error { return s.providersActionWithTimeout(ctx, func(ctx context.Context, provider datasource.Provider) error { return provider.Shutdown(ctx) - }, defaultShutdownTimeout, "pubsub provider shutdown timed out") + }, defaultShutdownTimeout, "pubsub provider shutdown timed out", false) } -func (s *graphServer) providersActionWithTimeout(ctx context.Context, action func(ctx context.Context, provider datasource.Provider) error, timeout time.Duration, timeoutMessage string) error { +// providerUnavailableMarker is implemented by providers that can be marked unavailable +// so that requests to their fields fail gracefully instead of using a provider that +// could not be started. +type providerUnavailableMarker interface { + MarkUnavailable() +} + +// providersActionWithTimeout runs action against every pubsub provider concurrently, +// bounding each by timeout. When continueOnError is true, a provider whose action fails +// or times out is logged and marked unavailable instead of aborting the whole action; +// this is used at startup so an unreachable provider does not prevent the router from +// starting. +func (s *graphServer) providersActionWithTimeout(ctx context.Context, action func(ctx context.Context, provider datasource.Provider) error, timeout time.Duration, timeoutMessage string, continueOnError bool) error { cancellableCtx, cancel := context.WithCancel(ctx) defer cancel() - timer := time.NewTimer(timeout) - defer timer.Stop() - providersGroup := new(errgroup.Group) for _, provider := range s.pubSubProviders { providersGroup.Go(func() error { + // Each provider is bounded by its own timer. A single shared timer would only + // ever deliver its fire to one goroutine, so multiple slow/unreachable providers + // would block the others forever and hang startup. + timer := time.NewTimer(timeout) + defer timer.Stop() + actionDone := make(chan error, 1) go func() { actionDone <- action(cancellableCtx, provider) }() + var actionErr error select { case err := <-actionDone: - return err + actionErr = err case <-timer.C: - return errors.New(timeoutMessage) + actionErr = errors.New(timeoutMessage) } + if actionErr == nil || !continueOnError { + return actionErr + } + // Lenient mode: tolerate the failure only if the provider can be marked + // unavailable, so that requests to its fields fail gracefully instead of using + // an unconnected adapter. If it cannot be marked, do not swallow the error. + marker, ok := provider.(providerUnavailableMarker) + if !ok { + return actionErr + } + s.logger.Error("EDFS provider is unavailable; the router will keep running and the fields backed by this provider will be unavailable", + zap.String("provider_id", provider.ID()), + zap.String("provider_type", provider.TypeID()), + zap.Error(actionErr), + ) + marker.MarkUnavailable() + return nil }) } diff --git a/router/core/graph_server_test.go b/router/core/graph_server_test.go index e312add8f2..3dd5ddc733 100644 --- a/router/core/graph_server_test.go +++ b/router/core/graph_server_test.go @@ -1,16 +1,129 @@ package core import ( + "context" + "errors" "slices" "testing" + "time" "golang.org/x/exp/constraints" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" "github.com/wundergraph/cosmo/router/pkg/config" + "github.com/wundergraph/cosmo/router/pkg/pubsub/datasource" + "go.uber.org/zap" ) +// unavailableTestSubConfig is a minimal SubscriptionEventConfiguration used to probe +// whether a provider rejects subscriptions after being marked unavailable. +type unavailableTestSubConfig struct{} + +func (unavailableTestSubConfig) ProviderID() string { return "redis-1" } +func (unavailableTestSubConfig) ProviderType() datasource.ProviderType { return datasource.ProviderTypeRedis } +func (unavailableTestSubConfig) RootFieldName() string { return "field" } + +func TestStartupPubSubProviders_SkipUnavailableProviders(t *testing.T) { + t.Parallel() + + newServer := func(skip bool, providers ...datasource.Provider) *graphServer { + return &graphServer{ + Config: &Config{ + logger: zap.NewNop(), + eventsConfig: config.EventsConfiguration{SkipUnavailableProviders: skip}, + }, + pubSubProviders: providers, + } + } + + t.Run("aborts startup when a provider fails and the flag is disabled", func(t *testing.T) { + t.Parallel() + + adapter := datasource.NewMockProvider(t) + adapter.On("Startup", mock.Anything).Return(errors.New("connection refused")) + provider := datasource.NewPubSubProvider("redis-1", "redis", adapter, zap.NewNop(), nil) + + s := newServer(false, provider) + require.Error(t, s.startupPubSubProviders(context.Background())) + }) + + t.Run("starts anyway and marks the provider unavailable when the flag is enabled", func(t *testing.T) { + t.Parallel() + + adapter := datasource.NewMockProvider(t) + adapter.On("Startup", mock.Anything).Return(errors.New("connection refused")) + provider := datasource.NewPubSubProvider("redis-1", "redis", adapter, zap.NewNop(), nil) + + s := newServer(true, provider) + require.NoError(t, s.startupPubSubProviders(context.Background())) + + // The provider was marked unavailable: a subscribe attempt returns an error + // without ever reaching the (unconnected) adapter, so the router does not panic. + err := provider.Subscribe(context.Background(), unavailableTestSubConfig{}, nil) + require.Error(t, err) + }) +} + +// TestProvidersActionWithTimeout_MultipleHangingProvidersDoNotDeadlock guards against a +// regression to a single shared timer: with per-provider timers, several providers whose +// action never returns must each independently time out instead of blocking startup. +func TestProvidersActionWithTimeout_MultipleHangingProvidersDoNotDeadlock(t *testing.T) { + t.Parallel() + + p1 := datasource.NewPubSubProvider("p1", "redis", datasource.NewMockProvider(t), zap.NewNop(), nil) + p2 := datasource.NewPubSubProvider("p2", "nats", datasource.NewMockProvider(t), zap.NewNop(), nil) + + s := &graphServer{ + Config: &Config{logger: zap.NewNop()}, + pubSubProviders: []datasource.Provider{p1, p2}, + } + + // The action blocks until its context is cancelled, simulating two providers whose + // Startup hangs (e.g. unreachable brokers). + hangingAction := func(ctx context.Context, _ datasource.Provider) error { + <-ctx.Done() + return ctx.Err() + } + + done := make(chan error, 1) + go func() { + done <- s.providersActionWithTimeout(context.Background(), hangingAction, 50*time.Millisecond, "timed out", true) + }() + + select { + case err := <-done: + require.NoError(t, err) // both timed out and were tolerated in lenient mode + case <-time.After(5 * time.Second): + t.Fatal("providersActionWithTimeout deadlocked with multiple hanging providers") + } + + // Both providers must have been marked unavailable by their own timeouts. + require.Error(t, p1.UnavailableError()) + require.Error(t, p2.UnavailableError()) +} + +// TestShutdownPubSubProviders_StaysStrictWithSkipUnavailableProviders pins that shutdown +// errors are never swallowed, even when skip_unavailable_providers is enabled. +func TestShutdownPubSubProviders_StaysStrictWithSkipUnavailableProviders(t *testing.T) { + t.Parallel() + + adapter := datasource.NewMockProvider(t) + adapter.On("Shutdown", mock.Anything).Return(errors.New("shutdown failed")) + provider := datasource.NewPubSubProvider("redis-1", "redis", adapter, zap.NewNop(), nil) + + s := &graphServer{ + Config: &Config{ + logger: zap.NewNop(), + eventsConfig: config.EventsConfiguration{SkipUnavailableProviders: true}, + }, + pubSubProviders: []datasource.Provider{provider}, + } + + require.Error(t, s.shutdownPubSubProviders(context.Background())) +} + func TestGetRoutingUrlGroupingForCircuitBreakers(t *testing.T) { t.Parallel() diff --git a/router/demo.config.yaml b/router/demo.config.yaml index ccea543c6d..1b94fbec6e 100644 --- a/router/demo.config.yaml +++ b/router/demo.config.yaml @@ -19,4 +19,4 @@ events: redis: - id: my-redis urls: - - "redis://localhost:6379/2" \ No newline at end of file + - "redis://localhost:6379/2" diff --git a/router/pkg/config/config.go b/router/pkg/config/config.go index 6c066096ab..bcbd772d51 100644 --- a/router/pkg/config/config.go +++ b/router/pkg/config/config.go @@ -801,13 +801,15 @@ type EventProviders struct { type EventsConfiguration struct { Providers EventProviders `yaml:"providers,omitempty"` - // SkipMissingProviders allows the router to start even when the execution config - // references an event provider that is not defined in the router configuration. - // When enabled, the router logs an error and skips the data sources that depend on - // the missing provider instead of failing to start. Only the affected fields become - // unavailable; the rest of the graph keeps serving traffic. - SkipMissingProviders bool `yaml:"skip_missing_providers" envDefault:"false" env:"EVENTS_SKIP_MISSING_PROVIDERS"` - Handlers StreamsHandlerConfiguration `yaml:"handlers,omitempty"` + // SkipUnavailableProviders allows the router to start even when an event provider + // referenced by the execution config is unavailable: either not defined in the router + // configuration, or defined but unreachable at startup (e.g. the broker is down). + // When enabled, the router logs an error and starts anyway instead of failing. A + // provider that is not defined has its data sources skipped; a provider that fails to + // connect is marked unavailable so requests to the affected fields return an error + // instead of crashing the router. The rest of the graph keeps serving traffic. + SkipUnavailableProviders bool `yaml:"skip_unavailable_providers" envDefault:"false" env:"EVENTS_SKIP_UNAVAILABLE_PROVIDERS"` + Handlers StreamsHandlerConfiguration `yaml:"handlers,omitempty"` } type StreamsHandlerConfiguration struct { diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json index 3929c60da3..4b4850d348 100644 --- a/router/pkg/config/config.schema.json +++ b/router/pkg/config/config.schema.json @@ -2891,10 +2891,10 @@ "description": "The configuration for EDFS. See https://cosmo-docs.wundergraph.com/router/event-driven-federated-subscriptions-edfs for more information.", "additionalProperties": false, "properties": { - "skip_missing_providers": { + "skip_unavailable_providers": { "type": "boolean", "default": false, - "description": "Skip the data sources that reference an event provider not defined in this configuration, instead of aborting startup. When enabled, the router logs an error and starts anyway, leaving only the affected fields unavailable. Use this when a missing provider should not prevent the router from serving the rest of the graph. Defaults to false, in which case a missing provider prevents the router from starting." + "description": "Start the router even when an event provider referenced by the execution config is unavailable, instead of aborting startup. This covers both a provider that is not defined in this configuration and a provider that is defined but cannot be reached at startup (e.g. the broker is down). The router logs an error and starts anyway, leaving only the fields backed by that provider unavailable; requests to them return an error instead of crashing the router. Defaults to false, in which case an unavailable provider prevents the router from starting." }, "providers": { "type": "object", diff --git a/router/pkg/config/config_test.go b/router/pkg/config/config_test.go index e0ca9d62e8..e57701b7cf 100644 --- a/router/pkg/config/config_test.go +++ b/router/pkg/config/config_test.go @@ -177,7 +177,7 @@ engine: }) } -func TestEventsSkipMissingProvidersConfigLoading(t *testing.T) { +func TestEventsSkipUnavailableProvidersConfigLoading(t *testing.T) { t.Run("defaults to false", func(t *testing.T) { t.Parallel() @@ -190,7 +190,7 @@ graph: cfg, err := LoadConfig([]string{f}) require.NoError(t, err) - require.False(t, cfg.Config.Events.SkipMissingProviders) + require.False(t, cfg.Config.Events.SkipUnavailableProviders) }) t.Run("can be enabled from yaml", func(t *testing.T) { @@ -203,16 +203,16 @@ graph: token: "token" events: - skip_missing_providers: true + skip_unavailable_providers: true `) cfg, err := LoadConfig([]string{f}) require.NoError(t, err) - require.True(t, cfg.Config.Events.SkipMissingProviders) + require.True(t, cfg.Config.Events.SkipUnavailableProviders) }) t.Run("can be enabled from env", func(t *testing.T) { - t.Setenv("EVENTS_SKIP_MISSING_PROVIDERS", "true") + t.Setenv("EVENTS_SKIP_UNAVAILABLE_PROVIDERS", "true") f := createTempFileFromFixture(t, ` version: "1" @@ -223,7 +223,7 @@ graph: cfg, err := LoadConfig([]string{f}) require.NoError(t, err) - require.True(t, cfg.Config.Events.SkipMissingProviders) + require.True(t, cfg.Config.Events.SkipUnavailableProviders) }) } diff --git a/router/pkg/config/fixtures/full.yaml b/router/pkg/config/fixtures/full.yaml index 06976b88fb..257ece80b2 100644 --- a/router/pkg/config/fixtures/full.yaml +++ b/router/pkg/config/fixtures/full.yaml @@ -358,7 +358,7 @@ cdn: cache_size: 100MB events: - skip_missing_providers: true + skip_unavailable_providers: true providers: nats: - id: default diff --git a/router/pkg/config/testdata/config_defaults.json b/router/pkg/config/testdata/config_defaults.json index 6ed523cb17..cfc44cb210 100644 --- a/router/pkg/config/testdata/config_defaults.json +++ b/router/pkg/config/testdata/config_defaults.json @@ -373,7 +373,7 @@ "Kafka": null, "Redis": null }, - "SkipMissingProviders": false, + "SkipUnavailableProviders": false, "Handlers": { "OnReceiveEvents": { "MaxConcurrentHandlers": 100, diff --git a/router/pkg/config/testdata/config_full.json b/router/pkg/config/testdata/config_full.json index 6759f9b85d..068411d897 100644 --- a/router/pkg/config/testdata/config_full.json +++ b/router/pkg/config/testdata/config_full.json @@ -783,7 +783,7 @@ } ] }, - "SkipMissingProviders": true, + "SkipUnavailableProviders": true, "Handlers": { "OnReceiveEvents": { "MaxConcurrentHandlers": 100, diff --git a/router/pkg/pubsub/datasource/pubsubprovider.go b/router/pkg/pubsub/datasource/pubsubprovider.go index 74ff9ff9bb..6750a2b9a9 100644 --- a/router/pkg/pubsub/datasource/pubsubprovider.go +++ b/router/pkg/pubsub/datasource/pubsubprovider.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "slices" + "sync/atomic" "go.uber.org/zap" "go.uber.org/zap/zapcore" @@ -16,6 +17,11 @@ type PubSubProvider struct { Logger *zap.Logger hooks Hooks eventBuilder EventBuilderFn + // unavailable is set when the provider could not be started (e.g. the broker is + // unreachable) and the router was configured to start anyway. Once set, Subscribe + // and Publish return an error instead of using the underlying adapter, so the + // affected fields fail gracefully rather than crashing the router. + unavailable atomic.Bool } // applyPublishEventHooks processes events through a chain of hook functions @@ -67,7 +73,31 @@ func (p *PubSubProvider) Startup(ctx context.Context) error { return nil } +// MarkUnavailable flags the provider as unavailable. It is called when the provider +// failed to start and the router was configured (events.skip_unavailable_providers) +// to keep running. Subsequent Subscribe/Publish calls return an error instead of +// using the underlying adapter, which may not have an established connection. +func (p *PubSubProvider) MarkUnavailable() { + p.unavailable.Store(true) +} + +// UnavailableError returns a non-nil error if the provider has been marked unavailable. +// Data sources that reach into the underlying adapter directly (e.g. NATS request/reply) +// should call this before using it, so a provider that failed to start is not used. +func (p *PubSubProvider) UnavailableError() error { + if p.unavailable.Load() { + return NewError(fmt.Sprintf("event provider %q (%s) is unavailable", p.id, p.typeID), nil) + } + return nil +} + func (p *PubSubProvider) Shutdown(ctx context.Context) error { + // A provider marked unavailable failed to start, so it has no connection to tear down. + // Skip the adapter: its connection fields may still be written by an in-flight Startup + // goroutine, and reading them here would race that write. + if p.unavailable.Load() { + return nil + } if err := p.Adapter.Shutdown(ctx); err != nil { return err } @@ -75,10 +105,17 @@ func (p *PubSubProvider) Shutdown(ctx context.Context) error { } func (p *PubSubProvider) Subscribe(ctx context.Context, cfg SubscriptionEventConfiguration, updater SubscriptionEventUpdater) error { + if err := p.UnavailableError(); err != nil { + return err + } return p.Adapter.Subscribe(ctx, cfg, updater) } func (p *PubSubProvider) Publish(ctx context.Context, cfg PublishEventConfiguration, events []StreamEvent) error { + if err := p.UnavailableError(); err != nil { + return err + } + if len(p.hooks.OnPublishEvents.Handlers) == 0 { return p.Adapter.Publish(ctx, cfg, events) } diff --git a/router/pkg/pubsub/datasource/pubsubprovider_test.go b/router/pkg/pubsub/datasource/pubsubprovider_test.go index b956ab38f0..781819693c 100644 --- a/router/pkg/pubsub/datasource/pubsubprovider_test.go +++ b/router/pkg/pubsub/datasource/pubsubprovider_test.go @@ -169,6 +169,53 @@ func TestProvider_Subscribe_Error(t *testing.T) { assert.Equal(t, expectedError, err) } +func TestProvider_MarkUnavailable_SubscribeReturnsErrorWithoutCallingAdapter(t *testing.T) { + // The adapter must NOT be reached once the provider is marked unavailable; the mock + // has no expectations set, so any adapter call would fail the test. + mockAdapter := NewMockProvider(t) + config := &testSubscriptionConfig{ + providerID: "test-provider", + providerType: ProviderTypeRedis, + fieldName: "testField", + } + + provider := PubSubProvider{ + id: "test-provider", + typeID: "redis", + Adapter: mockAdapter, + } + provider.MarkUnavailable() + + err := provider.Subscribe(context.Background(), config, nil) + + assert.Error(t, err) + var providerErr *Error + assert.ErrorAs(t, err, &providerErr) +} + +func TestProvider_MarkUnavailable_PublishReturnsErrorWithoutCallingAdapter(t *testing.T) { + mockAdapter := NewMockProvider(t) + config := &testPublishConfig{ + providerID: "test-provider", + providerType: ProviderTypeRedis, + fieldName: "testField", + } + events := []StreamEvent{&testEvent{mutableTestEvent("test data")}} + + provider := PubSubProvider{ + id: "test-provider", + typeID: "redis", + Adapter: mockAdapter, + } + provider.MarkUnavailable() + + err := provider.Publish(context.Background(), config, events) + + assert.Error(t, err) + var providerErr *Error + assert.ErrorAs(t, err, &providerErr) +} + func TestProvider_Publish_NoHooks_Success(t *testing.T) { mockAdapter := NewMockProvider(t) config := &testPublishConfig{ diff --git a/router/pkg/pubsub/nats/engine_datasource.go b/router/pkg/pubsub/nats/engine_datasource.go index 6739c418ba..d2a0d7c3f3 100644 --- a/router/pkg/pubsub/nats/engine_datasource.go +++ b/router/pkg/pubsub/nats/engine_datasource.go @@ -247,6 +247,12 @@ func (s *NatsRequestDataSource) Load(ctx context.Context, headers http.Header, i return nil, fmt.Errorf("adapter for provider %s is not of the right type", publishData.Provider) } + // The request path reaches into the underlying adapter directly, so it must honor the + // unavailable flag itself (Subscribe/Publish are guarded by the provider wrapper). + if err := providerBase.UnavailableError(); err != nil { + return nil, err + } + adapter, ok := providerBase.Adapter.(Adapter) if !ok { return nil, fmt.Errorf("adapter for provider %s is not of the right type", publishData.Provider) diff --git a/router/pkg/pubsub/nats/engine_datasource_test.go b/router/pkg/pubsub/nats/engine_datasource_test.go index bb38a1ce66..6914bdf2db 100644 --- a/router/pkg/pubsub/nats/engine_datasource_test.go +++ b/router/pkg/pubsub/nats/engine_datasource_test.go @@ -210,6 +210,24 @@ func TestNatsRequestDataSource_Load(t *testing.T) { } } +func TestNatsRequestDataSource_Load_UnavailableProvider(t *testing.T) { + t.Parallel() + + // No Request expectation is set on the mock, so the request path must not reach the + // adapter once the provider is marked unavailable. + mockAdapter := NewMockAdapter(t) + provider := datasource.NewPubSubProvider("test-provider", "nats", mockAdapter, zap.NewNop(), testNatsEventBuilder) + provider.MarkUnavailable() + + dataSource := &NatsRequestDataSource{ + pubSub: provider, + } + + input := []byte(`{"subject":"test-subject", "event": {"data":{"message":"hello"}}, "providerId":"test-provider"}`) + _, err := dataSource.Load(context.Background(), nil, input) + require.Error(t, err) +} + func TestNatsRequestDataSource_LoadWithFiles(t *testing.T) { dataSource := &NatsRequestDataSource{} assert.Panics(t, func() { diff --git a/router/pkg/pubsub/pubsub.go b/router/pkg/pubsub/pubsub.go index 7360ff030e..ffb4a50adb 100644 --- a/router/pkg/pubsub/pubsub.go +++ b/router/pkg/pubsub/pubsub.go @@ -70,8 +70,8 @@ func BuildProvidersAndDataSources( logger = zap.NewNop() } - if config.SkipMissingProviders { - logger.Warn("EDFS lenient mode is enabled (events.skip_missing_providers=true): the router will start even if an event provider referenced by the execution config is not defined, disabling only the affected fields") + if config.SkipUnavailableProviders { + logger.Warn("EDFS lenient mode is enabled (events.skip_unavailable_providers=true): the router will start even if an event provider referenced by the execution config is undefined or unreachable, disabling only the affected fields") } var pubSubProviders []pubsub_datasource.Provider @@ -86,7 +86,7 @@ func BuildProvidersAndDataSources( events: dsConf.Configuration.GetCustomEvents().GetKafka(), }) } - kafkaPubSubProviders, kafkaOuts, err := build(ctx, kafkaBuilder, config.Providers.Kafka, kafkaDsConfsWithEvents, store, hooks, logger, config.SkipMissingProviders) + kafkaPubSubProviders, kafkaOuts, err := build(ctx, kafkaBuilder, config.Providers.Kafka, kafkaDsConfsWithEvents, store, hooks, logger, config.SkipUnavailableProviders) if err != nil { return nil, nil, err } @@ -104,7 +104,7 @@ func BuildProvidersAndDataSources( events: dsConf.Configuration.GetCustomEvents().GetNats(), }) } - natsPubSubProviders, natsOuts, err := build(ctx, natsBuilder, config.Providers.Nats, natsDsConfsWithEvents, store, hooks, logger, config.SkipMissingProviders) + natsPubSubProviders, natsOuts, err := build(ctx, natsBuilder, config.Providers.Nats, natsDsConfsWithEvents, store, hooks, logger, config.SkipUnavailableProviders) if err != nil { return nil, nil, err } @@ -122,7 +122,7 @@ func BuildProvidersAndDataSources( events: dsConf.Configuration.GetCustomEvents().GetRedis(), }) } - redisPubSubProviders, redisOuts, err := build(ctx, redisBuilder, config.Providers.Redis, redisDsConfsWithEvents, store, hooks, logger, config.SkipMissingProviders) + redisPubSubProviders, redisOuts, err := build(ctx, redisBuilder, config.Providers.Redis, redisDsConfsWithEvents, store, hooks, logger, config.SkipUnavailableProviders) if err != nil { return nil, nil, err } @@ -141,7 +141,7 @@ func build[P GetID, E GetEngineEventConfiguration]( store metric.StreamMetricStore, hooks pubsub_datasource.Hooks, logger *zap.Logger, - skipMissingProviders bool, + skipUnavailableProviders bool, ) (map[string]pubsub_datasource.Provider, []plan.DataSource, error) { pubSubProviders := make(map[string]pubsub_datasource.Provider) var outs []plan.DataSource @@ -181,7 +181,7 @@ func build[P GetID, E GetEngineEventConfiguration]( ProviderID: providerId, ProviderTypeID: builder.TypeID(), } - if !skipMissingProviders { + if !skipUnavailableProviders { return pubSubProviders, nil, err } // Lenient mode: do not prevent the router from starting. Log the error so it @@ -198,7 +198,7 @@ func build[P GetID, E GetEngineEventConfiguration]( for _, dsConf := range dsConfs { for i, event := range dsConf.events { // Skip events that reference a provider which could not be initialized - // (only possible when skipMissingProviders is enabled). + // (only possible when skipUnavailableProviders is enabled). if _, ok := missingProviderIds[event.GetEngineEventConfiguration().GetProviderId()]; ok { continue } diff --git a/router/pkg/pubsub/pubsub_test.go b/router/pkg/pubsub/pubsub_test.go index 4b90833790..01369830c2 100644 --- a/router/pkg/pubsub/pubsub_test.go +++ b/router/pkg/pubsub/pubsub_test.go @@ -449,7 +449,7 @@ func TestBuildProvidersAndDataSources_Nats_OK(t *testing.T) { assert.False(t, dataSources[0].HasRootNode("Type1", "Field2")) } -func TestBuildProvidersAndDataSources_SkipMissingProviders(t *testing.T) { +func TestBuildProvidersAndDataSources_SkipUnavailableProviders(t *testing.T) { ctx := context.Background() dsMeta := &plan.DataSourceMetadata{ @@ -503,7 +503,7 @@ func TestBuildProvidersAndDataSources_SkipMissingProviders(t *testing.T) { assert.IsType(t, &ProviderNotDefinedError{}, err) // With the flag enabled, the router starts and only the affected data source is skipped. - eventsConfig.SkipMissingProviders = true + eventsConfig.SkipUnavailableProviders = true providers, dataSources, err := BuildProvidersAndDataSources(ctx, eventsConfig, nil, zap.NewNop(), dsConfs, "host", "addr", datasource.Hooks{}) require.NoError(t, err) require.Len(t, providers, 1) diff --git a/router/pkg/pubsub/redis/adapter.go b/router/pkg/pubsub/redis/adapter.go index 69c55d7bea..0c54393cdd 100644 --- a/router/pkg/pubsub/redis/adapter.go +++ b/router/pkg/pubsub/redis/adapter.go @@ -96,6 +96,14 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri zap.String("method", "subscribe"), zap.Strings("channels", subConf.Channels), ) + + // Defensive backstop behind PubSubProvider.UnavailableError(): a provider that failed + // to connect is normally short-circuited before reaching the adapter, but guard the + // nil connection here too so an unstarted adapter returns an error instead of panicking. + if p.conn == nil { + return datasource.NewError("redis connection not initialized", nil) + } + sub := p.conn.PSubscribe(ctx, subConf.Channels...) msgChan := sub.Channel() diff --git a/router/pkg/pubsub/redis/adapter_test.go b/router/pkg/pubsub/redis/adapter_test.go index 78ae91342e..261cd2d7e1 100644 --- a/router/pkg/pubsub/redis/adapter_test.go +++ b/router/pkg/pubsub/redis/adapter_test.go @@ -19,6 +19,26 @@ func (n *noopUpdater) Complete() {} func (n *noopUpdater) Done() {} func (n *noopUpdater) SetHooks(_ datasource.Hooks) {} +func TestProviderAdapter_SubscribeWithoutStartupReturnsError(t *testing.T) { + t.Parallel() + + ctx := context.Background() + // The adapter is created but Startup is never called, so p.conn stays nil. This + // mirrors the state of a provider that failed to connect under skip_unavailable_providers. + adapter := NewProviderAdapter(ctx, zaptest.NewLogger(t), []string{"redis://localhost:6379"}, false, datasource.ProviderOpts{}) + + conf := &SubscriptionEventConfiguration{ + Provider: "test-provider", + Channels: []string{"test-channel"}, + FieldName: "testField", + } + + require.NotPanics(t, func() { + err := adapter.Subscribe(ctx, conf, &noopUpdater{}) + require.Error(t, err) + }) +} + func TestProviderAdapter_ConnectionCleanupOnUnsubscribe(t *testing.T) { t.Parallel() From bc3fc3d252541a0aa181ce0ef8c0c693aecb5e5a Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Wed, 24 Jun 2026 09:18:23 +0200 Subject: [PATCH 3/7] chore: improve format --- router/core/graph_server_test.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/router/core/graph_server_test.go b/router/core/graph_server_test.go index 3dd5ddc733..5aed89adc2 100644 --- a/router/core/graph_server_test.go +++ b/router/core/graph_server_test.go @@ -21,9 +21,11 @@ import ( // whether a provider rejects subscriptions after being marked unavailable. type unavailableTestSubConfig struct{} -func (unavailableTestSubConfig) ProviderID() string { return "redis-1" } -func (unavailableTestSubConfig) ProviderType() datasource.ProviderType { return datasource.ProviderTypeRedis } -func (unavailableTestSubConfig) RootFieldName() string { return "field" } +func (unavailableTestSubConfig) ProviderID() string { return "redis-1" } +func (unavailableTestSubConfig) ProviderType() datasource.ProviderType { + return datasource.ProviderTypeRedis +} +func (unavailableTestSubConfig) RootFieldName() string { return "field" } func TestStartupPubSubProviders_SkipUnavailableProviders(t *testing.T) { t.Parallel() From 59a2927caa449ec84897768d5f33616eb67994e7 Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Mon, 29 Jun 2026 17:26:14 +0200 Subject: [PATCH 4/7] feat: if provider not available, router now try to connect to it again --- docs-website/router/configuration.mdx | 2 +- docs-website/router/cosmo-streams.mdx | 11 +- .../events/kafka_skip_unavailable_test.go | 255 ++++++++++++++++++ .../events/nats_skip_unavailable_test.go | 193 +++++++++++++ .../events/redis_skip_unavailable_test.go | 191 +++++++++++++ router-tests/testenv/tcpproxy.go | 177 ++++++++++++ router/core/graph_server.go | 39 ++- router/core/graph_server_test.go | 24 +- router/internal/rediscloser/rediscloser.go | 15 +- .../internal/rediscloser/rediscloser_test.go | 5 +- router/pkg/config/config.go | 5 +- router/pkg/pubsub/datasource/provider.go | 6 + .../pkg/pubsub/datasource/pubsubprovider.go | 42 +-- .../pubsub/datasource/pubsubprovider_test.go | 47 ---- router/pkg/pubsub/kafka/adapter.go | 18 ++ router/pkg/pubsub/nats/adapter.go | 35 ++- router/pkg/pubsub/nats/engine_datasource.go | 6 - .../pkg/pubsub/nats/engine_datasource_test.go | 18 -- router/pkg/pubsub/nats/provider_builder.go | 15 +- .../pkg/pubsub/nats/provider_builder_test.go | 44 ++- router/pkg/pubsub/pubsub.go | 3 +- router/pkg/pubsub/redis/adapter.go | 30 ++- 22 files changed, 996 insertions(+), 185 deletions(-) create mode 100644 router-tests/events/kafka_skip_unavailable_test.go create mode 100644 router-tests/events/nats_skip_unavailable_test.go create mode 100644 router-tests/events/redis_skip_unavailable_test.go create mode 100644 router-tests/testenv/tcpproxy.go diff --git a/docs-website/router/configuration.mdx b/docs-website/router/configuration.mdx index 6263e3b3e6..1fc053444a 100644 --- a/docs-website/router/configuration.mdx +++ b/docs-website/router/configuration.mdx @@ -1848,7 +1848,7 @@ events: | Environment Variable | YAML | Required | Description | Default Value | | --------------------------------- | -------------------------- | ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------- | -| EVENTS_SKIP_UNAVAILABLE_PROVIDERS | skip_unavailable_providers | | Start the router even when an event provider referenced by the execution config is unavailable: either not defined under `providers`, or defined but unreachable at startup (e.g. the broker is down). The router logs an error and disables only the affected fields instead of failing to start. By default this prevents startup. | false | +| EVENTS_SKIP_UNAVAILABLE_PROVIDERS | skip_unavailable_providers | | Start the router even when an event provider referenced by the execution config is unavailable: either not defined under `providers`, or defined but unreachable at startup (e.g. the broker is down). The router logs an error and disables only the affected fields instead of failing to start. A provider that is unreachable reconnects in the background and its fields recover without a restart once the broker is back. By default this prevents startup. | false | ### Provider diff --git a/docs-website/router/cosmo-streams.mdx b/docs-website/router/cosmo-streams.mdx index bedb84a80b..340d3b703b 100644 --- a/docs-website/router/cosmo-streams.mdx +++ b/docs-website/router/cosmo-streams.mdx @@ -190,12 +190,17 @@ If you run `make edfs-demo` in the Cosmo Monorepo, you'll automatically get a NA By default the router fails to start if an event provider referenced by the execution config is unavailable. This happens in two cases: -- The provider is not defined under `events.providers`. The router reports an error such as `redis provider with ID my-provider is not defined`. -- The provider is defined but cannot be reached at startup (e.g. the broker is down). +- The provider is not defined under `events.providers`. The router reports an error such as `redis provider with ID my-provider is not defined`. This applies to NATS, Kafka, and Redis. +- The provider is defined but cannot be reached at startup (e.g. the broker is down). This applies to NATS and Redis. The Kafka client connects lazily, so a router with an unreachable Kafka broker still starts by default. This stops a router from coming up with a broken event setup. -Set `skip_unavailable_providers: true` (or the `EVENTS_SKIP_UNAVAILABLE_PROVIDERS` environment variable) to start the router anyway. The router logs the error and disables only the fields backed by the unavailable provider. The rest of the graph keeps serving traffic. A field whose provider is not defined is dropped (it cannot be queried); a field whose provider failed to connect returns an error at query time. Set up an alert on the logged error to detect the misconfiguration or outage. +Set `skip_unavailable_providers: true` (or the `EVENTS_SKIP_UNAVAILABLE_PROVIDERS` environment variable) to start the router anyway. The router logs the error and disables only the fields backed by the unavailable provider. The rest of the graph keeps serving traffic. The two cases use distinct log messages and behave differently: + +- A provider that is not defined has its data sources dropped. The affected fields cannot be queried until the provider is added to the configuration. +- A provider that is defined but unreachable keeps a client that retries the connection in the background. Its fields are temporarily unavailable. They recover automatically once the broker becomes reachable, without a router restart. This applies to NATS, Kafka, and Redis. + +Set up an alert on the logged error to detect the misconfiguration or outage. ```yaml config.yaml events: diff --git a/router-tests/events/kafka_skip_unavailable_test.go b/router-tests/events/kafka_skip_unavailable_test.go new file mode 100644 index 0000000000..92420aea7c --- /dev/null +++ b/router-tests/events/kafka_skip_unavailable_test.go @@ -0,0 +1,255 @@ +package events_test + +import ( + "strings" + "testing" + "time" + + "github.com/hasura/go-graphql-client" + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router-tests/events" + "github.com/wundergraph/cosmo/router-tests/testenv" + "github.com/wundergraph/cosmo/router/pkg/config" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +const kafkaBrokerAddr = "127.0.0.1:9092" + +func TestKafkaSkipUnavailableProviders(t *testing.T) { + if testing.Short() { + t.Skip("skipping test in short mode.") + } + + // router start but one or more provider needed is not defined. + t.Run("router starts when a Kafka provider is not defined and the flag is enabled", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate, + EnableKafka: false, + NoRetryClient: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + cfg.Providers.Kafka = nil + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `{ employees { id } }`, + }) + require.Contains(t, res.Body, `"id":1`) + + notDefined := xEnv.Observer().FilterMessage(providerNotDefinedLogMsg) + require.NotZero(t, notDefined.Len()) + require.NotZero(t, notDefined.FilterField(zap.String("provider_type", "kafka")).Len()) + require.Zero(t, xEnv.Observer().FilterMessage(providerCouldNotConnectMsg).Len()) + + // The mutation backed by the undefined provider is unavailable. + errRes, err := xEnv.MakeGraphQLRequest(testenv.GraphQLRequest{ + Query: `mutation { updateEmployeeMyKafka(employeeID: 3, update: {name: "x"}) { success } }`, + }) + require.NoError(t, err) + require.Contains(t, errRes.Body, "errors") + require.NotContains(t, errRes.Body, `"updateEmployeeMyKafka":{"success":true}`) + }) + }) + + // router start but one or more provider are not reachable. + t.Run("router starts when a Kafka broker is unreachable and the flag is enabled", func(t *testing.T) { + t.Parallel() + + proxy := testenv.NewToggleableProxy(t, kafkaBrokerAddr) + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate, + EnableKafka: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + pointKafkaProvidersAt(cfg, []string{proxy.Addr()}) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `{ employees { id } }`, + }) + require.Contains(t, res.Body, `"id":1`) + + require.Eventually(t, func() bool { + return xEnv.Observer().FilterMessage(providerCouldNotConnectMsg). + FilterField(zap.String("provider_type", "kafka")).Len() > 0 + }, EventWaitTimeout, 100*time.Millisecond) + require.Zero(t, xEnv.Observer().FilterMessage(providerNotDefinedLogMsg).Len()) + }) + }) + + // if the provider is not reachable but became reachable, should start working without + // router restarting. + t.Run("Kafka provider recovers without a restart once the broker becomes reachable", func(t *testing.T) { + t.Parallel() + + proxy := testenv.NewToggleableProxy(t, kafkaBrokerAddr) + topics := []string{"employeeUpdated"} + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate, + EnableKafka: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + pointKafkaProvidersAt(cfg, []string{proxy.Addr()}) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + events.KafkaEnsureTopicExists(t, xEnv, EventWaitTimeout, topics...) + + // The broker was unreachable at startup. + require.Eventually(t, func() bool { + return xEnv.Observer().FilterMessage(providerCouldNotConnectMsg). + FilterField(zap.String("provider_type", "kafka")).Len() > 0 + }, EventWaitTimeout, 100*time.Millisecond) + + // The subscription is established while the broker is still down; kgo connects + // lazily and reconnects on its own once it becomes reachable. + var subscriptionOne struct { + employeeUpdatedMyKafka struct { + ID float64 `graphql:"id"` + Details struct { + Forename string `graphql:"forename"` + Surname string `graphql:"surname"` + } `graphql:"details"` + } `graphql:"employeeUpdatedMyKafka(employeeID: 3)"` + } + + client := graphql.NewSubscriptionClient(xEnv.GraphQLWebSocketSubscriptionURL()) + t.Cleanup(func() { _ = client.Close() }) + + subscriptionArgsCh := make(chan kafkaSubscriptionArgs) + _, err := client.Subscribe(&subscriptionOne, nil, func(dataValue []byte, errValue error) error { + subscriptionArgsCh <- kafkaSubscriptionArgs{dataValue: dataValue, errValue: errValue} + return nil + }) + require.NoError(t, err) + + clientRunCh := make(chan error) + go func() { + clientRunCh <- client.Run() + }() + + xEnv.WaitForSubscriptionCount(1, EventWaitTimeout) + xEnv.WaitForTriggerCount(1, EventWaitTimeout) + + // The broker becomes reachable. + proxy.SetReachable(true) + + // Without restarting the router, the subscription starts receiving messages. + xEnv.KafkaPublishUntilReceived(topics[0], `{"__typename":"Employee","id": 1,"update":{"name":"foo"}}`, 1, EventWaitTimeout) + + testenv.AwaitChannelWithT(t, EventWaitTimeout, subscriptionArgsCh, func(t *testing.T, args kafkaSubscriptionArgs) { + require.NoError(t, args.errValue) + require.JSONEq(t, `{"employeeUpdatedMyKafka":{"id":1,"details":{"forename":"Jens","surname":"Neuse"}}}`, string(args.dataValue)) + }) + }) + }) + + // The startup connectivity probe checks the producer (writeClient); this asserts that the + // producer itself recovers and can publish once the broker becomes reachable, without a + // router restart. (The subscription test above exercises a separate consumer client.) + t.Run("Kafka producer recovers without a restart once the broker becomes reachable", func(t *testing.T) { + t.Parallel() + + proxy := testenv.NewToggleableProxy(t, kafkaBrokerAddr) + topics := []string{"employeeUpdated"} + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate, + EnableKafka: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + pointKafkaProvidersAt(cfg, []string{proxy.Addr()}) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + events.KafkaEnsureTopicExists(t, xEnv, EventWaitTimeout, topics...) + + // The broker was unreachable at startup (the producer failed the connectivity probe). + require.Eventually(t, func() bool { + return xEnv.Observer().FilterMessage(providerCouldNotConnectMsg). + FilterField(zap.String("provider_type", "kafka")).Len() > 0 + }, EventWaitTimeout, 100*time.Millisecond) + + // The broker becomes reachable. + proxy.SetReachable(true) + + // Without restarting the router, a publish mutation (which uses the producer that + // failed the startup probe) succeeds again. + require.Eventually(t, func() bool { + res, err := xEnv.MakeGraphQLRequest(testenv.GraphQLRequest{ + Query: `mutation { updateEmployeeMyKafka(employeeID: 3, update: {name: "recovered"}) { success } }`, + }) + return err == nil && strings.Contains(res.Body, `"updateEmployeeMyKafka":{"success":true}`) + }, EventWaitTimeout, 500*time.Millisecond) + }) + }) + + // Unlike NATS and Redis, the Kafka client connects lazily, so in strict mode (flag + // disabled) the router still starts when the broker is unreachable: connectivity is only + // probed under skip_unavailable_providers, which is where the distinct error is logged. + t.Run("router starts with an unreachable Kafka broker in strict mode because Kafka connects lazily", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate, + EnableKafka: false, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + // flag intentionally left disabled; the address is closed so a probe (if any) + // would fail, but kgo connects lazily so the router still starts. + pointKafkaProvidersAt(cfg, []string{"127.0.0.1:1"}) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `{ employees { id } }`, + }) + require.Contains(t, res.Body, `"id":1`) + + // No connectivity probe runs in strict mode, so no "could not connect" log. + require.Zero(t, xEnv.Observer().FilterMessage(providerCouldNotConnectMsg).Len()) + }) + }) +} + +// pointKafkaProvidersAt rewrites every configured Kafka provider to use brokers. +// +// Note on the proxy with Kafka: kgo uses the seed broker only to bootstrap cluster +// metadata, then connects directly to each broker's advertised listener (localhost:9092 +// here). So while the proxy is unreachable, metadata bootstrap fails and the router cannot +// connect (this is what the unreachable/recovery tests rely on); once the proxy is made +// reachable, the metadata bootstrap succeeds and steady-state produce/fetch traffic flows +// directly to the advertised address. The proxy therefore models a broker that is +// unreachable at startup and then becomes reachable, which is exactly the scenario under +// test. It does not model an outage of an already-bootstrapped data connection. +func pointKafkaProvidersAt(cfg *config.EventsConfiguration, brokers []string) { + if len(cfg.Providers.Kafka) == 0 { + for _, sourceName := range testenv.DemoKafkaProviders { + cfg.Providers.Kafka = append(cfg.Providers.Kafka, config.KafkaEventSource{ID: sourceName}) + } + } + for i := range cfg.Providers.Kafka { + cfg.Providers.Kafka[i].Brokers = brokers + } +} diff --git a/router-tests/events/nats_skip_unavailable_test.go b/router-tests/events/nats_skip_unavailable_test.go new file mode 100644 index 0000000000..d320773cc9 --- /dev/null +++ b/router-tests/events/nats_skip_unavailable_test.go @@ -0,0 +1,193 @@ +package events_test + +import ( + "strconv" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router-tests/testenv" + "github.com/wundergraph/cosmo/router/pkg/config" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +// These messages are emitted by the router when events.skip_unavailable_providers is +// enabled. They must stay distinct so operators can tell an undefined provider (a +// configuration mistake) from a provider whose broker is unreachable (an availability +// problem that recovers on its own). +const ( + providerNotDefinedLogMsg = "Event provider referenced by the execution config is not defined; skipping affected data sources, the corresponding fields will be unavailable" + providerCouldNotConnectMsg = "EDFS provider could not be started at startup; the router will keep running and the fields backed by this provider are temporarily unavailable. An unreachable broker reconnects and recovers automatically without a restart; see the error for the cause" +) + +const natsBrokerAddr = "127.0.0.1:4222" + +func TestNatsSkipUnavailableProviders(t *testing.T) { + if testing.Short() { + t.Skip("skipping test in short mode.") + } + + // router start but one or more provider needed is not defined. + t.Run("router starts when a NATS provider is not defined and the flag is enabled", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, + // EnableNats is false: no broker is started and no providers are configured, so + // the NATS data sources reference an undefined provider. + EnableNats: false, + // The skipped field returns an error response; don't let the retry client spin. + NoRetryClient: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + cfg.Providers.Nats = nil + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + // The router started: a non-EDFS query is served normally. + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `{ employees { id } }`, + }) + require.Contains(t, res.Body, `"id":1`) + + // The undefined provider is reported with the "not defined" message and not the + // "could not connect" message. + notDefined := xEnv.Observer().FilterMessage(providerNotDefinedLogMsg) + require.NotZero(t, notDefined.Len()) + require.NotZero(t, notDefined.FilterField(zap.String("provider_type", "nats")).Len()) + require.Zero(t, xEnv.Observer().FilterMessage(providerCouldNotConnectMsg).Len()) + + // The field backed by the undefined provider is unavailable: its data source was + // skipped, so the query cannot be resolved. + errRes, err := xEnv.MakeGraphQLRequest(testenv.GraphQLRequest{ + Query: `query { employeeFromEvent(id: 3) { id } }`, + }) + require.NoError(t, err) + require.Contains(t, errRes.Body, "errors") + require.NotContains(t, errRes.Body, `"employeeFromEvent":{`) + }) + }) + + // router start but one or more provider are not reachable. + t.Run("router starts when a NATS broker is unreachable and the flag is enabled", func(t *testing.T) { + t.Parallel() + + // The proxy stays unreachable for the whole test, simulating a broker that is down. + proxy := testenv.NewToggleableProxy(t, natsBrokerAddr) + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, + EnableNats: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + pointNatsProvidersAt(cfg, "nats://"+proxy.Addr()) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + // The router started despite the unreachable broker and serves normal traffic. + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `{ employees { id } }`, + }) + require.Contains(t, res.Body, `"id":1`) + + // The unreachable broker is reported with the "could not connect" message and + // not the "not defined" message. + require.Eventually(t, func() bool { + return xEnv.Observer().FilterMessage(providerCouldNotConnectMsg). + FilterField(zap.String("provider_type", "nats")).Len() > 0 + }, EventWaitTimeout, 100*time.Millisecond) + require.Zero(t, xEnv.Observer().FilterMessage(providerNotDefinedLogMsg).Len()) + }) + }) + + // if the provider is not reachable but became reachable, should start working without + // router restarting. + t.Run("NATS provider recovers without a restart once the broker becomes reachable", func(t *testing.T) { + t.Parallel() + + proxy := testenv.NewToggleableProxy(t, natsBrokerAddr) + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, + EnableNats: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + pointNatsProvidersAt(cfg, "nats://"+proxy.Addr()) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + // The broker was unreachable at startup. + require.Eventually(t, func() bool { + return xEnv.Observer().FilterMessage(providerCouldNotConnectMsg). + FilterField(zap.String("provider_type", "nats")).Len() > 0 + }, EventWaitTimeout, 100*time.Millisecond) + + // A responder on the real broker answers the EDFS request field. + sub, err := xEnv.NatsConnectionDefault.Subscribe(xEnv.GetPubSubName("getEmployee.3"), func(msg *nats.Msg) { + _ = msg.Respond([]byte(`{"id": 3, "__typename": "Employee"}`)) + }) + require.NoError(t, err) + require.NoError(t, xEnv.NatsConnectionDefault.Flush()) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + // The broker becomes reachable. + proxy.SetReachable(true) + + // Without restarting the router, the EDFS field starts working again as the NATS + // client reconnects in the background. + require.Eventually(t, func() bool { + res, reqErr := xEnv.MakeGraphQLRequest(testenv.GraphQLRequest{ + Query: `query { employeeFromEvent(id: 3) { id } }`, + }) + return reqErr == nil && strings.Contains(res.Body, `"employeeFromEvent":{"id":3}`) + }, EventWaitTimeout, 500*time.Millisecond) + }) + }) + + // strict mode (flag disabled) must still abort startup when a broker is unreachable. + t.Run("router fails to start when a NATS broker is unreachable and the flag is disabled", func(t *testing.T) { + t.Parallel() + + listener := testenv.NewWaitingListener(t, time.Second*10) + listener.Start() + defer listener.Close() + + err := testenv.RunWithError(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate, + EnableNats: false, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + // flag intentionally left disabled + pointNatsProvidersAt(cfg, "nats://127.0.0.1:"+strconv.Itoa(listener.Port())) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + assert.Fail(t, "router should not start when a broker is unreachable in strict mode") + }) + require.Error(t, err) + }) +} + +// pointNatsProvidersAt rewrites every configured NATS provider URL to addr. +func pointNatsProvidersAt(cfg *config.EventsConfiguration, url string) { + if len(cfg.Providers.Nats) == 0 { + // EnableNats populated the demo providers; if a test cleared them, recreate them. + for _, sourceName := range testenv.DemoNatsProviders { + cfg.Providers.Nats = append(cfg.Providers.Nats, config.NatsEventSource{ID: sourceName}) + } + } + for i := range cfg.Providers.Nats { + cfg.Providers.Nats[i].URL = url + } +} diff --git a/router-tests/events/redis_skip_unavailable_test.go b/router-tests/events/redis_skip_unavailable_test.go new file mode 100644 index 0000000000..2bf26d349b --- /dev/null +++ b/router-tests/events/redis_skip_unavailable_test.go @@ -0,0 +1,191 @@ +package events_test + +import ( + "strconv" + "testing" + "time" + + "github.com/hasura/go-graphql-client" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router-tests/testenv" + "github.com/wundergraph/cosmo/router/pkg/config" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +const redisBrokerAddr = "127.0.0.1:6379" + +func TestRedisSkipUnavailableProviders(t *testing.T) { + if testing.Short() { + t.Skip("skipping test in short mode.") + } + + // router start but one or more provider needed is not defined. + t.Run("router starts when a Redis provider is not defined and the flag is enabled", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsRedisJSONTemplate, + EnableRedis: false, + NoRetryClient: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + cfg.Providers.Redis = nil + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `{ employees { id } }`, + }) + require.Contains(t, res.Body, `"id":1`) + + notDefined := xEnv.Observer().FilterMessage(providerNotDefinedLogMsg) + require.NotZero(t, notDefined.Len()) + require.NotZero(t, notDefined.FilterField(zap.String("provider_type", "redis")).Len()) + require.Zero(t, xEnv.Observer().FilterMessage(providerCouldNotConnectMsg).Len()) + + // The mutation backed by the undefined provider is unavailable. + errRes, err := xEnv.MakeGraphQLRequest(testenv.GraphQLRequest{ + Query: `mutation { updateEmployeeMyRedis(id: 3, update: {name: "x"}) { success } }`, + }) + require.NoError(t, err) + require.Contains(t, errRes.Body, "errors") + require.NotContains(t, errRes.Body, `"updateEmployeeMyRedis":{"success":true}`) + }) + }) + + // router start but one or more provider are not reachable. + t.Run("router starts when a Redis broker is unreachable and the flag is enabled", func(t *testing.T) { + t.Parallel() + + proxy := testenv.NewToggleableProxy(t, redisBrokerAddr) + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsRedisJSONTemplate, + EnableRedis: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + pointRedisProvidersAt(cfg, "redis://"+proxy.Addr()) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: `{ employees { id } }`, + }) + require.Contains(t, res.Body, `"id":1`) + + require.Eventually(t, func() bool { + return xEnv.Observer().FilterMessage(providerCouldNotConnectMsg). + FilterField(zap.String("provider_type", "redis")).Len() > 0 + }, EventWaitTimeout, 100*time.Millisecond) + require.Zero(t, xEnv.Observer().FilterMessage(providerNotDefinedLogMsg).Len()) + }) + }) + + // if the provider is not reachable but became reachable, should start working without + // router restarting. + t.Run("Redis provider recovers without a restart once the broker becomes reachable", func(t *testing.T) { + t.Parallel() + + proxy := testenv.NewToggleableProxy(t, redisBrokerAddr) + + testenv.Run(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsRedisJSONTemplate, + EnableRedis: true, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.InfoLevel, + }, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + cfg.SkipUnavailableProviders = true + pointRedisProvidersAt(cfg, "redis://"+proxy.Addr()) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + // The broker was unreachable at startup. + require.Eventually(t, func() bool { + return xEnv.Observer().FilterMessage(providerCouldNotConnectMsg). + FilterField(zap.String("provider_type", "redis")).Len() > 0 + }, EventWaitTimeout, 100*time.Millisecond) + + // The subscription is established while the broker is still down; go-redis + // reconnects on its own once it becomes reachable. + var subscriptionOne struct { + employeeUpdates struct { + ID float64 `graphql:"id"` + Details struct { + Forename string `graphql:"forename"` + Surname string `graphql:"surname"` + } `graphql:"details"` + } `graphql:"employeeUpdates"` + } + + client := graphql.NewSubscriptionClient(xEnv.GraphQLWebSocketSubscriptionURL()) + t.Cleanup(func() { _ = client.Close() }) + + subscriptionArgsCh := make(chan subscriptionArgs) + _, err := client.Subscribe(&subscriptionOne, nil, func(dataValue []byte, errValue error) error { + subscriptionArgsCh <- subscriptionArgs{dataValue, errValue} + return nil + }) + require.NoError(t, err) + + runCh := make(chan error) + go func() { + runCh <- client.Run() + }() + + xEnv.WaitForSubscriptionCount(1, EventWaitTimeout) + xEnv.WaitForTriggerCount(1, EventWaitTimeout) + + // The broker becomes reachable. + proxy.SetReachable(true) + + // Without restarting the router, the subscription starts receiving messages. + xEnv.RedisPublishUntilReceived(`employeeUpdatedMyRedis`, `{"__typename":"Employee","id": 1,"update":{"name":"foo"}}`, EventWaitTimeout) + + testenv.AwaitChannelWithT(t, EventWaitTimeout, subscriptionArgsCh, func(t *testing.T, args subscriptionArgs) { + require.NoError(t, args.errValue) + require.JSONEq(t, `{"employeeUpdates":{"id":1,"details":{"forename":"Jens","surname":"Neuse"}}}`, string(args.dataValue)) + }) + }) + }) + + // strict mode (flag disabled) must still abort startup when a broker is unreachable. + t.Run("router fails to start when a Redis broker is unreachable and the flag is disabled", func(t *testing.T) { + t.Parallel() + + listener := testenv.NewWaitingListener(t, time.Second*10) + listener.Start() + defer listener.Close() + + err := testenv.RunWithError(t, &testenv.Config{ + RouterConfigJSONTemplate: testenv.ConfigWithEdfsRedisJSONTemplate, + EnableRedis: false, + ModifyEventsConfiguration: func(cfg *config.EventsConfiguration) { + pointRedisProvidersAt(cfg, "redis://127.0.0.1:"+strconv.Itoa(listener.Port())) + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + assert.Fail(t, "router should not start when a broker is unreachable in strict mode") + }) + require.Error(t, err) + }) +} + +// pointRedisProvidersAt rewrites every configured Redis provider to use url. +func pointRedisProvidersAt(cfg *config.EventsConfiguration, url string) { + if len(cfg.Providers.Redis) == 0 { + for _, sourceName := range testenv.DemoRedisProviders { + cfg.Providers.Redis = append(cfg.Providers.Redis, config.RedisEventSource{ID: sourceName}) + } + } + for i := range cfg.Providers.Redis { + cfg.Providers.Redis[i].URLs = []string{url} + } +} diff --git a/router-tests/testenv/tcpproxy.go b/router-tests/testenv/tcpproxy.go new file mode 100644 index 0000000000..aa5eec670c --- /dev/null +++ b/router-tests/testenv/tcpproxy.go @@ -0,0 +1,177 @@ +package testenv + +import ( + "fmt" + "io" + "net" + "sync" + "testing" + + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router-tests/freeport" +) + +// ToggleableProxy is a TCP proxy used to simulate an event broker that is unreachable at +// router startup and later becomes reachable, without changing the address the router is +// configured with. +// +// While not reachable, accepted connections are closed immediately, so a client observes +// the broker as down (failed handshake / connection reset). After SetReachable(true) the +// proxy transparently forwards traffic to the real broker, simulating it coming back. This +// lets a test point an EDFS provider at the proxy address, start the router while the +// broker is "down", then bring it "up" and assert the provider recovers without a restart. +type ToggleableProxy struct { + listener net.Listener + target string + + mu sync.Mutex + reachable bool + conns map[net.Conn]struct{} + closed bool + + wg sync.WaitGroup +} + +// NewToggleableProxy starts a proxy that listens on a random loopback port and forwards to +// target (host:port) when reachable. It starts unreachable. The proxy is closed +// automatically when the test finishes. +func NewToggleableProxy(t testing.TB, target string) *ToggleableProxy { + t.Helper() + + listener, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", freeport.GetOne(t))) + require.NoError(t, err) + + p := &ToggleableProxy{ + listener: listener, + target: target, + conns: make(map[net.Conn]struct{}), + } + + p.wg.Go(p.acceptLoop) + + t.Cleanup(func() { + _ = p.Close() + }) + + return p +} + +// Addr returns the host:port the proxy is listening on. +func (p *ToggleableProxy) Addr() string { + return p.listener.Addr().String() +} + +// Port returns the port the proxy is listening on. +func (p *ToggleableProxy) Port() int { + return p.listener.Addr().(*net.TCPAddr).Port +} + +// SetReachable toggles whether the proxy forwards to the target. Setting it to false also +// closes any in-flight forwarded connections so the client observes the broker going away. +func (p *ToggleableProxy) SetReachable(reachable bool) { + p.mu.Lock() + p.reachable = reachable + var toClose []net.Conn + if !reachable { + for c := range p.conns { + toClose = append(toClose, c) + } + } + p.mu.Unlock() + + for _, c := range toClose { + _ = c.Close() + } +} + +func (p *ToggleableProxy) isReachable() bool { + p.mu.Lock() + defer p.mu.Unlock() + return p.reachable +} + +func (p *ToggleableProxy) trackConn(c net.Conn) bool { + p.mu.Lock() + defer p.mu.Unlock() + if p.closed { + return false + } + p.conns[c] = struct{}{} + return true +} + +func (p *ToggleableProxy) untrackConn(c net.Conn) { + p.mu.Lock() + defer p.mu.Unlock() + delete(p.conns, c) +} + +func (p *ToggleableProxy) acceptLoop() { + for { + client, err := p.listener.Accept() + if err != nil { + return + } + if !p.isReachable() { + // Broker is "down": drop the connection so the client sees it as unreachable. + _ = client.Close() + continue + } + p.wg.Go(func() { + p.handle(client) + }) + } +} + +func (p *ToggleableProxy) handle(client net.Conn) { + defer client.Close() + + upstream, err := net.Dial("tcp", p.target) + if err != nil { + return + } + defer upstream.Close() + + if !p.trackConn(client) || !p.trackConn(upstream) { + return + } + defer p.untrackConn(client) + defer p.untrackConn(upstream) + + // Closing either side on copy completion unblocks the other copy goroutine. + var copyWg sync.WaitGroup + copyWg.Add(2) + go func() { + defer copyWg.Done() + _, _ = io.Copy(upstream, client) + _ = upstream.Close() + }() + go func() { + defer copyWg.Done() + _, _ = io.Copy(client, upstream) + _ = client.Close() + }() + copyWg.Wait() +} + +// Close stops the proxy and waits for all goroutines to finish. +func (p *ToggleableProxy) Close() error { + p.mu.Lock() + if p.closed { + p.mu.Unlock() + return nil + } + p.closed = true + var toClose []net.Conn + for c := range p.conns { + toClose = append(toClose, c) + } + p.mu.Unlock() + + err := p.listener.Close() + for _, c := range toClose { + _ = c.Close() + } + p.wg.Wait() + return err +} diff --git a/router/core/graph_server.go b/router/core/graph_server.go index dc6ad9e8e1..ebaace9de7 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -2202,9 +2202,11 @@ func (s *graphServer) startupPubSubProviders(ctx context.Context) error { // Default timeout for pubsub provider startup const defaultStartupTimeout = 5 * time.Second - // When skip_unavailable_providers is enabled, a provider that fails to start (e.g. the - // broker is unreachable) must not prevent the router from starting. The provider is - // logged and marked unavailable so requests to its fields fail gracefully instead. + // When skip_unavailable_providers is enabled, a provider that fails to connect (e.g. the + // broker is unreachable) must not prevent the router from starting. The error is logged + // and the router keeps running; the adapter holds a resilient client that reconnects in + // the background, so the fields backed by that provider recover without a restart once + // the broker becomes reachable again. return s.providersActionWithTimeout(ctx, func(ctx context.Context, provider datasource.Provider) error { return provider.Startup(ctx) }, defaultStartupTimeout, "pubsub provider startup timed out", s.eventsConfig.SkipUnavailableProviders) @@ -2222,18 +2224,12 @@ func (s *graphServer) shutdownPubSubProviders(ctx context.Context) error { }, defaultShutdownTimeout, "pubsub provider shutdown timed out", false) } -// providerUnavailableMarker is implemented by providers that can be marked unavailable -// so that requests to their fields fail gracefully instead of using a provider that -// could not be started. -type providerUnavailableMarker interface { - MarkUnavailable() -} - // providersActionWithTimeout runs action against every pubsub provider concurrently, // bounding each by timeout. When continueOnError is true, a provider whose action fails -// or times out is logged and marked unavailable instead of aborting the whole action; -// this is used at startup so an unreachable provider does not prevent the router from -// starting. +// or times out is logged instead of aborting the whole action; this is used at startup +// (events.skip_unavailable_providers) so an unreachable provider does not prevent the +// router from starting. The provider's adapter keeps a resilient client that reconnects +// in the background, so the affected fields recover once the broker becomes reachable. func (s *graphServer) providersActionWithTimeout(ctx context.Context, action func(ctx context.Context, provider datasource.Provider) error, timeout time.Duration, timeoutMessage string, continueOnError bool) error { cancellableCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -2261,19 +2257,18 @@ func (s *graphServer) providersActionWithTimeout(ctx context.Context, action fun if actionErr == nil || !continueOnError { return actionErr } - // Lenient mode: tolerate the failure only if the provider can be marked - // unavailable, so that requests to its fields fail gracefully instead of using - // an unconnected adapter. If it cannot be marked, do not swallow the error. - marker, ok := provider.(providerUnavailableMarker) - if !ok { - return actionErr - } - s.logger.Error("EDFS provider is unavailable; the router will keep running and the fields backed by this provider will be unavailable", + // Lenient mode (events.skip_unavailable_providers): the provider failed to start. + // Log a distinct error and let the router keep running with the affected fields + // unavailable. When the cause is an unreachable broker the adapter retains a + // resilient client that reconnects in the background, so those fields recover + // without a restart once the broker is reachable. When the cause is a + // configuration error (e.g. an invalid URL) the provider cannot recover until the + // configuration is fixed; the underlying cause is attached as the error field. + s.logger.Error("EDFS provider could not be started at startup; the router will keep running and the fields backed by this provider are temporarily unavailable. An unreachable broker reconnects and recovers automatically without a restart; see the error for the cause", zap.String("provider_id", provider.ID()), zap.String("provider_type", provider.TypeID()), zap.Error(actionErr), ) - marker.MarkUnavailable() return nil }) } diff --git a/router/core/graph_server_test.go b/router/core/graph_server_test.go index 5aed89adc2..dfb9c6ae9b 100644 --- a/router/core/graph_server_test.go +++ b/router/core/graph_server_test.go @@ -17,16 +17,6 @@ import ( "go.uber.org/zap" ) -// unavailableTestSubConfig is a minimal SubscriptionEventConfiguration used to probe -// whether a provider rejects subscriptions after being marked unavailable. -type unavailableTestSubConfig struct{} - -func (unavailableTestSubConfig) ProviderID() string { return "redis-1" } -func (unavailableTestSubConfig) ProviderType() datasource.ProviderType { - return datasource.ProviderTypeRedis -} -func (unavailableTestSubConfig) RootFieldName() string { return "field" } - func TestStartupPubSubProviders_SkipUnavailableProviders(t *testing.T) { t.Parallel() @@ -51,20 +41,18 @@ func TestStartupPubSubProviders_SkipUnavailableProviders(t *testing.T) { require.Error(t, s.startupPubSubProviders(context.Background())) }) - t.Run("starts anyway and marks the provider unavailable when the flag is enabled", func(t *testing.T) { + t.Run("starts anyway when a provider fails and the flag is enabled", func(t *testing.T) { t.Parallel() + // The provider could not connect at startup, but lenient mode lets the router start + // anyway. The adapter keeps a resilient client that reconnects in the background, so + // the provider recovers without a restart once the broker becomes reachable. adapter := datasource.NewMockProvider(t) adapter.On("Startup", mock.Anything).Return(errors.New("connection refused")) provider := datasource.NewPubSubProvider("redis-1", "redis", adapter, zap.NewNop(), nil) s := newServer(true, provider) require.NoError(t, s.startupPubSubProviders(context.Background())) - - // The provider was marked unavailable: a subscribe attempt returns an error - // without ever reaching the (unconnected) adapter, so the router does not panic. - err := provider.Subscribe(context.Background(), unavailableTestSubConfig{}, nil) - require.Error(t, err) }) } @@ -100,10 +88,6 @@ func TestProvidersActionWithTimeout_MultipleHangingProvidersDoNotDeadlock(t *tes case <-time.After(5 * time.Second): t.Fatal("providersActionWithTimeout deadlocked with multiple hanging providers") } - - // Both providers must have been marked unavailable by their own timeouts. - require.Error(t, p1.UnavailableError()) - require.Error(t, p2.UnavailableError()) } // TestShutdownPubSubProviders_StaysStrictWithSkipUnavailableProviders pins that shutdown diff --git a/router/internal/rediscloser/rediscloser.go b/router/internal/rediscloser/rediscloser.go index 2bbc7c3459..032d387400 100644 --- a/router/internal/rediscloser/rediscloser.go +++ b/router/internal/rediscloser/rediscloser.go @@ -21,6 +21,11 @@ type RedisCloserOptions struct { URLs []string ClusterEnabled bool Password string + // Context bounds the initial connectivity check (Ping). When nil, context.Background() + // is used. Callers that must guarantee NewRedisCloser returns within a deadline (e.g. a + // bounded provider startup) should pass a context with a timeout so a black-holed broker + // cannot block the ping for the full go-redis dial timeout. + Context context.Context } func NewRedisCloser(opts *RedisCloserOptions) (RDCloser, error) { @@ -73,7 +78,11 @@ func NewRedisCloser(opts *RedisCloserOptions) (RDCloser, error) { } } - if isFunctioning, err := IsFunctioningClient(rdb); !isFunctioning { + pingCtx := opts.Context + if pingCtx == nil { + pingCtx = context.Background() + } + if isFunctioning, err := IsFunctioningClient(pingCtx, rdb); !isFunctioning { return rdb, fmt.Errorf("failed to create a functioning redis client with the provided URLs: %w", err) } @@ -110,12 +119,12 @@ func addClusterUrlsToQuery(opts *RedisCloserOptions, parsedUrl *url.URL) { parsedUrl.RawQuery = queryVals.Encode() } -func IsFunctioningClient(rdb RDCloser) (bool, error) { +func IsFunctioningClient(ctx context.Context, rdb RDCloser) (bool, error) { if rdb == nil { return false, nil } - res, err := rdb.Ping(context.Background()).Result() + res, err := rdb.Ping(ctx).Result() return err == nil && res == "PONG", err } diff --git a/router/internal/rediscloser/rediscloser_test.go b/router/internal/rediscloser/rediscloser_test.go index 79da15f237..2b47af820e 100644 --- a/router/internal/rediscloser/rediscloser_test.go +++ b/router/internal/rediscloser/rediscloser_test.go @@ -1,6 +1,7 @@ package rediscloser import ( + "context" "fmt" "testing" @@ -31,7 +32,7 @@ func TestRedisCloser(t *testing.T) { require.NoError(t, err) require.NotNil(t, cl) - isFunctioning, err := IsFunctioningClient(cl) + isFunctioning, err := IsFunctioningClient(context.Background(), cl) require.True(t, isFunctioning) require.NoError(t, err) require.False(t, isClusterClient(cl)) @@ -49,7 +50,7 @@ func TestRedisCloser(t *testing.T) { require.NoError(t, err) require.NotNil(t, cl) - isFunctioning, err := IsFunctioningClient(cl) + isFunctioning, err := IsFunctioningClient(context.Background(), cl) require.True(t, isFunctioning) require.NoError(t, err) require.False(t, isClusterClient(cl)) diff --git a/router/pkg/config/config.go b/router/pkg/config/config.go index bcbd772d51..062614d0cc 100644 --- a/router/pkg/config/config.go +++ b/router/pkg/config/config.go @@ -806,8 +806,9 @@ type EventsConfiguration struct { // configuration, or defined but unreachable at startup (e.g. the broker is down). // When enabled, the router logs an error and starts anyway instead of failing. A // provider that is not defined has its data sources skipped; a provider that fails to - // connect is marked unavailable so requests to the affected fields return an error - // instead of crashing the router. The rest of the graph keeps serving traffic. + // connect keeps a resilient client that reconnects in the background, so the affected + // fields are only temporarily unavailable and recover without a restart once the broker + // becomes reachable again. The rest of the graph keeps serving traffic throughout. SkipUnavailableProviders bool `yaml:"skip_unavailable_providers" envDefault:"false" env:"EVENTS_SKIP_UNAVAILABLE_PROVIDERS"` Handlers StreamsHandlerConfiguration `yaml:"handlers,omitempty"` } diff --git a/router/pkg/pubsub/datasource/provider.go b/router/pkg/pubsub/datasource/provider.go index fd02ffccf6..c1e9fea184 100644 --- a/router/pkg/pubsub/datasource/provider.go +++ b/router/pkg/pubsub/datasource/provider.go @@ -113,4 +113,10 @@ type PublishEventConfiguration interface { type ProviderOpts struct { StreamMetricStore metric.StreamMetricStore + // SkipUnavailableProviders mirrors events.skip_unavailable_providers. When true the + // adapter keeps a resilient, auto-reconnecting client even if the broker is unreachable + // at startup. Startup still reports the connection error (so the router can log a + // distinct "could not connect" message), but the provider recovers without a restart + // once the broker becomes reachable again. + SkipUnavailableProviders bool } diff --git a/router/pkg/pubsub/datasource/pubsubprovider.go b/router/pkg/pubsub/datasource/pubsubprovider.go index 6750a2b9a9..257c90c096 100644 --- a/router/pkg/pubsub/datasource/pubsubprovider.go +++ b/router/pkg/pubsub/datasource/pubsubprovider.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "slices" - "sync/atomic" "go.uber.org/zap" "go.uber.org/zap/zapcore" @@ -17,11 +16,6 @@ type PubSubProvider struct { Logger *zap.Logger hooks Hooks eventBuilder EventBuilderFn - // unavailable is set when the provider could not be started (e.g. the broker is - // unreachable) and the router was configured to start anyway. Once set, Subscribe - // and Publish return an error instead of using the underlying adapter, so the - // affected fields fail gracefully rather than crashing the router. - unavailable atomic.Bool } // applyPublishEventHooks processes events through a chain of hook functions @@ -73,31 +67,12 @@ func (p *PubSubProvider) Startup(ctx context.Context) error { return nil } -// MarkUnavailable flags the provider as unavailable. It is called when the provider -// failed to start and the router was configured (events.skip_unavailable_providers) -// to keep running. Subsequent Subscribe/Publish calls return an error instead of -// using the underlying adapter, which may not have an established connection. -func (p *PubSubProvider) MarkUnavailable() { - p.unavailable.Store(true) -} - -// UnavailableError returns a non-nil error if the provider has been marked unavailable. -// Data sources that reach into the underlying adapter directly (e.g. NATS request/reply) -// should call this before using it, so a provider that failed to start is not used. -func (p *PubSubProvider) UnavailableError() error { - if p.unavailable.Load() { - return NewError(fmt.Sprintf("event provider %q (%s) is unavailable", p.id, p.typeID), nil) - } - return nil -} - func (p *PubSubProvider) Shutdown(ctx context.Context) error { - // A provider marked unavailable failed to start, so it has no connection to tear down. - // Skip the adapter: its connection fields may still be written by an in-flight Startup - // goroutine, and reading them here would race that write. - if p.unavailable.Load() { - return nil - } + // The adapter is always torn down, including when the provider could not connect at + // startup under events.skip_unavailable_providers: in that case the adapter still holds + // a resilient client that is reconnecting in the background, and it must be closed to + // avoid leaking that goroutine. Adapters guard their (possibly nil) connection fields, + // so this is safe even if startup never established a connection. if err := p.Adapter.Shutdown(ctx); err != nil { return err } @@ -105,17 +80,10 @@ func (p *PubSubProvider) Shutdown(ctx context.Context) error { } func (p *PubSubProvider) Subscribe(ctx context.Context, cfg SubscriptionEventConfiguration, updater SubscriptionEventUpdater) error { - if err := p.UnavailableError(); err != nil { - return err - } return p.Adapter.Subscribe(ctx, cfg, updater) } func (p *PubSubProvider) Publish(ctx context.Context, cfg PublishEventConfiguration, events []StreamEvent) error { - if err := p.UnavailableError(); err != nil { - return err - } - if len(p.hooks.OnPublishEvents.Handlers) == 0 { return p.Adapter.Publish(ctx, cfg, events) } diff --git a/router/pkg/pubsub/datasource/pubsubprovider_test.go b/router/pkg/pubsub/datasource/pubsubprovider_test.go index 781819693c..b956ab38f0 100644 --- a/router/pkg/pubsub/datasource/pubsubprovider_test.go +++ b/router/pkg/pubsub/datasource/pubsubprovider_test.go @@ -169,53 +169,6 @@ func TestProvider_Subscribe_Error(t *testing.T) { assert.Equal(t, expectedError, err) } -func TestProvider_MarkUnavailable_SubscribeReturnsErrorWithoutCallingAdapter(t *testing.T) { - // The adapter must NOT be reached once the provider is marked unavailable; the mock - // has no expectations set, so any adapter call would fail the test. - mockAdapter := NewMockProvider(t) - config := &testSubscriptionConfig{ - providerID: "test-provider", - providerType: ProviderTypeRedis, - fieldName: "testField", - } - - provider := PubSubProvider{ - id: "test-provider", - typeID: "redis", - Adapter: mockAdapter, - } - provider.MarkUnavailable() - - err := provider.Subscribe(context.Background(), config, nil) - - assert.Error(t, err) - var providerErr *Error - assert.ErrorAs(t, err, &providerErr) -} - -func TestProvider_MarkUnavailable_PublishReturnsErrorWithoutCallingAdapter(t *testing.T) { - mockAdapter := NewMockProvider(t) - config := &testPublishConfig{ - providerID: "test-provider", - providerType: ProviderTypeRedis, - fieldName: "testField", - } - events := []StreamEvent{&testEvent{mutableTestEvent("test data")}} - - provider := PubSubProvider{ - id: "test-provider", - typeID: "redis", - Adapter: mockAdapter, - } - provider.MarkUnavailable() - - err := provider.Publish(context.Background(), config, events) - - assert.Error(t, err) - var providerErr *Error - assert.ErrorAs(t, err, &providerErr) -} - func TestProvider_Publish_NoHooks_Success(t *testing.T) { mockAdapter := NewMockProvider(t) config := &testPublishConfig{ diff --git a/router/pkg/pubsub/kafka/adapter.go b/router/pkg/pubsub/kafka/adapter.go index c90d15555c..43dff34909 100644 --- a/router/pkg/pubsub/kafka/adapter.go +++ b/router/pkg/pubsub/kafka/adapter.go @@ -41,6 +41,10 @@ type ProviderAdapter struct { closeWg sync.WaitGroup cancel context.CancelFunc streamMetricStore metric.StreamMetricStore + // skipUnavailable mirrors events.skip_unavailable_providers. When true, Startup probes + // connectivity (kgo connects lazily otherwise) so an unreachable broker surfaces a + // distinct "could not connect" error, consistent with the NATS and Redis adapters. + skipUnavailable bool } type PollerOpts struct { @@ -289,6 +293,19 @@ func (p *ProviderAdapter) Startup(ctx context.Context) (err error) { return err } + // In lenient mode probe connectivity so an unreachable broker surfaces a distinct + // "could not connect" error at startup, consistent with the NATS and Redis adapters. + // The client is kept regardless: kgo connects lazily and reconnects on its own, so the + // provider recovers without a restart once the broker is reachable. The probe is bounded + // so it completes well within the startup timeout. + if p.skipUnavailable { + pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + if pingErr := p.writeClient.Ping(pingCtx); pingErr != nil { + return datasource.NewError("kafka provider could not connect to the configured brokers; it will keep retrying in the background", pingErr) + } + } + return } @@ -338,6 +355,7 @@ func NewProviderAdapter(ctx context.Context, logger *zap.Logger, opts []kgo.Opt, closeWg: sync.WaitGroup{}, cancel: cancel, streamMetricStore: store, + skipUnavailable: providerOpts.SkipUnavailableProviders, }, nil } diff --git a/router/pkg/pubsub/nats/adapter.go b/router/pkg/pubsub/nats/adapter.go index ac3dd70fb2..a8ba5c3c7e 100644 --- a/router/pkg/pubsub/nats/adapter.go +++ b/router/pkg/pubsub/nats/adapter.go @@ -52,6 +52,11 @@ type ProviderAdapter struct { flushTimeout time.Duration streamMetricStore metric.StreamMetricStore consumerConfig consumerConfig + // skipUnavailable mirrors events.skip_unavailable_providers. When true, the connection + // options enable retry-on-failed-connect with unlimited reconnects, and Startup reports + // a connection error without aborting so the router can keep running while the client + // reconnects in the background. + skipUnavailable bool } // getInstanceIdentifier returns an identifier for the current instance. @@ -370,12 +375,22 @@ func (p *ProviderAdapter) flush(ctx context.Context) error { func (p *ProviderAdapter) Startup(ctx context.Context) (err error) { p.client, err = nats.Connect(p.url, p.opts...) if err != nil { + // In lenient mode the connection options enable retry-on-failed-connect, so a + // connection failure does not surface here; any error returned is a fatal + // misconfiguration (e.g. invalid URL/TLS) that cannot recover by retrying. return err } p.js, err = jetstream.New(p.client) if err != nil { return err } + // With retry-on-failed-connect (lenient mode) nats.Connect returns a client in the + // reconnecting state instead of an error when the broker is unreachable. Report the + // connection failure so the caller can log it, but keep the client: it reconnects in + // the background and the provider recovers without a restart once the broker is back. + if p.skipUnavailable && !p.client.IsConnected() { + return datasource.NewError(fmt.Sprintf("nats provider could not connect to %q; it will keep retrying in the background", p.url), nil) + } return nil } @@ -394,14 +409,19 @@ func (p *ProviderAdapter) Shutdown(ctx context.Context) error { p.deleteDurableConsumers(ctx) } - fErr := p.flush(ctx) - if fErr != nil { - shutdownErr = errors.Join(shutdownErr, fErr) - } + // Only flush and drain an established connection. Under skip_unavailable_providers the + // client may have never connected (it is still reconnecting in the background), in which + // case there is nothing to flush or drain and attempting it only returns spurious errors. + if p.client.IsConnected() { + fErr := p.flush(ctx) + if fErr != nil { + shutdownErr = errors.Join(shutdownErr, fErr) + } - drainErr := p.client.Drain() - if drainErr != nil { - shutdownErr = errors.Join(shutdownErr, drainErr) + drainErr := p.client.Drain() + if drainErr != nil { + shutdownErr = errors.Join(shutdownErr, drainErr) + } } // Close the client @@ -513,6 +533,7 @@ func NewAdapter(ctx context.Context, logger *zap.Logger, url string, opts []nats consumerConfig: consumerConfig{ deleteOnShutdown: deleteConsumersOnShutdown, }, + skipUnavailable: providerOpts.SkipUnavailableProviders, }, nil } diff --git a/router/pkg/pubsub/nats/engine_datasource.go b/router/pkg/pubsub/nats/engine_datasource.go index d2a0d7c3f3..6739c418ba 100644 --- a/router/pkg/pubsub/nats/engine_datasource.go +++ b/router/pkg/pubsub/nats/engine_datasource.go @@ -247,12 +247,6 @@ func (s *NatsRequestDataSource) Load(ctx context.Context, headers http.Header, i return nil, fmt.Errorf("adapter for provider %s is not of the right type", publishData.Provider) } - // The request path reaches into the underlying adapter directly, so it must honor the - // unavailable flag itself (Subscribe/Publish are guarded by the provider wrapper). - if err := providerBase.UnavailableError(); err != nil { - return nil, err - } - adapter, ok := providerBase.Adapter.(Adapter) if !ok { return nil, fmt.Errorf("adapter for provider %s is not of the right type", publishData.Provider) diff --git a/router/pkg/pubsub/nats/engine_datasource_test.go b/router/pkg/pubsub/nats/engine_datasource_test.go index 6914bdf2db..bb38a1ce66 100644 --- a/router/pkg/pubsub/nats/engine_datasource_test.go +++ b/router/pkg/pubsub/nats/engine_datasource_test.go @@ -210,24 +210,6 @@ func TestNatsRequestDataSource_Load(t *testing.T) { } } -func TestNatsRequestDataSource_Load_UnavailableProvider(t *testing.T) { - t.Parallel() - - // No Request expectation is set on the mock, so the request path must not reach the - // adapter once the provider is marked unavailable. - mockAdapter := NewMockAdapter(t) - provider := datasource.NewPubSubProvider("test-provider", "nats", mockAdapter, zap.NewNop(), testNatsEventBuilder) - provider.MarkUnavailable() - - dataSource := &NatsRequestDataSource{ - pubSub: provider, - } - - input := []byte(`{"subject":"test-subject", "event": {"data":{"message":"hello"}}, "providerId":"test-provider"}`) - _, err := dataSource.Load(context.Background(), nil, input) - require.Error(t, err) -} - func TestNatsRequestDataSource_LoadWithFiles(t *testing.T) { dataSource := &NatsRequestDataSource{} assert.Panics(t, func() { diff --git a/router/pkg/pubsub/nats/provider_builder.go b/router/pkg/pubsub/nats/provider_builder.go index 1b6267a612..96356e686a 100644 --- a/router/pkg/pubsub/nats/provider_builder.go +++ b/router/pkg/pubsub/nats/provider_builder.go @@ -76,7 +76,7 @@ func (p *ProviderBuilder) BuildProvider(provider config.NatsEventSource, provide return pubSubProvider, nil } -func buildNatsOptions(eventSource config.NatsEventSource, logger *zap.Logger) ([]nats.Option, error) { +func buildNatsOptions(eventSource config.NatsEventSource, logger *zap.Logger, skipUnavailable bool) ([]nats.Option, error) { opts := []nats.Option{ nats.Name(fmt.Sprintf("cosmo.router.edfs.nats.%s", eventSource.ID)), nats.ReconnectJitter(500*time.Millisecond, 2*time.Second), @@ -152,11 +152,22 @@ func buildNatsOptions(eventSource config.NatsEventSource, logger *zap.Logger) ([ opts = append(opts, nats.Secure(tlsCfg)) } + if skipUnavailable { + // Lenient mode (events.skip_unavailable_providers): do not fail the initial connect + // when the broker is unreachable. nats.Connect then returns a client in the + // reconnecting state instead of an error, and unlimited reconnects ensure the + // provider recovers on its own once the broker becomes reachable — no restart needed. + opts = append(opts, + nats.RetryOnFailedConnect(true), + nats.MaxReconnects(-1), + ) + } + return opts, nil } func buildProvider(ctx context.Context, provider config.NatsEventSource, logger *zap.Logger, hostName string, routerListenAddr string, providerOpts datasource.ProviderOpts) (datasource.Provider, error) { - options, err := buildNatsOptions(provider, logger) + options, err := buildNatsOptions(provider, logger, providerOpts.SkipUnavailableProviders) if err != nil { return nil, fmt.Errorf("failed to build options for Nats provider with ID \"%s\": %w", provider.ID, err) } diff --git a/router/pkg/pubsub/nats/provider_builder_test.go b/router/pkg/pubsub/nats/provider_builder_test.go index b6d51f8cfe..e5ee2bf8cf 100644 --- a/router/pkg/pubsub/nats/provider_builder_test.go +++ b/router/pkg/pubsub/nats/provider_builder_test.go @@ -38,11 +38,33 @@ func TestBuildNatsOptions(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) require.NotEmpty(t, opts) }) + t.Run("skip unavailable providers enables retry on failed connect with unlimited reconnects", func(t *testing.T) { + cfg := config.NatsEventSource{ + ID: "test-nats", + URL: "nats://localhost:4222", + } + logger := zaptest.NewLogger(t) + + // Strict mode must keep the default fail-fast behavior on the initial connect. + strict, err := buildNatsOptions(cfg, logger, false) + require.NoError(t, err) + strictOpts := applyNatsOptions(t, strict) + require.False(t, strictOpts.RetryOnFailedConnect) + + // Lenient mode retries the initial connect and reconnects forever so the provider + // recovers without a restart once the broker becomes reachable. + lenient, err := buildNatsOptions(cfg, logger, true) + require.NoError(t, err) + lenientOpts := applyNatsOptions(t, lenient) + require.True(t, lenientOpts.RetryOnFailedConnect) + require.Equal(t, -1, lenientOpts.MaxReconnect) + }) + t.Run("with token authentication", func(t *testing.T) { token := "test-token" cfg := config.NatsEventSource{ @@ -56,7 +78,7 @@ func TestBuildNatsOptions(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) require.NotEmpty(t, opts) // Can't directly check for token options, but we can verify options are present @@ -78,7 +100,7 @@ func TestBuildNatsOptions(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) require.NotEmpty(t, opts) // Can't directly check for auth options, but we can verify options are present @@ -95,7 +117,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) natsOpts := applyNatsOptions(t, opts) @@ -116,7 +138,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) natsOpts := applyNatsOptions(t, opts) @@ -135,7 +157,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - _, err := buildNatsOptions(cfg, logger) + _, err := buildNatsOptions(cfg, logger, false) require.ErrorContains(t, err, "failed to read CA file") }) @@ -149,7 +171,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - _, err := buildNatsOptions(cfg, logger) + _, err := buildNatsOptions(cfg, logger, false) require.ErrorContains(t, err, "both cert_file and key_file must be provided") }) @@ -163,7 +185,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - _, err := buildNatsOptions(cfg, logger) + _, err := buildNatsOptions(cfg, logger, false) require.ErrorContains(t, err, "both cert_file and key_file must be provided") }) @@ -179,7 +201,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) natsOpts := applyNatsOptions(t, opts) @@ -200,7 +222,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) natsOpts := applyNatsOptions(t, opts) @@ -224,7 +246,7 @@ func TestBuildNatsOptionsWithTLS(t *testing.T) { } logger := zaptest.NewLogger(t) - opts, err := buildNatsOptions(cfg, logger) + opts, err := buildNatsOptions(cfg, logger, false) require.NoError(t, err) natsOpts := applyNatsOptions(t, opts) diff --git a/router/pkg/pubsub/pubsub.go b/router/pkg/pubsub/pubsub.go index ffb4a50adb..47249940fc 100644 --- a/router/pkg/pubsub/pubsub.go +++ b/router/pkg/pubsub/pubsub.go @@ -162,7 +162,8 @@ func build[P GetID, E GetEngineEventConfiguration]( continue } provider, err := builder.BuildProvider(providerData, pubsub_datasource.ProviderOpts{ - StreamMetricStore: store, + StreamMetricStore: store, + SkipUnavailableProviders: skipUnavailableProviders, }) if err != nil { return nil, nil, err diff --git a/router/pkg/pubsub/redis/adapter.go b/router/pkg/pubsub/redis/adapter.go index 0c54393cdd..606a473e96 100644 --- a/router/pkg/pubsub/redis/adapter.go +++ b/router/pkg/pubsub/redis/adapter.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sync" + "time" "github.com/wundergraph/cosmo/router/pkg/metric" @@ -41,6 +42,7 @@ func NewProviderAdapter(ctx context.Context, logger *zap.Logger, urls []string, urls: urls, clusterEnabled: clusterEnabled, streamMetricStore: store, + skipUnavailable: opts.SkipUnavailableProviders, } } @@ -53,15 +55,37 @@ type ProviderAdapter struct { urls []string clusterEnabled bool streamMetricStore metric.StreamMetricStore + // skipUnavailable mirrors events.skip_unavailable_providers. When true, Startup keeps + // the resilient client even if the initial connection check fails, so go-redis can + // reconnect on a later command and the provider recovers without a restart. + skipUnavailable bool } func (p *ProviderAdapter) Startup(ctx context.Context) error { + // Bound the initial connectivity check so a black-holed broker (SYN dropped) cannot block + // Startup for the full go-redis dial timeout, which could exceed the caller's startup + // timeout and leave this call running in a leaked goroutine. + pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() rdCloser, err := rd.NewRedisCloser(&rd.RedisCloserOptions{ Logger: p.logger, URLs: p.urls, ClusterEnabled: p.clusterEnabled, + Context: pingCtx, }) if err != nil { + if p.skipUnavailable && rdCloser != nil { + // Lenient mode: retain the client even though the initial connection check + // failed. go-redis is lazy and reconnects on the next command, so the provider + // recovers without a restart once the broker is reachable. The error is still + // returned so the caller can log a distinct "could not connect" message. + p.conn = rdCloser + return err + } + // Strict mode: the router will abort startup, so close the client we won't keep. + if rdCloser != nil { + _ = rdCloser.Close() + } return err } @@ -97,9 +121,9 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri zap.Strings("channels", subConf.Channels), ) - // Defensive backstop behind PubSubProvider.UnavailableError(): a provider that failed - // to connect is normally short-circuited before reaching the adapter, but guard the - // nil connection here too so an unstarted adapter returns an error instead of panicking. + // Guard the possibly-nil connection: in strict mode a failed Startup leaves p.conn nil + // (under skip_unavailable_providers the resilient client is retained instead), so return + // an error rather than panicking if Subscribe is somehow reached without a connection. if p.conn == nil { return datasource.NewError("redis connection not initialized", nil) } From fccef7e81e1833bb93d9328c9651f71424ac90bd Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Mon, 29 Jun 2026 17:26:27 +0200 Subject: [PATCH 5/7] chore: format --- router/core/router.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/router/core/router.go b/router/core/router.go index 5dd087acd1..dd2508e558 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -1553,7 +1553,7 @@ func (r *Router) Start(ctx context.Context) error { } /** - * Server logging after features has been initialized / disabled + * Server logging after features has been initialized / disabled */ if r.localhostFallbackInsideDocker && docker.Inside() { From e56155d536dc38360ff6231ba8a74dba544ede87 Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Tue, 30 Jun 2026 11:45:56 +0200 Subject: [PATCH 6/7] fix: tcpproxy was mismanaging some edge cases --- router-tests/testenv/tcpproxy.go | 20 +++++- router-tests/testenv/tcpproxy_test.go | 89 +++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 3 deletions(-) create mode 100644 router-tests/testenv/tcpproxy_test.go diff --git a/router-tests/testenv/tcpproxy.go b/router-tests/testenv/tcpproxy.go index aa5eec670c..533c9bc256 100644 --- a/router-tests/testenv/tcpproxy.go +++ b/router-tests/testenv/tcpproxy.go @@ -90,10 +90,16 @@ func (p *ToggleableProxy) isReachable() bool { return p.reachable } +// trackConn registers c so SetReachable(false) and Close can tear it down, but only while +// the proxy is open and reachable. Rejecting registration when p.reachable is false, under +// the same lock SetReachable uses, closes the race where a connection accepted just before +// SetReachable(false) is added to p.conns after SetReachable has already snapshot-closed the +// set: such a connection is never tracked, so handle closes it and bails instead of +// forwarding. It returns false if the connection must not be used. func (p *ToggleableProxy) trackConn(c net.Conn) bool { p.mu.Lock() defer p.mu.Unlock() - if p.closed { + if p.closed || !p.reachable { return false } p.conns[c] = struct{}{} @@ -126,16 +132,24 @@ func (p *ToggleableProxy) acceptLoop() { func (p *ToggleableProxy) handle(client net.Conn) { defer client.Close() + // Register the client before doing any work. If the proxy is (or becomes) unreachable, + // trackConn rejects it and the connection is closed without ever forwarding, so a + // connection accepted just before SetReachable(false) cannot leak through the toggle. + if !p.trackConn(client) { + return + } + defer p.untrackConn(client) + upstream, err := net.Dial("tcp", p.target) if err != nil { return } defer upstream.Close() - if !p.trackConn(client) || !p.trackConn(upstream) { + // Re-check reachability after the dial: SetReachable(false) may have run while dialing. + if !p.trackConn(upstream) { return } - defer p.untrackConn(client) defer p.untrackConn(upstream) // Closing either side on copy completion unblocks the other copy goroutine. diff --git a/router-tests/testenv/tcpproxy_test.go b/router-tests/testenv/tcpproxy_test.go new file mode 100644 index 0000000000..bb09eb9c42 --- /dev/null +++ b/router-tests/testenv/tcpproxy_test.go @@ -0,0 +1,89 @@ +package testenv + +import ( + "io" + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// startEchoServer starts an in-process TCP server that echoes whatever it receives, used as +// the upstream "broker" for the proxy tests. It is torn down when the test finishes. +func startEchoServer(t *testing.T) string { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + + go func() { + for { + conn, acceptErr := ln.Accept() + if acceptErr != nil { + return + } + go func(c net.Conn) { + defer c.Close() + _, _ = io.Copy(c, c) + }(conn) + } + }() + + return ln.Addr().String() +} + +// proxyRoundTrip dials addr, sends a probe, and returns what it reads back. A dropped +// (unreachable) connection yields an error instead of the echoed probe. +func proxyRoundTrip(addr string) (string, error) { + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + return "", err + } + defer conn.Close() + + if err = conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + return "", err + } + if _, err = conn.Write([]byte("ping")); err != nil { + return "", err + } + buf := make([]byte, 4) + if _, err = io.ReadFull(conn, buf); err != nil { + return "", err + } + return string(buf), nil +} + +func TestToggleableProxy(t *testing.T) { + t.Parallel() + + echoAddr := startEchoServer(t) + proxy := NewToggleableProxy(t, echoAddr) + + // It starts unreachable: connections are accepted then dropped, so no echo comes back. + _, err := proxyRoundTrip(proxy.Addr()) + require.Error(t, err) + + // Once reachable, traffic is forwarded to the upstream and echoed back. + proxy.SetReachable(true) + require.Eventually(t, func() bool { + got, rtErr := proxyRoundTrip(proxy.Addr()) + return rtErr == nil && got == "ping" + }, 5*time.Second, 50*time.Millisecond) + + // Toggled back to unreachable: new connections must not forward anymore. + proxy.SetReachable(false) + require.Eventually(t, func() bool { + _, rtErr := proxyRoundTrip(proxy.Addr()) + return rtErr != nil + }, 5*time.Second, 50*time.Millisecond) + + // Reachable again: it recovers without recreating the proxy. + proxy.SetReachable(true) + require.Eventually(t, func() bool { + got, rtErr := proxyRoundTrip(proxy.Addr()) + return rtErr == nil && got == "ping" + }, 5*time.Second, 50*time.Millisecond) +} From ca6bb15277cf72f844bb0acaaa5383b677e11d3a Mon Sep 17 00:00:00 2001 From: Alessandro Pagnin Date: Tue, 30 Jun 2026 11:54:22 +0200 Subject: [PATCH 7/7] fix: two tests were not checking on Run return value --- router-tests/events/kafka_skip_unavailable_test.go | 11 +++++++++-- router-tests/events/redis_skip_unavailable_test.go | 11 +++++++++-- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/router-tests/events/kafka_skip_unavailable_test.go b/router-tests/events/kafka_skip_unavailable_test.go index 92420aea7c..64c1ef0b86 100644 --- a/router-tests/events/kafka_skip_unavailable_test.go +++ b/router-tests/events/kafka_skip_unavailable_test.go @@ -130,7 +130,6 @@ func TestKafkaSkipUnavailableProviders(t *testing.T) { } client := graphql.NewSubscriptionClient(xEnv.GraphQLWebSocketSubscriptionURL()) - t.Cleanup(func() { _ = client.Close() }) subscriptionArgsCh := make(chan kafkaSubscriptionArgs) _, err := client.Subscribe(&subscriptionOne, nil, func(dataValue []byte, errValue error) error { @@ -139,7 +138,9 @@ func TestKafkaSkipUnavailableProviders(t *testing.T) { }) require.NoError(t, err) - clientRunCh := make(chan error) + // Buffered so the goroutine never blocks delivering Run()'s result, even if the + // drain below times out before reading it. + clientRunCh := make(chan error, 1) go func() { clientRunCh <- client.Run() }() @@ -157,6 +158,12 @@ func TestKafkaSkipUnavailableProviders(t *testing.T) { require.NoError(t, args.errValue) require.JSONEq(t, `{"employeeUpdatedMyKafka":{"id":1,"details":{"forename":"Jens","surname":"Neuse"}}}`, string(args.dataValue)) }) + + // Close the client and verify Run() returned cleanly. + require.NoError(t, client.Close()) + testenv.AwaitChannelWithT(t, EventWaitTimeout, clientRunCh, func(t *testing.T, runErr error) { + require.NoError(t, runErr) + }, "unable to close client before timeout") }) }) diff --git a/router-tests/events/redis_skip_unavailable_test.go b/router-tests/events/redis_skip_unavailable_test.go index 2bf26d349b..68275e0eea 100644 --- a/router-tests/events/redis_skip_unavailable_test.go +++ b/router-tests/events/redis_skip_unavailable_test.go @@ -127,7 +127,6 @@ func TestRedisSkipUnavailableProviders(t *testing.T) { } client := graphql.NewSubscriptionClient(xEnv.GraphQLWebSocketSubscriptionURL()) - t.Cleanup(func() { _ = client.Close() }) subscriptionArgsCh := make(chan subscriptionArgs) _, err := client.Subscribe(&subscriptionOne, nil, func(dataValue []byte, errValue error) error { @@ -136,7 +135,9 @@ func TestRedisSkipUnavailableProviders(t *testing.T) { }) require.NoError(t, err) - runCh := make(chan error) + // Buffered so the goroutine never blocks delivering Run()'s result, even if the + // drain below times out before reading it. + runCh := make(chan error, 1) go func() { runCh <- client.Run() }() @@ -154,6 +155,12 @@ func TestRedisSkipUnavailableProviders(t *testing.T) { require.NoError(t, args.errValue) require.JSONEq(t, `{"employeeUpdates":{"id":1,"details":{"forename":"Jens","surname":"Neuse"}}}`, string(args.dataValue)) }) + + // Close the client and verify Run() returned cleanly. + require.NoError(t, client.Close()) + testenv.AwaitChannelWithT(t, EventWaitTimeout, runCh, func(t *testing.T, runErr error) { + require.NoError(t, runErr) + }, "unable to close client before timeout") }) })