[sync] Introduce HttpPostProvider::SetAllowBatching()

The method allows the embedder's network stack to batch a POST request
so that the network stack can schedule network requests in efficient
ways (e.g. batch requests to save power consumption when the device's
radio state is inactive).

Batching is only enabled for GET_UPDATES messages.

Bug: 1293657
Change-Id: I92481aad8d0ab2b3f72555598d0a39251555bd2a
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/3599405
Commit-Queue: Kenichi Ishibashi <bashi@chromium.org>
Reviewed-by: Marc Treib <treib@chromium.org>
Cr-Commit-Position: refs/heads/main@{#997004}

components/sync/engine/loopback_server/loopback_connection_manager.cc

@@ -18,6 +18,7 @@ LoopbackConnectionManager::~LoopbackConnectionManager() = default;
 HttpResponse LoopbackConnectionManager::PostBuffer(
     const std::string& buffer_in,
     const std::string& access_token,
+    bool allow_batching,
     std::string* buffer_out) {
   buffer_out->clear();
 

components/sync/engine/loopback_server/loopback_connection_manager.h

@@ -30,6 +30,7 @@ class LoopbackConnectionManager : public ServerConnectionManager {
   // Overridden ServerConnectionManager functions.
   HttpResponse PostBuffer(const std::string& buffer_in,
                           const std::string& access_token,
+                          bool allow_batching,
                           std::string* buffer_out) override;
 
   // The loopback server that will handle the requests locally.

components/sync/engine/net/http_bridge.cc

@@ -100,6 +100,11 @@ void HttpBridge::SetExtraRequestHeaders(const char* headers) {
   extra_headers_.assign(headers);
 }
 
+void HttpBridge::SetAllowBatching(bool allow_batching) {
+  DCHECK(!fetch_state_.url_loader);
+  allow_batching_ = allow_batching;
+}
+
 void HttpBridge::SetURL(const GURL& url) {
 #if DCHECK_IS_ON()
   DCHECK(thread_checker_.CalledOnValidThread());
@@ -245,8 +250,10 @@ void HttpBridge::MakeAsynchronousPost() {
       std::move(resource_request), traffic_annotation);
   network::SimpleURLLoader* url_loader = fetch_state_.url_loader.get();
 
-  if (network::SimpleURLLoaderThrottle::IsBatchingEnabled(traffic_annotation))
+  if (allow_batching_ &&
+      network::SimpleURLLoaderThrottle::IsBatchingEnabled(traffic_annotation)) {
     url_loader->SetAllowBatching();
+  }
 
   std::string request_to_send;
   compression::GzipCompress(request_content_, &request_to_send);

components/sync/engine/net/http_bridge.h

@@ -54,6 +54,7 @@ class HttpBridge : public HttpPostProvider {
   void SetPostPayload(const char* content_type,
                       int content_length,
                       const char* content) override;
+  void SetAllowBatching(bool allow_batching) override;
   bool MakeSynchronousPost(int* net_error_code, int* http_status_code) override;
   void Abort() override;
 
@@ -132,6 +133,11 @@ class HttpBridge : public HttpPostProvider {
   std::string request_content_;
   std::string extra_headers_;
 
+  // When true `fetch_state_.url_loader` is configured so that it can be
+  // batched in the network layer. See the comment in
+  // network::SimpleURLLoader::SetAllowBatching().
+  bool allow_batching_ = false;
+
   // A waitable event we use to provide blocking semantics to
   // MakeSynchronousPost. We block created_on_loop_ while the IO loop fetches
   // network request.

components/sync/engine/net/http_post_provider.h

@@ -36,6 +36,10 @@ class HttpPostProvider : public base::RefCountedThreadSafe<HttpPostProvider> {
                               int content_length,
                               const char* content) = 0;
 
+  // Set whether the POST message can be batched in the network stack of the
+  // embedding application.
+  virtual void SetAllowBatching(bool allow_batching) = 0;
+
   // Returns true if the URL request succeeded. If the request failed,
   // error() may be non-zero and hence contain more information.
   virtual bool MakeSynchronousPost(int* net_error_code,

components/sync/engine/net/server_connection_manager.cc

@@ -151,9 +151,11 @@ void ServerConnectionManager::NotifyStatusChanged() {
 
 HttpResponse ServerConnectionManager::PostBufferWithCachedAuth(
     const std::string& buffer_in,
+    bool allow_batching,
     std::string* buffer_out) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-  HttpResponse http_response = PostBuffer(buffer_in, access_token_, buffer_out);
+  HttpResponse http_response =
+      PostBuffer(buffer_in, access_token_, allow_batching, buffer_out);
   SetServerResponse(http_response);
   return http_response;
 }

components/sync/engine/net/server_connection_manager.h

@@ -98,8 +98,13 @@ class ServerConnectionManager {
   virtual ~ServerConnectionManager();
 
   // POSTs |buffer_in| and reads the body of the response into |buffer_out|.
-  // Uses the currently set access token in the headers.
+  // Uses the currently set access token in the headers. When |allow_batching|
+  // is true, the embedder's network stack may batch the post request depending
+  // on the network quality to streamline network access.
+  // TODO(https://crbug.com/1293657): Consider integrating batching logic into
+  // the sync code rather than relying on the embedder's network stack.
   HttpResponse PostBufferWithCachedAuth(const std::string& buffer_in,
+                                        bool allow_batching,
                                         std::string* buffer_out);
 
   void AddListener(ServerConnectionEventListener* listener);
@@ -136,6 +141,7 @@ class ServerConnectionManager {
   // implement.
   virtual HttpResponse PostBuffer(const std::string& buffer_in,
                                   const std::string& access_token,
+                                  bool allow_batching,
                                   std::string* buffer_out) = 0;
 
   void ClearAccessToken();

components/sync/engine/net/sync_server_connection_manager.cc

@@ -35,7 +35,8 @@ class Connection : public CancelationSignal::Observer {
 
   HttpResponse Init(const GURL& connection_url,
                     const std::string& access_token,
-                    const std::string& payload);
+                    const std::string& payload,
+                    bool allow_batching);
   bool ReadBufferResponse(std::string* buffer_out, HttpResponse* response);
 
   // CancelationSignal::Observer overrides.
@@ -69,7 +70,8 @@ Connection::~Connection() = default;
 
 HttpResponse Connection::Init(const GURL& sync_request_url,
                               const std::string& access_token,
-                              const std::string& payload) {
+                              const std::string& payload,
+                              bool allow_batching) {
   post_provider_->SetURL(sync_request_url);
 
   if (!access_token.empty()) {
@@ -82,6 +84,8 @@ HttpResponse Connection::Init(const GURL& sync_request_url,
   post_provider_->SetPostPayload("application/octet-stream", payload.length(),
                                  payload.data());
 
+  post_provider_->SetAllowBatching(allow_batching);
+
   // Issue the POST, blocking until it finishes.
   if (!cancelation_signal_->TryRegisterHandler(this)) {
     // Return early because cancelation signal was signaled.
@@ -157,6 +161,7 @@ SyncServerConnectionManager::~SyncServerConnectionManager() = default;
 HttpResponse SyncServerConnectionManager::PostBuffer(
     const std::string& buffer_in,
     const std::string& access_token,
+    bool allow_batching,
     std::string* buffer_out) {
   if (access_token.empty()) {
     // Print a log to distinguish this "known failure" from others.
@@ -174,8 +179,8 @@ HttpResponse SyncServerConnectionManager::PostBuffer(
 
   // Note that the post may be aborted by now, which will just cause Init to
   // fail with CONNECTION_UNAVAILABLE.
-  HttpResponse http_response =
-      connection->Init(sync_request_url_, access_token, buffer_in);
+  HttpResponse http_response = connection->Init(sync_request_url_, access_token,
+                                                buffer_in, allow_batching);
 
   if (http_response.server_status == HttpResponse::SYNC_AUTH_ERROR) {
     ClearAccessToken();

components/sync/engine/net/sync_server_connection_manager.h

@@ -37,6 +37,7 @@ class SyncServerConnectionManager : public ServerConnectionManager {
 
   HttpResponse PostBuffer(const std::string& buffer_in,
                           const std::string& access_token,
+                          bool allow_batching,
                           std::string* buffer_out) override;
 
  private:

components/sync/engine/net/sync_server_connection_manager_unittest.cc

@@ -31,6 +31,7 @@ class BlockingHttpPost : public HttpPostProvider {
   void SetPostPayload(const char* content_type,
                       int content_length,
                       const char* content) override {}
+  void SetAllowBatching(bool allow_batching) override {}
   bool MakeSynchronousPost(int* net_error_code,
                            int* http_status_code) override {
     wait_for_abort_.TimedWait(TestTimeouts::action_max_timeout());
@@ -71,7 +72,8 @@ TEST(SyncServerConnectionManagerTest, VeryEarlyAbortPost) {
       &signal);
 
   std::string buffer_out;
-  HttpResponse http_response = server.PostBuffer("", "testauth", &buffer_out);
+  HttpResponse http_response =
+      server.PostBuffer("", "testauth", /*allow_batching=*/false, &buffer_out);
 
   EXPECT_EQ(HttpResponse::CONNECTION_UNAVAILABLE, http_response.server_status);
 }
@@ -85,7 +87,8 @@ TEST(SyncServerConnectionManagerTest, EarlyAbortPost) {
 
   signal.Signal();
   std::string buffer_out;
-  HttpResponse http_response = server.PostBuffer("", "testauth", &buffer_out);
+  HttpResponse http_response =
+      server.PostBuffer("", "testauth", /*allow_batching=*/false, &buffer_out);
 
   EXPECT_EQ(HttpResponse::CONNECTION_UNAVAILABLE, http_response.server_status);
 }
@@ -105,7 +108,8 @@ TEST(SyncServerConnectionManagerTest, AbortPost) {
       TestTimeouts::tiny_timeout());
 
   std::string buffer_out;
-  HttpResponse http_response = server.PostBuffer("", "testauth", &buffer_out);
+  HttpResponse http_response =
+      server.PostBuffer("", "testauth", /*allow_batching=*/false, &buffer_out);
 
   EXPECT_EQ(HttpResponse::CONNECTION_UNAVAILABLE, http_response.server_status);
   abort_thread.Stop();
@@ -123,6 +127,7 @@ class FailingHttpPost : public HttpPostProvider {
   void SetPostPayload(const char* content_type,
                       int content_length,
                       const char* content) override {}
+  void SetAllowBatching(bool allow_batching) override {}
   bool MakeSynchronousPost(int* net_error_code,
                            int* http_status_code) override {
     *net_error_code = net_error_code_;
@@ -168,7 +173,8 @@ TEST(SyncServerConnectionManagerTest, FailPostWithTimedOut) {
       std::make_unique<FailingHttpPostFactory>(net::ERR_TIMED_OUT), &signal);
 
   std::string buffer_out;
-  HttpResponse http_response = server.PostBuffer("", "testauth", &buffer_out);
+  HttpResponse http_response =
+      server.PostBuffer("", "testauth", /*allow_batching=*/false, &buffer_out);
 
   EXPECT_EQ(HttpResponse::CONNECTION_UNAVAILABLE, http_response.server_status);
 }

components/sync/engine/sync_manager_impl_unittest.cc

@@ -61,6 +61,7 @@ class TestHttpPostProvider : public HttpPostProvider {
   void SetPostPayload(const char* content_type,
                       int content_length,
                       const char* content) override {}
+  void SetAllowBatching(bool allow_batching) override {}
   bool MakeSynchronousPost(int* net_error_code,
                            int* http_status_code) override {
     return false;

components/sync/engine/syncer_proto_util.cc

@@ -348,10 +348,19 @@ bool SyncerProtoUtil::PostAndProcessHeaders(ServerConnectionManager* scm,
 
   const base::Time start_time = base::Time::Now();
 
+  // User-initiated sync messages should not be batched. GET_UPDATES messages
+  // are mostly safe to consider non-user-initiated.
+  // TODO(https://crbug.com/1293657): Confirm that treating GET_UPDATES as
+  // non-user-initiated is reasonable. GET_UPDATES messages could be latency
+  // sensitive since these requests most commonly happen because of some
+  // user-initiated changes on a different device.
+  bool allow_batching =
+      msg.message_contents() == ClientToServerMessage::GET_UPDATES;
+
   // Fills in buffer_out.
   std::string buffer_out;
   HttpResponse http_response =
-      scm->PostBufferWithCachedAuth(buffer_in, &buffer_out);
+      scm->PostBufferWithCachedAuth(buffer_in, allow_batching, &buffer_out);
   if (http_response.server_status != HttpResponse::SERVER_CONNECTION_OK) {
     LOG(WARNING) << "Error posting from syncer:" << http_response;
     return false;

components/sync/engine/syncer_proto_util_unittest.cc

@@ -165,6 +165,7 @@ class DummyConnectionManager : public ServerConnectionManager {
 
   HttpResponse PostBuffer(const std::string& buffer_in,
                           const std::string& access_token,
+                          bool allow_batching,
                           std::string* buffer_out) override {
     if (send_error_) {
       return HttpResponse::ForIoError();

components/sync/test/engine/mock_connection_manager.cc

@@ -68,6 +68,7 @@ void MockConnectionManager::SetMidCommitObserver(
 
 HttpResponse MockConnectionManager::PostBuffer(const std::string& buffer_in,
                                                const std::string& access_token,
+                                               bool allow_batching,
                                                std::string* buffer_out) {
   ClientToServerMessage post;
   if (!post.ParseFromString(buffer_in)) {

components/sync/test/engine/mock_connection_manager.h

@@ -58,6 +58,7 @@ class MockConnectionManager : public ServerConnectionManager {
   // Overridden ServerConnectionManager functions.
   HttpResponse PostBuffer(const std::string& buffer_in,
                           const std::string& access_token,
+                          bool allow_batching,
                           std::string* buffer_out) override;
 
   // Control of commit response.

components/sync/test/fake_server/fake_server_http_post_provider.cc

@@ -66,6 +66,8 @@ void FakeServerHttpPostProvider::SetPostPayload(const char* content_type,
   request_content_.assign(content, content_length);
 }
 
+void FakeServerHttpPostProvider::SetAllowBatching(bool allow_batching) {}
+
 bool FakeServerHttpPostProvider::MakeSynchronousPost(int* net_error_code,
                                                      int* http_status_code) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);

components/sync/test/fake_server/fake_server_http_post_provider.h

@@ -37,6 +37,7 @@ class FakeServerHttpPostProvider : public syncer::HttpPostProvider {
   void SetPostPayload(const char* content_type,
                       int content_length,
                       const char* content) override;
+  void SetAllowBatching(bool allow_batching) override;
   bool MakeSynchronousPost(int* net_error_code, int* http_status_code) override;
   void Abort() override;
   int GetResponseContentLength() const override;