Implement new invalidations ack tracking system

sync/notifier/sync_invalidation_listener.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/notifier/sync_invalidation_listener.h"
 
 #include <vector>
 
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/tracked_objects.h"
 #include "google/cacheinvalidation/include/invalidation-client.h"
 #include "google/cacheinvalidation/include/types.h"
 #include "google/cacheinvalidation/types.pb.h"
 #include "jingle/notifier/listener/push_client.h"
 #include "sync/notifier/invalidation_util.h"
+#include "sync/notifier/object_id_invalidation_map.h"
 #include "sync/notifier/registration_manager.h"
 
 namespace {
 
 const char kApplicationName[] = "chrome-sync";
 
-static const int64 kUnknownVersion = -1;
-
 }  // namespace
 
 namespace syncer {
 
 SyncInvalidationListener::Delegate::~Delegate() {}
 
 SyncInvalidationListener::SyncInvalidationListener(
-    base::TickClock* tick_clock,
     scoped_ptr<notifier::PushClient> push_client)
-    : ack_tracker_(tick_clock, this),
-      push_client_(push_client.get()),
+    : push_client_(push_client.get()),
       sync_system_resources_(push_client.Pass(), this),
       delegate_(NULL),
       ticl_state_(DEFAULT_INVALIDATION_ERROR),
       push_client_state_(DEFAULT_INVALIDATION_ERROR),
       weak_ptr_factory_(this) {
   DCHECK(CalledOnValidThread());
   push_client_->AddObserver(this);
 }
 
 SyncInvalidationListener::~SyncInvalidationListener() {
   DCHECK(CalledOnValidThread());
   push_client_->RemoveObserver(this);
   Stop();
   DCHECK(!delegate_);
 }
 
 void SyncInvalidationListener::Start(
     const CreateInvalidationClientCallback&
         create_invalidation_client_callback,
     const std::string& client_id, const std::string& client_info,
     const std::string& invalidation_bootstrap_data,
-    const InvalidationStateMap& initial_invalidation_state_map,
+    const UnackedInvalidationsMap& initial_unacked_invalidations,
     const WeakHandle<InvalidationStateTracker>& invalidation_state_tracker,
     Delegate* delegate) {
   DCHECK(CalledOnValidThread());
   Stop();
 
   sync_system_resources_.set_platform(client_info);
   sync_system_resources_.Start();
 
   // The Storage resource is implemented as a write-through cache.  We populate
   // it with the initial state on startup, so subsequent writes go to disk and
   // update the in-memory cache, while reads just return the cached state.
   sync_system_resources_.storage()->SetInitialState(
       invalidation_bootstrap_data);
 
-  invalidation_state_map_ = initial_invalidation_state_map;
-  if (invalidation_state_map_.empty()) {
-    DVLOG(2) << "No initial max invalidation versions for any id";
-  } else {
-    for (InvalidationStateMap::const_iterator it =
-             invalidation_state_map_.begin();
-         it != invalidation_state_map_.end(); ++it) {
-      DVLOG(2) << "Initial max invalidation version for "
-               << ObjectIdToString(it->first) << " is "
-               << it->second.version;
-    }
-  }
+  unacked_invalidations_map_ = initial_unacked_invalidations;
   invalidation_state_tracker_ = invalidation_state_tracker;
   DCHECK(invalidation_state_tracker_.IsInitialized());
 
   DCHECK(!delegate_);
   DCHECK(delegate);
   delegate_ = delegate;
 
 #if defined(OS_IOS)
   int client_type = ipc::invalidation::ClientType::CHROME_SYNC_IOS;
 #else
   int client_type = ipc::invalidation::ClientType::CHROME_SYNC;
 #endif
   invalidation_client_.reset(
       create_invalidation_client_callback.Run(
           &sync_system_resources_, client_type, client_id,
           kApplicationName, this));
   invalidation_client_->Start();
 
   registration_manager_.reset(
       new RegistrationManager(invalidation_client_.get()));
-
-  // 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(
     const std::string& email, const std::string& token) {
   DCHECK(CalledOnValidThread());
   sync_system_resources_.network()->UpdateCredentials(email, token);
 }
 
 void SyncInvalidationListener::UpdateRegisteredIds(const ObjectIdSet& ids) {
   DCHECK(CalledOnValidThread());
   registered_ids_ = ids;
   // |ticl_state_| can go to INVALIDATIONS_ENABLED even without a
   // working XMPP connection (as observed by us), so check it instead
   // of GetState() (see http://crbug.com/139424).
   if (ticl_state_ == INVALIDATIONS_ENABLED && registration_manager_) {
     DoRegistrationUpdate();
   }
 }
 
-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());
   DCHECK_EQ(client, invalidation_client_.get());
   ticl_state_ = INVALIDATIONS_ENABLED;
   EmitStateChange();
   DoRegistrationUpdate();
 }
 
 void SyncInvalidationListener::Invalidate(
     invalidation::InvalidationClient* client,
     const invalidation::Invalidation& invalidation,
     const invalidation::AckHandle& ack_handle) {
   DCHECK(CalledOnValidThread());
   DCHECK_EQ(client, invalidation_client_.get());
-  DVLOG(1) << "Invalidate: " << InvalidationToString(invalidation);
+  client->Acknowledge(ack_handle);
 
   const invalidation::ObjectId& id = invalidation.object_id();
 
-  // The invalidation API spec allows for the possibility of redundant
-  // invalidations, so keep track of the max versions and drop
-  // invalidations with old versions.
-  //
-  // TODO(akalin): Now that we keep track of registered ids, we
-  // should drop invalidations for unregistered ids.  We may also
-  // have to filter it at a higher level, as invalidations for
-  // newly-unregistered ids may already be in flight.
-  InvalidationStateMap::const_iterator it = invalidation_state_map_.find(id);
-  if ((it != invalidation_state_map_.end()) &&
-      (invalidation.version() <= it->second.version)) {
-    // Drop redundant invalidations.
-    client->Acknowledge(ack_handle);
-    return;
-  }
-
   std::string payload;
   // payload() CHECK()'s has_payload(), so we must check it ourselves first.
   if (invalidation.has_payload())
     payload = invalidation.payload();
 
-  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);
+  DVLOG(2) << "Received invalidation with version " << invalidation.version()
+           << " for " << ObjectIdToString(id);
 
-  ObjectIdSet ids;
-  ids.insert(id);
-  PrepareInvalidation(ids, invalidation.version(), payload, client, ack_handle);
+  ObjectIdInvalidationMap invalidations;
+  Invalidation inv = Invalidation::Init(id, invalidation.version(), payload);
+  inv.set_ack_handler(GetThisAsAckHandler());
+  invalidations.Insert(inv);
+
+  DispatchInvalidations(invalidations);
 }
 
 void SyncInvalidationListener::InvalidateUnknownVersion(
     invalidation::InvalidationClient* client,
     const invalidation::ObjectId& object_id,
     const invalidation::AckHandle& ack_handle) {
   DCHECK(CalledOnValidThread());
   DCHECK_EQ(client, invalidation_client_.get());
   DVLOG(1) << "InvalidateUnknownVersion";
+  client->Acknowledge(ack_handle);
 
-  ObjectIdSet ids;
-  ids.insert(object_id);
-  PrepareInvalidation(
-      ids,
-      kUnknownVersion,
-      std::string(),
-      client,
-      ack_handle);
+  ObjectIdInvalidationMap invalidations;
+  Invalidation unknown_version = Invalidation::InitUnknownVersion(object_id);
+  unknown_version.set_ack_handler(GetThisAsAckHandler());
+  invalidations.Insert(unknown_version);
+
+  DispatchInvalidations(invalidations);
 }
 
 // This should behave as if we got an invalidation with version
 // UNKNOWN_OBJECT_VERSION for all known data types.
 void SyncInvalidationListener::InvalidateAll(
     invalidation::InvalidationClient* client,
     const invalidation::AckHandle& ack_handle) {
   DCHECK(CalledOnValidThread());
   DCHECK_EQ(client, invalidation_client_.get());
   DVLOG(1) << "InvalidateAll";
+  client->Acknowledge(ack_handle);
 
-  PrepareInvalidation(
-      registered_ids_,
-      kUnknownVersion,
-      std::string(),
-      client,
-      ack_handle);
+  ObjectIdInvalidationMap invalidations;
+  for (ObjectIdSet::iterator it = registered_ids_.begin();
+       it != registered_ids_.end(); ++it) {
+    Invalidation unknown_version = Invalidation::InitUnknownVersion(*it);
+    unknown_version.set_ack_handler(GetThisAsAckHandler());
+    invalidations.Insert(unknown_version);
+  }
+
+  DispatchInvalidations(invalidations);
 }
 
-void SyncInvalidationListener::PrepareInvalidation(
-    const ObjectIdSet& ids,
-    int64 version,
-    const std::string& payload,
-    invalidation::InvalidationClient* client,
-    const invalidation::AckHandle& ack_handle) {
+// If a handler is registered, emit right away.  Otherwise, save it for later.
+void SyncInvalidationListener::DispatchInvalidations(
+    const ObjectIdInvalidationMap& invalidations) {
   DCHECK(CalledOnValidThread());
 
-  // A server invalidation resets the local retry count.
-  ack_tracker_.Ack(ids);
+  ObjectIdInvalidationMap to_save = invalidations;
+  ObjectIdInvalidationMap to_emit =
+      invalidations.GetSubsetWithObjectIds(registered_ids_);
+
+  SaveInvalidations(to_save);
+  EmitInvalidations(to_emit);
+}
+
+void SyncInvalidationListener::SaveInvalidations(
+    const ObjectIdInvalidationMap& to_save) {
+  ObjectIdSet objects_to_save = to_save.GetObjectIds();
+  for (ObjectIdSet::iterator it = objects_to_save.begin();

[tim (not reviewing), 2013/11/18 21:59:05]

const_iterator?

[rlarocque, 2013/11/19 00:35:49]

Done.

+       it != objects_to_save.end(); ++it) {
+    UnackedInvalidationsMap::iterator lookup =
+        unacked_invalidations_map_.find(*it);
+    if (lookup == unacked_invalidations_map_.end()) {
+      lookup = unacked_invalidations_map_.insert(
+          std::make_pair(*it, UnackedInvalidationSet(*it))).first;
+    }
+    lookup->second.AddSet(to_save.ForObject(*it));
+  }
+
   invalidation_state_tracker_.Call(

[rlarocque, 2013/11/21 20:09:27]

What happens if this call doesn't make it through before shutdown?

If that happens, we could very well drop an invalidation.  This is not a
regression, though.  The old code had more complicated thread logic, but it
still acknowledged the the invalidation to the server before it wrote the
invalidation to disk.  

What the old system did accomplish but this one does not is ensure that  it
waited until the invalidation has been written to the prefs system (in memory
only; it's guaranteed to be written during a graceful shutdown but not during a
crash) before the invalidation is communicated to the observers.  I'm not sure
why it did this.  Maybe it had something to do with the periodic retries.

Could we actually close this loophole and prevent invalidations from being
dropped here?  Probably.  We have a reasonable amount of time (tens of seconds)
to acknowledge the invalidation to the server before anything bad happens.  We
can probably delay that acknowledgement at least until the invalidation has been
written to the prefs system, if not to disk.  But the server periodically pings
the client with retries if we don't acknowledge right away, so that could
complicate matters.  I'd prefer to deal with those issues in a separate CL.
Because this is not a regression, I feel justified in delaying this work until
later.

[tim (not reviewing), 2013/11/22 01:24:54]

The server won't ping periodicially within the 10s interval, will it? 

OK, as long as we don't forget / track this somewhere. Is there a bug?

[rlarocque, 2013/11/22 21:03:32]

Yes, it will send the same invalidation repeatedly for a while.  I'm counting
the time of the periodic retry period in the "tens of seconds" estimate.  The
first retry might arrive after a much shorter period of time.  (My memory is
foggy on this point.  I'd have to look it up to be sure.)

I filed http://crbug.com/322668.

       FROM_HERE,
-      &InvalidationStateTracker::GenerateAckHandles,
-      ids,
-      base::MessageLoopProxy::current(),
-      base::Bind(&SyncInvalidationListener::EmitInvalidation,
-                 weak_ptr_factory_.GetWeakPtr(),
-                 ids,
-                 version,
-                 payload,
-                 client,
-                 ack_handle));
+      &InvalidationStateTracker::SetSavedInvalidations,
+      unacked_invalidations_map_);
 }
 
-void SyncInvalidationListener::EmitInvalidation(
-    const ObjectIdSet& ids,
-    int64 version,
-    const std::string& payload,
-    invalidation::InvalidationClient* client,
-    const invalidation::AckHandle& ack_handle,
-    const AckHandleMap& local_ack_handles) {
-  DCHECK(CalledOnValidThread());
-
-  ObjectIdInvalidationMap invalidation_map;
-  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;
-
-    if (version == kUnknownVersion) {
-      Invalidation inv = Invalidation::InitUnknownVersion(it->first);
-      inv.set_ack_handle(it->second);
-      invalidation_map.Insert(inv);
-    } else {
-      Invalidation inv = Invalidation::Init(it->first, version, payload);
-      inv.set_ack_handle(it->second);
-      invalidation_map.Insert(inv);
-    }
-  }
-  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) {
-    if (invalidation_state_map_[*it].version == kUnknownVersion) {
-      Invalidation inv = Invalidation::InitUnknownVersion(*it);
-      inv.set_ack_handle(invalidation_state_map_[*it].expected);
-      invalidation_map.Insert(inv);
-    } else {
-      Invalidation inv = Invalidation::Init(
-          *it,
-          invalidation_state_map_[*it].version,
-          invalidation_state_map_[*it].payload);
-      inv.set_ack_handle(invalidation_state_map_[*it].expected);
-      invalidation_map.Insert(inv);
-    }
-  }
-  delegate_->OnInvalidate(invalidation_map);
+void SyncInvalidationListener::EmitInvalidations(
+    const ObjectIdInvalidationMap& to_emit) {

[tim (not reviewing), 2013/11/20 18:27:04]

|to_emit| is a subset of unacked_invalidations_map_, right?

[rlarocque, 2013/11/21 20:09:27]

Yep, they should be.  For every invalidation we emit, we should have a copy of
it sitting in the unacked_invalidations_map_, periodically saved the the storage
(the preferences store), and waiting for acknowledgement to tell us we no longer
need to keep saving them.

+  DVLOG(2) << "Emitting invalidations: " << to_emit.ToString();
+  delegate_->OnInvalidate(to_emit);
 }
 
 void SyncInvalidationListener::InformRegistrationStatus(
       invalidation::InvalidationClient* client,
       const invalidation::ObjectId& object_id,
       InvalidationListener::RegistrationState new_state) {
   DCHECK(CalledOnValidThread());
   DCHECK_EQ(client, invalidation_client_.get());
   DVLOG(1) << "InformRegistrationStatus: "
            << ObjectIdToString(object_id) << " " << new_state;
@@ skipping 52 matching lines @@
              << error_info.error_message()
              << " (transient = " << error_info.is_transient() << ")";
   if (error_info.error_reason() == invalidation::ErrorReason::AUTH_FAILURE) {
     ticl_state_ = INVALIDATION_CREDENTIALS_REJECTED;
   } else {
     ticl_state_ = TRANSIENT_INVALIDATION_ERROR;
   }
   EmitStateChange();
 }
 
+void SyncInvalidationListener::Acknowledge(
+  const invalidation::ObjectId& id,
+  const syncer::AckHandle& handle) {
+  UnackedInvalidationsMap::iterator lookup =
+      unacked_invalidations_map_.find(id);
+  if (lookup == unacked_invalidations_map_.end()) {
+    DLOG(WARNING) << "Received acknowledgement for untracked object ID";
+    return;
+  }
+  lookup->second.Acknowledge(handle);
+  invalidation_state_tracker_.Call(
+      FROM_HERE,
+      &InvalidationStateTracker::SetSavedInvalidations,
+      unacked_invalidations_map_);
+}
+
+void SyncInvalidationListener::Drop(
+    const invalidation::ObjectId& id,
+    const syncer::AckHandle& handle) {
+  UnackedInvalidationsMap::iterator lookup =
+      unacked_invalidations_map_.find(id);
+  if (lookup == unacked_invalidations_map_.end()) {
+    DLOG(WARNING) << "Received drop for untracked object ID";
+    return;
+  }
+  lookup->second.Drop(handle);
+  invalidation_state_tracker_.Call(

[tim (not reviewing), 2013/11/20 18:27:04]

What happens if this doesn't make it through before Chrome shuts down?

[rlarocque, 2013/11/21 20:09:27]

In this case it's no big deal.  

We'll fail to permanently drop the invalidation.  We'll fail to record the
"missing invalidations" flag-equivalent (though it may be still set from
earlier) and fail to remove this invalidation from the list of saved
invalidations.

On the next restart, the client will likely receive this invalidation and act on
it.  So we actually get a better result than we would have gotten if this
invalidation hadn't gotten through.

You may have meant to ask this question about a different call to
SetSavedInvalidations.  I'll add a comment there.

+      FROM_HERE,
+      &InvalidationStateTracker::SetSavedInvalidations,
+      unacked_invalidations_map_);
+}
+
 void SyncInvalidationListener::WriteState(const std::string& state) {
   DCHECK(CalledOnValidThread());
   DVLOG(1) << "WriteState";
   invalidation_state_tracker_.Call(
       FROM_HERE, &InvalidationStateTracker::SetBootstrapData, state);
 }
 
 void SyncInvalidationListener::DoRegistrationUpdate() {
   DCHECK(CalledOnValidThread());
   const ObjectIdSet& unregistered_ids =
       registration_manager_->UpdateRegisteredIds(registered_ids_);
-  for (ObjectIdSet::const_iterator it = unregistered_ids.begin();
+  for (ObjectIdSet::iterator it = unregistered_ids.begin();
        it != unregistered_ids.end(); ++it) {
-    invalidation_state_map_.erase(*it);
+    unacked_invalidations_map_.erase(*it);
   }
   invalidation_state_tracker_.Call(
-      FROM_HERE, &InvalidationStateTracker::Forget, unregistered_ids);
-  ack_tracker_.Ack(unregistered_ids);
+      FROM_HERE,
+      &InvalidationStateTracker::SetSavedInvalidations,
+      unacked_invalidations_map_);
+
+  ObjectIdInvalidationMap object_id_invalidation_map;
+  for (UnackedInvalidationsMap::iterator map_it =
+       unacked_invalidations_map_.begin();
+       map_it != unacked_invalidations_map_.end(); ++map_it) {
+    if (registered_ids_.find(map_it->first) == registered_ids_.end()) {
+      continue;
+    }
+    map_it->second.ExportInvalidations(
+        GetThisAsAckHandler(),
+        &object_id_invalidation_map);
+  }
+  EmitInvalidations(object_id_invalidation_map);

[tim (not reviewing), 2013/11/20 18:27:04]

Add a comment before this line that we need only Emit, not Dispatch, because the
invalidations are already saved (we just read from them).

In general it seems like you never want to call Emit directly, right? This makes
me wonder if we couldn't name Emit/Dispatch more clearly.  Is it a prerequisite
for Emit to operate on saved invalidations?  Does it need to take a parameter? 
Could we distinguish it that way by calling it EmitSavedInvalidations()?

[tim (not reviewing), 2013/11/20 18:27:44]

s/never/rarely

[rlarocque, 2013/11/21 20:09:27]

You're right.  The rename makes sense.

I've added a comment here.

I've also added some comments to the Dispatch, Emit, and Save functions in the
header file.

 }
 
 void SyncInvalidationListener::StopForTest() {
   DCHECK(CalledOnValidThread());
   Stop();
 }
 
-InvalidationStateMap SyncInvalidationListener::GetStateMapForTest() const {
-  DCHECK(CalledOnValidThread());
-  return invalidation_state_map_;
-}
-
-AckTracker* SyncInvalidationListener::GetAckTrackerForTest() {
-  return &ack_tracker_;
-}
-
 void SyncInvalidationListener::Stop() {
   DCHECK(CalledOnValidThread());
   if (!invalidation_client_) {
     return;
   }
 
-  ack_tracker_.Clear();
-
   registration_manager_.reset();
   sync_system_resources_.Stop();
   invalidation_client_->Stop();
 
   invalidation_client_.reset();
   delegate_ = NULL;
 
-  invalidation_state_tracker_.Reset();
-  invalidation_state_map_.clear();
   ticl_state_ = DEFAULT_INVALIDATION_ERROR;
   push_client_state_ = DEFAULT_INVALIDATION_ERROR;
 }
 
 InvalidatorState SyncInvalidationListener::GetState() const {
   DCHECK(CalledOnValidThread());
   if (ticl_state_ == INVALIDATION_CREDENTIALS_REJECTED ||
       push_client_state_ == INVALIDATION_CREDENTIALS_REJECTED) {
     // If either the ticl or the push client rejected our credentials,
     // return INVALIDATION_CREDENTIALS_REJECTED.
     return INVALIDATION_CREDENTIALS_REJECTED;
   }
   if (ticl_state_ == INVALIDATIONS_ENABLED &&
       push_client_state_ == INVALIDATIONS_ENABLED) {
     // If the ticl is ready and the push client notifications are
     // enabled, return INVALIDATIONS_ENABLED.
     return INVALIDATIONS_ENABLED;
   }
   // Otherwise, we have a transient error.
   return TRANSIENT_INVALIDATION_ERROR;
 }
 
 void SyncInvalidationListener::EmitStateChange() {
   DCHECK(CalledOnValidThread());
   delegate_->OnInvalidatorStateChange(GetState());
 }
 
+WeakHandle<AckHandler> SyncInvalidationListener::GetThisAsAckHandler() {
+  DCHECK(CalledOnValidThread());
+  return WeakHandle<AckHandler>(weak_ptr_factory_.GetWeakPtr());
+}
+
 void SyncInvalidationListener::OnNotificationsEnabled() {
   DCHECK(CalledOnValidThread());
   push_client_state_ = INVALIDATIONS_ENABLED;
   EmitStateChange();
 }
 
 void SyncInvalidationListener::OnNotificationsDisabled(
     notifier::NotificationsDisabledReason reason) {
   DCHECK(CalledOnValidThread());
   push_client_state_ = FromNotifierReason(reason);
   EmitStateChange();
 }
 
 void SyncInvalidationListener::OnIncomingNotification(
     const notifier::Notification& notification) {
   DCHECK(CalledOnValidThread());
   // Do nothing, since this is already handled by |invalidation_client_|.
 }
 
 }  // namespace syncer