Implement Invalidator::Acknowledge.

We implement this by creating a local queue of entries we've received invalidations for and then immediately acknowledging to Tango. When InvalidationHandlers acknowledge that they've finished processing for an id, we erase their entry from the queue; otherwise, we send reminder invalidations on an exponentially increasing delay.

BUG=124149

Review URL: https://chromiumcodereview.appspot.com/10911084

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

19 files changed, 498 insertions, 105 deletions

diff --git a/sync/internal_api/public/sync_manager.h b/sync/internal_api/public/sync_manager.h
index 6e23ba8..acd5394 100644
--- a/sync/internal_api/public/sync_manager.h
+++ b/sync/internal_api/public/sync_manager.h
@@ -354,6 +354,11 @@ class SYNC_EXPORT SyncManager {
   virtual void UnregisterInvalidationHandler(
       InvalidationHandler* handler) = 0;
 
+  // Forwards to the underlying notifier (see comments in invalidator.h).
+  virtual void AcknowledgeInvalidation(
+      const invalidation::ObjectId& id,
+      const syncer::AckHandle& ack_handle) = 0;
+
   // Put the syncer in normal mode ready to perform nudges and polls.
   virtual void StartSyncingNormally(
       const ModelSafeRoutingInfo& routing_info) = 0;
diff --git a/sync/internal_api/public/test/fake_sync_manager.h b/sync/internal_api/public/test/fake_sync_manager.h
index 49c443f..2d0f9de 100644
--- a/sync/internal_api/public/test/fake_sync_manager.h
+++ b/sync/internal_api/public/test/fake_sync_manager.h
@@ -102,6 +102,9 @@ class FakeSyncManager : public SyncManager {
       const ObjectIdSet& ids) OVERRIDE;
   virtual void UnregisterInvalidationHandler(
       InvalidationHandler* handler) OVERRIDE;
+  virtual void AcknowledgeInvalidation(
+      const invalidation::ObjectId& id,
+      const syncer::AckHandle& ack_handle) OVERRIDE;
   virtual void StartSyncingNormally(
       const ModelSafeRoutingInfo& routing_info) OVERRIDE;
   virtual void ConfigureSyncer(
diff --git a/sync/internal_api/sync_manager_impl.cc b/sync/internal_api/sync_manager_impl.cc
index b0d826a..e9d8d1c 100644
--- a/sync/internal_api/sync_manager_impl.cc
+++ b/sync/internal_api/sync_manager_impl.cc
@@ -635,6 +635,13 @@ void SyncManagerImpl::UnregisterInvalidationHandler(
   invalidator_->UnregisterHandler(handler);
 }
 
+void SyncManagerImpl::AcknowledgeInvalidation(
+    const invalidation::ObjectId& id, const syncer::AckHandle& ack_handle) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(initialized_);
+  invalidator_->Acknowledge(id, ack_handle);
+}
+
 void SyncManagerImpl::AddObserver(SyncManager::Observer* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   observers_.AddObserver(observer);
@@ -1238,6 +1245,16 @@ void SyncManagerImpl::OnInvalidatorStateChange(InvalidatorState state) {
 void SyncManagerImpl::OnIncomingInvalidation(
     const ObjectIdInvalidationMap& invalidation_map) {
   DCHECK(thread_checker_.CalledOnValidThread());
+
+  // TODO(dcheng): Acknowledge immediately for now. Fix this once the
+  // invalidator doesn't repeatedly ping for unacknowledged invaliations, since
+  // it conflicts with the sync scheduler's internal backoff algorithm.
+  // See http://crbug.com/124149 for more information.
+  for (ObjectIdInvalidationMap::const_iterator it = invalidation_map.begin();
+       it != invalidation_map.end(); ++it) {
+    invalidator_->Acknowledge(it->first, it->second.ack_handle);
+  }
+
   const ModelTypeInvalidationMap& type_invalidation_map =
       ObjectIdInvalidationMapToModelTypeInvalidationMap(invalidation_map);
   if (type_invalidation_map.empty()) {
diff --git a/sync/internal_api/sync_manager_impl.h b/sync/internal_api/sync_manager_impl.h
index 8baf44d..d58a1f2 100644
--- a/sync/internal_api/sync_manager_impl.h
+++ b/sync/internal_api/sync_manager_impl.h
@@ -96,6 +96,9 @@ class SYNC_EXPORT_PRIVATE SyncManagerImpl :
       const ObjectIdSet& ids) OVERRIDE;
   virtual void UnregisterInvalidationHandler(
       InvalidationHandler* handler) OVERRIDE;
+  virtual void AcknowledgeInvalidation(
+      const invalidation::ObjectId& id,
+      const syncer::AckHandle& ack_handle) OVERRIDE;
   virtual void StartSyncingNormally(
       const ModelSafeRoutingInfo& routing_info) OVERRIDE;
   virtual void ConfigureSyncer(
diff --git a/sync/internal_api/test/fake_sync_manager.cc b/sync/internal_api/test/fake_sync_manager.cc
index bc9c4dc..57c6c77 100644
--- a/sync/internal_api/test/fake_sync_manager.cc
+++ b/sync/internal_api/test/fake_sync_manager.cc
@@ -172,6 +172,11 @@ void FakeSyncManager::UnregisterInvalidationHandler(
   registrar_.UnregisterHandler(handler);
 }
 
+void FakeSyncManager::AcknowledgeInvalidation(const invalidation::ObjectId& id,
+                                              const AckHandle& ack_handle) {
+  // Do nothing.
+}
+
 void FakeSyncManager::StartSyncingNormally(
       const ModelSafeRoutingInfo& routing_info) {
   // Do nothing.
diff --git a/sync/notifier/ack_tracker.cc b/sync/notifier/ack_tracker.cc
index 759b869..6461571 100644
--- a/sync/notifier/ack_tracker.cc
+++ b/sync/notifier/ack_tracker.cc
@@ -97,6 +97,7 @@ void AckTracker::Clear() {
 
 void AckTracker::Track(const ObjectIdSet& ids) {
   DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!ids.empty());
 
   scoped_ptr<Entry> entry(new Entry(
       create_backoff_entry_callback_.Run(&kDefaultBackoffPolicy), ids));
diff --git a/sync/notifier/fake_invalidator.cc b/sync/notifier/fake_invalidator.cc
index b7cfd4d..c8216af 100644
--- a/sync/notifier/fake_invalidator.cc
+++ b/sync/notifier/fake_invalidator.cc
@@ -50,7 +50,7 @@ void FakeInvalidator::RegisterHandler(InvalidationHandler* handler) {
 }
 
 void FakeInvalidator::UpdateRegisteredIds(InvalidationHandler* handler,
-                                           const ObjectIdSet& ids) {
+                                          const ObjectIdSet& ids) {
   registrar_.UpdateRegisteredIds(handler, ids);
 }
 
@@ -58,6 +58,11 @@ void FakeInvalidator::UnregisterHandler(InvalidationHandler* handler) {
   registrar_.UnregisterHandler(handler);
 }
 
+void FakeInvalidator::Acknowledge(const invalidation::ObjectId& id,
+                                  const AckHandle& ack_handle) {
+  // Do nothing.
+}
+
 InvalidatorState FakeInvalidator::GetInvalidatorState() const {
   return registrar_.GetInvalidatorState();
 }
diff --git a/sync/notifier/fake_invalidator.h b/sync/notifier/fake_invalidator.h
index 13c65d7..e20fcc0 100644
--- a/sync/notifier/fake_invalidator.h
+++ b/sync/notifier/fake_invalidator.h
@@ -34,6 +34,8 @@ class FakeInvalidator : public Invalidator {
   virtual void UpdateRegisteredIds(InvalidationHandler* handler,
                                    const ObjectIdSet& ids) OVERRIDE;
   virtual void UnregisterHandler(InvalidationHandler* handler) OVERRIDE;
+  virtual void Acknowledge(const invalidation::ObjectId& id,
+                           const AckHandle& ack_handle) OVERRIDE;
   virtual InvalidatorState GetInvalidatorState() const OVERRIDE;
   virtual void SetUniqueId(const std::string& unique_id) OVERRIDE;
   virtual void UpdateCredentials(
diff --git a/sync/notifier/invalidation_notifier.cc b/sync/notifier/invalidation_notifier.cc
index 1b9830b..30e5df3 100644
--- a/sync/notifier/invalidation_notifier.cc
+++ b/sync/notifier/invalidation_notifier.cc
@@ -28,7 +28,7 @@ InvalidationNotifier::InvalidationNotifier(
       invalidation_state_tracker_(invalidation_state_tracker),
       client_info_(client_info),
       invalidation_bootstrap_data_(invalidation_bootstrap_data),
-      invalidation_listener_(push_client.Pass()) {
+      invalidation_listener_(&tick_clock_, push_client.Pass()) {
 }
 
 InvalidationNotifier::~InvalidationNotifier() {
@@ -52,6 +52,12 @@ void InvalidationNotifier::UnregisterHandler(InvalidationHandler* handler) {
   registrar_.UnregisterHandler(handler);
 }
 
+void InvalidationNotifier::Acknowledge(const invalidation::ObjectId& id,
+                                       const AckHandle& ack_handle) {
+  DCHECK(CalledOnValidThread());
+  invalidation_listener_.Acknowledge(id, ack_handle);
+}
+
 InvalidatorState InvalidationNotifier::GetInvalidatorState() const {
   DCHECK(CalledOnValidThread());
   return registrar_.GetInvalidatorState();
diff --git a/sync/notifier/invalidation_notifier.h b/sync/notifier/invalidation_notifier.h
index aac30dc..602ba73 100644
--- a/sync/notifier/invalidation_notifier.h
+++ b/sync/notifier/invalidation_notifier.h
@@ -7,7 +7,7 @@
 // up to the invalidation client.
 //
 // You probably don't want to use this directly; use
-// NonBlockingInvalidationNotifier.
+// NonBlockingInvalidator.
 
 #ifndef SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
 #define SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
@@ -18,6 +18,7 @@
 #include "base/compiler_specific.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/threading/non_thread_safe.h"
+#include "base/time/default_tick_clock.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/util/weak_handle.h"
@@ -55,6 +56,8 @@ class SYNC_EXPORT_PRIVATE InvalidationNotifier
   virtual void UpdateRegisteredIds(InvalidationHandler* handler,
                                    const ObjectIdSet& ids) OVERRIDE;
   virtual void UnregisterHandler(InvalidationHandler* handler) OVERRIDE;
+  virtual void Acknowledge(const invalidation::ObjectId& id,
+                           const AckHandle& ack_handle) OVERRIDE;
   virtual InvalidatorState GetInvalidatorState() const OVERRIDE;
   virtual void SetUniqueId(const std::string& unique_id) OVERRIDE;
   virtual void UpdateCredentials(
@@ -97,6 +100,10 @@ class SYNC_EXPORT_PRIVATE InvalidationNotifier
   // The initial bootstrap data to pass to |invalidation_listener_|.
   const std::string invalidation_bootstrap_data_;
 
+  // TODO(akalin): Clean up this reference to DefaultTickClock. Ideally, we
+  // should simply be using TaskRunner's tick clock. See http://crbug.com/179211
+  base::DefaultTickClock tick_clock_;
+
   // The invalidation listener.
   SyncInvalidationListener invalidation_listener_;
 
diff --git a/sync/notifier/invalidator.h b/sync/notifier/invalidator.h
index 0c2b311..7854b28 100644
--- a/sync/notifier/invalidator.h
+++ b/sync/notifier/invalidator.h
@@ -69,6 +69,10 @@ class SYNC_EXPORT Invalidator {
   // associated with |handler|.
   virtual void UnregisterHandler(InvalidationHandler* handler) = 0;
 
+  // Acknowledge that an invalidation for |id| was handled.
+  virtual void Acknowledge(const invalidation::ObjectId& id,
+                           const AckHandle& ack_handle) = 0;
+
   // Returns the current invalidator state.  When called from within
   // InvalidationHandler::OnInvalidatorStateChange(), this must return
   // the updated state.
diff --git a/sync/notifier/invalidator_registrar_unittest.cc b/sync/notifier/invalidator_registrar_unittest.cc
index f833940..fe4b409 100644
--- a/sync/notifier/invalidator_registrar_unittest.cc
+++ b/sync/notifier/invalidator_registrar_unittest.cc
@@ -43,6 +43,11 @@ class RegistrarInvalidator : public Invalidator {
     registrar_.UnregisterHandler(handler);
   }
 
+  virtual void Acknowledge(const invalidation::ObjectId& id,
+                           const AckHandle& ack_handle) OVERRIDE {
+    // Do nothing.
+  }
+
   virtual InvalidatorState GetInvalidatorState() const OVERRIDE {
     return registrar_.GetInvalidatorState();
   }
diff --git a/sync/notifier/non_blocking_invalidator.cc b/sync/notifier/non_blocking_invalidator.cc
index a306aa3..32ffd9e 100644
--- a/sync/notifier/non_blocking_invalidator.cc
+++ b/sync/notifier/non_blocking_invalidator.cc
@@ -36,6 +36,8 @@ class NonBlockingInvalidator::Core
       const std::string& client_info);
   void Teardown();
   void UpdateRegisteredIds(const ObjectIdSet& ids);
+  void Acknowledge(const invalidation::ObjectId& id,
+                   const AckHandle& ack_handle);
   void SetUniqueId(const std::string& unique_id);
   void UpdateCredentials(const std::string& email, const std::string& token);
 
@@ -102,6 +104,12 @@ void NonBlockingInvalidator::Core::UpdateRegisteredIds(const ObjectIdSet& ids) {
   invalidation_notifier_->UpdateRegisteredIds(this, ids);
 }
 
+void NonBlockingInvalidator::Core::Acknowledge(const invalidation::ObjectId& id,
+                                               const AckHandle& ack_handle) {
+  DCHECK(network_task_runner_->BelongsToCurrentThread());
+  invalidation_notifier_->Acknowledge(id, ack_handle);
+}
+
 void NonBlockingInvalidator::Core::SetUniqueId(const std::string& unique_id) {
   DCHECK(network_task_runner_->BelongsToCurrentThread());
   invalidation_notifier_->SetUniqueId(unique_id);
@@ -130,7 +138,7 @@ void NonBlockingInvalidator::Core::OnIncomingInvalidation(
 
 NonBlockingInvalidator::NonBlockingInvalidator(
     const notifier::NotifierOptions& notifier_options,
-    const InvalidationStateMap& initial_max_invalidation_versions,
+    const InvalidationStateMap& initial_invalidation_state_map,
     const std::string& invalidation_bootstrap_data,
     const WeakHandle<InvalidationStateTracker>&
         invalidation_state_tracker,
@@ -148,7 +156,7 @@ NonBlockingInvalidator::NonBlockingInvalidator(
               &NonBlockingInvalidator::Core::Initialize,
               core_.get(),
               notifier_options,
-              initial_max_invalidation_versions,
+              initial_invalidation_state_map,
               invalidation_bootstrap_data,
               invalidation_state_tracker,
               client_info))) {
@@ -190,6 +198,20 @@ void NonBlockingInvalidator::UnregisterHandler(InvalidationHandler* handler) {
   registrar_.UnregisterHandler(handler);
 }
 
+void NonBlockingInvalidator::Acknowledge(const invalidation::ObjectId& id,
+                                         const AckHandle& ack_handle) {
+  DCHECK(parent_task_runner_->BelongsToCurrentThread());
+  if (!network_task_runner_->PostTask(
+          FROM_HERE,
+          base::Bind(
+              &NonBlockingInvalidator::Core::Acknowledge,
+              core_.get(),
+              id,
+              ack_handle))) {
+    NOTREACHED();
+  }
+}
+
 InvalidatorState NonBlockingInvalidator::GetInvalidatorState() const {
   DCHECK(parent_task_runner_->BelongsToCurrentThread());
   return registrar_.GetInvalidatorState();
diff --git a/sync/notifier/non_blocking_invalidator.h b/sync/notifier/non_blocking_invalidator.h
index 0908b48..0da2e4c1 100644
--- a/sync/notifier/non_blocking_invalidator.h
+++ b/sync/notifier/non_blocking_invalidator.h
@@ -51,6 +51,8 @@ class SYNC_EXPORT_PRIVATE NonBlockingInvalidator
   virtual void UpdateRegisteredIds(InvalidationHandler* handler,
                                    const ObjectIdSet& ids) OVERRIDE;
   virtual void UnregisterHandler(InvalidationHandler* handler) OVERRIDE;
+  virtual void Acknowledge(const invalidation::ObjectId& id,
+                           const AckHandle& ack_handle) OVERRIDE;
   virtual InvalidatorState GetInvalidatorState() const OVERRIDE;
   virtual void SetUniqueId(const std::string& unique_id) OVERRIDE;
   virtual void UpdateCredentials(
diff --git a/sync/notifier/p2p_invalidator.cc b/sync/notifier/p2p_invalidator.cc
index 5894391..e71d1e3c 100644
--- a/sync/notifier/p2p_invalidator.cc
+++ b/sync/notifier/p2p_invalidator.cc
@@ -178,6 +178,12 @@ void P2PInvalidator::UnregisterHandler(InvalidationHandler* handler) {
   registrar_.UnregisterHandler(handler);
 }
 
+void P2PInvalidator::Acknowledge(const invalidation::ObjectId& id,
+                                 const AckHandle& ack_handle) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  // Do nothing for the P2P implementation.
+}
+
 InvalidatorState P2PInvalidator::GetInvalidatorState() const {
   DCHECK(thread_checker_.CalledOnValidThread());
   return registrar_.GetInvalidatorState();
diff --git a/sync/notifier/p2p_invalidator.h b/sync/notifier/p2p_invalidator.h
index cfb9644..952299f 100644
--- a/sync/notifier/p2p_invalidator.h
+++ b/sync/notifier/p2p_invalidator.h
@@ -102,6 +102,8 @@ class SYNC_EXPORT_PRIVATE P2PInvalidator
   virtual void UpdateRegisteredIds(InvalidationHandler* handler,
                                    const ObjectIdSet& ids) OVERRIDE;
   virtual void UnregisterHandler(InvalidationHandler* handler) OVERRIDE;
+  virtual void Acknowledge(const invalidation::ObjectId& id,
+                           const AckHandle& ack_handle) OVERRIDE;
   virtual InvalidatorState GetInvalidatorState() const OVERRIDE;
   virtual void SetUniqueId(const std::string& unique_id) OVERRIDE;
   virtual void UpdateCredentials(
diff --git a/sync/notifier/sync_invalidation_listener.cc b/sync/notifier/sync_invalidation_listener.cc
index 4e57bb8..28b37bc 100644
--- a/sync/notifier/sync_invalidation_listener.cc
+++ b/sync/notifier/sync_invalidation_listener.cc
@@ -4,9 +4,9 @@
 
 #include "sync/notifier/sync_invalidation_listener.h"
 
-#include <string>
 #include <vector>
 
+#include "base/bind.h"
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/logging.h"
@@ -29,8 +29,11 @@ namespace syncer {
 SyncInvalidationListener::Delegate::~Delegate() {}
 
 SyncInvalidationListener::SyncInvalidationListener(
+    base::TickClock* tick_clock,
     scoped_ptr<notifier::PushClient> push_client)
-    : push_client_(push_client.get()),
+    : weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
+      ack_tracker_(tick_clock, ALLOW_THIS_IN_INITIALIZER_LIST(this)),
+      push_client_(push_client.get()),
       sync_system_resources_(push_client.Pass(),
                              ALLOW_THIS_IN_INITIALIZER_LIST(this)),
       delegate_(NULL),
@@ -105,6 +108,19 @@ void SyncInvalidationListener::Start(
       FROM_HERE,
       &InvalidationStateTracker::SetInvalidatorClientId,
       client_id);
+
+  // Set up reminders for any invalidations that have not been locally
+  // acknowledged.
+  ObjectIdSet unacknowledged_ids;
+  for (InvalidationStateMap::const_iterator it =
+           invalidation_state_map_.begin();
+       it != invalidation_state_map_.end(); ++it) {
+    if (it->second.expected.Equals(it->second.current))
+      continue;
+    unacknowledged_ids.insert(it->first);
+  }
+  if (!unacknowledged_ids.empty())
+    ack_tracker_.Track(unacknowledged_ids);
 }
 
 void SyncInvalidationListener::UpdateCredentials(
@@ -124,6 +140,27 @@ void SyncInvalidationListener::UpdateRegisteredIds(const ObjectIdSet& ids) {
   }
 }
 
+void SyncInvalidationListener::Acknowledge(const invalidation::ObjectId& id,
+                                           const AckHandle& ack_handle) {
+  DCHECK(CalledOnValidThread());
+  InvalidationStateMap::iterator state_it = invalidation_state_map_.find(id);
+  if (state_it == invalidation_state_map_.end())
+    return;
+  invalidation_state_tracker_.Call(
+      FROM_HERE,
+      &InvalidationStateTracker::Acknowledge,
+      id,
+      ack_handle);
+  state_it->second.current = ack_handle;
+  if (state_it->second.expected.Equals(ack_handle)) {
+    // If the received ack matches the expected ack, then we no longer need to
+    // keep track of |id| since it is up-to-date.
+    ObjectIdSet ids;
+    ids.insert(id);
+    ack_tracker_.Ack(ids);
+  }
+}
+
 void SyncInvalidationListener::Ready(
     invalidation::InvalidationClient* client) {
   DCHECK(CalledOnValidThread());
@@ -167,17 +204,15 @@ void SyncInvalidationListener::Invalidate(
   DVLOG(2) << "Setting max invalidation version for " << ObjectIdToString(id)
            << " to " << invalidation.version();
   invalidation_state_map_[id].version = invalidation.version();
+  invalidation_state_map_[id].payload = payload;
   invalidation_state_tracker_.Call(
       FROM_HERE,
       &InvalidationStateTracker::SetMaxVersionAndPayload,
       id, invalidation.version(), payload);
 
-  ObjectIdInvalidationMap invalidation_map;
-  invalidation_map[id].payload = payload;
-  EmitInvalidation(invalidation_map);
-  // TODO(akalin): We should really acknowledge only after we get the
-  // updates from the sync server. (see http://crbug.com/78462).
-  client->Acknowledge(ack_handle);
+  ObjectIdSet ids;
+  ids.insert(id);
+  PrepareInvalidation(ids, payload, client, ack_handle);
 }
 
 void SyncInvalidationListener::InvalidateUnknownVersion(
@@ -188,12 +223,9 @@ void SyncInvalidationListener::InvalidateUnknownVersion(
   DCHECK_EQ(client, invalidation_client_.get());
   DVLOG(1) << "InvalidateUnknownVersion";
 
-  ObjectIdInvalidationMap invalidation_map;
-  invalidation_map[object_id].payload = std::string();
-  EmitInvalidation(invalidation_map);
-  // TODO(akalin): We should really acknowledge only after we get the
-  // updates from the sync server. (see http://crbug.com/78462).
-  client->Acknowledge(ack_handle);
+  ObjectIdSet ids;
+  ids.insert(object_id);
+  PrepareInvalidation(ids, std::string(), client, ack_handle);
 }
 
 // This should behave as if we got an invalidation with version
@@ -205,17 +237,60 @@ void SyncInvalidationListener::InvalidateAll(
   DCHECK_EQ(client, invalidation_client_.get());
   DVLOG(1) << "InvalidateAll";
 
-  const ObjectIdInvalidationMap& invalidation_map =
-      ObjectIdSetToInvalidationMap(registered_ids_, std::string());
-  EmitInvalidation(invalidation_map);
-  // TODO(akalin): We should really acknowledge only after we get the
-  // updates from the sync server. (see http://crbug.com/76482).
-  client->Acknowledge(ack_handle);
+  PrepareInvalidation(registered_ids_, std::string(), client, ack_handle);
+}
+
+void SyncInvalidationListener::PrepareInvalidation(
+    const ObjectIdSet& ids,
+    const std::string& payload,
+    invalidation::InvalidationClient* client,
+    const invalidation::AckHandle& ack_handle) {
+  DCHECK(CalledOnValidThread());
+
+  // A server invalidation resets the local retry count.
+  ack_tracker_.Ack(ids);
+  invalidation_state_tracker_.Call(
+      FROM_HERE,
+      &InvalidationStateTracker::GenerateAckHandles,
+      ids,
+      base::MessageLoopProxy::current(),
+      base::Bind(&SyncInvalidationListener::EmitInvalidation,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 ids,
+                 payload,
+                 client,
+                 ack_handle));
 }
 
 void SyncInvalidationListener::EmitInvalidation(
-    const ObjectIdInvalidationMap& invalidation_map) {
+    const ObjectIdSet& ids,
+    const std::string& payload,
+    invalidation::InvalidationClient* client,
+    const invalidation::AckHandle& ack_handle,
+    const AckHandleMap& local_ack_handles) {
   DCHECK(CalledOnValidThread());
+  ObjectIdInvalidationMap invalidation_map =
+      ObjectIdSetToInvalidationMap(ids, payload);
+  for (AckHandleMap::const_iterator it = local_ack_handles.begin();
+       it != local_ack_handles.end(); ++it) {
+    // Update in-memory copy of the invalidation state.
+    invalidation_state_map_[it->first].expected = it->second;
+    invalidation_map[it->first].ack_handle = it->second;
+  }
+  ack_tracker_.Track(ids);
+  delegate_->OnInvalidate(invalidation_map);
+  client->Acknowledge(ack_handle);
+}
+
+void SyncInvalidationListener::OnTimeout(const ObjectIdSet& ids) {
+  ObjectIdInvalidationMap invalidation_map;
+  for (ObjectIdSet::const_iterator it = ids.begin(); it != ids.end(); ++it) {
+    Invalidation invalidation;
+    invalidation.ack_handle = invalidation_state_map_[*it].expected;
+    invalidation.payload = invalidation_state_map_[*it].payload;
+    invalidation_map.insert(std::make_pair(*it, invalidation));
+  }
+
   delegate_->OnInvalidate(invalidation_map);
 }
 
@@ -306,6 +381,7 @@ void SyncInvalidationListener::DoRegistrationUpdate() {
   }
   invalidation_state_tracker_.Call(
       FROM_HERE, &InvalidationStateTracker::Forget, unregistered_ids);
+  ack_tracker_.Ack(unregistered_ids);
 }
 
 void SyncInvalidationListener::StopForTest() {
@@ -318,12 +394,18 @@ InvalidationStateMap SyncInvalidationListener::GetStateMapForTest() const {
   return invalidation_state_map_;
 }
 
+AckTracker* SyncInvalidationListener::GetAckTrackerForTest() {
+  return &ack_tracker_;
+}
+
 void SyncInvalidationListener::Stop() {
   DCHECK(CalledOnValidThread());
   if (!invalidation_client_.get()) {
     return;
   }
 
+  ack_tracker_.Clear();
+
   registration_manager_.reset();
   sync_system_resources_.Stop();
   invalidation_client_->Stop();
diff --git a/sync/notifier/sync_invalidation_listener.h b/sync/notifier/sync_invalidation_listener.h
index 345a671..3613551 100644
--- a/sync/notifier/sync_invalidation_listener.h
+++ b/sync/notifier/sync_invalidation_listener.h
@@ -20,12 +20,17 @@
 #include "jingle/notifier/listener/push_client_observer.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/util/weak_handle.h"
+#include "sync/notifier/ack_tracker.h"
 #include "sync/notifier/invalidation_state_tracker.h"
 #include "sync/notifier/invalidator_state.h"
 #include "sync/notifier/object_id_invalidation_map.h"
 #include "sync/notifier/state_writer.h"
 #include "sync/notifier/sync_system_resources.h"
 
+namespace base {
+class TickClock;
+}  // namespace base
+
 namespace buzz {
 class XmppTaskParentInterface;
 }  // namespace buzz
@@ -44,7 +49,8 @@ class SYNC_EXPORT_PRIVATE SyncInvalidationListener
     : public NON_EXPORTED_BASE(invalidation::InvalidationListener),
       public StateWriter,
       public NON_EXPORTED_BASE(notifier::PushClientObserver),
-      public base::NonThreadSafe {
+      public base::NonThreadSafe,
+      public AckTracker::Delegate {
  public:
   typedef base::Callback<invalidation::InvalidationClient*(
       invalidation::SystemResources*,
@@ -64,6 +70,7 @@ class SYNC_EXPORT_PRIVATE SyncInvalidationListener
   };
 
   explicit SyncInvalidationListener(
+      base::TickClock* tick_clock,
       scoped_ptr<notifier::PushClient> push_client);
 
   // Calls Stop().
@@ -85,6 +92,9 @@ class SYNC_EXPORT_PRIVATE SyncInvalidationListener
   // Update the set of object IDs that we're interested in getting
   // notifications for.  May be called at any time.
   void UpdateRegisteredIds(const ObjectIdSet& ids);
+  // Acknowledge that an invalidation for |id| was handled.
+  void Acknowledge(const invalidation::ObjectId& id,
+                   const AckHandle& ack_handle);
 
   // invalidation::InvalidationListener implementation.
   virtual void Ready(
@@ -131,6 +141,7 @@ class SYNC_EXPORT_PRIVATE SyncInvalidationListener
 
   void StopForTest();
   InvalidationStateMap GetStateMapForTest() const;
+  AckTracker* GetAckTrackerForTest();
 
  private:
   void Stop();
@@ -139,7 +150,21 @@ class SYNC_EXPORT_PRIVATE SyncInvalidationListener
 
   void EmitStateChange();
 
-  void EmitInvalidation(const ObjectIdInvalidationMap& invalidation_map);
+  void PrepareInvalidation(const ObjectIdSet& ids,
+                           const std::string& payload,
+                           invalidation::InvalidationClient* client,
+                           const invalidation::AckHandle& ack_handle);
+  void EmitInvalidation(const ObjectIdSet& ids,
+                        const std::string& payload,
+                        invalidation::InvalidationClient* client,
+                        const invalidation::AckHandle& ack_handle,
+                        const AckHandleMap& local_ack_handles);
+
+  // AckTracker::Delegate implementation.
+  virtual void OnTimeout(const ObjectIdSet& ids) OVERRIDE;
+
+  base::WeakPtrFactory<SyncInvalidationListener> weak_ptr_factory_;
+  AckTracker ack_tracker_;
 
   // Owned by |sync_system_resources_|.
   notifier::PushClient* const push_client_;
diff --git a/sync/notifier/sync_invalidation_listener_unittest.cc b/sync/notifier/sync_invalidation_listener_unittest.cc
index 41fc86d..44c1725 100644
--- a/sync/notifier/sync_invalidation_listener_unittest.cc
+++ b/sync/notifier/sync_invalidation_listener_unittest.cc
@@ -8,10 +8,15 @@
 
 #include "base/compiler_specific.h"
 #include "base/message_loop.h"
+#include "base/stl_util.h"
+#include "base/time.h"
+#include "base/time/tick_clock.h"
 #include "google/cacheinvalidation/include/invalidation-client.h"
 #include "google/cacheinvalidation/include/types.h"
 #include "jingle/notifier/listener/fake_push_client.h"
+#include "sync/internal_api/public/base/invalidation_test_util.h"
 #include "sync/internal_api/public/util/weak_handle.h"
+#include "sync/notifier/ack_tracker.h"
 #include "sync/notifier/fake_invalidation_state_tracker.h"
 #include "sync/notifier/invalidation_util.h"
 #include "sync/notifier/sync_invalidation_listener.h"
@@ -131,7 +136,9 @@ class FakeInvalidationClient : public invalidation::InvalidationClient {
 // and state.
 class FakeDelegate : public SyncInvalidationListener::Delegate {
  public:
-  FakeDelegate() : state_(TRANSIENT_INVALIDATION_ERROR) {}
+  explicit FakeDelegate(SyncInvalidationListener* listener)
+      : listener_(listener),
+        state_(TRANSIENT_INVALIDATION_ERROR) {}
   virtual ~FakeDelegate() {}
 
   int GetInvalidationCount(const ObjectId& id) const {
@@ -148,6 +155,10 @@ class FakeDelegate : public SyncInvalidationListener::Delegate {
     return state_;
   }
 
+  void Acknowledge(const ObjectId& id) {
+    listener_->Acknowledge(id, invalidations_[id].ack_handle);
+  }
+
   // SyncInvalidationListener::Delegate implementation.
 
   virtual void OnInvalidate(
@@ -167,6 +178,7 @@ class FakeDelegate : public SyncInvalidationListener::Delegate {
   typedef std::map<ObjectId, int, ObjectIdLessThan> ObjectIdCountMap;
   ObjectIdCountMap invalidation_counts_;
   ObjectIdInvalidationMap invalidations_;
+  SyncInvalidationListener* listener_;
   InvalidatorState state_;
 };
 
@@ -181,6 +193,50 @@ invalidation::InvalidationClient* CreateFakeInvalidationClient(
   return *fake_invalidation_client;
 }
 
+// TODO(dcheng): FakeTickClock and FakeBackoffEntry ought to be factored out
+// into a helpers file so it can be shared with the AckTracker unittest.
+class FakeTickClock : public base::TickClock {
+ public:
+  FakeTickClock() {}
+  virtual ~FakeTickClock() {}
+
+  void LeapForward(int seconds) {
+    ASSERT_GT(seconds, 0);
+    fake_now_ticks_ += base::TimeDelta::FromSeconds(seconds);
+  }
+
+  virtual base::TimeTicks NowTicks() OVERRIDE {
+    return fake_now_ticks_;
+  }
+
+ private:
+  base::TimeTicks fake_now_ticks_;
+
+  DISALLOW_COPY_AND_ASSIGN(FakeTickClock);
+};
+
+class FakeBackoffEntry : public net::BackoffEntry {
+ public:
+  FakeBackoffEntry(const Policy *const policy, base::TickClock* tick_clock)
+      : BackoffEntry(policy), tick_clock_(tick_clock) {
+  }
+
+ protected:
+  virtual base::TimeTicks ImplGetTimeNow() const OVERRIDE {
+    return tick_clock_->NowTicks();
+  }
+
+ private:
+  base::TickClock* const tick_clock_;
+};
+
+scoped_ptr<net::BackoffEntry> CreateMockEntry(
+    base::TickClock* tick_clock,
+    const net::BackoffEntry::Policy *const policy) {
+  return scoped_ptr<net::BackoffEntry>(
+      new FakeBackoffEntry(policy, tick_clock));
+}
+
 class SyncInvalidationListenerTest : public testing::Test {
  protected:
   SyncInvalidationListenerTest()
@@ -190,14 +246,16 @@ class SyncInvalidationListenerTest : public testing::Test {
         kAppsId_(kChromeSyncSourceId, "APP"),
         fake_push_client_(new notifier::FakePushClient()),
         fake_invalidation_client_(NULL),
-        client_(scoped_ptr<notifier::PushClient>(fake_push_client_)) {}
+        listener_(&tick_clock_,
+                  scoped_ptr<notifier::PushClient>(fake_push_client_)),
+        fake_delegate_(&listener_) {}
 
   virtual void SetUp() {
     StartClient();
 
     registered_ids_.insert(kBookmarksId_);
     registered_ids_.insert(kPreferencesId_);
-    client_.UpdateRegisteredIds(registered_ids_);
+    listener_.UpdateRegisteredIds(registered_ids_);
   }
 
   virtual void TearDown() {
@@ -210,6 +268,34 @@ class SyncInvalidationListenerTest : public testing::Test {
     StartClient();
   }
 
+  void StartClient() {
+    fake_invalidation_client_ = NULL;
+    listener_.Start(base::Bind(&CreateFakeInvalidationClient,
+                               &fake_invalidation_client_),
+                    kClientId, kClientInfo, kState,
+                    fake_tracker_.GetAllInvalidationStates(),
+                    MakeWeakHandle(fake_tracker_.AsWeakPtr()),
+                    &fake_delegate_);
+    DCHECK(fake_invalidation_client_);
+
+    // TODO(rlarocque): This is necessary for the deferred write of the client
+    // ID to take place.  We can remove this statement when we remove the
+    // WriteInvalidatorClientId test.  See crbug.com/124142.
+    message_loop_.RunUntilIdle();
+  }
+
+  void StopClient() {
+    // listener_.StopForTest() stops the invalidation scheduler, which
+    // deletes any pending tasks without running them.  Some tasks
+    // "run and delete" another task, so they must be run in order to
+    // avoid leaking the inner task.  listener_.StopForTest() does not
+    // schedule any tasks, so it's both necessary and sufficient to
+    // drain the task queue before calling it.
+    message_loop_.RunUntilIdle();
+    fake_invalidation_client_ = NULL;
+    listener_.StopForTest();
+  }
+
   int GetInvalidationCount(const ObjectId& id) const {
     return fake_delegate_.GetInvalidationCount(id);
   }
@@ -249,31 +335,37 @@ class SyncInvalidationListenerTest : public testing::Test {
     }
     const AckHandle ack_handle("fakedata");
     fake_invalidation_client_->ClearAckedHandles();
-    client_.Invalidate(fake_invalidation_client_, inv, ack_handle);
-    EXPECT_TRUE(fake_invalidation_client_->IsAckedHandle(ack_handle));
-    // Pump message loop to trigger
-    // InvalidationStateTracker::SetMaxVersion().
+    listener_.Invalidate(fake_invalidation_client_, inv, ack_handle);
+    // Pump message loop to trigger InvalidationStateTracker::SetMaxVersion()
+    // and callback from InvalidationStateTracker::GenerateAckHandles().
     message_loop_.RunUntilIdle();
+    EXPECT_TRUE(fake_invalidation_client_->IsAckedHandle(ack_handle));
   }
 
   // |payload| can be NULL, but not |type_name|.
   void FireInvalidateUnknownVersion(const ObjectId& object_id) {
     const AckHandle ack_handle("fakedata_unknown");
     fake_invalidation_client_->ClearAckedHandles();
-    client_.InvalidateUnknownVersion(fake_invalidation_client_, object_id,
+    listener_.InvalidateUnknownVersion(fake_invalidation_client_, object_id,
                                      ack_handle);
+    // Pump message loop to trigger callback from
+    // InvalidationStateTracker::GenerateAckHandles().
+    message_loop_.RunUntilIdle();
     EXPECT_TRUE(fake_invalidation_client_->IsAckedHandle(ack_handle));
   }
 
   void FireInvalidateAll() {
     const AckHandle ack_handle("fakedata_all");
     fake_invalidation_client_->ClearAckedHandles();
-    client_.InvalidateAll(fake_invalidation_client_, ack_handle);
+    listener_.InvalidateAll(fake_invalidation_client_, ack_handle);
+    // Pump message loop to trigger callback from
+    // InvalidationStateTracker::GenerateAckHandles().
+    message_loop_.RunUntilIdle();
     EXPECT_TRUE(fake_invalidation_client_->IsAckedHandle(ack_handle));
   }
 
   void WriteState(const std::string& new_state) {
-    client_.WriteState(new_state);
+    listener_.WriteState(new_state);
     // Pump message loop to trigger
     // InvalidationStateTracker::WriteState().
     message_loop_.RunUntilIdle();
@@ -287,6 +379,29 @@ class SyncInvalidationListenerTest : public testing::Test {
     fake_push_client_->DisableNotifications(reason);
   }
 
+  void VerifyUnacknowledged(const ObjectId& object_id) {
+    InvalidationStateMap state_map = fake_tracker_.GetAllInvalidationStates();
+    EXPECT_THAT(state_map[object_id].current,
+                Not(Eq(state_map[object_id].expected)));
+    EXPECT_EQ(listener_.GetStateMapForTest(), state_map);
+  }
+
+  void VerifyAcknowledged(const ObjectId& object_id) {
+    InvalidationStateMap state_map = fake_tracker_.GetAllInvalidationStates();
+    EXPECT_THAT(state_map[object_id].current,
+                Eq(state_map[object_id].expected));
+    EXPECT_EQ(listener_.GetStateMapForTest(), state_map);
+  }
+
+  void AcknowledgeAndVerify(const ObjectId& object_id) {
+    VerifyUnacknowledged(object_id);
+    fake_delegate_.Acknowledge(object_id);
+    // Pump message loop to trigger
+    // InvalidationStateTracker::Acknowledge().
+    message_loop_.RunUntilIdle();
+    VerifyAcknowledged(object_id);
+  }
+
   const ObjectId kBookmarksId_;
   const ObjectId kPreferencesId_;
   const ObjectId kExtensionsId_;
@@ -295,44 +410,18 @@ class SyncInvalidationListenerTest : public testing::Test {
   ObjectIdSet registered_ids_;
 
  private:
-  void StartClient() {
-    fake_invalidation_client_ = NULL;
-    client_.Start(base::Bind(&CreateFakeInvalidationClient,
-                             &fake_invalidation_client_),
-                  kClientId, kClientInfo, kState,
-                  fake_tracker_.GetAllInvalidationStates(),
-                  MakeWeakHandle(fake_tracker_.AsWeakPtr()),
-                  &fake_delegate_);
-    DCHECK(fake_invalidation_client_);
-
-    // TODO(rlarocque): This is necessary for the deferred write of the client
-    // ID to take place.  We can remove this statement when we remove the
-    // WriteInvalidatorClientId test.  See crbug.com/124142.
-    message_loop_.RunUntilIdle();
-  }
-
-  void StopClient() {
-    // client_.StopForTest() stops the invalidation scheduler, which
-    // deletes any pending tasks without running them.  Some tasks
-    // "run and delete" another task, so they must be run in order to
-    // avoid leaking the inner task.  client_.StopForTest() does not
-    // schedule any tasks, so it's both necessary and sufficient to
-    // drain the task queue before calling it.
-    message_loop_.RunUntilIdle();
-    fake_invalidation_client_ = NULL;
-    client_.StopForTest();
-  }
-
   MessageLoop message_loop_;
-
-  FakeDelegate fake_delegate_;
   FakeInvalidationStateTracker fake_tracker_;
   notifier::FakePushClient* const fake_push_client_;
 
  protected:
   // Tests need to access these directly.
   FakeInvalidationClient* fake_invalidation_client_;
-  SyncInvalidationListener client_;
+  FakeTickClock tick_clock_;
+  SyncInvalidationListener listener_;
+
+ private:
+  FakeDelegate fake_delegate_;
 };
 
 // Verify the client ID is written to the state tracker on client start.
@@ -363,6 +452,7 @@ TEST_F(SyncInvalidationListenerTest, InvalidateNoPayload) {
   EXPECT_EQ(1, GetInvalidationCount(id));
   EXPECT_EQ("", GetPayload(id));
   EXPECT_EQ(kVersion1, GetMaxVersion(id));
+  AcknowledgeAndVerify(id);
 }
 
 // Fire an invalidation with an empty payload.  It should be
@@ -376,6 +466,7 @@ TEST_F(SyncInvalidationListenerTest, InvalidateEmptyPayload) {
   EXPECT_EQ(1, GetInvalidationCount(id));
   EXPECT_EQ("", GetPayload(id));
   EXPECT_EQ(kVersion1, GetMaxVersion(id));
+  AcknowledgeAndVerify(id);
 }
 
 // Fire an invalidation with a payload.  It should be processed, and
@@ -388,11 +479,12 @@ TEST_F(SyncInvalidationListenerTest, InvalidateWithPayload) {
   EXPECT_EQ(1, GetInvalidationCount(id));
   EXPECT_EQ(kPayload1, GetPayload(id));
   EXPECT_EQ(kVersion1, GetMaxVersion(id));
+  AcknowledgeAndVerify(id);
 }
 
-// Fire an invalidation with a payload.  It should still be processed,
-// and both the payload and the version should be updated.
-TEST_F(SyncInvalidationListenerTest, InvalidateUnregistered) {
+// Fire an invalidation for an unregistered object ID with a payload.  It should
+// still be processed, and both the payload and the version should be updated.
+TEST_F(SyncInvalidationListenerTest, InvalidateUnregisteredWithPayload) {
   const ObjectId kUnregisteredId(
       kChromeSyncSourceId, "unregistered");
   const ObjectId& id = kUnregisteredId;
@@ -401,11 +493,12 @@ TEST_F(SyncInvalidationListenerTest, InvalidateUnregistered) {
   EXPECT_EQ("", GetPayload(id));
   EXPECT_EQ(kMinVersion, GetMaxVersion(id));
 
-  FireInvalidate(id, kVersion1, NULL);
+  FireInvalidate(id, kVersion1, "unregistered payload");
 
   EXPECT_EQ(1, GetInvalidationCount(id));
-  EXPECT_EQ("", GetPayload(id));
+  EXPECT_EQ("unregistered payload", GetPayload(id));
   EXPECT_EQ(kVersion1, GetMaxVersion(id));
+  AcknowledgeAndVerify(id);
 }
 
 // Fire an invalidation, then fire another one with a lower version.
@@ -419,12 +512,14 @@ TEST_F(SyncInvalidationListenerTest, InvalidateVersion) {
   EXPECT_EQ(1, GetInvalidationCount(id));
   EXPECT_EQ(kPayload2, GetPayload(id));
   EXPECT_EQ(kVersion2, GetMaxVersion(id));
+  AcknowledgeAndVerify(id);
 
   FireInvalidate(id, kVersion1, kPayload1);
 
   EXPECT_EQ(1, GetInvalidationCount(id));
   EXPECT_EQ(kPayload2, GetPayload(id));
   EXPECT_EQ(kVersion2, GetMaxVersion(id));
+  VerifyAcknowledged(id);
 }
 
 // Fire an invalidation with an unknown version twice.  It shouldn't
@@ -438,12 +533,14 @@ TEST_F(SyncInvalidationListenerTest, InvalidateUnknownVersion) {
   EXPECT_EQ(1, GetInvalidationCount(id));
   EXPECT_EQ("", GetPayload(id));
   EXPECT_EQ(kMinVersion, GetMaxVersion(id));
+  AcknowledgeAndVerify(id);
 
   FireInvalidateUnknownVersion(id);
 
   EXPECT_EQ(2, GetInvalidationCount(id));
   EXPECT_EQ("", GetPayload(id));
   EXPECT_EQ(kMinVersion, GetMaxVersion(id));
+  AcknowledgeAndVerify(id);
 }
 
 // Fire an invalidation for all enabled IDs.  It shouldn't update the
@@ -456,6 +553,7 @@ TEST_F(SyncInvalidationListenerTest, InvalidateAll) {
     EXPECT_EQ(1, GetInvalidationCount(*it));
     EXPECT_EQ("", GetPayload(*it));
     EXPECT_EQ(kMinVersion, GetMaxVersion(*it));
+    AcknowledgeAndVerify(*it);
   }
 }
 
@@ -466,12 +564,14 @@ TEST_F(SyncInvalidationListenerTest, InvalidateMultipleIds) {
   EXPECT_EQ(1, GetInvalidationCount(kBookmarksId_));
   EXPECT_EQ("", GetPayload(kBookmarksId_));
   EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+  AcknowledgeAndVerify(kBookmarksId_);
 
   FireInvalidate(kExtensionsId_, 2, NULL);
 
   EXPECT_EQ(1, GetInvalidationCount(kExtensionsId_));
   EXPECT_EQ("", GetPayload(kExtensionsId_));
   EXPECT_EQ(2, GetMaxVersion(kExtensionsId_));
+  AcknowledgeAndVerify(kExtensionsId_);
 
   // Invalidations with lower version numbers should be ignored.
 
@@ -494,14 +594,19 @@ TEST_F(SyncInvalidationListenerTest, InvalidateMultipleIds) {
   EXPECT_EQ(2, GetInvalidationCount(kBookmarksId_));
   EXPECT_EQ("", GetPayload(kBookmarksId_));
   EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+  AcknowledgeAndVerify(kBookmarksId_);
 
   EXPECT_EQ(1, GetInvalidationCount(kPreferencesId_));
   EXPECT_EQ("", GetPayload(kPreferencesId_));
   EXPECT_EQ(kMinVersion, GetMaxVersion(kPreferencesId_));
+  AcknowledgeAndVerify(kPreferencesId_);
 
+  // Note that kExtensionsId_ is not registered, so InvalidateAll() shouldn't
+  // affect it.
   EXPECT_EQ(1, GetInvalidationCount(kExtensionsId_));
   EXPECT_EQ("", GetPayload(kExtensionsId_));
   EXPECT_EQ(2, GetMaxVersion(kExtensionsId_));
+  VerifyAcknowledged(kExtensionsId_);
 
   // Invalidations with higher version numbers should be processed.
 
@@ -509,16 +614,101 @@ TEST_F(SyncInvalidationListenerTest, InvalidateMultipleIds) {
   EXPECT_EQ(2, GetInvalidationCount(kPreferencesId_));
   EXPECT_EQ("", GetPayload(kPreferencesId_));
   EXPECT_EQ(5, GetMaxVersion(kPreferencesId_));
+  AcknowledgeAndVerify(kPreferencesId_);
 
   FireInvalidate(kExtensionsId_, 3, NULL);
   EXPECT_EQ(2, GetInvalidationCount(kExtensionsId_));
   EXPECT_EQ("", GetPayload(kExtensionsId_));
   EXPECT_EQ(3, GetMaxVersion(kExtensionsId_));
+  AcknowledgeAndVerify(kExtensionsId_);
 
   FireInvalidate(kBookmarksId_, 4, NULL);
   EXPECT_EQ(3, GetInvalidationCount(kBookmarksId_));
   EXPECT_EQ("", GetPayload(kBookmarksId_));
   EXPECT_EQ(4, GetMaxVersion(kBookmarksId_));
+  AcknowledgeAndVerify(kBookmarksId_);
+}
+
+// Various tests for the local invalidation feature.
+// Tests a "normal" scenario. We allow one timeout period to expire by sending
+// ack handles that are not the "latest" ack handle. Once the timeout expires,
+// we verify that we get a second callback and then acknowledge it. Once
+// acknowledged, no further timeouts should occur.
+TEST_F(SyncInvalidationListenerTest, InvalidateOneTimeout) {
+  listener_.GetAckTrackerForTest()->SetCreateBackoffEntryCallbackForTest(
+      base::Bind(&CreateMockEntry, &tick_clock_));
+
+  // Trigger the initial invalidation.
+  FireInvalidate(kBookmarksId_, 3, NULL);
+  EXPECT_EQ(1, GetInvalidationCount(kBookmarksId_));
+  EXPECT_EQ("", GetPayload(kBookmarksId_));
+  EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+  VerifyUnacknowledged(kBookmarksId_);
+
+  // Trigger one timeout.
+  tick_clock_.LeapForward(60);
+  EXPECT_TRUE(listener_.GetAckTrackerForTest()->TriggerTimeoutAtForTest(
+      tick_clock_.NowTicks()));
+  EXPECT_EQ(2, GetInvalidationCount(kBookmarksId_));
+  // Other properties should remain the same.
+  EXPECT_EQ("", GetPayload(kBookmarksId_));
+  EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+
+  AcknowledgeAndVerify(kBookmarksId_);
+
+  // No more invalidations should remain in the queue.
+  EXPECT_TRUE(listener_.GetAckTrackerForTest()->IsQueueEmptyForTest());
+}
+
+// Test that an unacknowledged invalidation triggers reminders if the listener
+// is restarted.
+TEST_F(SyncInvalidationListenerTest, InvalidationTimeoutRestart) {
+  listener_.GetAckTrackerForTest()->SetCreateBackoffEntryCallbackForTest(
+      base::Bind(&CreateMockEntry, &tick_clock_));
+
+  FireInvalidate(kBookmarksId_, 3, NULL);
+  EXPECT_EQ(1, GetInvalidationCount(kBookmarksId_));
+  EXPECT_EQ("", GetPayload(kBookmarksId_));
+  EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+
+  // Trigger one timeout.
+  tick_clock_.LeapForward(60);
+  EXPECT_TRUE(listener_.GetAckTrackerForTest()->TriggerTimeoutAtForTest(
+      tick_clock_.NowTicks()));
+  EXPECT_EQ(2, GetInvalidationCount(kBookmarksId_));
+  // Other properties should remain the same.
+  EXPECT_EQ("", GetPayload(kBookmarksId_));
+  EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+
+  // Restarting the client should reset the retry count and the timeout period
+  // (e.g. it shouldn't increase to 120 seconds). Skip ahead 1200 seconds to be
+  // on the safe side.
+  StopClient();
+  tick_clock_.LeapForward(1200);
+  StartClient();
+
+  // The bookmark invalidation state should not have changed.
+  EXPECT_EQ(2, GetInvalidationCount(kBookmarksId_));
+  EXPECT_EQ("", GetPayload(kBookmarksId_));
+  EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+
+  // Now trigger the invalidation reminder after the client restarts.
+  tick_clock_.LeapForward(60);
+  EXPECT_TRUE(listener_.GetAckTrackerForTest()->TriggerTimeoutAtForTest(
+      tick_clock_.NowTicks()));
+  EXPECT_EQ(3, GetInvalidationCount(kBookmarksId_));
+  // Other properties should remain the same.
+  EXPECT_EQ("", GetPayload(kBookmarksId_));
+  EXPECT_EQ(3, GetMaxVersion(kBookmarksId_));
+
+  AcknowledgeAndVerify(kBookmarksId_);
+
+  // No more invalidations should remain in the queue.
+  EXPECT_TRUE(listener_.GetAckTrackerForTest()->IsQueueEmptyForTest());
+
+  // The queue should remain empty when we restart now.
+  RestartClient();
+  EXPECT_TRUE(listener_.GetAckTrackerForTest()->IsQueueEmptyForTest());
 }
 
 // Registration tests.
@@ -533,7 +723,7 @@ TEST_F(SyncInvalidationListenerTest, RegisterEnableReady) {
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 }
@@ -544,7 +734,7 @@ TEST_F(SyncInvalidationListenerTest, RegisterEnableReady) {
 TEST_F(SyncInvalidationListenerTest, RegisterReadyEnable) {
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 
@@ -557,7 +747,7 @@ TEST_F(SyncInvalidationListenerTest, RegisterReadyEnable) {
 // ready the client.  The IDs should be registered only after the
 // client is readied.
 TEST_F(SyncInvalidationListenerTest, EnableRegisterReady) {
-  client_.UpdateRegisteredIds(ObjectIdSet());
+  listener_.UpdateRegisteredIds(ObjectIdSet());
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
@@ -565,11 +755,11 @@ TEST_F(SyncInvalidationListenerTest, EnableRegisterReady) {
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.UpdateRegisteredIds(registered_ids_);
+  listener_.UpdateRegisteredIds(registered_ids_);
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 }
@@ -578,7 +768,7 @@ TEST_F(SyncInvalidationListenerTest, EnableRegisterReady) {
 // re-register the IDs.  The IDs should be registered only after the
 // client is readied.
 TEST_F(SyncInvalidationListenerTest, EnableReadyRegister) {
-  client_.UpdateRegisteredIds(ObjectIdSet());
+  listener_.UpdateRegisteredIds(ObjectIdSet());
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
@@ -586,11 +776,11 @@ TEST_F(SyncInvalidationListenerTest, EnableReadyRegister) {
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.UpdateRegisteredIds(registered_ids_);
+  listener_.UpdateRegisteredIds(registered_ids_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 }
@@ -599,7 +789,7 @@ TEST_F(SyncInvalidationListenerTest, EnableReadyRegister) {
 // re-register the IDs.  The IDs should be registered only after the
 // client is readied.
 TEST_F(SyncInvalidationListenerTest, ReadyEnableRegister) {
-  client_.UpdateRegisteredIds(ObjectIdSet());
+  listener_.UpdateRegisteredIds(ObjectIdSet());
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
@@ -607,11 +797,11 @@ TEST_F(SyncInvalidationListenerTest, ReadyEnableRegister) {
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.UpdateRegisteredIds(registered_ids_);
+  listener_.UpdateRegisteredIds(registered_ids_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 }
@@ -622,15 +812,15 @@ TEST_F(SyncInvalidationListenerTest, ReadyEnableRegister) {
 //
 // This test is important: see http://crbug.com/139424.
 TEST_F(SyncInvalidationListenerTest, ReadyRegisterEnable) {
-  client_.UpdateRegisteredIds(ObjectIdSet());
+  listener_.UpdateRegisteredIds(ObjectIdSet());
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.UpdateRegisteredIds(registered_ids_);
+  listener_.UpdateRegisteredIds(registered_ids_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 
@@ -644,7 +834,7 @@ TEST_F(SyncInvalidationListenerTest, ReadyRegisterEnable) {
 TEST_F(SyncInvalidationListenerTest, RegisterTypesPreserved) {
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 
@@ -652,7 +842,7 @@ TEST_F(SyncInvalidationListenerTest, RegisterTypesPreserved) {
 
   EXPECT_TRUE(GetRegisteredIds().empty());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(registered_ids_, GetRegisteredIds());
 }
@@ -660,21 +850,22 @@ TEST_F(SyncInvalidationListenerTest, RegisterTypesPreserved) {
 // Make sure that state is correctly purged from the local invalidation state
 // map cache when an ID is unregistered.
 TEST_F(SyncInvalidationListenerTest, UnregisterCleansUpStateMapCache) {
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
-  InvalidationStateMap state_map;
-  state_map[kBookmarksId_].version = 1;
+  EXPECT_TRUE(listener_.GetStateMapForTest().empty());
   FireInvalidate(kBookmarksId_, 1, "hello");
-  EXPECT_EQ(state_map, client_.GetStateMapForTest());
-  state_map[kPreferencesId_].version = 2;
+  EXPECT_EQ(1U, listener_.GetStateMapForTest().size());
+  EXPECT_TRUE(ContainsKey(listener_.GetStateMapForTest(), kBookmarksId_));
   FireInvalidate(kPreferencesId_, 2, "world");
-  EXPECT_EQ(state_map, client_.GetStateMapForTest());
+  EXPECT_EQ(2U, listener_.GetStateMapForTest().size());
+  EXPECT_TRUE(ContainsKey(listener_.GetStateMapForTest(), kBookmarksId_));
+  EXPECT_TRUE(ContainsKey(listener_.GetStateMapForTest(), kPreferencesId_));
 
   ObjectIdSet ids;
   ids.insert(kBookmarksId_);
-  client_.UpdateRegisteredIds(ids);
-  state_map.erase(kPreferencesId_);
-  EXPECT_EQ(state_map, client_.GetStateMapForTest());
+  listener_.UpdateRegisteredIds(ids);
+  EXPECT_EQ(1U, listener_.GetStateMapForTest().size());
+  EXPECT_TRUE(ContainsKey(listener_.GetStateMapForTest(), kBookmarksId_));
 }
 
 // Without readying the client, disable notifications, then enable
@@ -706,7 +897,7 @@ TEST_F(SyncInvalidationListenerTest, EnableNotificationsThenReady) {
 
   EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
 }
@@ -716,7 +907,7 @@ TEST_F(SyncInvalidationListenerTest, EnableNotificationsThenReady) {
 TEST_F(SyncInvalidationListenerTest, ReadyThenEnableNotifications) {
   EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
 
@@ -730,7 +921,7 @@ TEST_F(SyncInvalidationListenerTest, ReadyThenEnableNotifications) {
 // delegate should go into an auth error mode and then back out.
 TEST_F(SyncInvalidationListenerTest, PushClientAuthError) {
   EnableNotifications();
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
 
@@ -750,11 +941,11 @@ TEST_F(SyncInvalidationListenerTest, PushClientAuthError) {
 // auth error mode and come out of it only after the client is ready.
 TEST_F(SyncInvalidationListenerTest, InvalidationClientAuthError) {
   EnableNotifications();
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
 
-  client_.InformError(
+  listener_.InformError(
       fake_invalidation_client_,
       invalidation::ErrorInfo(
           invalidation::ErrorReason::AUTH_FAILURE,
@@ -776,7 +967,7 @@ TEST_F(SyncInvalidationListenerTest, InvalidationClientAuthError) {
 
   EXPECT_EQ(INVALIDATION_CREDENTIALS_REJECTED, GetInvalidatorState());
 
-  client_.Ready(fake_invalidation_client_);
+  listener_.Ready(fake_invalidation_client_);
 
   EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
 }