[conversions] Add report sending to internals page

Adds a button to the internals page which sends all pending reports
immediately.

This button simply sends all reports for now. A more
complicated UI was considered, which would allow individual rows to
be selected and sent. However, this requires changes to the
ConversionStorage interface, and the existing interface can be
extended if this functionality is desirable.

This change also restructures communication between
ConversionManagerImpl and ConversionReporterImpl to be callback based.
This is done to allow binding the HandleSentReport callback to a
BarrierClosure which holds the WebUI callback. This allows the WebUI
to update once all reports have finished sending.

Bug: 1057240
Change-Id: If82773b01b9df98d6d1f52cf35aee8fd6bdcf751
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2192436
Commit-Queue: John Delaney <johnidel@chromium.org>
Reviewed-by: Ken Buchanan <kenrb@chromium.org>
Reviewed-by: Charlie Harrison <csharrison@chromium.org>
Cr-Commit-Position: refs/heads/master@{#768561}

content/browser/conversions/conversion_internals.mojom

@@ -43,6 +43,11 @@ interface ConversionInternalsHandler {
   // being sent.
   GetPendingReports() => (array<WebUIConversionReport> reports);
 
+  // Sends all stored reports, ignoring delay, returning when the
+  // operation has been completed and all reports have been cleared from
+  // storage.
+  SendPendingReports() => ();
+
   // Deletes all persisted data for the Conversion API, returning when the
   // operation has been completed.
   ClearStorage() => ();

content/browser/conversions/conversion_internals_browsertest.cc

@@ -183,7 +183,7 @@ IN_PROC_BROWSER_TEST_F(ConversionInternalsWebUiBrowserTest,
   ConversionReport report(
       ImpressionBuilder(base::Time::Now()).SetData("100").Build(),
       "7" /* conversion_data */, base::Time::Now() /* report_time */,
-      base::nullopt /* conversion_id */);
+      1 /* conversion_id */);
   manager.SetReportsForWebUI({report});
   OverrideWebUIConversionManager(&manager);
 
@@ -211,7 +211,7 @@ IN_PROC_BROWSER_TEST_F(ConversionInternalsWebUiBrowserTest,
   ConversionReport report(
       ImpressionBuilder(base::Time::Now()).SetData("100").Build(),
       "7" /* conversion_data */, base::Time::Now() /* report_time */,
-      base::nullopt /* conversion_id */);
+      1 /* conversion_id */);
   manager.SetReportsForWebUI({report});
   OverrideWebUIConversionManager(&manager);
 
@@ -241,4 +241,44 @@ IN_PROC_BROWSER_TEST_F(ConversionInternalsWebUiBrowserTest,
   EXPECT_EQ(kDeleteTitle, delete_title_watcher.WaitAndGetTitle());
 }
 
+// TODO(johnidel): Use a real ConversionManager here and verify that the reports
+// are actually sent.
+IN_PROC_BROWSER_TEST_F(ConversionInternalsWebUiBrowserTest,
+                       WebUISendReports_ReportsRemoved) {
+  EXPECT_TRUE(NavigateToURL(shell(), GURL(kConversionInternalsUrl)));
+
+  TestConversionManager manager;
+  ConversionReport report(
+      ImpressionBuilder(base::Time::Now()).SetData("100").Build(),
+      "7" /* conversion_data */, base::Time::Now() /* report_time */,
+      1 /* conversion_id */);
+  manager.SetReportsForWebUI({report});
+  OverrideWebUIConversionManager(&manager);
+
+  std::string wait_script = R"(
+    let table = document.getElementById("report-table-body");
+    let obs = new MutationObserver(() => {
+      if (table.children.length === 1 &&
+          table.children[0].children[1].innerText === "0x7") {
+        document.title = $1;
+      }
+    });
+    obs.observe(table, {'childList': true});)";
+  EXPECT_TRUE(ExecJsInWebUI(JsReplace(wait_script, kCompleteTitle)));
+
+  // Wait for the table to rendered.
+  TitleWatcher title_watcher(shell()->web_contents(), kCompleteTitle);
+  ClickRefreshButton();
+  EXPECT_EQ(kCompleteTitle, title_watcher.WaitAndGetTitle());
+
+  // Click the send reports button and expect that the report table is emptied.
+  const base::string16 kSentTitle = base::ASCIIToUTF16("Sent");
+  TitleWatcher sent_title_watcher(shell()->web_contents(), kSentTitle);
+  SetTitleOnReportsTableEmpty(kSentTitle);
+
+  EXPECT_TRUE(
+      ExecJsInWebUI("document.getElementById('send-reports').click();"));
+  EXPECT_EQ(kSentTitle, sent_title_watcher.WaitAndGetTitle());
+}
+
 }  // namespace content

content/browser/conversions/conversion_internals_handler_impl.cc

@@ -100,6 +100,16 @@ void ConversionInternalsHandlerImpl::GetPendingReports(
   }
 }
 
+void ConversionInternalsHandlerImpl::SendPendingReports(
+    ::mojom::ConversionInternalsHandler::SendPendingReportsCallback callback) {
+  if (ConversionManager* manager =
+          manager_provider_->GetManager(web_ui_->GetWebContents())) {
+    manager->SendReportsForWebUI(std::move(callback));
+  } else {
+    std::move(callback).Run();
+  }
+}
+
 void ConversionInternalsHandlerImpl::ClearStorage(
     ::mojom::ConversionInternalsHandler::ClearStorageCallback callback) {
   if (ConversionManager* manager =

content/browser/conversions/conversion_internals_handler_impl.h

@@ -35,6 +35,9 @@ class ConversionInternalsHandlerImpl
   void GetPendingReports(
       ::mojom::ConversionInternalsHandler::GetPendingReportsCallback callback)
       override;
+  void SendPendingReports(
+      ::mojom::ConversionInternalsHandler::SendPendingReportsCallback callback)
+      override;
   void ClearStorage(::mojom::ConversionInternalsHandler::ClearStorageCallback
                         callback) override;
 

content/browser/conversions/conversion_manager.h

@@ -47,10 +47,6 @@ 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;
-
   // Get all impressions that are currently stored in this partition. Used for
   // populating WebUI.
   virtual void GetActiveImpressionsForWebUI(
@@ -62,6 +58,10 @@ class CONTENT_EXPORT ConversionManager {
       base::OnceCallback<void(std::vector<ConversionReport>)> callback,
       base::Time max_report_time) = 0;
 
+  // Sends all pending reports immediately, and runs |done| once they have all
+  // been sent.
+  virtual void SendReportsForWebUI(base::OnceClosure done) = 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

@@ -6,6 +6,7 @@
 
 #include <utility>
 
+#include "base/barrier_closure.h"
 #include "base/bind.h"
 #include "base/command_line.h"
 #include "base/memory/ptr_util.h"
@@ -52,7 +53,6 @@ ConversionManagerImpl::ConversionManagerImpl(
     : ConversionManagerImpl(
           std::make_unique<ConversionReporterImpl>(
               storage_partition,
-              this,
               base::DefaultClock::GetInstance()),
           std::make_unique<ConversionPolicy>(
               base::CommandLine::ForCurrentProcess()->HasSwitch(
@@ -124,13 +124,6 @@ void ConversionManagerImpl::HandleConversion(
     GetAndQueueReportsForNextInterval();
 }
 
-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));
-}
-
 void ConversionManagerImpl::GetActiveImpressionsForWebUI(
     base::OnceCallback<void(std::vector<StorableImpression>)> callback) {
   // Unretained is safe because any task to delete |storage_| will be posted
@@ -145,13 +138,14 @@ void ConversionManagerImpl::GetActiveImpressionsForWebUI(
 void ConversionManagerImpl::GetReportsForWebUI(
     base::OnceCallback<void(std::vector<ConversionReport>)> callback,
     base::Time max_report_time) {
-  // Unretained is safe because any task to delete |storage_| will be posted
-  // after this one because |storage_| uses base::OnTaskRunnerDeleter.
-  base::PostTaskAndReplyWithResult(
-      storage_task_runner_.get(), FROM_HERE,
-      base::BindOnce(&ConversionStorage::GetConversionsToReport,
-                     base::Unretained(storage_.get()), max_report_time),
-      std::move(callback));
+  GetAndHandleReports(std::move(callback), max_report_time);
+}
+
+void ConversionManagerImpl::SendReportsForWebUI(base::OnceClosure done) {
+  GetAndHandleReports(
+      base::BindOnce(&ConversionManagerImpl::HandleReportsSentFromWebUI,
+                     weak_factory_.GetWeakPtr(), std::move(done)),
+      base::Time::Max());
 }
 
 const ConversionPolicy& ConversionManagerImpl::GetConversionPolicy() const {
@@ -181,7 +175,8 @@ void ConversionManagerImpl::OnInitCompleted(bool success) {
   // Chrome was not running and handle these specially.
   GetAndHandleReports(
       base::BindOnce(&ConversionManagerImpl::HandleReportsExpiredAtStartup,
-                     weak_factory_.GetWeakPtr()));
+                     weak_factory_.GetWeakPtr()),
+      clock_->Now() + kConversionManagerQueueReportsInterval);
 
   // Start a repeating timer that will fetch reports once every
   // |kConversionManagerQueueReportsInterval| and add them to |reporter_|.
@@ -191,12 +186,12 @@ void ConversionManagerImpl::OnInitCompleted(bool success) {
 }
 
 void ConversionManagerImpl::GetAndHandleReports(
-    ReportsHandlerFunc handler_function) {
+    ReportsHandlerFunc handler_function,
+    base::Time max_report_time) {
   base::PostTaskAndReplyWithResult(
       storage_task_runner_.get(), FROM_HERE,
       base::BindOnce(&ConversionStorage::GetConversionsToReport,
-                     base::Unretained(storage_.get()),
-                     clock_->Now() + kConversionManagerQueueReportsInterval),
+                     base::Unretained(storage_.get()), max_report_time),
       std::move(handler_function));
 }
 
@@ -204,13 +199,20 @@ 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()));
+                                     weak_factory_.GetWeakPtr()),
+                      clock_->Now() + kConversionManagerQueueReportsInterval);
 }
 
 void ConversionManagerImpl::QueueReports(
     std::vector<ConversionReport> reports) {
-  if (!reports.empty())
-    reporter_->AddReportsToQueue(std::move(reports));
+  if (!reports.empty()) {
+    // |reporter_| is owned by |this|, so base::Unretained() is safe as the
+    // reporter and callbacks will be deleted first.
+    reporter_->AddReportsToQueue(
+        std::move(reports),
+        base::BindRepeating(&ConversionManagerImpl::OnReportSent,
+                            base::Unretained(this)));
+  }
 }
 
 void ConversionManagerImpl::HandleReportsExpiredAtStartup(
@@ -232,4 +234,49 @@ void ConversionManagerImpl::HandleReportsExpiredAtStartup(
   QueueReports(std::move(reports));
 }
 
+void ConversionManagerImpl::HandleReportsSentFromWebUI(
+    base::OnceClosure done,
+    std::vector<ConversionReport> reports) {
+  if (reports.empty()) {
+    std::move(done).Run();
+    return;
+  }
+
+  // All reports should be sent immediately.
+  for (ConversionReport& report : reports) {
+    report.report_time = base::Time::Min();
+  }
+
+  // Wraps |done| so that is will run once all of the reports have finished
+  // sending.
+  base::RepeatingClosure all_reports_sent =
+      base::BarrierClosure(reports.size(), std::move(done));
+
+  // |reporter_| is owned by |this|, so base::Unretained() is safe as the
+  // reporter and callbacks will be deleted first.
+  reporter_->AddReportsToQueue(
+      std::move(reports),
+      base::BindRepeating(&ConversionManagerImpl::OnReportSentFromWebUI,
+                          base::Unretained(this), std::move(all_reports_sent)));
+}
+
+void ConversionManagerImpl::OnReportSent(int64_t conversion_id) {
+  storage_task_runner_->PostTask(
+      FROM_HERE,
+      base::BindOnce(base::IgnoreResult(&ConversionStorage::DeleteConversion),
+                     base::Unretained(storage_.get()), conversion_id));
+}
+
+void ConversionManagerImpl::OnReportSentFromWebUI(
+    base::OnceClosure reports_sent_barrier,
+    int64_t conversion_id) {
+  // |reports_sent_barrier| is a OnceClosure view of a RepeatingClosure obtained
+  // by base::BarrierClosure().
+  storage_task_runner_->PostTaskAndReply(
+      FROM_HERE,
+      base::BindOnce(base::IgnoreResult(&ConversionStorage::DeleteConversion),
+                     base::Unretained(storage_.get()), conversion_id),
+      std::move(reports_sent_barrier));
+}
+
 }  // namespace content

content/browser/conversions/conversion_manager_impl.h

@@ -56,10 +56,12 @@ class CONTENT_EXPORT ConversionManagerImpl : public ConversionManager {
    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;
+    // Adds |reports| to a shared queue of reports that need to be sent. Runs
+    // |report_sent_callback| for every report that is sent, with the associated
+    // |conversion_id| of the report.
+    virtual void AddReportsToQueue(
+        std::vector<ConversionReport> reports,
+        base::RepeatingCallback<void(int64_t)> report_sent_callback) = 0;
   };
 
   static std::unique_ptr<ConversionManagerImpl> CreateForTesting(
@@ -81,13 +83,13 @@ class CONTENT_EXPORT ConversionManagerImpl : public ConversionManager {
   // ConversionManager:
   void HandleImpression(const StorableImpression& impression) override;
   void HandleConversion(const StorableConversion& conversion) override;
-  void HandleSentReport(int64_t conversion_id) override;
   void GetActiveImpressionsForWebUI(
       base::OnceCallback<void(std::vector<StorableImpression>)> callback)
       override;
   void GetReportsForWebUI(
       base::OnceCallback<void(std::vector<ConversionReport>)> callback,
       base::Time max_report_time) override;
+  void SendReportsForWebUI(base::OnceClosure done) override;
   const ConversionPolicy& GetConversionPolicy() const override;
   void ClearData(base::Time delete_begin,
                  base::Time delete_end,
@@ -104,9 +106,12 @@ class CONTENT_EXPORT ConversionManagerImpl : public ConversionManager {
 
   void OnInitCompleted(bool success);
 
+  // Retrieves reports from storage whose |report_time| <= |max_report_time|,
+  // and calls |handler_function| on them.
   using ReportsHandlerFunc =
       base::OnceCallback<void(std::vector<ConversionReport>)>;
-  void GetAndHandleReports(ReportsHandlerFunc handler_function);
+  void GetAndHandleReports(ReportsHandlerFunc handler_function,
+                           base::Time max_report_time);
 
   // 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
@@ -118,6 +123,18 @@ class CONTENT_EXPORT ConversionManagerImpl : public ConversionManager {
 
   void HandleReportsExpiredAtStartup(std::vector<ConversionReport> reports);
 
+  void HandleReportsSentFromWebUI(base::OnceClosure done,
+                                  std::vector<ConversionReport> reports);
+
+  // Notify storage to delete the given |conversion_id| when it's associated
+  // report has been sent.
+  void OnReportSent(int64_t conversion_id);
+
+  // Similar to OnReportSent, but invokes |reports_sent_barrier| when the
+  // report has been removed from storage.
+  void OnReportSentFromWebUI(base::OnceClosure reports_sent_barrier,
+                             int64_t conversion_id);
+
   // Friend to expose the ConversionStorage and task runner, consider changing
   // to just expose the storage if it moves to SequenceBound.
   friend std::vector<ConversionReport> GetConversionsToReportForTesting(

content/browser/conversions/conversion_manager_impl_unittest.cc

@@ -51,15 +51,27 @@ class TestConversionReporter
   ~TestConversionReporter() override = default;
 
   // ConversionManagerImpl::ConversionReporter
-  void AddReportsToQueue(std::vector<ConversionReport> reports) override {
+  void AddReportsToQueue(
+      std::vector<ConversionReport> reports,
+      base::RepeatingCallback<void(int64_t)> report_sent_callback) override {
     num_reports_ += reports.size();
     last_conversion_id_ = *reports.back().conversion_id;
     last_report_time_ = reports.back().report_time;
 
+    if (should_run_report_sent_callbacks_) {
+      for (const auto& report : reports) {
+        report_sent_callback.Run(*report.conversion_id);
+      }
+    }
+
     if (quit_closure_ && num_reports_ >= expected_num_reports_)
       std::move(quit_closure_).Run();
   }
 
+  void ShouldRunReportSentCallbacks(bool should_run_report_sent_callbacks) {
+    should_run_report_sent_callbacks_ = should_run_report_sent_callbacks;
+  }
+
   size_t num_reports() { return num_reports_; }
 
   int64_t last_conversion_id() { return last_conversion_id_; }
@@ -77,6 +89,7 @@ class TestConversionReporter
   }
 
  private:
+  bool should_run_report_sent_callbacks_ = false;
   size_t expected_num_reports_ = 0u;
   size_t num_reports_ = 0u;
   int64_t last_conversion_id_ = 0UL;
@@ -213,6 +226,7 @@ TEST_F(ConversionManagerImplTest, QueuedReportNotSent_QueuedAgain) {
 }
 
 TEST_F(ConversionManagerImplTest, QueuedReportSent_NotQueuedAgain) {
+  test_reporter_->ShouldRunReportSentCallbacks(true);
   conversion_manager_->HandleImpression(
       ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
   conversion_manager_->HandleConversion(DefaultConversion());
@@ -220,9 +234,6 @@ TEST_F(ConversionManagerImplTest, QueuedReportSent_NotQueuedAgain) {
                                   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());
@@ -256,9 +267,9 @@ TEST_F(ConversionManagerImplTest, ExpiredReportsAtStartup_Queued) {
 
   // Create the manager and check that the first report is queued immediately.
   CreateManager();
+  test_reporter_->ShouldRunReportSentCallbacks(true);
   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);
@@ -289,6 +300,19 @@ TEST_F(ConversionManagerImplTest, ClearData) {
   }
 }
 
+TEST_F(ConversionManagerImplTest, ConversionsSentFromUI_ReportedImmediately) {
+  conversion_manager_->HandleImpression(
+      ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
+  conversion_manager_->HandleImpression(
+      ImpressionBuilder(clock().Now()).SetExpiry(kImpressionExpiry).Build());
+  conversion_manager_->HandleConversion(DefaultConversion());
+  EXPECT_EQ(0u, test_reporter_->num_reports());
+
+  conversion_manager_->SendReportsForWebUI(base::DoNothing());
+  task_environment_.FastForwardBy(base::TimeDelta::FromMinutes(0));
+  EXPECT_EQ(2u, test_reporter_->num_reports());
+}
+
 TEST_F(ConversionManagerImplTest, ExpiredReportsAtStartup_Delayed) {
   // Create a report that will be reported at t= 2 days.
   base::Time start_time = clock().Now();

content/browser/conversions/conversion_network_sender_impl.cc

@@ -44,6 +44,11 @@ enum class Status {
 // Called when a network request is started for |report|, for logging metrics.
 void LogMetricsOnReportSend(ConversionReport* report) {
   DCHECK(report);
+
+  // Reports sent from the WebUI should not log metrics.
+  if (report->report_time == base::Time::Min())
+    return;
+
   // Use a large time range to capture users that might not open the browser for
   // a long time while a conversion report is pending. Revisit this range if it
   // is non-ideal for real world data.

content/browser/conversions/conversion_reporter_impl.cc

@@ -15,19 +15,16 @@ namespace content {
 
 ConversionReporterImpl::ConversionReporterImpl(
     StoragePartition* storage_partition,
-    ConversionManager* conversion_manager,
     const base::Clock* clock)
-    : conversion_manager_(conversion_manager),
-      clock_(clock),
+    : clock_(clock),
       network_sender_(
-          std::make_unique<ConversionNetworkSenderImpl>(storage_partition)) {
-  DCHECK(conversion_manager_);
-}
+          std::make_unique<ConversionNetworkSenderImpl>(storage_partition)) {}
 
 ConversionReporterImpl::~ConversionReporterImpl() = default;
 
 void ConversionReporterImpl::AddReportsToQueue(
-    std::vector<ConversionReport> reports) {
+    std::vector<ConversionReport> reports,
+    base::RepeatingCallback<void(int64_t)> report_sent_callback) {
   DCHECK(!reports.empty());
 
   std::vector<std::unique_ptr<ConversionReport>> swappable_reports;
@@ -45,8 +42,9 @@ void ConversionReporterImpl::AddReportsToQueue(
 
   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;
+    bool inserted = conversion_report_callbacks_
+                        .emplace(*(report->conversion_id), report_sent_callback)
+                        .second;
     if (inserted)
       report_queue_.push(std::move(report));
   }
@@ -84,14 +82,18 @@ void ConversionReporterImpl::MaybeScheduleNextReport() {
   // 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,
+      FROM_HERE,
+      (report_time < current_time) ? base::TimeDelta()
+                                   : 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);
+  auto it = conversion_report_callbacks_.find(conversion_id);
+  DCHECK(it != conversion_report_callbacks_.end());
+  std::move(it->second).Run(conversion_id);
+  conversion_report_callbacks_.erase(it);
 }
 
 bool ConversionReporterImpl::ReportComparator::operator()(

content/browser/conversions/conversion_reporter_impl.h

@@ -10,7 +10,8 @@
 #include <queue>
 #include <vector>
 
-#include "base/containers/flat_set.h"
+#include "base/callback.h"
+#include "base/containers/flat_map.h"
 #include "base/time/time.h"
 #include "base/timer/timer.h"
 #include "content/browser/conversions/conversion_manager_impl.h"
@@ -24,7 +25,6 @@ class Clock;
 
 namespace content {
 
-class ConversionManager;
 class StoragePartition;
 
 // This class is responsible for managing the dispatch of conversion reports to
@@ -52,14 +52,15 @@ class CONTENT_EXPORT ConversionReporterImpl
   };
 
   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 AddReportsToQueue(
+      std::vector<ConversionReport> reports,
+      base::RepeatingCallback<void(int64_t)> report_sent_callback) override;
 
   void SetNetworkSenderForTesting(
       std::unique_ptr<NetworkSender> network_sender);
@@ -86,13 +87,12 @@ class CONTENT_EXPORT ConversionReporterImpl
                       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_;
+  // Map of all conversion ids that are currently in |report_queue| or are being
+  // sent by |network_sender_|, and their associated report sent callbacks. The
+  // number of concurrent conversion reports being sent at any time is expected
+  // to be small, so a flat_map is used.
+  base::flat_map<int64_t, base::OnceCallback<void(int64_t)>>
+      conversion_report_callbacks_;
 
   const base::Clock* clock_;
 

content/browser/conversions/conversion_reporter_impl_unittest.cc

@@ -4,9 +4,13 @@
 
 #include "content/browser/conversions/conversion_reporter_impl.h"
 
+#include <stdint.h>
+
+#include "base/bind.h"
 #include "base/sequenced_task_runner.h"
 #include "base/strings/strcat.h"
 #include "base/task/post_task.h"
+#include "base/test/bind_test_util.h"
 #include "base/test/simple_test_clock.h"
 #include "content/browser/conversions/conversion_manager.h"
 #include "content/browser/conversions/conversion_test_utils.h"
@@ -60,7 +64,6 @@ class ConversionReporterImplTest : public testing::Test {
         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();
@@ -74,32 +77,35 @@ class ConversionReporterImplTest : public testing::Test {
   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)});
+  reporter_->AddReportsToQueue({GetReport(clock().Now(), /*conversion_id=*/1)},
+                               base::BindRepeating([](int64_t conversion_id) {
+                                 EXPECT_EQ(1L, conversion_id);
+                               }));
 
   // 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)});
+  reporter_->AddReportsToQueue(
+      {GetReport(clock().Now() - base::TimeDelta::FromHours(10),
+                 /*conversion_id=*/1)},
+      base::BindRepeating(
+          [](int64_t conversion_id) { EXPECT_EQ(1L, conversion_id); }));
 
   // 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,
@@ -107,7 +113,8 @@ TEST_F(ConversionReporterImplTest,
   const base::TimeDelta delay = base::TimeDelta::FromMinutes(30);
 
   reporter_->AddReportsToQueue(
-      {GetReport(clock().Now() + delay, /*conversion_id=*/1)});
+      {GetReport(clock().Now() + delay, /*conversion_id=*/1)},
+      base::DoNothing());
   task_environment_.FastForwardBy(base::TimeDelta());
   EXPECT_EQ(0u, sender_->num_reports_sent());
 
@@ -119,36 +126,46 @@ TEST_F(ConversionReporterImplTest,
 }
 
 TEST_F(ConversionReporterImplTest, DuplicateReportScheduled_Ignored) {
-  reporter_->AddReportsToQueue({GetReport(
-      clock().Now() + base::TimeDelta::FromMinutes(1), /*conversion_id=*/1)});
+  reporter_->AddReportsToQueue(
+      {GetReport(clock().Now() + base::TimeDelta::FromMinutes(1),
+                 /*conversion_id=*/1)},
+      base::DoNothing());
 
   // A duplicate report should not be scheduled.
-  reporter_->AddReportsToQueue({GetReport(
-      clock().Now() + base::TimeDelta::FromMinutes(1), /*conversion_id=*/1)});
+  reporter_->AddReportsToQueue(
+      {GetReport(clock().Now() + base::TimeDelta::FromMinutes(1),
+                 /*conversion_id=*/1)},
+      base::DoNothing());
   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)});
+  reporter_->AddReportsToQueue({GetReport(clock().Now(), /*conversion_id=*/1)},
+                               base::DoNothing());
   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)});
+  reporter_->AddReportsToQueue({GetReport(clock().Now(), /*conversion_id=*/1)},
+                               base::DoNothing());
   task_environment_.FastForwardBy(base::TimeDelta());
   EXPECT_EQ(2u, sender_->num_reports_sent());
 }
 
 TEST_F(ConversionReporterImplTest, ManyReportsAddedAtOnce_SentInOrder) {
   std::vector<ConversionReport> reports;
+  int64_t last_report_id = 0UL;
   for (int i = 1; i < 10; i++) {
     reports.push_back(GetReport(clock().Now() + base::TimeDelta::FromMinutes(i),
                                 /*conversion_id=*/i));
   }
-  reporter_->AddReportsToQueue(reports);
+  reporter_->AddReportsToQueue(
+      reports, base::BindLambdaForTesting([&](int64_t conversion_id) {
+        last_report_id = conversion_id;
+      }));
   task_environment_.FastForwardBy(base::TimeDelta());
   EXPECT_EQ(0u, sender_->num_reports_sent());
 
@@ -157,15 +174,19 @@ TEST_F(ConversionReporterImplTest, ManyReportsAddedAtOnce_SentInOrder) {
 
     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());
+    EXPECT_EQ(static_cast<int64_t>(i), last_report_id);
   }
 }
 
 TEST_F(ConversionReporterImplTest, ManyReportsAddedSeparately_SentInOrder) {
+  int64_t last_report_id = 0;
+  auto report_sent_callback = base::BindLambdaForTesting(
+      [&](int64_t conversion_id) { last_report_id = conversion_id; });
   for (int i = 1; i < 10; i++) {
     reporter_->AddReportsToQueue(
         {GetReport(clock().Now() + base::TimeDelta::FromMinutes(i),
-                   /*conversion_id=*/i)});
+                   /*conversion_id=*/i)},
+        report_sent_callback);
   }
   task_environment_.FastForwardBy(base::TimeDelta());
   EXPECT_EQ(0u, sender_->num_reports_sent());
@@ -175,7 +196,7 @@ TEST_F(ConversionReporterImplTest, ManyReportsAddedSeparately_SentInOrder) {
 
     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());
+    EXPECT_EQ(static_cast<int64_t>(i), last_report_id);
   }
 }
 

content/browser/conversions/conversion_test_utils.cc

@@ -52,10 +52,6 @@ void TestConversionManager::HandleConversion(
   num_conversions_++;
 }
 
-void TestConversionManager::HandleSentReport(int64_t conversion_id) {
-  last_sent_report_id_ = conversion_id;
-}
-
 void TestConversionManager::GetActiveImpressionsForWebUI(
     base::OnceCallback<void(std::vector<StorableImpression>)> callback) {
   std::move(callback).Run(impressions_);
@@ -67,6 +63,11 @@ void TestConversionManager::GetReportsForWebUI(
   std::move(callback).Run(reports_);
 }
 
+void TestConversionManager::SendReportsForWebUI(base::OnceClosure done) {
+  reports_.clear();
+  std::move(done).Run();
+}
+
 const ConversionPolicy& TestConversionManager::GetConversionPolicy() const {
   return policy_;
 }
@@ -94,7 +95,6 @@ void TestConversionManager::SetReportsForWebUI(
 void TestConversionManager::Reset() {
   num_impressions_ = 0u;
   num_conversions_ = 0u;
-  last_sent_report_id_ = 0UL;
 }
 
 // Builds an impression with default values. This is done as a builder because

content/browser/conversions/conversion_test_utils.h

@@ -57,13 +57,13 @@ class TestConversionManager : public ConversionManager {
   // ConversionManager:
   void HandleImpression(const StorableImpression& impression) override;
   void HandleConversion(const StorableConversion& conversion) override;
-  void HandleSentReport(int64_t conversion_id) override;
   void GetActiveImpressionsForWebUI(
       base::OnceCallback<void(std::vector<StorableImpression>)> callback)
       override;
   void GetReportsForWebUI(
       base::OnceCallback<void(std::vector<ConversionReport>)> callback,
       base::Time max_report_time) override;
+  void SendReportsForWebUI(base::OnceClosure done) override;
   const ConversionPolicy& GetConversionPolicy() const override;
   void ClearData(base::Time delete_begin,
                  base::Time delete_end,
@@ -80,13 +80,10 @@ class TestConversionManager : public ConversionManager {
   size_t num_impressions() const { return num_impressions_; }
   size_t num_conversions() const { return num_conversions_; }
 
-  int64_t last_sent_report_id() { return last_sent_report_id_; }
-
  private:
   ConversionPolicy policy_;
   size_t num_impressions_ = 0;
   size_t num_conversions_ = 0;
-  int64_t last_sent_report_id_ = 0L;
 
   std::vector<StorableImpression> impressions_;
   std::vector<ConversionReport> reports_;

content/browser/resources/conversions/conversion_internals.html

@@ -104,6 +104,9 @@
         </tbody>
       </table>
     </div>
+    <div>
+      <button id="send-reports">Send All Reports</button>
+    </div>
   </div>
 </div>
 

content/browser/resources/conversions/conversion_internals.js

@@ -155,12 +155,30 @@ function clearStorage() {
   });
 }
 
+/**
+ * Sends all conversion reports, and updates the page once they are sent.
+ * Disables the button while the reports are still being sent.
+ */
+function sendReports() {
+  const button = $('send-reports');
+  const previousText = $('send-reports').innerText;
+
+  button.disabled = true;
+  button.innerText = 'Sending...';
+  pageHandler.sendPendingReports().then(() => {
+    updatePageData();
+    button.disabled = false;
+    button.innerText = previousText;
+  });
+}
+
 document.addEventListener('DOMContentLoaded', function() {
   // Setup the mojo interface.
   pageHandler = mojom.ConversionInternalsHandler.getRemote();
 
   $('refresh').addEventListener('click', updatePageData);
   $('clear-data').addEventListener('click', clearStorage);
+  $('send-reports').addEventListener('click', sendReports);
 
   // Automatically refresh every 2 minutes.
   setInterval(updatePageData, 2 * 60 * 1000);