Fix reentrant subscribe in StreamingNettyByteBody

[yawkat]

If a subscriber to a split streaming body writes a response when reading data from the body, this can ultimately lead to another split of the same body being closed. Closing the body in turn is a subscribe operation, which must not happen during a read in a reentrant fashion. This would lead to a failure in the assertion that guards against such reentrant operations (assert !working;), and various downstream issues like buffer leaks. In particular, MaxRequestSizeSpec was affected by this bug occasionally.

This patch replaces the use of EventLoopFlow with more suitable code. In particular, EventLoopFlow does not support reentrant or concurrent calls, it only ensures serialization. The new logic supports reentrant or concurrent calls and still ensures serialization where it matters.

The new test does not work yet due to netty/netty#13730 . This PR is a draft until that patch is released.

[graemerocher]

@yawkat now that the netty upgrade is done can this proceed?

[yawkat]

Sure I'll update

[sonarqubecloud]

Quality Gate failed

Failed conditions
64.3% Coverage on New Code (required ≥ 70%)

See analysis details on SonarCloud

[shawnone]

@yawkat Hi, it this issue merged to 4.7? I am using 4.7 and reproduce this issue

[yawkat]

This is fixed in 4.7, yes. You are likely seeing a different issue.

[mwz1994]

This is fixed in 4.7, yes. You are likely seeing a different issue.

Maybe it's a similar exception.
4.7.5 throws the following stack message.But 4.6.x works fine

ExceptionMessage:
Cannot invoke "io.micronaut.http.netty.EventLoopFlow.executeNow(java.lang.Runnable)" because "this.flow" is null
ExceptionStackTrace:
java.lang.NullPointerException: Cannot invoke "io.micronaut.http.netty.EventLoopFlow.executeNow(java.lang.Runnable)" because "this.flow" is null
at io.micronaut.http.client.netty.StreamWriter.add(StreamWriter.java:90)
at io.micronaut.http.netty.body.StreamingNettyByteBody$SharedBuffer.subscribe0(StreamingNettyByteBody.java:348)
at io.micronaut.http.netty.body.StreamingNettyByteBody$SharedBuffer.subscribe(StreamingNettyByteBody.java:326)
at io.micronaut.http.netty.body.StreamingNettyByteBody.primary(StreamingNettyByteBody.java:92)
at io.micronaut.http.client.netty.StreamWriter.(StreamWriter.java:53)
at io.micronaut.http.client.netty.DefaultHttpClient.lambda$sendRawRequest$48(DefaultHttpClient.java:1667)
at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:61)
at reactor.core.publisher.Mono.subscribe(Mono.java:4576)
at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:126)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:96)
at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:106)
at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:1583)

[yawkat]

looks like a bug, please make a new issue.

[mwz1994]

looks like a bug, please make a new issue.

Thanks for the reply, I created a new issue. （#11364）