storage_queue.h 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408
  1. // Copyright 2020 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #ifndef COMPONENTS_REPORTING_STORAGE_STORAGE_QUEUE_H_
  5. #define COMPONENTS_REPORTING_STORAGE_STORAGE_QUEUE_H_
  6. #include <list>
  7. #include <map>
  8. #include <memory>
  9. #include <string>
  10. #include "base/callback.h"
  11. #include "base/containers/flat_map.h"
  12. #include "base/containers/flat_set.h"
  13. #include "base/files/file.h"
  14. #include "base/files/file_enumerator.h"
  15. #include "base/files/file_path.h"
  16. #include "base/memory/ref_counted.h"
  17. #include "base/memory/ref_counted_delete_on_sequence.h"
  18. #include "base/memory/scoped_refptr.h"
  19. #include "base/strings/string_piece.h"
  20. #include "base/task/sequenced_task_runner.h"
  21. #include "base/threading/thread.h"
  22. #include "base/threading/thread_task_runner_handle.h"
  23. #include "base/timer/timer.h"
  24. #include "components/reporting/compression/compression_module.h"
  25. #include "components/reporting/encryption/encryption_module_interface.h"
  26. #include "components/reporting/proto/synced/record.pb.h"
  27. #include "components/reporting/storage/storage_configuration.h"
  28. #include "components/reporting/storage/storage_uploader_interface.h"
  29. #include "components/reporting/util/status.h"
  30. #include "components/reporting/util/statusor.h"
  31. #include "third_party/abseil-cpp/absl/types/optional.h"
  32. namespace reporting {
  33. namespace test {
  34. // Storage Queue operation kind used to associate operations with failures for
  35. // testing purposes
  36. enum class StorageQueueOperationKind {
  37. kReadBlock,
  38. kWriteBlock,
  39. kWriteMetadata
  40. };
  41. } // namespace test
  42. // Storage queue represents single queue of data to be collected and stored
  43. // persistently. It allows to add whole data records as necessary,
  44. // flush previously collected records and confirm records up to certain
  45. // sequencing id to be eliminated.
  46. class StorageQueue : public base::RefCountedDeleteOnSequence<StorageQueue> {
  47. public:
  48. // Creates StorageQueue instance with the specified options, and returns it
  49. // with the |completion_cb| callback. |async_start_upload_cb| is a factory
  50. // callback that instantiates UploaderInterface every time the queue starts
  51. // uploading records - periodically or immediately after Write (and in the
  52. // near future - upon explicit Flush request).
  53. static void Create(
  54. const QueueOptions& options,
  55. UploaderInterface::AsyncStartUploaderCb async_start_upload_cb,
  56. scoped_refptr<EncryptionModuleInterface> encryption_module,
  57. scoped_refptr<CompressionModule> compression_module,
  58. base::OnceCallback<void(StatusOr<scoped_refptr<StorageQueue>>)>
  59. completion_cb);
  60. // Wraps and serializes Record (taking ownership of it), encrypts and writes
  61. // the resulting blob into the StorageQueue (the last file of it) with the
  62. // next sequencing id assigned. The write is a non-blocking operation -
  63. // caller can "fire and forget" it (|completion_cb| allows to verify that
  64. // record has been successfully enqueued). If file is going to become too
  65. // large, it is closed and new file is created.
  66. // Helper methods: AssignLastFile, WriteHeaderAndBlock, OpenNewWriteableFile,
  67. // WriteMetadata, DeleteOutdatedMetadata.
  68. void Write(Record record, base::OnceCallback<void(Status)> completion_cb);
  69. // Confirms acceptance of the records up to |sequencing_id| (inclusively).
  70. // All records with sequencing ids <= this one can be removed from
  71. // the StorageQueue, and can no longer be uploaded.
  72. // If |force| is false (which is used in most cases), |sequencing_id| is
  73. // only accepted if no higher ids were confirmed before; otherwise it is
  74. // accepted unconditionally.
  75. // Helper methods: RemoveConfirmedData.
  76. void Confirm(absl::optional<int64_t> sequencing_id,
  77. bool force,
  78. base::OnceCallback<void(Status)> completion_cb);
  79. // Initiates upload of collected records. Called periodically by timer, based
  80. // on upload_period of the queue, and can also be called explicitly - for
  81. // a queue with an infinite or very large upload period. Multiple |Flush|
  82. // calls can safely run in parallel.
  83. // Starts by calling |async_start_upload_cb_| that instantiates
  84. // |UploaderInterface uploader|. Then repeatedly reads EncryptedRecord(s) one
  85. // by one from the StorageQueue starting from |first_sequencing_id_|, handing
  86. // each one over to |uploader|->ProcessRecord (keeping ownership of the
  87. // buffer) and resuming after result callback returns 'true'. Only files that
  88. // have been closed are included in reading; |Upload| makes sure to close the
  89. // last writeable file and create a new one before starting to send records to
  90. // the |uploader|. If some records are not available or corrupt,
  91. // |uploader|->ProcessGap is called. If the monotonic order of sequencing is
  92. // broken, INTERNAL error Status is reported. |Upload| can be stopped after
  93. // any record by returning 'false' to |processed_cb| callback - in that case
  94. // |Upload| will behave as if the end of data has been reached. While one or
  95. // more |Upload|s are active, files can be added to the StorageQueue but
  96. // cannot be deleted. If processing of the record takes significant time,
  97. // |uploader| implementation should be offset to another thread to avoid
  98. // locking StorageQueue. Helper methods: SwitchLastFileIfNotEmpty,
  99. // CollectFilesForUpload.
  100. void Flush();
  101. // Test only: makes specified records fail on specified operation kind.
  102. void TestInjectErrorsForOperation(
  103. const test::StorageQueueOperationKind operation_kind,
  104. std::initializer_list<int64_t> sequencing_ids);
  105. // Access queue options.
  106. const QueueOptions& options() const { return options_; }
  107. StorageQueue(const StorageQueue& other) = delete;
  108. StorageQueue& operator=(const StorageQueue& other) = delete;
  109. protected:
  110. virtual ~StorageQueue();
  111. private:
  112. friend class base::RefCountedDeleteOnSequence<StorageQueue>;
  113. friend class base::DeleteHelper<StorageQueue>;
  114. // Private data structures for Read and Write (need access to the private
  115. // StorageQueue fields).
  116. class WriteContext;
  117. class ReadContext;
  118. class ConfirmContext;
  119. // Private envelope class for single file in a StorageQueue.
  120. class SingleFile : public base::RefCountedThreadSafe<SingleFile> {
  121. public:
  122. // Factory method creates a SingleFile object for existing
  123. // or new file (of zero size). In case of any error (e.g. insufficient disk
  124. // space) returns status.
  125. static StatusOr<scoped_refptr<SingleFile>> Create(
  126. const base::FilePath& filename,
  127. int64_t size,
  128. scoped_refptr<ResourceInterface> memory_resource,
  129. scoped_refptr<ResourceInterface> disk_space_resource);
  130. // Returns the file sequence ID (the first sequence ID in the file) if the
  131. // sequence ID can be extracted from the extension. Otherwise, returns an
  132. // error status.
  133. static StatusOr<int64_t> GetFileSequenceIdFromPath(
  134. const base::FilePath& file_name);
  135. Status Open(bool read_only); // No-op if already opened.
  136. void Close(); // No-op if not opened.
  137. void DeleteWarnIfFailed();
  138. // Attempts to read |size| bytes from position |pos| and returns
  139. // reference to the data that were actually read (no more than |size|).
  140. // End of file is indicated by empty data.
  141. // |max_buffer_size| specifies the largest allowed buffer, which
  142. // must accommodate the largest possible data block plus header and
  143. // overhead.
  144. // |expect_readonly| must match to is_readonly() (when set to false,
  145. // the file is expected to be writeable; this only happens when scanning
  146. // files restarting the queue).
  147. StatusOr<base::StringPiece> Read(uint32_t pos,
  148. uint32_t size,
  149. size_t max_buffer_size,
  150. bool expect_readonly = true);
  151. // Appends data to the file.
  152. StatusOr<uint32_t> Append(base::StringPiece data);
  153. bool is_opened() const { return handle_.get() != nullptr; }
  154. bool is_readonly() const {
  155. DCHECK(is_opened());
  156. return is_readonly_.value();
  157. }
  158. uint64_t size() const { return size_; }
  159. std::string name() const { return filename_.MaybeAsASCII(); }
  160. protected:
  161. virtual ~SingleFile();
  162. private:
  163. friend class base::RefCountedThreadSafe<SingleFile>;
  164. // Private constructor, called by factory method only.
  165. SingleFile(const base::FilePath& filename,
  166. int64_t size,
  167. scoped_refptr<ResourceInterface> memory_resource,
  168. scoped_refptr<ResourceInterface> disk_space_resource);
  169. // Flag (valid for opened file only): true if file was opened for reading
  170. // only, false otherwise.
  171. absl::optional<bool> is_readonly_;
  172. const base::FilePath filename_; // relative to the StorageQueue directory
  173. uint64_t size_ = 0; // tracked internally rather than by filesystem
  174. std::unique_ptr<base::File> handle_; // Set only when opened/created.
  175. scoped_refptr<ResourceInterface> memory_resource_;
  176. scoped_refptr<ResourceInterface> disk_space_resource_;
  177. // When reading the file, this is the buffer and data positions.
  178. // If the data is read sequentially, buffered portions are reused
  179. // improving performance. When the sequential order is broken (e.g.
  180. // we start reading the same file in parallel from different position),
  181. // the buffer is reset.
  182. size_t data_start_ = 0;
  183. size_t data_end_ = 0;
  184. uint64_t file_position_ = 0;
  185. size_t buffer_size_ = 0;
  186. std::unique_ptr<char[]> buffer_;
  187. };
  188. // Private constructor, to be called by Create factory method only.
  189. StorageQueue(scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner,
  190. const QueueOptions& options,
  191. UploaderInterface::AsyncStartUploaderCb async_start_upload_cb,
  192. scoped_refptr<EncryptionModuleInterface> encryption_module,
  193. scoped_refptr<CompressionModule> compression_module);
  194. // Initializes the object by enumerating files in the assigned directory
  195. // and determines the sequence information of the last record.
  196. // Must be called once and only once after construction.
  197. // Returns OK or error status, if anything failed to initialize.
  198. // Called once, during initialization.
  199. // Helper methods: EnumerateDataFiles, ScanLastFile, RestoreMetadata.
  200. Status Init();
  201. // Retrieves last record digest (does not exist at a generation start).
  202. absl::optional<std::string> GetLastRecordDigest() const;
  203. // Helper method for Init(): process single data file.
  204. // Return sequencing_id from <prefix>.<sequencing_id> file name, or Status
  205. // in case there is any error.
  206. StatusOr<int64_t> AddDataFile(
  207. const base::FilePath& full_name,
  208. const base::FileEnumerator::FileInfo& file_info);
  209. // Helper method for Init(): sets generation id based on data file name.
  210. // For backwards compatibility, accepts file name without generation too.
  211. Status SetGenerationId(const base::FilePath& full_name);
  212. // Helper method for Init(): enumerates all data files in the directory.
  213. // Valid file names are <prefix>.<sequencing_id>, any other names are ignored.
  214. // Adds used data files to the set.
  215. Status EnumerateDataFiles(base::flat_set<base::FilePath>* used_files_set);
  216. // Helper method for Init(): scans the last file in StorageQueue, if there are
  217. // files at all, and learns the latest sequencing id. Otherwise (if there
  218. // are no files) sets it to 0.
  219. Status ScanLastFile();
  220. // Helper method for Write(): increments sequencing id and assigns last
  221. // file to place record in. |size| parameter indicates the size of data that
  222. // comprise the record expected to be appended; if appending the record will
  223. // make the file too large, the current last file will be closed, and a new
  224. // file will be created and assigned to be the last one.
  225. StatusOr<scoped_refptr<SingleFile>> AssignLastFile(size_t size);
  226. // Helper method for Write() and Read(): creates and opens a new empty
  227. // writeable file, adding it to |files_|.
  228. StatusOr<scoped_refptr<SingleFile>> OpenNewWriteableFile();
  229. // Helper method for Write(): stores a file with metadata to match the
  230. // incoming new record. Synchronously composes metadata to record, then
  231. // asynchronously writes it into a file with next sequencing id and then
  232. // notifies the Write operation that it can now complete. After that it
  233. // asynchronously deletes all other files with lower sequencing id
  234. // (multiple Writes can see the same files and attempt to delete them, and
  235. // that is not an error).
  236. Status WriteMetadata(base::StringPiece current_record_digest);
  237. // Helper method for RestoreMetadata(): loads and verifies metadata file
  238. // contents. If accepted, adds the file to the set.
  239. Status ReadMetadata(const base::FilePath& meta_file_path,
  240. size_t size,
  241. int64_t sequencing_id,
  242. base::flat_set<base::FilePath>* used_files_set);
  243. // Helper method for Init(): locates file with metadata that matches the
  244. // last sequencing id and loads metadata from it.
  245. // Adds used metadata file to the set.
  246. Status RestoreMetadata(base::flat_set<base::FilePath>* used_files_set);
  247. // Delete all files except those listed in |used_file_set|.
  248. void DeleteUnusedFiles(
  249. const base::flat_set<base::FilePath>& used_files_set) const;
  250. // Helper method for Write(): deletes meta files up to, but not including
  251. // |sequencing_id_to_keep|. Any errors are ignored.
  252. void DeleteOutdatedMetadata(int64_t sequencing_id_to_keep);
  253. // Helper method for Write(): composes record header and writes it to the
  254. // file, followed by data. Stores record digest in the queue, increments
  255. // next sequencing id.
  256. Status WriteHeaderAndBlock(base::StringPiece data,
  257. base::StringPiece current_record_digest,
  258. scoped_refptr<SingleFile> file);
  259. // Helper method for Upload: if the last file is not empty (has at least one
  260. // record), close it and create the new one, so that its records are also
  261. // included in the reading.
  262. Status SwitchLastFileIfNotEmpty();
  263. // Helper method for Upload: collects and sets aside |files| in the
  264. // StorageQueue that have data for the Upload (all files that have records
  265. // with sequencing ids equal or higher than |sequencing_id|).
  266. std::map<int64_t, scoped_refptr<SingleFile>> CollectFilesForUpload(
  267. int64_t sequencing_id) const;
  268. // Helper method for Confirm: Moves |first_sequencing_id_| to
  269. // (|sequencing_id|+1) and removes files that only have records with seq
  270. // ids below or equal to |sequencing_id| (below |first_sequencing_id_|).
  271. Status RemoveConfirmedData(int64_t sequencing_id);
  272. // Helper method to release all file instances held by the queue.
  273. // Files on the disk remain as they were.
  274. void ReleaseAllFileInstances();
  275. // Helper method to retry upload if prior one failed or if some events below
  276. // |next_sequencing_id| were not uploaded.
  277. void CheckBackUpload(Status status, int64_t next_sequencing_id);
  278. // Helper method called by periodic time to upload data.
  279. void PeriodicUpload();
  280. // Sequential task runner for all activities in this StorageQueue
  281. // (must be first member in class).
  282. scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
  283. SEQUENCE_CHECKER(storage_queue_sequence_checker_);
  284. // Immutable options, stored at the time of creation.
  285. const QueueOptions options_;
  286. // Current generation id, unique per device and queue.
  287. // Set up once during initialization by reading from the 'gen_id.NNNN' file
  288. // matching the last sequencing id, or generated anew as a random number if no
  289. // such file found (files do not match the id).
  290. int64_t generation_id_ = 0;
  291. // Digest of the last written record (loaded at queue initialization, absent
  292. // if the new generation has just started, and no records where stored yet).
  293. absl::optional<std::string> last_record_digest_;
  294. // Queue of the write context instances in the order of creation, sequencing
  295. // ids and record digests. Context is always removed from this queue before
  296. // being destructed. We use std::list rather than std::queue, because
  297. // if the write fails, it needs to be removed from the queue regardless of
  298. // whether it is at the head, tail or middle.
  299. std::list<WriteContext*> write_contexts_queue_;
  300. // Next sequencing id to store (not assigned yet).
  301. int64_t next_sequencing_id_ = 0;
  302. // First sequencing id store still has (no records with lower
  303. // sequencing id exist in store).
  304. int64_t first_sequencing_id_ = 0;
  305. // First unconfirmed sequencing id (no records with lower
  306. // sequencing id will be ever uploaded). Set by the first
  307. // Confirm call.
  308. // If first_unconfirmed_sequencing_id_ < first_sequencing_id_,
  309. // [first_unconfirmed_sequencing_id_, first_sequencing_id_) is a gap
  310. // that cannot be filled in and is uploaded as such.
  311. absl::optional<int64_t> first_unconfirmed_sequencing_id_;
  312. // Latest metafile. May be null.
  313. scoped_refptr<SingleFile> meta_file_;
  314. // Ordered map of the files by ascending sequencing id.
  315. std::map<int64_t, scoped_refptr<SingleFile>> files_;
  316. // Counter of the Read operations. When not 0, none of the files_ can be
  317. // deleted. Incremented by |ReadContext::OnStart|, decremented by
  318. // |ReadContext::OnComplete|. Accessed by |RemoveConfirmedData|.
  319. // All accesses take place on sequenced_task_runner_.
  320. int32_t active_read_operations_ = 0;
  321. // Upload timer (active only if options_.upload_period() is not 0).
  322. base::RepeatingTimer upload_timer_;
  323. // Upload provider callback.
  324. const UploaderInterface::AsyncStartUploaderCb async_start_upload_cb_;
  325. // Encryption module.
  326. scoped_refptr<EncryptionModuleInterface> encryption_module_;
  327. // Compression module.
  328. scoped_refptr<CompressionModule> compression_module_;
  329. // Test only: records specified to fail for a given operation kind.
  330. base::flat_map<test::StorageQueueOperationKind, base::flat_set<int64_t>>
  331. test_injected_failures_;
  332. // Weak pointer factory (must be last member in class).
  333. base::WeakPtrFactory<StorageQueue> weakptr_factory_{this};
  334. };
  335. } // namespace reporting
  336. #endif // COMPONENTS_REPORTING_STORAGE_STORAGE_QUEUE_H_