RetriableStream permanently leaked in UncommittedRetriableStreamsRegistry when cancel() races ClientCall.start()

[michael3lyb]

What version of gRPC-Java are you using?

Reproduced on 1.84.0 (latest) and 1.81.0.

RetriableStream.java, ClientCallImpl.java and ClientCalls.java are byte-identical between v1.81.0 and master, and the ManagedChannelImpl diff over that range does not touch the registry, prestart() or postCommit() — so master is affected too. All line numbers below are master.

What is your environment?

macOS / JDK 17, grpc-inprocess transport in the attached reproducer. Originally found in production on Linux/JDK11 over grpc-netty; the defect is transport-independent (it is entirely inside ClientCallImpl + RetriableStream).

What did you expect to see?

A ClientCall.cancel() should never leave the RetriableStream registered in ManagedChannelImpl$UncommittedRetriableStreamsRegistry, regardless of when the cancel lands relative to start().

What did you see instead?

On any channel built with enableRetry(), a cancel() that lands in the window between stream = clientStreamProvider.newStream(...) and stream.start(...) inside ClientCallImpl.startInternal() leaks the RetriableStream permanently into that channel-scoped, process-lifetime HashSet.

The order of operations inverts:

1. cancel() commits the stream and runs postCommit() → registry.remove(this) — a no-op, because the stream has not been registered yet;
2. start() then runs prestart() → registry.add(this);
3. commit() is one-shot, so no second post-commit task can ever be produced.

The entry is never removed for the life of the channel. Each leaked entry retains its Substream, Metadata, CallOptions and Context, so retained size is substantial. A leaked entry also permanently blocks delayedTransport.shutdown() on channel shutdown, since that is gated on the registry reaching isEmpty().

This is not hedging-specific — it reproduces with hedging disabled, i.e. on any enableRetry() channel.

Mechanism

ClientCallImpl.startInternal():

250:  stream = clientStreamProvider.newStream(method, callOptions, headers, context);
      //  ... ~8 setters, reportCallStarted() ...
286:  stream.start(new ClientStreamListenerImpl(observer));   // -> RetriableStream.start()
292:  cancellationHandler.setUp();

RetriableStream.start() registers only at step 286:

392:  public final void start(ClientStreamListener listener) {
393:    masterListener = listener;
395:    Status shutdownStatus = prestart();     // ManagedChannelImpl: registry.add(this)

ClientCallImpl.cancelInternal() is explicitly documented to run in this window, and the stream != null guard is already true from L250:

469:  // Cancel is called in exception handling cases, so it may be the case that the
470:  // stream was never successfully created or start has never been called.
471:  if (stream != null) {
482:    stream.cancel(status);      // non-null since L250 -- but not yet registered

RetriableStream.cancel() commits and runs the post-commit task inline:

526:  public final void cancel(final Status reason) {
527:    Substream noopSubstream = new Substream(0);
528:    noopSubstream.stream = new NoopClientStream();
529:    Runnable runnable = commit(noopSubstream);   // sets state.winningSubstream
532:    if (runnable != null) {
535:      runnable.run();          // -> postCommit() -> registry.remove(this)  == NO-OP

and commit() is one-shot:

155:  private Runnable commit(final Substream winningSubstream) {
156:    synchronized (lock) {
157:      if (state.winningSubstream != null) {
158:        return null;           // no further post-commit task, ever

postCommit() is the only caller of UncommittedRetriableStreamsRegistry.remove(...).

Steps to reproduce the bug

Single file, stock gRPC jars only, no protobuf and no third-party code (source inlined below). It uses the in-process transport, an enableRetry() channel, and a second thread calling call.cancel() with randomised nanosecond jitter around call.start(); it then reflects on ManagedChannelImpl.uncommittedRetriableStreamsRegistry.uncommittedRetriableStreams.size().

javac -cp <grpc jars> LeakRepro.java
java -cp .:<grpc jars> LeakRepro race        # leaks
java -cp .:<grpc jars> LeakRepro blocking    # control, 0
java -cp .:<grpc jars> -Dpolicy=none LeakRepro race   # hedging off -- still leaks

(The mode is a positional arg and defaults to blocking; running it with no args exercises the control path.)

Results — 200 calls each, after quiescence + System.gc()

run	version	cancel races start()	retry policy	leaked entries
A	1.84.0	yes	hedging	22, 60, 50 (three runs)
B	1.81.0	yes	hedging	54
C	1.84.0	yes	none (hedging off)	9
D (control)	1.84.0	no	hedging	0
E (control)	1.81.0	no	hedging	0

The exact count is timing-dependent and varies run to run; what is stable is leaked > 0 in every race run and exactly 0 in every control run, and registry size growing monotonically with call count in A/B/C. Run C is what widens the scope: the leak needs only enableRetry(), not hedging.

The controls are not vacuous — the same reflection probe reports a non-zero peak registry size during run E (max registry size ever observed = 1, tracking the single in-flight call), so it demonstrably can see entries and the final 0 is a real empty registry rather than a broken probe.

Every leaked entry has the same shape, which is exactly what the mechanism predicts:

leaked entry shape: total=50 committed(winningSubstream!=null)=50 inFlight=0
  cancelled=false passThrough=true hedgingFrozen=false hedgingAttemptCount=0
  buffer=null  drainedSubstreams=size=1  winningSubstream.stream=NoopClientStream

winningSubstream.stream == NoopClientStream is constructed only in RetriableStream.cancel() (L528), so each leaked entry provably committed via cancel() and was registered afterwards.

The window is measurable: widening the injected jitter from 4 µs to 400 µs takes the leak from 95 entries down to 1, consistent with a window of tens of microseconds that narrows further once startInternal is JIT-compiled.

Production impact

Found in a high-QPS Java service on 1.81.0: a heap dump held 499,418 leaked ManagedChannelImpl$ChannelStreamProvider$1RetryStream instances, roughly 14 GB, producing a GC death spiral. Because the retention is channel-scoped, the only recovery is a process restart.

Suggested fix

Register before the cancellable window, or make de-registration idempotent — any of:

1. Register in the constructor / newStream() rather than in start(), so add() happens-before any reachable cancel().
2. Re-check after prestart(): in RetriableStream.start(), if the stream is already committed immediately after prestart(), call postCommit() again.
3. Do not cancel an unstarted stream in ClientCallImpl — track "started" and defer the cancel until after stream.start().

(2) is the smallest change and is local to RetriableStream.start():

   Status shutdownStatus = prestart();
   if (shutdownStatus != null) {
     cancel(shutdownStatus);
     return;
   }
+  // cancel() may have committed this stream before prestart() registered it, in which case
+  // its postCommit() removed nothing -- and commit() is one-shot, so there is no second
+  // chance to de-register.
+  synchronized (lock) {
+    if (state.winningSubstream != null) {
+      postCommit();
+      return;
+    }
+  }

A regression test can assert that uncommittedRetriableStreams is empty after a cancel()-racing-start() sequence on an enableRetry() channel.

Related issues checked

I could not find an existing issue for this. Searched uncommittedRetriableStreams (0 hits), UncommittedRetriableStreamsRegistry (#4073 lock granularity, #7227 refactor, #7362 pending-call drain — none this), and RetriableStream leak cancel (#10209 framer/ByteBuf, #9340, #9185, #12891, #3839 — none this). #3537 / #3557 concern a different defect (ThreadlessExecutor discarding runnables after shutdown()), which I initially suspected here and then ruled out — 360,000 hedged blocking calls over Netty produced 0 dropped runnables and 0 registry growth, with GRPC_CLIENT_CALL_REJECT_RUNNABLE=true and a passing positive control for the drop detector.

LeakRepro.java (click to expand)

import io.grpc.*;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.stub.ClientCalls;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;

/**
 * Standalone reproducer for a permanent RetriableStream leak in
 * ManagedChannelImpl$UncommittedRetriableStreamsRegistry.
 *
 * Bug:
 *   ClientCallImpl.startInternal() assigns `stream = clientStreamProvider.newStream(...)`
 *   (~L250) but the stream only registers itself much later, from
 *   RetriableStream.start() -> prestart() -> registry.add(this), reached at ~L286.
 *   A ClientCall.cancel() landing in that window reaches RetriableStream.cancel(), which
 *   commits and runs the post-commit task INLINE -> postCommit() -> registry.remove(this),
 *   which is a NO-OP because the stream is not registered yet. start() then proceeds and
 *   registers it. commit() is one-shot ("if (state.winningSubstream != null) return null"),
 *   so no second post-commit task is ever produced and the entry is pinned for the life of
 *   the channel.
 *
 * Requires only an enableRetry() channel -- hedging is NOT required (see -Dpolicy=none).
 *
 * Prediction:
 *   mode=race     -> registry size grows monotonically and stays > 0 after quiescence
 *   mode=blocking -> registry size returns to 0 (no cancel races start())
 *
 * Usage: java LeakRepro <race|blocking|future> [iterations]
 *   NOTE: the mode is a POSITIONAL arg and defaults to "blocking"; running this class with
 *   no arguments exercises the control path and does NOT reproduce the leak.
 *   -Dpolicy=none disables hedging while keeping enableRetry().
 */
public class LeakRepro {

  static final MethodDescriptor<String, String> METHOD =
      MethodDescriptor.<String, String>newBuilder()
          .setType(MethodDescriptor.MethodType.UNARY)
          .setFullMethodName("leak.Svc/Call")
          .setRequestMarshaller(new StringMarshaller())
          .setResponseMarshaller(new StringMarshaller())
          .build();

  /** Trivial marshaller so the repro needs no protobuf dependency. */
  static final class StringMarshaller implements MethodDescriptor.Marshaller<String> {
    @Override public InputStream stream(String value) {
      return new ByteArrayInputStream(value.getBytes(StandardCharsets.UTF_8));
    }
    @Override public String parse(InputStream stream) {
      try {
        return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  }

  /**
   * A handler that accepts the call and then never responds. The client's deadline is
   * what ends every call, which is exactly the production shape (callee slow / hedged
   * attempts outliving the caller's deadline).
   */
  static ServerServiceDefinition blackHoleService() {
    ServerCallHandler<String, String> handler =
        (call, headers) -> {
          call.request(2);
          return new ServerCall.Listener<String>() {};
        };
    return ServerServiceDefinition.builder("leak.Svc")
        .addMethod(METHOD, handler)
        .build();
  }

  /** Hedging policy equivalent to what a D2/service-config-driven client would receive. */
  static Map<String, Object> hedgingServiceConfig(String hedgingDelay, int maxAttempts) {
    Map<String, Object> name = new LinkedHashMap<>();
    name.put("service", "leak.Svc");

    Map<String, Object> hedging = new LinkedHashMap<>();
    hedging.put("maxAttempts", (double) maxAttempts);
    hedging.put("hedgingDelay", hedgingDelay);
    hedging.put("nonFatalStatusCodes", Arrays.asList("UNAVAILABLE", "DEADLINE_EXCEEDED"));

    Map<String, Object> methodConfig = new LinkedHashMap<>();
    methodConfig.put("name", Collections.singletonList(name));
    methodConfig.put("hedgingPolicy", hedging);

    Map<String, Object> serviceConfig = new LinkedHashMap<>();
    serviceConfig.put("methodConfig", Collections.singletonList(methodConfig));
    return serviceConfig;
  }

  /** Retry policy — an alternative to hedging; also wraps the call in RetriableStream. */
  static Map<String, Object> retryServiceConfig(int maxAttempts) {
    Map<String, Object> name = new LinkedHashMap<>();
    name.put("service", "leak.Svc");

    Map<String, Object> retry = new LinkedHashMap<>();
    retry.put("maxAttempts", (double) maxAttempts);
    retry.put("initialBackoff", "0.1s");
    retry.put("maxBackoff", "1s");
    retry.put("backoffMultiplier", 2.0D);
    retry.put("retryableStatusCodes", Arrays.asList("UNAVAILABLE"));

    Map<String, Object> methodConfig = new LinkedHashMap<>();
    methodConfig.put("name", Collections.singletonList(name));
    methodConfig.put("retryPolicy", retry);

    Map<String, Object> serviceConfig = new LinkedHashMap<>();
    serviceConfig.put("methodConfig", Collections.singletonList(methodConfig));
    return serviceConfig;
  }

  /** Reflect down to ManagedChannelImpl.uncommittedRetriableStreamsRegistry.uncommittedRetriableStreams. */
  static Collection<?> registry(ManagedChannel ch) throws Exception {
    Object o = ch;
    while (!o.getClass().getName().equals("io.grpc.internal.ManagedChannelImpl")) {
      Field delegate = null;
      for (Class<?> c = o.getClass(); c != null && delegate == null; c = c.getSuperclass()) {
        for (Field f : c.getDeclaredFields()) {
          if (ManagedChannel.class.isAssignableFrom(f.getType())) {
            delegate = f;
            break;
          }
        }
      }
      if (delegate == null) {
        throw new IllegalStateException("cannot unwrap channel class " + o.getClass().getName());
      }
      delegate.setAccessible(true);
      o = delegate.get(o);
    }
    Field regField = o.getClass().getDeclaredField("uncommittedRetriableStreamsRegistry");
    regField.setAccessible(true);
    Object reg = regField.get(o);
    Field setField = reg.getClass().getDeclaredField("uncommittedRetriableStreams");
    setField.setAccessible(true);
    return (Collection<?>) setField.get(reg);
  }

  /**
   * For leaked entries, confirm they match the production heap signature:
   * RetriableStream.state.winningSubstream != null  (i.e. commit() DID run, so the
   * stream is not merely "in flight" -- it is committed-but-never-removed).
   */
  static String describeLeaked(Collection<?> reg) throws Exception {
    int committed = 0, inFlight = 0, total = 0;
    StringBuilder sample = new StringBuilder();
    for (Object stream : new ArrayList<>(reg)) {
      total++;
      Field stateF = null;
      for (Class<?> c = stream.getClass(); c != null; c = c.getSuperclass()) {
        try {
          stateF = c.getDeclaredField("state");
          break;
        } catch (NoSuchFieldException ignored) { }
      }
      if (stateF == null) continue;
      stateF.setAccessible(true);
      Object state = stateF.get(stream);
      if (state == null) continue;
      Field winF = state.getClass().getDeclaredField("winningSubstream");
      winF.setAccessible(true);
      Object win = winF.get(state);
      if (win != null) committed++; else inFlight++;

      if (total <= 3) {
        // Compare field-for-field against the production heap dump signature.
        sample.append(String.format("%n    sample#%d: ", total));
        for (String fn : new String[]{"cancelled", "passThrough", "hedgingFrozen",
                                      "hedgingAttemptCount", "buffer", "drainedSubstreams"}) {
          Field f = state.getClass().getDeclaredField(fn);
          f.setAccessible(true);
          Object v = f.get(state);
          if (fn.equals("drainedSubstreams") && v instanceof Collection) {
            v = "size=" + ((Collection<?>) v).size();
          } else if (fn.equals("buffer")) {
            v = (v == null ? "null" : "size=" + ((Collection<?>) v).size());
          }
          sample.append(fn).append('=').append(v).append("  ");
        }
        if (win != null) {
          Field sf = win.getClass().getDeclaredField("stream");
          sf.setAccessible(true);
          Object s = sf.get(win);
          sample.append("winningSubstream.stream=")
              .append(s == null ? "null" : s.getClass().getSimpleName());
        }
      }
    }
    return String.format("total=%d committed(winningSubstream!=null)=%d inFlight=%d%s",
        total, committed, inFlight, sample);
  }

  public static void main(String[] args) throws Exception {
    String mode = args.length > 0 ? args[0] : "blocking";
    int iterations = args.length > 1 ? Integer.parseInt(args[1]) : 200;
    long deadlineMs = 60;

    String serverName = InProcessServerBuilder.generateName();
    Server server = InProcessServerBuilder.forName(serverName)
        .addService(blackHoleService())
        .build()
        .start();

    // -Dpolicy=hedging (default) | retry | none
    // "retry" models a fallback where hedging.nonFatalStatusCodes is left blank.
    // "none" models enableRetry() with no policy at all — still wraps in RetriableStream.
    String policy = System.getProperty("policy", "hedging");
    InProcessChannelBuilder cb = InProcessChannelBuilder.forName(serverName);
    if ("off".equals(policy)) {
      cb.disableRetry();          // no RetriableStream is ever created
    } else {
      cb.enableRetry();
    }
    // -DhedgingDelay varies the hedging delay. The delay gates only when a hedge FIRES;
    // the timer is scheduled on every call regardless. The leak race happens at start()
    // before any hedge fires, so this knob makes no difference -- and -Dpolicy=none, which
    // removes hedging entirely, still leaks.
    String hedgingDelay = System.getProperty("hedgingDelay", "0.010s");
    if ("hedging".equals(policy)) {
      cb.maxHedgedAttempts(3).defaultServiceConfig(hedgingServiceConfig(hedgingDelay, 3));
    } else if ("retry".equals(policy)) {
      cb.maxRetryAttempts(3).defaultServiceConfig(retryServiceConfig(3));
    }
    ManagedChannel channel = cb.build();

    Collection<?> reg = registry(channel);
    System.out.printf("mode=%s iterations=%d deadline=%dms policy=%s hedgingDelay=%s%n",
        mode, iterations, deadlineMs, policy, hedgingDelay);
    System.out.printf("registry class=%s initial size=%d%n",
        reg.getClass().getName(), reg.size());

    // POSITIVE CONTROL for the probe itself: if the registry never goes above 0 even
    // transiently, then no RetriableStream is being created at all (hedging inactive),
    // and a final size of 0 says nothing about cleanup. Sample continuously.
    final int[] maxSeen = {0};
    Thread sampler = new Thread(() -> {
      while (!Thread.currentThread().isInterrupted()) {
        try {
          int s = reg.size();
          synchronized (maxSeen) { if (s > maxSeen[0]) maxSeen[0] = s; }
          Thread.sleep(0, 200_000);
        } catch (InterruptedException e) {
          return;
        } catch (ConcurrentModificationException ignored) {
          // HashSet is not thread-safe; a racing size() read may throw. Ignore and resample.
        }
      }
    });
    sampler.setDaemon(true);
    sampler.start();

    int deadlineExceeded = 0, other = 0;
    if (mode.equals("race")) {
      // Reproduce the cancel-during-startInternal window:
      // ClientCallImpl assigns `stream` (line ~250) well before calling stream.start()
      // (line ~286), which is where RetriableStream.prestart() registers the stream.
      // A cancel() landing in that window runs postCommit() -> registry.remove(this)
      // on a stream that has not been added yet, and the subsequent start() then adds
      // it -- permanently, because nothing will ever commit it again.
      ExecutorService cancellers = Executors.newFixedThreadPool(8);
      java.util.Random rnd = new java.util.Random(42);
      for (int i = 1; i <= iterations; i++) {
        CallOptions opts = CallOptions.DEFAULT.withDeadlineAfter(deadlineMs, TimeUnit.MILLISECONDS);
        final ClientCall<String, String> call = channel.newCall(METHOD, opts);
        final long jitter = rnd.nextInt(4000);
        cancellers.submit(() -> {
          java.util.concurrent.locks.LockSupport.parkNanos(jitter);
          try {
            call.cancel("race", null);
          } catch (RuntimeException ignored) { }
        });
        try {
          call.start(new ClientCall.Listener<String>() { }, new Metadata());
          call.request(2);
          call.sendMessage("req");
          call.halfClose();
          other++;
        } catch (RuntimeException e) {
          other++;
        }
        if (i % Math.max(1, iterations / 10) == 0) {
          System.out.printf("  after %5d calls: registry size = %d%n", i, reg.size());
        }
      }
      cancellers.shutdown();
      cancellers.awaitTermination(30, TimeUnit.SECONDS);
    } else {
      for (int i = 1; i <= iterations; i++) {
      CallOptions opts = CallOptions.DEFAULT.withDeadlineAfter(deadlineMs, TimeUnit.MILLISECONDS);
      try {
        if (mode.equals("blocking")) {
          ClientCalls.blockingUnaryCall(channel, METHOD, opts, "req");
        } else {
          Future<String> f = ClientCalls.futureUnaryCall(channel.newCall(METHOD, opts), "req");
          f.get(deadlineMs + 500, TimeUnit.MILLISECONDS);
        }
        other++;
      } catch (StatusRuntimeException e) {
        if (e.getStatus().getCode() == Status.Code.DEADLINE_EXCEEDED) deadlineExceeded++; else other++;
      } catch (ExecutionException e) {
        Throwable c = e.getCause();
        if (c instanceof StatusRuntimeException
            && ((StatusRuntimeException) c).getStatus().getCode() == Status.Code.DEADLINE_EXCEEDED) {
          deadlineExceeded++;
        } else other++;
      } catch (Exception e) {
        other++;
      }

      if (i % Math.max(1, iterations / 10) == 0) {
        System.out.printf("  after %4d calls: registry size = %d%n", i, reg.size());
      }
    }
    }

    // Give any late async cleanup a chance to run, so a slow-but-eventual removal is not
    // misread as a permanent leak.
    Thread.sleep(3000);
    System.gc();
    Thread.sleep(1000);

    sampler.interrupt();
    System.out.printf("PROBE CONTROL: max registry size ever observed during run = %d%n", maxSeen[0]);
    System.out.printf("FINAL mode=%s calls=%d deadlineExceeded=%d other=%d registrySize=%d%n",
        mode, iterations, deadlineExceeded, other, reg.size());
    System.out.println("leaked entry shape: " + describeLeaked(reg));

    channel.shutdownNow();
    server.shutdownNow();
  }
}