fix(monitoring): prevent catalog pool deadlock in AddCDCBatchTablesForFlow (#3557)

Move PeerDBMetricsRecordAggregatesEnabled check before transaction start
  to avoid nested catalog pool connection acquisition. Previously, the
  function would:
  
  1. Acquire connection for transaction (pool.Begin)
2. Call PeerDBMetricsRecordAggregatesEnabled which internally acquires
another connection via `dynLookup > GetCatalogConnectionPoolFromEnv`

With concurrent calls and pool size of 3, all connections would be held
  by transactions while each waited for a 4th connection to check the
dynamic config, causing deadlock with idle transactions stuck in BEGIN.

  Fixes deadlock observed with 3 connections in "idle in transaction"
  state blocking all subsequent catalog operations.

Author: iskakaushik

flow/connectors/utils/monitoring/monitoring.go

@@ -123,6 +123,11 @@ func AddCDCBatchTablesForFlow(
 	tableNameRowsMapping map[string]*model.RecordTypeCounts,
 	otelManager *otel_metrics.OtelManager,
 ) error {
+	var recordAggregateMetrics bool
+	if enabled, err := internal.PeerDBMetricsRecordAggregatesEnabled(ctx, nil); err == nil && enabled {
+		recordAggregateMetrics = true
+	}
+
 	insertBatchTablesTx, err := pool.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("error while beginning transaction for inserting statistics: %w", err)
@@ -132,10 +137,7 @@ func AddCDCBatchTablesForFlow(
 		op    string
 		count int32
 	}
-	var recordAggregateMetrics bool
-	if enabled, err := internal.PeerDBMetricsRecordAggregatesEnabled(ctx, nil); err == nil && enabled {
-		recordAggregateMetrics = true
-	}
+
 	var syncedTablesCount int64
 	tableNameOperations := make(map[string][3]opWithValue, len(tableNameRowsMapping))
 	for destinationTableName, rowCounts := range tableNameRowsMapping {