fix(tracing): keep postgres.run span open across row fetch

Review finding #1: postgres.run used `#[tracing::instrument]`, whose span closed as soon as `run()` returned the result stream, so it covered only query setup (bind/execute/flush) and not the row fetching that dominates query latency. Create the span explicitly and attach it to the returned stream via `instrument_stream` so it stays entered while rows are received.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: grahamc

sqlx-postgres/src/connection/executor.rs

@@ -16,6 +16,7 @@ use futures_core::stream::BoxStream;
 use futures_core::Stream;
 use futures_util::TryStreamExt;
 use sqlx_core::arguments::Arguments;
+use sqlx_core::instrument_stream::InstrumentStream;
 use sqlx_core::sql_str::SqlStr;
 use sqlx_core::Either;
 use std::{pin::pin, sync::Arc};
@@ -210,24 +211,24 @@ impl PgConnection {
         Ok(statement)
     }
 
-    #[tracing::instrument(
-        target = "sqlx::query",
-        name = "postgres.run",
-        skip_all,
-        fields(
-            db.system = "postgresql",
-            db.operation.parameters = arguments.as_ref().map_or(0, |a| a.len()),
-            db.postgresql.prepared = arguments.is_some(),
-        ),
-        level = "debug",
-    )]
     pub(crate) async fn run<'e, 'c: 'e, 'q: 'e>(
         &'c mut self,
         query: SqlStr,
         arguments: Option<PgArguments>,
         persistent: bool,
         metadata_opt: Option<Arc<PgStatementMetadata>>,
     ) -> Result<impl Stream<Item = Result<Either<PgQueryResult, PgRow>, Error>> + 'e, Error> {
+        // The span is attached to the returned stream (see `instrument_stream`)
+        // rather than via `#[tracing::instrument]` so it stays open while rows
+        // are fetched, not just while the query is set up.
+        let span = tracing::debug_span!(
+            target: "sqlx::query",
+            "postgres.run",
+            db.system = "postgresql",
+            db.operation.parameters = arguments.as_ref().map_or(0, |a| a.len()),
+            db.postgresql.prepared = arguments.is_some(),
+        );
+
         let mut logger = QueryLogger::new(query, self.inner.log_settings.clone());
         let sql = logger.sql().as_str();
 
@@ -397,7 +398,8 @@ impl PgConnection {
             }
 
             Ok(())
-        })
+        }
+        .instrument_stream(span))
     }
 }