Address review comments on PR grpc#12807: inline CONNECTING_RESULT in RoundRobinLoadBalancer and refine PendingStream delay telemetry synchronization

Author: AgraVator

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

@@ -428,6 +428,9 @@ private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers,
      * structured transition event is appended to the active span without span re-creation.
      */
     synchronized void updateDelay(@Nullable String newType, @Nullable String newReason) {
+      if (getRealStream() != null) {
+        return;
+      }
       if (!Objects.equals(activeDelayType, newType)) {
         // Delay type changed (e.g., from RLS lookup to connecting). End the previous delay.
         if (activeDelayType != null) {
@@ -466,7 +469,12 @@ synchronized void endDelay() {
     }
 
     Runnable setStreamAndEndDelay(ClientStream stream) {
-      endDelay();
+      synchronized (this) {
+        if (getRealStream() != null) {
+          return null;
+        }
+        endDelay();
+      }
       return setStream(stream);
     }
 
@@ -493,7 +501,6 @@ private Runnable createRealStream(ClientTransport transport, String authorityOve
 
     @Override
     public void cancel(Status reason) {
-      endDelay();
       super.cancel(reason);
       synchronized (lock) {
         if (reportTransportTerminated != null) {
@@ -512,6 +519,7 @@ public void cancel(Status reason) {
 
     @Override
     protected void onEarlyCancellation(Status reason) {
+      endDelay();
       for (ClientStreamTracer tracer : tracers) {
         tracer.streamClosed(reason);
       }

core/src/test/java/io/grpc/internal/DelayedClientTransportTest.java

@@ -939,6 +939,39 @@ public PickResult pickSubchannel(PickSubchannelArgs args) {
     };
   }
 
+  @Test
+  public void streamDelayMetrics_updateDelayAfterCancellation_isNoOp() {
+    final FakeStreamTracer fakeTracer = new FakeStreamTracer();
+    ClientStreamTracer[] customTracers = new ClientStreamTracer[] { fakeTracer };
+
+    // Create a pending stream
+    delayedTransport.reprocess(fakePicker(
+        PickResult.withNoResult("connecting", "pick_first: attempting to connect")));
+    final ClientStream stream = delayedTransport.newStream(
+        method, headers, callOptions, customTracers);
+    stream.start(streamListener);
+
+    assertEquals(Collections.singletonList("connecting"), fakeTracer.startedDelayTypes);
+    assertEquals(0, fakeTracer.delayEndedCount);
+
+    // Reprocess with a custom picker that cancels the stream during the pick.
+    // This exactly simulates the concurrent timing race.
+    SubchannelPicker racePicker = new SubchannelPicker() {
+      @Override
+      public PickResult pickSubchannel(PickSubchannelArgs args) {
+        stream.cancel(Status.CANCELLED);
+        return PickResult.withNoResult("connecting", "retry-backoff");
+      }
+    };
+
+    delayedTransport.reprocess(racePicker);
+
+    // VERIFICATION:
+    // The updateDelay called after the cancellation must be ignored because of our check.
+    assertEquals(Collections.singletonList("connecting"), fakeTracer.startedDelayTypes);
+    assertEquals(1, fakeTracer.delayEndedCount);
+  }
+
   private static TransportProvider newTransportProvider(final ClientTransport transport) {
     return new TransportProvider() {
       @Override

util/src/main/java/io/grpc/util/RoundRobinLoadBalancer.java

@@ -44,16 +44,12 @@ final class RoundRobinLoadBalancer extends MultiChildLoadBalancer {
   private static final PickResult CONNECTING_RESULT = PickResult.withNoResult("connecting",
       "round_robin connecting: TCP/TLS handshake in progress to child balancers");
   private final AtomicInteger sequence = new AtomicInteger(new Random().nextInt());
-  private SubchannelPicker currentPicker = new FixedResultPicker(connectingResult());
+  private SubchannelPicker currentPicker = new FixedResultPicker(CONNECTING_RESULT);
 
   public RoundRobinLoadBalancer(Helper helper) {
     super(helper);
   }
 
-  private PickResult connectingResult() {
-    return CONNECTING_RESULT;
-  }
-
   /**
    * Updates picker with the list of active subchannels (state == READY).
    */
@@ -74,7 +70,7 @@ protected void updateOverallBalancingState() {
       }
 
       if (isConnecting) {
-        updateBalancingState(CONNECTING, new FixedResultPicker(connectingResult()));
+        updateBalancingState(CONNECTING, new FixedResultPicker(CONNECTING_RESULT));
       } else {
         updateBalancingState(TRANSIENT_FAILURE, createReadyPicker(getChildLbStates()));
       }