fix: metrics flush timeout canceled shutdown

router/core/graph_server.go

@@ -997,12 +997,6 @@ func (s *graphMux) Shutdown(ctx context.Context) error {
 		}
 	}
 
-	if s.streamMetricStore != nil {
-		if aErr := s.streamMetricStore.Shutdown(ctx); aErr != nil {
-			err = errors.Join(err, aErr)
-		}
-	}
-
 	if s.prometheusMetricsExporter != nil {
 		if aErr := s.prometheusMetricsExporter.Shutdown(ctx); aErr != nil {
 			err = errors.Join(err, aErr)
@@ -2100,6 +2094,38 @@ func (s *graphServer) wait(ctx context.Context) error {
 	}
 }
 
+// metricsFlushTimeout bounds the single, central flush of the shared meter
+// providers during graph server shutdown.
+const metricsFlushTimeout = 30 * time.Second
+
+// flushMeterProviders flushes the OTLP and Prometheus meter providers once. These
+// providers are shared by every metric store (request, connection, stream,
+// engine, runtime), so a single flush drains all of their metrics.
+func (s *graphServer) flushMeterProviders(ctx context.Context) error {
+	wg := &sync.WaitGroup{}
+
+	var otlpErr error
+	var promErr error
+
+	if s.otlpMeterProvider != nil {
+		wg.Go(func() {
+			if err := s.otlpMeterProvider.ForceFlush(ctx); err != nil {
+				otlpErr = errors.Join(otlpErr, fmt.Errorf("failed to flush otlp metrics: %w", err))
+			}
+		})
+	}
+	if s.promMeterProvider != nil {
+		wg.Go(func() {
+			if err := s.promMeterProvider.ForceFlush(ctx); err != nil {
+				promErr = errors.Join(promErr, fmt.Errorf("failed to flush prometheus metrics: %w", err))
+			}
+		})
+	}
+	wg.Wait()
+
+	return errors.Join(otlpErr, promErr)
+}
+
 // Shutdown gracefully shutdown the server and waits for all in-flight requests to finish.
 // After all requests are done, it will shut down the metric store and runtime metrics.
 // Shutdown does cancel the context after all non-hijacked requests such as WebSockets has been handled.
@@ -2125,6 +2151,16 @@ func (s *graphServer) Shutdown(ctx context.Context) error {
 		zap.String("config_version", s.baseRouterConfigVersion),
 	)
 
+	// Flush the meter providers exactly once, with their own timeout,
+	// before tearing down the individual metric stores.
+	// As all the stores share the same meter providers, we only need to flush once
+	// before initiating the shutdown of the individual stores.
+	flushCtx, flushCancel := context.WithTimeout(ctx, metricsFlushTimeout)
+	if err := s.flushMeterProviders(flushCtx); err != nil {
+		finalErr = errors.Join(finalErr, fmt.Errorf("failed to flush metrics: %w", err))
+	}
+	flushCancel()
+
 	// Ensure that we don't wait indefinitely for shutdown
 	if s.routerGracePeriod > 0 {
 		newCtx, cancel := context.WithTimeout(ctx, s.routerGracePeriod)

router/core/graphql_prehandler.go

@@ -1244,21 +1244,18 @@ func (h *PreHandler) flushMetrics(ctx context.Context, requestLogger *zap.Logger
 	now := time.Now()
 
 	wg := &sync.WaitGroup{}
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+
+	wg.Go(func() {
 		if err := h.metrics.MetricStore().Flush(ctx); err != nil {
 			requestLogger.Error("Failed to flush OTEL metrics", zap.Error(err))
 		}
-	}()
+	})
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		if err := h.tracerProvider.ForceFlush(ctx); err != nil {
 			requestLogger.Error("Failed to flush OTEL tracer", zap.Error(err))
 		}
-	}()
+	})
 
 	wg.Wait()
 

router/pkg/metric/connection_metric_store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+
 	"github.com/wundergraph/cosmo/router/pkg/otel"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -17,14 +18,12 @@ import (
 type ConnectionMetricProvider interface {
 	MeasureConnectionAcquireDuration(ctx context.Context, duration float64, opts ...otelmetric.RecordOption)
 	MeasureMaxConnections(ctx context.Context, count int64, opts ...otelmetric.RecordOption)
-	Flush(ctx context.Context) error
 	Shutdown() error
 }
 
 // ConnectionMetricStore is the interface for connection and pool metrics only.
 type ConnectionMetricStore interface {
 	MeasureConnectionAcquireDuration(ctx context.Context, duration float64, attrs ...attribute.KeyValue)
-	Flush(ctx context.Context) error
 	Shutdown(ctx context.Context) error
 }
 
@@ -90,31 +89,12 @@ func (c *ConnectionMetrics) MeasureConnectionAcquireDuration(ctx context.Context
 	c.promConnectionMetrics.MeasureConnectionAcquireDuration(ctx, duration, opts)
 }
 
-// Flush flushes the metrics to the backend synchronously.
-func (h *ConnectionMetrics) Flush(ctx context.Context) error {
-	var err error
-
-	errOtlp := h.otlpConnectionMetrics.Flush(ctx)
-	if errOtlp != nil {
-		err = errors.Join(err, fmt.Errorf("failed to flush otlp metrics: %w", errOtlp))
-	}
-
-	errProm := h.promConnectionMetrics.Flush(ctx)
-	if errProm != nil {
-		err = errors.Join(err, fmt.Errorf("failed to flush prometheus metrics: %w", errProm))
-	}
-
-	return err
-}
-
-// Shutdown flushes the metrics and stops the runtime metrics.
-func (h *ConnectionMetrics) Shutdown(ctx context.Context) error {
+// Shutdown stops the metric instruments. It does not flush: the shared meter
+// providers are flushed once centrally during graph server shutdown to avoid
+// redundant ForceFlush calls that all compete for a single shutdown deadline.
+func (h *ConnectionMetrics) Shutdown(_ context.Context) error {
 	var err error
 
-	if errFlush := h.Flush(ctx); errFlush != nil {
-		err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errFlush))
-	}
-
 	if errProm := h.promConnectionMetrics.Shutdown(); errProm != nil {
 		err = errors.Join(err, fmt.Errorf("failed to shutdown prom metrics: %w", errProm))
 	}

router/pkg/metric/metric_store.go

@@ -479,21 +479,16 @@ func (h *Metrics) Flush(ctx context.Context) error {
 	return err
 }
 
-// Shutdown flushes the metrics and stops the runtime metrics.
-func (h *Metrics) Shutdown(ctx context.Context) error {
-
+// Shutdown stops the metric instruments. It does not flush: the shared meter
+// providers are flushed once centrally during graph server shutdown to avoid
+// redundant ForceFlush calls that all compete for a single shutdown deadline.
+func (h *Metrics) Shutdown(_ context.Context) error {
 	var err error
 
-	if errFlush := h.Flush(ctx); errFlush != nil {
-		err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errFlush))
-	}
-
-	errProm := h.promRequestMetrics.Shutdown()
-	if err != nil {
+	if errProm := h.promRequestMetrics.Shutdown(); errProm != nil {
 		err = errors.Join(err, fmt.Errorf("failed to shutdown prom metrics: %w", errProm))
 	}
-	errOtlp := h.otlpRequestMetrics.Shutdown()
-	if err != nil {
+	if errOtlp := h.otlpRequestMetrics.Shutdown(); errOtlp != nil {
 		err = errors.Join(err, fmt.Errorf("failed to shutdown otlp metrics: %w", errOtlp))
 	}
 

router/pkg/metric/oltp_connection_metric_store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+
 	"github.com/wundergraph/cosmo/router/pkg/otel"
 	"go.opentelemetry.io/otel/attribute"
 
@@ -92,16 +93,12 @@ func (h *otlpConnectionMetrics) MeasureMaxConnections(ctx context.Context, count
 	h.instruments.maxConnections.Record(ctx, count, opts...)
 }
 
-func (h *otlpConnectionMetrics) Flush(ctx context.Context) error {
-	return h.meterProvider.ForceFlush(ctx)
-}
-
 func (h *otlpConnectionMetrics) Shutdown() error {
 	var err error
 
 	for _, reg := range h.instrumentRegistrations {
 		if regErr := reg.Unregister(); regErr != nil {
-			err = errors.Join(regErr)
+			err = errors.Join(err, regErr)
 		}
 	}
 

router/pkg/metric/oltp_stream_metric_store.go

@@ -46,7 +46,3 @@ func (o *otlpStreamEventMetrics) Produce(ctx context.Context, opts ...otelmetric
 func (o *otlpStreamEventMetrics) Consume(ctx context.Context, opts ...otelmetric.AddOption) {
 	o.instruments.consumedMessages.Add(ctx, 1, opts...)
 }
-
-func (o *otlpStreamEventMetrics) Flush(ctx context.Context) error {
-	return o.meterProvider.ForceFlush(ctx)
-}

router/pkg/metric/prom_connection_metric_store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+
 	"github.com/wundergraph/cosmo/router/pkg/otel"
 	"go.opentelemetry.io/otel/attribute"
 
@@ -91,10 +92,6 @@ func (m *promConnectionMetrics) MeasureMaxConnections(ctx context.Context, count
 	m.instruments.maxConnections.Record(ctx, count, opts...)
 }
 
-func (m *promConnectionMetrics) Flush(ctx context.Context) error {
-	return m.meterProvider.ForceFlush(ctx)
-}
-
 func (h *promConnectionMetrics) Shutdown() error {
 	var err error
 

router/pkg/metric/prom_metric_store.go

@@ -172,7 +172,7 @@ func (h *PromMetricStore) Shutdown() error {
 
 	for _, reg := range h.instrumentRegistrations {
 		if regErr := reg.Unregister(); regErr != nil {
-			err = errors.Join(regErr)
+			err = errors.Join(err, regErr)
 		}
 	}
 

router/pkg/metric/stream_metric_store.go

@@ -2,7 +2,6 @@ package metric
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -34,16 +33,11 @@ type StreamsEvent struct {
 type StreamMetricProvider interface {
 	Produce(ctx context.Context, opts ...otelmetric.AddOption)
 	Consume(ctx context.Context, opts ...otelmetric.AddOption)
-
-	Flush(ctx context.Context) error
 }
 
 type StreamMetricStore interface {
 	Produce(ctx context.Context, event StreamsEvent)
 	Consume(ctx context.Context, event StreamsEvent)
-
-	Flush(ctx context.Context) error
-	Shutdown(ctx context.Context) error
 }
 
 // StreamMetrics is the store for Event (Kafka/Redis/NATS) metrics.
@@ -127,27 +121,3 @@ func (e *StreamMetrics) Consume(ctx context.Context, event StreamsEvent) {
 		provider.Consume(ctx, opt)
 	}
 }
-
-// Flush flushes the metrics to the backend synchronously.
-func (e *StreamMetrics) Flush(ctx context.Context) error {
-	var err error
-
-	for _, provider := range e.providers {
-		if errOtlp := provider.Flush(ctx); errOtlp != nil {
-			err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errOtlp))
-		}
-	}
-
-	return err
-}
-
-// Shutdown flushes the metrics and stops observers if any.
-func (e *StreamMetrics) Shutdown(ctx context.Context) error {
-	var err error
-
-	if errFlush := e.Flush(ctx); errFlush != nil {
-		err = errors.Join(err, fmt.Errorf("failed to flush metrics: %w", errFlush))
-	}
-
-	return err
-}