Timeout leaves connection in the pool in configured state, blocking a pool slot about reactor-netty

[chemicL]

It looks like for 9ms you hit a race where the cancellation reaches your doOnCancel but at the same time the connection is being delivered. That's my expectation as the logs show the connection was established. The connect timeout would apply if the connection didn't succeed to be established, but it succeeds and is delivered to your delayElement() operator. That operator keeps a reference to the connection. When the delay is longer than the argument to timeout() it means that the scheduled task that holds the reference to the connection is disposed and the reference is discarded. If there is no discard hook defined on the reactive pipeline then the reference is just abandoned and the connection is not returned to the pool. That's my understanding of the situation. reactor-netty's timeout configuration have no insight into the pipeline that you attach to the results and can't control your code that refers to the connection.

Just trying to help - please check if adding the discard hook that releases the connection helps.

[violetagg]

@jkippen With this reproducible example, I'm seeing that the connection is always emitted but your pipeline delays it intentionally. If I add log like below I can see that connection was delivered.

Mono
				.defer(() -> client
						.connect()
						.log()
						.doOnCancel(() -> log.debug("********* cancelling"))
...

As the connection is delivered, you have to guarantee that the connection is disposed regardless whether it is cancellation, error or normal flow.

The logs:

10:01:02.562 [reactor-tcp-nio-3] DEBUG r.n.r.DefaultPooledConnectionProvider - [054ae908, L:/127.0.0.1:50560 - R:localhost/127.0.0.1:5555] onStateChange(PooledConnection{channel=[id: 0x054ae908, L:/127.0.0.1:50560 - R:localhost/127.0.0.1:5555]}, [connected])
10:01:02.562 [reactor-tcp-nio-3] INFO  reactor.netty.tcp.TcpServerTests - Client Connection State: [id: 0x054ae908, L:/127.0.0.1:50560 - R:localhost/127.0.0.1:5555] // [connected] // reactor-tcp-nio-3
10:01:02.569 [reactor-tcp-nio-3] DEBUG r.n.r.DefaultPooledConnectionProvider - [054ae908, L:/127.0.0.1:50560 - R:localhost/127.0.0.1:5555] onStateChange(ChannelOperations{PooledConnection{channel=[id: 0x054ae908, L:/127.0.0.1:50560 - R:localhost/127.0.0.1:5555]}}, [configured])
10:01:02.570 [reactor-tcp-nio-4] DEBUG reactor.netty.tcp.TcpServer - [3d8e3123, L:/127.0.0.1:5555 - R:/127.0.0.1:50560] Handler is being applied: reactor.netty.tcp.TcpServerTests$$Lambda$407/1621513804@55d2037f
10:01:02.570 [reactor-tcp-nio-3] INFO  reactor.Mono.Create.2 - onNext(ChannelOperations{PooledConnection{channel=[id: 0x054ae908, L:/127.0.0.1:50560 - R:localhost/127.0.0.1:5555]}})
10:01:02.570 [reactor-tcp-nio-4] INFO  reactor.netty.tcp.TcpServerTests - ******** Server receive ChannelOperations{SimpleConnection{channel=[id: 0x3d8e3123, L:/127.0.0.1:5555 - R:/127.0.0.1:50560]}}
10:01:02.570 [reactor-tcp-nio-3] INFO  reactor.Mono.Create.2 - onComplete()
10:01:02.570 [reactor-tcp-nio-3] INFO  reactor.netty.tcp.TcpServerTests - Client Connection State: [id: 0x054ae908, L:/127.0.0.1:50560 - R:localhost/127.0.0.1:5555] // [configured] // reactor-tcp-nio-3

[jkippen]

I may have not explained the issue correctly - I intentionally add a delay to simulate a firing of the Timeout before the connection is delivered, to try and see if there is a race condition between connect() delivering a connection, Timeout firing, and the connection not being aborted.

Even when the doOnCancel fires, I still see this connection hanging around in configured state, taking a slot in the connection pool.

[jkippen]

The doOnCancel I just added for some observability into the abort of the connection

[violetagg]

@jkippen With the current example I cannot reproduce what you are explaining. It seems that you can, so please provide logs from an execution where you see the behaviour that you are explaining.

[jkippen]

I added a couple screenshots from my client and server logs, is it visible?

[jkippen]

Here's smaller snippet of code if it helps. In the client log I attached above, the first exception is the Timeout firing, and the second is the acquire timeout from the second connect() attempt.

ConnectionProvider provider = ConnectionProvider.create("test", 1, true);

/// This connect times out, leaves a connection in configured state
Mono
        .defer(() -> client
                        .connect()
                        .delayElement(Duration.ofMillis(1000))
                        .timeout(Duration.ofMillis(900))                      
                        .flatMap(connection -> sendAndReceive(connection))
        ).block();

/// This connect cannot acquire a connection
 Mono
        .defer(() -> client
                        .connect()
                        .flatMap(connection -> sendAndReceive(connection))
                        .doOnError(e -> {
                            e.printStackTrace();
                        })     
                        .onErrorComplete()
                )
                .block();

[jkippen]

oh I see what you're saying - I removed some stuff to share cleaned up code... will share again in a moment

[chemicL]

I have a feeling this issue is very similar to this one from reactor-pool. Please have a look at my comment, perhaps using .doOnDiscard() you will be able to fix it.

[violetagg]

I want to see logs from connect(), whether we delivered the connection or not. So @jkippen please add .log() after the connect()
I need to know whether the issue is a cancelation while the connection still is not delivered or after that.

[jkippen]

Okay, here's the code for the first connect() attempt that times out (I had removed onErrorComplete - will add log and send the output shortly

  // timeout via reactive        
        Mono
        .defer(() -> client
                        .connect()
                        .doOnCancel(() -> log.debug("********* cancelling"))
                        .delayElement(Duration.ofMillis(Random.from(RandomGenerator.getDefault()).nextInt(900, 1000)))
                        .timeout(Duration.ofMillis(9))                      
                        .flatMap(connection -> sendAndReceive(connection))
                        .doOnError(e -> {
                            e.printStackTrace();
                        })
                        .onErrorComplete()
        ).block();

[jkippen]

Here's the log with log() output

[jkippen]

@chemicL to be honest, I'm assuming that using the connect timeout on the TcpClient itself would avoid the problem (though I'm not sure, and not sure how to validate that). Even so (and I don't know much about Reactor), if the responsibility of timeout(..) is to cancel the mono, shouldn't it abort the connection regardless of whether it was emitted from connect() or not?

[jkippen]

In my example, if you change the timeout to something closer to the delay (i.e. 900ms) the doOnCancel also doesn't run. In the output of my last test, doOnCancel did run (the timeout was 9ms)

Here's an output of the log when doOnCancel doesn't run, if it's helpful

[violetagg]

@jkippen One additional question - you are testing with this ancient version -> implementation platform('io.projectreactor:reactor-bom:2020.0.9') or with the latest releases?

[jkippen]

Ah - I was using the ancient version. However, I updated my gradle version to io.projectreactor:reactor-bom:2023.0.4, cleaned my workspace, broke my workspace, fixed it, and retested. Got the same result.

[violetagg]

@jkippen I agree with the comment made by @chemicL. You need to handle cancellation and error events. From Reactor Netty point of view it is dangerous to handle these events and just close the connection, We don't know what kind of protocol is implemented and whether this protocol needs to do some cleanup in these situations.