[Sync] Add support for loading, updating and querying delete journals in

sync/syncable/directory.cc

 // Copyright (c) 2012 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/syncable/directory.h"
 
 #include "base/debug/trace_event.h"
 #include "base/perftimer.h"
 #include "base/stl_util.h"
 #include "base/string_number_conversions.h"
@@ skipping 83 matching lines @@
   download_progress[model_type].set_data_type_id(
       GetSpecificsFieldNumberFromModelType(model_type));
   // An empty-string token indicates no prior knowledge.
   download_progress[model_type].set_token(std::string());
 }
 
 Directory::SaveChangesSnapshot::SaveChangesSnapshot()
     : kernel_info_status(KERNEL_SHARE_INFO_INVALID) {
 }
 
-Directory::SaveChangesSnapshot::~SaveChangesSnapshot() {}
+Directory::SaveChangesSnapshot::~SaveChangesSnapshot() {
+  STLDeleteElements(&dirty_metas);

[tim (not reviewing), 2012/12/14 20:12:38]

Where was this being deleted before?

[haitaol1, 2012/12/14 23:15:36]

It doesn't need to be deleted because it's not a pointer before.

On 2012/12/14 20:12:38, timsteele wrote:

+  STLDeleteElements(&delete_journals);
+}
 
 Directory::Kernel::Kernel(
     const std::string& name,
     const KernelLoadInfo& info, DirectoryChangeDelegate* delegate,
     const WeakHandle<TransactionObserver>& transaction_observer)
     : next_write_transaction_id(0),
       name(name),
       metahandles_index(new Directory::MetahandlesIndex),
       ids_index(new Directory::IdsIndex),
       parent_id_child_index(new Directory::ParentIdChildIndex),
       client_tag_index(new Directory::ClientTagIndex),
       unsynced_metahandles(new MetahandleSet),
       dirty_metahandles(new MetahandleSet),
       metahandles_to_purge(new MetahandleSet),
       info_status(Directory::KERNEL_SHARE_INFO_VALID),
       persisted_info(info.kernel_info),
       cache_guid(info.cache_guid),
       next_metahandle(info.max_metahandle + 1),
       delegate(delegate),
-      transaction_observer(transaction_observer) {
+      transaction_observer(transaction_observer),
+      delete_journals_(new Directory::IdsIndex),
+      delete_journals_to_purge_(new MetahandleSet) {
   DCHECK(delegate);
   DCHECK(transaction_observer.IsInitialized());
 }
 
 Directory::Kernel::~Kernel() {
   delete unsynced_metahandles;
   delete dirty_metahandles;
   delete metahandles_to_purge;
   delete parent_id_child_index;
   delete client_tag_index;
   delete ids_index;
   STLDeleteElements(metahandles_index);
   delete metahandles_index;
+  STLDeleteElements(delete_journals_);
+  delete delete_journals_;
+  delete delete_journals_to_purge_;
 }
 
 Directory::Directory(
     DirectoryBackingStore* store,
     UnrecoverableErrorHandler* unrecoverable_error_handler,
     ReportUnrecoverableErrorFunction report_unrecoverable_error_function,
     NigoriHandler* nigori_handler,
     Cryptographer* cryptographer)
     : kernel_(NULL),
       store_(store),
@@ skipping 41 matching lines @@
     }
     DCHECK(!entry->is_dirty());
   }
 }
 
 DirOpenResult Directory::OpenImpl(
     const string& name,
     DirectoryChangeDelegate* delegate,
     const WeakHandle<TransactionObserver>&
         transaction_observer) {
-
   KernelLoadInfo info;
   // Temporary indices before kernel_ initialized in case Load fails. We 0(1)
   // swap these later.
   MetahandlesIndex metas_bucket;
-  DirOpenResult result = store_->Load(&metas_bucket, &info);
+  IdsIndex delete_journals;
+
+  DirOpenResult result = store_->Load(&metas_bucket, &delete_journals, &info);
   if (OPENED != result)
     return result;
 
   kernel_ = new Kernel(name, info, delegate, transaction_observer);
   kernel_->metahandles_index->swap(metas_bucket);
+  kernel_->delete_journals_->swap(delete_journals);
   InitializeIndices();
 
   // Write back the share info to reserve some space in 'next_id'.  This will
   // prevent local ID reuse in the case of an early crash.  See the comments in
   // TakeSnapshotForSaveChanges() or crbug.com/142987 for more information.
   kernel_->info_status = KERNEL_SHARE_INFO_DIRTY;
   if (!SaveChanges())
     return FAILED_INITIAL_WRITE;
 
   return OPENED;
 }
 
 void Directory::Close() {
   store_.reset();
   if (kernel_) {
     delete kernel_;
     kernel_ = NULL;
   }
 }
 
 void Directory::OnUnrecoverableError(const BaseTransaction* trans,
                                      const tracked_objects::Location& location,
                                      const std::string & message) {
   DCHECK(trans != NULL);
   unrecoverable_error_set_ = true;
   unrecoverable_error_handler_->OnUnrecoverableError(location,
                                                      message);
 }
 
+void Directory::UpdateDeleteJournals(BaseTransaction* trans,

[tim (not reviewing), 2012/12/13 23:41:30]

UpdateDeleteJournals sounds a bit more general than this is, since we only want
to do so / use this method when the server tells us an item is deleted. The code
assumes that is the case, and also, we should be on the sync thread when this
happens due to working on the SERVER_ namespace.
UpdateDeleteJournalForServerDelete?

[haitaol1, 2012/12/14 19:22:38]

Done.

+                                     bool was_deleted,
+                                     const EntryKernel& entry) {
+  if (!(IsDeleteJournalEnabled(
+          GetModelTypeFromSpecifics(entry.ref(SPECIFICS))) ||
+      IsDeleteJournalEnabled(entry.GetServerModelType()))) {

[tim (not reviewing), 2012/12/13 23:41:30]

Do you expect local specifics and server specifics to be different when this is
called?  Can one be an assertion? It's a bit intimidating to be unable to assume
anything / any preconditions about the state that |entry| is in at the time this
is called.

[haitaol1, 2012/12/14 19:22:38]

The specifics would be different because this is called during processing update
before server specs being copied to local specs. It should be sufficient to
check only server type in reality. But not all tests are set up like real and it
seems harmless to check both.

On 2012/12/13 23:41:30, timsteele wrote:

+    return;
+  }
+
+  ScopedKernelLock lock(this);
+  IdsIndex::const_iterator it =
+      kernel_->delete_journals_->find(const_cast<EntryKernel*>(&entry));

[tim (not reviewing), 2012/12/13 23:41:30]

Why remove the const-ness?  Everything about delete_journals and your comparison
routine seem to take const EntryKernel*s.

[haitaol1, 2012/12/14 19:22:38]

It's because IdIndex stores non-const pointer. Changed to use needle to be
absolutely safe.

On 2012/12/13 23:41:30, timsteele wrote:

+
+  if (entry.ref(SERVER_IS_DEL)) {
+    if (it == kernel_->delete_journals_->end()) {
+      // New delete.
+      EntryKernel* t = new EntryKernel(entry);
+      kernel_->delete_journals_->insert(t);
+      kernel_->delete_journals_to_purge_->erase(t->ref(META_HANDLE));
+    }
+  } else {
+    // Undelete. This could happen in two cases:
+    // * An entry was actually deleted and undeleted: was_deleted = true.

[tim (not reviewing), 2012/12/13 23:41:30]

Where would this undelete have come from?  The server?

[haitaol1, 2012/12/14 19:22:38]

Yes. Looks like if an entry was deleted on client A but has unsynced data on
client B, B's update sent to server will undelete the entry effectively.


On 2012/12/13 23:41:30, timsteele wrote:

[tim (not reviewing), 2012/12/14 20:12:38]

So, it was deleted (and possibly applied to the native model), but still remains
in the journal.  Can you comment that?

[haitaol1, 2012/12/14 23:15:36]

Done.

+    // * A data type was broken in last sync session and all its entries

[tim (not reviewing), 2012/12/13 23:41:30]

Define broken? 

[haitaol1, 2012/12/14 19:22:38]

Done.

+    //   were duplicated in delete journals. On restart, entries are recreated
+    //   from downloads and recreation calls UpdateDeleteJournals() to remove
+    //   live entries from delete journals, thus only deleted entries remain in
+    //   journals.
+    if (it != kernel_->delete_journals_->end()) {
+      kernel_->delete_journals_to_purge_->insert((*it)->ref(META_HANDLE));
+      delete *it;
+      kernel_->delete_journals_->erase(it);
+    } else if (was_deleted) {
+      kernel_->delete_journals_to_purge_->insert((*it)->ref(META_HANDLE));
+    }
+  }
+}
+
+void Directory::GetDeleteJournals(BaseTransaction* trans,
+                                  ModelType type,
+                                  EntryKernelSet* deleted_entries) {
+  ScopedKernelLock lock(this);
+  DCHECK(!passive_delete_journal_types_.Has(type));
+  for (IdsIndex::const_iterator it = kernel_->delete_journals_->begin();
+      it != kernel_->delete_journals_->end(); ++it) {
+    if (GetModelTypeFromSpecifics((*it)->ref(SPECIFICS)) == type ||
+        (*it)->GetServerModelType() == type) {
+      deleted_entries->insert(*it);
+    }
+  }
+  passive_delete_journal_types_.Put(type);
+}
+
+void Directory::PurgeDeleteJournals(BaseTransaction* trans,
+                                    const MetahandleSet& to_purge) {
+  ScopedKernelLock lock(this);
+  IdsIndex::const_iterator it = kernel_->delete_journals_->begin();
+  while (it != kernel_->delete_journals_->end()) {
+    int64 handle = (*it)->ref(META_HANDLE);
+    if (to_purge.count(handle)) {
+      delete *it;
+      kernel_->delete_journals_->erase(it++);
+    } else {
+      ++it;
+    }
+  }
+  kernel_->delete_journals_to_purge_->insert(to_purge.begin(), to_purge.end());
+}
 
 EntryKernel* Directory::GetEntryById(const Id& id) {
   ScopedKernelLock lock(this);
   return GetEntryById(id, &lock);
 }
 
 EntryKernel* Directory::GetEntryById(const Id& id,
                                      ScopedKernelLock* const lock) {
   DCHECK(kernel_);
   // Find it in the in memory ID index.
@@ skipping 210 matching lines @@
   // Deep copy dirty entries from kernel_->metahandles_index into snapshot and
   // clear dirty flags.
   for (MetahandleSet::const_iterator i = kernel_->dirty_metahandles->begin();
        i != kernel_->dirty_metahandles->end(); ++i) {
     EntryKernel* entry = GetEntryByHandle(*i, &lock);
     if (!entry)
       continue;
     // Skip over false positives; it happens relatively infrequently.
     if (!entry->is_dirty())
       continue;
-    snapshot->dirty_metas.insert(snapshot->dirty_metas.end(), *entry);
+    snapshot->dirty_metas.insert(snapshot->dirty_metas.end(),
+                                 new EntryKernel(*entry));
     DCHECK_EQ(1U, kernel_->dirty_metahandles->count(*i));
     // We don't bother removing from the index here as we blow the entire thing
     // in a moment, and it unnecessarily complicates iteration.
     entry->clear_dirty(NULL);
   }
   ClearDirtyMetahandles();
 
   // Set purged handles.
   DCHECK(snapshot->metahandles_to_purge.empty());
   snapshot->metahandles_to_purge.swap(*(kernel_->metahandles_to_purge));
 
   // Fill kernel_info_status and kernel_info.
   snapshot->kernel_info = kernel_->persisted_info;
   // To avoid duplicates when the process crashes, we record the next_id to be
   // greater magnitude than could possibly be reached before the next save
   // changes.  In other words, it's effectively impossible for the user to
   // generate 65536 new bookmarks in 3 seconds.
   snapshot->kernel_info.next_id -= 65536;
   snapshot->kernel_info_status = kernel_->info_status;
   // This one we reset on failure.
   kernel_->info_status = KERNEL_SHARE_INFO_VALID;
+
+  // Move passive delete journals to snapshot. Will copy back if snapshot fails
+  // to save.
+  MetahandlesIndex::const_iterator it = kernel_->delete_journals_->begin();
+  while (it != kernel_->delete_journals_->end()) {
+    if (passive_delete_journal_types_.Has((*it)->GetServerModelType()) ||
+        passive_delete_journal_types_.Has(GetModelTypeFromSpecifics(
+            (*it)->ref(SPECIFICS)))) {
+      snapshot->delete_journals.insert(*it);
+      kernel_->delete_journals_->erase(it++);
+    } else {
+      ++it;
+    }
+  }
+  snapshot->delete_journals_to_purge.swap(
+      *kernel_->delete_journals_to_purge_);
 }
 
 bool Directory::SaveChanges() {
   bool success = false;
 
   base::AutoLock scoped_lock(kernel_->save_changes_mutex);
 
   // Snapshot and save.
   SaveChangesSnapshot snapshot;
   TakeSnapshotForSaveChanges(&snapshot);
@@ skipping 10 matching lines @@
 bool Directory::VacuumAfterSaveChanges(const SaveChangesSnapshot& snapshot) {
   if (snapshot.dirty_metas.empty())
     return true;
 
   // Need a write transaction as we are about to permanently purge entries.
   WriteTransaction trans(FROM_HERE, VACUUM_AFTER_SAVE, this);
   ScopedKernelLock lock(this);
   // Now drop everything we can out of memory.
   for (EntryKernelSet::const_iterator i = snapshot.dirty_metas.begin();
        i != snapshot.dirty_metas.end(); ++i) {
-    kernel_->needle.put(META_HANDLE, i->ref(META_HANDLE));
+    kernel_->needle.put(META_HANDLE, (*i)->ref(META_HANDLE));
     MetahandlesIndex::iterator found =
         kernel_->metahandles_index->find(&kernel_->needle);
     EntryKernel* entry = (found == kernel_->metahandles_index->end() ?
                           NULL : *found);
     if (entry && SafeToPurgeFromMemory(&trans, entry)) {
       // We now drop deleted metahandles that are up to date on both the client
       // and the server.
       size_t num_erased = 0;
       num_erased = kernel_->ids_index->erase(entry);
       DCHECK_EQ(1u, num_erased);
       num_erased = kernel_->metahandles_index->erase(entry);
       DCHECK_EQ(1u, num_erased);
 
       // Might not be in it
       num_erased = kernel_->client_tag_index->erase(entry);
       DCHECK_EQ(entry->ref(UNIQUE_CLIENT_TAG).empty(), !num_erased);
       if (!SyncAssert(!kernel_->parent_id_child_index->count(entry),
                       FROM_HERE,
                       "Deleted entry still present",
                       (&trans)))
         return false;
       delete entry;
     }
     if (trans.unrecoverable_error_set())
       return false;
   }
   return true;
 }
 
 bool Directory::PurgeEntriesWithTypeIn(ModelTypeSet types) {

[tim (not reviewing), 2012/12/13 23:41:30]

I don't see any changes here for the delete journals, meaning server-deletes are
sticky across disable / re-enable of a datatype. I'm wondering if there are any
unintended side effects when migration or chip-machine fixup occurs...

[haitaol1, 2012/12/14 19:22:38]

I'm a little ahead of myself here. This is in another cl I'm gonna send out.

On 2012/12/13 23:41:30, timsteele wrote:

   if (types.Empty())
     return true;
 
   {
     WriteTransaction trans(FROM_HERE, PURGE_ENTRIES, this);
     {
       ScopedKernelLock lock(this);
       MetahandlesIndex::iterator it = kernel_->metahandles_index->begin();
       while (it != kernel_->metahandles_index->end()) {
         const sync_pb::EntitySpecifics& local_specifics = (*it)->ref(SPECIFICS);
@@ skipping 47 matching lines @@
   ScopedKernelLock lock(this);
   kernel_->info_status = KERNEL_SHARE_INFO_DIRTY;
 
   // Because we optimistically cleared the dirty bit on the real entries when
   // taking the snapshot, we must restore it on failure.  Not doing this could
   // cause lost data, if no other changes are made to the in-memory entries
   // that would cause the dirty bit to get set again. Setting the bit ensures
   // that SaveChanges will at least try again later.
   for (EntryKernelSet::const_iterator i = snapshot.dirty_metas.begin();
        i != snapshot.dirty_metas.end(); ++i) {
-    kernel_->needle.put(META_HANDLE, i->ref(META_HANDLE));
+    kernel_->needle.put(META_HANDLE, (*i)->ref(META_HANDLE));
     MetahandlesIndex::iterator found =
         kernel_->metahandles_index->find(&kernel_->needle);
     if (found != kernel_->metahandles_index->end()) {
       (*found)->mark_dirty(kernel_->dirty_metahandles);
     }
   }
 
   kernel_->metahandles_to_purge->insert(snapshot.metahandles_to_purge.begin(),
                                         snapshot.metahandles_to_purge.end());
+
+  // Restore delete journals.
+  for (EntryKernelSet::const_iterator i = snapshot.delete_journals.begin();
+       i != snapshot.delete_journals.end(); ++i) {
+    kernel_->needle.put(ID, (*i)->ref(ID));
+    if (kernel_->delete_journals_->find(&kernel_->needle) ==
+        kernel_->delete_journals_->end()) {
+      kernel_->delete_journals_->insert(new EntryKernel(**i));
+    }
+  }
+  kernel_->delete_journals_to_purge_->insert(
+      snapshot.delete_journals_to_purge.begin(),
+      snapshot.delete_journals_to_purge.end());
+
 }
 
 void Directory::GetDownloadProgress(
     ModelType model_type,
     sync_pb::DataTypeProgressMarker* value_out) const {
   ScopedKernelLock lock(this);
   return value_out->CopyFrom(
       kernel_->persisted_info.download_progress[model_type]);
 }
 
@@ skipping 622 matching lines @@
     // ordering.
     if (entry->ref(PREV_ID).IsRoot() ||
         entry->ref(PREV_ID) != entry->ref(NEXT_ID)) {
       return entry;
     }
   }
   // There were no children in the linked list.
   return NULL;
 }
 
+/* static */
+bool Directory::IsDeleteJournalEnabled(ModelType type) {
+  switch (type) {
+    case BOOKMARKS:
+      return true;
+    default:
+      return false;
+  }
+}
+
 ScopedKernelLock::ScopedKernelLock(const Directory* dir)
   :  scoped_lock_(dir->kernel_->mutex), dir_(const_cast<Directory*>(dir)) {
 }
 
 }  // namespace syncable
 }  // namespace syncer