Add ExchangeFilterFunction that enforces limit on the response size [SPR-16989]

[spring-projects-issues]

Sergey Galkin opened SPR-16989 and commented

I'm sending lots of http request with notifications to other system. I expected small acknowledgment in response (less than 1 KB), but from time to time one of this system start to return huge responses (several MB), which provide no value to me and only consume resources.

For such cases it will be handy if I will be able to configure WebClient to read only limited amount of data from response and keep using bodyToMono method:

retrieve(1024).bodyToMono()

Currently we to replaced bodyToMono with custom extractor as a workaround.

...
.exchange()
.flatMap(response -> response
  .body((t, m) -> t
  .getBody()
  .takeUntil(buffer -> byteCounter.addAndGet(buffer.readableByteCount()) > logLimit))
.collect(...)

---

Affects: 5.0.7

Issue Links:

• DataBufferUtils#takeUntilByteCount checks one extra buffer but does not release it [SPR-17118] #21655 DataBufferUtils#takeUntilByteCount checks one extra buffer but does not release it
• DataBufferUtils#join could leak buffers in case of error from the source [SPR-17025] #21563 DataBufferUtils#join could leak buffers in case of error from the source

Referenced from: commits aec9826

[spring-projects-issues]

Sergey Galkin commented

My workaround is not byte-precise because WebClient have no control over netty's buffer size (see io.netty.channel.DefaultChannelConfig and  io.netty.channel.AdaptiveRecvByteBufAllocator). Maximum buffer size is 65K which is OK for me. If "non-precise limit" is OK as general solution, a can make a PR.

[spring-projects-issues]

Rossen Stoyanchev commented

So do you use this with existing HttpMessageReader or Decoder implementations, such that if the amount of data is less than the limit then all is good, or do you also have a custom Decoder to go with it? What happens if the limit is exceeded? At that point the content to decode from is incomplete, and only a custom decoder could get what it needs and still produce a response object, essentially ignoring the rest of the content but closing the connection.

One way or another overloading retrieve doesn't seem like the right place. Someone might want the same option when going through exchange() but first let me understand this better. For example if it is only a matter of detecting more than a certain amount of bytes and raising an error, then an ExchangeFilterFunction might be something to try. 

 

[spring-projects-issues]

Sergey Galkin commented

1.  I do not use existing HttpMessageReader / Decoder implementations, custom decoder is used.
2. HTTP body is used for logging purpose. In case of limit exceed we log part of the body and append "... (truncated)".
3. In future we may need to deserialize body, so preferred behavior is following:
   1. below limit: bodyToMono returns deserialized object
   2. above limit: some kind of TooLargeEntityException which contains body (part until limit)

[spring-projects-issues]

Rossen Stoyanchev commented

This seems like a good fit for a client filter which can be applied independent of what part of the API is used (e.g. exchange vs retrieve). We could include a built-in implementation of such a filter like we do for basicAuthentication in ExchangeFilterFunctions.

[spring-projects-issues]

Sergey Galkin commented

I've tried with client filter. It looks better then my initial solution:

    class ResponseBodyLimitFunction implements ExchangeFilterFunction {

        private final int bodyByteLimit;

        ResponseBodyLimitFunction(int bodyByteLimit) {
            this.bodyByteLimit = bodyByteLimit;
        }

        @Override
        public Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
            return next.exchange(request).flatMap(this::filter);
        }

        private Mono<ClientResponse> filter(ClientResponse response) {
            final AtomicInteger byteCounter = new AtomicInteger(0);

            Flux<DataBuffer> buffers = response.body((message, ctx) -> message
                    .getBody()
                    .takeUntil(buffer -> byteCounter.addAndGet(buffer.readableByteCount()) > bodyByteLimit))
                    .collectList()
                    .map(list -> {
                        if (byteCounter.get() > bodyByteLimit) {
                            throw new TooLargeHttpBodyException(toByteArray(list, byteCounter.get()));
                        } else {
                            return list;
                        }
                    })
                    .flatMapMany(Flux::fromIterable);

            return Mono.just(ClientResponse.create(response.statusCode()).body(buffers).build());
        }

        private byte[] toByteArray(List<DataBuffer> dataBuffers, int bytesCount) {
            ByteBuffer total = ByteBuffer.allocate(bytesCount);
            dataBuffers.forEach(buffer -> total.put(buffer.asByteBuffer()));
            return total.array();
        }
    }

Is it worth to create PR to incorporate this behavior into ExchangeFilterFunctions?

 

[spring-projects-issues]

Rossen Stoyanchev commented

Modified title: was: "Response size limit for org.springframework.web.reactive.function.client.Webclient"

[spring-projects-issues]

Rossen Stoyanchev commented

Yes I think this would be good to have.

For the implementation, when the exception is thrown the data buffers would need to be released since nothing else will do that after. However I'd prefer if by default the filter did not buffer at all but simply counted the number of accumulated bytes + enforced the configured limit.

Buffering becomes a separate concern. It's quite easy to do via DataBufferUtils#join which some decoders do already but one could also add a filter to buffer the content in order to shield from errors which could happen for reasons other than the size limit.

The consumer will have the (partial) content at that point, so no need to include it in the exception. Also since the response may have a content-length header, it may be possible to enforce the limit even without reading any content.

[spring-projects-issues]

Rossen Stoyanchev commented

I've also created #21563 which if fixed will make it quite easy to add DataBufferUtils#join to add the buffering with an Exception containing the content received before the limit.

[spring-projects-issues]

Sergey Galkin commented

Rossen Stoyanchev Please take a look at #1909

[spring-projects-issues]

Rossen Stoyanchev commented

Hm, this looks more like a 'buffering" or "aggregating" filter. For a "limit" filter shouldn't there be an exception, or else the consumer would not know the content was truncated.

[spring-projects-issues]

Sergey Galkin commented

I've added throwing exception as an option and updated PR.

[spring-projects-issues]

Rossen Stoyanchev commented

After discovering and addressing #21655, I've added only one static factory method method in ExchangeFilterFunctions that delegates to DatabufferUtils#takeUntilByteCount. I think this is the simplest solution and most universally useful part. It does not prevent use of DataBufferUtils#join, or obtaining count+1 to see if more data comes, on top of that.

Thanks for the PR and for the suggestion!