diff --git a/router/core/executor.go b/router/core/executor.go index ae72771f96..69e4b02cf7 100644 --- a/router/core/executor.go +++ b/router/core/executor.go @@ -8,6 +8,7 @@ import ( "go.uber.org/zap" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" 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/grpcconnector" @@ -51,6 +52,19 @@ type Executor struct { TrackUsageInfo bool } +// Close releases schema and planner references held by the executor so a replaced +// graph mux can be garbage-collected after config reload. +func (e *Executor) Close() { + if e == nil { + return + } + e.ClientSchema = nil + e.RouterSchema = nil + e.PlanConfig = plan.Configuration{} + e.RenameTypeNames = nil + e.Resolver = nil +} + type ExecutorBuildOptions struct { EngineConfig *nodev1.EngineConfiguration Subgraphs []*nodev1.Subgraph @@ -62,10 +76,11 @@ type ExecutorBuildOptions struct { TraceClientRequired bool PluginsEnabled bool InstanceData InstanceData + WebSocketConfiguration *config.WebSocketConfiguration } func (b *ExecutorConfigurationBuilder) Build(ctx context.Context, opts *ExecutorBuildOptions) (*Executor, []pubsub_datasource.Provider, error) { - planConfig, providers, err := b.buildPlannerConfiguration(ctx, opts.EngineConfig, opts.Subgraphs, opts.RouterEngineConfig, opts.PluginsEnabled) + planConfig, providers, err := b.buildPlannerConfiguration(ctx, opts) if err != nil { return nil, nil, fmt.Errorf("failed to build planner configuration: %w", err) } @@ -215,29 +230,43 @@ func (b *ExecutorConfigurationBuilder) Build(ctx context.Context, opts *Executor }, providers, nil } -func (b *ExecutorConfigurationBuilder) buildPlannerConfiguration(ctx context.Context, engineConfig *nodev1.EngineConfiguration, subgraphs []*nodev1.Subgraph, routerEngineCfg *RouterEngineConfiguration, pluginsEnabled bool) (*plan.Configuration, []pubsub_datasource.Provider, error) { +func (b *ExecutorConfigurationBuilder) buildPlannerConfiguration(ctx context.Context, opts *ExecutorBuildOptions) (*plan.Configuration, []pubsub_datasource.Provider, error) { // this loader is used to take the engine config and create a plan config // the plan config is what the engine uses to turn a GraphQL Request into an execution plan // the plan config is stateful as it carries connection pools and other things + subscriptionClientOptions := b.subscriptionClientOptions + if subscriptionClientOptions == nil { + subscriptionClientOptions = &SubscriptionClientOptions{} + } + resolvedSubscriptionClientOptions := *subscriptionClientOptions + if mondaytweaks.UseNoopUpstreamSubscriptionClientWhenUnused { + resolvedSubscriptionClientOptions.UseNoopClient = shouldUseNoopUpstreamSubscriptionClient( + opts.EngineConfig.GetGraphqlSchema(), + opts.EngineConfig, + opts.RouterEngineConfig.Events, + opts.WebSocketConfiguration, + ) + } + loader := NewLoader(ctx, b.trackUsageInfo, NewDefaultFactoryResolver( ctx, b.transportOptions, - b.subscriptionClientOptions, + &resolvedSubscriptionClientOptions, b.baseTripper, b.subgraphTrippers, b.pluginHost, b.logger, - routerEngineCfg.Execution.EnableNetPoll, + opts.RouterEngineConfig.Execution.EnableNetPoll, b.instanceData, ), b.logger, b.subscriptionHooks) // this generates the plan config using the data source factories from the config package - planConfig, providers, err := loader.Load(engineConfig, subgraphs, routerEngineCfg, pluginsEnabled) + planConfig, providers, err := loader.Load(opts.EngineConfig, opts.Subgraphs, opts.RouterEngineConfig, opts.PluginsEnabled) if err != nil { return nil, nil, fmt.Errorf("failed to load configuration: %w", err) } - debug := &routerEngineCfg.Execution.Debug + debug := &opts.RouterEngineConfig.Execution.Debug planConfig.Debug = plan.DebugConfiguration{ PrintOperationTransformations: debug.PrintOperationTransformations, PrintOperationEnableASTRefs: debug.PrintOperationEnableASTRefs, @@ -248,19 +277,19 @@ func (b *ExecutorConfigurationBuilder) buildPlannerConfiguration(ctx context.Con PlanningVisitor: debug.PlanningVisitor, DatasourceVisitor: debug.DatasourceVisitor, } - planConfig.MinifySubgraphOperations = routerEngineCfg.Execution.MinifySubgraphOperations + planConfig.MinifySubgraphOperations = opts.RouterEngineConfig.Execution.MinifySubgraphOperations - planConfig.EnableOperationNamePropagation = routerEngineCfg.Execution.EnableSubgraphFetchOperationName + planConfig.EnableOperationNamePropagation = opts.RouterEngineConfig.Execution.EnableSubgraphFetchOperationName - planConfig.BuildFetchReasons = routerEngineCfg.Execution.EnableRequireFetchReasons || routerEngineCfg.Execution.ValidateRequiredExternalFields - planConfig.ValidateRequiredExternalFields = routerEngineCfg.Execution.ValidateRequiredExternalFields - planConfig.RelaxSubgraphOperationFieldSelectionMergingNullability = routerEngineCfg.Execution.RelaxSubgraphOperationFieldSelectionMergingNullability + planConfig.BuildFetchReasons = opts.RouterEngineConfig.Execution.EnableRequireFetchReasons || opts.RouterEngineConfig.Execution.ValidateRequiredExternalFields + planConfig.ValidateRequiredExternalFields = opts.RouterEngineConfig.Execution.ValidateRequiredExternalFields + planConfig.RelaxSubgraphOperationFieldSelectionMergingNullability = opts.RouterEngineConfig.Execution.RelaxSubgraphOperationFieldSelectionMergingNullability // Enable cost computation when cost control is enabled - if routerEngineCfg.CostControl != nil && routerEngineCfg.CostControl.Enabled { + if opts.RouterEngineConfig.CostControl != nil && opts.RouterEngineConfig.CostControl.Enabled { planConfig.ComputeCosts = true - planConfig.StaticCostDefaultListSize = routerEngineCfg.CostControl.EstimatedListSize - planConfig.IgnoreImplementingTypeWeights = routerEngineCfg.CostControl.IgnoreImplementingTypeWeights + planConfig.StaticCostDefaultListSize = opts.RouterEngineConfig.CostControl.EstimatedListSize + planConfig.IgnoreImplementingTypeWeights = opts.RouterEngineConfig.CostControl.IgnoreImplementingTypeWeights } return planConfig, providers, nil diff --git a/router/core/executor_test.go b/router/core/executor_test.go new file mode 100644 index 0000000000..a157b09a49 --- /dev/null +++ b/router/core/executor_test.go @@ -0,0 +1,37 @@ +package core + +import ( + "testing" + + "github.com/stretchr/testify/require" + "github.com/wundergraph/graphql-go-tools/v2/pkg/ast" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/plan" +) + +func TestExecutorCloseReleasesSchemaReferences(t *testing.T) { + t.Parallel() + + executor := &Executor{ + ClientSchema: &ast.Document{}, + RouterSchema: &ast.Document{}, + PlanConfig: plan.Configuration{DataSources: []plan.DataSource{nil}}, + RenameTypeNames: nil, + } + + executor.Close() + + require.Nil(t, executor.ClientSchema) + require.Nil(t, executor.RouterSchema) + require.Empty(t, executor.PlanConfig.DataSources) + require.Nil(t, executor.RenameTypeNames) + require.Nil(t, executor.Resolver) +} + +func TestExecutorCloseNilSafe(t *testing.T) { + t.Parallel() + + var executor *Executor + require.NotPanics(t, func() { + executor.Close() + }) +} diff --git a/router/core/factoryresolver.go b/router/core/factoryresolver.go index d0a94d9a51..46db85e1b8 100644 --- a/router/core/factoryresolver.go +++ b/router/core/factoryresolver.go @@ -7,6 +7,7 @@ import ( "net/http" "net/url" "slices" + "sync" "time" "github.com/buger/jsonparser" @@ -17,6 +18,7 @@ import ( 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/grpcconnector" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" rmetric "github.com/wundergraph/cosmo/router/pkg/metric" "github.com/wundergraph/cosmo/router/pkg/pubsub" pubsub_datasource "github.com/wundergraph/cosmo/router/pkg/pubsub/datasource" @@ -78,6 +80,10 @@ type DefaultFactoryResolver struct { transportFactory ApiTransportFactory defaultSubgraphRequestTimeout time.Duration subscriptionClientOptions []graphql_datasource.SubscriptionClientOption + useNoopSubscriptionClient bool + + subscriptionClient graphql_datasource.GraphQLSubscriptionClient + subscriptionClientOnce sync.Once } func NewDefaultFactoryResolver( @@ -131,7 +137,9 @@ func NewDefaultFactoryResolver( graphql_datasource.WithLogger(factoryLogger), } + useNoopSubscriptionClient := false if subscriptionClientOptions != nil { + useNoopSubscriptionClient = subscriptionClientOptions.UseNoopClient if subscriptionClientOptions.PingInterval > 0 { options = append(options, graphql_datasource.WithPingInterval(subscriptionClientOptions.PingInterval)) } @@ -164,6 +172,7 @@ func NewDefaultFactoryResolver( transportFactory: transportFactory, defaultSubgraphRequestTimeout: transportOptions.SubgraphTransportOptions.RequestTimeout, subscriptionClientOptions: options, + useNoopSubscriptionClient: useNoopSubscriptionClient, } } @@ -183,10 +192,40 @@ func (d *DefaultFactoryResolver) ResolveGraphqlFactory(subgraphName string) (pla if d.transportFactory == nil || d.baseTransport == nil { // dummy implementation for plan generator that doesn't make requests - subscriptionClient := graphql_datasource.NewGraphQLSubscriptionClient(d.engineCtx, + return graphql_datasource.NewFactory(d.engineCtx, http.DefaultClient, d.subscriptionClientForFactory()) + } + + defaultHTTPClient := &http.Client{ + Timeout: d.defaultSubgraphRequestTimeout, + Transport: d.transportFactory.RoundTripper(d.baseTransport), + } + + if subgraphClient, ok := d.subgraphHTTPClients[subgraphName]; ok { + // it's intentional that we're not using the subgraphClient for subscriptions + // custom subgraph clients are intended to be used for custom timeouts, which is not relevant for subscriptions + return graphql_datasource.NewFactory(d.engineCtx, subgraphClient, d.subscriptionClientForFactory()) + } + + return graphql_datasource.NewFactory(d.engineCtx, defaultHTTPClient, d.subscriptionClientForFactory()) +} + +func (d *DefaultFactoryResolver) subscriptionClientForFactory() graphql_datasource.GraphQLSubscriptionClient { + if mondaytweaks.ShareUpstreamSubscriptionClient { + return d.sharedSubscriptionClient() + } + return d.newSubscriptionClient() +} + +func (d *DefaultFactoryResolver) newSubscriptionClient() graphql_datasource.GraphQLSubscriptionClient { + if d.useNoopSubscriptionClient { + return noopGraphQLSubscriptionClientInstance + } + + if d.transportFactory == nil || d.baseTransport == nil { + return graphql_datasource.NewGraphQLSubscriptionClient( + d.engineCtx, d.subscriptionClientOptions..., ) - return graphql_datasource.NewFactory(d.engineCtx, http.DefaultClient, subscriptionClient) } defaultHTTPClient := &http.Client{ @@ -198,18 +237,49 @@ func (d *DefaultFactoryResolver) ResolveGraphqlFactory(subgraphName string) (pla Transport: d.transportFactory.RoundTripper(d.baseTransport), } - subscriptionClient := graphql_datasource.NewGraphQLSubscriptionClient( + return graphql_datasource.NewGraphQLSubscriptionClient( d.engineCtx, - append([]graphql_datasource.SubscriptionClientOption{graphql_datasource.WithUpgradeClient(defaultHTTPClient), graphql_datasource.WithStreamingClient(streamingClient)}, d.subscriptionClientOptions...)..., + append([]graphql_datasource.SubscriptionClientOption{ + graphql_datasource.WithUpgradeClient(defaultHTTPClient), + graphql_datasource.WithStreamingClient(streamingClient), + }, d.subscriptionClientOptions...)..., ) +} - if subgraphClient, ok := d.subgraphHTTPClients[subgraphName]; ok { - // it's intentional that we're not using the subgraphClient for subscriptions - // custom subgraph clients are intended to be used for custom timeouts, which is not relevant for subscriptions - return graphql_datasource.NewFactory(d.engineCtx, subgraphClient, subscriptionClient) - } +func (d *DefaultFactoryResolver) sharedSubscriptionClient() graphql_datasource.GraphQLSubscriptionClient { + d.subscriptionClientOnce.Do(func() { + if d.useNoopSubscriptionClient { + d.subscriptionClient = noopGraphQLSubscriptionClientInstance + return + } + + if d.transportFactory == nil || d.baseTransport == nil { + d.subscriptionClient = graphql_datasource.NewGraphQLSubscriptionClient( + d.engineCtx, + d.subscriptionClientOptions..., + ) + return + } + + defaultHTTPClient := &http.Client{ + Timeout: d.defaultSubgraphRequestTimeout, + Transport: d.transportFactory.RoundTripper(d.baseTransport), + } + + streamingClient := &http.Client{ + Transport: d.transportFactory.RoundTripper(d.baseTransport), + } + + d.subscriptionClient = graphql_datasource.NewGraphQLSubscriptionClient( + d.engineCtx, + append([]graphql_datasource.SubscriptionClientOption{ + graphql_datasource.WithUpgradeClient(defaultHTTPClient), + graphql_datasource.WithStreamingClient(streamingClient), + }, d.subscriptionClientOptions...)..., + ) + }) - return graphql_datasource.NewFactory(d.engineCtx, defaultHTTPClient, subscriptionClient) + return d.subscriptionClient } func (d *DefaultFactoryResolver) ResolveStaticFactory() (factory plan.PlannerFactory[staticdatasource.Configuration], err error) { diff --git a/router/core/graph_server.go b/router/core/graph_server.go index af206997be..1e47962df3 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -23,6 +23,7 @@ import ( "github.com/golang-jwt/jwt/v5" "github.com/klauspost/compress/gzhttp" "github.com/klauspost/compress/gzip" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" "github.com/wundergraph/cosmo/router/pkg/routerconfig" "go.opentelemetry.io/otel/attribute" otelmetric "go.opentelemetry.io/otel/metric" @@ -681,6 +682,10 @@ type graphMux struct { mux *chi.Mux reused atomic.Bool + wsHandler *WebsocketHandler + executor *Executor + planCacheOnEvictEnabled atomic.Bool + planCache *ristretto.Cache[uint64, *planWithMetaData] planFallbackCache *slowplancache.Cache[*planWithMetaData] persistedOperationCache *ristretto.Cache[uint64, NormalizationCacheEntry] @@ -721,11 +726,21 @@ func (s *graphMux) buildOperationCaches(srv *graphServer) (computeSha256 bool, e BufferItems: 64, } if srv.cacheWarmup != nil && srv.cacheWarmup.Enabled && srv.cacheWarmup.InMemoryFallback { - planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) { - // This could be called before planFallbackCache is set, but it's not a problem - // because there is a nil guard inside, as well as items should not really be evicted - // on startup - s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration) + if mondaytweaks.SkipPlanCacheOnEvictDuringMuxShutdown { + s.planCacheOnEvictEnabled.Store(true) + planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) { + // This could be called before planFallbackCache is set, but it's not a problem + // because there is a nil guard inside, as well as items should not really be evicted + // on startup + if !s.planCacheOnEvictEnabled.Load() { + return + } + s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration) + } + } else { + planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) { + s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration) + } } } s.planCache, err = ristretto.NewCache[uint64, *planWithMetaData](planCacheConfig) @@ -957,10 +972,31 @@ func (s *graphMux) configureCacheMetrics(srv *graphServer, baseOtelAttributes [] return nil } -func (s *graphMux) Shutdown(ctx context.Context) error { - // cancel the graph muxes context to close its resources like websocket connections, resolvers, etc. - s.cancel() +func closeRistrettoCacheUint64[V any](cache **ristretto.Cache[uint64, V]) { + if *cache != nil { + (*cache).Close() + *cache = nil + } +} + +// releaseOperationCaches drops references to closed Ristretto caches so the old +// graphMux can be collected after shutdown (Close clears entries but retains structs). +func (s *graphMux) releaseOperationCaches() { + closeRistrettoCacheUint64(&s.planCache) + if s.planFallbackCache != nil { + s.planFallbackCache.Close() + s.planFallbackCache = nil + } + closeRistrettoCacheUint64(&s.persistedOperationCache) + closeRistrettoCacheUint64(&s.normalizationCache) + closeRistrettoCacheUint64(&s.variablesNormalizationCache) + closeRistrettoCacheUint64(&s.remapVariablesCache) + closeRistrettoCacheUint64(&s.complexityCalculationCache) + closeRistrettoCacheUint64(&s.validationCache) + closeRistrettoCacheUint64(&s.operationHashCache) +} +func (s *graphMux) closeOperationCachesLegacy() { s.planCache.Close() s.planFallbackCache.Close() s.persistedOperationCache.Close() @@ -970,6 +1006,43 @@ func (s *graphMux) Shutdown(ctx context.Context) error { s.complexityCalculationCache.Close() s.validationCache.Close() s.operationHashCache.Close() +} + +func (s *graphMux) Shutdown(ctx context.Context) error { + if mondaytweaks.DrainWebsocketSubscriptionsBeforeCacheClose { + // Close websocket subscriptions synchronously before tearing down plan caches so + // active preparedPlan and executor references are released first. + if s.wsHandler != nil { + s.wsHandler.ShutdownConnections() + } + } + + // cancel the graph muxes context to close its resources like websocket connections, resolvers, etc. + s.cancel() + + if mondaytweaks.SkipPlanCacheOnEvictDuringMuxShutdown { + // ristretto Close() clears all entries and invokes OnEvict for each one. Disable + // migration into the slow-plan fallback cache during intentional mux shutdown. + s.planCacheOnEvictEnabled.Store(false) + if s.planFallbackCache != nil { + s.planFallbackCache.Wait() + } + } + + if mondaytweaks.CloseExecutorOnGraphMuxShutdown { + if s.executor != nil { + s.executor.Close() + s.executor = nil + } + } + + if mondaytweaks.NilGraphMuxCachesOnShutdown { + s.releaseOperationCaches() + s.wsHandler = nil + s.mux = nil + } else { + s.closeOperationCachesLegacy() + } var err error @@ -1446,22 +1519,30 @@ func (s *graphServer) buildGraphMux( return nil, fmt.Errorf("failed to process retry options: %w", err) } + subscriptionClientOptions := &SubscriptionClientOptions{ + PingInterval: s.engineExecutionConfiguration.WebSocketClientPingInterval, + PingTimeout: s.engineExecutionConfiguration.WebSocketClientPingTimeout, + WriteTimeout: s.engineExecutionConfiguration.WebSocketClientWriteTimeout, + AckTimeout: s.engineExecutionConfiguration.WebSocketClientAckTimeout, + ReadLimit: int64(s.engineExecutionConfiguration.WebSocketClientReadLimit), + DefaultErrorExtensionCode: s.subgraphErrorPropagation.DefaultExtensionCode, + } + // Client-facing WebSocket subscriptions are disabled; skip upstream ping loops + // that would otherwise start one goroutine per subgraph datasource factory. + if mondaytweaks.DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled && + s.webSocketConfiguration != nil && !s.webSocketConfiguration.Enabled { + subscriptionClientOptions.PingInterval = 0 + } + ecb := &ExecutorConfigurationBuilder{ - introspection: s.introspection, - baseURL: s.baseURL, - baseTripper: s.baseTransport, - subgraphTrippers: subgraphTippers, - pluginHost: s.connector, - logger: s.logger, - trackUsageInfo: s.graphqlMetricsConfig.Enabled || s.metricConfig.Prometheus.PromSchemaFieldUsage.Enabled, - subscriptionClientOptions: &SubscriptionClientOptions{ - PingInterval: s.engineExecutionConfiguration.WebSocketClientPingInterval, - PingTimeout: s.engineExecutionConfiguration.WebSocketClientPingTimeout, - WriteTimeout: s.engineExecutionConfiguration.WebSocketClientWriteTimeout, - AckTimeout: s.engineExecutionConfiguration.WebSocketClientAckTimeout, - ReadLimit: int64(s.engineExecutionConfiguration.WebSocketClientReadLimit), - DefaultErrorExtensionCode: s.subgraphErrorPropagation.DefaultExtensionCode, - }, + introspection: s.introspection, + baseURL: s.baseURL, + baseTripper: s.baseTransport, + subgraphTrippers: subgraphTippers, + pluginHost: s.connector, + logger: s.logger, + trackUsageInfo: s.graphqlMetricsConfig.Enabled || s.metricConfig.Prometheus.PromSchemaFieldUsage.Enabled, + subscriptionClientOptions: subscriptionClientOptions, transportOptions: &TransportOptions{ SubgraphTransportOptions: s.subgraphTransportOptions, PreHandlers: s.preOriginHandlers, @@ -1492,11 +1573,15 @@ func (s *graphServer) buildGraphMux( HeartbeatInterval: s.subscriptionHeartbeatInterval, PluginsEnabled: s.plugins.Enabled, InstanceData: s.instanceData, + WebSocketConfiguration: s.webSocketConfiguration, }, ) if err != nil { return nil, fmt.Errorf("failed to build plan configuration: %w", err) } + if mondaytweaks.CloseExecutorOnGraphMuxShutdown { + gm.executor = executor + } s.pubSubProviders = providers if pubSubStartupErr := s.startupPubSubProviders(s.graphServerCtx); pubSubStartupErr != nil { @@ -1830,7 +1915,7 @@ func (s *graphServer) buildGraphMux( }) if s.webSocketConfiguration != nil && s.webSocketConfiguration.Enabled { - wsMiddleware := NewWebsocketMiddleware(graphMuxCtx, WebsocketMiddlewareOptions{ + wsMiddleware, wsHandler := NewWebsocketMiddleware(graphMuxCtx, WebsocketMiddlewareOptions{ OperationProcessor: operationProcessor, OperationBlocker: operationBlocker, Planner: operationPlanner, @@ -1850,6 +1935,9 @@ func (s *graphServer) buildGraphMux( DisableVariablesRemapping: s.engineExecutionConfiguration.DisableVariablesRemapping, ApolloCompatibilityFlags: s.apolloCompatibilityFlags, }) + if mondaytweaks.DrainWebsocketSubscriptionsBeforeCacheClose { + gm.wsHandler = wsHandler + } // When the playground path is equal to the graphql path, we need to handle // ws upgrades and html requests on the same route. @@ -2172,6 +2260,9 @@ func (s *graphServer) Shutdown(ctx context.Context) error { if err := mux.Shutdown(ctx); err != nil { finalErr = errors.Join(finalErr, err) } + if mondaytweaks.NilGraphMuxCachesOnShutdown { + delete(s.graphMuxList, name) + } } // Close idle connections on base and subgraph transports diff --git a/router/core/noop_graphql_subscription_client.go b/router/core/noop_graphql_subscription_client.go new file mode 100644 index 0000000000..79af8d33f5 --- /dev/null +++ b/router/core/noop_graphql_subscription_client.go @@ -0,0 +1,105 @@ +package core + +import ( + "errors" + + nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" + "github.com/wundergraph/cosmo/router/pkg/config" + "github.com/wundergraph/graphql-go-tools/v2/pkg/ast" + "github.com/wundergraph/graphql-go-tools/v2/pkg/astparser" + "github.com/wundergraph/graphql-go-tools/v2/pkg/asttransform" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/graphql_datasource" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve" +) + +var errUpstreamGraphQLSubscriptionsDisabled = errors.New("upstream GraphQL subscriptions are disabled") + +// noopGraphQLSubscriptionClient satisfies graphql-go-tools NewFactory's non-nil +// subscription client requirement without initializing upstream WS/SSE transports. +type noopGraphQLSubscriptionClient struct{} + +func (c *noopGraphQLSubscriptionClient) Subscribe(_ *resolve.Context, _ graphql_datasource.GraphQLSubscriptionOptions, _ resolve.SubscriptionUpdater) error { + return errUpstreamGraphQLSubscriptionsDisabled +} + +var noopGraphQLSubscriptionClientInstance graphql_datasource.GraphQLSubscriptionClient = &noopGraphQLSubscriptionClient{} + +func shouldUseNoopUpstreamSubscriptionClient( + graphqlSchema string, + engineConfig *nodev1.EngineConfiguration, + eventsConfig config.EventsConfiguration, + webSocketConfiguration *config.WebSocketConfiguration, +) bool { + if !schemaHasSubscriptionRootFields(graphqlSchema) { + return true + } + if !clientWebSocketSubscriptionsEnabled(webSocketConfiguration) && !eventSubscriptionsEnabled(engineConfig, eventsConfig) { + return true + } + return false +} + +func schemaHasSubscriptionRootFields(graphqlSchema string) bool { + if graphqlSchema == "" { + return false + } + + doc, report := astparser.ParseGraphqlDocumentString(graphqlSchema) + if report.HasErrors() { + return false + } + if err := asttransform.MergeDefinitionWithBaseSchema(&doc); err != nil { + return false + } + + return subscriptionRootFieldCount(&doc) > 0 +} + +func subscriptionRootFieldCount(doc *ast.Document) int { + if doc.Index.SubscriptionTypeName == nil { + return 0 + } + + node, ok := doc.Index.FirstNodeByNameBytes(doc.Index.SubscriptionTypeName) + if !ok || node.Kind != ast.NodeKindObjectTypeDefinition { + return 0 + } + + return len(doc.ObjectTypeDefinitions[node.Ref].FieldsDefinition.Refs) +} + +func clientWebSocketSubscriptionsEnabled(webSocketConfiguration *config.WebSocketConfiguration) bool { + if webSocketConfiguration == nil { + return true + } + return webSocketConfiguration.Enabled +} + +func eventSubscriptionsEnabled(engineConfig *nodev1.EngineConfiguration, eventsConfig config.EventsConfiguration) bool { + if len(eventsConfig.Providers.Nats) > 0 || + len(eventsConfig.Providers.Kafka) > 0 || + len(eventsConfig.Providers.Redis) > 0 { + return true + } + + if engineConfig == nil { + return false + } + + for _, ds := range engineConfig.GetDatasourceConfigurations() { + if ds.GetKind() == nodev1.DataSourceKind_PUBSUB { + return true + } + customEvents := ds.GetCustomEvents() + if customEvents == nil { + continue + } + if len(customEvents.GetNats()) > 0 || + len(customEvents.GetKafka()) > 0 || + len(customEvents.GetRedis()) > 0 { + return true + } + } + + return false +} diff --git a/router/core/noop_graphql_subscription_client_test.go b/router/core/noop_graphql_subscription_client_test.go new file mode 100644 index 0000000000..a4a286c62c --- /dev/null +++ b/router/core/noop_graphql_subscription_client_test.go @@ -0,0 +1,89 @@ +package core + +import ( + "testing" + + nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" + "github.com/wundergraph/cosmo/router/pkg/config" + "github.com/stretchr/testify/require" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/graphql_datasource" +) + +func TestShouldUseNoopUpstreamSubscriptionClient_NoSubscriptionRootFields(t *testing.T) { + schema := `type Query { hello: String }` + + require.True(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + nil, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: true}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_EmptySubscriptionType(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { }` + + require.True(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + nil, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: true}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_ClientWSDisabledWithoutEvents(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { onUpdate: String }` + + require.True(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + &nodev1.EngineConfiguration{}, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: false}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_ClientWSDisabledWithPubSubDatasource(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { onUpdate: String }` + + engineConfig := &nodev1.EngineConfiguration{ + DatasourceConfigurations: []*nodev1.DataSourceConfiguration{ + {Kind: nodev1.DataSourceKind_PUBSUB}, + }, + } + + require.False(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + engineConfig, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: false}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_UpstreamSubscriptionsNeeded(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { onUpdate: String }` + + require.False(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + &nodev1.EngineConfiguration{}, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: true}, + )) +} + +func TestNoopGraphQLSubscriptionClient_SubscribeReturnsError(t *testing.T) { + err := noopGraphQLSubscriptionClientInstance.Subscribe(nil, graphql_datasource.GraphQLSubscriptionOptions{}, nil) + require.ErrorIs(t, err, errUpstreamGraphQLSubscriptionsDisabled) +} + +func TestSharedSubscriptionClient_UsesNoopWhenConfigured(t *testing.T) { + resolver := &DefaultFactoryResolver{ + useNoopSubscriptionClient: true, + } + + client := resolver.sharedSubscriptionClient() + require.Same(t, noopGraphQLSubscriptionClientInstance, client) +} diff --git a/router/core/operation_planner.go b/router/core/operation_planner.go index f9da57396f..037b23021a 100644 --- a/router/core/operation_planner.go +++ b/router/core/operation_planner.go @@ -18,9 +18,9 @@ import ( ) type planWithMetaData struct { - preparedPlan plan.Plan - operationDocument, schemaDocument *ast.Document - typeFieldUsageInfo []*graphqlschemausage.TypeFieldUsageInfo + preparedPlan plan.Plan + operationDocument *ast.Document + typeFieldUsageInfo []*graphqlschemausage.TypeFieldUsageInfo argumentUsageInfo []*graphqlmetricsv1.ArgumentUsageInfo content string operationName string @@ -96,7 +96,6 @@ func (p *OperationPlanner) planOperation(content string, name string, includeQue return &planWithMetaData{ preparedPlan: preparedPlan, operationDocument: &doc, - schemaDocument: p.executor.RouterSchema, }, nil } diff --git a/router/core/router.go b/router/core/router.go index 5dd087acd1..e1fe1cfbeb 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -19,6 +19,7 @@ import ( "connectrpc.com/connect" "github.com/mitchellh/mapstructure" "github.com/nats-io/nuid" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" "github.com/wundergraph/cosmo/router/pkg/routerconfig" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -27,6 +28,7 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.uber.org/zap" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/graphqlmetrics/v1/graphqlmetricsv1connect" nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" @@ -627,6 +629,12 @@ func (r *Router) serverTLSConfig() (*tls.Config, error) { // newGraphServer creates a new server. func (r *Router) newServer(ctx context.Context, response *routerconfig.Response) error { + // Extract slow-plan cache entries before building the new graph server, which + // overwrites ReloadPersistentState cache references and before the old graphMux shuts down. + if mondaytweaks.CallOnRouterConfigReloadOnHotReload { + r.reloadPersistentState.OnRouterConfigReload() + } + server, err := newGraphServer(ctx, r, response, r.proxy) if err != nil { r.logger.Error("Failed to create graph server. Keeping the old server", zap.Error(err)) @@ -1098,6 +1106,11 @@ func (r *Router) bootstrap(ctx context.Context) error { } r.staticExecutionConfig = executionConfig + + if hash, hashErr := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path); hashErr == nil && mondaytweaks.SkipManifestReloadWhenMapperUnchanged { + r.lastManifestMapperHash = hash + r.manifestMapperHashSeen = true + } } if err := r.buildClients(ctx); err != nil { @@ -1717,6 +1730,20 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) return } + if mondaytweaks.SkipManifestReloadWhenMapperUnchanged { + mapperHash, err := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path) + if err != nil { + ll.Error("Failed to hash manifest mapper", zap.Error(err)) + return + } + + if r.manifestMapperHashSeen && mapperHash == r.lastManifestMapperHash { + ll.Debug("Manifest mapper unchanged, skipping reload", + zap.String("path", r.manifestConfig.Path)) + return + } + } + cfg, err := routerconfig.AssembleStaticExecutionConfigFromManifest( r.manifestConfig.Path, routerconfig.AssembleConfigRules{ SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, @@ -1734,6 +1761,24 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) ll.Error("Failed to update server with new config", zap.Error(err)) return } + + if mondaytweaks.SkipManifestReloadWhenMapperUnchanged { + mapperHash, err := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path) + if err != nil { + ll.Error("Failed to hash manifest mapper", zap.Error(err)) + return + } + r.lastManifestMapperHash = mapperHash + r.manifestMapperHashSeen = true + } + + if mondaytweaks.ResetExecutionConfigProtoOnReload { + if old := r.staticExecutionConfig; old != nil && old != cfg { + proto.Reset(old) + } + } + r.staticExecutionConfig = cfg + r.trackExecutionConfigUsage(cfg, true) }, }) diff --git a/router/core/router_config.go b/router/core/router_config.go index 6b380ede12..062ae1409f 100644 --- a/router/core/router_config.go +++ b/router/core/router_config.go @@ -153,6 +153,9 @@ type Config struct { grpcPluginDialOptions []grpc.DialOption tracingAttributes []config.CustomAttribute subscriptionHooks subscriptionHooks + // lastManifestMapperHash skips manifest reload when mapper.json content is unchanged. + lastManifestMapperHash [32]byte + manifestMapperHashSeen bool } // Usage returns an anonymized version of the config for usage tracking diff --git a/router/core/transport.go b/router/core/transport.go index 609229c341..32afe6745e 100644 --- a/router/core/transport.go +++ b/router/core/transport.go @@ -223,6 +223,8 @@ type SubscriptionClientOptions struct { AckTimeout time.Duration ReadLimit int64 DefaultErrorExtensionCode string + // UseNoopClient skips upstream WS/SSE transport initialization when subscriptions are not needed. + UseNoopClient bool } func NewTransport(opts *TransportOptions) *TransportFactory { diff --git a/router/core/websocket.go b/router/core/websocket.go index 95f83864b6..e4cf5b85a0 100644 --- a/router/core/websocket.go +++ b/router/core/websocket.go @@ -70,7 +70,7 @@ type WebsocketMiddlewareOptions struct { ApolloCompatibilityFlags config.ApolloCompatibilityFlags } -func NewWebsocketMiddleware(ctx context.Context, opts WebsocketMiddlewareOptions) func(http.Handler) http.Handler { +func NewWebsocketMiddleware(ctx context.Context, opts WebsocketMiddlewareOptions) (func(http.Handler) http.Handler, *WebsocketHandler) { handler := &WebsocketHandler{ ctx: ctx, operationProcessor: opts.OperationProcessor, @@ -148,7 +148,16 @@ func NewWebsocketMiddleware(ctx context.Context, opts WebsocketMiddlewareOptions } handler.handleUpgradeRequest(w, r) }) + }, handler +} + +// ShutdownConnections closes all active websocket connections and unsubscribes +// any live GraphQL subscriptions before graph mux caches are torn down. +func (h *WebsocketHandler) ShutdownConnections() { + if h == nil { + return } + h.closeAllConnections() } // wsConnectionWrapper is a wrapper around websocket.Conn that allows diff --git a/router/pkg/mondaytweaks/mondaytweaks.go b/router/pkg/mondaytweaks/mondaytweaks.go index 226bbbf8c4..ab53923cf1 100644 --- a/router/pkg/mondaytweaks/mondaytweaks.go +++ b/router/pkg/mondaytweaks/mondaytweaks.go @@ -10,4 +10,49 @@ const ( // struct itself is GC'd — which may be delayed by goroutines still referencing // the owning graphMux — causing ~200-300 MB of retained memory per config reload. ClearSlowPlanCacheOnClose = true + + // OmitSchemaDocumentFromCachedPlans removes the unused schemaDocument field from + // planWithMetaData (compile-time structural change in operation_planner.go). + OmitSchemaDocumentFromCachedPlans = true + + // CallOnRouterConfigReloadOnHotReload invokes ReloadPersistentState.OnRouterConfigReload + // at the start of Router.newServer(), matching the supervisor restart path. + CallOnRouterConfigReloadOnHotReload = true + + // SkipPlanCacheOnEvictDuringMuxShutdown disables ristretto OnEvict migration into + // slowplancache while a graphMux is shutting down intentionally. + SkipPlanCacheOnEvictDuringMuxShutdown = true + + // DrainWebsocketSubscriptionsBeforeCacheClose closes client websocket subscriptions + // synchronously before plan caches are torn down on graphMux shutdown. + DrainWebsocketSubscriptionsBeforeCacheClose = true + + // CloseExecutorOnGraphMuxShutdown nils federation schema refs held by Executor after + // graphMux drain, allowing the old graph generation to be garbage-collected. + CloseExecutorOnGraphMuxShutdown = true + + // NilGraphMuxCachesOnShutdown closes and nils Ristretto caches on shut-down graphMux, + // drops wsHandler/mux references, and removes the mux from graphMuxList. + NilGraphMuxCachesOnShutdown = true + + // ResetExecutionConfigProtoOnReload proto.Resets the previous staticExecutionConfig + // after a successful manifest reload so decoded protojson strings can be collected. + ResetExecutionConfigProtoOnReload = true + + // SkipManifestReloadWhenMapperUnchanged skips manifest watcher reload when mapper.json + // bytes are unchanged. Disabled: latest.json / feature-flag files can change without + // mapper.json changing, which would serve stale config. + SkipManifestReloadWhenMapperUnchanged = false + + // ShareUpstreamSubscriptionClient uses one upstream GraphQLSubscriptionClient per + // DefaultFactoryResolver instead of one per subgraph factory (behavior-altering). + ShareUpstreamSubscriptionClient = true + + // UseNoopUpstreamSubscriptionClientWhenUnused skips upstream WS/SSE transport init + // when subscriptions are not used (behavior-altering). + UseNoopUpstreamSubscriptionClientWhenUnused = true + + // DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled sets PingInterval=0 on + // upstream subscription clients when client-facing websocket is disabled. + DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled = true ) diff --git a/router/pkg/routerconfig/routerconfig.go b/router/pkg/routerconfig/routerconfig.go index 71ba30e846..bda805ac0f 100644 --- a/router/pkg/routerconfig/routerconfig.go +++ b/router/pkg/routerconfig/routerconfig.go @@ -18,6 +18,7 @@ package routerconfig import ( + "crypto/sha256" "encoding/json" "fmt" "io/fs" @@ -106,6 +107,16 @@ func readMapperFile(path string) (map[string]string, error) { return mapper, nil } +// ManifestMapperSHA256 returns the SHA-256 digest of mapper.json bytes. +// Used to skip manifest reload when only the file mtime changed. +func ManifestMapperSHA256(manifestConfigPath string) ([32]byte, error) { + data, err := os.ReadFile(filepath.Join(manifestConfigPath, "mapper.json")) + if err != nil { + return [32]byte{}, fmt.Errorf("failed to read mapper file: %w", err) + } + return sha256.Sum256(data), nil +} + // assembleConfig assembles the router execution config from the base config and the feature flag configs. // The base config is the latest.json file in the manifest directory. // The feature flag configs are the feature-flags/.json files in the manifest directory.