Send reports for completed conversions

This CL implements reporting for the Conversion Measurement API.

Reports are periodically fetched from ConversionStorage once every
30 minutes by the ConversionManager, which puts them into a queue to be
sent.

ConversionReporter implements this priority_queue and keeps a timer
which fires when the next report is eligible. Reports are sent via
SimpleUrlLoaders using the browsers SharedUrlLoaderFactory.

Once a report is sent, it is deleted from storage. There is currently
no logic for retries due to connection errors.

Change-Id: I5ad28ebc6940618fe24b965b371e6097dbabf177
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2050848
Commit-Queue: John Delaney <johnidel@chromium.org>
Reviewed-by: Charlie Harrison <csharrison@chromium.org>
Reviewed-by: Christian Dullweber <dullweber@chromium.org>
Reviewed-by: Avi Drissman <avi@chromium.org>
Cr-Commit-Position: refs/heads/master@{#758279}

content/browser/BUILD.gn

@@ -684,10 +684,14 @@ jumbo_source_set("browser") {
     "conversions/conversion_manager.h",
     "conversions/conversion_manager_impl.cc",
     "conversions/conversion_manager_impl.h",
+    "conversions/conversion_network_sender_impl.cc",
+    "conversions/conversion_network_sender_impl.h",
     "conversions/conversion_policy.cc",
     "conversions/conversion_policy.h",
     "conversions/conversion_report.cc",
     "conversions/conversion_report.h",
+    "conversions/conversion_reporter_impl.cc",
+    "conversions/conversion_reporter_impl.h",
     "conversions/conversion_storage.h",
     "conversions/conversion_storage_delegate_impl.cc",
     "conversions/conversion_storage_delegate_impl.h",

content/browser/conversions/conversion_host_unittest.cc

@@ -8,6 +8,7 @@
 
 #include "base/test/scoped_feature_list.h"
 #include "content/browser/conversions/conversion_manager.h"
+#include "content/browser/conversions/conversion_test_utils.h"
 #include "content/browser/web_contents/web_contents_impl.h"
 #include "content/public/common/content_features.h"
 #include "content/public/test/test_renderer_host.h"
@@ -36,36 +37,6 @@ Impression CreateValidImpression() {
 
 }  // namespace
 
-class TestConversionManager : public ConversionManager {
- public:
-  TestConversionManager() = default;
-  ~TestConversionManager() override = default;
-
-  void HandleImpression(const StorableImpression& impression) override {
-    num_impressions_++;
-  }
-
-  void HandleConversion(const StorableConversion& impression) override {
-    num_conversions_++;
-  }
-
-  const ConversionPolicy& GetConversionPolicy() const override {
-    return policy;
-  }
-
-  size_t num_impressions() const { return num_impressions_; }
-
-  void Reset() {
-    num_impressions_ = 0u;
-    num_conversions_ = 0u;
-  }
-
- private:
-  ConversionPolicy policy;
-  size_t num_impressions_ = 0u;
-  size_t num_conversions_ = 0u;
-};
-
 class TestManagerProvider : public ConversionManager::Provider {
  public:
   explicit TestManagerProvider(ConversionManager* manager)

content/browser/conversions/conversion_manager.h

@@ -41,6 +41,10 @@ class CONTENT_EXPORT ConversionManager {
   // conversion reports to storage.
   virtual void HandleConversion(const StorableConversion& conversion) = 0;
 
+  // Notify storage to delete the given |conversion_id| when it's associated
+  // report has been sent.
+  virtual void HandleSentReport(int64_t conversion_id) = 0;
+
   // Returns the ConversionPolicy that is used to control API policies such
   // as noise.
   virtual const ConversionPolicy& GetConversionPolicy() const = 0;

content/browser/conversions/conversion_manager_impl.cc

@@ -7,18 +7,49 @@
 #include <memory>
 
 #include "base/bind.h"
+#include "base/memory/ptr_util.h"
 #include "base/task_runner_util.h"
 #include "base/time/default_clock.h"
+#include "content/browser/conversions/conversion_reporter_impl.h"
 #include "content/browser/conversions/conversion_storage_delegate_impl.h"
 #include "content/browser/conversions/conversion_storage_sql.h"
 
 namespace content {
 
+const constexpr base::TimeDelta kConversionManagerQueueReportsInterval =
+    base::TimeDelta::FromMinutes(30);
+
+// static
+std::unique_ptr<ConversionManagerImpl> ConversionManagerImpl::CreateForTesting(
+    std::unique_ptr<ConversionReporter> reporter,
+    const base::Clock* clock,
+    const base::FilePath& user_data_directory,
+    scoped_refptr<base::SequencedTaskRunner> storage_task_runner) {
+  return base::WrapUnique<ConversionManagerImpl>(
+      new ConversionManagerImpl(std::move(reporter), clock, user_data_directory,
+                                std::move(storage_task_runner)));
+}
+
 ConversionManagerImpl::ConversionManagerImpl(
+    StoragePartition* storage_partition,
     const base::FilePath& user_data_directory,
     scoped_refptr<base::SequencedTaskRunner> task_runner)
-    : storage_task_runner_(std::move(task_runner)),
-      clock_(base::DefaultClock::GetInstance()),
+    : ConversionManagerImpl(std::make_unique<ConversionReporterImpl>(
+                                storage_partition,
+                                this,
+                                base::DefaultClock::GetInstance()),
+                            base::DefaultClock::GetInstance(),
+                            user_data_directory,
+                            std::move(task_runner)) {}
+
+ConversionManagerImpl::ConversionManagerImpl(
+    std::unique_ptr<ConversionReporter> reporter,
+    const base::Clock* clock,
+    const base::FilePath& user_data_directory,
+    scoped_refptr<base::SequencedTaskRunner> storage_task_runner)
+    : storage_task_runner_(std::move(storage_task_runner)),
+      clock_(clock),
+      reporter_(std::move(reporter)),
       storage_(new ConversionStorageSql(
                    user_data_directory,
                    std::make_unique<ConversionStorageDelegateImpl>(),
@@ -66,13 +97,64 @@ void ConversionManagerImpl::HandleConversion(
           base::Unretained(storage_.get()), conversion));
 }
 
+void ConversionManagerImpl::HandleSentReport(int64_t conversion_id) {
+  storage_task_runner_->PostTask(
+      FROM_HERE,
+      base::BindOnce(base::IgnoreResult(&ConversionStorage::DeleteConversion),
+                     base::Unretained(storage_.get()), conversion_id));
+}
+
 const ConversionPolicy& ConversionManagerImpl::GetConversionPolicy() const {
   return *conversion_policy_;
 }
 
 void ConversionManagerImpl::OnInitCompleted(bool success) {
-  if (!success)
+  if (!success) {
     storage_.reset();
+    return;
+  }
+
+  // Once the database is loaded, get all reports that may have expired while
+  // Chrome was not running and handle these specially.
+  GetAndHandleReports(
+      base::BindOnce(&ConversionManagerImpl::HandleReportsExpiredAtStartup,
+                     weak_factory_.GetWeakPtr()));
+
+  // Start a repeating timer that will fetch reports once every
+  // |kConversionManagerQueueReportsInterval| and add them to |reporter_|.
+  get_and_queue_reports_timer_.Start(
+      FROM_HERE, kConversionManagerQueueReportsInterval, this,
+      &ConversionManagerImpl::GetAndQueueReportsForNextInterval);
+}
+
+void ConversionManagerImpl::GetAndHandleReports(
+    ReportsHandlerFunc handler_function) {
+  base::PostTaskAndReplyWithResult(
+      storage_task_runner_.get(), FROM_HERE,
+      base::BindOnce(&ConversionStorage::GetConversionsToReport,
+                     base::Unretained(storage_.get()),
+                     clock_->Now() + kConversionManagerQueueReportsInterval),
+      std::move(handler_function));
+}
+
+void ConversionManagerImpl::GetAndQueueReportsForNextInterval() {
+  // Get all the reports that will be reported in the next interval and them to
+  // the |reporter_|.
+  GetAndHandleReports(base::BindOnce(&ConversionManagerImpl::QueueReports,
+                                     weak_factory_.GetWeakPtr()));
+}
+
+void ConversionManagerImpl::QueueReports(
+    std::vector<ConversionReport> reports) {
+  if (!reports.empty())
+    reporter_->AddReportsToQueue(std::move(reports));
+}
+
+void ConversionManagerImpl::HandleReportsExpiredAtStartup(
+    std::vector<ConversionReport> reports) {
+  // TODO(https://crbug.com/1054119): We need to add special logic to ensure
+  // that these reports are not temporally joinable.
+  QueueReports(std::move(reports));
 }
 
 }  // namespace content

content/browser/conversions/conversion_manager_impl.h

@@ -12,6 +12,7 @@
 #include "base/memory/scoped_refptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/sequenced_task_runner.h"
+#include "base/timer/timer.h"
 #include "content/browser/conversions/conversion_manager.h"
 #include "content/browser/conversions/conversion_policy.h"
 #include "content/browser/conversions/conversion_storage.h"
@@ -22,12 +23,39 @@ class Clock;
 
 namespace content {
 
+// Frequency we pull ConversionReports from storage and queue them to be
+// reported.
+extern CONTENT_EXPORT const base::TimeDelta
+    kConversionManagerQueueReportsInterval;
+
+class ConversionStorage;
+class StoragePartition;
+
 // UI thread class that manages the lifetime of the underlying conversion
 // storage. Owned by the storage partition.
-class ConversionManagerImpl : public ConversionManager {
+class CONTENT_EXPORT ConversionManagerImpl : public ConversionManager {
  public:
+  // Interface which manages the ownership, queuing, and sending of pending
+  // conversion reports. Owned by |this|.
+  class ConversionReporter {
+   public:
+    virtual ~ConversionReporter() = default;
+
+    // Adds |reports| to a shared queue of reports that need to be sent. The
+    // reporter needs to notify it's owning manager when a report has been sent
+    // via ConversionManager::HandleSentReport().
+    virtual void AddReportsToQueue(std::vector<ConversionReport> reports) = 0;
+  };
+
+  static std::unique_ptr<ConversionManagerImpl> CreateForTesting(
+      std::unique_ptr<ConversionReporter> reporter,
+      const base::Clock* clock,
+      const base::FilePath& user_data_directory,
+      scoped_refptr<base::SequencedTaskRunner> storage_task_runner);
+
   // |storage_task_runner| should run with base::TaskPriority::BEST_EFFORT.
   ConversionManagerImpl(
+      StoragePartition* storage_partition,
       const base::FilePath& user_data_directory,
       scoped_refptr<base::SequencedTaskRunner> storage_task_runner);
   ConversionManagerImpl(const ConversionManagerImpl& other) = delete;
@@ -37,16 +65,44 @@ class ConversionManagerImpl : public ConversionManager {
   // ConversionManager:
   void HandleImpression(const StorableImpression& impression) override;
   void HandleConversion(const StorableConversion& conversion) override;
+  void HandleSentReport(int64_t conversion_id) override;
   const ConversionPolicy& GetConversionPolicy() const override;
 
  private:
+  ConversionManagerImpl(
+      std::unique_ptr<ConversionReporter> reporter,
+      const base::Clock* clock,
+      const base::FilePath& user_data_directory,
+      scoped_refptr<base::SequencedTaskRunner> storage_task_runner);
+
   void OnInitCompleted(bool success);
 
+  using ReportsHandlerFunc =
+      base::OnceCallback<void(std::vector<ConversionReport>)>;
+  void GetAndHandleReports(ReportsHandlerFunc handler_function);
+
+  // Get the next set of reports from storage that need to be sent before the
+  // next call from |get_and_queue_reports_timer_|. Adds the reports to
+  // |reporter|.
+  void GetAndQueueReportsForNextInterval();
+
+  // Queue the given |reports| on |reporter_|.
+  void QueueReports(std::vector<ConversionReport> reports);
+
+  void HandleReportsExpiredAtStartup(std::vector<ConversionReport> reports);
+
   // Task runner used to perform operations on |storage_|. Runs with
   // base::TaskPriority::BEST_EFFORT.
   scoped_refptr<base::SequencedTaskRunner> storage_task_runner_;
 
-  base::Clock* clock_;
+  const base::Clock* clock_;
+
+  // Timer which administers calls to GetAndQueueReportsForNextInterval().
+  base::RepeatingTimer get_and_queue_reports_timer_;
+
+  // Handle keeping track of conversion reports to send. Reports are fetched
+  // from |storage_| and added to |reporter_| by |get_reports_timer_|.
+  std::unique_ptr<ConversionReporter> reporter_;
 
   // ConversionStorage instance which is scoped to lifetime of
   // |storage_task_runner_|. |storage_| should be accessed by calling

content/browser/conversions/conversion_manager_impl_unittest.cc

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/conversions/conversion_manager_impl.h"
+
+#include <stdint.h>
+#include <memory>
+
+#include "base/callback_forward.h"
+#include "base/files/scoped_temp_dir.h"
+#include "base/run_loop.h"
+#include "base/sequenced_task_runner.h"
+#include "base/time/clock.h"
+#include "base/time/time.h"
+#include "content/browser/conversions/conversion_test_utils.h"
+#include "content/public/test/browser_task_environment.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace content {
+
+namespace {
+
+// Mock reporter that tracks reports being queued by the ConversionManager.
+class TestConversionReporter
+    : public ConversionManagerImpl::ConversionReporter {
+ public:
+  TestConversionReporter() = default;
+  ~TestConversionReporter() override = default;
+
+  // ConversionManagerImpl::ConversionReporter
+  void AddReportsToQueue(std::vector<ConversionReport> reports) override {
+    num_reports_ += reports.size();
+    last_conversion_id_ = *reports.back().conversion_id;
+
+    if (quit_closure_ && num_reports_ >= expected_num_reports_)
+      std::move(quit_closure_).Run();
+  }
+
+  size_t num_reports() { return num_reports_; }
+
+  int64_t last_conversion_id() { return last_conversion_id_; }
+
+  void WaitForNumReports(size_t expected_num_reports) {
+    if (num_reports_ >= expected_num_reports)
+      return;
+
+    expected_num_reports_ = expected_num_reports;
+    base::RunLoop wait_loop;
+    quit_closure_ = wait_loop.QuitClosure();
+    wait_loop.Run();
+    ;
+  }
+
+ private:
+  size_t expected_num_reports_ = 0u;
+  size_t num_reports_ = 0u;
+  int64_t last_conversion_id_ = 0UL;
+  base::OnceClosure quit_closure_;
+};
+
+// Time after impression that a conversion can first be sent. See
+// ConversionStorageDelegateImpl::GetReportTimeForConversion().
+constexpr base::TimeDelta kFirstReportingWindow = base::TimeDelta::FromDays(2);
+
+// Give impressions a sufficiently long expiry.
+constexpr base::TimeDelta kImpressionExpiry = base::TimeDelta::FromDays(30);
+
+}  // namespace
+
+class ConversionManagerImplTest : public testing::Test {
+ public:
+  ConversionManagerImplTest()
+      : task_environment_(base::test::TaskEnvironment::TimeSource::MOCK_TIME) {
+    EXPECT_TRUE(dir_.CreateUniqueTempDir());
+    CreateManager();
+  }
+
+  void CreateManager() {
+    auto reporter = std::make_unique<TestConversionReporter>();
+    test_reporter_ = reporter.get();
+    conversion_manager_ = ConversionManagerImpl::CreateForTesting(
+        std::move(reporter), task_environment_.GetMockClock(), dir_.GetPath(),
+        base::ThreadPool::CreateSequencedTaskRunner({base::MayBlock()}));
+  }
+
+  const base::Clock& clock() { return *task_environment_.GetMockClock(); }
+
+ protected:
+  base::ScopedTempDir dir_;
+  BrowserTaskEnvironment task_environment_;
+  std::unique_ptr<ConversionManagerImpl> conversion_manager_;
+  TestConversionReporter* test_reporter_ = nullptr;
+};
+
+TEST_F(ConversionManagerImplTest, ImpressionConverted_ReportQueued) {
+  conversion_manager_->HandleImpression(
+      ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
+  conversion_manager_->HandleConversion(DefaultConversion());
+
+  // Reports are queued in intervals ahead of when they should be
+  // sent. Make sure the report is not queued earlier than this.
+  task_environment_.FastForwardBy(kFirstReportingWindow -
+                                  kConversionManagerQueueReportsInterval -
+                                  base::TimeDelta::FromMinutes(1));
+  EXPECT_EQ(0u, test_reporter_->num_reports());
+
+  task_environment_.FastForwardBy(base::TimeDelta::FromMinutes(1));
+  EXPECT_EQ(1u, test_reporter_->num_reports());
+}
+
+TEST_F(ConversionManagerImplTest, QueuedReportNotSent_QueuedAgain) {
+  conversion_manager_->HandleImpression(
+      ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
+  conversion_manager_->HandleConversion(DefaultConversion());
+  task_environment_.FastForwardBy(kFirstReportingWindow -
+                                  kConversionManagerQueueReportsInterval);
+  EXPECT_EQ(1u, test_reporter_->num_reports());
+
+  // If the report is not sent, it should be added to the queue again.
+  task_environment_.FastForwardBy(kConversionManagerQueueReportsInterval);
+  EXPECT_EQ(2u, test_reporter_->num_reports());
+}
+
+TEST_F(ConversionManagerImplTest, QueuedReportSent_NotQueuedAgain) {
+  conversion_manager_->HandleImpression(
+      ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
+  conversion_manager_->HandleConversion(DefaultConversion());
+  task_environment_.FastForwardBy(kFirstReportingWindow -
+                                  kConversionManagerQueueReportsInterval);
+  EXPECT_EQ(1u, test_reporter_->num_reports());
+
+  // Notify the manager that the report has been sent.
+  conversion_manager_->HandleSentReport(test_reporter_->last_conversion_id());
+
+  // The report should not be added to the queue again.
+  task_environment_.FastForwardBy(kConversionManagerQueueReportsInterval);
+  EXPECT_EQ(1u, test_reporter_->num_reports());
+}
+
+// Add a conversion to storage and reset the manager to mimic a report being
+// available at startup.
+TEST_F(ConversionManagerImplTest, ExpiredReportsAtStartup_Queued) {
+  // Create a report that will be reported at t= 2 days.
+  conversion_manager_->HandleImpression(
+      ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
+  conversion_manager_->HandleConversion(DefaultConversion());
+
+  // Create another conversion that will be reported at t=
+  // (kFirstReportingWindow + 2 * kConversionManagerQueueReportsInterval).
+  task_environment_.FastForwardBy(2 * kConversionManagerQueueReportsInterval);
+  conversion_manager_->HandleImpression(
+      ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
+  conversion_manager_->HandleConversion(DefaultConversion());
+
+  EXPECT_EQ(0u, test_reporter_->num_reports());
+
+  // Reset the manager to simulate shutdown.
+  conversion_manager_.reset();
+
+  // Fast forward past the expected report time of the first conversion, t =
+  // (kFirstReportingWindow+ 1 minute).
+  task_environment_.FastForwardBy(kFirstReportingWindow -
+                                  (2 * kConversionManagerQueueReportsInterval) +
+                                  base::TimeDelta::FromMinutes(1));
+
+  // Create the manager and check that the first report is queued immediately.
+  CreateManager();
+  test_reporter_->WaitForNumReports(1);
+  EXPECT_EQ(1u, test_reporter_->num_reports());
+  conversion_manager_->HandleSentReport(test_reporter_->last_conversion_id());
+
+  // The second report is still queued at the correct time.
+  task_environment_.FastForwardBy(kConversionManagerQueueReportsInterval);
+  EXPECT_EQ(2u, test_reporter_->num_reports());
+}
+
+}  // namespace content

content/browser/conversions/conversion_network_sender_impl.cc

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/conversions/conversion_network_sender_impl.h"
+
+#include "base/bind.h"
+#include "base/strings/strcat.h"
+#include "base/strings/string_number_conversions.h"
+#include "base/time/time.h"
+#include "content/public/browser/storage_partition.h"
+#include "net/traffic_annotation/network_traffic_annotation.h"
+#include "services/network/public/cpp/resource_request.h"
+#include "services/network/public/cpp/simple_url_loader.h"
+#include "url/gurl.h"
+#include "url/origin.h"
+#include "url/url_canon.h"
+
+namespace content {
+
+namespace {
+
+GURL GetReportUrl(const content::ConversionReport& report) {
+  url::Replacements<char> replacements;
+  const char kEndpointPath[] = "/.well-known/register-conversion";
+  replacements.SetPath(kEndpointPath, url::Component(0, strlen(kEndpointPath)));
+  std::string query = base::StrCat(
+      {"impression-data=", report.impression.impression_data(),
+       "&conversion-data=", report.conversion_data,
+       "&credit=", base::NumberToString(report.attribution_credit)});
+  replacements.SetQuery(query.c_str(), url::Component(0, query.length()));
+  return report.impression.reporting_origin().GetURL().ReplaceComponents(
+      replacements);
+}
+
+}  // namespace
+
+ConversionNetworkSenderImpl::ConversionNetworkSenderImpl(
+    StoragePartition* storage_partition)
+    : storage_partition_(storage_partition) {}
+
+ConversionNetworkSenderImpl::~ConversionNetworkSenderImpl() = default;
+
+void ConversionNetworkSenderImpl::SendReport(ConversionReport* report,
+                                             ReportSentCallback sent_callback) {
+  // The browser process URLLoaderFactory is not created by default, so don't
+  // create it until it is directly needed.
+  if (!url_loader_factory_) {
+    url_loader_factory_ =
+        storage_partition_->GetURLLoaderFactoryForBrowserProcess();
+  }
+
+  auto resource_request = std::make_unique<network::ResourceRequest>();
+  resource_request->url = GetReportUrl(*report);
+  resource_request->referrer = report->impression.conversion_origin().GetURL();
+  resource_request->method = "POST";
+  resource_request->credentials_mode = network::mojom::CredentialsMode::kOmit;
+
+  // TODO(https://crbug.com/1058018): Update the "policy" field in the traffic
+  // annotation when a setting to disable the API is properly
+  // surfaced/implemented.
+  net::NetworkTrafficAnnotationTag traffic_annotation =
+      net::DefineNetworkTrafficAnnotation("conversion_measurement_report", R"(
+        semantics {
+          sender: "Event-level Conversion Measurement API"
+          description:
+            "The Conversion Measurement API allows sites to measure "
+            "conversions (e.g. purchases) and attribute them to clicked ads, "
+            "without using cross-site persistent identifiers like third party "
+            "cookies."
+          trigger:
+            "When a registered conversion has become eligible for reporting."
+          data:
+            "A high-entropy identifier declared by the site in which the user "
+            "clicked on an impression. A noisy low entropy data value declared "
+            "on the conversion site. A browser generated value that denotes "
+            "if this was the last impression clicked prior to conversion."
+          destination:OTHER
+        }
+        policy {
+          cookies_allowed: NO
+          setting:
+            "This feature cannot be disabled by settings."
+          policy_exception_justification: "Not implemented."
+        })");
+
+  auto simple_url_loader = network::SimpleURLLoader::Create(
+      std::move(resource_request), traffic_annotation);
+  network::SimpleURLLoader* simple_url_loader_ptr = simple_url_loader.get();
+
+  auto it = loaders_in_progress_.insert(loaders_in_progress_.begin(),
+                                        std::move(simple_url_loader));
+  simple_url_loader_ptr->SetTimeoutDuration(base::TimeDelta::FromSeconds(30));
+
+  // Unretained is safe because the URLLoader is owned by |this| and will be
+  // deleted before |this|.
+  simple_url_loader_ptr->DownloadHeadersOnly(
+      url_loader_factory_.get(),
+      base::BindOnce(&ConversionNetworkSenderImpl::OnReportSent,
+                     base::Unretained(this), std::move(it),
+                     std::move(sent_callback)));
+}
+
+void ConversionNetworkSenderImpl::SetURLLoaderFactoryForTesting(
+    scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory) {
+  url_loader_factory_ = url_loader_factory;
+}
+
+void ConversionNetworkSenderImpl::OnReportSent(
+    UrlLoaderList::iterator it,
+    ReportSentCallback sent_callback,
+    scoped_refptr<net::HttpResponseHeaders> headers) {
+  // TODO(https://crbug.com/1054127): Log metrics for success/failure of sending
+  // reports. This should inspect the HTTP response code from the headers HTTP
+  // failures and SimpleUrlLoader::NetError() for internal errors/timeouts.
+  loaders_in_progress_.erase(it);
+  std::move(sent_callback).Run();
+}
+
+}  // namespace content

content/browser/conversions/conversion_network_sender_impl.h

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CONTENT_BROWSER_CONVERSIONS_CONVERSION_NETWORK_SENDER_IMPL_H_
+#define CONTENT_BROWSER_CONVERSIONS_CONVERSION_NETWORK_SENDER_IMPL_H_
+
+#include <stdint.h>
+#include <list>
+#include <memory>
+
+#include "base/callback.h"
+#include "content/browser/conversions/conversion_report.h"
+#include "content/browser/conversions/conversion_reporter_impl.h"
+#include "content/common/content_export.h"
+#include "services/network/public/cpp/shared_url_loader_factory.h"
+
+namespace network {
+class SimpleURLLoader;
+}  // namespace network
+
+namespace content {
+
+class StoragePartition;
+
+// Implemented a NetworkSender capable of issuing POST requests for complete
+// conversions. Maintains a set of all ongoing UrlLoaders used for posting
+// conversion reports. Created and owned by ConversionReporterImpl.
+class CONTENT_EXPORT ConversionNetworkSenderImpl
+    : public ConversionReporterImpl::NetworkSender {
+ public:
+  explicit ConversionNetworkSenderImpl(StoragePartition* storage_partition);
+  ConversionNetworkSenderImpl(const ConversionNetworkSenderImpl&) = delete;
+  ConversionNetworkSenderImpl& operator=(const ConversionNetworkSenderImpl&) =
+      delete;
+  ~ConversionNetworkSenderImpl() override;
+
+  // Generates a resource request for |report| and creates a new UrlLoader to
+  // send it. A report is only attempted to be sent once, with a timeout of 30
+  // seconds. |report| is destroyed after this call finishes.
+  // |sent_callback| is run after the request finishes, whether or not it
+  // succeeded,
+  void SendReport(ConversionReport* report,
+                  ReportSentCallback sent_callback) override;
+
+  // Tests inject a TestURLLoaderFactory so they can mock the network response.
+  void SetURLLoaderFactoryForTesting(
+      scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory);
+
+ private:
+  // This is a std::list so that iterators remain valid during modifications.
+  using UrlLoaderList = std::list<std::unique_ptr<network::SimpleURLLoader>>;
+
+  // Called when headers are available for a sent report.
+  void OnReportSent(UrlLoaderList::iterator it,
+                    ReportSentCallback sent_callback,
+                    scoped_refptr<net::HttpResponseHeaders> headers);
+
+  // Reports that are actively being sent.
+  UrlLoaderList loaders_in_progress_;
+
+  // Must outlive |this|.
+  StoragePartition* storage_partition_;
+
+  // Lazily accessed URLLoaderFactory used for network requests.
+  scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory_;
+};
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_CONVERSIONS_CONVERSION_NETWORK_SENDER_IMPL_H_

content/browser/conversions/conversion_network_sender_impl_unittest.cc

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/conversions/conversion_network_sender_impl.h"
+
+#include "base/bind.h"
+#include "base/strings/strcat.h"
+#include "base/task/post_task.h"
+#include "base/test/scoped_feature_list.h"
+#include "base/test/simple_test_clock.h"
+#include "content/browser/conversions/conversion_test_utils.h"
+#include "content/browser/storage_partition_impl.h"
+#include "content/public/browser/browser_context.h"
+#include "content/public/browser/storage_partition.h"
+#include "content/public/common/content_features.h"
+#include "content/public/test/browser_task_environment.h"
+#include "content/public/test/test_browser_context.h"
+#include "services/network/public/cpp/resource_request.h"
+#include "services/network/public/cpp/shared_url_loader_factory.h"
+#include "services/network/public/cpp/weak_wrapper_shared_url_loader_factory.h"
+#include "services/network/test/test_url_loader_factory.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace content {
+
+namespace {
+
+// Gets a report url which matches reports created by GetReport().
+std::string GetReportUrl(std::string impression_data) {
+  return base::StrCat(
+      {"https://report.test/.well-known/register-conversion?impression-data=",
+       impression_data, "&conversion-data=", impression_data, "&credit=0"});
+}
+
+// Create a simple report where impression data/conversion data/conversion id
+// are all the same.
+ConversionReport GetReport(int64_t conversion_id) {
+  return ConversionReport(
+      ImpressionBuilder(base::Time())
+          .SetData(base::NumberToString(conversion_id))
+          .Build(),
+      /*conversion_data=*/base::NumberToString(conversion_id),
+      /*report_time=*/base::Time(),
+      /*conversion_id=*/conversion_id);
+}
+
+}  // namespace
+
+class ConversionNetworkSenderTest : public testing::Test {
+ public:
+  ConversionNetworkSenderTest()
+      : task_environment_(base::test::TaskEnvironment::TimeSource::MOCK_TIME),
+        network_sender_(std::make_unique<ConversionNetworkSenderImpl>(
+            /*sotrage_partition=*/nullptr)),
+        shared_url_loader_factory_(
+            base::MakeRefCounted<network::WeakWrapperSharedURLLoaderFactory>(
+                &test_url_loader_factory_)) {
+    network_sender_->SetURLLoaderFactoryForTesting(shared_url_loader_factory_);
+  }
+
+  ConversionReporterImpl::NetworkSender::ReportSentCallback GetSentCallback() {
+    return base::BindOnce(&ConversionNetworkSenderTest::OnReportSent,
+                          base::Unretained(this));
+  }
+
+ protected:
+  size_t num_reports_sent_ = 0u;
+
+  // |task_enviorment_| must be initialized first.
+  content::BrowserTaskEnvironment task_environment_;
+
+  // Unique ptr so it can be reset during testing.
+  std::unique_ptr<ConversionNetworkSenderImpl> network_sender_;
+  network::TestURLLoaderFactory test_url_loader_factory_;
+
+ private:
+  void OnReportSent() { num_reports_sent_++; }
+
+  scoped_refptr<network::SharedURLLoaderFactory> shared_url_loader_factory_;
+};
+
+TEST_F(ConversionNetworkSenderTest,
+       ConversionReportReceived_NetworkRequestMade) {
+  auto report = GetReport(/*conversion_id=*/1);
+  network_sender_->SendReport(&report, std::move(base::DoNothing()));
+  EXPECT_EQ(1, test_url_loader_factory_.NumPending());
+  EXPECT_TRUE(test_url_loader_factory_.SimulateResponseForPendingRequest(
+      GetReportUrl("1"), ""));
+}
+
+TEST_F(ConversionNetworkSenderTest, ReportSent_CallbackFired) {
+  auto report = GetReport(/*conversion_id=*/1);
+  network_sender_->SendReport(&report, GetSentCallback());
+  EXPECT_EQ(1, test_url_loader_factory_.NumPending());
+  EXPECT_TRUE(test_url_loader_factory_.SimulateResponseForPendingRequest(
+      GetReportUrl("1"), ""));
+  EXPECT_EQ(1u, num_reports_sent_);
+}
+
+TEST_F(ConversionNetworkSenderTest, SenderDeletedDuringRequest_NoCrash) {
+  auto report = GetReport(/*conversion_id=*/1);
+  network_sender_->SendReport(&report, GetSentCallback());
+  EXPECT_EQ(1, test_url_loader_factory_.NumPending());
+  network_sender_.reset();
+  EXPECT_FALSE(test_url_loader_factory_.SimulateResponseForPendingRequest(
+      GetReportUrl("1"), ""));
+  EXPECT_EQ(0u, num_reports_sent_);
+}
+
+TEST_F(ConversionNetworkSenderTest, ReportRequestHangs_TimesOut) {
+  auto report = GetReport(/*conversion_id=*/1);
+  network_sender_->SendReport(&report, GetSentCallback());
+  EXPECT_EQ(1, test_url_loader_factory_.NumPending());
+
+  // The request should time out after 30 seconds.
+  task_environment_.FastForwardBy(base::TimeDelta::FromSeconds(30));
+
+  EXPECT_EQ(0, test_url_loader_factory_.NumPending());
+
+  // Also verify that the sent callback runs if the request times out.
+  EXPECT_EQ(1u, num_reports_sent_);
+}
+
+TEST_F(ConversionNetworkSenderTest, ReportSent_QueryParamsSetCorrectly) {
+  auto impression =
+      ImpressionBuilder(base::Time())
+          .SetData("impression")
+          .SetReportingOrigin(url::Origin::Create(GURL("https://a.com")))
+          .Build();
+  ConversionReport report(impression,
+                          /*conversion_data=*/"conversion",
+                          /*report_time=*/base::Time(),
+                          /*conversion_id=*/1);
+  report.attribution_credit = 50;
+  network_sender_->SendReport(&report, base::DoNothing());
+
+  std::string expected_report_url(
+      "https://a.com/.well-known/"
+      "register-conversion?impression-data=impression&conversion-data="
+      "conversion&credit=50");
+  EXPECT_TRUE(test_url_loader_factory_.SimulateResponseForPendingRequest(
+      expected_report_url, ""));
+}
+
+TEST_F(ConversionNetworkSenderTest, ReportSent_RequestAttributesSet) {
+  auto impression =
+      ImpressionBuilder(base::Time())
+          .SetData("1")
+          .SetReportingOrigin(url::Origin::Create(GURL("https://a.com")))
+          .SetConversionOrigin(url::Origin::Create(GURL("https://b.com")))
+          .Build();
+  ConversionReport report(impression,
+                          /*conversion_data=*/"1",
+                          /*report_time=*/base::Time(),
+                          /*conversion_id=*/1);
+  network_sender_->SendReport(&report, base::DoNothing());
+
+  const network::ResourceRequest* pending_request;
+  std::string expected_report_url(
+      "https://a.com/.well-known/"
+      "register-conversion?impression-data=1&conversion-data=1&credit=0");
+  EXPECT_TRUE(test_url_loader_factory_.IsPending(expected_report_url,
+                                                 &pending_request));
+
+  // Ensure that the request is sent with no credentials.
+  EXPECT_EQ(network::mojom::CredentialsMode::kOmit,
+            pending_request->credentials_mode);
+  EXPECT_EQ("POST", pending_request->method);
+  EXPECT_EQ(GURL("https://b.com"), pending_request->referrer);
+}
+
+TEST_F(ConversionNetworkSenderTest, ReportResultsInHttpError_SentCallbackRuns) {
+  auto report = GetReport(/*conversion_id=*/1);
+  network_sender_->SendReport(&report, GetSentCallback());
+  EXPECT_EQ(0u, num_reports_sent_);
+
+  // We should run the sent callback even if there is an http error.
+  EXPECT_TRUE(test_url_loader_factory_.SimulateResponseForPendingRequest(
+      GetReportUrl("1"), "", net::HttpStatusCode::HTTP_BAD_REQUEST));
+  EXPECT_EQ(1u, num_reports_sent_);
+}
+
+TEST_F(ConversionNetworkSenderTest, ManyReports_AllSentSuccessfully) {
+  for (int i = 0; i < 10; i++) {
+    auto report = GetReport(/*conversion_id=*/i);
+    network_sender_->SendReport(&report, GetSentCallback());
+  }
+  EXPECT_EQ(10, test_url_loader_factory_.NumPending());
+
+  // Send reports out of order to guarantee that callback conversion_ids are
+  // properly handled.
+  for (int i = 9; i >= 0; i--) {
+    std::string report_id = base::NumberToString(i);
+
+    EXPECT_TRUE(test_url_loader_factory_.SimulateResponseForPendingRequest(
+        GetReportUrl(report_id), ""));
+  }
+  EXPECT_EQ(10u, num_reports_sent_);
+  EXPECT_EQ(0, test_url_loader_factory_.NumPending());
+}
+
+}  // namespace content

content/browser/conversions/conversion_reporter_impl.cc

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/conversions/conversion_reporter_impl.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/rand_util.h"
+#include "base/time/clock.h"
+#include "content/browser/conversions/conversion_manager.h"
+#include "content/browser/conversions/conversion_network_sender_impl.h"
+
+namespace content {
+
+ConversionReporterImpl::ConversionReporterImpl(
+    StoragePartition* storage_partition,
+    ConversionManager* conversion_manager,
+    const base::Clock* clock)
+    : conversion_manager_(conversion_manager),
+      clock_(clock),
+      network_sender_(
+          std::make_unique<ConversionNetworkSenderImpl>(storage_partition)) {
+  DCHECK(conversion_manager_);
+}
+
+ConversionReporterImpl::~ConversionReporterImpl() = default;
+
+void ConversionReporterImpl::AddReportsToQueue(
+    std::vector<ConversionReport> reports) {
+  DCHECK(!reports.empty());
+
+  std::vector<std::unique_ptr<ConversionReport>> swappable_reports;
+  for (ConversionReport& report : reports) {
+    swappable_reports.push_back(
+        std::make_unique<ConversionReport>(std::move(report)));
+  }
+
+  // Shuffle new reports to provide plausible deniability on the ordering of
+  // reports that share the same |report_time|. This is important because
+  // multiple conversions for the same impression share the same report time if
+  // they are within the same reporting window, and we do not want to allow
+  // ordering on their conversion metadata bits.
+  base::RandomShuffle(swappable_reports.begin(), swappable_reports.end());
+
+  for (std::unique_ptr<ConversionReport>& report : swappable_reports) {
+    // If the given report is already being processed, ignore it.
+    bool inserted =
+        conversion_ids_being_processed_.insert(*(report->conversion_id)).second;
+    if (inserted)
+      report_queue_.push(std::move(report));
+  }
+  MaybeScheduleNextReport();
+}
+
+void ConversionReporterImpl::SetNetworkSenderForTesting(
+    std::unique_ptr<NetworkSender> network_sender) {
+  network_sender_ = std::move(network_sender);
+}
+
+void ConversionReporterImpl::SendNextReport() {
+  // Send the next report and remove it from the queue. Bind the conversion id
+  // to the sent callback so we know which conversion report has finished
+  // sending.
+  network_sender_->SendReport(
+      report_queue_.top().get(),
+      base::BindOnce(&ConversionReporterImpl::OnReportSent,
+                     base::Unretained(this),
+                     *report_queue_.top()->conversion_id));
+  report_queue_.pop();
+  MaybeScheduleNextReport();
+}
+
+void ConversionReporterImpl::MaybeScheduleNextReport() {
+  if (report_queue_.empty())
+    return;
+
+  send_report_timer_.Stop();
+  base::Time current_time = clock_->Now();
+  base::Time report_time = report_queue_.top()->report_time;
+
+  // Start a timer to wait until the next report is ready to be sent. This
+  // purposefully yields the thread for every report that gets scheduled.
+  // Unretained is safe because the task should never actually be posted if the
+  // timer itself is destroyed
+  send_report_timer_.Start(
+      FROM_HERE, report_time - current_time,
+      base::BindOnce(&ConversionReporterImpl::SendNextReport,
+                     base::Unretained(this)));
+}
+
+void ConversionReporterImpl::OnReportSent(int64_t conversion_id) {
+  conversion_ids_being_processed_.erase(conversion_id);
+  conversion_manager_->HandleSentReport(conversion_id);
+}
+
+bool ConversionReporterImpl::ReportComparator::operator()(
+    const std::unique_ptr<ConversionReport>& a,
+    const std::unique_ptr<ConversionReport>& b) const {
+  // Returns whether a should appear before b in ordering. Because
+  // std::priority_queue is max priority queue, we used greater then to make a
+  // min priority queue.
+  return a->report_time > b->report_time;
+}
+
+}  // namespace content

content/browser/conversions/conversion_reporter_impl.h

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CONTENT_BROWSER_CONVERSIONS_CONVERSION_REPORTER_IMPL_H_
+#define CONTENT_BROWSER_CONVERSIONS_CONVERSION_REPORTER_IMPL_H_
+
+#include <stdint.h>
+#include <memory>
+#include <queue>
+#include <vector>
+
+#include "base/containers/flat_set.h"
+#include "base/time/time.h"
+#include "base/timer/timer.h"
+#include "content/browser/conversions/conversion_manager_impl.h"
+#include "content/browser/conversions/conversion_report.h"
+#include "content/common/content_export.h"
+#include "services/network/public/cpp/shared_url_loader_factory.h"
+
+namespace base {
+class Clock;
+}  // namespace base
+
+namespace content {
+
+class ConversionManager;
+class StoragePartition;
+
+// This class is responsible for managing the dispatch of conversion reports to
+// a ConversionReporterImpl::NetworkSender. It maintains a queue of reports and
+// a timer to ensure all reports are sent at the correct time, since the time in
+// which a conversion report is sent is potentially sensitive information.
+// Created and owned by ConversionManager.
+class CONTENT_EXPORT ConversionReporterImpl
+    : public ConversionManagerImpl::ConversionReporter {
+ public:
+  // This class is responsible for sending conversion reports to their
+  // configured endpoints over the network.
+  class NetworkSender {
+   public:
+    virtual ~NetworkSender() = default;
+
+    // Callback used to notify caller that the requested report has been sent.
+    using ReportSentCallback = base::OnceCallback<void()>;
+
+    // Generates and sends a conversion report matching |report|. This should
+    // generate a secure POST quest with no-credentials. Does not persist the
+    // raw pointer.
+    virtual void SendReport(ConversionReport* report,
+                            ReportSentCallback sent_callback) = 0;
+  };
+
+  ConversionReporterImpl(StoragePartition* storage_partition,
+                         ConversionManager* manager,
+                         const base::Clock* clock);
+  ConversionReporterImpl(const ConversionReporterImpl&) = delete;
+  ConversionReporterImpl& operator=(const ConversionReporterImpl&) = delete;
+  ~ConversionReporterImpl() override;
+
+  // ConversionManagerImpl::ConversionReporter:
+  void AddReportsToQueue(std::vector<ConversionReport> reports) override;
+
+  void SetNetworkSenderForTesting(
+      std::unique_ptr<NetworkSender> network_sender);
+
+ private:
+  void MaybeScheduleNextReport();
+  void SendNextReport();
+
+  // Called when a conversion report sent via NetworkSender::SendReport() has
+  // completed loading.
+  void OnReportSent(int64_t conversion_id);
+
+  // Comparator used to order ConversionReports by their report time, with the
+  // smallest time at the top of |report_queue_|.
+  struct ReportComparator {
+    bool operator()(const std::unique_ptr<ConversionReport>& a,
+                    const std::unique_ptr<ConversionReport>& b) const;
+  };
+
+  // Priority queue which holds reports that are yet to be sent. Reports are
+  // removed from the queue when they are delivered to the NetworkSender.
+  std::priority_queue<std::unique_ptr<ConversionReport>,
+                      std::vector<std::unique_ptr<ConversionReport>>,
+                      ReportComparator>
+      report_queue_;
+
+  // Set of all conversion ids that are currently in |report_queue| or are being
+  // sent by |network_sender_|. The number of concurrent conversion reports
+  // being sent at any time is expected to be small, so a flat_set is used.
+  base::flat_set<int64_t> conversion_ids_being_processed_;
+
+  // Must outlive |this|.
+  ConversionManager* conversion_manager_;
+
+  const base::Clock* clock_;
+
+  // Timer which signals the next report in |report_queue_| should be sent.
+  base::OneShotTimer send_report_timer_;
+
+  // Responsible for issuing requests to network for report that need to be
+  // sent. Calls OnReportSent() when a report has finished sending.
+  //
+  // Should never be nullptr.
+  std::unique_ptr<NetworkSender> network_sender_;
+};
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_CONVERSIONS_CONVERSION_REPORTER_IMPL_H_

content/browser/conversions/conversion_reporter_impl_unittest.cc

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/conversions/conversion_reporter_impl.h"
+
+#include "base/sequenced_task_runner.h"
+#include "base/strings/strcat.h"
+#include "base/task/post_task.h"
+#include "base/test/simple_test_clock.h"
+#include "content/browser/conversions/conversion_manager.h"
+#include "content/browser/conversions/conversion_test_utils.h"
+#include "content/public/browser/browser_context.h"
+#include "content/public/browser/storage_partition.h"
+#include "content/public/test/browser_task_environment.h"
+#include "content/public/test/test_browser_context.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace content {
+
+namespace {
+
+// Create a report which should be sent at |report_time|. Impression
+// data/conversion data/conversion id are all the same for simplicity.
+ConversionReport GetReport(base::Time report_time, int64_t conversion_id) {
+  // Construct impressions with a null impression time as it is not used for
+  // reporting.
+  return ConversionReport(ImpressionBuilder(base::Time()).Build(),
+                          /*conversion_data=*/"", report_time,
+                          /*conversion_id=*/conversion_id);
+}
+
+// NetworkSender that keep track of the last sent report id.
+class MockNetworkSender : public ConversionReporterImpl::NetworkSender {
+ public:
+  MockNetworkSender() = default;
+
+  void SendReport(ConversionReport* conversion_report,
+                  ReportSentCallback sent_callback) override {
+    last_sent_report_id_ = *conversion_report->conversion_id;
+    num_reports_sent_++;
+    std::move(sent_callback).Run();
+  }
+
+  int64_t last_sent_report_id() { return last_sent_report_id_; }
+
+  size_t num_reports_sent() { return num_reports_sent_; }
+
+ private:
+  size_t num_reports_sent_ = 0u;
+  int64_t last_sent_report_id_ = -1;
+};
+
+}  // namespace
+
+class ConversionReporterImplTest : public testing::Test {
+ public:
+  ConversionReporterImplTest()
+      : task_environment_(base::test::TaskEnvironment::TimeSource::MOCK_TIME),
+        browser_context_(std::make_unique<TestBrowserContext>()),
+        reporter_(std::make_unique<ConversionReporterImpl>(
+            BrowserContext::GetDefaultStoragePartition(browser_context_.get()),
+            &test_manager_,
+            task_environment_.GetMockClock())) {
+    auto network_sender = std::make_unique<MockNetworkSender>();
+    sender_ = network_sender.get();
+    reporter_->SetNetworkSenderForTesting(std::move(network_sender));
+  }
+
+  const base::Clock& clock() { return *task_environment_.GetMockClock(); }
+
+ protected:
+  // |task_enviorment_| must be initialized first.
+  content::BrowserTaskEnvironment task_environment_;
+  std::unique_ptr<TestBrowserContext> browser_context_;
+
+  TestConversionManager test_manager_;
+  std::unique_ptr<ConversionReporterImpl> reporter_;
+  MockNetworkSender* sender_;
+};
+
+TEST_F(ConversionReporterImplTest,
+       ReportAddedWithImmediateReportTime_ReportSent) {
+  reporter_->AddReportsToQueue({GetReport(clock().Now(), /*conversion_id=*/1)});
+
+  // Fast forward by 0, as we yield the thread when a report is scheduled to be
+  // sent.
+  task_environment_.FastForwardBy(base::TimeDelta());
+  EXPECT_EQ(1, sender_->last_sent_report_id());
+  EXPECT_EQ(1, test_manager_.last_sent_report_id());
+}
+
+TEST_F(ConversionReporterImplTest,
+       ReportWithReportTimeBeforeCurrentTime_ReportSent) {
+  reporter_->AddReportsToQueue({GetReport(
+      clock().Now() - base::TimeDelta::FromHours(10), /*conversion_id=*/1)});
+
+  // Fast forward by 0, as we yield the thread when a report is scheduled to be
+  // sent.
+  task_environment_.FastForwardBy(base::TimeDelta());
+  EXPECT_EQ(1, sender_->last_sent_report_id());
+  EXPECT_EQ(1, test_manager_.last_sent_report_id());
+}
+
+TEST_F(ConversionReporterImplTest,
+       ReportWithDelayedReportTime_NotSentUntilDelay) {
+  const base::TimeDelta delay = base::TimeDelta::FromMinutes(30);
+
+  reporter_->AddReportsToQueue(
+      {GetReport(clock().Now() + delay, /*conversion_id=*/1)});
+  task_environment_.FastForwardBy(base::TimeDelta());
+  EXPECT_EQ(0u, sender_->num_reports_sent());
+
+  task_environment_.FastForwardBy(delay - base::TimeDelta::FromSeconds(1));
+  EXPECT_EQ(0u, sender_->num_reports_sent());
+
+  task_environment_.FastForwardBy(base::TimeDelta::FromSeconds(1));
+  EXPECT_EQ(1u, sender_->num_reports_sent());
+}
+
+TEST_F(ConversionReporterImplTest, DuplicateReportScheduled_Ignored) {
+  reporter_->AddReportsToQueue({GetReport(
+      clock().Now() + base::TimeDelta::FromMinutes(1), /*conversion_id=*/1)});
+
+  // A duplicate report should not be scheduled.
+  reporter_->AddReportsToQueue({GetReport(
+      clock().Now() + base::TimeDelta::FromMinutes(1), /*conversion_id=*/1)});
+  task_environment_.FastForwardBy(base::TimeDelta::FromMinutes(1));
+  EXPECT_EQ(1u, sender_->num_reports_sent());
+}
+
+TEST_F(ConversionReporterImplTest,
+       NewReportWithPreviouslySeenConversionId_Scheduled) {
+  reporter_->AddReportsToQueue({GetReport(clock().Now(), /*conversion_id=*/1)});
+  task_environment_.FastForwardBy(base::TimeDelta());
+  EXPECT_EQ(1u, sender_->num_reports_sent());
+
+  // We should schedule the new report because the previous report has been
+  // sent.
+  reporter_->AddReportsToQueue({GetReport(clock().Now(), /*conversion_id=*/1)});
+  task_environment_.FastForwardBy(base::TimeDelta());
+  EXPECT_EQ(2u, sender_->num_reports_sent());
+}
+
+TEST_F(ConversionReporterImplTest, ManyReportsAddedAtOnce_SentInOrder) {
+  std::vector<ConversionReport> reports;
+  for (int i = 1; i < 10; i++) {
+    reports.push_back(GetReport(clock().Now() + base::TimeDelta::FromMinutes(i),
+                                /*conversion_id=*/i));
+  }
+  reporter_->AddReportsToQueue(reports);
+  task_environment_.FastForwardBy(base::TimeDelta());
+  EXPECT_EQ(0u, sender_->num_reports_sent());
+
+  for (int i = 1; i < 10; i++) {
+    task_environment_.FastForwardBy(base::TimeDelta::FromMinutes(1));
+
+    EXPECT_EQ(static_cast<size_t>(i), sender_->num_reports_sent());
+    EXPECT_EQ(static_cast<int64_t>(i), sender_->last_sent_report_id());
+    EXPECT_EQ(static_cast<int64_t>(i), test_manager_.last_sent_report_id());
+  }
+}
+
+TEST_F(ConversionReporterImplTest, ManyReportsAddedSeparately_SentInOrder) {
+  for (int i = 1; i < 10; i++) {
+    reporter_->AddReportsToQueue(
+        {GetReport(clock().Now() + base::TimeDelta::FromMinutes(i),
+                   /*conversion_id=*/i)});
+  }
+  task_environment_.FastForwardBy(base::TimeDelta());
+  EXPECT_EQ(0u, sender_->num_reports_sent());
+
+  for (int i = 1; i < 10; i++) {
+    task_environment_.FastForwardBy(base::TimeDelta::FromMinutes(1));
+
+    EXPECT_EQ(static_cast<size_t>(i), sender_->num_reports_sent());
+    EXPECT_EQ(static_cast<int64_t>(i), sender_->last_sent_report_id());
+    EXPECT_EQ(static_cast<int64_t>(i), test_manager_.last_sent_report_id());
+  }
+}
+
+}  // namespace content

content/browser/conversions/conversion_storage_sql.cc

@@ -53,7 +53,7 @@ base::Time DeserializeTime(int64_t microseconds) {
 ConversionStorageSql::ConversionStorageSql(
     const base::FilePath& path_to_database_dir,
     std::unique_ptr<Delegate> delegate,
-    base::Clock* clock)
+    const base::Clock* clock)
     : path_to_database_(path_to_database_dir.Append(kDatabaseName)),
       clock_(clock),
       delegate_(std::move(delegate)),
@@ -332,7 +332,6 @@ bool ConversionStorageSql::DeleteConversion(int64_t conversion_id) {
   if (!statement.Run())
     return false;
 
-  DCHECK_EQ(1, db_.GetLastChangeCount());
   return db_.GetLastChangeCount() > 0;
 }
 

content/browser/conversions/conversion_storage_sql.h

@@ -29,7 +29,7 @@ class CONTENT_EXPORT ConversionStorageSql : public ConversionStorage {
  public:
   ConversionStorageSql(const base::FilePath& path_to_database_dir,
                        std::unique_ptr<Delegate> delegate,
-                       base::Clock* clock);
+                       const base::Clock* clock);
   ConversionStorageSql(const ConversionStorageSql& other) = delete;
   ConversionStorageSql& operator=(const ConversionStorageSql& other) = delete;
   ~ConversionStorageSql() override;
@@ -53,7 +53,7 @@ class CONTENT_EXPORT ConversionStorageSql : public ConversionStorage {
   sql::Database db_;
 
   // Must outlive |this|.
-  base::Clock* const clock_;
+  const base::Clock* clock_;
 
   std::unique_ptr<Delegate> delegate_;
 

content/browser/conversions/conversion_test_utils.cc

@@ -12,9 +12,9 @@ namespace content {
 
 namespace {
 
-const char kDefaultImpressionOrigin[] = "https:/impression.test/";
-const char kDefaultConversionOrigin[] = "https:/conversion.test/";
-const char kDefaultReportOrigin[] = "https:/report.test/";
+const char kDefaultImpressionOrigin[] = "https://impression.test/";
+const char kDefaultConversionOrigin[] = "https://conversion.test/";
+const char kDefaultReportOrigin[] = "https://report.test/";
 
 // Default expiry time for impressions for testing.
 const int64_t kExpiryTime = 30;
@@ -25,6 +25,24 @@ int EmptyStorageDelegate::GetMaxConversionsPerImpression() const {
   return 1;
 }
 
+void TestConversionManager::HandleImpression(
+    const StorableImpression& impression) {
+  num_impressions_++;
+}
+
+void TestConversionManager::HandleSentReport(int64_t conversion_id) {
+  last_sent_report_id_ = conversion_id;
+}
+
+const ConversionPolicy& TestConversionManager::GetConversionPolicy() const {
+  return policy_;
+}
+
+void TestConversionManager::Reset() {
+  num_impressions_ = 0u;
+  last_sent_report_id_ = 0UL;
+}
+
 // Builds an impression with default values. This is done as a builder because
 // all values needed to be provided at construction time.
 ImpressionBuilder::ImpressionBuilder(base::Time time)

content/browser/conversions/conversion_test_utils.h

@@ -9,6 +9,7 @@
 #include <vector>
 
 #include "base/time/time.h"
+#include "content/browser/conversions/conversion_manager.h"
 #include "content/browser/conversions/conversion_report.h"
 #include "content/browser/conversions/conversion_storage.h"
 #include "content/browser/conversions/storable_conversion.h"
@@ -30,6 +31,32 @@ class EmptyStorageDelegate : public ConversionStorage::Delegate {
   int GetMaxConversionsPerImpression() const override;
 };
 
+// Test ConversionManager which can be injected into tests to monitor calls to a
+// ConversionManager instance.
+class TestConversionManager : public ConversionManager {
+ public:
+  TestConversionManager() = default;
+  ~TestConversionManager() override = default;
+
+  // ConversionManager:
+  void HandleImpression(const StorableImpression& impression) override;
+  void HandleConversion(const StorableConversion& impression) override {}
+  void HandleSentReport(int64_t conversion_id) override;
+  const ConversionPolicy& GetConversionPolicy() const override;
+
+  // Resets all counters on this.
+  void Reset();
+
+  size_t num_impressions() const { return num_impressions_; }
+
+  int64_t last_sent_report_id() { return last_sent_report_id_; }
+
+ private:
+  ConversionPolicy policy_;
+  size_t num_impressions_ = 0;
+  int64_t last_sent_report_id_ = 0L;
+};
+
 // Helper class to construct a StorableImpression for tests using default data.
 // StorableImpression members are not mutable after construction requiring a
 // builder pattern.

content/browser/storage_partition_impl.cc

@@ -1569,8 +1569,9 @@ void StoragePartitionImpl::Initialize() {
   if (!is_in_memory_ &&
       base::FeatureList::IsEnabled(features::kConversionMeasurement)) {
     conversion_manager_ = std::make_unique<ConversionManagerImpl>(
-        path, base::ThreadPool::CreateSequencedTaskRunner(
-                  {base::MayBlock(), base::TaskPriority::BEST_EFFORT}));
+        this, path,
+        base::ThreadPool::CreateSequencedTaskRunner(
+            {base::MayBlock(), base::TaskPriority::BEST_EFFORT}));
   }
 
   GeneratedCodeCacheSettings settings =

content/test/BUILD.gn

@@ -1638,7 +1638,10 @@ test("content_unittests") {
     "../browser/content_index/content_index_database_unittest.cc",
     "../browser/content_index/content_index_service_impl_unittest.cc",
     "../browser/conversions/conversion_host_unittest.cc",
+    "../browser/conversions/conversion_manager_impl_unittest.cc",
+    "../browser/conversions/conversion_network_sender_impl_unittest.cc",
     "../browser/conversions/conversion_policy_unittest.cc",
+    "../browser/conversions/conversion_reporter_impl_unittest.cc",
     "../browser/conversions/conversion_storage_delegate_impl_unittest.cc",
     "../browser/conversions/conversion_storage_sql_unittest.cc",
     "../browser/conversions/conversion_storage_unittest.cc",

tools/traffic_annotation/summary/annotations.xml

@@ -55,6 +55,7 @@ Refer to README.md for content description and update process.
  <item id="content_hash_verification_job" hash_code="64733114" type="0" content_hash_code="127912411" os_list="linux,windows" file_path="extensions/browser/content_hash_fetcher.cc"/>
  <item id="content_resource_fetcher" hash_code="70796791" type="0" deprecated="2017-09-16" content_hash_code="135648626" file_path=""/>
  <item id="content_suggestion_get_favicon" hash_code="16653985" type="0" content_hash_code="134280933" os_list="linux,windows" file_path="components/ntp_snippets/content_suggestions_service.cc"/>
+ <item id="conversion_measurement_report" hash_code="113422320" type="0" content_hash_code="60688877" os_list="linux,windows" file_path="content/browser/conversions/conversion_network_sender_impl.cc"/>
  <item id="credenential_avatar" hash_code="53695122" type="0" content_hash_code="113035371" os_list="linux,windows" file_path="chrome/browser/ui/passwords/account_avatar_fetcher.cc"/>
  <item id="cros_recovery_image_download" hash_code="101725581" type="0" content_hash_code="23088027" os_list="linux,windows" file_path="chrome/browser/extensions/api/image_writer_private/write_from_url_operation.cc"/>
  <item id="cryptauth_device_sync_tickle" hash_code="96565489" type="1" second_id="29188932" deprecated="2018-03-15" content_hash_code="115714668" file_path=""/>