feat(grpc-gcp): prime scaled channels before publish (#14232)

With dynamic scaling enabled, a freshly built channel was published to
the pool the moment its delegate was constructed. Its first real RPCs
then paid for connection establishment, TLS, and any service-side
warm-up, so every scale-up event injected a latency spike into live
traffic right when the pool was already under load.

**Change**
- **Optional channel primer.** New `GcpChannelPrimer` hook
(`setChannelPrimer`) issues a cheap end-to-end RPC on each newly built
scale-up channel before it is published; for Cloud Spanner this can be a
`SELECT 1`. Channels in a scale-up batch are primed concurrently and
each one is published as soon as its own prime succeeds, so one slow
channel never holds back the rest. A `null` primer (the default) keeps
the existing publish-on-build path. Only dynamic scale-up channels are
primed; the initial pool is not.
- **Bounded attempts.** Each attempt is bounded by `channelPrimeTimeout`
(default 10s) and retried up to `channelPrimeMaxAttempts` (default 3)
with exponential backoff (100ms doubling, capped at 5s). The timeout
covers the whole attempt, including the synchronous `prime()` call. A
primer still blocked inside `prime()` at the timeout fails the channel
without a retry, since nothing can stop that call; primer invocations
run on a dedicated executor so a blocked primer cannot starve the shared
scheduler that drives scale-up, draining, and timeouts. Exhausted or
timed-out channels are closed and counted in the new
`scale_up_prime_failures` metric.
- **Scale-up accounting.** Channels still priming count toward both the
pool size cap and the desired size, so a second scale-up during a slow
prime batch cannot rebuild the same shortfall or construct delegates
past `maxSize`. A prime abandoned because its call never returned keeps
its slot until the call does, so a stuck primer can pin at most
`maxSize` slots rather than one thread per scale-up event.
- **Lifecycle.** Unpublished priming channels are never visible to
pickers; shutdown cancels every pending prime and closes its delegate.
Per-RPC pick and stream-accounting paths are unchanged, with no new
hot-path allocations.
- **Configuration.** `setChannelPrimeTimeout` rejects negative or
non-nanosecond-representable durations (zero uses the default);
`setChannelPrimeMaxAttempts` rejects negatives (zero uses the default).

Author: rahul2393

grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpChannelPrimeController.java

@@ -0,0 +1,72 @@
+/*
+ * 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 delegate channels built by dynamic scale-up before they are published to the pool. The
+ * initial pool and non-dynamic growth are not primed.
+ */
+@FunctionalInterface
+public interface GcpChannelPrimer {
+
+  /**
+   * Issues a cheap, read-only end-to-end RPC on {@code channel} so its transport, TLS session, and
+   * server-side connection are established before the channel serves live traffic. Scale-up batches
+   * invoke this method concurrently across channels 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 the attempt; the pool retries up to the configured attempt
+   * count and closes the channel when they are exhausted.
+   *
+   * <p>Implementations must observe the following:
+   *
+   * <ul>
+   *   <li><b>Return promptly and do the work inside the future.</b> Do not block in this method (no
+   *       blocking stub, {@code ResultSet.next()}, {@code Future.get()}, or waiting on other
+   *       threads). The configured prime timeout bounds each attempt as a whole: a future still
+   *       pending at the timeout is cancelled and the attempt retried, while a call that is still
+   *       blocked inside this method at the timeout fails the channel with no retry, because
+   *       nothing can stop it. Such a call keeps occupying its primer thread and a scale-up slot
+   *       until it returns.
+   *   <li><b>Set an explicit, short deadline on the RPC</b>, for example {@code
+   *       stub.withDeadlineAfter(5, SECONDS)}. Do not rely on the default deadline: many Google
+   *       Cloud APIs default to very long ones, such as one hour for Cloud Spanner's {@code
+   *       ExecuteStreamingSql}, so an unresponsive backend would leave the call lingering on the
+   *       server long after the pool has given up on it.
+   *   <li><b>Propagate cancellation to the gRPC call.</b> The pool calls {@code
+   *       future.cancel(true)} on the returned future when the attempt times out or the pool shuts
+   *       down. Cancelling the future must cancel the underlying {@link io.grpc.ClientCall} or
+   *       {@link io.grpc.Context.CancellableContext} so the stream and its server-side resources
+   *       are released promptly.
+   *   <li><b>Expect the channel to be shut down under a running call.</b> The pool may call {@code
+   *       shutdownNow()} on {@code channel} while this method is still running: when the call is
+   *       still blocked at the timeout, or when the pool shuts down. A channel-shutdown status from
+   *       an RPC on {@code channel} is then expected and is not a backend outage.
+   *   <li><b>Tolerate concurrent calls for one channel.</b> A retry may overlap work from a
+   *       cancelled future if that work does not promptly honor cancellation, so the implementation
+   *       must be safe for multiple invocations on the same channel at once.
+   *   <li><b>Keep the RPC trivial and side-effect free.</b> It should be strictly read-only and
+   *       need minimal server work, such as {@code SELECT 1} or a lightweight ping.
+   * </ul>
+   *
+   * @param channel the newly built, not yet published delegate channel to prime
+   * @return a future that succeeds when the priming RPC succeeds, or fails to reject the attempt
+   */
+  ListenableFuture<Void> prime(ManagedChannel channel);
+}

grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpChannelPrimer.java

@@ -76,6 +76,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.IntUnaryOperator;
+import java.util.function.LongUnaryOperator;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -84,6 +85,7 @@
 
 /** A channel management factory that implements grpc.Channel APIs. */
 public class GcpManagedChannel extends ManagedChannel {
+
   private static final Logger logger = Logger.getLogger(GcpManagedChannel.class.getName());
   static final AtomicInteger channelPoolIndex = new AtomicInteger();
 
@@ -181,6 +183,7 @@ private static int stateFromChannelId(int channelId) {
   private int errorPenaltyStep = 5;
   private Duration errorPenaltyDuration = Duration.ofSeconds(5);
   private long errorPenaltyDurationNanos = Duration.ofSeconds(5).toNanos();
+  @Nullable private GcpChannelPrimeController channelPrimeController;
   private boolean isDynamicScalingEnabled = false;
   private int maxConcurrentStreamsLowWatermark = DEFAULT_MAX_STREAM;
   private GcpManagedChannelOptions.ChannelPickStrategy channelPickStrategy =
@@ -302,6 +305,7 @@ private static int stateFromChannelId(int channelId) {
   private AtomicLong maxUnresponsiveDrops = new AtomicLong();
   private AtomicLong scaleUpCount = new AtomicLong();
   private AtomicLong scaleDownCount = new AtomicLong();
+  private final AtomicLong scaleUpPrimeFailures = new AtomicLong();
 
   // Clock supplier for nanoTime, injectable for testing.
   private NanoClock nanoClock = System::nanoTime;
@@ -354,6 +358,41 @@ int readyChannelCountForTest() {
     return readyChannels.get();
   }
 
+  @VisibleForTesting
+  long scaleUpPrimeFailuresForTest() {
+    return scaleUpPrimeFailures.get();
+  }
+
+  @VisibleForTesting
+  boolean scaleUpWorkerRunningForTest() {
+    return scaleUpWorkerRunning.get();
+  }
+
+  @VisibleForTesting
+  synchronized int inFlightPrimeCountForTest() {
+    return channelPrimeController == null ? 0 : channelPrimeController.inFlightCount();
+  }
+
+  @VisibleForTesting
+  synchronized int abandonedPrimeCountForTest() {
+    return channelPrimeController == null ? 0 : channelPrimeController.abandonedCount();
+  }
+
+  @VisibleForTesting
+  static long primeBackoffMillisForTest(int attempt) {
+    return GcpChannelPrimeController.backoffMillis(attempt);
+  }
+
+  @VisibleForTesting
+  static long primeBackoffMillisForTest(int attempt, LongUnaryOperator random) {
+    return GcpChannelPrimeController.backoffMillis(attempt, random);
+  }
+
+  @VisibleForTesting
+  static long primeBaseBackoffMillisForTest(int attempt) {
+    return GcpChannelPrimeController.baseBackoffMillis(attempt);
+  }
+
   @VisibleForTesting
   synchronized int drainTaskCountForTest() {
     return drainTasks.size();
@@ -742,13 +781,64 @@ private void initOptions() {
       errorPenaltyStep = poolOptions.getErrorPenaltyStep();
       errorPenaltyDuration = poolOptions.getErrorPenaltyDuration();
       errorPenaltyDurationNanos = errorPenaltyDuration.toNanos();
+      channelPrimeController = createChannelPrimeController(poolOptions);
       isDynamicScalingEnabled =
           minRpcPerChannel > 0 && maxRpcPerChannel > 0 && !scaleDownInterval.isZero();
       channelPickStrategy = poolOptions.getChannelPickStrategy();
     }
     initMetrics();
   }
 
+  @Nullable
+  private GcpChannelPrimeController createChannelPrimeController(
+      GcpManagedChannelOptions.GcpChannelPoolOptions poolOptions) {
+    GcpChannelPrimer primer = poolOptions.getChannelPrimer();
+    if (primer == null) {
+      return null;
+    }
+    GcpChannelPrimeController.Pool pool =
+        new GcpChannelPrimeController.Pool() {
+          @Override
+          public boolean isShuttingDown() {
+            return shuttingDown;
+          }
+
+          @Override
+          public boolean isFull() {
+            return channelRefs.size() >= maxSize;
+          }
+
+          @Override
+          public void addPrimedChannel(ManagedChannel channel) {
+            addBuiltChannel(channel);
+            scaleUpCount.incrementAndGet();
+          }
+
+          @Override
+          public void reportPrimeFailure(Throwable failure) {
+            scaleUpPrimeFailures.incrementAndGet();
+            logger.log(Level.WARNING, log("Scaled-up channel priming failed"), failure);
+          }
+
+          @Override
+          public void reportPrimeAbandoned() {
+            logger.warning(
+                log(
+                    "Channel priming abandoned without retry: prime() was still blocked at the"
+                        + " timeout and nothing can stop it. The channel is closed and the blocked"
+                        + " call keeps occupying a primer thread and a scale-up slot until it"
+                        + " returns."));
+          }
+        };
+    return new GcpChannelPrimeController(
+        this,
+        pool,
+        primer,
+        poolOptions.getChannelPrimeTimeout().toNanos(),
+        poolOptions.getChannelPrimeMaxAttempts(),
+        SHARED_BACKGROUND_SERVICE);
+  }
+
   private synchronized void initCleanupTask(Duration cleanupInterval) {
     if (cleanupInterval.isZero()) {
       return;
@@ -1060,6 +1150,13 @@ private void initMetrics() {
         this,
         GcpManagedChannel::reportScaleUp,
         GcpManagedChannel::reportScaleDown);
+
+    createDerivedLongCumulativeTimeSeries(
+        GcpMetricsConstants.METRIC_SCALE_UP_PRIME_FAILURES,
+        "The number of scaled-up channels rejected because priming failed or timed out.",
+        GcpMetricsConstants.COUNT,
+        this,
+        GcpManagedChannel::reportScaleUpPrimeFailures);
   }
 
   private void setupOtelCommonAttributes(GcpMetricsOptions metricsOptions) {
@@ -1315,6 +1412,14 @@ private void initOtelMetrics(Meter meter) {
               m.record(reportScaleUp(), withDirection(GcpMetricsConstants.DIRECTION_UP));
               m.record(reportScaleDown(), withDirection(GcpMetricsConstants.DIRECTION_DOWN));
             });
+
+    meter
+        .gaugeBuilder(metricPrefix + GcpMetricsConstants.METRIC_SCALE_UP_PRIME_FAILURES)
+        .ofLongs()
+        .setDescription(
+            "The number of scaled-up channels rejected because priming failed or timed out.")
+        .setUnit(GcpMetricsConstants.COUNT)
+        .buildWithCallback(m -> m.record(reportScaleUpPrimeFailures(), otelCommonAttributes));
   }
 
   private void logGauge(String key, long value) {
@@ -1343,6 +1448,7 @@ void logMetrics() {
     reportMaxAllowedChannels();
     reportScaleUp();
     reportScaleDown();
+    reportScaleUpPrimeFailures();
     reportNumChannelDisconnect();
     reportNumChannelConnect();
     reportMinReadinessTime();
@@ -1707,6 +1813,12 @@ private long reportScaleDown() {
     return value;
   }
 
+  private long reportScaleUpPrimeFailures() {
+    long value = scaleUpPrimeFailures.get();
+    logCumulative(GcpMetricsConstants.METRIC_SCALE_UP_PRIME_FAILURES, value);
+    return value;
+  }
+
   private void incReadyChannels(boolean connected) {
     if (connected) {
       numChannelConnect.incrementAndGet();
@@ -2234,7 +2346,11 @@ private void dynamicUpscale() {
     }
     final int channelsToBuild;
     synchronized (this) {
-      if (!isDynamicScalingEnabled || shuttingDown || channelRefs.size() >= maxSize) {
+      // Channels still priming are capacity already on its way: they count toward the size cap
+      // and the desired size, so overlapping scale-ups cannot build the same shortfall twice.
+      // Abandoned primes keep their slot until their blocked prime() call returns.
+      int priming = channelPrimeController == null ? 0 : channelPrimeController.primingCount();
+      if (!isDynamicScalingEnabled || shuttingDown || channelRefs.size() + priming >= maxSize) {
         return;
       }
       if (lastScaleUpNanos != Long.MIN_VALUE
@@ -2248,11 +2364,11 @@ private void dynamicUpscale() {
         return;
       }
       int desired = ceilDiv(pickerLoad(activeChannels, now), targetRpcPerChannel());
-      int add = desired - active;
+      int add = desired - active - priming;
       // Small pools may add two channels per event before percentage growth dominates.
       int percentCap = Math.max(2, ceilDiv((long) active * maxScaleUpPercent, 100));
       add = Math.min(add, percentCap);
-      add = Math.min(add, maxSize - active);
+      add = Math.min(add, maxSize - channelRefs.size() - priming);
       if (add <= 0) {
         return;
       }
@@ -2277,6 +2393,11 @@ private void dynamicUpscale() {
       throw failure;
     }
 
+    if (channelPrimeController != null) {
+      builtChannels.forEach(channelPrimeController::startPrime);
+      return;
+    }
+
     int added = 0;
     List<ManagedChannel> surplus = new ArrayList<>();
     synchronized (this) {
@@ -2555,7 +2676,7 @@ private String keyFromOptsCtx(CallOptions callOptions) {
     return key;
   }
 
-  private synchronized void cancelBackgroundTasks() {
+  private synchronized List<GcpChannelPrimeController.PendingPrime> cancelBackgroundTasks() {
     shuttingDown = true;
     scaleUpSignalPending.set(false);
     if (cleanupTask != null) {
@@ -2572,12 +2693,18 @@ private synchronized void cancelBackgroundTasks() {
     }
     drainTasks.values().forEach(task -> task.cancel(false));
     drainTasks.clear();
+    return channelPrimeController == null
+        ? Collections.emptyList()
+        : channelPrimeController.detachAll();
   }
 
   @Override
   public ManagedChannel shutdownNow() {
     logger.finer(log("Shutdown now started."));
-    cancelBackgroundTasks();
+    List<GcpChannelPrimeController.PendingPrime> primesToCancel = cancelBackgroundTasks();
+    if (channelPrimeController != null) {
+      channelPrimeController.cancel(primesToCancel, true);
+    }
     List<ChannelRef> activeSnapshot = new ArrayList<>(channelRefs);
     List<ChannelRef> removedSnapshot = new ArrayList<>(removedChannelRefs);
     for (ChannelRef channelRef : activeSnapshot) {
@@ -2599,7 +2726,10 @@ public ManagedChannel shutdownNow() {
   @Override
   public ManagedChannel shutdown() {
     logger.finer(log("Shutdown started."));
-    cancelBackgroundTasks();
+    List<GcpChannelPrimeController.PendingPrime> primesToCancel = cancelBackgroundTasks();
+    if (channelPrimeController != null) {
+      channelPrimeController.cancel(primesToCancel, false);
+    }
     List<ChannelRef> activeSnapshot = new ArrayList<>(channelRefs);
     List<ChannelRef> removedSnapshot = new ArrayList<>(removedChannelRefs);
     for (ChannelRef channelRef : activeSnapshot) {