ServiceWorkerCacheWriter should handle remote disconnection

It owns remotes to ServiceWorkerResource{Reader,Writer} which can be
disconnected at any time. Handle these disconnections to provide
graceful error propagations.

This CL doesn't add tests for some comparison cases (copying and
resuming). Follow-up CLs will add these tests.

Bug: 1133143
Change-Id: Ie8d76b93ff0ffbf3bfde4ddfe2dfe98b2ab920d7
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2525702Reviewed-by: Makoto Shimazu <shimazu@chromium.org>
Commit-Queue: Kenichi Ishibashi <bashi@chromium.org>
Cr-Commit-Position: refs/heads/master@{#825704}

content/browser/service_worker/service_worker_cache_writer.cc

@@ -174,7 +174,22 @@ ServiceWorkerCacheWriter::ServiceWorkerCacheWriter(
       compare_reader_(std::move(compare_reader)),
       copy_reader_(std::move(copy_reader)),
       writer_(std::move(writer)),
-      writer_resource_id_(writer_resource_id) {}
+      writer_resource_id_(writer_resource_id) {
+  if (compare_reader_) {
+    compare_reader_.set_disconnect_handler(
+        base::BindOnce(&ServiceWorkerCacheWriter::OnRemoteDisconnected,
+                       weak_factory_.GetWeakPtr()));
+  }
+  if (copy_reader_) {
+    copy_reader_.set_disconnect_handler(
+        base::BindOnce(&ServiceWorkerCacheWriter::OnRemoteDisconnected,
+                       weak_factory_.GetWeakPtr()));
+  }
+  DCHECK(writer_);
+  writer_.set_disconnect_handler(
+      base::BindOnce(&ServiceWorkerCacheWriter::OnRemoteDisconnected,
+                     weak_factory_.GetWeakPtr()));
+}
 
 ServiceWorkerCacheWriter::~ServiceWorkerCacheWriter() {}
 
@@ -184,7 +199,9 @@ ServiceWorkerCacheWriter::CreateForCopy(
     mojo::Remote<storage::mojom::ServiceWorkerResourceWriter> writer,
     int64_t writer_resource_id) {
   DCHECK(copy_reader);
+  DCHECK(copy_reader.is_connected());
   DCHECK(writer);
+  DCHECK(writer.is_connected());
   mojo::Remote<storage::mojom::ServiceWorkerResourceReader> null_remote;
   return base::WrapUnique(new ServiceWorkerCacheWriter(
       std::move(null_remote) /* compare_reader */, std::move(copy_reader),
@@ -197,6 +214,7 @@ ServiceWorkerCacheWriter::CreateForWriteBack(
     mojo::Remote<storage::mojom::ServiceWorkerResourceWriter> writer,
     int64_t writer_resource_id) {
   DCHECK(writer);
+  DCHECK(writer.is_connected());
   return base::WrapUnique(new ServiceWorkerCacheWriter(
       /*compare_reader=*/{}, /*copy_reader=*/{}, std::move(writer),
       writer_resource_id,
@@ -213,8 +231,11 @@ ServiceWorkerCacheWriter::CreateForComparison(
   // |compare_reader| reads data for the comparison. |copy_reader| reads
   // data for copy.
   DCHECK(compare_reader);
+  DCHECK(compare_reader.is_connected());
   DCHECK(copy_reader);
+  DCHECK(copy_reader.is_connected());
   DCHECK(writer);
+  DCHECK(writer.is_connected());
   return base::WrapUnique(new ServiceWorkerCacheWriter(
       std::move(compare_reader), std::move(copy_reader), std::move(writer),
       writer_resource_id, pause_when_not_identical));
@@ -226,6 +247,9 @@ net::Error ServiceWorkerCacheWriter::MaybeWriteHeaders(
   DCHECK(!io_pending_);
   DCHECK(!IsCopying());
 
+  if (!writer_.is_connected())
+    return net::ERR_FAILED;
+
   response_head_to_write_ = std::move(response_head);
   pending_callback_ = std::move(callback);
   DCHECK_EQ(STATE_START, state_);
@@ -255,6 +279,9 @@ net::Error ServiceWorkerCacheWriter::MaybeWriteData(
   DCHECK(!io_pending_);
   DCHECK(!IsCopying());
 
+  if (!writer_.is_connected())
+    return net::ERR_FAILED;
+
   data_to_write_ = buf;
   len_to_write_ = buf_size;
   pending_callback_ = std::move(callback);
@@ -295,6 +322,9 @@ net::Error ServiceWorkerCacheWriter::Resume(OnWriteCompleteCallback callback) {
   DCHECK(io_pending_);
   DCHECK(!IsCopying());
 
+  if (!copy_reader_.is_connected())
+    return net::ERR_FAILED;
+
   io_pending_ = false;
   pending_callback_ = std::move(callback);
   state_ = STATE_READ_HEADERS_FOR_COPY;
@@ -326,8 +356,11 @@ net::Error ServiceWorkerCacheWriter::Resume(OnWriteCompleteCallback callback) {
 net::Error ServiceWorkerCacheWriter::StartCopy(
     OnWriteCompleteCallback callback) {
   DCHECK(IsCopying());
-  pending_callback_ = std::move(callback);
 
+  if (!copy_reader_.is_connected())
+    return net::ERR_FAILED;
+
+  pending_callback_ = std::move(callback);
   int result = DoLoop(net::OK);
 
   // Synchronous completions are always STATE_DONE.
@@ -351,6 +384,15 @@ bool ServiceWorkerCacheWriter::IsCopying() const {
   return !compare_reader_ && copy_reader_;
 }
 
+void ServiceWorkerCacheWriter::FlushRemotesForTesting() {
+  if (copy_reader_)
+    copy_reader_.FlushForTesting();  // IN-TEST
+  if (compare_reader_)
+    compare_reader_.FlushForTesting();  // IN-TEST
+  DCHECK(writer_);
+  writer_.FlushForTesting();  // IN-TEST
+}
+
 int64_t ServiceWorkerCacheWriter::writer_resource_id() const {
   DCHECK_NE(writer_resource_id_, blink::mojom::kInvalidServiceWorkerResourceId);
   return writer_resource_id_;
@@ -377,6 +419,11 @@ int ServiceWorkerCacheWriter::DoReadHeadersForCompare(int result) {
   DCHECK(response_head_to_write_);
   DCHECK(compare_reader_);
 
+  if (!compare_reader_.is_connected()) {
+    state_ = STATE_DONE;
+    return net::ERR_FAILED;
+  }
+
   state_ = STATE_READ_HEADERS_FOR_COMPARE_DONE;
   return ReadResponseHead(compare_reader_.get());
 }
@@ -404,6 +451,10 @@ int ServiceWorkerCacheWriter::DoReadDataForCompare(int result) {
   // If this was an EOF, don't issue a read.
   if (len_to_write_ > 0) {
     DCHECK(compare_reader_);
+    if (!compare_reader_.is_connected()) {
+      state_ = STATE_DONE;
+      return net::ERR_FAILED;
+    }
     result = ReadDataHelper(compare_reader_.get(), compare_data_pipe_reader_,
                             data_to_read_.get(), len_to_read_);
   }
@@ -456,6 +507,10 @@ int ServiceWorkerCacheWriter::DoReadDataForCompareDone(int result) {
   // that this reuses the same IOBuffer.
   if (compare_offset_ < static_cast<size_t>(len_to_read_)) {
     DCHECK(compare_reader_);
+    if (!compare_reader_.is_connected()) {
+      state_ = STATE_DONE;
+      return net::ERR_FAILED;
+    }
     state_ = STATE_READ_DATA_FOR_COMPARE_DONE;
     return ReadDataHelper(compare_reader_.get(), compare_data_pipe_reader_,
                           data_to_read_.get(), len_to_read_ - compare_offset_);
@@ -478,6 +533,12 @@ int ServiceWorkerCacheWriter::DoReadDataForCompareDone(int result) {
 int ServiceWorkerCacheWriter::DoReadHeadersForCopy(int result) {
   DCHECK_GE(result, 0);
   DCHECK(copy_reader_);
+
+  if (!copy_reader_.is_connected()) {
+    state_ = STATE_DONE;
+    return net::ERR_FAILED;
+  }
+
   bytes_copied_ = 0;
   data_to_copy_ = base::MakeRefCounted<net::IOBuffer>(kCopyBufferSize);
   state_ = STATE_READ_HEADERS_FOR_COPY_DONE;
@@ -522,6 +583,12 @@ int ServiceWorkerCacheWriter::DoWriteHeadersForCopyDone(int result) {
 
 int ServiceWorkerCacheWriter::DoReadDataForCopy(int result) {
   DCHECK_GE(result, 0);
+  DCHECK(copy_reader_);
+
+  if (!copy_reader_.is_connected()) {
+    state_ = STATE_DONE;
+    return net::ERR_FAILED;
+  }
 
   // If the cache writer is only for copy, get the total size to read from
   // header data instead of |bytes_compared_| as no comparison is done.
@@ -731,6 +798,11 @@ int ServiceWorkerCacheWriter::ReadDataHelper(
 
 int ServiceWorkerCacheWriter::WriteResponseHeadToResponseWriter(
     network::mojom::URLResponseHeadPtr response_head) {
+  if (!writer_.is_connected()) {
+    state_ = STATE_DONE;
+    return net::ERR_FAILED;
+  }
+
   DCHECK(response_head);
   did_replace_ = true;
   net::CompletionOnceCallback run_callback = base::BindOnce(
@@ -762,6 +834,11 @@ int ServiceWorkerCacheWriter::WriteResponseHead(
 int ServiceWorkerCacheWriter::WriteDataToResponseWriter(
     scoped_refptr<net::IOBuffer> data,
     int length) {
+  if (!writer_.is_connected()) {
+    state_ = STATE_DONE;
+    return net::ERR_FAILED;
+  }
+
   net::CompletionOnceCallback run_callback = base::BindOnce(
       &ServiceWorkerCacheWriter::AsyncDoLoop, weak_factory_.GetWeakPtr());
   scoped_refptr<AsyncOnlyCompletionCallbackAdaptor> adaptor(
@@ -815,15 +892,28 @@ void ServiceWorkerCacheWriter::OnWillWriteDataCompleted(
     AsyncDoLoop(result);
 }
 
+void ServiceWorkerCacheWriter::OnRemoteDisconnected() {
+  if (state_ == STATE_DONE)
+    return;
+
+  state_ = STATE_DONE;
+  AsyncDoLoop(net::ERR_FAILED);
+}
+
 void ServiceWorkerCacheWriter::AsyncDoLoop(int result) {
   result = DoLoop(result);
   // If the result is ERR_IO_PENDING, the pending callback will be run by a
   // later invocation of AsyncDoLoop.
   if (result != net::ERR_IO_PENDING) {
-    OnWriteCompleteCallback callback = std::move(pending_callback_);
-    net::Error error = result >= 0 ? net::OK : static_cast<net::Error>(result);
     io_pending_ = false;
-    std::move(callback).Run(error);
+    // `pending_callback_` might be already consumed when mojo remotes are
+    // disconnected.
+    if (pending_callback_) {
+      OnWriteCompleteCallback callback = std::move(pending_callback_);
+      net::Error error =
+          result >= 0 ? net::OK : static_cast<net::Error>(result);
+      std::move(callback).Run(error);
+    }
     return;
   }
   if (state_ == STATE_PAUSING) {

content/browser/service_worker/service_worker_cache_writer.h

@@ -140,6 +140,8 @@ class CONTENT_EXPORT ServiceWorkerCacheWriter {
     write_observer_ = write_observer;
   }
 
+  void FlushRemotesForTesting();
+
  private:
   class ReadResponseHeadCallbackAdapter;
   class DataPipeReader;
@@ -270,6 +272,8 @@ class CONTENT_EXPORT ServiceWorkerCacheWriter {
                                 int length,
                                 net::Error error);
 
+  void OnRemoteDisconnected();
+
   // Callback used by the above helpers for their IO operations. This is only
   // run when those IO operations complete asynchronously, in which case it
   // invokes the synchronous DoLoop function and runs the client callback (the

content/browser/service_worker/service_worker_cache_writer_unittest.cc

@@ -117,7 +117,7 @@ class ServiceWorkerCacheWriterTest : public ::testing::Test {
   std::list<std::unique_ptr<MockServiceWorkerResourceWriter>> writers_;
   std::unique_ptr<ServiceWorkerCacheWriter> cache_writer_;
   bool write_complete_ = false;
-  net::Error last_error_;
+  net::Error last_error_ = net::OK;
 
   mojo::Remote<storage::mojom::ServiceWorkerResourceReader> CreateReader() {
     mojo::Remote<storage::mojom::ServiceWorkerResourceReader> remote;
@@ -1002,5 +1002,288 @@ TEST_F(ServiceWorkerCacheWriterTest, ObserverAsyncFail) {
   cache_writer_->set_write_observer(nullptr);
 }
 
+class ServiceWorkerCacheWriterDisconnectionTest
+    : public ServiceWorkerCacheWriterTest {
+ public:
+  ServiceWorkerCacheWriterDisconnectionTest() = default;
+  ~ServiceWorkerCacheWriterDisconnectionTest() override = default;
+
+  void InitializeForWriteBack() {
+    writer_ = std::make_unique<MockServiceWorkerResourceWriter>();
+    mojo::Remote<storage::mojom::ServiceWorkerResourceWriter> remote_writer;
+    remote_writer.Bind(writer_->BindNewPipeAndPassRemote(base::DoNothing()));
+
+    cache_writer_ = ServiceWorkerCacheWriter::CreateForWriteBack(
+        std::move(remote_writer), /*writer_resource_id=*/0);
+  }
+
+  void InitializeForCopy() {
+    writer_ = std::make_unique<MockServiceWorkerResourceWriter>();
+    mojo::Remote<storage::mojom::ServiceWorkerResourceWriter> remote_writer;
+    remote_writer.Bind(writer_->BindNewPipeAndPassRemote(base::DoNothing()));
+
+    copy_reader_ = std::make_unique<MockServiceWorkerResourceReader>();
+    mojo::Remote<storage::mojom::ServiceWorkerResourceReader>
+        remote_copy_reader;
+    remote_copy_reader.Bind(
+        copy_reader_->BindNewPipeAndPassRemote(base::DoNothing()));
+
+    cache_writer_ = ServiceWorkerCacheWriter::CreateForCopy(
+        std::move(remote_copy_reader), std::move(remote_writer),
+        /*writer_resource_id=*/0);
+  }
+
+  void InitializeForComparison() {
+    writer_ = std::make_unique<MockServiceWorkerResourceWriter>();
+    mojo::Remote<storage::mojom::ServiceWorkerResourceWriter> remote_writer;
+    remote_writer.Bind(writer_->BindNewPipeAndPassRemote(base::DoNothing()));
+
+    copy_reader_ = std::make_unique<MockServiceWorkerResourceReader>();
+    mojo::Remote<storage::mojom::ServiceWorkerResourceReader>
+        remote_copy_reader;
+    remote_copy_reader.Bind(
+        copy_reader_->BindNewPipeAndPassRemote(base::DoNothing()));
+
+    compare_reader_ = std::make_unique<MockServiceWorkerResourceReader>();
+    mojo::Remote<storage::mojom::ServiceWorkerResourceReader>
+        remote_compare_reader;
+    remote_compare_reader.Bind(
+        compare_reader_->BindNewPipeAndPassRemote(base::DoNothing()));
+
+    cache_writer_ = ServiceWorkerCacheWriter::CreateForComparison(
+        std::move(remote_compare_reader), std::move(remote_copy_reader),
+        std::move(remote_writer),
+        /*writer_resource_id=*/0, /*pause_when_not_identical=*/true);
+  }
+
+  void SimulateDisconnection() {
+    // Destroy readers and the writer to disconnect remotes in `cache_writer_`.
+    writer_.reset();
+    copy_reader_.reset();
+    compare_reader_.reset();
+    cache_writer_->FlushRemotesForTesting();
+  }
+
+ protected:
+  std::unique_ptr<MockServiceWorkerResourceWriter> writer_;
+  std::unique_ptr<MockServiceWorkerResourceReader> copy_reader_;
+  std::unique_ptr<MockServiceWorkerResourceReader> compare_reader_;
+};
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, WriteBackBeforeHeader) {
+  size_t kHeaderSize = 16;
+
+  InitializeForWriteBack();
+  writer_->ExpectWriteResponseHeadOk(kHeaderSize);
+
+  SimulateDisconnection();
+
+  net::Error error = WriteHeaders(kHeaderSize);
+  EXPECT_EQ(error, net::ERR_FAILED);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, WriteBackBeforeData) {
+  const std::string data1 = "abcdef";
+  const size_t response_size = data1.size();
+
+  InitializeForWriteBack();
+  writer_->ExpectWriteResponseHeadOk(response_size);
+  writer_->ExpectWriteDataOk(data1.size());
+
+  net::Error error = WriteHeaders(response_size);
+  EXPECT_EQ(error, net::ERR_IO_PENDING);
+  writer_->CompletePendingWrite();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_TRUE(write_complete_);
+  write_complete_ = false;
+
+  SimulateDisconnection();
+
+  error = WriteData(data1);
+  EXPECT_EQ(error, net::ERR_FAILED);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, WriteBackDuringData) {
+  const std::string data1 = "abcdef";
+  const std::string data2 = "ghijklmno";
+  const size_t response_size = data1.size() + data2.size();
+
+  InitializeForWriteBack();
+  writer_->ExpectWriteResponseHeadOk(response_size);
+  writer_->ExpectWriteDataOk(data1.size());
+  writer_->ExpectWriteDataOk(data2.size());
+
+  net::Error error = WriteHeaders(response_size);
+  EXPECT_EQ(net::ERR_IO_PENDING, error);
+  writer_->CompletePendingWrite();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_TRUE(write_complete_);
+  write_complete_ = false;
+
+  error = WriteData(data1);
+  EXPECT_EQ(error, net::ERR_IO_PENDING);
+  writer_->CompletePendingWrite();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_TRUE(write_complete_);
+  write_complete_ = false;
+
+  error = WriteData(data2);
+  EXPECT_EQ(error, net::ERR_IO_PENDING);
+
+  SimulateDisconnection();
+
+  EXPECT_TRUE(write_complete_);
+  EXPECT_EQ(last_error_, net::ERR_FAILED);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, CopyBeforeStart) {
+  const std::string data1 = "abcd";
+  const size_t response_size = data1.size();
+
+  InitializeForCopy();
+  writer_->ExpectWriteResponseHeadOk(response_size);
+  writer_->ExpectWriteDataOk(data1.size());
+
+  SimulateDisconnection();
+
+  net::Error error = cache_writer_->StartCopy(CreateWriteCallback());
+  EXPECT_EQ(error, net::ERR_FAILED);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, CopyBeforeHeaderRead) {
+  const std::string data1 = "abcd";
+  const size_t response_size = data1.size();
+
+  InitializeForCopy();
+  copy_reader_->ExpectReadResponseHeadOk(response_size);
+  copy_reader_->ExpectReadDataOk(data1);
+  writer_->ExpectWriteResponseHeadOk(response_size);
+  writer_->ExpectWriteDataOk(data1.size());
+
+  net::Error error = cache_writer_->StartCopy(CreateWriteCallback());
+  EXPECT_EQ(error, net::ERR_IO_PENDING);
+  EXPECT_FALSE(write_complete_);
+
+  SimulateDisconnection();
+
+  EXPECT_EQ(last_error_, net::ERR_FAILED);
+  EXPECT_TRUE(write_complete_);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, CopyBeforeDataRead) {
+  const std::string data1 = "abcd";
+  const size_t response_size = data1.size();
+
+  InitializeForCopy();
+  copy_reader_->ExpectReadResponseHeadOk(response_size);
+  copy_reader_->ExpectReadDataOk(data1);
+  writer_->ExpectWriteResponseHeadOk(response_size);
+  writer_->ExpectWriteDataOk(data1.size());
+
+  net::Error error = cache_writer_->StartCopy(CreateWriteCallback());
+  EXPECT_EQ(error, net::ERR_IO_PENDING);
+  EXPECT_FALSE(write_complete_);
+
+  // Completes the header read.
+  copy_reader_->CompletePendingRead();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_FALSE(write_complete_);
+
+  // Completes the header write.
+  writer_->CompletePendingWrite();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_FALSE(write_complete_);
+
+  SimulateDisconnection();
+
+  EXPECT_EQ(last_error_, net::ERR_FAILED);
+  EXPECT_TRUE(write_complete_);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, CopyDuringDataRead) {
+  const std::string data1 = "abcd";
+  const std::string data2 = "efgh";
+  const size_t response_size = data1.size() + data2.size();
+
+  InitializeForCopy();
+  copy_reader_->ExpectReadResponseHeadOk(response_size);
+  copy_reader_->ExpectReadDataOk(data1);
+  copy_reader_->ExpectReadDataOk(data2);
+  writer_->ExpectWriteResponseHeadOk(response_size);
+  writer_->ExpectWriteDataOk(data1.size());
+  writer_->ExpectWriteDataOk(data2.size());
+
+  net::Error error = cache_writer_->StartCopy(CreateWriteCallback());
+  EXPECT_EQ(error, net::ERR_IO_PENDING);
+  EXPECT_FALSE(write_complete_);
+
+  // Completes the header read.
+  copy_reader_->CompletePendingRead();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_FALSE(write_complete_);
+
+  // Completes the header write.
+  writer_->CompletePendingWrite();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_FALSE(write_complete_);
+
+  // Completes the read of the first data chunk.
+  copy_reader_->CompletePendingRead();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_FALSE(write_complete_);
+
+  // Completes the write of the first data chunk.
+  writer_->CompletePendingWrite();
+  EXPECT_EQ(last_error_, net::OK);
+  EXPECT_FALSE(write_complete_);
+
+  SimulateDisconnection();
+
+  EXPECT_EQ(last_error_, net::ERR_FAILED);
+  EXPECT_TRUE(write_complete_);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, ComparisonBeforeHeaderRead) {
+  const std::string data1 = "abcd";
+  const size_t response_size = data1.size();
+
+  InitializeForComparison();
+  compare_reader_->ExpectReadResponseHeadOk(response_size);
+  compare_reader_->ExpectReadDataOk(data1);
+
+  net::Error error = WriteHeaders(response_size);
+  EXPECT_EQ(net::ERR_IO_PENDING, error);
+  EXPECT_FALSE(write_complete_);
+
+  SimulateDisconnection();
+
+  EXPECT_EQ(last_error_, net::ERR_FAILED);
+  EXPECT_TRUE(write_complete_);
+}
+
+TEST_F(ServiceWorkerCacheWriterDisconnectionTest, ComparisonBeforeDataWrite) {
+  const std::string data1 = "abcd";
+  const size_t response_size = data1.size();
+
+  InitializeForComparison();
+  compare_reader_->ExpectReadResponseHeadOk(response_size);
+  compare_reader_->ExpectReadDataOk(data1);
+
+  // Completes the header read.
+  net::Error error = WriteHeaders(response_size);
+  EXPECT_EQ(net::ERR_IO_PENDING, error);
+  EXPECT_FALSE(write_complete_);
+  compare_reader_->CompletePendingRead();
+  EXPECT_TRUE(write_complete_);
+  write_complete_ = false;
+
+  SimulateDisconnection();
+
+  error = WriteData(data1);
+  EXPECT_EQ(error, net::ERR_FAILED);
+}
+
+// TODO(crbug.com/1133143): Add test for copying and resuming for comparison.
+
 }  // namespace
 }  // namespace content