ChannelSendOperator never cancels the source subscription when the write result errors before READY_TO_WRITE, leaking infinite streams on client disconnect #37527

Affects: verified on spring-web 7.0.9 (Spring Boot 4.1.1); the relevant code is unchanged on main, and the same structure exists in 6.x.

Summary

When the write function's result publisher signals onError before the HTTP server adapter's write subscriber has subscribed (i.e. while ChannelSendOperator.WriteBarrier is in state FIRST_SIGNAL_RECEIVED), the error is forwarded to the completion subscriber and the cached data item is released, but the subscription to the source publisher is never cancelled. For an infinite source (SSE stream, WebSocket-style feed, anything fed by a shared Sinks.Many), the whole upstream chain stays subscribed forever: doFinally/doOnCancel on the source never fire, and shared-sink subscriber lists grow monotonically.

With Reactor Netty this is hit by a mundane event: the client disconnects between the stream's first onNext and the attachment of the write subscriber. ChannelOperations.send* then short-circuits with

if (!channel().isActive()) {
    return then(Mono.error(AbortedException.beforeSend()));
}

so writeFunction.apply(this) returns an already-failed publisher, and result.subscribe(writeCompletionBarrier) delivers onError immediately. Nothing is logged (the error reaches an already-terminated connection and is dropped), and nothing upstream is cleaned up.

Where the cancel is lost

ChannelSendOperator.WriteCompletionBarrier (spring-web 7.0.9, identical on main):

@Override
public void onError(Throwable ex) {
    try {
        this.completionSubscriber.onError(ex);
    }
    finally {
        this.writeBarrier.releaseCachedItem();   // <-- releases the item, but never
    }                                            //     cancels writeBarrier.subscription
}

@Override
public void cancel() {
    this.writeBarrier.cancel();                  // <-- the only path that cancels the source
    ...
}

The catch (Throwable ex) blocks around writeFunction.apply(...) in WriteBarrier.onNext/onComplete have the same gap: they call writeCompletionBarrier.onError(ex) without cancelling the source subscription.

The leak requires the first signal to be onNext (after onComplete/onError the source has terminated itself, so nothing is left to cancel).

Minimal reproduction (pure spring-web unit test, no server needed)

@Test
void sourceIsCancelledWhenWriteResultErrorsBeforeReadyToWrite() {
    AtomicReference<SignalType> finallySignal = new AtomicReference<>();

    Flux<String> source = Flux.just("event-1").concatWith(Flux.never())
            .doFinally(finallySignal::set);

    Mono<Void> write = new ChannelSendOperator<>(source,
            p -> Mono.error(new IOException("Connection has been closed BEFORE send operation")));

    StepVerifier.create(write)
            .expectErrorMessage("Connection has been closed BEFORE send operation")
            .verify();

    // Expected: SignalType.CANCEL. Actual: null — the source is still subscribed
    // and, being infinite, stays subscribed forever.
    assertThat(finallySignal.get()).isEqualTo(SignalType.CANCEL);
}

Walkthrough: WriteBarrier.onSubscribe requests 1 → onNext("event-1") → state NEW → FIRST_SIGNAL_RECEIVED, item cached → writeFunction.apply(this) returns the error Mono → result.subscribe(writeCompletionBarrier) → onError → item released, error forwarded — source subscription left dangling.

How this bites in production

WebFlux SSE endpoint on Reactor Netty whose per-connection stream merges a shared Sinks.many().multicast().directBestEffort() with a heartbeat Flux.interval(...).share(). A load test with ~1,600 open SSE connections that are all closed simultaneously by the client (plus a few new connections still arriving during the teardown) reliably leaks the connections whose first merged event arrives just after their channel was closed — 10 out of 1,600 in our run. For each of them, a heap dump (taken after a full GC, so all objects reachable) shows the frozen state predicted above:

• WriteBarrier: state = FIRST_SIGNAL_RECEIVED, item = null (released by the error path), error = null, completed = false, writeSubscriber = null, upstream subscription still set and never cancelled;
• no MonoSendMany$SendManyInner exists for these responses — the write subscriber never subscribed;
• the per-connection HttpServerOperations: connection = DisposedConnection, inboundDone = true — Reactor Netty fully terminated its side;
• upstream of the barrier: FluxFlatMap$FlatMapMain with cancelled = false, wip = 0, FluxDoFinally$DoFinallySubscriber with once = 0, and the sink's inner subscriber still present in SinkManyBestEffort.subscribers[] with cancelled = false.

The GC-root path of every leaked chain runs through the application's singleton sink, so the retained set (including closed NioSocketChannels referenced back through the barrier) survives indefinitely. Besides the memory, every doOnSubscribe/doFinally-based connection gauge drifts upward by the number of leaked chains per incident, since the doFinally is never reached.

Nothing appears in the logs: the AbortedException is delivered to the completion subscriber after Reactor Netty has already terminated the connection, so it is dropped silently.

Suggested fix

In WriteCompletionBarrier.onError, cancel the write barrier instead of only releasing the cached item:

@Override
public void onError(Throwable ex) {
    try {
        this.completionSubscriber.onError(ex);
    }
    finally {
        this.writeBarrier.cancel();   // cancels the source subscription (if any) and releases the cached item
    }
}

WriteBarrier.cancel() is idempotent (it nulls the subscription and releases the item), and in READY_TO_WRITE the source is normally already cancelled through the write subscriber's own cancellation, so the extra call is harmless there. The catch blocks around writeFunction.apply(...) in WriteBarrier.onNext/onComplete need the same treatment.

Related

• Error signal not propagated if writeFunction in ChannelSendOperator fails immediately #23175 made sure the error from an immediately-failing write function reaches the completion subscriber — this report is about that same error path not releasing the source.
• spring-cloud-gateway#2221 shows the AbortedException → WriteCompletionBarrier.onError route occurring in the wild on client disconnects.

Environment: Spring Boot 4.1.1, spring-web/spring-webflux 7.0.9, reactor-core 3.8.7, reactor-netty 1.3.7, netty 4.2.17 (NIO transport), OpenJDK 25; observed on macOS aarch64 and Linux/Docker.