Implememt garbage collection of old entries according to

version_watermark specified in progress marker during update processing.

BUG=347253

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

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@254298 0039d316-1c4b-4281-b951-d872f2087c98

9 files changed, 209 insertions, 5 deletions

diff --git a/sync/engine/directory_update_handler.cc b/sync/engine/directory_update_handler.cc
index be36082..9d97765 100644
--- a/sync/engine/directory_update_handler.cc
+++ b/sync/engine/directory_update_handler.cc
@@ -37,7 +37,11 @@ void DirectoryUpdateHandler::ProcessGetUpdatesResponse(
     sessions::StatusController* status) {
   syncable::ModelNeutralWriteTransaction trans(FROM_HERE, SYNCER, dir_);
   UpdateSyncEntities(&trans, applicable_updates, status);
-  UpdateProgressMarker(progress_marker);
+
+  if (IsValidProgressMarker(progress_marker)) {
+    ExpireEntriesIfNeeded(&trans, progress_marker);
+    UpdateProgressMarker(progress_marker);
+  }
 }
 
 void DirectoryUpdateHandler::ApplyUpdates(sessions::StatusController* status) {
@@ -147,8 +151,8 @@ void DirectoryUpdateHandler::UpdateSyncEntities(
   ProcessDownloadedUpdates(dir_, trans, type_, applicable_updates, status);
 }
 
-void DirectoryUpdateHandler::UpdateProgressMarker(
-    const sync_pb::DataTypeProgressMarker& progress_marker) {
+bool DirectoryUpdateHandler::IsValidProgressMarker(
+    const sync_pb::DataTypeProgressMarker& progress_marker) const {
   int field_number = progress_marker.data_type_id();
   ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
   if (!IsRealDataType(model_type) || type_ != model_type) {
@@ -156,8 +160,50 @@ void DirectoryUpdateHandler::UpdateProgressMarker(
         << "Update handler of type " << ModelTypeToString(type_)
         << " asked to process progress marker with invalid type "
         << field_number;
+    return false;
+  }
+  return true;
+}
+
+void DirectoryUpdateHandler::UpdateProgressMarker(
+    const sync_pb::DataTypeProgressMarker& progress_marker) {
+  if (progress_marker.has_gc_directive() || !cached_gc_directive_) {
+    dir_->SetDownloadProgress(type_, progress_marker);
+  } else {
+    sync_pb::DataTypeProgressMarker merged_marker = progress_marker;
+    merged_marker.mutable_gc_directive()->CopyFrom(*cached_gc_directive_);
+    dir_->SetDownloadProgress(type_, merged_marker);
+  }
+}
+
+void DirectoryUpdateHandler::ExpireEntriesIfNeeded(
+    syncable::ModelNeutralWriteTransaction* trans,
+    const sync_pb::DataTypeProgressMarker& progress_marker) {
+  if (!cached_gc_directive_) {
+    sync_pb::DataTypeProgressMarker current_marker;
+    GetDownloadProgress(&current_marker);
+    if (current_marker.has_gc_directive()) {
+      cached_gc_directive_.reset(new sync_pb::GarbageCollectionDirective(
+          current_marker.gc_directive()));
+    }
+  }
+
+  if (!progress_marker.has_gc_directive())
+    return;
+
+  const sync_pb::GarbageCollectionDirective& new_gc_directive =
+      progress_marker.gc_directive();
+
+  if (new_gc_directive.has_version_watermark() &&
+      (!cached_gc_directive_ ||
+          cached_gc_directive_->version_watermark() <
+              new_gc_directive.version_watermark())) {
+    ExpireEntriesByVersion(dir_, trans, type_,
+                           new_gc_directive.version_watermark());
   }
-  dir_->SetDownloadProgress(type_, progress_marker);
+
+  cached_gc_directive_.reset(
+      new sync_pb::GarbageCollectionDirective(new_gc_directive));
 }
 
 }  // namespace syncer
diff --git a/sync/engine/directory_update_handler.h b/sync/engine/directory_update_handler.h
index 25acbd9..5cc150d 100644
--- a/sync/engine/directory_update_handler.h
+++ b/sync/engine/directory_update_handler.h
@@ -9,6 +9,7 @@
 
 #include "base/basictypes.h"
 #include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
 #include "sync/base/sync_export.h"
 #include "sync/engine/process_updates_util.h"
 #include "sync/engine/update_handler.h"
@@ -17,6 +18,7 @@
 
 namespace sync_pb {
 class DataTypeProgressMarker;
+class GarbageCollectionDirective;
 class GetUpdatesResponse;
 }
 
@@ -82,11 +84,19 @@ class SYNC_EXPORT_PRIVATE DirectoryUpdateHandler : public UpdateHandler {
       const SyncEntityList& applicable_updates,
       sessions::StatusController* status);
 
+  // Expires entries according to GC directives.
+  void ExpireEntriesIfNeeded(
+      syncable::ModelNeutralWriteTransaction* trans,
+      const sync_pb::DataTypeProgressMarker& progress_marker);
+
   // Stores the given progress marker in the directory.
   // Its type must match this update handler's type.
   void UpdateProgressMarker(
       const sync_pb::DataTypeProgressMarker& progress_marker);
 
+  bool IsValidProgressMarker(
+      const sync_pb::DataTypeProgressMarker& progress_marker) const;
+
   // Skips all checks and goes straight to applying the updates.
   SyncerError ApplyUpdatesImpl(sessions::StatusController* status);
 
@@ -94,6 +104,8 @@ class SYNC_EXPORT_PRIVATE DirectoryUpdateHandler : public UpdateHandler {
   ModelType type_;
   scoped_refptr<ModelSafeWorker> worker_;
 
+  scoped_ptr<sync_pb::GarbageCollectionDirective> cached_gc_directive_;
+
   DISALLOW_COPY_AND_ASSIGN(DirectoryUpdateHandler);
 };
 
diff --git a/sync/engine/directory_update_handler_unittest.cc b/sync/engine/directory_update_handler_unittest.cc
index 0352505..ada49e1 100644
--- a/sync/engine/directory_update_handler_unittest.cc
+++ b/sync/engine/directory_update_handler_unittest.cc
@@ -29,6 +29,8 @@ namespace syncer {
 
 using syncable::UNITTEST;
 
+static const int64 kDefaultVersion = 1000;
+
 // A test harness for tests that focus on processing updates.
 //
 // Update processing is what occurs when we first download updates.  It converts
@@ -75,6 +77,13 @@ class DirectoryUpdateHandlerProcessUpdateTest : public ::testing::Test {
     return ui_worker_;
   }
 
+  bool EntryExists(const std::string& id) {
+    syncable::ReadTransaction trans(FROM_HERE, dir());
+    syncable::Entry e(&trans, syncable::GET_BY_ID,
+                      syncable::Id::CreateFromServerId(id));
+    return e.good() && !e.GetIsDel();
+  }
+
  private:
   base::MessageLoop loop_;  // Needed to initialize the directory.
   TestDirectorySetterUpper dir_maker_;
@@ -91,7 +100,7 @@ DirectoryUpdateHandlerProcessUpdateTest::CreateUpdate(
   e->set_parent_id_string(parent);
   e->set_non_unique_name(id);
   e->set_name(id);
-  e->set_version(1000);
+  e->set_version(kDefaultVersion);
   AddDefaultFieldValue(type, e->mutable_specifics());
   return e.Pass();
 }
@@ -232,6 +241,59 @@ TEST_F(DirectoryUpdateHandlerProcessUpdateTest, ProcessNewProgressMarkers) {
   EXPECT_EQ(progress.data_type_id(), saved.data_type_id());
 }
 
+TEST_F(DirectoryUpdateHandlerProcessUpdateTest, GarbageCollectionByVersion) {
+  DirectoryUpdateHandler handler(dir(), SYNCED_NOTIFICATIONS, ui_worker());
+  sessions::StatusController status;
+
+  sync_pb::DataTypeProgressMarker progress;
+  progress.set_data_type_id(
+      GetSpecificsFieldNumberFromModelType(SYNCED_NOTIFICATIONS));
+  progress.set_token("token");
+  progress.mutable_gc_directive()->set_version_watermark(kDefaultVersion + 10);
+
+  scoped_ptr<sync_pb::SyncEntity> type_root =
+      CreateUpdate(SyncableIdToProto(syncable::Id::CreateFromServerId("root")),
+                   syncable::GetNullId().GetServerId(),
+                   SYNCED_NOTIFICATIONS);
+  type_root->set_server_defined_unique_tag(
+      ModelTypeToRootTag(SYNCED_NOTIFICATIONS));
+  type_root->set_folder(true);
+
+  scoped_ptr<sync_pb::SyncEntity> e1 =
+      CreateUpdate(SyncableIdToProto(syncable::Id::CreateFromServerId("e1")),
+                   type_root->id_string(),
+                   SYNCED_NOTIFICATIONS);
+
+  scoped_ptr<sync_pb::SyncEntity> e2 =
+      CreateUpdate(SyncableIdToProto(syncable::Id::CreateFromServerId("e2")),
+                   type_root->id_string(),
+                   SYNCED_NOTIFICATIONS);
+  e2->set_version(kDefaultVersion + 100);
+
+  // Add to the applicable updates list.
+  SyncEntityList updates;
+  updates.push_back(type_root.get());
+  updates.push_back(e1.get());
+  updates.push_back(e2.get());
+
+  // Process and apply updates.
+  handler.ProcessGetUpdatesResponse(progress, updates, &status);
+  handler.ApplyUpdates(&status);
+
+  // Verify none is deleted because they are unapplied during GC.
+  EXPECT_TRUE(EntryExists(type_root->id_string()));
+  EXPECT_TRUE(EntryExists(e1->id_string()));
+  EXPECT_TRUE(EntryExists(e2->id_string()));
+
+  // Process and apply again. Old entry is deleted but not root.
+  progress.mutable_gc_directive()->set_version_watermark(kDefaultVersion + 20);
+  handler.ProcessGetUpdatesResponse(progress, SyncEntityList(), &status);
+  handler.ApplyUpdates(&status);
+  EXPECT_TRUE(EntryExists(type_root->id_string()));
+  EXPECT_FALSE(EntryExists(e1->id_string()));
+  EXPECT_TRUE(EntryExists(e2->id_string()));
+}
+
 // A test harness for tests that focus on applying updates.
 //
 // Update application is performed when we want to take updates that were
diff --git a/sync/engine/get_updates_processor.cc b/sync/engine/get_updates_processor.cc
index 2ffc4b3..399c78c 100644
--- a/sync/engine/get_updates_processor.cc
+++ b/sync/engine/get_updates_processor.cc
@@ -175,6 +175,7 @@ void GetUpdatesProcessor::PrepareGetUpdates(
     sync_pb::DataTypeProgressMarker* progress_marker =
         get_updates->add_from_progress_marker();
     handler_it->second->GetDownloadProgress(progress_marker);
+    progress_marker->clear_gc_directive();
   }
   delegate_.HelpPopulateGuMessage(get_updates);
 }
diff --git a/sync/engine/process_updates_util.cc b/sync/engine/process_updates_util.cc
index b2cab4a..7cfb920 100644
--- a/sync/engine/process_updates_util.cc
+++ b/sync/engine/process_updates_util.cc
@@ -300,4 +300,28 @@ void ProcessDownloadedUpdates(
   }
 }
 
+void ExpireEntriesByVersion(syncable::Directory* dir,
+                            syncable::ModelNeutralWriteTransaction* trans,
+                            ModelType type,
+                            int64 version_watermark) {
+  syncable::Directory::Metahandles handles;
+  dir->GetMetaHandlesOfType(trans, type, &handles);
+  for (size_t i = 0; i < handles.size(); ++i) {
+    syncable::ModelNeutralMutableEntry entry(trans, syncable::GET_BY_HANDLE,
+                                             handles[i]);
+    if (!entry.good() || !entry.GetId().ServerKnows() ||
+        entry.GetUniqueServerTag() == ModelTypeToRootTag(type) ||
+        entry.GetIsUnappliedUpdate() || entry.GetIsUnsynced() ||
+        entry.GetIsDel() || entry.GetServerIsDel() ||
+        entry.GetBaseVersion() >= version_watermark) {
+      continue;
+    }
+
+    // Mark entry as deleted by server.
+    entry.PutServerIsDel(true);
+    entry.PutServerVersion(version_watermark);
+    entry.PutIsUnappliedUpdate(true);
+  }
+}
+
 }  // namespace syncer
diff --git a/sync/engine/process_updates_util.h b/sync/engine/process_updates_util.h
index 428ff57..4cbd103 100644
--- a/sync/engine/process_updates_util.h
+++ b/sync/engine/process_updates_util.h
@@ -34,6 +34,13 @@ void ProcessDownloadedUpdates(
     const SyncEntityList& applicable_updates,
     sessions::StatusController* status);
 
+// Tombstones all entries of |type| whose versions are older than
+// |version_watermark| unless they are type root or unsynced/unapplied.
+void ExpireEntriesByVersion(syncable::Directory* dir,
+                            syncable::ModelNeutralWriteTransaction* trans,
+                            ModelType type,
+                            int64 version_watermark);
+
 }  // namespace syncer
 
 #endif  // SYNC_ENGINE_PROCESS_UPDATES_UTIL_H_
diff --git a/sync/protocol/sync.proto b/sync/protocol/sync.proto
index 55bc9b3..6d8d4b2 100644
--- a/sync/protocol/sync.proto
+++ b/sync/protocol/sync.proto
@@ -473,6 +473,32 @@ message GetUpdateTriggers {
   optional bool server_dropped_hints = 6;
 }
 
+message GarbageCollectionDirective {
+  enum Type {
+    UNKNOWN = 0;
+    VERSION_WATERMARK = 1;
+    AGE_WATERMARK = 2;
+    MAX_ITEM_COUNT = 3;
+  }
+
+  optional Type type = 1 [default = UNKNOWN];
+
+  // This field specifies the watermark for the versions which should get
+  // garbage collected.  The client should purge all sync entities with a
+  // version smaller than version_watermark locally.
+  optional int64 version_watermark = 2;
+
+  // This field specifies the watermark in terms of age in days.  The client
+  // should purge all sync entities which are older than this specific value
+  // based on last modified time.
+  optional int32 age_watermark_in_days = 3;
+
+  // This field specifies the max number of items that the client should keep
+  // for a specific datatype.  If the number of items exceeds this limit, the
+  // client should purge the extra sync entities based on the LRU rule.
+  optional int32 max_number_of_items = 4;
+}
+
 message DataTypeProgressMarker {
   // An integer identifying the data type whose progress is tracked by this
   // marker.  The legitimate values of this field correspond to the protobuf
@@ -520,6 +546,12 @@ message DataTypeProgressMarker {
 
   // This field will be included only in GetUpdates with origin GU_TRIGGER.
   optional GetUpdateTriggers get_update_triggers = 5;
+
+  // The garbage collection directive for this data type.  The client should
+  // purge items locally based on this directive.  Since this directive is
+  // designed to be sent from server only, the client should persist it locally
+  // as needed and avoid sending it to the server.
+  optional GarbageCollectionDirective gc_directive = 6;
 }
 
 message GetUpdatesMessage {
diff --git a/sync/syncable/directory.cc b/sync/syncable/directory.cc
index d83b1f6..97e7071 100644
--- a/sync/syncable/directory.cc
+++ b/sync/syncable/directory.cc
@@ -903,6 +903,21 @@ void Directory::GetUnappliedUpdateMetaHandles(
   }
 }
 
+void Directory::GetMetaHandlesOfType(BaseTransaction* trans,
+                                     ModelType type,
+                                     std::vector<int64>* result) {
+  result->clear();
+  ScopedKernelLock lock(this);
+  for (MetahandlesMap::iterator it = kernel_->metahandles_map.begin();
+       it != kernel_->metahandles_map.end(); ++it) {
+    EntryKernel* entry = it->second;
+    const ModelType entry_type =
+        GetModelTypeFromSpecifics(entry->ref(SPECIFICS));
+    if (entry_type == type)
+      result->push_back(it->first);
+  }
+}
+
 void Directory::CollectMetaHandleCounts(
     std::vector<int>* num_entries_by_type,
     std::vector<int>* num_to_delete_entries_by_type) {
diff --git a/sync/syncable/directory.h b/sync/syncable/directory.h
index c5959b9..ca8f4f1 100644
--- a/sync/syncable/directory.h
+++ b/sync/syncable/directory.h
@@ -315,6 +315,11 @@ class SYNC_EXPORT Directory {
                                      FullModelTypeSet server_types,
                                      std::vector<int64>* result);
 
+  // Get all the metahandles of entries of |type|.
+  void GetMetaHandlesOfType(BaseTransaction* trans,
+                            ModelType type,
+                            Metahandles* result);
+
   // Get metahandle counts for various criteria to show on the
   // about:sync page. The information is computed on the fly
   // each time. If this results in a significant performance hit,