chore(spanner): implement dynamic channel pooling and transaction affinity

Adds dynamic channel pooling to the Spanner client, replacing single-channel stubs with an elastic pool of gRPC connections.

- Distributes requests across channels using the Power of Two Choices (P2C) algorithm to avoid hot connections.
- Automatically spins up and warms new channels under heavy load, and quietly drains idle ones when traffic drops.
- Pins multi-step transactions to the same channel so server state remains consistent.
- Supports both static and dynamic configurations, and respects the `SPANNER_NUM_CHANNELS` environment variable.
- Keeps the pool configuration internal (`pub(crate)`) for now while the API and behavior are finalized.

Author: olavloite

src/spanner/src/batch_read_only_transaction.rs

@@ -168,7 +168,7 @@ impl BatchReadOnlyTransaction {
             .partition_query(
                 request,
                 crate::RequestOptions::default(),
-                self.inner.context.channel_hint,
+                self.inner.context.affinity(),
             )
             .await?;
 
@@ -231,7 +231,7 @@ impl BatchReadOnlyTransaction {
             .partition_read(
                 request,
                 crate::RequestOptions::default(),
-                self.inner.context.channel_hint,
+                self.inner.context.affinity(),
             )
             .await?;
 
@@ -405,15 +405,11 @@ impl Partition {
         req: &ExecuteSqlRequest,
         gax_options: GaxRequestOptions,
     ) -> crate::Result<ResultSet> {
-        let channel_hint = client.next_channel_hint();
-        let gax_options = client.attach_request_id(gax_options, channel_hint);
+        let builder = client.execute_streaming_sql(req.clone(), gax_options, None);
+        let actual_gax_options = builder.options().clone();
         let (stream, attempt_start_time) =
-            Self::execute_partition_stream(client, "ExecuteStreamingSql", || {
-                client
-                    .execute_streaming_sql(req.clone(), gax_options.clone(), channel_hint)
-                    .send()
-            })
-            .await?;
+            Self::execute_partition_stream(client, "ExecuteStreamingSql", move || builder.send())
+                .await?;
 
         ResultSet::create(ResultSetParams {
             stream,
@@ -428,8 +424,7 @@ impl Partition {
             session_name: req.session.clone(),
             transaction_tag: None,
             operation: StreamOperation::Query(req.clone()),
-            channel_hint,
-            gax_options,
+            gax_options: actual_gax_options,
             method_name: "ExecuteStreamingSql",
             attempt_start_time: Some(attempt_start_time),
             operation_start_time: Some(attempt_start_time),
@@ -443,15 +438,10 @@ impl Partition {
         req: &ReadRequest,
         gax_options: GaxRequestOptions,
     ) -> crate::Result<ResultSet> {
-        let channel_hint = client.next_channel_hint();
-        let gax_options = client.attach_request_id(gax_options, channel_hint);
+        let builder = client.streaming_read(req.clone(), gax_options, None);
+        let actual_gax_options = builder.options().clone();
         let (stream, attempt_start_time) =
-            Self::execute_partition_stream(client, "StreamingRead", || {
-                client
-                    .streaming_read(req.clone(), gax_options.clone(), channel_hint)
-                    .send()
-            })
-            .await?;
+            Self::execute_partition_stream(client, "StreamingRead", move || builder.send()).await?;
 
         ResultSet::create(ResultSetParams {
             stream,
@@ -466,8 +456,7 @@ impl Partition {
             session_name: req.session.clone(),
             transaction_tag: None,
             operation: StreamOperation::Read(req.clone()),
-            channel_hint,
-            gax_options,
+            gax_options: actual_gax_options,
             method_name: "StreamingRead",
             attempt_start_time: Some(attempt_start_time),
             operation_start_time: Some(attempt_start_time),

src/spanner/src/batch_write_transaction.rs

@@ -209,12 +209,10 @@ impl BatchWriteTransactionBuilder {
     /// ```
     pub fn build(self) -> BatchWriteTransaction {
         let session_name = self.client.session_name();
-        let channel_hint = self.client.next_channel_hint();
         let gax_options = apply_defaults(self.gax_options);
         BatchWriteTransaction {
             session_name,
             client: self.client,
-            channel_hint,
             transaction_tag: self.transaction_tag,
             priority: self.priority,
             exclude_txn_from_change_streams: self.exclude_txn_from_change_streams,
@@ -230,7 +228,6 @@ impl BatchWriteTransactionBuilder {
 pub struct BatchWriteTransaction {
     session_name: String,
     client: DatabaseClient,
-    channel_hint: usize,
     transaction_tag: Option<String>,
     priority: Priority,
     exclude_txn_from_change_streams: bool,
@@ -291,7 +288,6 @@ impl BatchWriteTransaction {
         Ok(BatchWriteResponseStream {
             client: self.client,
             session_name: self.session_name,
-            channel_hint: self.channel_hint,
             transaction_tag: self.transaction_tag,
             priority: self.priority,
             exclude_txn_from_change_streams: self.exclude_txn_from_change_streams,
@@ -317,7 +313,6 @@ impl BatchWriteTransaction {
 pub struct BatchWriteResponseStream {
     client: DatabaseClient,
     session_name: String,
-    channel_hint: usize,
     transaction_tag: Option<String>,
     priority: Priority,
     exclude_txn_from_change_streams: bool,
@@ -436,7 +431,7 @@ impl BatchWriteResponseStream {
 
             let stream_result = self
                 .client
-                .batch_write(request, self.gax_options.clone(), self.channel_hint)
+                .batch_write(request, self.gax_options.clone(), None)
                 .send()
                 .await;
 
@@ -540,8 +535,6 @@ impl BatchWriteResponseStream {
         match self.check_retry(error) {
             Ok(()) => {
                 self.retry_count += 1;
-                // Rotate channel hint only when a retry is confirmed to distribute load across healthy connections.
-                self.channel_hint = self.client.next_channel_hint();
                 if let Some(policy) = self.gax_options.backoff_policy() {
                     let state = RetryState::new(true).set_attempt_count(self.retry_count as u32);
                     let delay = policy.on_failure(&state);