Spring Boot 2.3.0.RC1 - RSocket over Websocket transport - requestStream - produces additional request(255)

[maxim-bandurko-lsvt]

Had created the issue at Spring Boot, but may be this should be posted here:
spring-projects/spring-boot#21360

With latest 2.3.0.RC1 RSocket "requestStream" started producing additional request to Flux inside controller with quantity 255.

Configuration to reproduce the issue:

Spring Boot 2.3.0.RC1
RSocket plugged to server
Transport - Websockets
Using @MessageMapping controller
RSocket connection with Composite Metadata ('message/x.rsocket.composite-metadata.v0'), Binary MimeType ('application/octet-stream') and Buffer encoders
Tested with javascript RSocket client v0.0.19.

Previous version of Spring Boot 2.3.0.M4 hasn't such issue.

[OlegDokuka]

Hey @maxim-bandurko-lsvt.

This is expected.
It does not produce values over the network but produces it and keep in the local queue. We use the operator limitRate to ensure that a responder stream does not overflow the connection bandwidth when a request sends Long.MAX_VALUE.

Is such behavior breaks your business logic?

Cheers,
Oleh

[maxim-bandurko-lsvt]

Hello Oleg @OlegDokuka

Thing is that client is not sending request Long.MAX_VALUE. It is sending request(1) or request(10), etc. And Java server is getting this amount of request, but additionally triggers one more request(255). But server doesn't deliver 256 requests, it delivers only 1 to client (that was requested by it). So, other 255 got prefetched.

Here is a basic example on server side:

@MessageMapping("test")
    public Flux<DataBuffer> test(DataBuffer request) {
        
        return Flux.just("a", "b", "c", "d", "e", "f")
        .map(
            data -> {
                log.info("next: " + data);
                return new DefaultDataBufferFactory().wrap(data.getBytes());
            }
        )
        .doOnRequest(
            consumer -> {
                log.info("doOnRequest: " + consumer);
            }
        );
        
        
    }

And here is a log on server side:

doOnRequest: 1
next: a
doOnRequest: 255
next: b
next: c
next: d
next: e
next: f

Javascript client was requesting only 1 element and got only 1, but because on Java side were originated request of 256 in total, all of them are got processed and stored in cache.

In previous Spring Boot version (2.3.0.M4) that had rsocket-core-1.0.0-RC6 with rsocket-transport-netty-1.0.0-RC6 everything was working properly and how it supposed to do, there were no additional request(255) at server side, so backpresure was correct to whole inner streams flow logic.

With new Spring Boot version (2.3.0.RC1) that has rsocket-core-1.0.0-RC7 with rsocket-transport-netty-1.0.0-RC7 everything got changed and stream is getting additional request 255 internally, that is propagated to whole inner streams flow logic.

Based on this issue, server prefetches 255 items internally that got not actual data. As client can do request(10) for example in one hour later, and will get 10 elements that were generated 1 hour earlier with old data.

If such prefetch logic is designed on purpose, is there any way to turn it off for certain situations?

Thank you.

[OlegDokuka]

As I said this is expected. It is done to prevent potential cases when the possible overwhelming may happen in case there is Long.MAX_VALUE requested.

It means that starting from RC7 it will always pre-request in advance 256 elements regardless the number of elements was originally requested. (applies for requestStream and requestChannel cases only)

This is the result of applying the '.limitRate' operator which enables such behavior.

I mean we did that on purpose but I guess if it is critical for you, we can have a configuration to disable it, of course.

[maxim-bandurko-lsvt]

@OlegDokuka I see.

How can I configure to disable it? That is some kind critical and breaks all initial rules of backpressure logic that were originally developed.

Thank you!

[OlegDokuka]

Unfortunately, for now, you can not. I will create a PR that should provide a way to configure that.

Please stay tuned

[maxim-bandurko-lsvt]

@OlegDokuka btw, may be limitRate can have an option, to not add additional requests if the rate is not reached? I mean, in situation, if requested is lower the minimal rate.

[OlegDokuka]

@maxim-bandurko-lsvt

Previously we had such a mechanism, but then we decided to use a built-in option from ProjectReactor.

cc/ @simonbasle @rstoyanchev

[maxim-bandurko-lsvt]

@OlegDokuka I see,

So, it may be even makes sense to have such mechanism be implemented inside limitRate at ProjectReactor.

Other thing, is that such limiter is needed only at certain level of inner streams, not in the root level. Like, directly between the database operations, etc. So, developer can make a discussion himself, at what exact place to have such limiter injected. As if that will be injected on default, with not correct propagation of backpressure to server, it will break a lot of existing logic implemented by developers. And also, it makes confusing, as in some places using Reactive Streams backpressure in RSocket is not working as it supposed to be.

[OlegDokuka]

Funny) Feel free to leave your comment on my PR -> reactor/reactor-core#1879

[OlegDokuka]

Oh, well, you have already left one, which I was not aware of

[OlegDokuka]

Anyways, let me talk with @rstoyanchev on the closest day to provide a proper resolution for that.

1.0.0 is scheduled for the next Monday, so I really hope that we can solve it by that time

[maxim-bandurko-lsvt]

@OlegDokuka Thank you very much!

Also, is PR 1.0.0 will be inside upcoming PR of Spring Boot 2.3.0?

[OlegDokuka]

Yup. 1.0.0 is exactly dedicated for the SB release

[maxim-bandurko-lsvt]

@OlegDokuka Awesome! Thanks!!!