Refactor Storage to handle EncryptedRecord.

Since EncrypedRecord represents not only data but also sequencing
information, rely on it everywhere, including the unittests.
Extend test to verify sequencing too.

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

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

@@ -30,6 +30,7 @@
 #include "chrome/browser/browser_process.h"
 #include "chrome/browser/browser_process_platform_part.h"
 #include "chrome/browser/net/system_network_context_manager.h"
+#include "components/policy/proto/record.pb.h"
 #include "services/network/public/cpp/shared_url_loader_factory.h"
 
 #if defined(OS_CHROMEOS)
@@ -122,63 +123,42 @@ StatusOr<std::unique_ptr<Uploader>> Uploader::Create(
   return uploader;
 }
 
-void Uploader::ProcessBlob(Priority priority,
-                           StatusOr<base::span<const uint8_t>> data,
-                           base::OnceCallback<void(bool)> processed_cb) {
+void Uploader::ProcessRecord(StatusOr<EncryptedRecord> data,
+                             base::OnceCallback<void(bool)> processed_cb) {
   if (completed_ || !data.ok()) {
     std::move(processed_cb).Run(false);
     return;
   }
 
-  class ProcessBlobContext : public TaskRunnerContext<bool> {
+  class ProcessRecordContext : public TaskRunnerContext<bool> {
    public:
-    ProcessBlobContext(
-        base::span<const uint8_t> data,
+    ProcessRecordContext(
+        EncryptedRecord record,
         std::vector<EncryptedRecord>* records,
         base::OnceCallback<void(bool)> processed_callback,
         scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner)
         : TaskRunnerContext<bool>(std::move(processed_callback),
                                   sequenced_task_runner),
           records_(records),
-          data_(data.begin(), data.end()) {}
+          record_(std::move(record)) {}
 
    private:
-    ~ProcessBlobContext() override = default;
+    ~ProcessRecordContext() override = default;
 
     void OnStart() override {
-      if (data_.empty()) {
-        Complete(true);
-        return;
-      }
-      ProcessBlob();
-    }
-
-    void ProcessBlob() {
-      EncryptedRecord record;
-      if (!record.ParseFromArray(data_.data(), data_.size())) {
-        Complete(false);
-        return;
-      }
-      records_->push_back(record);
-      Complete(true);
-    }
-
-    void Complete(bool success) {
-      if (!success) {
-        LOG(ERROR) << "Unable to process blob";
-      }
-      Response(success);
+      records_->emplace_back(std::move(record_));
+      Response(true);
     }
 
     std::vector<EncryptedRecord>* const records_;
-    const std::vector<uint8_t> data_;
+    const EncryptedRecord record_;
   };
 
-  Start<ProcessBlobContext>(data.ValueOrDie(), encrypted_records_.get(),
-                            std::move(processed_cb), sequenced_task_runner_);
+  Start<ProcessRecordContext>(data.ValueOrDie(), encrypted_records_.get(),
+                              std::move(processed_cb), sequenced_task_runner_);
 }
 
-void Uploader::Completed(Priority priority, Status final_status) {
+void Uploader::Completed(Status final_status) {
   if (!final_status.ok()) {
     // No work to do - something went wrong with storage and it no longer wants
     // to upload the records. Let the records die with |this|.

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

@@ -18,6 +18,7 @@
 #include "chrome/browser/policy/messaging_layer/util/shared_queue.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
 #include "chrome/browser/policy/messaging_layer/util/task_runner_context.h"
+#include "components/policy/proto/record.pb.h"
 
 namespace reporting {
 
@@ -50,13 +51,10 @@ class ReportingClient {
     Uploader(const Uploader& other) = delete;
     Uploader& operator=(const Uploader& other) = delete;
 
-    // TODO(chromium:1078512) Priority is unused, remove it.
-    void ProcessBlob(Priority priority,
-                     StatusOr<base::span<const uint8_t>> data,
-                     base::OnceCallback<void(bool)> processed_cb) override;
+    void ProcessRecord(StatusOr<EncryptedRecord> data,
+                       base::OnceCallback<void(bool)> processed_cb) override;
 
-    // TODO(chromium:1078512) Priority is unused, remove it.
-    void Completed(Priority priority, Status final_status) override;
+    void Completed(Status final_status) override;
 
    private:
     explicit Uploader(UploadCallback upload_callback_);

chrome/browser/policy/messaging_layer/storage/storage.cc

@@ -18,6 +18,7 @@
 #include "chrome/browser/policy/messaging_layer/storage/storage_queue.h"
 #include "chrome/browser/policy/messaging_layer/util/status_macros.h"
 #include "chrome/browser/policy/messaging_layer/util/task_runner_context.h"
+#include "components/policy/proto/record.pb.h"
 
 namespace reporting {
 
@@ -123,12 +124,21 @@ class Storage::QueueUploaderInterface : public StorageQueue::UploaderInterface {
                                                     std::move(uploader));
   }
 
-  void ProcessBlob(StatusOr<base::span<const uint8_t>> data,
-                   base::OnceCallback<void(bool)> processed_cb) override {
-    storage_interface_->ProcessBlob(priority_, data, std::move(processed_cb));
+  void ProcessRecord(StatusOr<EncryptedRecord> encrypted_record,
+                     base::OnceCallback<void(bool)> processed_cb) override {
+    if (encrypted_record.ok()) {
+      // Update sequencing information: add Priority and Generation ID.
+      SequencingInformation* const sequencing_info =
+          encrypted_record.ValueOrDie().mutable_sequencing_information();
+      sequencing_info->set_priority(priority_);
+      // sequencing_info->set_generation_id(...);  Not supported yet.
+    }
+    storage_interface_->ProcessRecord(std::move(encrypted_record),
+                                      std::move(processed_cb));
   }
+
   void Completed(Status final_status) override {
-    storage_interface_->Completed(priority_, final_status);
+    storage_interface_->Completed(final_status);
   }
 
  private:
@@ -230,13 +240,13 @@ Storage::Storage(const Options& options, StartUploadCb start_upload_cb)
 Storage::~Storage() = default;
 
 void Storage::Write(Priority priority,
-                    base::span<const uint8_t> data,
+                    EncryptedRecord record,
                     base::OnceCallback<void(Status)> completion_cb) {
   // Note: queues_ never change after initialization is finished, so there is no
   // need to protect or serialize access to it.
   ASSIGN_OR_ONCE_CALLBACK_AND_RETURN(scoped_refptr<StorageQueue> queue,
                                      completion_cb, GetQueue(priority));
-  queue->Write(data, std::move(completion_cb));
+  queue->Write(std::move(record), std::move(completion_cb));
 }
 
 void Storage::Confirm(Priority priority,

chrome/browser/policy/messaging_layer/storage/storage.h

@@ -11,7 +11,6 @@
 
 #include "base/callback.h"
 #include "base/containers/flat_map.h"
-#include "base/containers/span.h"
 #include "base/files/file_path.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/scoped_refptr.h"
@@ -19,6 +18,7 @@
 #include "chrome/browser/policy/messaging_layer/storage/storage_queue.h"
 #include "chrome/browser/policy/messaging_layer/util/status.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "components/policy/proto/record.pb.h"
 #include "components/policy/proto/record_constants.pb.h"
 
 namespace reporting {
@@ -38,17 +38,16 @@ class Storage : public base::RefCountedThreadSafe<Storage> {
    public:
     virtual ~UploaderInterface() = default;
 
-    // Asynchronously processes every record (e.g. serializes and adds to the
-    // network message). Expects |processed_cb| to be called after the record
-    // has been processed, with true if next record needs to be delivered and
-    // false if the Uploader should stop.
-    virtual void ProcessBlob(Priority priority,
-                             StatusOr<base::span<const uint8_t>> data,
-                             base::OnceCallback<void(bool)> processed_cb) = 0;
+    // Unserializes every record and hands ownership over for processing (e.g.
+    // to add to the network message). Expects |processed_cb| to be called after
+    // the record or error status has been processed, with true if next record
+    // needs to be delivered and false if the Uploader should stop.
+    virtual void ProcessRecord(StatusOr<EncryptedRecord> record,
+                               base::OnceCallback<void(bool)> processed_cb) = 0;
 
     // Finalizes the upload (e.g. sends the message to the server and gets
     // response).
-    virtual void Completed(Priority priority, Status final_status) = 0;
+    virtual void Completed(Status final_status) = 0;
   };
 
   // Callback type for UploadInterface provider for specified queue.
@@ -82,11 +81,12 @@ class Storage : public base::RefCountedThreadSafe<Storage> {
       StartUploadCb start_upload_cb,
       base::OnceCallback<void(StatusOr<scoped_refptr<Storage>>)> completion_cb);
 
-  // Writes data blob into the Storage (the last file of it) according to the
+  // Serializes EncryptedRecord (taking ownership of it) and writes the
+  // resulting blob into the Storage (the last file of it) according to the
   // priority with the next sequencing number assigned. If file is going to
   // become too large, it is closed and new file is created.
   void Write(Priority priority,
-             base::span<const uint8_t> data,
+             EncryptedRecord record,
              base::OnceCallback<void(Status)> completion_cb);
 
   // Confirms acceptance of the records according to the priority up to
@@ -123,7 +123,7 @@ class Storage : public base::RefCountedThreadSafe<Storage> {
   // Returns OK or error status, if anything failed to initialize.
   Status Init();
 
-  // Helper function that selects queue by priority. Returns error
+  // Helper method that selects queue by priority. Returns error
   // if priority does not match any queue.
   // Note: queues_ never change after initialization is finished, so there is no
   // need to protect or serialize access to it.

chrome/browser/policy/messaging_layer/storage/storage_module.cc

@@ -25,11 +25,7 @@ StorageModule::~StorageModule() = default;
 void StorageModule::AddRecord(EncryptedRecord record,
                               Priority priority,
                               base::OnceCallback<void(Status)> callback) {
-  size_t record_size = record.ByteSizeLong();
-  auto data = std::make_unique<uint8_t[]>(record_size);
-  record.SerializeToArray(data.get(), record_size);
-  storage_->Write(priority, base::make_span(data.get(), record_size),
-                  std::move(callback));
+  storage_->Write(priority, std::move(record), std::move(callback));
 }
 
 void StorageModule::ReportSuccess(

chrome/browser/policy/messaging_layer/storage/storage_module.h

@@ -29,8 +29,9 @@ class StorageModule : public base::RefCountedThreadSafe<StorageModule> {
   StorageModule(const StorageModule& other) = delete;
   StorageModule& operator=(const StorageModule& other) = delete;
 
-  // AddRecord will add |record| to the |StorageModule| according to the
-  // provided |priority|. On completion, |callback| will be called.
+  // AddRecord will add |record| (taking ownership) to the |StorageModule|
+  // according to the provided |priority|. On completion, |callback| will be
+  // called.
   virtual void AddRecord(reporting::EncryptedRecord record,
                          reporting::Priority priority,
                          base::OnceCallback<void(Status)> callback);

chrome/browser/policy/messaging_layer/storage/storage_queue.h

@@ -11,7 +11,6 @@
 #include <vector>
 
 #include "base/callback.h"
-#include "base/containers/span.h"
 #include "base/files/file.h"
 #include "base/files/file_path.h"
 #include "base/memory/ref_counted.h"
@@ -24,6 +23,7 @@
 #include "base/timer/timer.h"
 #include "chrome/browser/policy/messaging_layer/util/status.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "components/policy/proto/record.pb.h"
 
 namespace reporting {
 
@@ -104,12 +104,12 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
    public:
     virtual ~UploaderInterface() = default;
 
-    // Asynchronously processes every record (e.g. serializes and adds to the
-    // network message). Expects |processed_cb| to be called after the record
-    // or error status has been processed, with true if next record needs to be
-    // delivered and false if the Uploader should stop.
-    virtual void ProcessBlob(StatusOr<base::span<const uint8_t>> data,
-                             base::OnceCallback<void(bool)> processed_cb) = 0;
+    // Unserializes every record and hands ownership over for processing (e.g.
+    // to add to the network message). Expects |processed_cb| to be called after
+    // the record or error status has been processed, with true if next record
+    // needs to be delivered and false if the Uploader should stop.
+    virtual void ProcessRecord(StatusOr<EncryptedRecord> record,
+                               base::OnceCallback<void(bool)> processed_cb) = 0;
 
     // Finalizes the upload (e.g. sends the message to server and gets
     // response). Called always, regardless of whether there were errors.
@@ -131,13 +131,14 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
       base::OnceCallback<void(StatusOr<scoped_refptr<StorageQueue>>)>
           completion_cb);
 
-  // Writes data blob into the StorageQueue (the last file of it) with the next
-  // sequencing number assigned. The write is a non-blocking operation -
-  // caller can "fire and forget" it (|completion_cb| allows to verify that
-  // record has been successfully enqueued). If file is going to become too
-  // large, it is closed and new file is created.
+  // Serializes EncryptedRecord (taking ownership of it) and writes the
+  // resulting blob into the StorageQueue (the last file of it) with the next
+  // sequencing number assigned. The write is a non-blocking operation - caller
+  // can "fire and forget" it (|completion_cb| allows to verify that record has
+  // been successfully enqueued). If file is going to become too large, it is
+  // closed and new file is created.
   // Helper methods: AssignLastFile, WriteHeaderAndBlock.
-  void Write(base::span<const uint8_t> data,
+  void Write(EncryptedRecord record,
              base::OnceCallback<void(Status)> completion_cb);
 
   // Confirms acceptance of the records up to |seq_number| (inclusively).
@@ -152,9 +153,9 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
   // a queue with an infinite or very large upload period. Multiple |Flush|
   // calls can safely run in parallel.
   // Starts by calling |start_upload_cb_| that instantiates |UploaderInterface
-  // uploader|. Then repeatedly reads data blob(s) one by one from the
+  // uploader|. Then repeatedly reads EncryptedRecord(s) one by one from the
   // StorageQueue starting from |first_seq_number_|, handing each one over to
-  // |uploader|->ProcessBlob (keeping ownership of the buffer) and resuming
+  // |uploader|->ProcessRecord (keeping ownership of the buffer) and resuming
   // after result callback returns 'true'. Only files that have been closed are
   // included in reading; |Upload| makes sure to close the last writeable file
   // and create a new one before starting to send records to the |uploader|. If
@@ -163,8 +164,8 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
   // |processed_cb| callback - in that case |Upload| will behave as if the end
   // of data has been reached. While one or more |Upload|s are active, files can
   // be added to the StorageQueue but cannot be deleted. If processing of the
-  // blob takes significant time, |uploader| implementation should be offset to
-  // another thread to avoid locking StorageQueue.
+  // record takes significant time, |uploader| implementation should be offset
+  // to another thread to avoid locking StorageQueue.
   // Helper methods: SwitchLastFileIfNotEmpty, CollectFilesForUpload.
   void Flush();
 
@@ -194,12 +195,12 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
     Status Delete();
 
     // Attempts to read |size| bytes from position |pos| and returns
-    // span of data that were actually read (no more than |size|).
-    // End of file is indicated by empty span.
-    StatusOr<base::span<const uint8_t>> Read(uint32_t pos, uint32_t size);
+    // reference to the data that were actually read (no more than |size|).
+    // End of file is indicated by empty data.
+    StatusOr<base::StringPiece> Read(uint32_t pos, uint32_t size);
 
     // Appends data to the file.
-    StatusOr<uint32_t> Append(base::span<const uint8_t> data);
+    StatusOr<uint32_t> Append(base::StringPiece data);
 
     bool is_opened() const { return handle_.get() != nullptr; }
     bool is_readonly() const {
@@ -232,7 +233,7 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
     size_t data_start_ = 0;
     size_t data_end_ = 0;
     uint64_t file_position_ = 0;
-    std::unique_ptr<uint8_t[]> buffer_;
+    std::unique_ptr<char[]> buffer_;
   };
 
   // Private constructor, to be called by Create factory method only.
@@ -242,8 +243,8 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
   // and determines the sequencing information of the last record.
   // Must be called once and only once after construction.
   // Returns OK or error status, if anything failed to initialize.
-  // Called once, during initialization. Helper methods: EnumerateDataFiles,
-  // ScanLastFile.
+  // Called once, during initialization.
+  // Helper methods: EnumerateDataFiles, ScanLastFile.
   Status Init();
 
   // Helper method for Init(): enumerates all data files in the directory.
@@ -268,7 +269,7 @@ class StorageQueue : public base::RefCountedThreadSafe<StorageQueue> {
 
   // Helper method for Write(): composes record header and writes it to the
   // file, followed by data.
-  Status WriteHeaderAndBlock(base::span<const uint8_t> data,
+  Status WriteHeaderAndBlock(base::StringPiece data,
                              scoped_refptr<SingleFile> file);
 
   // Helper method for Upload: if the last file is not empty (has at least one