Refactor upload: get sequencing info from response

Bug: 1078512
Change-Id: I2a49977934366c11f5d7a5d41f59162f6b93a2c7
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2574804
Commit-Queue: Leonid Baraz <lbaraz@chromium.org>
Reviewed-by: Sergey Poromov <poromov@chromium.org>
Reviewed-by: Zach Trudo <zatrudo@google.com>
Cr-Commit-Position: refs/heads/master@{#834342}

chrome/browser/policy/messaging_layer/upload/record_handler_impl.cc

@@ -31,6 +31,42 @@
 
 namespace reporting {
 
+// ReportUploader handles enqueuing events on the |report_queue_|,
+// and uploading those events with the |client_|.
+class RecordHandlerImpl::ReportUploader
+    : public TaskRunnerContext<DmServerUploadService::CompletionResponse> {
+ public:
+  ReportUploader(
+      std::unique_ptr<std::vector<EncryptedRecord>> records,
+      policy::CloudPolicyClient* client,
+      DmServerUploadService::CompletionCallback upload_complete_cb,
+      scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner);
+
+ private:
+  ~ReportUploader() override;
+
+  void OnStart() override;
+
+  void StartUpload(const EncryptedRecord& encrypted_record);
+  void OnUploadComplete(base::Optional<base::Value> response);
+  void HandleFailedUpload();
+  void HandleSuccessfulUpload();
+
+  void Complete(DmServerUploadService::CompletionResponse result);
+
+  std::unique_ptr<std::vector<EncryptedRecord>> records_;
+  policy::CloudPolicyClient* client_;
+
+  // Last successful response to be processed.
+  // Note: I could not find a way to pass it as a parameter,
+  // so it is a class member variable. |last_response_| must be processed before
+  // any attempt to retry calling the client, otherwise it will be overwritten.
+  base::Value last_response_;
+
+  // Set for the highest record being uploaded.
+  base::Optional<SequencingInformation> highest_sequencing_information_;
+};
+
 RecordHandlerImpl::ReportUploader::ReportUploader(
     std::unique_ptr<std::vector<EncryptedRecord>> records,
     policy::CloudPolicyClient* client,
@@ -81,7 +117,7 @@ void RecordHandlerImpl::ReportUploader::StartUpload(
       FROM_HERE, {content::BrowserThread::UI},
       base::BindOnce(
           [](policy::CloudPolicyClient* client, const EncryptedRecord& record,
-             base::OnceCallback<void(bool)> cb) {
+             base::OnceCallback<void(base::Optional<base::Value>)> cb) {
             client->UploadEncryptedReport(
                 record,
                 reporting::GetContext(ProfileManager::GetPrimaryUserProfile()),
@@ -90,12 +126,14 @@ void RecordHandlerImpl::ReportUploader::StartUpload(
           client_, encrypted_record, std::move(cb)));
 }
 
-void RecordHandlerImpl::ReportUploader::OnUploadComplete(bool success) {
-  if (!success) {
+void RecordHandlerImpl::ReportUploader::OnUploadComplete(
+    base::Optional<base::Value> response) {
+  if (!response.has_value()) {
     Schedule(&RecordHandlerImpl::ReportUploader::HandleFailedUpload,
              base::Unretained(this));
     return;
   }
+  last_response_ = std::move(response.value());
   Schedule(&RecordHandlerImpl::ReportUploader::HandleSuccessfulUpload,
            base::Unretained(this));
 }
@@ -116,7 +154,36 @@ void RecordHandlerImpl::ReportUploader::HandleFailedUpload() {
 }
 
 void RecordHandlerImpl::ReportUploader::HandleSuccessfulUpload() {
-  highest_sequencing_information_ = records_->back().sequencing_information();
+  // Decypher 'response' containing a base::Value dictionary that looks like:
+  //  {
+  //    "lastSucceedUploadedRecord": ... // SequencingInformation proto
+  //    "firstFailedUploadedRecord": {
+  //      "failedUploadedRecord": ... // SequencingInformation proto
+  //      "failureStatus": ... // Status proto
+  //    }
+  //  }
+  // TODO(b/169883262): Factor out the decoding into a separate class.
+
+  const base::Value* last_succeed_uploaded_record =
+      last_response_.FindDictKey("lastSucceedUploadedRecord");
+  if (last_succeed_uploaded_record != nullptr) {
+    SequencingInformation seq_info;
+    // Note: Fields below are 'int', should be converted into 'uint64_t'.
+    const auto sequencing_id =
+        last_succeed_uploaded_record->FindIntKey("sequencingId");
+    const auto generation_id =
+        last_succeed_uploaded_record->FindIntKey("generationId");
+    const auto priority = last_succeed_uploaded_record->FindIntKey("priority");
+    if (sequencing_id.has_value() && generation_id.has_value() &&
+        priority.has_value() && Priority_IsValid(priority.value())) {
+      seq_info.set_sequencing_id(sequencing_id.value());
+      seq_info.set_generation_id(generation_id.value());
+      seq_info.set_priority(Priority(priority.value()));
+      highest_sequencing_information_ = std::move(seq_info);
+    }
+  }
+  // TODO(b/169883262): Decode and handle failure information.
+  // TODO(b/170054326): Handle the encryption settings.
 
   // Pop the last record that was processed.
   records_->pop_back();

chrome/browser/policy/messaging_layer/upload/record_handler_impl.h

@@ -32,36 +32,6 @@ namespace reporting {
 // queue.
 class RecordHandlerImpl : public DmServerUploadService::RecordHandler {
  public:
-  // ReportUploader handles enqueuing events on the |report_queue_|,
-  // and uploading those events with the |client_|.
-  class ReportUploader
-      : public TaskRunnerContext<DmServerUploadService::CompletionResponse> {
-   public:
-    ReportUploader(
-        std::unique_ptr<std::vector<EncryptedRecord>> records,
-        policy::CloudPolicyClient* client,
-        DmServerUploadService::CompletionCallback upload_complete_cb,
-        scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner);
-
-   private:
-    ~ReportUploader() override;
-
-    void OnStart() override;
-
-    void StartUpload(const EncryptedRecord& encrypted_record);
-    void OnUploadComplete(bool success);
-    void HandleFailedUpload();
-    void HandleSuccessfulUpload();
-
-    void Complete(DmServerUploadService::CompletionResponse completion_result);
-
-    std::unique_ptr<std::vector<EncryptedRecord>> records_;
-    policy::CloudPolicyClient* client_;
-
-    // Set for the highest record being uploaded.
-    base::Optional<SequencingInformation> highest_sequencing_information_;
-  };
-
   explicit RecordHandlerImpl(policy::CloudPolicyClient* client);
   ~RecordHandlerImpl() override;
 
@@ -71,6 +41,14 @@ class RecordHandlerImpl : public DmServerUploadService::RecordHandler {
       DmServerUploadService::CompletionCallback upload_complete) override;
 
  private:
+  // Helper |ReportUploader| class handles enqueuing events on the
+  // |report_queue_|, and uploading those events with the |client_|.
+  class ReportUploader;
+
+  // Processes last JSON response received from the server in case of success,
+  // or nullopt in case of failures on all attempts.
+  void ProcessResponse(const base::Value& response);
+
   scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
 };
 

chrome/browser/policy/messaging_layer/upload/record_handler_impl_unittest.cc

@@ -81,6 +81,21 @@ class TestCompletionResponder {
               (DmServerUploadService::CompletionResponse));
 };
 
+// Helper function composes JSON represented as base::Value from Sequencing
+// information.
+base::Value ValueFromSucceededSequencingInfo(
+    const SequencingInformation& sequencing_information) {
+  base::Value sequencing_info(base::Value::Type::DICTIONARY);
+  sequencing_info.SetIntKey("sequencingId",
+                            sequencing_information.sequencing_id());
+  sequencing_info.SetIntKey("generationId",
+                            sequencing_information.generation_id());
+  sequencing_info.SetIntKey("priority", sequencing_information.priority());
+  base::Value result(base::Value::Type::DICTIONARY);
+  result.SetPath("lastSucceedUploadedRecord", std::move(sequencing_info));
+  return result;
+}
+
 class RecordHandlerImplTest : public testing::Test {
  public:
   RecordHandlerImplTest()
@@ -124,9 +139,12 @@ TEST_F(RecordHandlerImplTest, ForwardsRecordsToCloudPolicyClient) {
   TestCallbackWaiterWithCounter client_waiter{kNumTestRecords};
   EXPECT_CALL(*client_, UploadEncryptedReport(_, _, _))
       .Times(kNumTestRecords)
-      .WillRepeatedly(WithArgs<2>(
-          Invoke([&client_waiter](base::OnceCallback<void(bool)> callback) {
-            std::move(callback).Run(true);
+      .WillRepeatedly(WithArgs<0, 2>(
+          Invoke([&client_waiter](
+                     const ::reporting::EncryptedRecord& record,
+                     policy::CloudPolicyClient::ResponseCallback callback) {
+            std::move(callback).Run(ValueFromSucceededSequencingInfo(
+                record.sequencing_information()));
             client_waiter.Signal();
           })));
 
@@ -134,6 +152,7 @@ TEST_F(RecordHandlerImplTest, ForwardsRecordsToCloudPolicyClient) {
 
   TestCallbackWaiter responder_waiter;
   TestCompletionResponder responder;
+  ::testing::InSequence seq;
   EXPECT_CALL(responder, RecordsHandled(ValueEqualsProto(
                              test_records->back().sequencing_information())))
       .WillOnce(Invoke([&responder_waiter]() { responder_waiter.Signal(); }));
@@ -159,15 +178,19 @@ TEST_F(RecordHandlerImplTest, ReportsEarlyFailure) {
   ::testing::InSequence seq;
   EXPECT_CALL(*client_, UploadEncryptedReport(_, _, _))
       .Times(kNumSuccessfulUploads)
-      .WillRepeatedly(WithArgs<2>(
-          Invoke([&client_waiter](base::OnceCallback<void(bool)> callback) {
-            std::move(callback).Run(true);
+      .WillRepeatedly(WithArgs<0, 2>(Invoke(
+          [&client_waiter](
+              const ::reporting::EncryptedRecord& record,
+              base::OnceCallback<void(base::Optional<base::Value>)> callback) {
+            std::move(callback).Run(ValueFromSucceededSequencingInfo(
+                record.sequencing_information()));
             client_waiter.Signal();
           })));
   EXPECT_CALL(*client_, UploadEncryptedReport(_, _, _))
-      .WillOnce(WithArgs<2>(
-          Invoke([&client_waiter](base::OnceCallback<void(bool)> callback) {
-            std::move(callback).Run(false);
+      .WillOnce(WithArgs<2>(Invoke(
+          [&client_waiter](
+              base::OnceCallback<void(base::Optional<base::Value>)> callback) {
+            std::move(callback).Run(base::nullopt);
             client_waiter.Signal();
           })));
 

chrome/browser/policy/messaging_layer/upload/upload_client_unittest.cc

@@ -119,6 +119,21 @@ class TestCallbackWaiterWithCounter : public TestCallbackWaiter {
   std::atomic<size_t> counter_limit_;
 };
 
+// Helper function composes JSON represented as base::Value from Sequencing
+// information.
+base::Value ValueFromSucceededSequencingInfo(
+    const SequencingInformation& sequencing_information) {
+  base::Value sequencing_info(base::Value::Type::DICTIONARY);
+  sequencing_info.SetIntKey("sequencingId",
+                            sequencing_information.sequencing_id());
+  sequencing_info.SetIntKey("generationId",
+                            sequencing_information.generation_id());
+  sequencing_info.SetIntKey("priority", sequencing_information.priority());
+  base::Value result(base::Value::Type::DICTIONARY);
+  result.SetPath("lastSucceedUploadedRecord", std::move(sequencing_info));
+  return result;
+}
+
 class UploadClientTest : public ::testing::Test {
  public:
   UploadClientTest() = default;
@@ -195,9 +210,11 @@ TEST_F(UploadClientTest, CreateUploadClientAndUploadRecords) {
       policy::DMToken::CreateValidTokenForTesting("FAKE_DM_TOKEN").value());
 
   EXPECT_CALL(*client, UploadEncryptedReport(_, _, _))
-      .WillRepeatedly(WithArgs<2>(
-          Invoke([&waiter](base::OnceCallback<void(bool)> callback) {
-            std::move(callback).Run(true);
+      .WillRepeatedly(WithArgs<0, 2>(Invoke(
+          [&waiter](const ::reporting::EncryptedRecord& record,
+                    policy::CloudPolicyClient::ResponseCallback callback) {
+            std::move(callback).Run(ValueFromSucceededSequencingInfo(
+                record.sequencing_information()));
             base::ThreadPool::PostTask(
                 FROM_HERE, {base::TaskPriority::BEST_EFFORT},
                 base::BindOnce(&TestCallbackWaiterWithCounter::Signal,

components/policy/core/common/cloud/cloud_policy_client.cc

@@ -609,10 +609,10 @@ void CloudPolicyClient::UploadSecurityEventReport(
 void CloudPolicyClient::UploadEncryptedReport(
     const ::reporting::EncryptedRecord& record,
     base::Optional<base::Value> context,
-    StatusCallback callback) {
+    ResponseCallback callback) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   if (!is_registered()) {
-    std::move(callback).Run(false);
+    std::move(callback).Run(base::nullopt);
     return;
   }
 
@@ -1295,20 +1295,23 @@ void CloudPolicyClient::OnRealtimeReportUploadCompleted(
 
 // |job| can be null if the owning EncryptedReportingJobConfiguration is
 // destroyed prior to calling OnUploadComplete. In that case, callback will be
-// called with false value.
+// called with nullopt value.
 void CloudPolicyClient::OnEncryptedReportUploadCompleted(
-    StatusCallback callback,
+    ResponseCallback callback,
     DeviceManagementService::Job* job,
     DeviceManagementStatus status,
     int net_error,
     const base::Value& response) {
   if (job == nullptr) {
-    std::move(callback).Run(false);
+    std::move(callback).Run(base::nullopt);
     return;
   }
-
-  OnRealtimeReportUploadCompleted(std::move(callback), job, status, net_error,
-                                  response);
+  status_ = status;
+  if (status != DM_STATUS_SUCCESS) {
+    NotifyClientError();
+  }
+  std::move(callback).Run(response.Clone());
+  RemoveJob(job);
 }
 
 void CloudPolicyClient::OnRemoteCommandsFetched(

components/policy/core/common/cloud/cloud_policy_client.h

@@ -79,6 +79,11 @@ class POLICY_EXPORT CloudPolicyClient {
   using DeviceDMTokenCallback = base::RepeatingCallback<std::string(
       const std::vector<std::string>& user_affiliation_ids)>;
 
+  // A callback that processes response value received from the server,
+  // or nullopt, if there was a failure.
+  using ResponseCallback =
+      base::OnceCallback<void(base::Optional<base::Value>)>;
+
   using ClientCertProvisioningStartCsrCallback = base::OnceCallback<void(
       DeviceManagementStatus,
       base::Optional<
@@ -319,7 +324,7 @@ class POLICY_EXPORT CloudPolicyClient {
   // completes.
   virtual void UploadEncryptedReport(const ::reporting::EncryptedRecord& record,
                                      base::Optional<base::Value> context,
-                                     StatusCallback callback);
+                                     ResponseCallback callback);
 
   // Uploads a report on the status of app push-installs. The client must be in
   // a registered state. The |callback| will be called when the operation
@@ -647,7 +652,7 @@ class POLICY_EXPORT CloudPolicyClient {
                                        const base::Value& response);
 
   // Callback for encrypted report upload requests.
-  void OnEncryptedReportUploadCompleted(StatusCallback callback,
+  void OnEncryptedReportUploadCompleted(ResponseCallback callback,
                                         DeviceManagementService::Job* job,
                                         DeviceManagementStatus status,
                                         int net_error,

components/policy/core/common/cloud/cloud_policy_client_unittest.cc

@@ -46,10 +46,16 @@ using testing::_;
 using testing::DoAll;
 using testing::ElementsAre;
 using testing::Mock;
+using testing::Not;
 using testing::Return;
 using testing::SaveArg;
 using testing::StrictMock;
 
+// Matcher for base::Optional. Can be combined with Not().
+MATCHER(HasValue, "Has value") {
+  return arg.has_value();
+}
+
 // The type for variables containing an error from DM Server response.
 using CertProvisioningResponseErrorType =
     enterprise_management::ClientCertificateProvisioningResponse::Error;
@@ -142,6 +148,13 @@ class MockRobotAuthCodeCallbackObserver {
                void(DeviceManagementStatus, const std::string&));
 };
 
+class MockResponseCallbackObserver {
+ public:
+  MockResponseCallbackObserver() = default;
+
+  MOCK_METHOD1(OnResponseReceived, void(base::Optional<base::Value>));
+};
+
 }  // namespace
 
 class CloudPolicyClientTest : public testing::Test {
@@ -595,11 +608,11 @@ class CloudPolicyClientTest : public testing::Test {
   void AttemptUploadEncryptedWaitUntilIdle(
       const ::reporting::EncryptedRecord& record,
       base::Optional<base::Value> context = base::nullopt) {
-    CloudPolicyClient::StatusCallback callback =
-        base::BindOnce(&MockStatusCallbackObserver::OnCallbackComplete,
-                       base::Unretained(&callback_observer_));
+    CloudPolicyClient::ResponseCallback response_callback =
+        base::BindOnce(&MockResponseCallbackObserver::OnResponseReceived,
+                       base::Unretained(&response_callback_observer_));
     client_->UploadEncryptedReport(record, std::move(context),
-                                   std::move(callback));
+                                   std::move(response_callback));
     base::RunLoop().RunUntilIdle();
   }
 
@@ -652,6 +665,7 @@ class CloudPolicyClientTest : public testing::Test {
       device_dmtoken_callback_observer_;
   StrictMock<MockRobotAuthCodeCallbackObserver>
       robot_auth_code_callback_observer_;
+  StrictMock<MockResponseCallbackObserver> response_callback_observer_;
   FakeSigningService fake_signing_service_;
   std::unique_ptr<CloudPolicyClient> client_;
   network::TestURLLoaderFactory url_loader_factory_;
@@ -1676,7 +1690,8 @@ TEST_F(CloudPolicyClientTest, UploadEncryptedReport) {
   RegisterClient();
   ExpectEncryptedReport();
 
-  EXPECT_CALL(callback_observer_, OnCallbackComplete(true)).Times(1);
+  EXPECT_CALL(response_callback_observer_, OnResponseReceived(HasValue()))
+      .Times(1);
   AttemptUploadEncryptedWaitUntilIdle(record);
 
   EXPECT_EQ(
@@ -1691,17 +1706,23 @@ TEST_F(CloudPolicyClientTest, DenyPoorlyFormedEncryptedRecords) {
   // Create empty record
   ::reporting::EncryptedRecord record;
 
-  EXPECT_CALL(callback_observer_, OnCallbackComplete(false)).Times(4);
-
+  EXPECT_CALL(response_callback_observer_, OnResponseReceived(Not(HasValue())))
+      .Times(1);
   AttemptUploadEncryptedWaitUntilIdle(record);
 
   // Add encrypted_wrapped_record without sequencing information.
   record.set_encrypted_wrapped_record("Enterprise");
+
+  EXPECT_CALL(response_callback_observer_, OnResponseReceived(Not(HasValue())))
+      .Times(1);
   AttemptUploadEncryptedWaitUntilIdle(record);
 
   // Incorrectly set sequencing information by only setting sequencing id.
   auto* sequencing_information = record.mutable_sequencing_information();
   sequencing_information->set_sequencing_id(1701);
+
+  EXPECT_CALL(response_callback_observer_, OnResponseReceived(Not(HasValue())))
+      .Times(1);
   AttemptUploadEncryptedWaitUntilIdle(record);
 
   // Finish correctly setting sequencing information but incorrectly set
@@ -1712,12 +1733,15 @@ TEST_F(CloudPolicyClientTest, DenyPoorlyFormedEncryptedRecords) {
   auto* encryption_info = record.mutable_encryption_info();
   encryption_info->set_encryption_key("Key");
 
+  EXPECT_CALL(response_callback_observer_, OnResponseReceived(Not(HasValue())))
+      .Times(1);
   AttemptUploadEncryptedWaitUntilIdle(record);
 
   // Finish correctly setting encryption info - expect complete call.
   encryption_info->set_public_key_id(1234);
 
-  EXPECT_CALL(callback_observer_, OnCallbackComplete(true)).Times(1);
+  EXPECT_CALL(response_callback_observer_, OnResponseReceived(HasValue()))
+      .Times(1);
   ExpectEncryptedReport();
 
   AttemptUploadEncryptedWaitUntilIdle(record);

components/policy/core/common/cloud/mock_cloud_policy_client.h

@@ -113,7 +113,7 @@ class MockCloudPolicyClient : public CloudPolicyClient {
   MOCK_METHOD3(UploadEncryptedReport,
                void(const ::reporting::EncryptedRecord&,
                     base::Optional<base::Value>,
-                    StatusCallback));
+                    ResponseCallback));
 
   void UploadAppInstallReport(base::Value value,
                               StatusCallback callback) override {