Recommit attempt for rev 72685.

[SYNC] Refactor SyncSourceInfo and add support in chrome invalidation client and syncer thread for passing a datatype-specific payload originating in the invalidation server and directed at the sync frontend server. Also fixes bug with last_sync_time and PostTimeToTypeHistogram, which would get hit when the unit tests were being run. 

BUG=68572, 69558
TEST=unit

Review URL: http://codereview.chromium.org/6313018

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

21 files changed, 586 insertions, 95 deletions

diff --git a/chrome/browser/sync/engine/download_updates_command.cc b/chrome/browser/sync/engine/download_updates_command.cc
index 84fd6bf3..4c01784 100644
--- a/chrome/browser/sync/engine/download_updates_command.cc
+++ b/chrome/browser/sync/engine/download_updates_command.cc
@@ -60,8 +60,10 @@ void DownloadUpdatesCommand::ExecuteImpl(SyncSession* session) {
   get_updates->set_fetch_folders(true);
 
   // Set GetUpdatesMessage.GetUpdatesCallerInfo information.
+  // TODO(zea): send SyncSourceInfo's payloads to server once we know it's all
+  // working properly.
   get_updates->mutable_caller_info()->set_source(
-      session->TestAndSetSource().first);
+      session->TestAndSetSource().updates_source);
   get_updates->mutable_caller_info()->set_notifications_enabled(
       session->context()->notifications_enabled());
 
diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index bc11b31..a275ae9 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -41,6 +41,7 @@
 #include "chrome/browser/sync/protocol/sync.pb.h"
 #include "chrome/browser/sync/protocol/theme_specifics.pb.h"
 #include "chrome/browser/sync/protocol/typed_url_specifics.pb.h"
+#include "chrome/browser/sync/sessions/sync_session.h"
 #include "chrome/browser/sync/sessions/sync_session_context.h"
 #include "chrome/browser/sync/syncable/autofill_migration.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
@@ -83,6 +84,7 @@ typedef GoogleServiceAuthError AuthError;
 
 static const int kThreadExitTimeoutMsec = 60000;
 static const int kSSLPort = 443;
+static const int kSyncerThreadDelayMsec = 250;
 
 #if defined(OS_CHROMEOS)
 static const int kChromeOSNetworkChangeReactionDelayHackMsec = 5000;
@@ -2193,22 +2195,29 @@ void SyncManager::SyncInternal::TalkMediatorLogin(
 
 void SyncManager::SyncInternal::OnIncomingNotification(
     const IncomingNotificationData& notification_data) {
-  syncable::ModelTypeBitSet model_types;
+  browser_sync::sessions::TypePayloadMap model_types_with_payloads;
 
   // Check if the service url is a sync URL.  An empty service URL is
   // treated as a legacy sync notification.  If we're listening to
   // server-issued notifications, no need to check the service_url.
   if (notifier_options_.notification_method ==
       notifier::NOTIFICATION_SERVER) {
-    VLOG(1) << "Sync received server notification: " <<
+    VLOG(1) << "Sync received server notification from " <<
+        notification_data.service_url << ": " <<
+        notification_data.service_specific_data;
+    syncable::ModelTypeBitSet model_types;
+    const std::string& model_type_list = notification_data.service_url;
+    const std::string& notification_payload =
         notification_data.service_specific_data;
 
-    if (!syncable::ModelTypeBitSetFromString(
-            notification_data.service_specific_data,
-            &model_types)) {
+    if (!syncable::ModelTypeBitSetFromString(model_type_list, &model_types)) {
       LOG(DFATAL) << "Could not extract model types from server data.";
       model_types.set();
     }
+
+    model_types_with_payloads =
+        browser_sync::sessions::MakeTypePayloadMapFromBitSet(model_types,
+            notification_payload);
   } else if (notification_data.service_url.empty() ||
              (notification_data.service_url ==
               browser_sync::kSyncLegacyServiceUrl) ||
@@ -2216,20 +2225,24 @@ void SyncManager::SyncInternal::OnIncomingNotification(
               browser_sync::kSyncServiceUrl)) {
     VLOG(1) << "Sync received P2P notification.";
 
-    // Catch for sync integration tests (uses p2p). Just set all datatypes.
-    model_types.set();
+    // Catch for sync integration tests (uses p2p). Just set all enabled
+    // datatypes.
+    ModelSafeRoutingInfo routes;
+    registrar_->GetModelSafeRoutingInfo(&routes);
+    model_types_with_payloads =
+        browser_sync::sessions::MakeTypePayloadMapFromRoutingInfo(routes,
+            std::string());
   } else {
     LOG(WARNING) << "Notification fron unexpected source: "
                  << notification_data.service_url;
   }
 
-  if (model_types.any()) {
+  if (!model_types_with_payloads.empty()) {
     if (syncer_thread()) {
-     // Introduce a delay to help coalesce initial notifications.
-     syncer_thread()->NudgeSyncerWithDataTypes(
-         250,
-         SyncerThread::kNotification,
-         model_types);
+      syncer_thread()->NudgeSyncerWithPayloads(
+          kSyncerThreadDelayMsec,
+          SyncerThread::kNotification,
+          model_types_with_payloads);
     }
     allstatus_.IncrementNotificationsReceived();
   } else {
diff --git a/chrome/browser/sync/engine/syncer.cc b/chrome/browser/sync/engine/syncer.cc
index 70d585f..cbd7eef 100644
--- a/chrome/browser/sync/engine/syncer.cc
+++ b/chrome/browser/sync/engine/syncer.cc
@@ -78,7 +78,8 @@ void Syncer::SyncShare(sessions::SyncSession* session) {
   CHECK(dir.good());
 
   const sessions::SyncSourceInfo& source(session->source());
-  if (sync_pb::GetUpdatesCallerInfo::CLEAR_PRIVATE_DATA == source.first) {
+  if (sync_pb::GetUpdatesCallerInfo::CLEAR_PRIVATE_DATA ==
+      source.updates_source) {
     SyncShare(session, CLEAR_PRIVATE_DATA, SYNCER_END);
     return;
   } else {
diff --git a/chrome/browser/sync/engine/syncer_thread.cc b/chrome/browser/sync/engine/syncer_thread.cc
index 8813709..20ea665 100644
--- a/chrome/browser/sync/engine/syncer_thread.cc
+++ b/chrome/browser/sync/engine/syncer_thread.cc
@@ -5,8 +5,9 @@
 #include "chrome/browser/sync/engine/syncer_thread.h"
 
 #include <algorithm>
-#include <map>
 #include <queue>
+#include <string>
+#include <vector>
 
 #include "base/rand_util.h"
 #include "base/third_party/dynamic_annotations/dynamic_annotations.h"
@@ -34,6 +35,7 @@ namespace browser_sync {
 using sessions::SyncSession;
 using sessions::SyncSessionSnapshot;
 using sessions::SyncSourceInfo;
+using sessions::TypePayloadMap;
 
 // We use high values here to ensure that failure to receive poll updates from
 // the server doesn't result in rapid-fire polling from the client due to low
@@ -54,6 +56,18 @@ static const int kBackoffRandomizationFactor = 2;
 
 const int SyncerThread::kMaxBackoffSeconds = 60 * 60 * 4;  // 4 hours.
 
+void SyncerThread::NudgeSyncerWithPayloads(
+    int milliseconds_from_now,
+    NudgeSource source,
+    const TypePayloadMap& model_types_with_payloads) {
+  base::AutoLock lock(lock_);
+  if (vault_.syncer_ == NULL) {
+    return;
+  }
+
+  NudgeSyncImpl(milliseconds_from_now, source, model_types_with_payloads);
+}
+
 void SyncerThread::NudgeSyncerWithDataTypes(
     int milliseconds_from_now,
     NudgeSource source,
@@ -63,7 +77,9 @@ void SyncerThread::NudgeSyncerWithDataTypes(
     return;
   }
 
-  NudgeSyncImpl(milliseconds_from_now, source, model_types);
+  TypePayloadMap model_types_with_payloads =
+      sessions::MakeTypePayloadMapFromBitSet(model_types, std::string());
+  NudgeSyncImpl(milliseconds_from_now, source, model_types_with_payloads);
 }
 
 void SyncerThread::NudgeSyncer(
@@ -74,8 +90,12 @@ void SyncerThread::NudgeSyncer(
     return;
   }
 
-  syncable::ModelTypeBitSet model_types;  // All false by default.
-  NudgeSyncImpl(milliseconds_from_now, source, model_types);
+  // Set all enabled datatypes.
+  ModelSafeRoutingInfo routes;
+  session_context_->registrar()->GetModelSafeRoutingInfo(&routes);
+  TypePayloadMap model_types_with_payloads =
+      sessions::MakeTypePayloadMapFromRoutingInfo(routes, std::string());
+  NudgeSyncImpl(milliseconds_from_now, source, model_types_with_payloads);
 }
 
 SyncerThread::SyncerThread(sessions::SyncSessionContext* context)
@@ -342,11 +362,12 @@ void SyncerThread::ThreadMainLoop() {
     // Update timing information for how often these datatypes are triggering
     // nudges.
     base::TimeTicks now = TimeTicks::Now();
-    for (size_t i = syncable::FIRST_REAL_MODEL_TYPE;
-         i < session->source().second.size();
-         ++i) {
-      if (session->source().second[i]) {
-        syncable::PostTimeToTypeHistogram(syncable::ModelType(i),
+    if (!last_sync_time.is_null()) {
+      TypePayloadMap::const_iterator iter;
+      for (iter = session->source().types.begin();
+           iter != session->source().types.end();
+           ++iter) {
+        syncable::PostTimeToTypeHistogram(iter->first,
                                           now - last_sync_time);
       }
     }
@@ -574,7 +595,7 @@ SyncSourceInfo SyncerThread::GetAndResetNudgeSource(bool was_throttled,
                                                     bool* was_nudged) {
   bool nudged = false;
   NudgeSource nudge_source = kUnknown;
-  syncable::ModelTypeBitSet model_types;
+  TypePayloadMap model_types_with_payloads;
   // Has the previous sync cycle completed?
   if (continue_sync_cycle)
     nudge_source = kContinuation;
@@ -583,13 +604,13 @@ SyncSourceInfo SyncerThread::GetAndResetNudgeSource(bool was_throttled,
   if (!vault_.pending_nudge_time_.is_null()) {
     if (!was_throttled) {
       nudge_source = vault_.pending_nudge_source_;
-      model_types = vault_.pending_nudge_types_;
+      model_types_with_payloads = vault_.pending_nudge_types_;
       nudged = true;
     }
     VLOG(1) << "Clearing pending nudge from " << vault_.pending_nudge_source_
             << " at tick " << vault_.pending_nudge_time_.ToInternalValue();
     vault_.pending_nudge_source_ = kUnknown;
-    vault_.pending_nudge_types_.reset();
+    vault_.pending_nudge_types_.clear();
     vault_.pending_nudge_time_ = base::TimeTicks();
   }
 
@@ -599,11 +620,12 @@ SyncSourceInfo SyncerThread::GetAndResetNudgeSource(bool was_throttled,
   // from syncer having more work to do.  This will be handled properly with
   // the message loop based syncer thread, bug 26339.
   return MakeSyncSourceInfo(nudged || nudge_source == kContinuation,
-      nudge_source, model_types, initial_sync);
+      nudge_source, model_types_with_payloads, initial_sync);
 }
 
 SyncSourceInfo SyncerThread::MakeSyncSourceInfo(bool nudged,
-    NudgeSource nudge_source, const syncable::ModelTypeBitSet& nudge_types,
+    NudgeSource nudge_source,
+    const TypePayloadMap& model_types_with_payloads,
     bool* initial_sync) {
   sync_pb::GetUpdatesCallerInfo::GetUpdatesSource updates_source =
       sync_pb::GetUpdatesCallerInfo::UNKNOWN;
@@ -632,7 +654,19 @@ SyncSourceInfo SyncerThread::MakeSyncSourceInfo(bool nudged,
         break;
     }
   }
-  return SyncSourceInfo(updates_source, nudge_types);
+
+  TypePayloadMap sync_source_types;
+  if (model_types_with_payloads.empty()) {
+    // No datatypes requested. This must be a poll so set all enabled datatypes.
+    ModelSafeRoutingInfo routes;
+    session_context_->registrar()->GetModelSafeRoutingInfo(&routes);
+    sync_source_types = sessions::MakeTypePayloadMapFromRoutingInfo(routes,
+        std::string());
+  } else {
+    sync_source_types = model_types_with_payloads;
+  }
+
+  return SyncSourceInfo(updates_source, sync_source_types);
 }
 
 void SyncerThread::CreateSyncer(const std::string& dirname) {
@@ -734,9 +768,10 @@ int SyncerThread::CalculateSyncWaitTime(int last_interval, int user_idle_ms) {
 }
 
 // Called with mutex_ already locked.
-void SyncerThread::NudgeSyncImpl(int milliseconds_from_now,
-                                 NudgeSource source,
-                                 const syncable::ModelTypeBitSet& model_types) {
+void SyncerThread::NudgeSyncImpl(
+    int milliseconds_from_now,
+    NudgeSource source,
+    const TypePayloadMap& model_types_with_payloads) {
   // TODO(sync): Add the option to reset the backoff state machine.
   // This is needed so nudges that are a result of the user's desire
   // to download updates for a new data type can be satisfied quickly.
@@ -747,11 +782,12 @@ void SyncerThread::NudgeSyncImpl(int milliseconds_from_now,
     return;
   }
 
-  // Union the current bitset with any from nudges that may have already
+  // Union the current TypePayloadMap with any from nudges that may have already
   // posted (coalesce the nudge datatype information).
   // TODO(tim): It seems weird to do this if the sources don't match up (e.g.
   // if pending_source is kLocal and |source| is kClearPrivateData).
-  vault_.pending_nudge_types_ |= model_types;
+  sessions::CoalescePayloads(&vault_.pending_nudge_types_,
+                             model_types_with_payloads);
 
   const TimeTicks nudge_time = TimeTicks::Now() +
       TimeDelta::FromMilliseconds(milliseconds_from_now);
diff --git a/chrome/browser/sync/engine/syncer_thread.h b/chrome/browser/sync/engine/syncer_thread.h
index 22531d8..122e5df 100644
--- a/chrome/browser/sync/engine/syncer_thread.h
+++ b/chrome/browser/sync/engine/syncer_thread.h
@@ -49,6 +49,9 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest, NudgeWithDataTypes);
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest,
                            NudgeWithDataTypesCoalesced);
+  FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest, NudgeWithPayloads);
+  FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest,
+                           NudgeWithPayloadsCoalesced);
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest, Throttling);
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest, AuthInvalid);
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest, Pause);
@@ -138,7 +141,15 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
   virtual void NudgeSyncerWithDataTypes(
       int milliseconds_from_now,
       NudgeSource source,
-      const syncable::ModelTypeBitSet& model_type);
+      const syncable::ModelTypeBitSet& model_types);
+
+  // Same as |NudgeSyncer|, but supports including a payload for passing on to
+  // the download updates command. Datatypes with payloads are also considered
+  // to have caused a nudged to occur and treated accordingly.
+  virtual void NudgeSyncerWithPayloads(
+      int milliseconds_from_now,
+      NudgeSource source,
+      const sessions::TypePayloadMap& model_types_with_payloads);
 
   void SetNotificationsEnabled(bool notifications_enabled);
 
@@ -189,9 +200,13 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
     // check pending_nudge_time_.)
     NudgeSource pending_nudge_source_;
 
-    // BitSet of the datatypes that have triggered the current nudge
-    // (can be union of various bitsets when multiple nudges are coalesced)
-    syncable::ModelTypeBitSet pending_nudge_types_;
+    // Map of all datatypes that are requesting a nudge. Can be union
+    // from multiple nudges that are coalesced. In addition, we
+    // optionally track a payload associated with each datatype (most recent
+    // payload overwrites old ones). These payloads are used by the download
+    // updates command and can contain datatype specific information the server
+    // might use.
+    sessions::TypePayloadMap pending_nudge_types_;
 
     // null iff there is no pending nudge.
     base::TimeTicks pending_nudge_time_;
@@ -280,7 +295,7 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
   sessions::SyncSourceInfo MakeSyncSourceInfo(
       bool nudged,
       NudgeSource nudge_source,
-      const syncable::ModelTypeBitSet& nudge_types,
+      const sessions::TypePayloadMap& model_types_with_payloads,
       bool* initial_sync);
 
   int UserIdleTime();
@@ -327,7 +342,7 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
   void NudgeSyncImpl(
       int milliseconds_from_now,
       NudgeSource source,
-      const syncable::ModelTypeBitSet& model_types);
+      const sessions::TypePayloadMap& model_types_with_payloads);
 
 #if defined(OS_LINUX)
   // On Linux, we need this information in order to query idle time.
diff --git a/chrome/browser/sync/engine/syncer_thread2.cc b/chrome/browser/sync/engine/syncer_thread2.cc
index c53ea15..9ad29a2 100644
--- a/chrome/browser/sync/engine/syncer_thread2.cc
+++ b/chrome/browser/sync/engine/syncer_thread2.cc
@@ -17,6 +17,7 @@ namespace browser_sync {
 using sessions::SyncSession;
 using sessions::SyncSessionSnapshot;
 using sessions::SyncSourceInfo;
+using sessions::TypePayloadMap;
 using syncable::ModelTypeBitSet;
 using sync_pb::GetUpdatesCallerInfo;
 
@@ -126,7 +127,7 @@ bool SyncerThread::ShouldRunJob(SyncSessionJob::Purpose purpose,
   // being starved from running (e.g. due to a very, very low poll interval,
   // such as 0ms). It's rare that this would ever matter in practice.
   if (purpose == SyncSessionJob::POLL && (pending_nudge_.get() &&
-      pending_nudge_->session->source().first ==
+      pending_nudge_->session->source().updates_source ==
           GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION)) {
     return false;
   }
@@ -175,12 +176,27 @@ void SyncerThread::ScheduleNudge(const TimeDelta& delay,
     return;
   }
 
+  TypePayloadMap types_with_payloads =
+      sessions::MakeTypePayloadMapFromBitSet(types, std::string());
   thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-      this, &SyncerThread::ScheduleNudgeImpl, delay, source, types));
+      this, &SyncerThread::ScheduleNudgeImpl, delay, source,
+      types_with_payloads));
+}
+
+void SyncerThread::ScheduleNudgeWithPayloads(const TimeDelta& delay,
+    NudgeSource source, const TypePayloadMap& types_with_payloads) {
+  if (!thread_.IsRunning()) {
+    NOTREACHED();
+    return;
+  }
+
+  thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
+      this, &SyncerThread::ScheduleNudgeImpl, delay, source,
+      types_with_payloads));
 }
 
 void SyncerThread::ScheduleNudgeImpl(const TimeDelta& delay,
-    NudgeSource source, const ModelTypeBitSet& model_types) {
+    NudgeSource source, const TypePayloadMap& types_with_payloads) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
   TimeTicks rough_start = TimeTicks::Now() + delay;
   if (!ShouldRunJob(SyncSessionJob::NUDGE, rough_start))
@@ -193,7 +209,8 @@ void SyncerThread::ScheduleNudgeImpl(const TimeDelta& delay,
   std::vector<ModelSafeWorker*> workers;
   session_context_->registrar()->GetModelSafeRoutingInfo(&routes);
   session_context_->registrar()->GetWorkers(&workers);
-  SyncSourceInfo info(GetUpdatesFromNudgeSource(source), model_types);
+  SyncSourceInfo info(GetUpdatesFromNudgeSource(source),
+                      types_with_payloads);
 
   scoped_ptr<SyncSession> session(new SyncSession(
       session_context_.get(), this, info, routes, workers));
@@ -316,12 +333,12 @@ void SyncerThread::FinishSyncSessionJob(const SyncSessionJob& job) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
   // Update timing information for how often datatypes are triggering nudges.
   base::TimeTicks now = TimeTicks::Now();
-  for (size_t i = syncable::FIRST_REAL_MODEL_TYPE;
-       i < job.session->source().second.size() &&
-           !last_sync_session_end_time_.is_null();
-       ++i) {
-    if (job.session->source().second[i]) {
-      syncable::PostTimeToTypeHistogram(syncable::ModelTypeFromInt(i),
+  if (!last_sync_session_end_time_.is_null()) {
+    TypePayloadMap::const_iterator iter;
+    for (iter = job.session->source().types.begin();
+         iter != job.session->source().types.end();
+         ++iter) {
+      syncable::PostTimeToTypeHistogram(iter->first,
                                         now - last_sync_session_end_time_);
     }
   }
@@ -355,7 +372,7 @@ void SyncerThread::ScheduleNextSync(const SyncSessionJob& old_job) {
     return;
   }
 
-  if (old_job.session->source().first ==
+  if (old_job.session->source().updates_source ==
       GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION) {
     // We don't seem to have made forward progress. Start or extend backoff.
     HandleConsecutiveContinuationError(old_job);
@@ -369,7 +386,7 @@ void SyncerThread::ScheduleNextSync(const SyncSessionJob& old_job) {
     // We weren't continuing and we aren't in backoff.  Schedule a normal
     // continuation.
     ScheduleNudgeImpl(TimeDelta::FromSeconds(0), NUDGE_SOURCE_CONTINUATION,
-                      old_job.session->source().second);
+                      old_job.session->source().types);
   }
 }
 
@@ -456,7 +473,9 @@ void SyncerThread::PollTimerCallback() {
   std::vector<ModelSafeWorker*> w;
   session_context_->registrar()->GetModelSafeRoutingInfo(&r);
   session_context_->registrar()->GetWorkers(&w);
-  SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, ModelTypeBitSet());
+  TypePayloadMap types_with_payloads =
+      sessions::MakeTypePayloadMapFromRoutingInfo(r, std::string());
+  SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, types_with_payloads);
   SyncSession* s = new SyncSession(session_context_.get(), this, info, r, w);
   ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), SyncSessionJob::POLL, s);
 }
diff --git a/chrome/browser/sync/engine/syncer_thread2.h b/chrome/browser/sync/engine/syncer_thread2.h
index 138d2a5..b2e2395 100644
--- a/chrome/browser/sync/engine/syncer_thread2.h
+++ b/chrome/browser/sync/engine/syncer_thread2.h
@@ -60,6 +60,9 @@ class SyncerThread : public sessions::SyncSession::Delegate {
   // The meat and potatoes.
   void ScheduleNudge(const base::TimeDelta& delay, NudgeSource source,
                      const syncable::ModelTypeBitSet& types);
+  void ScheduleNudgeWithPayloads(
+      const base::TimeDelta& delay, NudgeSource source,
+      const sessions::TypePayloadMap& types_with_payloads);
   void ScheduleConfig(const base::TimeDelta& delay,
                       const syncable::ModelTypeBitSet& types);
 
@@ -150,7 +153,7 @@ class SyncerThread : public sessions::SyncSession::Delegate {
   void StartImpl(Mode mode);
   void ScheduleNudgeImpl(const base::TimeDelta& delay,
                          NudgeSource source,
-                         const syncable::ModelTypeBitSet& model_types);
+                         const sessions::TypePayloadMap& types_with_payloads);
   void ScheduleConfigImpl(const base::TimeDelta& delay,
                           const ModelSafeRoutingInfo& routing_info,
                           const std::vector<ModelSafeWorker*>& workers);
diff --git a/chrome/browser/sync/engine/syncer_thread2_unittest.cc b/chrome/browser/sync/engine/syncer_thread2_unittest.cc
index f5adfd7..3b2742c 100644
--- a/chrome/browser/sync/engine/syncer_thread2_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_thread2_unittest.cc
@@ -89,7 +89,7 @@ class SyncerThread2Test : public testing::Test {
       TimeTicks optimal_next_sync = optimal_start + poll_interval * i;
       EXPECT_GE(data[i], optimal_next_sync);
       EXPECT_EQ(GetUpdatesCallerInfo::PERIODIC,
-                records.snapshots[i]->source.first);
+                records.snapshots[i]->source.updates_source);
     }
   }
 
@@ -129,6 +129,21 @@ class SyncerThread2Test : public testing::Test {
     event->Signal();
   }
 
+  // Compare a ModelTyepBitSet to a TypePayloadMap, ignoring payload values.
+  bool CompareModelTypeBitSetToTypePayloadMap(
+      const syncable::ModelTypeBitSet& lhs,
+      const sessions::TypePayloadMap& rhs) {
+    size_t count = 0;
+    for (sessions::TypePayloadMap::const_iterator i = rhs.begin();
+         i != rhs.end(); ++i, ++count) {
+      if (!lhs.test(i->first))
+        return false;
+    }
+    if (lhs.count() != count)
+      return false;
+    return true;
+  }
+
  private:
   scoped_ptr<SyncerThread> syncer_thread_;
   SyncSessionContext* context_;
@@ -170,13 +185,32 @@ TEST_F(SyncerThread2Test, Nudge) {
 
   EXPECT_CALL(*syncer(), SyncShare(_))
       .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
-          WithArg<0>(RecordSyncShare(&records, 1U, &done))));
+          WithArg<0>(RecordSyncShare(&records, 1U, &done))))
+      .RetiresOnSaturation();
   syncer_thread()->ScheduleNudge(zero(), NUDGE_SOURCE_LOCAL, model_types);
   done.TimedWait(timeout());
 
   EXPECT_EQ(1U, records.snapshots.size());
-  EXPECT_EQ(model_types, records.snapshots[0]->source.second);
-  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL, records.snapshots[0]->source.first);
+  EXPECT_TRUE(CompareModelTypeBitSetToTypePayloadMap(model_types,
+      records.snapshots[0]->source.types));
+  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL,
+            records.snapshots[0]->source.updates_source);
+
+  // Make sure a second, later, nudge is unaffected by first (no coalescing).
+  SyncShareRecords records2;
+  model_types[syncable::BOOKMARKS] = false;
+  model_types[syncable::AUTOFILL] = true;
+  EXPECT_CALL(*syncer(), SyncShare(_))
+      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
+          WithArg<0>(RecordSyncShare(&records2, 1U, &done))));
+  syncer_thread()->ScheduleNudge(zero(), NUDGE_SOURCE_LOCAL, model_types);
+  done.TimedWait(timeout());
+
+  EXPECT_EQ(1U, records2.snapshots.size());
+  EXPECT_TRUE(CompareModelTypeBitSetToTypePayloadMap(model_types,
+      records2.snapshots[0]->source.types));
+  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL,
+            records2.snapshots[0]->source.updates_source);
 }
 
 // Test that nudges are coalesced.
@@ -200,8 +234,10 @@ TEST_F(SyncerThread2Test, NudgeCoalescing) {
 
   EXPECT_EQ(1U, r.snapshots.size());
   EXPECT_GE(r.times[0], optimal_time);
-  EXPECT_EQ(types1 | types2 | types3, r.snapshots[0]->source.second);
-  EXPECT_EQ(GetUpdatesCallerInfo::NOTIFICATION, r.snapshots[0]->source.first);
+  EXPECT_TRUE(CompareModelTypeBitSetToTypePayloadMap(types1 | types2 | types3,
+      r.snapshots[0]->source.types));
+  EXPECT_EQ(GetUpdatesCallerInfo::NOTIFICATION,
+            r.snapshots[0]->source.updates_source);
 
   SyncShareRecords r2;
   EXPECT_CALL(*syncer(), SyncShare(_))
@@ -210,8 +246,93 @@ TEST_F(SyncerThread2Test, NudgeCoalescing) {
   syncer_thread()->ScheduleNudge(zero(), NUDGE_SOURCE_NOTIFICATION, types3);
   done.TimedWait(timeout());
   EXPECT_EQ(1U, r2.snapshots.size());
-  EXPECT_EQ(types3, r2.snapshots[0]->source.second);
-  EXPECT_EQ(GetUpdatesCallerInfo::NOTIFICATION, r2.snapshots[0]->source.first);
+  EXPECT_TRUE(CompareModelTypeBitSetToTypePayloadMap(types3,
+      r2.snapshots[0]->source.types));
+  EXPECT_EQ(GetUpdatesCallerInfo::NOTIFICATION,
+            r2.snapshots[0]->source.updates_source);
+}
+
+// Test nudge scheduling.
+TEST_F(SyncerThread2Test, NudgeWithPayloads) {
+  syncer_thread()->Start(SyncerThread::NORMAL_MODE);
+  base::WaitableEvent done(false, false);
+  SyncShareRecords records;
+  sessions::TypePayloadMap model_types_with_payloads;
+  model_types_with_payloads[syncable::BOOKMARKS] = "test";
+
+  EXPECT_CALL(*syncer(), SyncShare(_))
+      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
+          WithArg<0>(RecordSyncShare(&records, 1U, &done))))
+      .RetiresOnSaturation();
+  syncer_thread()->ScheduleNudgeWithPayloads(zero(), NUDGE_SOURCE_LOCAL,
+      model_types_with_payloads);
+  done.TimedWait(timeout());
+
+  EXPECT_EQ(1U, records.snapshots.size());
+  EXPECT_EQ(model_types_with_payloads, records.snapshots[0]->source.types);
+  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL,
+            records.snapshots[0]->source.updates_source);
+
+  // Make sure a second, later, nudge is unaffected by first (no coalescing).
+  SyncShareRecords records2;
+  model_types_with_payloads.erase(syncable::BOOKMARKS);
+  model_types_with_payloads[syncable::AUTOFILL] = "test2";
+  EXPECT_CALL(*syncer(), SyncShare(_))
+      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
+          WithArg<0>(RecordSyncShare(&records2, 1U, &done))));
+  syncer_thread()->ScheduleNudgeWithPayloads(zero(), NUDGE_SOURCE_LOCAL,
+      model_types_with_payloads);
+  done.TimedWait(timeout());
+
+  EXPECT_EQ(1U, records2.snapshots.size());
+  EXPECT_EQ(model_types_with_payloads, records2.snapshots[0]->source.types);
+  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL,
+            records2.snapshots[0]->source.updates_source);
+}
+
+// Test that nudges are coalesced.
+TEST_F(SyncerThread2Test, NudgeWithPayloadsCoalescing) {
+  syncer_thread()->Start(SyncerThread::NORMAL_MODE);
+  base::WaitableEvent done(false, false);
+  SyncShareRecords r;
+  EXPECT_CALL(*syncer(), SyncShare(_))
+      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
+           WithArg<0>(RecordSyncShare(&r, 1U, &done))));
+  sessions::TypePayloadMap types1, types2, types3;
+  types1[syncable::BOOKMARKS] = "test1";
+  types2[syncable::AUTOFILL] = "test2";
+  types3[syncable::THEMES] = "test3";
+  TimeDelta delay = TimeDelta::FromMilliseconds(20);
+  TimeTicks optimal_time = TimeTicks::Now() + delay;
+  syncer_thread()->ScheduleNudgeWithPayloads(delay, NUDGE_SOURCE_UNKNOWN,
+      types1);
+  syncer_thread()->ScheduleNudgeWithPayloads(zero(), NUDGE_SOURCE_LOCAL,
+      types2);
+  syncer_thread()->ScheduleNudgeWithPayloads(zero(), NUDGE_SOURCE_NOTIFICATION,
+      types3);
+  done.TimedWait(timeout());
+
+  EXPECT_EQ(1U, r.snapshots.size());
+  EXPECT_GE(r.times[0], optimal_time);
+  sessions::TypePayloadMap coalesced_types;
+  sessions::CoalescePayloads(&coalesced_types, types1);
+  sessions::CoalescePayloads(&coalesced_types, types2);
+  sessions::CoalescePayloads(&coalesced_types, types3);
+  EXPECT_EQ(coalesced_types, r.snapshots[0]->source.types);
+  EXPECT_EQ(GetUpdatesCallerInfo::NOTIFICATION,
+            r.snapshots[0]->source.updates_source);
+
+  SyncShareRecords r2;
+  EXPECT_CALL(*syncer(), SyncShare(_))
+      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
+           WithArg<0>(RecordSyncShare(&r2, 1U, &done))));
+  syncer_thread()->ScheduleNudgeWithPayloads(zero(), NUDGE_SOURCE_NOTIFICATION,
+      types3);
+  done.TimedWait(timeout());
+  EXPECT_EQ(1U, r2.snapshots.size());
+  EXPECT_EQ(types3, r2.snapshots[0]->source.types);
+  EXPECT_EQ(GetUpdatesCallerInfo::NOTIFICATION,
+            r2.snapshots[0]->source.updates_source);
 }
 
 // Test that polling works as expected.
@@ -413,9 +534,10 @@ TEST_F(SyncerThread2Test, BackoffDropsJobs) {
 
   Mock::VerifyAndClearExpectations(syncer());
   EXPECT_EQ(2U, r.snapshots.size());
-  EXPECT_EQ(GetUpdatesCallerInfo::PERIODIC, r.snapshots[0]->source.first);
+  EXPECT_EQ(GetUpdatesCallerInfo::PERIODIC,
+            r.snapshots[0]->source.updates_source);
   EXPECT_EQ(GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION,
-            r.snapshots[1]->source.first);
+            r.snapshots[1]->source.updates_source);
 
   EXPECT_CALL(*syncer(), SyncShare(_)).Times(1)
       .WillOnce(DoAll(Invoke(sessions::test_util::SimulateCommitFailed),
@@ -430,7 +552,8 @@ TEST_F(SyncerThread2Test, BackoffDropsJobs) {
   Mock::VerifyAndClearExpectations(syncer());
   Mock::VerifyAndClearExpectations(delay());
   EXPECT_EQ(3U, r.snapshots.size());
-  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL, r.snapshots[2]->source.first);
+  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL,
+            r.snapshots[2]->source.updates_source);
 
   EXPECT_CALL(*syncer(), SyncShare(_)).Times(0);
   EXPECT_CALL(*delay(), GetDelay(_)).Times(0);
@@ -511,7 +634,7 @@ TEST_F(SyncerThread2Test, BackoffRelief) {
     EXPECT_GE(r.times[i], optimal_next_sync);
     EXPECT_EQ(i == 0 ? GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION
                      : GetUpdatesCallerInfo::PERIODIC,
-              r.snapshots[i]->source.first);
+              r.snapshots[i]->source.updates_source);
   }
 }
 
diff --git a/chrome/browser/sync/engine/syncer_thread_unittest.cc b/chrome/browser/sync/engine/syncer_thread_unittest.cc
index 9822547..0ac0715 100644
--- a/chrome/browser/sync/engine/syncer_thread_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_thread_unittest.cc
@@ -169,6 +169,34 @@ class SyncerThreadWithSyncerTest : public testing::Test,
     syncer_thread()->SetSyncerShortPollInterval(poll_interval);
   }
 
+  // Compare a provided TypePayloadMap to the pending nudge info stored in the
+  // SyncerThread vault.
+  bool CompareNudgeTypesToVault(const sessions::TypePayloadMap& lhs) {
+    const sessions::TypePayloadMap& vault_nudge_types =
+        syncer_thread()->vault_.pending_nudge_types_;
+    return lhs == vault_nudge_types;
+  }
+
+  // Compare a provided ModelTypeBitset to the pending nudge info stored in the
+  // SyncerThread vault. Nudge info in vault must not have any non-empty
+  // payloads.
+  bool CompareNudgeTypesBitSetToVault(const syncable::ModelTypeBitSet& lhs) {
+    sessions::TypePayloadMap model_types_with_payloads =
+        sessions::MakeTypePayloadMapFromBitSet(lhs, std::string());
+    size_t count = 0;
+    for (sessions::TypePayloadMap::const_iterator i =
+             syncer_thread()->vault_.pending_nudge_types_.begin();
+         i != syncer_thread()->vault_.pending_nudge_types_.end();
+         ++i, ++count) {
+      if (!lhs.test(i->first))
+        return false;
+    }
+    if (lhs.count() != count)
+      return false;
+    return true;
+  }
+
+
  private:
 
   virtual void OnSyncEngineEvent(const SyncEngineEvent& event) {
@@ -468,7 +496,6 @@ TEST_F(SyncerThreadTest, CalculatePollingWaitTime) {
   }
 
   {
-
     // Now try with unsynced local items.
     context->set_last_snapshot(SessionSnapshotForTest(0, 1));
     bool continue_sync_cycle_param = false;
@@ -512,7 +539,6 @@ TEST_F(SyncerThreadTest, CalculatePollingWaitTime) {
 
   // Regression for exponential backoff reset when the syncer is nudged.
   {
-
     context->set_last_snapshot(SessionSnapshotForTest(0, 1));
     bool continue_sync_cycle_param = false;
 
@@ -732,7 +758,7 @@ TEST_F(SyncerThreadWithSyncerTest, NudgeWithDataTypes) {
   syncer_thread()->NudgeSyncerWithDataTypes(5,
       SyncerThread::kUnknown,
       model_types);
-  EXPECT_EQ(model_types, syncer_thread()->vault_.pending_nudge_types_);
+  EXPECT_TRUE(CompareNudgeTypesBitSetToVault(model_types));
   syncer_thread()->RequestResume();
 
   interceptor.WaitForSyncShare(1, TimeDelta::FromSeconds(1));
@@ -741,7 +767,7 @@ TEST_F(SyncerThreadWithSyncerTest, NudgeWithDataTypes) {
 
   // SyncerThread should be waiting again.  Signal it to stop.
   EXPECT_TRUE(syncer_thread()->Stop(2000));
-  EXPECT_TRUE(syncer_thread()->vault_.pending_nudge_types_.none());
+  EXPECT_TRUE(syncer_thread()->vault_.pending_nudge_types_.empty());
 }
 
 TEST_F(SyncerThreadWithSyncerTest, NudgeWithDataTypesCoalesced) {
@@ -767,7 +793,7 @@ TEST_F(SyncerThreadWithSyncerTest, NudgeWithDataTypesCoalesced) {
   syncer_thread()->NudgeSyncerWithDataTypes(100,
       SyncerThread::kUnknown,
       model_types);
-  EXPECT_EQ(model_types, syncer_thread()->vault_.pending_nudge_types_);
+  EXPECT_TRUE(CompareNudgeTypesBitSetToVault(model_types));
 
   model_types[syncable::BOOKMARKS] = false;
   model_types[syncable::AUTOFILL] = true;
@@ -777,7 +803,88 @@ TEST_F(SyncerThreadWithSyncerTest, NudgeWithDataTypesCoalesced) {
 
   // Reset BOOKMARKS for expectations.
   model_types[syncable::BOOKMARKS] = true;
-  EXPECT_EQ(model_types, syncer_thread()->vault_.pending_nudge_types_);
+  EXPECT_TRUE(CompareNudgeTypesBitSetToVault(model_types));
+
+  syncer_thread()->RequestResume();
+
+  interceptor.WaitForSyncShare(1, TimeDelta::FromSeconds(1));
+  EXPECT_EQ(static_cast<unsigned int>(2),
+      interceptor.times_sync_occured().size());
+
+  // SyncerThread should be waiting again.  Signal it to stop.
+  EXPECT_TRUE(syncer_thread()->Stop(2000));
+  EXPECT_TRUE(syncer_thread()->vault_.pending_nudge_types_.empty());
+}
+
+TEST_F(SyncerThreadWithSyncerTest, NudgeWithPayloads) {
+  SyncShareIntercept interceptor;
+  connection()->SetMidCommitObserver(&interceptor);
+  // We don't want a poll to happen during this test (except the first one).
+  PreventThreadFromPolling();
+  EXPECT_TRUE(syncer_thread()->Start());
+  metadb()->Open();
+  syncer_thread()->CreateSyncer(metadb()->name());
+  const TimeDelta poll_interval = TimeDelta::FromMinutes(5);
+  interceptor.WaitForSyncShare(1, poll_interval + poll_interval);
+  EXPECT_EQ(static_cast<unsigned int>(1),
+            interceptor.times_sync_occured().size());
+
+  // The SyncerThread should be waiting for the poll now.  Nudge it to sync
+  // immediately (5ms).
+  sessions::TypePayloadMap nudge_types;
+  nudge_types[syncable::BOOKMARKS] = "test";
+
+  // Paused so we can verify the nudge types safely.
+  syncer_thread()->RequestPause();
+  syncer_thread()->NudgeSyncerWithPayloads(5,
+      SyncerThread::kUnknown,
+      nudge_types);
+  EXPECT_TRUE(CompareNudgeTypesToVault(nudge_types));
+  syncer_thread()->RequestResume();
+
+  interceptor.WaitForSyncShare(1, TimeDelta::FromSeconds(1));
+  EXPECT_EQ(static_cast<unsigned int>(2),
+      interceptor.times_sync_occured().size());
+
+  // SyncerThread should be waiting again.  Signal it to stop.
+  EXPECT_TRUE(syncer_thread()->Stop(2000));
+  EXPECT_TRUE(syncer_thread()->vault_.pending_nudge_types_.empty());
+}
+
+TEST_F(SyncerThreadWithSyncerTest, NudgeWithPayloadsCoalesced) {
+  SyncShareIntercept interceptor;
+  connection()->SetMidCommitObserver(&interceptor);
+  // We don't want a poll to happen during this test (except the first one).
+  PreventThreadFromPolling();
+  EXPECT_TRUE(syncer_thread()->Start());
+  metadb()->Open();
+  syncer_thread()->CreateSyncer(metadb()->name());
+  const TimeDelta poll_interval = TimeDelta::FromMinutes(5);
+  interceptor.WaitForSyncShare(1, poll_interval + poll_interval);
+  EXPECT_EQ(static_cast<unsigned int>(1),
+    interceptor.times_sync_occured().size());
+
+  // The SyncerThread should be waiting for the poll now.  Nudge it to sync
+  // immediately (5ms).
+  sessions::TypePayloadMap nudge_types;
+  nudge_types[syncable::BOOKMARKS] = "books";
+
+  // Paused so we can verify the nudge types safely.
+  syncer_thread()->RequestPause();
+  syncer_thread()->NudgeSyncerWithPayloads(100,
+      SyncerThread::kUnknown,
+      nudge_types);
+  EXPECT_TRUE(CompareNudgeTypesToVault(nudge_types));
+
+  nudge_types.erase(syncable::BOOKMARKS);
+  nudge_types[syncable::AUTOFILL] = "auto";
+  syncer_thread()->NudgeSyncerWithPayloads(0,
+      SyncerThread::kUnknown,
+      nudge_types);
+
+  // Reset BOOKMARKS for expectations.
+  nudge_types[syncable::BOOKMARKS] = "books";
+  EXPECT_TRUE(CompareNudgeTypesToVault(nudge_types));
 
   syncer_thread()->RequestResume();
 
@@ -787,7 +894,7 @@ TEST_F(SyncerThreadWithSyncerTest, NudgeWithDataTypesCoalesced) {
 
   // SyncerThread should be waiting again.  Signal it to stop.
   EXPECT_TRUE(syncer_thread()->Stop(2000));
-  EXPECT_TRUE(syncer_thread()->vault_.pending_nudge_types_.none());
+  EXPECT_TRUE(syncer_thread()->vault_.pending_nudge_types_.empty());
 }
 
 TEST_F(SyncerThreadWithSyncerTest, Throttling) {
diff --git a/chrome/browser/sync/engine/syncer_unittest.cc b/chrome/browser/sync/engine/syncer_unittest.cc
index 883b30b..9af46c6 100644
--- a/chrome/browser/sync/engine/syncer_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_unittest.cc
@@ -5,6 +5,8 @@
 // Syncer unit tests. Unfortunately a lot of these tests
 // are outdated and need to be reworked and updated.
 
+#include <algorithm>
+#include <limits>
 #include <list>
 #include <map>
 #include <set>
@@ -25,6 +27,7 @@
 #include "chrome/browser/sync/engine/syncproto.h"
 #include "chrome/browser/sync/protocol/sync.pb.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
+#include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/syncable/syncable.h"
 #include "chrome/common/deprecated/event_sys-inl.h"
 #include "chrome/test/sync/engine/mock_connection_manager.h"
@@ -150,8 +153,11 @@ class SyncerTest : public testing::Test,
     std::vector<ModelSafeWorker*> workers;
     GetModelSafeRoutingInfo(&info);
     GetWorkers(&workers);
-    return new SyncSession(context_.get(), this, sessions::SyncSourceInfo(),
-                           info, workers);
+    sessions::TypePayloadMap types =
+        sessions::MakeTypePayloadMapFromRoutingInfo(info, std::string());
+    return new SyncSession(context_.get(), this,
+        sessions::SyncSourceInfo(sync_pb::GetUpdatesCallerInfo::UNKNOWN, types),
+        info, workers);
   }
 
   bool SyncShareAsDelegate() {
diff --git a/chrome/browser/sync/notifier/chrome_invalidation_client.cc b/chrome/browser/sync/notifier/chrome_invalidation_client.cc
index 7b17f9a..5d0cfd2a 100644
--- a/chrome/browser/sync/notifier/chrome_invalidation_client.cc
+++ b/chrome/browser/sync/notifier/chrome_invalidation_client.cc
@@ -126,13 +126,18 @@ void ChromeInvalidationClient::Invalidate(
   VLOG(1) << "Invalidate: " << InvalidationToString(invalidation);
   syncable::ModelType model_type;
   if (ObjectIdToRealModelType(invalidation.object_id(), &model_type)) {
+    std::string payload;
+    // payload() CHECK()'s has_payload(), so we must check it ourselves first.
+    if (invalidation.has_payload())
+      payload = invalidation.payload();
+
     // TODO(akalin): This is a hack to make new sync data types work
     // with server-issued notifications.  Remove this when it's not
     // needed anymore.
     if (model_type == syncable::UNSPECIFIED) {
       listener_->OnInvalidateAll();
     } else {
-      listener_->OnInvalidate(model_type);
+      listener_->OnInvalidate(model_type, payload);
     }
   } else {
     LOG(WARNING) << "Could not get invalidation model type; "
diff --git a/chrome/browser/sync/notifier/chrome_invalidation_client.h b/chrome/browser/sync/notifier/chrome_invalidation_client.h
index 3d4e1d7..873b877 100644
--- a/chrome/browser/sync/notifier/chrome_invalidation_client.h
+++ b/chrome/browser/sync/notifier/chrome_invalidation_client.h
@@ -41,7 +41,8 @@ class ChromeInvalidationClient
    public:
     virtual ~Listener();
 
-    virtual void OnInvalidate(syncable::ModelType model_type) = 0;
+    virtual void OnInvalidate(syncable::ModelType model_type,
+                              const std::string& payload) = 0;
 
     virtual void OnInvalidateAll() = 0;
   };
diff --git a/chrome/browser/sync/notifier/chrome_invalidation_client_unittest.cc b/chrome/browser/sync/notifier/chrome_invalidation_client_unittest.cc
index 809e32a..13438b9 100644
--- a/chrome/browser/sync/notifier/chrome_invalidation_client_unittest.cc
+++ b/chrome/browser/sync/notifier/chrome_invalidation_client_unittest.cc
@@ -18,7 +18,8 @@ using ::testing::Return;
 
 class MockListener : public ChromeInvalidationClient::Listener {
  public:
-  MOCK_METHOD1(OnInvalidate, void(syncable::ModelType));
+  MOCK_METHOD2(OnInvalidate, void(syncable::ModelType,
+                                  const std::string& payload));
   MOCK_METHOD0(OnInvalidateAll, void());
 };
 
diff --git a/chrome/browser/sync/notifier/server_notifier_thread.cc b/chrome/browser/sync/notifier/server_notifier_thread.cc
index ae15943..1c45beb 100644
--- a/chrome/browser/sync/notifier/server_notifier_thread.cc
+++ b/chrome/browser/sync/notifier/server_notifier_thread.cc
@@ -66,7 +66,9 @@ void ServerNotifierThread::SendNotification(
                   "used";
 }
 
-void ServerNotifierThread::OnInvalidate(syncable::ModelType model_type) {
+void ServerNotifierThread::OnInvalidate(
+    syncable::ModelType model_type,
+    const std::string& payload) {
   DCHECK_EQ(MessageLoop::current(), worker_message_loop());
   DCHECK_GE(model_type, syncable::FIRST_REAL_MODEL_TYPE);
   DCHECK_LT(model_type, syncable::MODEL_TYPE_COUNT);
@@ -75,7 +77,8 @@ void ServerNotifierThread::OnInvalidate(syncable::ModelType model_type) {
   syncable::ModelTypeBitSet model_types;
   model_types[model_type] = true;
   IncomingNotificationData notification_data;
-  notification_data.service_specific_data = model_types.to_string();
+  notification_data.service_url = model_types.to_string();
+  notification_data.service_specific_data = payload;
   observers_->Notify(&Observer::OnIncomingNotification, notification_data);
 }
 
diff --git a/chrome/browser/sync/notifier/server_notifier_thread.h b/chrome/browser/sync/notifier/server_notifier_thread.h
index 55b60fe..b0b9819 100644
--- a/chrome/browser/sync/notifier/server_notifier_thread.h
+++ b/chrome/browser/sync/notifier/server_notifier_thread.h
@@ -59,7 +59,14 @@ class ServerNotifierThread
   virtual void SendNotification(const OutgoingNotificationData& data);
 
   // ChromeInvalidationClient::Listener implementation.
-  virtual void OnInvalidate(syncable::ModelType model_type);
+  // We pass on two pieces of information to observers through the
+  // IncomingNotificationData.
+  // - the model type being invalidated, through IncomingNotificationData's
+  //       service_url.
+  // - the invalidation payload for that model type, through
+  //       IncomingNotificationData's service_specific_data.
+  virtual void OnInvalidate(syncable::ModelType model_type,
+                            const std::string& payload);
   virtual void OnInvalidateAll();
 
   // StateWriter implementation.
diff --git a/chrome/browser/sync/sessions/session_state.cc b/chrome/browser/sync/sessions/session_state.cc
index fda99b8..112dc2f 100644
--- a/chrome/browser/sync/sessions/session_state.cc
+++ b/chrome/browser/sync/sessions/session_state.cc
@@ -13,6 +13,54 @@ using std::vector;
 namespace browser_sync {
 namespace sessions {
 
+TypePayloadMap MakeTypePayloadMapFromBitSet(
+    const syncable::ModelTypeBitSet& types,
+    const std::string& payload) {
+  TypePayloadMap types_with_payloads;
+  for (size_t i = syncable::FIRST_REAL_MODEL_TYPE;
+       i < types.size(); ++i) {
+    if (types[i]) {
+      types_with_payloads[syncable::ModelTypeFromInt(i)] = payload;
+    }
+  }
+  return types_with_payloads;
+}
+
+TypePayloadMap MakeTypePayloadMapFromRoutingInfo(
+    const ModelSafeRoutingInfo& routes,
+    const std::string& payload) {
+  TypePayloadMap types_with_payloads;
+  for (ModelSafeRoutingInfo::const_iterator i = routes.begin();
+       i != routes.end(); ++i) {
+    types_with_payloads[i->first] = payload;
+  }
+  return types_with_payloads;
+}
+
+void CoalescePayloads(TypePayloadMap* original,
+                      const TypePayloadMap& update) {
+  for (TypePayloadMap::const_iterator i = update.begin();
+       i != update.end(); ++i) {
+    if (original->count(i->first) == 0) {
+      // If this datatype isn't already in our map, add it with whatever payload
+      // it has.
+      (*original)[i->first] = i->second;
+    } else if (i->second.length() > 0) {
+      // If this datatype is already in our map, we only overwrite the payload
+      // if the new one is non-empty.
+      (*original)[i->first] = i->second;
+    }
+  }
+}
+
+SyncSourceInfo::SyncSourceInfo()
+    : updates_source(sync_pb::GetUpdatesCallerInfo::UNKNOWN) {}
+
+SyncSourceInfo::SyncSourceInfo(
+    const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource& u,
+    const TypePayloadMap& t)
+    : updates_source(u), types(t) {}
+
 SyncerStatus::SyncerStatus()
     : invalid_store(false),
       syncer_stuck(false),
diff --git a/chrome/browser/sync/sessions/session_state.h b/chrome/browser/sync/sessions/session_state.h
index 2d8381e3..a656316 100644
--- a/chrome/browser/sync/sessions/session_state.h
+++ b/chrome/browser/sync/sessions/session_state.h
@@ -32,8 +32,38 @@ namespace sessions {
 
 class UpdateProgress;
 
-typedef std::pair<sync_pb::GetUpdatesCallerInfo::GetUpdatesSource,
-    syncable::ModelTypeBitSet> SyncSourceInfo;
+// A container that contains a set of datatypes with possible string payloads.
+typedef std::map<syncable::ModelType, std::string> TypePayloadMap;
+
+// Helper utils for building TypePayloadMaps.
+// Make a TypePayloadMap from all the types in a ModelTypeBitSet using a
+// default payload.
+TypePayloadMap MakeTypePayloadMapFromBitSet(
+    const syncable::ModelTypeBitSet& types,
+    const std::string& payload);
+
+// Make a TypePayloadMap for all the enabled types in a ModelSafeRoutingInfo
+// using a default payload.
+TypePayloadMap MakeTypePayloadMapFromRoutingInfo(
+    const ModelSafeRoutingInfo& routes,
+    const std::string& payload);
+
+// Coalesce |update| into |original|, overwriting only when |update| has
+// a non-empty payload.
+void CoalescePayloads(TypePayloadMap* original, const TypePayloadMap& update);
+
+// A container for the source of a sync session. This includes the update
+// source, the datatypes triggering the sync session, and possible session
+// specific payloads which should be sent to the server.
+struct SyncSourceInfo {
+  SyncSourceInfo();
+  SyncSourceInfo(
+      const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource& u,
+      const TypePayloadMap& t);
+
+  sync_pb::GetUpdatesCallerInfo::GetUpdatesSource updates_source;
+  TypePayloadMap types;
+};
 
 // Data pertaining to the status of an active Syncer object.
 struct SyncerStatus {
diff --git a/chrome/browser/sync/sessions/sync_session.cc b/chrome/browser/sync/sessions/sync_session.cc
index 077b629..52a44fa 100644
--- a/chrome/browser/sync/sessions/sync_session.cc
+++ b/chrome/browser/sync/sessions/sync_session.cc
@@ -30,8 +30,10 @@ void SyncSession::Coalesce(const SyncSession& session) {
     return;
   }
 
-  source_ = SyncSourceInfo(session.source_.first,
-                           source_.second | session.source_.second);
+  // When we coalesce sessions, the sync update source gets overwritten with the
+  // most recent, while the type/payload map gets merged.
+  CoalescePayloads(&source_.types, session.source_.types);
+  source_.updates_source = session.source_.updates_source;
 
   std::vector<ModelSafeWorker*> temp;
   std::set_union(workers_.begin(), workers_.end(),
@@ -89,7 +91,7 @@ SyncSourceInfo SyncSession::TestAndSetSource() {
   SyncSourceInfo old_source = source_;
   source_ = SyncSourceInfo(
       sync_pb::GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION,
-      source_.second);
+      source_.types);
   return old_source;
 }
 
diff --git a/chrome/browser/sync/sessions/sync_session.h b/chrome/browser/sync/sessions/sync_session.h
index 3626d11..693d652 100644
--- a/chrome/browser/sync/sessions/sync_session.h
+++ b/chrome/browser/sync/sessions/sync_session.h
@@ -15,16 +15,20 @@
 #define CHROME_BROWSER_SYNC_SESSIONS_SYNC_SESSION_H_
 #pragma once
 
+#include <map>
+#include <string>
 #include <utility>
 #include <vector>
 
 #include "base/basictypes.h"
 #include "base/scoped_ptr.h"
 #include "base/time.h"
+#include "chrome/browser/sync/engine/model_safe_worker.h"
 #include "chrome/browser/sync/sessions/ordered_commit_set.h"
 #include "chrome/browser/sync/sessions/session_state.h"
 #include "chrome/browser/sync/sessions/status_controller.h"
 #include "chrome/browser/sync/sessions/sync_session_context.h"
+#include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/util/extensions_activity_monitor.h"
 
 namespace syncable {
diff --git a/chrome/browser/sync/sessions/sync_session_unittest.cc b/chrome/browser/sync/sessions/sync_session_unittest.cc
index 681c0ad..0f433de 100644
--- a/chrome/browser/sync/sessions/sync_session_unittest.cc
+++ b/chrome/browser/sync/sessions/sync_session_unittest.cc
@@ -10,6 +10,7 @@
 #include "chrome/browser/sync/engine/syncer_types.h"
 #include "chrome/browser/sync/engine/syncer_util.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
+#include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/syncable/syncable.h"
 #include "chrome/test/sync/engine/test_directory_setter_upper.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -265,10 +266,15 @@ TEST_F(SyncSessionTest, ResetTransientState) {
 TEST_F(SyncSessionTest, Coalesce) {
   std::vector<ModelSafeWorker*> workers_one, workers_two;
   ModelSafeRoutingInfo routes_one, routes_two;
-  SyncSourceInfo source_one(sync_pb::GetUpdatesCallerInfo::PERIODIC,
-                            ParamsMeaningJustOneEnabledType());
-  SyncSourceInfo source_two(sync_pb::GetUpdatesCallerInfo::LOCAL,
-                            ParamsMeaningAllEnabledTypes());
+  TypePayloadMap one_type = sessions::MakeTypePayloadMapFromBitSet(
+      ParamsMeaningJustOneEnabledType(),
+      std::string());;
+  TypePayloadMap all_types = sessions::MakeTypePayloadMapFromBitSet(
+      ParamsMeaningAllEnabledTypes(),
+      std::string());;
+  SyncSourceInfo source_one(sync_pb::GetUpdatesCallerInfo::PERIODIC, one_type);
+  SyncSourceInfo source_two(sync_pb::GetUpdatesCallerInfo::LOCAL, all_types);
+
   scoped_refptr<MockDBModelWorker> db_worker(new MockDBModelWorker());
   scoped_refptr<MockUIModelWorker> ui_worker(new MockUIModelWorker());
   workers_one.push_back(db_worker);
@@ -282,8 +288,8 @@ TEST_F(SyncSessionTest, Coalesce) {
 
   one.Coalesce(two);
 
-  EXPECT_EQ(two.source().first, one.source().first);
-  EXPECT_EQ(ParamsMeaningAllEnabledTypes(), one.source().second);
+  EXPECT_EQ(two.source().updates_source, one.source().updates_source);
+  EXPECT_EQ(all_types, one.source().types);
   std::vector<ModelSafeWorker*>::const_iterator it_db =
       std::find(one.workers().begin(), one.workers().end(), db_worker);
   std::vector<ModelSafeWorker*>::const_iterator it_ui =
@@ -293,6 +299,64 @@ TEST_F(SyncSessionTest, Coalesce) {
   EXPECT_EQ(routes_two, one.routing_info());
 }
 
+TEST_F(SyncSessionTest, MakeTypePayloadMapFromBitSet) {
+  syncable::ModelTypeBitSet types;
+  std::string payload = "test";
+  TypePayloadMap types_with_payloads = MakeTypePayloadMapFromBitSet(types,
+      payload);
+  EXPECT_TRUE(types_with_payloads.empty());
+
+  types[syncable::BOOKMARKS] = true;
+  types[syncable::PASSWORDS] = true;
+  types[syncable::AUTOFILL] = true;
+  payload = "test2";
+  types_with_payloads = MakeTypePayloadMapFromBitSet(types, payload);
+
+  ASSERT_EQ(3U, types_with_payloads.size());
+  EXPECT_EQ(types_with_payloads[syncable::BOOKMARKS], payload);
+  EXPECT_EQ(types_with_payloads[syncable::PASSWORDS], payload);
+  EXPECT_EQ(types_with_payloads[syncable::AUTOFILL], payload);
+}
+
+TEST_F(SyncSessionTest, MakeTypePayloadMapFromRoutingInfo) {
+  std::string payload = "test";
+  TypePayloadMap types_with_payloads
+      = MakeTypePayloadMapFromRoutingInfo(routes_, payload);
+  ASSERT_EQ(routes_.size(), types_with_payloads.size());
+  for (ModelSafeRoutingInfo::iterator iter = routes_.begin();
+       iter != routes_.end();
+       ++iter) {
+    EXPECT_EQ(payload, types_with_payloads[iter->first]);
+  }
+}
+
+TEST_F(SyncSessionTest, CoalescePayloads) {
+  TypePayloadMap original;
+  std::string empty_payload;
+  std::string payload1 = "payload1";
+  std::string payload2 = "payload2";
+  std::string payload3 = "payload3";
+  original[syncable::BOOKMARKS] = empty_payload;
+  original[syncable::PASSWORDS] = payload1;
+  original[syncable::AUTOFILL] = payload2;
+  original[syncable::THEMES] = payload3;
+
+  TypePayloadMap update;
+  update[syncable::BOOKMARKS] = empty_payload;  // Same.
+  update[syncable::PASSWORDS] = empty_payload;  // Overwrite with empty.
+  update[syncable::AUTOFILL] = payload1;        // Overwrite with non-empty.
+  update[syncable::SESSIONS] = payload2;        // New.
+  // Themes untouched.
+
+  CoalescePayloads(&original, update);
+  ASSERT_EQ(5U, original.size());
+  EXPECT_EQ(empty_payload, original[syncable::BOOKMARKS]);
+  EXPECT_EQ(payload1, original[syncable::PASSWORDS]);
+  EXPECT_EQ(payload1, original[syncable::AUTOFILL]);
+  EXPECT_EQ(payload2, original[syncable::SESSIONS]);
+  EXPECT_EQ(payload3, original[syncable::THEMES]);
+}
+
 }  // namespace
 }  // namespace sessions
 }  // namespace browser_sync
diff --git a/chrome/browser/sync/tools/sync_listen_notifications.cc b/chrome/browser/sync/tools/sync_listen_notifications.cc
index 842c230..a483eb0 100644
--- a/chrome/browser/sync/tools/sync_listen_notifications.cc
+++ b/chrome/browser/sync/tools/sync_listen_notifications.cc
@@ -153,9 +153,10 @@ class ChromeInvalidationListener
  public:
   ChromeInvalidationListener() {}
 
-  virtual void OnInvalidate(syncable::ModelType model_type) {
+  virtual void OnInvalidate(syncable::ModelType model_type,
+                            const std::string& payload) {
     LOG(INFO) << "OnInvalidate: "
-              << syncable::ModelTypeToString(model_type);
+              << syncable::ModelTypeToString(model_type) << " - " << payload;
     // A real implementation would respond to the invalidation.
   }