diff --git a/docs-website/router/configuration.mdx b/docs-website/router/configuration.mdx index 715d853953..391e361cdd 100644 --- a/docs-website/router/configuration.mdx +++ b/docs-website/router/configuration.mdx @@ -1832,6 +1832,7 @@ We support NATS, Kafka and Redis as event bus provider. version: "1" events: + skip_unavailable_providers: false providers: nats: - id: default @@ -1858,9 +1859,15 @@ events: redis: - id: my-redis urls: - - "localhost:9092" + - "redis://localhost:6379" ``` +### Options + +| 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. 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 | 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..340d3b703b 100644 --- a/docs-website/router/cosmo-streams.mdx +++ b/docs-website/router/cosmo-streams.mdx @@ -186,6 +186,31 @@ 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. +### Unavailable providers + +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`. 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. 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: + skip_unavailable_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-tests/events/kafka_skip_unavailable_test.go b/router-tests/events/kafka_skip_unavailable_test.go new file mode 100644 index 0000000000..64c1ef0b86 --- /dev/null +++ b/router-tests/events/kafka_skip_unavailable_test.go @@ -0,0 +1,262 @@ +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()) + + 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) + + // 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() + }() + + 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)) + }) + + // 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") + }) + }) + + // 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..68275e0eea --- /dev/null +++ b/router-tests/events/redis_skip_unavailable_test.go @@ -0,0 +1,198 @@ +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()) + + 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) + + // 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() + }() + + 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)) + }) + + // 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") + }) + }) + + // 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..533c9bc256 --- /dev/null +++ b/router-tests/testenv/tcpproxy.go @@ -0,0 +1,191 @@ +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 +} + +// 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 || !p.reachable { + 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() + + // 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() + + // Re-check reachability after the dial: SetReachable(false) may have run while dialing. + if !p.trackConn(upstream) { + return + } + 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-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) +} diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 37cdf2dabc..66e2ca7232 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -2310,16 +2310,39 @@ func providersActionWithTimeout(ctx context.Context, providers []datasource.Prov providersGroup := new(errgroup.Group) for _, provider := range providers { 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(timeoutCtx, provider) }() + var actionErr error select { case err := <-actionDone: return err case <-timeoutCtx.Done(): return errors.New(timeoutMessage) } + if actionErr == nil || !continueOnError { + return actionErr + } + // 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), + ) + return nil }) } diff --git a/router/core/graph_server_test.go b/router/core/graph_server_test.go index 066a17e187..9d7a99d75a 100644 --- a/router/core/graph_server_test.go +++ b/router/core/graph_server_test.go @@ -20,6 +20,99 @@ import ( "github.com/wundergraph/cosmo/router/pkg/routerconfig" ) +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 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())) + }) +} + +// 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") + } +} + +// 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/core/router.go b/router/core/router.go index 49b1dc0fbb..7f1e05e808 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -1561,7 +1561,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() { 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 a53fd35f22..02ad9bc2b7 100644 --- a/router/pkg/config/config.go +++ b/router/pkg/config/config.go @@ -809,8 +809,17 @@ type EventProviders struct { } type EventsConfiguration struct { - Providers EventProviders `yaml:"providers,omitempty"` - Handlers StreamsHandlerConfiguration `yaml:"handlers,omitempty"` + Providers EventProviders `yaml:"providers,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 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"` } type StreamsHandlerConfiguration struct { diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json index 7f7cf2b897..9bcd8b3b98 100644 --- a/router/pkg/config/config.schema.json +++ b/router/pkg/config/config.schema.json @@ -2939,6 +2939,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_unavailable_providers": { + "type": "boolean", + "default": false, + "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", "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 bc047d8fba..e53905ff99 100644 --- a/router/pkg/config/config_test.go +++ b/router/pkg/config/config_test.go @@ -177,6 +177,56 @@ engine: }) } +func TestEventsSkipUnavailableProvidersConfigLoading(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.SkipUnavailableProviders) + }) + + t.Run("can be enabled from yaml", func(t *testing.T) { + t.Parallel() + + f := createTempFileFromFixture(t, ` +version: "1" + +graph: + token: "token" + +events: + skip_unavailable_providers: true +`) + + cfg, err := LoadConfig([]string{f}) + require.NoError(t, err) + require.True(t, cfg.Config.Events.SkipUnavailableProviders) + }) + + t.Run("can be enabled from env", func(t *testing.T) { + t.Setenv("EVENTS_SKIP_UNAVAILABLE_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.SkipUnavailableProviders) + }) +} + // 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 2847a043a2..31f7889fc9 100644 --- a/router/pkg/config/fixtures/full.yaml +++ b/router/pkg/config/fixtures/full.yaml @@ -366,6 +366,7 @@ cdn: cache_size: 100MB events: + 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 903baa297d..bc74bbf801 100644 --- a/router/pkg/config/testdata/config_defaults.json +++ b/router/pkg/config/testdata/config_defaults.json @@ -385,6 +385,7 @@ "Kafka": null, "Redis": null }, + "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 727d2d7230..e5f9a96f83 100644 --- a/router/pkg/config/testdata/config_full.json +++ b/router/pkg/config/testdata/config_full.json @@ -795,6 +795,7 @@ } ] }, + "SkipUnavailableProviders": true, "Handlers": { "OnReceiveEvents": { "MaxConcurrentHandlers": 100, 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 74ff9ff9bb..257c90c096 100644 --- a/router/pkg/pubsub/datasource/pubsubprovider.go +++ b/router/pkg/pubsub/datasource/pubsubprovider.go @@ -68,6 +68,11 @@ func (p *PubSubProvider) Startup(ctx context.Context) error { } func (p *PubSubProvider) Shutdown(ctx context.Context) error { + // 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 } 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/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 19c908712e..47249940fc 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.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 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.SkipUnavailableProviders) 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.SkipUnavailableProviders) 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.SkipUnavailableProviders) 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, + skipUnavailableProviders bool, ) (map[string]pubsub_datasource.Provider, []plan.DataSource, error) { pubSubProviders := make(map[string]pubsub_datasource.Provider) var outs []plan.DataSource @@ -153,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 @@ -163,18 +173,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 !skipUnavailableProviders { + 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 skipUnavailableProviders 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..01369830c2 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_SkipUnavailableProviders(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.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) + 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() diff --git a/router/pkg/pubsub/redis/adapter.go b/router/pkg/pubsub/redis/adapter.go index 69c55d7bea..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 } @@ -96,6 +120,14 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri zap.String("method", "subscribe"), zap.Strings("channels", subConf.Channels), ) + + // 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) + } + 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()