Make ReportClient::CreateReportQueue() async

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

chrome/browser/policy/messaging_layer/public/report_client.h

@@ -8,11 +8,14 @@
 #include <memory>
 #include <utility>
 
+#include "base/containers/queue.h"
+#include "base/memory/singleton.h"
 #include "chrome/browser/policy/messaging_layer/encryption/encryption_module.h"
 #include "chrome/browser/policy/messaging_layer/public/report_queue.h"
 #include "chrome/browser/policy/messaging_layer/public/report_queue_configuration.h"
 #include "chrome/browser/policy/messaging_layer/storage/storage_module.h"
 #include "chrome/browser/policy/messaging_layer/upload/upload_client.h"
+#include "chrome/browser/policy/messaging_layer/util/shared_queue.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
 #include "chrome/browser/policy/messaging_layer/util/task_runner_context.h"
 
@@ -67,36 +70,181 @@ class ReportingClient {
     scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
   };
 
+  struct Configuration {
+    Configuration();
+    ~Configuration();
+
+    scoped_refptr<StorageModule> storage_;
+    scoped_refptr<EncryptionModule> encryption_;
+  };
+
+  using CreateReportQueueResponse = StatusOr<std::unique_ptr<ReportQueue>>;
+
+  using CreateReportQueueCallback =
+      base::OnceCallback<void(CreateReportQueueResponse)>;
+
+  using UpdateConfigurationCallback =
+      base::OnceCallback<void(const Configuration&,
+                              base::OnceCallback<void(Status)>)>;
+
+  using InitCompleteCallback = base::OnceCallback<void(Status)>;
+
+  using InitializationStatusCallback = base::OnceCallback<void(Status)>;
+
+  class InitializationStateTracker
+      : public base::RefCountedThreadSafe<InitializationStateTracker> {
+   public:
+    using ReleaseLeaderCallback = base::OnceCallback<void(bool)>;
+    using LeaderPromotionRequestCallback =
+        base::OnceCallback<void(StatusOr<ReleaseLeaderCallback>)>;
+    using GetInitStateCallback = base::OnceCallback<void(bool)>;
+
+    static scoped_refptr<InitializationStateTracker> Create();
+
+    // Will call |get_init_state_cb| with |is_initialized_| value.
+    void GetInitState(GetInitStateCallback get_init_state_cb);
+
+    // Will promote one initializer to leader at a time. Will deny
+    // initialization requests if the ReportingClient is already initialized. If
+    // there are no errors will return a ReleaseLeaderCallback for releasing the
+    // initializing leadership.
+    //
+    // Error code responses:
+    // RESOURCE_EXHAUSTED - Returned when a promotion is requested when there is
+    //     already a leader.
+    // FAILED_PRECONDITION - Returned when a promotion is requested when
+    //     ReportingClient is already initialized.
+    void RequestLeaderPromotion(
+        LeaderPromotionRequestCallback promo_request_cb);
+
+   private:
+    friend class base::RefCountedThreadSafe<InitializationStateTracker>;
+    InitializationStateTracker();
+    virtual ~InitializationStateTracker();
+
+    void OnIsInitializedRequest(GetInitStateCallback get_init_state_cb);
+
+    void OnLeaderPromotionRequest(
+        LeaderPromotionRequestCallback promo_request_cb);
+
+    void ReleaseLeader(bool initialization_successful);
+    void OnLeaderRelease(bool initialization_successful);
+
+    bool has_promoted_initializing_context_{false};
+    bool is_initialized_{false};
+
+    scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
+  };
+
+  class InitializingContext : public TaskRunnerContext<Status> {
+   public:
+    InitializingContext(
+        Storage::StartUploadCb start_upload_cb,
+        UpdateConfigurationCallback update_config_cb,
+        InitCompleteCallback init_complete_cb,
+        scoped_refptr<InitializationStateTracker> init_state_tracker,
+        scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner);
+
+   private:
+    ~InitializingContext() override;
+
+    // OnStart will begin the process of configuring the ReportClient.
+    void OnStart() override;
+    void OnLeaderPromotionResult(
+        StatusOr<InitializationStateTracker::ReleaseLeaderCallback>
+            promo_result);
+
+    // ConfigureStorageModule will build a StorageModule and add it to the
+    // |client_config_|.
+    void ConfigureStorageModule();
+    void OnStorageModuleConfigured(
+        StatusOr<scoped_refptr<StorageModule>> storage_result);
+
+    // ConfigureEncryptionModule will build an |EncryptionModule| and add it
+    // to the |client_config_|.
+    void ConfigureEncryptionModule();
+    void OnEncryptionModuleConfigured(
+        StatusOr<scoped_refptr<EncryptionModule>> encryption_result);
+
+    void UpdateConfiguration();
+
+    // Complete calls response with |client_config_|
+    void Complete(Status status);
+
+    Storage::StartUploadCb start_upload_cb_;
+    UpdateConfigurationCallback update_config_cb_;
+    scoped_refptr<InitializationStateTracker> init_state_tracker_;
+
+    InitializationStateTracker::ReleaseLeaderCallback release_leader_cb_;
+    Configuration client_config_;
+  };
+
   ~ReportingClient();
   ReportingClient(const ReportingClient& other) = delete;
   ReportingClient& operator=(const ReportingClient& other) = delete;
 
-  // Allows a user to synchronously create a |ReportQueue|. Will create an
-  // underlying ReportingClient if it doesn't exists. This call can fail if
-  // |storage_| or |encryption_| cannot be instantiated for any reason.
+  // Allows a user to asynchronously create a |ReportQueue|. Will create an
+  // underlying ReportingClient if it doesn't exists. The callback will contain
+  // an error if |storage_| or |encryption_| cannot be instantiated for any
+  // reason.
   //
   // TODO(chromium:1078512): Once the StorageModule is ready, update this
   // comment with concrete failure conditions.
   // TODO(chromium:1078512): Once the EncryptionModule is ready, update this
   // comment with concrete failure conditions.
-  static StatusOr<std::unique_ptr<ReportQueue>> CreateReportQueue(
-      std::unique_ptr<ReportQueueConfiguration> config);
+  static void CreateReportQueue(
+      std::unique_ptr<ReportQueueConfiguration> config,
+      CreateReportQueueCallback create_cb);
+
+  // Resets the singleton object. Should only be used in tests when the current
+  // TaskEnvironment will be invalidated.
+  static void Reset_test();
 
  private:
-  explicit ReportingClient(scoped_refptr<StorageModule> storage);
+  // Holds the creation request for a ReportQueue.
+  class CreateReportQueueRequest {
+   public:
+    CreateReportQueueRequest(std::unique_ptr<ReportQueueConfiguration> config,
+                             CreateReportQueueCallback create_cb);
+    ~CreateReportQueueRequest();
+    CreateReportQueueRequest(CreateReportQueueRequest&& other);
+
+    std::unique_ptr<ReportQueueConfiguration> config();
+    CreateReportQueueCallback create_cb();
+
+   private:
+    std::unique_ptr<ReportQueueConfiguration> config_;
+    CreateReportQueueCallback create_cb_;
+  };
+
+  friend struct base::DefaultSingletonTraits<ReportingClient>;
+
+  ReportingClient();
+  static ReportingClient* GetInstance();
 
-  static StatusOr<ReportingClient*> GetInstance();
+  void OnPushComplete();
+  void OnInitState(bool reporting_client_configured);
+  void OnConfigResult(const Configuration& config,
+                      base::OnceCallback<void(Status)> continue_init_cb);
+  void OnInitializationComplete(Status init_status);
 
-  // ReportingClient is not meant to be used directly.
-  static StatusOr<std::unique_ptr<ReportingClient>> Create();
+  void ClearRequestQueue(base::queue<CreateReportQueueRequest> failed_requests);
+  void BuildRequestQueue(StatusOr<CreateReportQueueRequest> pop_result);
 
   // TODO(chromium:1078512) Priority is unused, remove it.
   static StatusOr<std::unique_ptr<Storage::UploaderInterface>> BuildUploader(
       Priority priority);
 
+  // Queue for storing creation requests while the ReportingClient is
+  // initializing.
+  scoped_refptr<SharedQueue<CreateReportQueueRequest>> create_request_queue_;
+
+  scoped_refptr<InitializationStateTracker> init_state_tracker_;
+
   scoped_refptr<StorageModule> storage_;
   scoped_refptr<EncryptionModule> encryption_;
   std::unique_ptr<UploadClient> upload_client_;
+  Configuration config_;
 };
 
 }  // namespace reporting

chrome/browser/policy/messaging_layer/public/report_client_unittest.cc

@@ -4,6 +4,8 @@
 
 #include "chrome/browser/policy/messaging_layer/public/report_client.h"
 
+#include "base/memory/singleton.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/test/task_environment.h"
 #include "chrome/browser/policy/messaging_layer/public/report_queue.h"
 #include "chrome/browser/policy/messaging_layer/public/report_queue_configuration.h"
@@ -21,10 +23,31 @@ using policy::DMToken;
 using reporting::Destination;
 using reporting::Priority;
 
-class ReportingClientTest : public testing::Test {
+class TestCallbackWaiter {
+ public:
+  TestCallbackWaiter()
+      : completed_(base::WaitableEvent::ResetPolicy::MANUAL,
+                   base::WaitableEvent::InitialState::NOT_SIGNALED) {}
+
+  virtual void Signal() {
+    DCHECK(!completed_.IsSignaled());
+    completed_.Signal();
+  }
+
+  void Wait() { completed_.Wait(); }
+  void Reset() { completed_.Reset(); }
+
  protected:
-  base::test::TaskEnvironment task_envrionment_;
+  base::WaitableEvent completed_;
+};
 
+class ReportingClientTest : public testing::Test {
+ public:
+  void TearDown() override { ReportingClient::Reset_test(); }
+
+ protected:
+  base::test::TaskEnvironment task_envrionment_{
+      base::test::TaskEnvironment::TimeSource::MOCK_TIME};
   const DMToken dm_token_ = DMToken::CreateValidTokenForTesting("TOKEN");
   const Destination destination_ = Destination::UPLOAD_EVENTS;
   const Priority priority_ = Priority::IMMEDIATE;
@@ -36,12 +59,24 @@ class ReportingClientTest : public testing::Test {
 TEST_F(ReportingClientTest, CreatesReportQueue) {
   auto config_result = ReportQueueConfiguration::Create(
       dm_token_, destination_, priority_, policy_checker_callback_);
-  EXPECT_TRUE(config_result.ok());
+  ASSERT_OK(config_result);
 
-  auto report_queue_result =
-      ReportingClient::CreateReportQueue(std::move(config_result.ValueOrDie()));
+  TestCallbackWaiter waiter;
+  StatusOr<std::unique_ptr<ReportQueue>> result;
+  auto create_report_queue_cb = base::BindOnce(
+      [](TestCallbackWaiter* waiter,
+         StatusOr<std::unique_ptr<ReportQueue>>* result,
+         StatusOr<std::unique_ptr<ReportQueue>> create_result) {
+        *result = std::move(create_result);
+        waiter->Signal();
+      },
+      &waiter, &result);
+  ReportingClient::CreateReportQueue(std::move(config_result.ValueOrDie()),
+                                     std::move(create_report_queue_cb));
 
-  EXPECT_TRUE(report_queue_result.ok());
+  waiter.Wait();
+  waiter.Reset();
+  ASSERT_OK(result);
 }
 
 // Ensures that created ReportQueues are actually different.
@@ -50,20 +85,41 @@ TEST_F(ReportingClientTest, CreatesTwoDifferentReportQueues) {
       dm_token_, destination_, priority_, policy_checker_callback_);
   EXPECT_TRUE(config_result.ok());
 
-  auto report_queue_result_1 =
-      ReportingClient::CreateReportQueue(std::move(config_result.ValueOrDie()));
-  EXPECT_TRUE(report_queue_result_1.ok());
+  TestCallbackWaiter waiter;
+  StatusOr<std::unique_ptr<ReportQueue>> result;
+  auto create_report_queue_cb = base::BindOnce(
+      [](TestCallbackWaiter* waiter,
+         StatusOr<std::unique_ptr<ReportQueue>>* result,
+         StatusOr<std::unique_ptr<ReportQueue>> create_result) {
+        *result = std::move(create_result);
+        waiter->Signal();
+      },
+      &waiter, &result);
+  ReportingClient::CreateReportQueue(std::move(config_result.ValueOrDie()),
+                                     std::move(create_report_queue_cb));
+  waiter.Wait();
+  waiter.Reset();
+  ASSERT_OK(result);
+  auto report_queue_1 = std::move(result.ValueOrDie());
 
   config_result = ReportQueueConfiguration::Create(
       dm_token_, destination_, priority_, policy_checker_callback_);
-  EXPECT_TRUE(config_result.ok());
+  create_report_queue_cb = base::BindOnce(
+      [](TestCallbackWaiter* waiter,
+         StatusOr<std::unique_ptr<ReportQueue>>* result,
+         StatusOr<std::unique_ptr<ReportQueue>> create_result) {
+        *result = std::move(create_result);
+        waiter->Signal();
+      },
+      &waiter, &result);
+  ReportingClient::CreateReportQueue(std::move(config_result.ValueOrDie()),
+                                     std::move(create_report_queue_cb));
+  waiter.Wait();
+  ASSERT_OK(result);
 
-  auto report_queue_result_2 =
-      ReportingClient::CreateReportQueue(std::move(config_result.ValueOrDie()));
-  EXPECT_TRUE(report_queue_result_2.ok());
+  auto report_queue_2 = std::move(result.ValueOrDie());
 
-  EXPECT_NE(report_queue_result_1.ValueOrDie().get(),
-            report_queue_result_2.ValueOrDie().get());
+  EXPECT_NE(report_queue_1.get(), report_queue_2.get());
 }
 
 }  // namespace