Implement new invalidations ack tracking system

chrome/browser/policy/cloud/cloud_policy_invalidator.cc

 // Copyright 2013 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 "chrome/browser/policy/cloud/cloud_policy_invalidator.h"
 
 #include "base/bind.h"
 #include "base/command_line.h"
 #include "base/hash.h"
 #include "base/location.h"
@@ skipping 24 matching lines @@
     : state_(UNINITIALIZED),
       core_(core),
       task_runner_(task_runner),
       invalidation_service_(NULL),
       invalidations_enabled_(false),
       invalidation_service_enabled_(false),
       registered_timestamp_(0),
       invalid_(false),
       invalidation_version_(0),
       unknown_version_invalidation_count_(0),
-      ack_handle_(syncer::AckHandle::InvalidAckHandle()),
       weak_factory_(this),
       max_fetch_delay_(kMaxFetchDelayDefault),
       policy_hash_value_(0) {
   DCHECK(core);
   DCHECK(task_runner.get());
 }
 
 CloudPolicyInvalidator::~CloudPolicyInvalidator() {
   DCHECK(state_ == SHUT_DOWN);
 }
@@ skipping 35 matching lines @@
 void CloudPolicyInvalidator::OnIncomingInvalidation(
     const syncer::ObjectIdInvalidationMap& invalidation_map) {
   DCHECK(state_ == STARTED);
   DCHECK(thread_checker_.CalledOnValidThread());
   const syncer::SingleObjectInvalidationSet& list =
       invalidation_map.ForObject(object_id_);
   if (list.IsEmpty()) {
     NOTREACHED();
     return;
   }
+
+  // Acknowledge all except the invalidation with the highest version.
+  syncer::SingleObjectInvalidationSet::const_reverse_iterator it =
+      list.rbegin();
+  ++it;
+  for ( ; it != list.rend(); ++it) {
+    it->Acknowledge();
+  }
+
+  // Handle the highest version invalidation.
   HandleInvalidation(list.back());
 }
 
 void CloudPolicyInvalidator::OnCoreConnected(CloudPolicyCore* core) {}
 
 void CloudPolicyInvalidator::OnRefreshSchedulerStarted(CloudPolicyCore* core) {
   DCHECK(state_ == STOPPED);
   DCHECK(thread_checker_.CalledOnValidThread());
   state_ = STARTED;
   OnStoreLoaded(core_->store());
@@ skipping 36 matching lines @@
   }
 
   UpdateRegistration(store->policy());
   UpdateMaxFetchDelay(store->policy_map());
 }
 
 void CloudPolicyInvalidator::OnStoreError(CloudPolicyStore* store) {}
 
 void CloudPolicyInvalidator::HandleInvalidation(
     const syncer::Invalidation& invalidation) {
-  // The invalidation service may send an invalidation more than once if there
-  // is a delay in acknowledging it. Duplicate invalidations are ignored.
-  if (invalid_ && ack_handle_.Equals(invalidation.ack_handle()))
+  // Ignore old invalidations.
+  if (invalid_ &&
+      !invalidation.is_unknown_version() &&
+      invalidation.version() <= invalidation_version_) {
     return;
+  }
 
   // If there is still a pending invalidation, acknowledge it, since we only
   // care about the latest invalidation.
   if (invalid_)
     AcknowledgeInvalidation();
 
   // Update invalidation state.
   invalid_ = true;
-  ack_handle_ = invalidation.ack_handle();
+  invalidation_.reset(new syncer::Invalidation(invalidation));
 
   // When an invalidation with unknown version is received, use negative
   // numbers based on the number of such invalidations received. This
   // ensures that the version numbers do not collide with "real" versions
   // (which are positive) or previous invalidations with unknown version.
   if (invalidation.is_unknown_version()) {
     invalidation_version_ = -(++unknown_version_invalidation_count_);
   } else {
     invalidation_version_ = invalidation.version();
   }
@@ skipping 135 matching lines @@
   // the client yet, so set it now that the required time has elapsed.
   if (is_missing_payload)
     core_->client()->SetInvalidationInfo(invalidation_version_, std::string());
   core_->refresh_scheduler()->RefreshSoon();
 }
 
 void CloudPolicyInvalidator::AcknowledgeInvalidation() {
   DCHECK(invalid_);
   invalid_ = false;
   core_->client()->SetInvalidationInfo(0, std::string());
-  invalidation_service_->AcknowledgeInvalidation(object_id_, ack_handle_);
+  invalidation_->Acknowledge();
+  invalidation_.reset();
   // Cancel any scheduled policy refreshes.
   weak_factory_.InvalidateWeakPtrs();
 }
 
 bool CloudPolicyInvalidator::IsPolicyChanged(
     const enterprise_management::PolicyData* policy) {
   // Determine if the policy changed by comparing its hash value to the
   // previous policy's hash value.
   uint32 new_hash_value = 0;
   if (policy && policy->has_policy_value())
@@ skipping 10 matching lines @@
     if (invalidations_enabled_)
       return METRIC_POLICY_REFRESH_CHANGED;
     return METRIC_POLICY_REFRESH_CHANGED_NO_INVALIDATIONS;
   }
   if (invalid_)
     return METRIC_POLICY_REFRESH_INVALIDATED_UNCHANGED;
   return METRIC_POLICY_REFRESH_UNCHANGED;
 }
 
 }  // namespace policy