[http-cache-filter] limit cache read chunk size to buffer size (envoy…

…proxy#32307)

* [cache-filter] limit cache read chunk size

---------

Signed-off-by: Raven Black <ravenblack@dropbox.com>

source/extensions/filters/http/cache/cache_filter.cc

@@ -23,6 +23,16 @@ namespace {
 inline bool isResponseNotModified(const Http::ResponseHeaderMap& response_headers) {
   return Http::Utility::getResponseStatus(response_headers) == enumToInt(Http::Code::NotModified);
 }
+
+// This value is only used if there is no encoderBufferLimit on the stream;
+// without *some* constraint here, a very large chunk can be requested and
+// attempt to load into a memory buffer.
+//
+// This default is quite large to minimize the chance of being a surprise
+// behavioral change when a constraint is added.
+//
+// And everyone knows 64MB should be enough for anyone.
+static const size_t MAX_BYTES_TO_FETCH_FROM_CACHE_PER_REQUEST = 64 * 1024 * 1024;
 } // namespace
 
 struct CacheResponseCodeDetailValues {
@@ -326,10 +336,21 @@ void CacheFilter::getBody() {
   // posted callback.
   CacheFilterWeakPtr self = weak_from_this();
 
+  // We don't want to request more than a buffer-size at a time from the cache.
+  uint64_t fetch_size_limit = encoder_callbacks_->encoderBufferLimit();
+  // If there is no buffer size limit, we still want *some* constraint.
+  if (fetch_size_limit == 0) {
+    fetch_size_limit = MAX_BYTES_TO_FETCH_FROM_CACHE_PER_REQUEST;
+  }
+  AdjustedByteRange fetch_range = {remaining_ranges_[0].begin(),
+                                   (remaining_ranges_[0].length() > fetch_size_limit)
+                                       ? (remaining_ranges_[0].begin() + fetch_size_limit)
+                                       : remaining_ranges_[0].end()};
+
   // The dispatcher needs to be captured because there's no guarantee that
   // decoder_callbacks_->dispatcher() is thread-safe.
-  lookup_->getBody(remaining_ranges_[0], [self, &dispatcher = decoder_callbacks_->dispatcher()](
-                                             Buffer::InstancePtr&& body) {
+  lookup_->getBody(fetch_range, [self, &dispatcher = decoder_callbacks_->dispatcher()](
+                                    Buffer::InstancePtr&& body) {
     // The callback is posted to the dispatcher to make sure it is called on the worker thread.
     dispatcher.post([self, body = std::move(body)]() mutable {
       if (CacheFilterSharedPtr cache_filter = self.lock()) {

test/extensions/filters/http/cache/cache_filter_test.cc

@@ -490,6 +490,69 @@ TEST_F(CacheFilterTest, FilterDestroyedWhileWatermarkedSendsLowWatermarkEvent) {
   }
 }
 
+MATCHER_P2(RangeMatcher, begin, end, "") {
+  return testing::ExplainMatchResult(begin, arg.begin(), result_listener) &&
+         testing::ExplainMatchResult(end, arg.end(), result_listener);
+}
+
+TEST_F(CacheFilterTest, BodyReadFromCacheLimitedToBufferSizeChunks) {
+  request_headers_.setHost("CacheHitWithBody");
+  // Set the buffer limit to 5 bytes, and we will have the file be of size
+  // 8 bytes.
+  EXPECT_CALL(encoder_callbacks_, encoderBufferLimit()).WillRepeatedly(::testing::Return(5));
+  auto mock_http_cache = std::make_shared<MockHttpCache>();
+  auto mock_lookup_context = std::make_unique<MockLookupContext>();
+  EXPECT_CALL(*mock_http_cache, makeLookupContext(_, _))
+      .WillOnce([&](LookupRequest&&,
+                    Http::StreamDecoderFilterCallbacks&) -> std::unique_ptr<LookupContext> {
+        return std::move(mock_lookup_context);
+      });
+  EXPECT_CALL(*mock_lookup_context, getHeaders(_)).WillOnce([&](LookupHeadersCallback&& cb) {
+    std::unique_ptr<Http::ResponseHeaderMap> response_headers =
+        std::make_unique<Http::TestResponseHeaderMapImpl>(response_headers_);
+    cb(LookupResult{CacheEntryStatus::Ok, std::move(response_headers), 8, absl::nullopt});
+  });
+  EXPECT_CALL(*mock_lookup_context, getBody(RangeMatcher(0, 5), _))
+      .WillOnce([&](const AdjustedByteRange&, LookupBodyCallback&& cb) {
+        cb(std::make_unique<Buffer::OwnedImpl>("abcde"));
+      });
+  EXPECT_CALL(*mock_lookup_context, getBody(RangeMatcher(5, 8), _))
+      .WillOnce([&](const AdjustedByteRange&, LookupBodyCallback&& cb) {
+        cb(std::make_unique<Buffer::OwnedImpl>("fgh"));
+      });
+  EXPECT_CALL(*mock_lookup_context, onDestroy());
+
+  CacheFilterSharedPtr filter = makeFilter(mock_http_cache, false);
+
+  // The filter should encode cached headers.
+  EXPECT_CALL(decoder_callbacks_, encodeHeaders_(IsSupersetOfHeaders(response_headers_), false));
+
+  // The filter should encode cached data in two pieces.
+  EXPECT_CALL(
+      decoder_callbacks_,
+      encodeData(testing::Property(&Buffer::Instance::toString, testing::Eq("abcde")), false));
+  EXPECT_CALL(decoder_callbacks_,
+              encodeData(testing::Property(&Buffer::Instance::toString, testing::Eq("fgh")), true));
+
+  // The filter should stop decoding iteration when decodeHeaders is called as a cache lookup is
+  // in progress.
+  EXPECT_EQ(filter->decodeHeaders(request_headers_, true),
+            Http::FilterHeadersStatus::StopAllIterationAndWatermark);
+
+  // The filter should not continue decoding when the cache lookup result is ready, as the
+  // expected result is a hit.
+  EXPECT_CALL(decoder_callbacks_, continueDecoding).Times(0);
+
+  // The cache lookup callback should be posted to the dispatcher.
+  // Run events on the dispatcher so that the callback is invoked.
+  // The posted lookup callback will cause another callback to be posted (when getBody() is
+  // called) which should also be invoked.
+  dispatcher_->run(Event::Dispatcher::RunType::Block);
+
+  filter->onDestroy();
+  filter.reset();
+}
+
 TEST_F(CacheFilterTest, CacheInsertAbortedByCache) {
   request_headers_.setHost("CacheHitWithBody");
   const std::string body = "abc";