[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);
+  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,
+                                     bool was_deleted,
+                                     const EntryKernel* entry) {

[tim (not reviewing), 2012/12/13 02:48:10]

How come this can't be a ref?

[haitaol1, 2012/12/13 21:34:21]

Done.

+  if (!IsDeleteJournalEnabledForType(entry->GetServerModelType()))
+    return;
+
+  ScopedKernelLock lock(this);
+  IdsIndex::const_iterator it =
+      kernel_->delete_journals_->find(const_cast<EntryKernel*>(entry));
+
+  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.
+    // * A data type was broken in last sync session and all its entries
+    //   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 ((*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())) {
+      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);
@@ skipping 77 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