reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)

[lkolisko]

Using RSocket REQUEST_STREAM to implement a query where results are streamed back to the client. The client sends REQUEST_N frame 512. It is successfully receiving results. Once the client successfully receives >256 results, it requests an additional 256. The communication works fine. FrameLogger shows expected receiving and sending frames. After some time the following exception is thrown on the event of additional REQUEST_N on client;

2020-11-18 13:27:47.944 DEBUG 1 --- [tor-tcp-epoll-2] io.rsocket.FrameLogger                   : receiving -> 
Frame => Stream ID: 1 Type: NEXT Flags: 0b100000 Length: 296
Data:
......

020-11-18 13:27:47.945 DEBUG 1 --- [pool-5-thread-5] io.rsocket.FrameLogger                   : sending -> 
Frame => Stream ID: 1 Type: REQUEST_N Flags: 0b0 Length: 10 RequestN: 256
Data:

2020-11-18 13:27:47.956 ERROR 1 --- [tor-tcp-epoll-2]  : 

reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)
	at reactor.core.Exceptions.failWithOverflow(Exceptions.java:221)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.FluxSourceFuseable] :
	reactor.core.publisher.Flux.from(Flux.java:957)
	io.rsocket.core.RSocketRequester.handleRequestStream(RSocketRequester.java:338)
Error has been observed at the following site(s):
	|_        Flux.from ? at io.rsocket.core.RSocketRequester.handleRequestStream(RSocketRequester.java:338)
	|_ Flux.subscribeOn ? at io.rsocket.core.RSocketRequester.handleRequestStream(RSocketRequester.java:393)

This happens as part of a large application. Threfore I do not have a standalone replication example at the moment.

• RSocket version(s) used: 1.0.3
• Other relevant libraries versions: spring-boot: 2.3.5, netty 4.1.53
• Platform: OpenJDK 64-Bit Server VM AdoptOpenJDK (build 11.0.8+10, mixed mode)

[OlegDokuka]

Looks like it is possible with fusion enabled. I will try to arrange something similar with the .publisheOn operator to see if this reproduces

[koldat]

It is happening for us on more places. So far did not found any workaround. How to disable fusion?

[koldat]

@OlegDokuka why is in handleRequestStream/handleChannel the assumption that the receiver queue has bounded size 1? I would understand it in handleRequestResponse, but not in Flux variants.

io.rsocket.core.RSocketRequester.handleRequestStream(Payload)

final UnicastProcessor<Payload> receiver = UnicastProcessor.create(Queues.<Payload>one().get());

Handle frame function can consume network frames as fast as it can so it can happen that processor did not yet dispatched previous frame downstream. When I used unbounded queue it started working without exception.

[OlegDokuka]

@koldat

It is happening for us on more places. So far did not found any workaround. How to disable fusion?

There is no way to do so on your end.
For now, to work around what you have, you can try to add a RequesterInterceptor like the following :

new RSocketProxy(source) {

   Flux<Payload> requestStream(Paload p) {
      return source.requestStream(p).hide();
   }

   Flux<Payload> requestChannel(Flux<Paload> p) {
      return source.requestChannel(p).hide();
   }
	
}

please find a full example here -> https://github.com/rsocket/rsocket-java/blob/master/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/plugins/LimitRateInterceptorExample.java

The .hide operator should disable fusion and no issue should appear after that.

@OlegDokuka why is in handleRequestStream/handleChannel the assumption that the receiver queue has bounded size 1? I would understand it in handleRequestResponse, but not in Flux variants.

io.rsocket.core.RSocketRequester.handleRequestStream(Payload)

final UnicastProcessor receiver = UnicastProcessor.create(Queues.one().get());
Handle frame function can consume network frames as fast as it can so it can happen that processor did not yet dispatched previous frame downstream. When I used unbounded queue it started working without exception.

there is a strict reactive streams relationship. If you requested N elements, you clearly stated that you are ready to receive N elements. Following reactive streams, the producer follows your demand and produce exactly N elements (or less).

That said, there is no reason for us (rsocket core logic) to cache N elements anywhere before your events handler. That is why to avoid any redundant allocations we use a queue of a single element since we expect no values will be enqueued at all

The mentioned issue is the result of Reactor optimizations that we have not taken into account. Which fuse with the unicastProcessor queue and does not take into account the queue capacity

[koldat]

@OlegDokuka I am not sure I described that properly.

Ofcourse I know that N requested elements are ready to be consumed. But the problem is that there is also asynchronous effect. So downstream can consume, but it can take a time. Our application uses usual backpressure (512 elements) and normal operators.

As I said issue is fixed when queue is unbounded, because in rare case it hits the condition where in queue there is one element still not dispatched (randomly after 10k - 1M items received). But it also sometimes happen with much shorter streams.

I tries to remove fusion everywhere. I will try also your suggestion, but again:
why is in handleRequestStream/handleChannel the assumption that the receiver queue has bounded size 1? When receiving side is processed in other threads than consuming side.

[OlegDokuka]

here is the point. Your backpressure operator is responsible for storing up to 512 elements, RSocket has no clue about your request. it just transparently transforms your request N into a binary frame and ensures that that frame is decoded back upon receiveal into the same requestN method invocation on your subscription on the receiver side.

The problem is that Reactor is a smart framework and tries to do lots of optimizations.
That said, if you have a workflow of UnicastProcessor[Queue] -> PublishOn[Queue]

then Reactor will optimize your workflow and make it as the following PublishOn[UnicastProcessor[Queue]], which means that 1 less queue will be allocated.

But here comes the problem. We use UnicastProcessor just because it is a convenient API that lets us send elements manually. On the other hand, we don't need to store any elements, because we know the strict relationship between publisher and subscriber, which says that if subscriber say request(5) it is ready to consume(5) if later on, it says request(10), but previous 5 was not yet delivered, it means it is ready to receive 15 (5 + 10) (and again we don't have to store anything).

The problems come from the fact that UnicastProcessor fuses with any operator which creates its queue. This means now if we create queue(1) and publishOn wanted queue(512), the UnicastProcessor will have no clue about that requirement. hence we see overflow exceptions because of requirements mismatch.

As I said, to avoid this issue for now, just place the .hide operator before your publishOn or any other operator like groupBy. which can fuse with our internal UnicastProcessor.

Also, feel free to migrate to 1.1.0 which provides custom implementations of request operators and does not use UnicastProcessor anymore

[OlegDokuka]

On the other hand, we have subscribeOn between UnicastProcessor and your logic. SubscribeOn should disable fusion. Thus now I'm at the point of I don't have a clue what is going on there.

@lkolisko appologies for the confusion. Let me reproduce that issue locally and then I can provide you with more updates. For now please, use either 1.0.2 or 1.1.0

[koldat]

I am working with Lukas on this issue. We use spring-boot unfortunately and I am not sure we can upgrade to 1.1.0 and it brings some incompatibilities on onterfaces. On the other hand we tried 1.0.2 and it did not help. We placed ".hide" almost everywhere without any effect as we cannot go under "subscribedOn" in handleRequestStream. The only fix that helped is to use unbounded queue in handleRequestStream to cover that glitch. In worst case there will be 2 elements in memory, but still better to not crash the stream.

[koldat]

The change (unbounded vs. one()) was introduced in 18050ed after 1.0.2. Hmm... We will try again if it is really happening even for 1.0.2.

[OlegDokuka]

@koldat @lkolisko It should not happening in 1.0.2. We will do our best to land a fix for that ASAP and do 1.0.4

[koldat]

You are right, we have just verified that. The issue was that we got same exception text (overrun by more signals than expected) from server side as well thus were confused. So it looked the same, but was not. It means similar problem is also on the other side (RSocketResponder) . When we changed server and client issue was solved.

Thanks a lot for looking into this.

[OlegDokuka]

@koldat one more question. Do you have any custom publishers on the producer side?

I did a couple of different test cases, and I can not reproduce this issue. Also, we have an extensive number of tests inside RSocket which means if we have a bug, it would eventually appear on the CI but it's not. Thus, I suspect that something is on your end. Thus to make sure it is our bug, I would appreciate if you share more details on what kind of operators do you use: what is the producer, what is the consumer; what Reactor operators do you use (if not reactor, what libraries do you use in combination with rsocket)

[koldat]

We use Spring boot 2.3.5. Flux we return is flatmap:

    public Flux<MetadataMetricQueryMatch> query(Mono<MetadataMetricQueryReactiveRequest> request, RoleBasedSession session) {
        return request.flatMapMany(query ->  metadataManager.executeAsFlux(session, query));
     }

where inner flux is using and fromIterable:

        return Flux.using(() -> new AtomicReference<>(new DatabaseIteratorContext()),
                    DatabaseIterator<MetadataMetricQueryMatch> it = execute(session, query, matcher, ctx.get(), false);
                    return Flux.fromIterable(() -> it);
            }, ctx -> {
                ctx.get().close();
            });