components/reporting/client/report_queue_impl.cc

// Copyright 2020 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "components/reporting/client/report_queue_impl.h"

#include <memory>
#include <optional>
#include <queue>
#include <string>
#include <utility>

#include "base/check.h"
#include "base/check_op.h"
#include "base/functional/bind.h"
#include "base/functional/callback.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/memory/ptr_util.h"
#include "base/memory/scoped_refptr.h"
#include "base/memory/weak_ptr.h"
#include "base/metrics/histogram_functions.h"
#include "base/notreached.h"
#include "base/sequence_checker.h"
#include "base/strings/strcat.h"
#include "base/strings/string_number_conversions.h"
#include "base/task/bind_post_task.h"
#include "base/task/sequenced_task_runner.h"
#include "base/task/task_traits.h"
#include "base/task/thread_pool.h"
#include "base/time/time.h"
#include "base/types/expected.h"
#include "components/reporting/client/report_queue_configuration.h"
#include "components/reporting/proto/synced/metric_data.pb.h"
#include "components/reporting/proto/synced/record.pb.h"
#include "components/reporting/proto/synced/record_constants.pb.h"
#include "components/reporting/storage/storage_module_interface.h"
#include "components/reporting/util/reporting_errors.h"
#include "components/reporting/util/status.h"
#include "components/reporting/util/statusor.h"

namespace reporting {
namespace {

// UTC time of 2122-01-01T00:00:00Z since Unix epoch 1970-01-01T00:00:00Z in
// microseconds.
constexpr int64_t kTime2122 = 4'796'668'800'000'000;

// Returns true if record is allowed to go to `destination`. Returns false
// otherwise.
static bool RecordMayGoToDestination(const std::string& record_data,
                                     Destination destination) {
  // All records sent to destination *_METRIC must be MetricData
  // protos due to the way the server is implemented.
  if (destination == Destination::EVENT_METRIC ||
      destination == Destination::TELEMETRY_METRIC ||
      destination == Destination::INFO_METRIC) {
    MetricData metric_data;
    const bool is_metric_data =
        metric_data.ParseFromString(record_data) &&
        (metric_data.has_event_data() || metric_data.has_telemetry_data() ||
         metric_data.has_info_data());
    LOG_IF(ERROR, !is_metric_data)
        << "Only MetricData records may be enqueued with destinations: "
           "EVENT_METRIC, TELEMETRY_METRIC, or INFO_METRIC";
    return is_metric_data;
  }
  return true;
}

StatusOr<Record> ProduceRecord(std::string dm_token,
                               Destination destination,
                               int64_t reserved_space,
                               std::optional<SourceInfo> source_info,
                               ReportQueue::RecordProducer record_producer) {
  // Generate record data.
  auto record_result = std::move(record_producer).Run();
  if (!record_result.has_value()) {
    return base::unexpected(record_result.error());
  }

  CHECK(RecordMayGoToDestination(record_result.value(), destination));

  // Augment data.
  Record record;
  *record.mutable_data() = std::move(record_result.value());
  record.set_destination(destination);
  if (reserved_space > 0L) {
    record.set_reserved_space(reserved_space);
  }

  // Additional record augmentation for keeping local record copy.
  // Note: that must be done before calling `storage->AddRecord` below,
  // because later the handler might call it with no need to set this flag.
  switch (destination) {
    case LOG_UPLOAD:
      // It would be better to base the decision on `upload_settings` presence
      // in the event, but that would require protobuf reflecion, that is not
      // included in Chromium build. So instead we just use `destination`.
      record.set_needs_local_unencrypted_copy(true);
      break;
    default:  // Do nothing.
      break;
  }

  // |record| with no DM token is assumed to be associated with device DM
  // token.
  if (!dm_token.empty()) {
    *record.mutable_dm_token() = std::move(dm_token);
  }

  // Augment source info if available.
  if (source_info.has_value()) {
    *record.mutable_source_info() = std::move(source_info.value());
  }

  // Calculate timestamp in microseconds - to match Spanner expectations.
  const int64_t time_since_epoch_us =
      base::Time::Now().InMillisecondsSinceUnixEpoch() *
      base::Time::kMicrosecondsPerMillisecond;
  if (time_since_epoch_us > kTime2122) {
    // Unusual timestamp. Reject the record even though the record is good
    // otherwise, because we can't obtain a reasonable timestamp. We have this
    // code block here because server very occasionally detects very large
    // timestamps. The reason could come from occasional irregular system
    // time. Filtering out irregular timestamps here should address the
    // problem without leaving timestamp-related bugs in the ERP undiscovered
    // (should there be any).
    base::UmaHistogramBoolean("Browser.ERP.UnusualEnqueueTimestamp", true);
    return base::unexpected(Status(
        error::FAILED_PRECONDITION,
        base::StrCat(
            {"Abnormal system timestamp obtained. Microseconds since epoch: ",
             base::NumberToString(time_since_epoch_us)})));
  }
  record.set_timestamp_us(time_since_epoch_us);
  return std::move(record);
}

// Calls |record_producer|, checks the result and in case of success, forwards
// it to the storage. In production code should be invoked asynchronously, on
// a thread pool (no synchronization expected).
void AddRecordToStorage(scoped_refptr<StorageModuleInterface> storage,
                        Priority priority,
                        WrappedRateLimiter::AsyncAcquireCb acquire_cb,
                        std::string dm_token,
                        Destination destination,
                        int64_t reserved_space,
                        std::optional<SourceInfo> source_info,
                        ReportQueue::RecordProducer record_producer,
                        StorageModuleInterface::EnqueueCallback callback) {
  auto record_result =
      ProduceRecord(dm_token, destination, reserved_space, source_info,
                    std::move(record_producer));

  if (!record_result.has_value()) {
    std::move(callback).Run(record_result.error());
    return;
  }

  const auto record_size = record_result.value().ByteSizeLong();

  // Prepare `Storage::AddRecord` as a callback.
  auto add_record_cb =
      base::BindOnce(&StorageModuleInterface::AddRecord, storage, priority,
                     std::move(record_result.value()));

  // Rate-limit event, if required.
  if (!acquire_cb) {
    // No rate limiter, just add resulting Record to the storage.
    std::move(add_record_cb).Run(std::move(callback));
    return;
  }

  // Add Record only if rate limiter approves.
  acquire_cb.Run(
      record_size,
      base::BindOnce(
          [](size_t record_size,
             base::OnceCallback<void(StorageModuleInterface::EnqueueCallback
                                         callback)> add_record_cb,
             StorageModuleInterface::EnqueueCallback callback, bool acquired) {
            if (!acquired) {
              std::move(callback).Run(
                  Status(error::OUT_OF_RANGE,
                         base::StrCat({"Event size ",
                                       base::NumberToString(record_size),
                                       " rejected by rate limiter"})));
              return;
            }
            // Add resulting Record to the storage.
            std::move(add_record_cb).Run(std::move(callback));
          },
          record_size, std::move(add_record_cb), std::move(callback)));
}
}  // namespace

void ReportQueueImpl::Create(
    std::unique_ptr<ReportQueueConfiguration> config,
    scoped_refptr<StorageModuleInterface> storage,
    base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)> cb) {
  std::move(cb).Run(base::WrapUnique<ReportQueueImpl>(
      new ReportQueueImpl(std::move(config), storage)));
}

ReportQueueImpl::ReportQueueImpl(
    std::unique_ptr<ReportQueueConfiguration> config,
    scoped_refptr<StorageModuleInterface> storage)
    : config_(std::move(config)), storage_(storage) {
  CHECK(config_);
}

ReportQueueImpl::~ReportQueueImpl() = default;

void ReportQueueImpl::AddProducedRecord(RecordProducer record_producer,
                                        Priority priority,
                                        EnqueueCallback callback) const {
  const Status status = config_->CheckPolicy();
  if (!status.ok()) {
    std::move(callback).Run(status);
    return;
  }

  if (priority == Priority::UNDEFINED_PRIORITY) {
    std::move(callback).Run(
        Status(error::INVALID_ARGUMENT, "Priority must be defined"));
    return;
  }

  // Execute |record_producer| on arbitrary thread, analyze the result and send
  // it to the Storage, returning with the callback.
  base::ThreadPool::PostTask(
      FROM_HERE, {base::TaskPriority::BEST_EFFORT},
      base::BindOnce(&AddRecordToStorage, storage_, priority,
                     config_->is_event_allowed_cb(), config_->dm_token(),
                     config_->destination(), config_->reserved_space(),
                     config_->source_info(), std::move(record_producer),
                     // EnqueueCallback must be run on the current thread, we
                     // need to bind to make sure it's posted to correct thread
                     // from the ThreadPool.
                     base::BindPostTaskToCurrentDefault(std::move(callback))));
}

void ReportQueueImpl::Flush(Priority priority, FlushCallback callback) {
  storage_->Flush(priority, std::move(callback));
}

base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
ReportQueueImpl::PrepareToAttachActualQueue() const {
  NOTREACHED();
}

Destination ReportQueueImpl::GetDestination() const {
  return config_->destination();
}

// Implementation of SpeculativeReportQueueImpl::PendingRecordProducer

SpeculativeReportQueueImpl::PendingRecordProducer::PendingRecordProducer(
    RecordProducer producer,
    EnqueueCallback callback,
    Priority priority)
    : record_producer(std::move(producer)),
      record_callback(std::move(callback)),
      record_priority(priority) {}

SpeculativeReportQueueImpl::PendingRecordProducer::PendingRecordProducer(
    PendingRecordProducer&& other)
    : record_producer(std::move(other.record_producer)),
      record_callback(std::move(other.record_callback)),
      record_priority(other.record_priority) {}

SpeculativeReportQueueImpl::PendingRecordProducer::~PendingRecordProducer() =
    default;

SpeculativeReportQueueImpl::PendingRecordProducer&
SpeculativeReportQueueImpl::PendingRecordProducer::operator=(
    PendingRecordProducer&& other) {
  record_producer = std::move(other.record_producer);
  record_callback = std::move(other.record_callback);
  record_priority = other.record_priority;
  return *this;
}

// static
std::unique_ptr<SpeculativeReportQueueImpl, base::OnTaskRunnerDeleter>
SpeculativeReportQueueImpl::Create(
    const SpeculativeConfigSettings& config_settings) {
  scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner =
      base::ThreadPool::CreateSequencedTaskRunner(
          {base::TaskPriority::BEST_EFFORT, base::MayBlock()});
  return std::unique_ptr<SpeculativeReportQueueImpl, base::OnTaskRunnerDeleter>(
      new SpeculativeReportQueueImpl(config_settings, sequenced_task_runner),
      base::OnTaskRunnerDeleter(sequenced_task_runner));
}

SpeculativeReportQueueImpl::SpeculativeReportQueueImpl(
    const SpeculativeConfigSettings& config_settings,
    scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner)
    : sequenced_task_runner_(sequenced_task_runner),
      config_settings_(config_settings) {
  DETACH_FROM_SEQUENCE(sequence_checker_);
}

SpeculativeReportQueueImpl::~SpeculativeReportQueueImpl() {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  PurgePendingProducers(
      Status(error::DATA_LOSS, "The queue is being destructed"));
}

void SpeculativeReportQueueImpl::Flush(Priority priority,
                                       FlushCallback callback) {
  sequenced_task_runner_->PostTask(
      FROM_HERE,
      base::BindOnce(
          [](Priority priority, FlushCallback callback,
             base::WeakPtr<SpeculativeReportQueueImpl> self) {
            if (!self) {
              std::move(callback).Run(
                  Status(error::UNAVAILABLE, "Queue has been destructed"));
              base::UmaHistogramEnumeration(
                  reporting::kUmaUnavailableErrorReason,
                  UnavailableErrorReason::REPORT_QUEUE_DESTRUCTED,
                  UnavailableErrorReason::MAX_VALUE);
              return;
            }
            DCHECK_CALLED_ON_VALID_SEQUENCE(self->sequence_checker_);
            if (!self->actual_report_queue_.has_value()) {
              std::move(callback).Run(Status(error::FAILED_PRECONDITION,
                                             "ReportQueue is not ready yet."));
              return;
            }
            const std::unique_ptr<ReportQueue>& report_queue =
                self->actual_report_queue_.value();
            report_queue->Flush(priority, std::move(callback));
          },
          priority, std::move(callback), weak_ptr_factory_.GetWeakPtr()));
}

void SpeculativeReportQueueImpl::AddProducedRecord(
    RecordProducer record_producer,
    Priority priority,
    EnqueueCallback callback) const {
  // Invoke producer on a thread pool, then enqueue record on sequenced task
  // runner.
  sequenced_task_runner_->PostTask(
      FROM_HERE,
      base::BindOnce(&SpeculativeReportQueueImpl::MaybeEnqueueRecordProducer,
                     weak_ptr_factory_.GetWeakPtr(), priority,
                     base::BindPostTaskToCurrentDefault(std::move(callback)),
                     std::move(record_producer)));
}

void SpeculativeReportQueueImpl::MaybeEnqueueRecordProducer(
    Priority priority,
    EnqueueCallback callback,
    RecordProducer record_producer) const {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  if (!actual_report_queue_.has_value()) {
    // Queue is not ready yet, store the record in the memory queue.
    pending_record_producers_.emplace(std::move(record_producer),
                                      std::move(callback), priority);
    return;
  }
  // Queue is ready. If memory queue is empty, just forward the record.
  if (pending_record_producers_.empty()) {
    const std::unique_ptr<ReportQueue>& report_queue =
        actual_report_queue_.value();
    report_queue->AddProducedRecord(std::move(record_producer), priority,
                                    std::move(callback));
    return;
  }
  // If memory queue is not empty, attach the new record at the
  // end and initiate enqueuing of everything from there.
  pending_record_producers_.emplace(std::move(record_producer),
                                    std::move(callback), priority);
  EnqueuePendingRecordProducers();
}

void SpeculativeReportQueueImpl::EnqueuePendingRecordProducers() const {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  CHECK(actual_report_queue_.has_value());
  if (pending_record_producers_.empty()) {
    return;
  }
  const std::unique_ptr<ReportQueue>& report_queue =
      actual_report_queue_.value();
  auto head = std::move(pending_record_producers_.front());
  pending_record_producers_.pop();
  if (pending_record_producers_.empty()) {
    // Last of the pending records.
    report_queue->AddProducedRecord(std::move(head.record_producer),
                                    head.record_priority,
                                    std::move(head.record_callback));
    return;
  }
  report_queue->AddProducedRecord(
      std::move(head.record_producer), head.record_priority,
      base::BindPostTask(
          sequenced_task_runner_,
          base::BindOnce(
              [](base::WeakPtr<const SpeculativeReportQueueImpl> self,
                 EnqueueCallback callback, Status status) {
                if (!status.ok()) {
                  std::move(callback).Run(status);
                  return;
                }
                if (!self) {
                  std::move(callback).Run(
                      Status(error::UNAVAILABLE, "Queue has been destructed"));
                  base::UmaHistogramEnumeration(
                      reporting::kUmaUnavailableErrorReason,
                      UnavailableErrorReason::REPORT_QUEUE_DESTRUCTED,
                      UnavailableErrorReason::MAX_VALUE);
                  return;
                }
                std::move(callback).Run(status);
                self->EnqueuePendingRecordProducers();
              },
              weak_ptr_factory_.GetWeakPtr(),
              std::move(head.record_callback))));
}

base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
SpeculativeReportQueueImpl::PrepareToAttachActualQueue() const {
  return base::BindPostTask(
      sequenced_task_runner_,
      base::BindOnce(&SpeculativeReportQueueImpl::AttachActualQueue,
                     weak_ptr_factory_.GetMutableWeakPtr()));
}

Destination SpeculativeReportQueueImpl::GetDestination() const {
  return config_settings_.destination;
}

void SpeculativeReportQueueImpl::AttachActualQueue(
    StatusOr<std::unique_ptr<ReportQueue>> status_or_actual_queue) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  if (actual_report_queue_.has_value()) {
    // Already attached, do nothing.
    return;
  }
  if (!status_or_actual_queue.has_value()) {
    // Failed to create actual queue.
    // Flush all pending records with this status.
    PurgePendingProducers(status_or_actual_queue.error());
    return;
  }
  // Actual report queue succeeded, store it (never to change later).
  CHECK_EQ(config_settings_.destination,
           status_or_actual_queue.value()->GetDestination());
  actual_report_queue_ = std::move(status_or_actual_queue.value());
  EnqueuePendingRecordProducers();
}

void SpeculativeReportQueueImpl::PurgePendingProducers(Status status) const {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  while (!pending_record_producers_.empty()) {
    auto head = std::move(pending_record_producers_.front());
    pending_record_producers_.pop();
    base::UmaHistogramEnumeration(
        reporting::kUmaDataLossErrorReason,
        DataLossErrorReason::
            SPECULATIVE_REPORT_QUEUE_DESTRUCTED_BEFORE_RECORDS_ENQUEUED,
        DataLossErrorReason::MAX_VALUE);
    std::move(head.record_callback).Run(status);
  }
}
}  // namespace reporting