googleapis · rahul2393 · Aug 26, 2026 · Aug 26, 2026 · Aug 26, 2026 · Aug 27, 2026
@@ -0,0 +1,11 @@
+# Dynamic channel pool lifecycle
+
+Dynamic scaling removes channels from picker candidates before closing them. Removed channels drain
+existing work, then close after their active stream count reaches zero and the configured idle grace
+period passes.
+
+| Routing state | Behavior while channel drains |
+| --- | --- |
+| Unaffinitized calls | Pickers skip the draining channel. |
+| Session-scoped affinity key | The key is unbound and its next call selects an active channel. |
+| Caller-owned `ChannelAffinityRef` | The reference stays on the open draining channel so transaction RPCs remain ordered. Calls still contribute to that channel's active stream count and delay closure. The reference selects an active channel after delegate shutdown or `useDifferentChannelOnNextCall()`. |
@@ -63,6 +63,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <site.installationModule>grpc-gcp</site.installationModule>
     <api-common.version>2.68.0-SNAPSHOT</api-common.version><!-- {x-version-update:api-common:current} -->
+    <awaitility.version>4.3.0</awaitility.version>
     <auto-value.version>1.11.0</auto-value.version>
     <error-prone-annotations.version>2.50.0</error-prone-annotations.version>
     <google-http-client.version>2.2.0</google-http-client.version>
@@ -172,6 +173,12 @@
       <version>${junit.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <version>${awaitility.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>com.google.truth</groupId>
       <artifactId>truth</artifactId>

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.grpc;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import io.grpc.ManagedChannel;
+
+/** Primes newly built delegate channels concurrently before dynamic-pool publication. */
+@FunctionalInterface
+public interface GcpChannelPrimer {
+
+  /**
+   * Issues a cheap end-to-end RPC on {@code channel} so its connection is warm before real traffic.
+   * Scale-up batches invoke this method concurrently and publish each channel as soon as its own
+   * future succeeds. For example, a Cloud Spanner implementation can execute {@code SELECT 1}.
+   * Return a failed future to reject and close the channel.
+   */
+  ListenableFuture<Void> prime(ManagedChannel channel);
+}
@@ -17,6 +17,7 @@
 package com.google.cloud.grpc;
 
 import com.google.cloud.grpc.proto.AffinityConfig;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import io.grpc.Attributes;
 import io.grpc.CallOptions;
@@ -30,7 +31,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Queue;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 
@@ -52,16 +53,28 @@ public class GcpClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
   private ClientCall<ReqT, RespT> delegateCall = null;
   private List<String> keys = null;
   private boolean received = false;
-  private final AtomicBoolean decremented = new AtomicBoolean(false);
+  // 0 = not counted, 1 = counted, 2 = finished.
+  private final AtomicInteger countState = new AtomicInteger();
 
   @GuardedBy("this")
   private final Queue<Runnable> calls = new ArrayDeque<>();
 
   @GuardedBy("this")
   private boolean started;
 
+  @GuardedBy("this")
+  private boolean cancelQueued;
+
+  @GuardedBy("this")
+  private boolean cancelled;
+
   private long startNanos = 0;
 
+  @VisibleForTesting
+  synchronized int queuedCallCountForTest() {
+    return calls.size();
+  }
+
   protected GcpClientCall(
       GcpManagedChannel delegateChannel,
       MethodDescriptor<ReqT, RespT> methodDescriptor,
@@ -90,7 +103,22 @@ public void setMessageCompression(boolean enabled) {
 
   @Override
   public void cancel(@Nullable String message, @Nullable Throwable cause) {
-    checkSendMessage(() -> checkedCancel(message, cause));
+    synchronized (this) {
+      if (cancelQueued || cancelled) {
+        return;
+      }
+      cancelQueued = true;
+      Runnable cancelCall =
+          () -> {
+            cancelled = true;
+            checkedCancel(message, cause);
+          };
+      if (started) {
+        cancelCall.run();
+      } else {
+        calls.add(cancelCall);
+      }
+    }
   }
 
   @Override
@@ -104,6 +132,7 @@ public void halfClose() {
    */
   @Override
   public void sendMessage(ReqT message) {
+    boolean send;
     synchronized (this) {
       if (!started) {
         startNanos = System.nanoTime();
@@ -123,17 +152,30 @@ public void sendMessage(ReqT message) {
           delegateChannelRef = delegateChannel.getChannelRef(key);
         }
         delegateChannelRef.activeStreamsCountIncr();
-
-        // Create the client call and do the previous operations.
-        delegateCall = delegateChannelRef.getChannel().newCall(methodDescriptor, callOptions);
-        for (Runnable call : calls) {
-          call.run();
+        countState.set(1);
+
+        try {
+          // Create the client call and do the previous operations.
+          CallOptions callOptionsWithChannelId =
+              callOptions.withOption(GcpManagedChannel.CHANNEL_ID_KEY, delegateChannelRef.getId());
+          delegateCall =
+              delegateChannelRef.getChannel().newCall(methodDescriptor, callOptionsWithChannelId);
+          for (Runnable call : calls) {
+            call.run();
+          }
+        } catch (RuntimeException | Error failure) {
+          finishCount(Status.fromThrowable(failure), true);
+          throw failure;
+        } finally {
+          calls.clear();
         }
-        calls.clear();
         started = true;
       }
+      send = !cancelled;
+    }
+    if (send) {
+      delegateCall.sendMessage(message);
     }
-    delegateCall.sendMessage(message);
   }
 
   /** Calls that send exactly one message should not check this method. */
@@ -162,14 +204,21 @@ public String toString() {
   }
 
   private void checkedCancel(@Nullable String message, @Nullable Throwable cause) {
-    if (!decremented.getAndSet(true)) {
-      delegateChannelRef.activeStreamsCountDecr(startNanos, Status.CANCELLED, true);
-    }
+    finishCount(Status.CANCELLED, true);
     delegateCall.cancel(message, cause);
   }
 
+  private void finishCount(Status status, boolean fromClientSide) {
+    if (countState.compareAndSet(1, 2)) {
+      delegateChannelRef.activeStreamsCountDecr(startNanos, status, fromClientSide);
+    }
+  }
+
   private void checkSendMessage(Runnable call) {
     synchronized (this) {
+      if (cancelQueued || cancelled) {
+        return;
+      }
       if (started) {
         call.run();
       } else {
@@ -185,9 +234,7 @@ private Listener<RespT> getListener(final Listener<RespT> responseListener) {
       // Decrement the stream number by one when the call is closed.
       @Override
       public void onClose(Status status, Metadata trailers) {
-        if (!decremented.getAndSet(true)) {
-          delegateChannelRef.activeStreamsCountDecr(startNanos, status, false);
-        }
+        finishCount(status, false);
         // If the operation completed successfully, bind/unbind the affinity key.
         if (keys != null && status.getCode() == Status.Code.OK) {
           if (affinity.getCommand() == AffinityConfig.Command.UNBIND) {
@@ -219,7 +266,8 @@ public void onMessage(RespT message) {
    * A simple wrapper of ClientCall.
    *
    * <p>It defines the callback function to manage the number of active streams of a ChannelRef
-   * everytime a call is started/closed.
+   * every time a call is created/closed. Stream capacity is reserved in the constructor, before
+   * {@link #start(Listener, Metadata)}, and remains reserved until close or cancel.
    */
   public static class SimpleGcpClientCall<ReqT, RespT> extends ForwardingClientCall<ReqT, RespT> {
 
@@ -230,7 +278,8 @@ public static class SimpleGcpClientCall<ReqT, RespT> extends ForwardingClientCal
     private final boolean unbindOnComplete;
     private long startNanos = 0;
 
-    private final AtomicBoolean decremented = new AtomicBoolean(false);
+    // 0 = not counted, 1 = counted, 2 = finished.
+    private final AtomicInteger countState = new AtomicInteger();
 
     protected SimpleGcpClientCall(
         GcpManagedChannel delegateChannel,
@@ -244,8 +293,16 @@ protected SimpleGcpClientCall(
       // Set the actual channel ID in callOptions so downstream interceptors can access it.
       CallOptions callOptionsWithChannelId =
           callOptions.withOption(GcpManagedChannel.CHANNEL_ID_KEY, channelRef.getId());
-      this.delegateCall =
-          channelRef.getChannel().newCall(methodDescriptor, callOptionsWithChannelId);
+      startNanos = System.nanoTime();
+      channelRef.activeStreamsCountIncr();
+      countState.set(1);
+      try {
+        this.delegateCall =
+            channelRef.getChannel().newCall(methodDescriptor, callOptionsWithChannelId);
+      } catch (RuntimeException | Error failure) {
+        finishCount(Status.fromThrowable(failure), true);
+        throw failure;
+      }
     }
 
     @Override
@@ -255,16 +312,12 @@ protected ClientCall<ReqT, RespT> delegate() {
 
     @Override
     public void start(Listener<RespT> responseListener, Metadata headers) {
-      startNanos = System.nanoTime();
-
       Listener<RespT> listener =
           new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(
               responseListener) {
             @Override
             public void onClose(Status status, Metadata trailers) {
-              if (!decremented.getAndSet(true)) {
-                channelRef.activeStreamsCountDecr(startNanos, status, false);
-              }
+              finishCount(status, false);
               // Unbind the affinity key when the caller explicitly requests it
               // (e.g., on terminal RPCs like Commit or Rollback) to prevent
               // unbounded growth of the affinity map.
@@ -281,20 +334,28 @@ public void onMessage(RespT message) {
             }
           };
 
-      channelRef.activeStreamsCountIncr();
-      delegateCall.start(listener, headers);
+      try {
+        delegateCall.start(listener, headers);
+      } catch (RuntimeException | Error failure) {
+        finishCount(Status.fromThrowable(failure), true);
+        throw failure;
+      }
     }
 
     @Override
     public void cancel(String message, Throwable cause) {
-      if (!decremented.getAndSet(true)) {
-        channelRef.activeStreamsCountDecr(startNanos, Status.CANCELLED, true);
-      }
+      finishCount(Status.CANCELLED, true);
       // Always unbind on cancel — the transaction is being abandoned.
       if (affinityKey != null) {
         delegateChannel.unbind(Collections.singletonList(affinityKey));
       }
       delegateCall.cancel(message, cause);
     }
+
+    private void finishCount(Status status, boolean fromClientSide) {
+      if (countState.compareAndSet(1, 2)) {
+        channelRef.activeStreamsCountDecr(startNanos, status, fromClientSide);
+      }
+    }
   }
 }