Skip to content

chore(router): remove mondaytweaks memory-leak flags after upstream #3035 #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
arutkowski00 merged 1 commit into from
Jul 7, 2026
Merged

chore(router): remove mondaytweaks memory-leak flags after upstream #3035 #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 1 addition & 14 deletions router/core/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,10 @@ 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"
"github.com/wundergraph/cosmo/router/pkg/mondaytweaks"
pubsub_datasource "github.com/wundergraph/cosmo/router/pkg/pubsub/datasource"

"github.com/wundergraph/graphql-go-tools/v2/pkg/ast"
Expand Down Expand Up @@ -52,19 +52,6 @@ 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
Expand Down
37 changes: 0 additions & 37 deletions router/core/executor_test.go
View file
Open in desktop

This file was deleted.

134 changes: 34 additions & 100 deletions router/core/graph_server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -701,10 +701,6 @@ type graphMux struct {
reused atomic.Bool
finalized atomic.Bool

wsHandler *WebsocketHandler
executor *Executor
planCacheOnEvictEnabled atomic.Bool

planCache *ristretto.Cache[uint64, *planWithMetaData]
planFallbackCache *slowplancache.Cache[*planWithMetaData]
persistedOperationCache *ristretto.Cache[uint64, NormalizationCacheEntry]
Expand Down Expand Up @@ -755,21 +751,8 @@ func (s *graphMux) buildOperationCaches(srv *graphServer) (computeSha256 bool, e
BufferItems: 64,
}
if srv.cacheWarmup != nil && srv.cacheWarmup.Enabled && srv.cacheWarmup.InMemoryFallback {
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)
}
planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) {
s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration)
}
}
s.planCache, err = ristretto.NewCache(planCacheConfig)
Expand Down Expand Up @@ -1020,81 +1003,41 @@ func (s *graphMux) stopPubsubProviders(ctx context.Context) error {
}, providerTimeout, "pubsub provider shutdown timed out")
}

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()
s.normalizationCache.Close()
s.variablesNormalizationCache.Close()
s.remapVariablesCache.Close()
s.complexityCalculationCache.Close()
s.validationCache.Close()
s.operationHashCache.Close()
}

func (s *graphMux) Shutdown(ctx context.Context) error {
// Make sure we do not shutdown the mux multiple times
if !s.finalized.CompareAndSwap(false, true) {
return nil
}

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 s.planCache != nil {
s.planCache.Close()
}

if mondaytweaks.CloseExecutorOnGraphMuxShutdown {
if s.executor != nil {
s.executor.Close()
s.executor = nil
}
if s.planFallbackCache != nil {
s.planFallbackCache.Close()
}

if mondaytweaks.NilGraphMuxCachesOnShutdown {
s.releaseOperationCaches()
s.wsHandler = nil
s.mux = nil
} else {
s.closeOperationCachesLegacy()
if s.persistedOperationCache != nil {
s.persistedOperationCache.Close()
}
if s.normalizationCache != nil {
s.normalizationCache.Close()
}
if s.variablesNormalizationCache != nil {
s.variablesNormalizationCache.Close()
}
if s.remapVariablesCache != nil {
s.remapVariablesCache.Close()
}
if s.complexityCalculationCache != nil {
s.complexityCalculationCache.Close()
}
if s.validationCache != nil {
s.validationCache.Close()
}
if s.operationHashCache != nil {
s.operationHashCache.Close()
}

var err error
Expand Down Expand Up @@ -1605,13 +1548,13 @@ func (s *graphServer) buildGraphMux(

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,
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,
Expand Down Expand Up @@ -1648,9 +1591,6 @@ func (s *graphServer) buildGraphMux(
if err != nil {
return nil, fmt.Errorf("failed to build plan configuration: %w", err)
}
if mondaytweaks.CloseExecutorOnGraphMuxShutdown {
gm.executor = executor
}

if s.engineStats != nil && executor.Resolver != nil {
s.engineStats.RegisterResolver(executor.Resolver)
Expand Down Expand Up @@ -1994,7 +1934,7 @@ func (s *graphServer) buildGraphMux(
})

if s.webSocketConfiguration != nil && s.webSocketConfiguration.Enabled {
wsMiddleware, wsHandler := NewWebsocketMiddleware(graphMuxCtx, WebsocketMiddlewareOptions{
wsMiddleware, _ := NewWebsocketMiddleware(graphMuxCtx, WebsocketMiddlewareOptions{
OperationProcessor: operationProcessor,
OperationBlocker: operationBlocker,
Planner: operationPlanner,
Expand All @@ -2014,9 +1954,6 @@ 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.
Expand Down Expand Up @@ -2381,9 +2318,6 @@ 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
Expand Down
43 changes: 0 additions & 43 deletions router/core/router.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ 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"
Expand All @@ -28,7 +27,6 @@ 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"
Expand Down Expand Up @@ -629,12 +627,6 @@ 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))
Expand Down Expand Up @@ -1114,11 +1106,6 @@ 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 {
Expand Down Expand Up @@ -1738,20 +1725,6 @@ 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,
Expand All @@ -1769,22 +1742,6 @@ 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)
},
Expand Down
3 changes: 0 additions & 3 deletions router/core/router_config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,9 +158,6 @@ 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
Expand Down
Loading
Loading