Queue data from bytes_consumer_ when ResponseBodyLoader is suspended

We should queue all the data read from bytes_consumer_ and send them
later when the ResponseBodyLoader resumes. This is needed to support
handling network requests while a page is in the back-forward cache.
Note that other sources of data (that will call DidReceiveData) are out
of scope and will trigger bfcache eviction.

See doc: https://docs.google.com/document/d/1KmiwSOD3Kr3Etjrl8xqMra3QrL7s4Rl7EQJ5s1Wxc0Y/edit#

Bug: 1137682
Change-Id: Iaef4bdea95392bf8ef0a8885bc6774c3e85d4fdf
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2519347
Commit-Queue: Rakina Zata Amni <rakina@chromium.org>
Reviewed-by: Yutaka Hirano <yhirano@chromium.org>
Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Cr-Commit-Position: refs/heads/master@{#825326}

third_party/blink/renderer/platform/loader/fetch/response_body_loader.cc

@@ -433,7 +433,7 @@ void ResponseBodyLoader::OnStateChange() {
 
   size_t num_bytes_consumed = 0;
 
-  while (!aborted_ && !suspended_) {
+  while (!aborted_) {
     if (kMaxNumConsumedBytesInTask == num_bytes_consumed) {
       // We've already consumed many bytes in this task. Defer the remaining
       // to the next task.
@@ -443,6 +443,22 @@ void ResponseBodyLoader::OnStateChange() {
       return;
     }
 
+    if (!suspended_ && bytes_remaining_in_buffer_ > 0) {
+      // We need to empty |buffered_data_| first before reading more from
+      // |bytes_consumer_|.
+      auto* start_position = buffered_data_.end() - bytes_remaining_in_buffer_;
+      size_t size_to_send =
+          std::min(bytes_remaining_in_buffer_,
+                   kMaxNumConsumedBytesInTask - num_bytes_consumed);
+      DidReceiveData(
+          base::make_span(start_position, start_position + size_to_send));
+      bytes_remaining_in_buffer_ -= size_to_send;
+      num_bytes_consumed += size_to_send;
+      if (bytes_remaining_in_buffer_ == 0)
+        buffered_data_.clear();
+      continue;
+    }
+
     const char* buffer = nullptr;
     size_t available = 0;
     auto result = bytes_consumer_->BeginRead(&buffer, &available);
@@ -455,7 +471,13 @@ void ResponseBodyLoader::OnStateChange() {
 
       available =
           std::min(available, kMaxNumConsumedBytesInTask - num_bytes_consumed);
-      DidReceiveData(base::make_span(buffer, available));
+      if (suspended_) {
+        // When suspended, save the read data into |buffered_data_| instead.
+        buffered_data_.insert(buffered_data_.size(), buffer, available);
+        bytes_remaining_in_buffer_ += available;
+      } else {
+        DidReceiveData(base::make_span(buffer, available));
+      }
       result = bytes_consumer_->EndRead(available);
       in_two_phase_read_ = false;
       num_bytes_consumed += available;
@@ -466,6 +488,11 @@ void ResponseBodyLoader::OnStateChange() {
       }
     }
     DCHECK_NE(result, BytesConsumer::Result::kShouldWait);
+    if (suspended_ && result != BytesConsumer::Result::kOk) {
+      // Don't dispatch finish/failure messages when suspended. We'll dispatch
+      // them later when we call OnStateChange again after resuming.
+      return;
+    }
     if (result == BytesConsumer::Result::kDone) {
       DidFinishLoadingBody();
       return;

third_party/blink/renderer/platform/loader/fetch/response_body_loader.h

@@ -14,6 +14,7 @@
 #include "third_party/blink/renderer/platform/loader/fetch/response_body_loader_client.h"
 #include "third_party/blink/renderer/platform/platform_export.h"
 #include "third_party/blink/renderer/platform/wtf/text/wtf_string.h"
+#include "third_party/blink/renderer/platform/wtf/vector.h"
 
 namespace base {
 class SingleThreadTaskRunner;
@@ -124,6 +125,8 @@ class PLATFORM_EXPORT ResponseBodyLoader final
   Member<DelegatingBytesConsumer> delegating_bytes_consumer_;
   const Member<ResponseBodyLoaderClient> client_;
   const scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  Vector<char> buffered_data_;
+  size_t bytes_remaining_in_buffer_ = 0;
   bool started_ = false;
   bool aborted_ = false;
   bool suspended_ = false;

third_party/blink/renderer/platform/loader/fetch/response_body_loader_test.cc

@@ -431,6 +431,70 @@ TEST_F(ResponseBodyLoaderTest, DrainAsDataPipe) {
   EXPECT_EQ("xyzabc", client->GetData());
 }
 
+TEST_F(ResponseBodyLoaderTest, ReadDataFromConsumerWhileSuspended) {
+  auto task_runner = base::MakeRefCounted<scheduler::FakeTaskRunner>();
+  auto* consumer = MakeGarbageCollected<ReplayingBytesConsumer>(task_runner);
+  auto* client = MakeGarbageCollected<TestClient>();
+  auto* body_loader =
+      MakeGarbageCollected<ResponseBodyLoader>(*consumer, *client, task_runner);
+  consumer->Add(Command(Command::kData, "he"));
+  body_loader->Start();
+  task_runner->RunUntilIdle();
+  EXPECT_EQ("he", client->GetData());
+  EXPECT_FALSE(client->LoadingIsFinished());
+  EXPECT_FALSE(client->LoadingIsFailed());
+
+  // Suspend, then add some data to |consumer|.
+  body_loader->Suspend();
+  consumer->Add(Command(Command::kData, "llo"));
+  consumer->Add(Command(Command::kWait));
+  consumer->Add(Command(Command::kData, "wo"));
+  task_runner->RunUntilIdle();
+  EXPECT_EQ("he", client->GetData());
+  EXPECT_FALSE(client->LoadingIsFinished());
+  EXPECT_FALSE(client->LoadingIsFailed());
+
+  // The data received while suspended will be processed after resuming, before
+  // processing newer data.
+  body_loader->Resume();
+  consumer->Add(Command(Command::kData, "rld"));
+  consumer->Add(Command(Command::kDone));
+
+  task_runner->RunUntilIdle();
+  EXPECT_EQ("helloworld", client->GetData());
+  EXPECT_TRUE(client->LoadingIsFinished());
+  EXPECT_FALSE(client->LoadingIsFailed());
+}
+
+TEST_F(ResponseBodyLoaderTest, ReadDataFromConsumerWhileSuspendedLong) {
+  auto task_runner = base::MakeRefCounted<scheduler::FakeTaskRunner>();
+  auto* consumer = MakeGarbageCollected<ReplayingBytesConsumer>(task_runner);
+  auto* client = MakeGarbageCollected<TestClient>();
+  auto* body_loader =
+      MakeGarbageCollected<ResponseBodyLoader>(*consumer, *client, task_runner);
+  body_loader->Start();
+  task_runner->RunUntilIdle();
+  EXPECT_EQ("", client->GetData());
+  EXPECT_FALSE(client->LoadingIsFinished());
+  EXPECT_FALSE(client->LoadingIsFailed());
+
+  // Suspend, then add a long response body to |consumer|.
+  body_loader->Suspend();
+  std::string body(70000, '*');
+  consumer->Add(Command(Command::kDataAndDone, body.c_str()));
+  task_runner->RunUntilIdle();
+  EXPECT_EQ("", client->GetData());
+  EXPECT_FALSE(client->LoadingIsFinished());
+  EXPECT_FALSE(client->LoadingIsFailed());
+
+  // The data received while suspended will be processed after resuming.
+  body_loader->Resume();
+  task_runner->RunUntilIdle();
+  EXPECT_EQ(AtomicString(body.c_str()), client->GetData());
+  EXPECT_TRUE(client->LoadingIsFinished());
+  EXPECT_FALSE(client->LoadingIsFailed());
+}
+
 TEST_F(ResponseBodyLoaderTest, DrainAsDataPipeAndReportError) {
   mojo::ScopedDataPipeConsumerHandle consumer_end;
   mojo::ScopedDataPipeProducerHandle producer_end;