[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/6182004

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

16 files changed, 386 insertions, 59 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..b6868c0 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::ModelTypeBitSetToTypePayloadMap(
+            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::RoutingInfoToTypePayloadMap(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..db35a7f 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::ModelTypeBitSetToTypePayloadMap(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::RoutingInfoToTypePayloadMap(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::RoutingInfoToTypePayloadMap(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_thread_unittest.cc b/chrome/browser/sync/engine/syncer_thread_unittest.cc
index 9822547..5fbd818 100644
--- a/chrome/browser/sync/engine/syncer_thread_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_thread_unittest.cc
@@ -169,6 +169,24 @@ class SyncerThreadWithSyncerTest : public testing::Test,
     syncer_thread()->SetSyncerShortPollInterval(poll_interval);
   }
 
+  // Compare a provided SyncSouceInfo::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::ModelTypeBitSetToTypePayloadMap(lhs, std::string());
+    return CompareNudgeTypesToVault(model_types_with_payloads);
+  }
+
+
  private:
 
   virtual void OnSyncEngineEvent(const SyncEngineEvent& event) {
@@ -468,7 +486,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 +529,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 +748,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 +757,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 +783,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 +793,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 +884,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..14fcb45 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::RoutingInfoToTypePayloadMap(
+        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/sync_session.cc b/chrome/browser/sync/sessions/sync_session.cc
index 077b629..c5876f9 100644
--- a/chrome/browser/sync/sessions/sync_session.cc
+++ b/chrome/browser/sync/sessions/sync_session.cc
@@ -9,6 +9,53 @@
 namespace browser_sync {
 namespace sessions {
 
+TypePayloadMap ModelTypeBitSetToTypePayloadMap(
+    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 RoutingInfoToTypePayloadMap(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) {}
+
 SyncSession::SyncSession(SyncSessionContext* context, Delegate* delegate,
     SyncSourceInfo source,
     const ModelSafeRoutingInfo& routing_info,
@@ -89,7 +136,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..fb605f1 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 {
@@ -36,6 +40,35 @@ class ModelSafeWorker;
 
 namespace sessions {
 
+// 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.
+// Convert a ModelTypeBitset into a TypePayloadMap using a default payload.
+TypePayloadMap ModelTypeBitSetToTypePayloadMap(
+    const syncable::ModelTypeBitSet& types,
+    const std::string& payload);
+// Convert a ModelSafeRoutingInfo into a TypePayloadMap using a default payload.
+TypePayloadMap RoutingInfoToTypePayloadMap(
+    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;
+};
+
 class SyncSession {
  public:
   // The Delegate services events that occur during the session requiring an
diff --git a/chrome/browser/sync/sessions/sync_session_unittest.cc b/chrome/browser/sync/sessions/sync_session_unittest.cc
index 681c0ad..57ea030 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"
@@ -252,6 +253,64 @@ TEST_F(SyncSessionTest, MoreToSyncIfConflictsResolved) {
   EXPECT_TRUE(session_->HasMoreToSync());
 }
 
+TEST_F(SyncSessionTest, ModelTypeBitSetToTypePayloadMap) {
+  syncable::ModelTypeBitSet types;
+  std::string payload = "test";
+  TypePayloadMap types_with_payloads =
+      ModelTypeBitSetToTypePayloadMap(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 = ModelTypeBitSetToTypePayloadMap(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, RoutingInfoToTypePayloadMap) {
+  std::string payload = "test";
+  TypePayloadMap types_with_payloads
+      = RoutingInfoToTypePayloadMap(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]);
+}
+
 TEST_F(SyncSessionTest, ResetTransientState) {
   status()->update_conflicts_resolved(true);
   status()->increment_num_successful_commits();
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.
   }