report_queue_provider.cc 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274
  1. // Copyright 2021 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "components/reporting/client/report_queue_provider.h"
  5. #include <memory>
  6. #include <string>
  7. #include "base/bind.h"
  8. #include "base/callback.h"
  9. #include "base/feature_list.h"
  10. #include "base/memory/ptr_util.h"
  11. #include "base/memory/ref_counted.h"
  12. #include "base/memory/scoped_refptr.h"
  13. #include "base/sequence_checker.h"
  14. #include "base/task/bind_post_task.h"
  15. #include "base/task/sequenced_task_runner.h"
  16. #include "base/task/task_traits.h"
  17. #include "base/task/thread_pool.h"
  18. #include "components/reporting/client/report_queue.h"
  19. #include "components/reporting/client/report_queue_configuration.h"
  20. #include "components/reporting/client/report_queue_impl.h"
  21. #include "components/reporting/proto/synced/record_constants.pb.h"
  22. #include "components/reporting/storage/storage_module_interface.h"
  23. #include "components/reporting/util/status.h"
  24. #include "components/reporting/util/status_macros.h"
  25. #include "components/reporting/util/statusor.h"
  26. namespace reporting {
  27. using InitCompleteCallback = base::OnceCallback<void(Status)>;
  28. // Report queue creation request. Recorded in the `create_request_queue_` when
  29. // provider cannot create queues yet.
  30. class ReportQueueProvider::CreateReportQueueRequest {
  31. public:
  32. static void New(std::unique_ptr<ReportQueueConfiguration> config,
  33. CreateReportQueueCallback create_cb) {
  34. auto* const provider = GetInstance();
  35. DCHECK(provider)
  36. << "Provider must exist, otherwise it is an internal error";
  37. auto request = base::WrapUnique(
  38. new CreateReportQueueRequest(std::move(config), std::move(create_cb)));
  39. provider->sequenced_task_runner_->PostTask(
  40. FROM_HERE,
  41. base::BindOnce(
  42. [](base::WeakPtr<ReportQueueProvider> provider,
  43. std::unique_ptr<CreateReportQueueRequest> request) {
  44. if (!provider) {
  45. std::move(request->release_create_cb())
  46. .Run(Status(error::UNAVAILABLE,
  47. "Provider has been shut down"));
  48. return;
  49. }
  50. DCHECK_CALLED_ON_VALID_SEQUENCE(provider->sequence_checker_);
  51. provider->create_request_queue_.push(std::move(request));
  52. provider->CheckInitializationState();
  53. },
  54. provider->GetWeakPtr(), std::move(request)));
  55. }
  56. CreateReportQueueRequest(const CreateReportQueueRequest& other) = delete;
  57. CreateReportQueueRequest& operator=(const CreateReportQueueRequest& other) =
  58. delete;
  59. ~CreateReportQueueRequest() = default;
  60. std::unique_ptr<ReportQueueConfiguration> release_config() {
  61. DCHECK(config_) << "Can only be released once";
  62. return std::move(config_);
  63. }
  64. ReportQueueProvider::CreateReportQueueCallback release_create_cb() {
  65. DCHECK(create_cb_) << "Can only be released once";
  66. return std::move(create_cb_);
  67. }
  68. private:
  69. // Constructor is only called by `New` factory method above.
  70. CreateReportQueueRequest(std::unique_ptr<ReportQueueConfiguration> config,
  71. CreateReportQueueCallback create_cb)
  72. : config_(std::move(config)), create_cb_(std::move(create_cb)) {}
  73. std::unique_ptr<ReportQueueConfiguration> config_;
  74. CreateReportQueueCallback create_cb_;
  75. };
  76. // ReportQueueProvider core implementation.
  77. // static
  78. bool ReportQueueProvider::IsEncryptedReportingPipelineEnabled() {
  79. return base::FeatureList::IsEnabled(kEncryptedReportingPipeline);
  80. }
  81. // static
  82. const base::Feature ReportQueueProvider::kEncryptedReportingPipeline{
  83. "EncryptedReportingPipeline", base::FEATURE_ENABLED_BY_DEFAULT};
  84. ReportQueueProvider::ReportQueueProvider(
  85. StorageModuleCreateCallback storage_create_cb)
  86. : storage_create_cb_(storage_create_cb),
  87. sequenced_task_runner_(base::ThreadPool::CreateSequencedTaskRunner(
  88. {base::TaskPriority::BEST_EFFORT, base::MayBlock()})) {
  89. DETACH_FROM_SEQUENCE(sequence_checker_);
  90. }
  91. ReportQueueProvider::~ReportQueueProvider() = default;
  92. base::WeakPtr<ReportQueueProvider> ReportQueueProvider::GetWeakPtr() {
  93. return weak_ptr_factory_.GetWeakPtr();
  94. }
  95. scoped_refptr<StorageModuleInterface> ReportQueueProvider::storage() const {
  96. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  97. return storage_;
  98. }
  99. scoped_refptr<base::SequencedTaskRunner>
  100. ReportQueueProvider::sequenced_task_runner() const {
  101. return sequenced_task_runner_;
  102. }
  103. void ReportQueueProvider::CreateNewQueue(
  104. std::unique_ptr<ReportQueueConfiguration> config,
  105. CreateReportQueueCallback cb) {
  106. sequenced_task_runner_->PostTask(
  107. FROM_HERE,
  108. base::BindOnce(
  109. [](base::WeakPtr<ReportQueueProvider> provider,
  110. std::unique_ptr<ReportQueueConfiguration> config,
  111. CreateReportQueueCallback cb) {
  112. if (!provider) {
  113. std::move(cb).Run(
  114. Status(error::UNAVAILABLE, "Provider has been shut down"));
  115. return;
  116. }
  117. // Configure report queue config with an appropriate DM token and
  118. // proceed to create the queue if configuration was successful.
  119. DCHECK_CALLED_ON_VALID_SEQUENCE(provider->sequence_checker_);
  120. auto report_queue_configured_cb = base::BindOnce(
  121. [](scoped_refptr<StorageModuleInterface> storage,
  122. CreateReportQueueCallback cb,
  123. StatusOr<std::unique_ptr<ReportQueueConfiguration>>
  124. config_result) {
  125. // If configuration hit an error, we abort and
  126. // report this through the callback
  127. if (!config_result.ok()) {
  128. std::move(cb).Run(config_result.status());
  129. return;
  130. }
  131. // Proceed to create the queue on arbitrary thread.
  132. base::ThreadPool::PostTask(
  133. FROM_HERE,
  134. base::BindOnce(&ReportQueueImpl::Create,
  135. std::move(config_result.ValueOrDie()),
  136. storage, std::move(cb)));
  137. },
  138. provider->storage_, std::move(cb));
  139. provider->ConfigureReportQueue(
  140. std::move(config), std::move(report_queue_configured_cb));
  141. },
  142. GetWeakPtr(), std::move(config), std::move(cb)));
  143. }
  144. StatusOr<std::unique_ptr<ReportQueue, base::OnTaskRunnerDeleter>>
  145. ReportQueueProvider::CreateNewSpeculativeQueue() {
  146. return SpeculativeReportQueueImpl::Create();
  147. }
  148. void ReportQueueProvider::OnInitCompleted() {}
  149. // static
  150. void ReportQueueProvider::CreateQueue(
  151. std::unique_ptr<ReportQueueConfiguration> config,
  152. CreateReportQueueCallback create_cb) {
  153. if (!IsEncryptedReportingPipelineEnabled()) {
  154. Status not_enabled = Status(
  155. error::FAILED_PRECONDITION,
  156. "The Encrypted Reporting Pipeline is not enabled. Please enable it on "
  157. "the command line using --enable-features=EncryptedReportingPipeline");
  158. VLOG(1) << not_enabled;
  159. std::move(create_cb).Run(not_enabled);
  160. return;
  161. }
  162. CreateReportQueueRequest::New(std::move(config), std::move(create_cb));
  163. }
  164. // static
  165. StatusOr<std::unique_ptr<ReportQueue, base::OnTaskRunnerDeleter>>
  166. ReportQueueProvider::CreateSpeculativeQueue(
  167. std::unique_ptr<ReportQueueConfiguration> config) {
  168. if (!IsEncryptedReportingPipelineEnabled()) {
  169. Status not_enabled = Status(
  170. error::FAILED_PRECONDITION,
  171. "The Encrypted Reporting Pipeline is not enabled. Please enable it on "
  172. "the command line using --enable-features=EncryptedReportingPipeline");
  173. VLOG(1) << not_enabled;
  174. return not_enabled;
  175. }
  176. // Instantiate speculative queue, bail out in case of an error.
  177. ASSIGN_OR_RETURN(auto speculative_queue,
  178. GetInstance()->CreateNewSpeculativeQueue());
  179. // Initiate underlying queue creation.
  180. CreateReportQueueRequest::New(
  181. std::move(config), speculative_queue->PrepareToAttachActualQueue());
  182. return speculative_queue;
  183. }
  184. void ReportQueueProvider::CheckInitializationState() {
  185. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  186. if (!storage_) {
  187. // Provider not ready.
  188. DCHECK(!create_request_queue_.empty()) << "Request queue cannot be empty";
  189. if (create_request_queue_.size() > 1) {
  190. // More than one request in the queue - it means Storage creation has
  191. // already been started.
  192. return;
  193. }
  194. // Start Storage creation on an arbitrary thread. Upon completion resume on
  195. // sequenced task runner.
  196. base::ThreadPool::PostTask(
  197. FROM_HERE,
  198. base::BindOnce(
  199. [](StorageModuleCreateCallback storage_create_cb,
  200. OnStorageModuleCreatedCallback on_storage_created_cb) {
  201. storage_create_cb.Run(std::move(on_storage_created_cb));
  202. },
  203. storage_create_cb_,
  204. base::BindPostTask(
  205. sequenced_task_runner_,
  206. base::BindOnce(&ReportQueueProvider::OnStorageModuleConfigured,
  207. GetWeakPtr()))));
  208. return;
  209. }
  210. // Storage ready, create all report queues that were submitted.
  211. // Note that `CreateNewQueue` call offsets heavy work to arbitrary threads.
  212. while (!create_request_queue_.empty()) {
  213. auto& report_queue_request = create_request_queue_.front();
  214. CreateNewQueue(report_queue_request->release_config(),
  215. report_queue_request->release_create_cb());
  216. create_request_queue_.pop();
  217. }
  218. }
  219. void ReportQueueProvider::OnStorageModuleConfigured(
  220. StatusOr<scoped_refptr<StorageModuleInterface>> storage_result) {
  221. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  222. if (!storage_result.ok()) {
  223. // Storage creation failed, kill all requests.
  224. while (!create_request_queue_.empty()) {
  225. auto& report_queue_request = create_request_queue_.front();
  226. std::move(report_queue_request->release_create_cb())
  227. .Run(Status(error::UNAVAILABLE, "Unable to build a ReportQueue"));
  228. create_request_queue_.pop();
  229. }
  230. return;
  231. }
  232. // Storage ready, create all report queues that were submitted.
  233. // Note that `CreateNewQueue` call offsets heavy work to arbitrary threads.
  234. DCHECK(!storage_) << "Storage module already recorded";
  235. OnInitCompleted();
  236. storage_ = storage_result.ValueOrDie();
  237. while (!create_request_queue_.empty()) {
  238. auto& report_queue_request = create_request_queue_.front();
  239. CreateNewQueue(report_queue_request->release_config(),
  240. report_queue_request->release_create_cb());
  241. create_request_queue_.pop();
  242. }
  243. }
  244. } // namespace reporting