Make connection-close telemetry flush blocking

Add flushSync method to metricsAggregator that holds the lock only while
draining the batch, then calls exporter.export synchronously. Update
Interceptor.Close to use flushSync so telemetry is guaranteed to be
delivered before the connection returns to the pool, matching JDBC's
blocking flush(true) behavior on session close.

Co-authored-by: Isaac

Author: samikshya-db

telemetry/aggregator.go

@@ -197,13 +197,29 @@ func (agg *metricsAggregator) flushLoop() {
 	}
 }
 
-// flush flushes pending metrics to exporter.
+// flush flushes pending metrics to exporter asynchronously (fire-and-forget).
 func (agg *metricsAggregator) flush(ctx context.Context) {
 	agg.mu.Lock()
 	defer agg.mu.Unlock()
 	agg.flushUnlocked(ctx)
 }
 
+// flushSync flushes pending metrics synchronously, blocking until the export
+// completes. Used on connection close to guarantee delivery before returning.
+func (agg *metricsAggregator) flushSync(ctx context.Context) {
+	agg.mu.Lock()
+	if len(agg.batch) == 0 {
+		agg.mu.Unlock()
+		return
+	}
+	metrics := make([]*telemetryMetric, len(agg.batch))
+	copy(metrics, agg.batch)
+	agg.batch = agg.batch[:0]
+	agg.mu.Unlock()
+
+	agg.exporter.export(ctx, metrics)
+}
+
 // flushUnlocked flushes without locking (caller must hold lock).
 func (agg *metricsAggregator) flushUnlocked(ctx context.Context) {
 	if len(agg.batch) == 0 {

telemetry/interceptor.go

@@ -200,6 +200,6 @@ func (i *Interceptor) Close(ctx context.Context) error {
 		return nil
 	}
 
-	i.aggregator.flush(ctx)
+	i.aggregator.flushSync(ctx)
 	return nil
 }