Make GenericChangeProcessor upload attachments on startup.

Convert some uses of AttachmentIdList to AttachmentIdSet.

Remove the no longer needed UploadAttachments method from
GenericChangeProcessor.

BUG=372622

Review URL: https://codereview.chromium.org/582913002

Cr-Commit-Position: refs/heads/master@{#295988}

components/sync_driver/generic_change_processor.cc

@@ -112,6 +112,7 @@ GenericChangeProcessor::GenericChangeProcessor(
     attachment_service_proxy_.reset(new syncer::AttachmentServiceProxy(
         base::MessageLoopProxy::current(),
         attachment_service_weak_ptr_factory_->GetWeakPtr()));
+    UploadAllAttachmentsNotOnServer();
   } else {
     attachment_service_proxy_.reset(new syncer::AttachmentServiceProxy(
         base::MessageLoopProxy::current(),
@@ -408,7 +409,7 @@ syncer::SyncError GenericChangeProcessor::ProcessSyncChanges(
 
   // Keep track of brand new attachments so we can persist them on this device
   // and upload them to the server.
-  syncer::AttachmentIdList new_attachments;
+  syncer::AttachmentIdSet new_attachments;
 
   syncer::WriteTransaction trans(from_here, share_handle());
 
@@ -470,7 +471,7 @@ syncer::SyncError GenericChangeProcessor::ProcessSyncChanges(
       NOTREACHED();
       return error;
     }
-    UploadAttachments(new_attachments);
+    attachment_service_->UploadAttachments(new_attachments);
   }
 
   return syncer::SyncError();
@@ -485,7 +486,7 @@ syncer::SyncError GenericChangeProcessor::HandleActionAdd(
     const std::string& type_str,
     const syncer::WriteTransaction& trans,
     syncer::WriteNode* sync_node,
-    syncer::AttachmentIdList* new_attachments) {
+    syncer::AttachmentIdSet* new_attachments) {
   // TODO(sync): Handle other types of creation (custom parents, folders,
   // etc.).
   syncer::ReadNode root_node(&trans);
@@ -553,8 +554,7 @@ syncer::SyncError GenericChangeProcessor::HandleActionAdd(
   SetAttachmentMetadata(attachment_ids, sync_node);
 
   // Return any newly added attachments.
-  new_attachments->insert(
-      new_attachments->end(), attachment_ids.begin(), attachment_ids.end());
+  new_attachments->insert(attachment_ids.begin(), attachment_ids.end());
   if (merge_result_.get()) {
     merge_result_->set_num_items_added(merge_result_->num_items_added() + 1);
   }
@@ -569,7 +569,7 @@ syncer::SyncError GenericChangeProcessor::HandleActionUpdate(
     const std::string& type_str,
     const syncer::WriteTransaction& trans,
     syncer::WriteNode* sync_node,
-    syncer::AttachmentIdList* new_attachments) {
+    syncer::AttachmentIdSet* new_attachments) {
   // TODO(zea): consider having this logic for all possible changes?
 
   const syncer::SyncDataLocal sync_data_local(change.sync_data());
@@ -655,8 +655,7 @@ syncer::SyncError GenericChangeProcessor::HandleActionUpdate(
   SetAttachmentMetadata(attachment_ids, sync_node);
 
   // Return any newly added attachments.
-  new_attachments->insert(
-      new_attachments->end(), attachment_ids.begin(), attachment_ids.end());
+  new_attachments->insert(attachment_ids.begin(), attachment_ids.end());
 
   if (merge_result_.get()) {
     merge_result_->set_num_items_modified(merge_result_->num_items_modified() +
@@ -703,14 +702,17 @@ syncer::UserShare* GenericChangeProcessor::share_handle() const {
   return share_handle_;
 }
 
-void GenericChangeProcessor::UploadAttachments(
-    const syncer::AttachmentIdList& attachment_ids) {
+void GenericChangeProcessor::UploadAllAttachmentsNotOnServer() {
   DCHECK(CalledOnValidThread());
-  DCHECK(attachment_service_.get() != NULL);
-
-  syncer::AttachmentIdSet attachment_id_set;
-  attachment_id_set.insert(attachment_ids.begin(), attachment_ids.end());
-  attachment_service_->UploadAttachments(attachment_id_set);
+  DCHECK(attachment_service_.get());
+  syncer::AttachmentIdSet id_set;
+  {
+    syncer::ReadTransaction trans(FROM_HERE, share_handle());
+    trans.GetAttachmentIdsToUpload(type_, &id_set);
+  }
+  if (!id_set.empty()) {
+    attachment_service_->UploadAttachments(id_set);
+  }
 }
 
 }  // namespace sync_driver

components/sync_driver/generic_change_processor.h

@@ -114,7 +114,7 @@ class GenericChangeProcessor : public ChangeProcessor,
                                     const std::string& type_str,
                                     const syncer::WriteTransaction& trans,
                                     syncer::WriteNode* sync_node,
-                                    syncer::AttachmentIdList* new_attachments);
+                                    syncer::AttachmentIdSet* new_attachments);
 
   // Logically part of ProcessSyncChanges.
   //
@@ -125,13 +125,11 @@ class GenericChangeProcessor : public ChangeProcessor,
       const std::string& type_str,
       const syncer::WriteTransaction& trans,
       syncer::WriteNode* sync_node,
-      syncer::AttachmentIdList* new_attachments);
+      syncer::AttachmentIdSet* new_attachments);
 
-  // Upload |attachments| to the sync server.
-  //
-  // This function assumes that attachments were already stored in
-  // AttachmentStore.
-  void UploadAttachments(const syncer::AttachmentIdList& attachment_ids);
+  // Begin uploading attachments that have not yet been uploaded to the sync
+  // server.
+  void UploadAllAttachmentsNotOnServer();
 
   const syncer::ModelType type_;
 

components/sync_driver/generic_change_processor_unittest.cc

@@ -11,6 +11,7 @@
 #include "base/strings/stringprintf.h"
 #include "components/sync_driver/data_type_error_handler_mock.h"
 #include "components/sync_driver/sync_api_component_factory.h"
+#include "sync/api/attachments/attachment_id.h"
 #include "sync/api/attachments/fake_attachment_store.h"
 #include "sync/api/fake_syncable_service.h"
 #include "sync/api/sync_change.h"
@@ -142,9 +143,12 @@ class SyncGenericChangeProcessorTest : public testing::Test {
                                         test_user_share_->user_share());
     }
     test_user_share_->encryption_handler()->Init();
+    ConstructGenericChangeProcessor(type);
+  }
+
+  void ConstructGenericChangeProcessor(syncer::ModelType type) {
     scoped_refptr<syncer::AttachmentStore> attachment_store(
         new syncer::FakeAttachmentStore(base::MessageLoopProxy::current()));
-
     scoped_ptr<MockAttachmentService> mock_attachment_service(
         new MockAttachmentService(attachment_store));
     // GenericChangeProcessor takes ownership of the AttachmentService, but we
@@ -446,6 +450,36 @@ TEST_F(SyncGenericChangeProcessorTest, AttachmentUploaded) {
   EXPECT_EQ(1U, attachment_ids.size());
 }
 
+// Verify that upon construction, all attachments not yet on the server are
+// scheduled for upload.
+TEST_F(SyncGenericChangeProcessorTest, UploadAllAttachmentsNotOnServer) {
+  // Create two attachment ids.  id2 will be marked as "on server".
+  syncer::AttachmentId id1 = syncer::AttachmentId::Create();
+  syncer::AttachmentId id2 = syncer::AttachmentId::Create();
+  {
+    // Write an entry containing these two attachment ids.
+    syncer::WriteTransaction trans(FROM_HERE, user_share());
+    syncer::ReadNode root(&trans);
+    ASSERT_EQ(syncer::BaseNode::INIT_OK, root.InitTypeRoot(kType));
+    syncer::WriteNode node(&trans);
+    node.InitUniqueByCreation(kType, root, "some node");
+    sync_pb::AttachmentMetadata metadata;
+    sync_pb::AttachmentMetadataRecord* record1 = metadata.add_record();
+    *record1->mutable_id() = id1.GetProto();
+    sync_pb::AttachmentMetadataRecord* record2 = metadata.add_record();
+    *record2->mutable_id() = id2.GetProto();
+    record2->set_is_on_server(true);
+    node.SetAttachmentMetadata(metadata);
+  }
+
+  // Construct the GenericChangeProcessor and see that it asks the
+  // AttachmentService to upload id1 only.
+  ConstructGenericChangeProcessor(kType);
+  ASSERT_EQ(1U, mock_attachment_service()->attachment_id_sets()->size());
+  ASSERT_THAT(mock_attachment_service()->attachment_id_sets()->front(),
+              testing::UnorderedElementsAre(id1));
+}
+
 }  // namespace
 
 }  // namespace sync_driver

sync/api/attachments/README

@@ -2,5 +2,5 @@ This directory contains the sync attachment interface code that is used by both
 consumers of sync and sync itself.
 
 Because parts of sync may depend on this code, it's important that it remains
-"leafy" and never depends on sync/internal_api/ or else we may end up with
-cycles.
+"leafy" and never depends on sync/internal_api/ or sync/syncable/, or else we
+may end up with cycles.

sync/internal_api/public/read_transaction.h

@@ -6,6 +6,7 @@
 #define SYNC_INTERNAL_API_PUBLIC_READ_TRANSACTION_H_
 
 #include "base/compiler_specific.h"
+#include "sync/api/attachments/attachment_id.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/base_transaction.h"
 
@@ -45,6 +46,10 @@ class SYNC_EXPORT ReadTransaction : public BaseTransaction {
   void GetDataTypeContext(ModelType type,
                           sync_pb::DataTypeContext* context) const;
 
+  // Clears |id_set| and fills it with the ids of attachments that need to be
+  // uploaded to the sync server.
+  void GetAttachmentIdsToUpload(ModelType type, AttachmentIdSet* id_set);
+
  private:
   void* operator new(size_t size);  // Transaction is meant for stack use only.
 

sync/internal_api/read_transaction.cc

@@ -47,4 +47,11 @@ void ReadTransaction::GetDataTypeContext(
       transaction_, type, context);
 }
 
+void ReadTransaction::GetAttachmentIdsToUpload(ModelType type,
+                                               AttachmentIdSet* id_set) {
+  DCHECK(id_set);
+  transaction_->directory()->GetAttachmentIdsToUpload(
+      transaction_, type, id_set);
+}
+
 }  // namespace syncer

sync/syncable/DEPS

 include_rules = [
   "+net/base/escape.h",
   "+sql",
+  "+sync/api/attachments",
   "+sync/base",
   "+sync/internal_api/public/base",
   "+sync/internal_api/public/engine",

sync/syncable/directory.cc

@@ -4,6 +4,7 @@
 
 #include "sync/syncable/directory.h"
 
+#include <algorithm>
 #include <iterator>
 
 #include "base/base64.h"
@@ -1063,8 +1064,15 @@ void Directory::GetUnappliedUpdateMetaHandles(
 void Directory::GetMetaHandlesOfType(BaseTransaction* trans,
                                      ModelType type,
                                      std::vector<int64>* result) {
-  result->clear();
   ScopedKernelLock lock(this);
+  GetMetaHandlesOfType(lock, trans, type, result);
+}
+
+void Directory::GetMetaHandlesOfType(const ScopedKernelLock& lock,
+                                     BaseTransaction* trans,
+                                     ModelType type,
+                                     std::vector<int64>* result) {
+  result->clear();
   for (MetahandlesMap::iterator it = kernel_->metahandles_map.begin();
        it != kernel_->metahandles_map.end(); ++it) {
     EntryKernel* entry = it->second;
@@ -1470,5 +1478,59 @@ void Directory::UnmarkDirtyEntry(WriteTransaction* trans, Entry* entry) {
   entry->kernel_->clear_dirty(&kernel_->dirty_metahandles);
 }
 
+void Directory::GetAttachmentIdsToUpload(BaseTransaction* trans,
+                                         ModelType type,
+                                         AttachmentIdSet* id_set) {
+  // TODO(maniscalco): Maintain an index by ModelType and rewrite this method to
+  // use it.  The approach below is likely very expensive because it iterates
+  // all entries (bug 415199).
+  DCHECK(trans);
+  DCHECK(id_set);
+  id_set->clear();
+  AttachmentIdSet on_server_id_set;
+  AttachmentIdSet not_on_server_id_set;
+  std::vector<int64> metahandles;
+  {
+    ScopedKernelLock lock(this);
+    GetMetaHandlesOfType(lock, trans, type, &metahandles);
+    std::vector<int64>::const_iterator iter = metahandles.begin();
+    const std::vector<int64>::const_iterator end = metahandles.end();
+    // For all of this type's entries...
+    for (; iter != end; ++iter) {
+      EntryKernel* entry = GetEntryByHandle(*iter, &lock);
+      DCHECK(entry);
+      const sync_pb::AttachmentMetadata metadata =
+          entry->ref(ATTACHMENT_METADATA);
+      // for each of this entry's attachments...
+      for (int i = 0; i < metadata.record_size(); ++i) {
+        AttachmentId id =
+            AttachmentId::CreateFromProto(metadata.record(i).id());
+        // if this attachment is known to be on the server, remember it for
+        // later,
+        if (metadata.record(i).is_on_server()) {
+          on_server_id_set.insert(id);
+        } else {
+          // otherwise, add it to id_set.
+          not_on_server_id_set.insert(id);
+        }
+      }
+    }
+  }
+  // Why did we bother keeping a set of ids known to be on the server?  The
+  // is_on_server flag is stored denormalized so we can end up with two entries
+  // with the same attachment id where one says it's on the server and the other
+  // says it's not.  When this happens, we trust the one that says it's on the
+  // server.  To avoid re-uploading the same attachment mulitple times, we
+  // remove any ids known to be on the server from the id_set we are about to
+  // return.
+  //
+  // TODO(maniscalco): Eliminate redundant metadata storage (bug 415203).
+  std::set_difference(not_on_server_id_set.begin(),
+                      not_on_server_id_set.end(),
+                      on_server_id_set.begin(),
+                      on_server_id_set.end(),
+                      std::inserter(*id_set, id_set->end()));
+}
+
 }  // namespace syncable
 }  // namespace syncer

sync/syncable/directory.h

@@ -15,6 +15,7 @@
 #include "base/files/file_util.h"
 #include "base/gtest_prod_util.h"
 #include "base/values.h"
+#include "sync/api/attachments/attachment_id.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/util/report_unrecoverable_error_function.h"
 #include "sync/internal_api/public/util/weak_handle.h"
@@ -413,6 +414,12 @@ class SYNC_EXPORT Directory {
   // preserve sync preferences in DB on disk.
   void UnmarkDirtyEntry(WriteTransaction* trans, Entry* entry);
 
+  // Clears |id_set| and fills it with the ids of attachments that need to be
+  // uploaded to the sync server.
+  void GetAttachmentIdsToUpload(BaseTransaction* trans,
+                                ModelType type,
+                                AttachmentIdSet* id_set);
+
  protected:  // for friends, mainly used by Entry constructors
   virtual EntryKernel* GetEntryByHandle(int64 handle);
   virtual EntryKernel* GetEntryByHandle(int64 metahandle,
@@ -603,6 +610,13 @@ class SYNC_EXPORT Directory {
       const sync_pb::AttachmentMetadata& attachment_metadata,
       const ScopedKernelLock& lock);
 
+  // A private version of the public GetMetaHandlesOfType for when you already
+  // have a ScopedKernelLock.
+  void GetMetaHandlesOfType(const ScopedKernelLock& lock,
+                            BaseTransaction* trans,
+                            ModelType type,
+                            std::vector<int64>* result);
+
   Kernel* kernel_;
 
   scoped_ptr<DirectoryBackingStore> store_;

sync/syncable/directory_unittest.cc

@@ -1719,6 +1719,53 @@ TEST_F(SyncableDirectoryTest, Directory_LastReferenceUnlinksAttachments) {
   ASSERT_FALSE(dir()->IsAttachmentLinked(attachment_id_proto));
 }
 
+TEST_F(SyncableDirectoryTest, Directory_GetAttachmentIdsToUpload) {
+  // Create one attachment, referenced by two entries.
+  AttachmentId attachment_id = AttachmentId::Create();
+  sync_pb::AttachmentIdProto attachment_id_proto = attachment_id.GetProto();
+  sync_pb::AttachmentMetadata attachment_metadata;
+  sync_pb::AttachmentMetadataRecord* record = attachment_metadata.add_record();
+  *record->mutable_id() = attachment_id_proto;
+  const Id id1 = TestIdFactory::FromNumber(-1);
+  const Id id2 = TestIdFactory::FromNumber(-2);
+  CreateEntryWithAttachmentMetadata(
+      PREFERENCES, "some entry", id1, attachment_metadata);
+  CreateEntryWithAttachmentMetadata(
+      PREFERENCES, "some other entry", id2, attachment_metadata);
+
+  // See that Directory reports that this attachment is not on the server.
+  AttachmentIdSet id_set;
+  {
+    ReadTransaction trans(FROM_HERE, dir().get());
+    dir()->GetAttachmentIdsToUpload(&trans, PREFERENCES, &id_set);
+  }
+  ASSERT_EQ(1U, id_set.size());
+  ASSERT_EQ(attachment_id, *id_set.begin());
+
+  // Call again, but this time with a ModelType for which there are no entries.
+  // See that Directory correctly reports that there are none.
+  {
+    ReadTransaction trans(FROM_HERE, dir().get());
+    dir()->GetAttachmentIdsToUpload(&trans, PASSWORDS, &id_set);
+  }
+  ASSERT_TRUE(id_set.empty());
+
+  // Now, mark the attachment as "on the server" via entry_1.
+  {
+    WriteTransaction trans(FROM_HERE, UNITTEST, dir().get());
+    MutableEntry entry_1(&trans, GET_BY_ID, id1);
+    entry_1.MarkAttachmentAsOnServer(attachment_id_proto);
+  }
+
+  // See that Directory no longer reports that this attachment is not on the
+  // server.
+  {
+    ReadTransaction trans(FROM_HERE, dir().get());
+    dir()->GetAttachmentIdsToUpload(&trans, PREFERENCES, &id_set);
+  }
+  ASSERT_TRUE(id_set.empty());
+}
+
 }  // namespace syncable
 
 }  // namespace syncer