Introduce AttachmentStore metadata record

sync/internal_api/attachments/on_disk_attachment_store.cc

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "sync/internal_api/public/attachments/on_disk_attachment_store.h"
 
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/location.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/sequenced_task_runner.h"
+#include "sync/internal_api/attachments/proto/attachment_store.pb.h"
 #include "sync/protocol/attachments.pb.h"
 #include "third_party/leveldatabase/src/include/leveldb/db.h"
 #include "third_party/leveldatabase/src/include/leveldb/options.h"
 #include "third_party/leveldatabase/src/include/leveldb/slice.h"
 #include "third_party/leveldatabase/src/include/leveldb/status.h"
 
 namespace syncer {
 
 namespace {
 
 // Prefix for records containing attachment data.
 const char kDataPrefix[] = "data-";
 
+const char kDatabaseMetadataKey[] = "database-metadata";
+
+const int32 kCurrentSchemaVersion = 1;
+
 const base::FilePath::CharType kLeveldbDirectory[] =
     FILE_PATH_LITERAL("leveldb");
 }  // namespace
 
 OnDiskAttachmentStore::OnDiskAttachmentStore(
     const scoped_refptr<base::SequencedTaskRunner>& callback_task_runner)
     : callback_task_runner_(callback_task_runner) {
 }
 
 OnDiskAttachmentStore::~OnDiskAttachmentStore() {
 }
 
 void OnDiskAttachmentStore::Read(const AttachmentIdList& ids,
                                  const ReadCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK(db_);
   scoped_ptr<AttachmentMap> result_map(new AttachmentMap());
   scoped_ptr<AttachmentIdList> unavailable_attachments(new AttachmentIdList());
 
   leveldb::ReadOptions read_options;
   // Attachment content is typically large and only read once. Don't cache it on
   // db level.
   read_options.fill_cache = false;
   read_options.verify_checksums = true;
 
   AttachmentIdList::const_iterator iter = ids.begin();
   const AttachmentIdList::const_iterator end = ids.end();
   for (; iter != end; ++iter) {
-    const std::string key = CreateDataKeyFromAttachmentId(*iter);
+    const std::string key = MakeDataKeyFromAttachmentId(*iter);
     std::string data_str;
     leveldb::Status status = db_->Get(read_options, key, &data_str);
     if (!status.ok()) {
       DVLOG(1) << "DB::Get failed: status=" << status.ToString();
       unavailable_attachments->push_back(*iter);
       continue;
     }
     scoped_refptr<base::RefCountedMemory> data =
         base::RefCountedString::TakeString(&data_str);
     Attachment attachment = Attachment::CreateWithId(*iter, data);
@@ skipping 19 matching lines @@
   leveldb::ReadOptions read_options;
   read_options.fill_cache = false;
   read_options.verify_checksums = true;
 
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
   AttachmentList::const_iterator iter = attachments.begin();
   const AttachmentList::const_iterator end = attachments.end();
   for (; iter != end; ++iter) {
-    const std::string key = CreateDataKeyFromAttachmentId(iter->GetId());
+    const std::string key = MakeDataKeyFromAttachmentId(iter->GetId());
 
     std::string data_str;
     // TODO(pavely): crbug/424304 This read is expensive. When I add metadata
     // records this read will target metadata record instead of payload record.
     leveldb::Status status = db_->Get(read_options, key, &data_str);
     if (status.ok()) {
       // Entry exists, don't overwrite.
       continue;
     } else if (!status.IsNotFound()) {
       // Entry exists but failed to read.
@@ skipping 18 matching lines @@
 void OnDiskAttachmentStore::Drop(const AttachmentIdList& ids,
                                  const DropCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK(db_);
   Result result_code = SUCCESS;
   leveldb::WriteOptions write_options;
   write_options.sync = true;
   AttachmentIdList::const_iterator iter = ids.begin();
   const AttachmentIdList::const_iterator end = ids.end();
   for (; iter != end; ++iter) {
-    const std::string key = CreateDataKeyFromAttachmentId(*iter);
+    const std::string key = MakeDataKeyFromAttachmentId(*iter);
     leveldb::Status status = db_->Delete(write_options, key);
     if (!status.ok()) {
       // DB::Delete doesn't check if record exists, it returns ok just like
       // AttachmentStore::Drop should.
       DVLOG(1) << "DB::Delete failed: status=" << status.ToString();
       result_code = UNSPECIFIED_ERROR;
     }
   }
   callback_task_runner_->PostTask(FROM_HERE, base::Bind(callback, result_code));
 }
 
 AttachmentStore::Result OnDiskAttachmentStore::OpenOrCreate(
     const base::FilePath& path) {
   DCHECK(CalledOnValidThread());
   DCHECK(!db_);
-  Result result_code = UNSPECIFIED_ERROR;
   base::FilePath leveldb_path = path.Append(kLeveldbDirectory);
 
-  leveldb::DB* db;
+  leveldb::DB* db_raw;
+  scoped_ptr<leveldb::DB> db;
   leveldb::Options options;
   options.create_if_missing = true;
   // TODO(pavely): crbug/424287 Consider adding info_log, block_cache and
   // filter_policy to options.
   leveldb::Status status =
-      leveldb::DB::Open(options, leveldb_path.AsUTF8Unsafe(), &db);
+      leveldb::DB::Open(options, leveldb_path.AsUTF8Unsafe(), &db_raw);
   if (!status.ok()) {
     DVLOG(1) << "DB::Open failed: status=" << status.ToString()
              << ", path=" << path.AsUTF8Unsafe();
-  } else {
-    db_.reset(db);
-    result_code = SUCCESS;
+    return UNSPECIFIED_ERROR;
   }
-  return result_code;
+
+  db.reset(db_raw);
+
+  attachment_store_pb::AttachmentStoreMetadata metadata;
+  status = ReadStoreMetadata(db.get(), &metadata);
+  if (!status.ok() && !status.IsNotFound()) {
+    DVLOG(1) << "ReadStoreMetadata failed: status=" << status.ToString();
+    return UNSPECIFIED_ERROR;
+  }
+  if (status.IsNotFound()) {
+    // Brand new database.
+    metadata.set_schema_version(kCurrentSchemaVersion);
+    status = UpdateStoreMetadata(db.get(), metadata);
+    if (!status.ok()) {
+      DVLOG(1) << "UpdateStoreMetadata failed: status=" << status.ToString();
+      return UNSPECIFIED_ERROR;
+    }
+  }
+  DCHECK(status.ok());
+
+  // Upgrade code goes here.
+
+  if (metadata.schema_version() != kCurrentSchemaVersion) {
+    DVLOG(1) << "Unknown schema version: " << metadata.schema_version();
+    return UNSPECIFIED_ERROR;
+  }
+
+  db_ = db.Pass();
+  return SUCCESS;
 }
 
-std::string OnDiskAttachmentStore::CreateDataKeyFromAttachmentId(
+leveldb::Status OnDiskAttachmentStore::ReadStoreMetadata(
+    leveldb::DB* db,
+    attachment_store_pb::AttachmentStoreMetadata* metadata) {
+  std::string data_str;
+  leveldb::ReadOptions read_options;

[maniscalco, 2014/10/21 21:41:03]

Since we do this in a couple places, consider creating a helper function that
returns a ReadOptions object with the values we use.

(same may apply to WriteOptions)

[pavely, 2014/10/24 20:42:55]

I've created function for WriteOptions.
I want to defer doing the same for ReadingOptions to the change that does
attachment record metadata. Attachment store will use different combination of
read options for different things.

[maniscalco, 2014/10/24 21:52:39]

SGTM

+  read_options.fill_cache = false;
+  read_options.verify_checksums = true;
+
+  leveldb::Status status =
+      db->Get(read_options, kDatabaseMetadataKey, &data_str);
+  if (!status.ok())
+    return status;
+  if (!metadata->ParseFromString(data_str))
+    return leveldb::Status::Corruption("Metadata record corruption");
+  return leveldb::Status::OK();
+}
+
+leveldb::Status OnDiskAttachmentStore::UpdateStoreMetadata(

[maniscalco, 2014/10/21 21:41:03]

Light suggestion: since this function replaces the existing metadata record,
consider renaming to Write.  Update might be interpreted as updating only the
fields that have been set in the proto.

[pavely, 2014/10/24 20:42:55]

Done.

+    leveldb::DB* db,
+    const attachment_store_pb::AttachmentStoreMetadata& metadata) {
+  std::string data_str;
+  leveldb::WriteOptions write_options;
+  write_options.sync = true;
+
+  metadata.SerializeToString(&data_str);
+  return db->Put(write_options, kDatabaseMetadataKey, data_str);
+}
+
+std::string OnDiskAttachmentStore::MakeDataKeyFromAttachmentId(
     const AttachmentId& attachment_id) {
   std::string key = kDataPrefix + attachment_id.GetProto().unique_id();
   return key;
 }
 
 }  // namespace syncer