Finish commit 146665: [Sync] Refactor sync configuration logic.

Only the last patchset (which was a partial diff) was committed..

Original codereview at: https://chromiumcodereview.appspot.com/10780002/

TBR=rlarocque@chromium.org
BUG=129665
TEST=

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

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

12 files changed, 614 insertions, 256 deletions

diff --git a/sync/engine/sync_scheduler.cc b/sync/engine/sync_scheduler.cc
index f55627d..04b30f0 100644
--- a/sync/engine/sync_scheduler.cc
+++ b/sync/engine/sync_scheduler.cc
@@ -65,6 +65,24 @@ bool IsActionableError(
 }
 }  // namespace
 
+ConfigurationParams::ConfigurationParams()
+    : source(GetUpdatesCallerInfo::UNKNOWN),
+      keystore_key_status(KEYSTORE_KEY_UNNECESSARY) {}
+ConfigurationParams::ConfigurationParams(
+    const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource& source,
+    const syncer::ModelTypeSet& types_to_download,
+    const syncer::ModelSafeRoutingInfo& routing_info,
+    KeystoreKeyStatus keystore_key_status,
+    const base::Closure& ready_task)
+    : source(source),
+      types_to_download(types_to_download),
+      routing_info(routing_info),
+      keystore_key_status(keystore_key_status),
+      ready_task(ready_task) {
+  DCHECK(!ready_task.is_null());
+}
+ConfigurationParams::~ConfigurationParams() {}
+
 SyncScheduler::DelayProvider::DelayProvider() {}
 SyncScheduler::DelayProvider::~DelayProvider() {}
 
@@ -96,12 +114,16 @@ SyncScheduler::SyncSessionJob::~SyncSessionJob() {}
 
 SyncScheduler::SyncSessionJob::SyncSessionJob(SyncSessionJobPurpose purpose,
     base::TimeTicks start,
-    linked_ptr<sessions::SyncSession> session, bool is_canary_job,
-    const tracked_objects::Location& from_here) : purpose(purpose),
-        scheduled_start(start),
-        session(session),
-        is_canary_job(is_canary_job),
-        from_here(from_here) {
+    linked_ptr<sessions::SyncSession> session,
+    bool is_canary_job,
+    const ConfigurationParams& config_params,
+    const tracked_objects::Location& from_here)
+    : purpose(purpose),
+      scheduled_start(start),
+      session(session),
+      is_canary_job(is_canary_job),
+      config_params(config_params),
+      from_here(from_here) {
 }
 
 const char* SyncScheduler::SyncSessionJob::GetPurposeString(
@@ -243,7 +265,7 @@ void SyncScheduler::UpdateServerConnectionManagerStatus(
   connection_code_ = code;
 }
 
-void SyncScheduler::Start(Mode mode, const base::Closure& callback) {
+void SyncScheduler::Start(Mode mode) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   std::string thread_name = MessageLoop::current()->thread_name();
   if (thread_name.empty())
@@ -260,8 +282,6 @@ void SyncScheduler::Start(Mode mode, const base::Closure& callback) {
   Mode old_mode = mode_;
   mode_ = mode;
   AdjustPolling(NULL);  // Will kick start poll timer if needed.
-  if (!callback.is_null())
-    callback.Run();
 
   if (old_mode != mode_) {
     // We just changed our mode. See if there are any pending jobs that we could
@@ -280,6 +300,112 @@ void SyncScheduler::SendInitialSnapshot() {
   session_context_->NotifyListeners(event);
 }
 
+namespace {
+
+// Helper to extract the routing info and workers corresponding to types in
+// |types| from |current_routes| and |current_workers|.
+void BuildModelSafeParams(
+    const ModelTypeSet& types_to_download,
+    const ModelSafeRoutingInfo& current_routes,
+    const std::vector<ModelSafeWorker*>& current_workers,
+    ModelSafeRoutingInfo* result_routes,
+    std::vector<ModelSafeWorker*>* result_workers) {
+  std::set<ModelSafeGroup> active_groups;
+  active_groups.insert(GROUP_PASSIVE);
+  for (ModelTypeSet::Iterator iter = types_to_download.First(); iter.Good();
+       iter.Inc()) {
+    syncer::ModelType type = iter.Get();
+    ModelSafeRoutingInfo::const_iterator route = current_routes.find(type);
+    DCHECK(route != current_routes.end());
+    ModelSafeGroup group = route->second;
+    (*result_routes)[type] = group;
+    active_groups.insert(group);
+  }
+
+  for(std::vector<ModelSafeWorker*>::const_iterator iter =
+          current_workers.begin(); iter != current_workers.end(); ++iter) {
+    if (active_groups.count((*iter)->GetModelSafeGroup()) > 0)
+      result_workers->push_back(*iter);
+  }
+}
+
+}  // namespace.
+
+bool SyncScheduler::ScheduleConfiguration(const ConfigurationParams& params) {
+  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
+  DCHECK_EQ(CONFIGURATION_MODE, mode_);
+  DCHECK(!params.ready_task.is_null());
+  SDVLOG(2) << "Reconfiguring syncer.";
+
+  // Only one configuration is allowed at a time. Verify we're not waiting
+  // for a pending configure job.
+  DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job.get());
+
+  // TODO(sync): now that ModelChanging commands only use those workers within
+  // the routing info, we don't really need |restricted_workers|. Remove it.
+  // crbug.com/133030
+  syncer::ModelSafeRoutingInfo restricted_routes;
+  std::vector<ModelSafeWorker*> restricted_workers;
+  BuildModelSafeParams(params.types_to_download,
+                       params.routing_info,
+                       session_context_->workers(),
+                       &restricted_routes,
+                       &restricted_workers);
+  session_context_->set_routing_info(params.routing_info);
+
+  // We rely on this not failing, so don't need to worry about checking for
+  // success. In addition, this will be removed as part of crbug.com/131433.
+  SyncSessionJob cleanup_job(
+      SyncSessionJob::CLEANUP_DISABLED_TYPES,
+      TimeTicks::Now(),
+      make_linked_ptr(CreateSyncSession(SyncSourceInfo())),
+      false,
+      ConfigurationParams(),
+      FROM_HERE);
+  DoSyncSessionJob(cleanup_job);
+
+  if (params.keystore_key_status == ConfigurationParams::KEYSTORE_KEY_NEEDED) {
+    // TODO(zea): implement in such a way that we can handle failures and the
+    // subsequent retrys the scheduler might perform. See crbug.com/129665.
+    NOTIMPLEMENTED();
+  }
+
+  // Only reconfigure if we have types to download.
+  if (!params.types_to_download.Empty()) {
+    DCHECK(!restricted_routes.empty());
+    linked_ptr<SyncSession> session(new SyncSession(
+        session_context_,
+        this,
+        SyncSourceInfo(params.source,
+                       ModelSafeRoutingInfoToPayloadMap(
+                           restricted_routes,
+                           std::string())),
+        restricted_routes,
+        restricted_workers));
+    SyncSessionJob job(SyncSessionJob::CONFIGURATION,
+                       TimeTicks::Now(),
+                       session,
+                       false,
+                       params,
+                       FROM_HERE);
+    DoSyncSessionJob(job);
+
+    // If we failed, the job would have been saved as the pending configure
+    // job and a wait interval would have been set.
+    if (!session->Succeeded()) {
+      DCHECK(wait_interval_.get() &&
+             wait_interval_->pending_configure_job.get());
+      return false;
+    }
+  } else {
+    SDVLOG(2) << "No change in routing info, calling ready task directly.";
+    params.ready_task.Run();
+  }
+
+  return true;
+}
+
 SyncScheduler::JobProcessDecision SyncScheduler::DecideWhileInWaitInterval(
     const SyncSessionJob& job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
@@ -376,7 +502,8 @@ void SyncScheduler::InitOrCoalescePendingJob(const SyncSessionJob& job) {
         s->delegate(), s->source(), s->routing_info(), s->workers()));
 
     SyncSessionJob new_job(SyncSessionJob::NUDGE, job.scheduled_start,
-        make_linked_ptr(session.release()), false, job.from_here);
+        make_linked_ptr(session.release()), false,
+        ConfigurationParams(), job.from_here);
     pending_nudge_.reset(new SyncSessionJob(new_job));
 
     return;
@@ -421,11 +548,17 @@ void SyncScheduler::SaveJob(const SyncSessionJob& job) {
     DCHECK(wait_interval_.get());
     DCHECK(mode_ == CONFIGURATION_MODE);
 
+    // Config params should always get set.
+    DCHECK(!job.config_params.ready_task.is_null());
     SyncSession* old = job.session.get();
     SyncSession* s(new SyncSession(session_context_, this, old->source(),
                                    old->routing_info(), old->workers()));
-    SyncSessionJob new_job(job.purpose, TimeTicks::Now(),
-                           make_linked_ptr(s), false, job.from_here);
+    SyncSessionJob new_job(job.purpose,
+                           TimeTicks::Now(),
+                           make_linked_ptr(s),
+                           false,
+                           job.config_params,
+                           job.from_here);
     wait_interval_->pending_configure_job.reset(new SyncSessionJob(new_job));
   } // drop the rest.
   // TODO(sync): Is it okay to drop the rest?  It's weird that
@@ -442,15 +575,6 @@ struct ModelSafeWorkerGroupIs {
   ModelSafeGroup group;
 };
 
-void SyncScheduler::CleanupDisabledTypes() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SyncSessionJob job(SyncSessionJob::CLEANUP_DISABLED_TYPES, TimeTicks::Now(),
-                     make_linked_ptr(CreateSyncSession(SyncSourceInfo())),
-                     false,
-                     FROM_HERE);
-  DoSyncSessionJob(job);
-}
-
 void SyncScheduler::ScheduleNudgeAsync(
     const TimeDelta& delay,
     NudgeSource source, ModelTypeSet types,
@@ -508,7 +632,7 @@ void SyncScheduler::ScheduleNudgeImpl(
   SyncSession* session(CreateSyncSession(info));
   SyncSessionJob job(SyncSessionJob::NUDGE, TimeTicks::Now() + delay,
                      make_linked_ptr(session), is_canary_job,
-                     nudge_location);
+                     ConfigurationParams(), nudge_location);
 
   session = NULL;
   if (!ShouldRunJob(job))
@@ -539,76 +663,6 @@ void SyncScheduler::ScheduleNudgeImpl(
   ScheduleSyncSessionJob(job);
 }
 
-// Helper to extract the routing info and workers corresponding to types in
-// |types| from |current_routes| and |current_workers|.
-void GetModelSafeParamsForTypes(ModelTypeSet types,
-    const ModelSafeRoutingInfo& current_routes,
-    const std::vector<ModelSafeWorker*>& current_workers,
-    ModelSafeRoutingInfo* result_routes,
-    std::vector<ModelSafeWorker*>* result_workers) {
-  bool passive_group_added = false;
-
-  typedef std::vector<ModelSafeWorker*>::const_iterator iter;
-  for (ModelTypeSet::Iterator it = types.First();
-       it.Good(); it.Inc()) {
-    const syncer::ModelType t = it.Get();
-    ModelSafeRoutingInfo::const_iterator route = current_routes.find(t);
-    DCHECK(route != current_routes.end());
-    ModelSafeGroup group = route->second;
-
-    (*result_routes)[t] = group;
-    iter w_tmp_it = std::find_if(current_workers.begin(), current_workers.end(),
-                                 ModelSafeWorkerGroupIs(group));
-    if (w_tmp_it != current_workers.end()) {
-      iter result_workers_it = std::find_if(
-          result_workers->begin(), result_workers->end(),
-          ModelSafeWorkerGroupIs(group));
-      if (result_workers_it == result_workers->end())
-        result_workers->push_back(*w_tmp_it);
-
-      if (group == GROUP_PASSIVE)
-        passive_group_added = true;
-    } else {
-        NOTREACHED();
-    }
-  }
-
-  // Always add group passive.
-  if (passive_group_added == false) {
-    iter it = std::find_if(current_workers.begin(), current_workers.end(),
-                           ModelSafeWorkerGroupIs(GROUP_PASSIVE));
-    if (it != current_workers.end())
-      result_workers->push_back(*it);
-    else
-      NOTREACHED();
-  }
-}
-
-void SyncScheduler::ScheduleConfiguration(
-    ModelTypeSet types,
-    GetUpdatesCallerInfo::GetUpdatesSource source) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  DCHECK(IsConfigRelatedUpdateSourceValue(source));
-  SDVLOG(2) << "Scheduling a config";
-
-  ModelSafeRoutingInfo routes;
-  std::vector<ModelSafeWorker*> workers;
-  GetModelSafeParamsForTypes(types,
-                             session_context_->routing_info(),
-                             session_context_->workers(),
-                             &routes, &workers);
-
-  SyncSession* session = new SyncSession(session_context_, this,
-      SyncSourceInfo(source,
-                     ModelSafeRoutingInfoToPayloadMap(routes, std::string())),
-      routes, workers);
-  SyncSessionJob job(SyncSessionJob::CONFIGURATION, TimeTicks::Now(),
-                     make_linked_ptr(session),
-                     false,
-                     FROM_HERE);
-  DoSyncSessionJob(job);
-}
-
 const char* SyncScheduler::GetModeString(SyncScheduler::Mode mode) {
   switch (mode) {
     ENUM_CASE(CONFIGURATION_MODE);
@@ -807,6 +861,11 @@ void SyncScheduler::FinishSyncSessionJob(const SyncSessionJob& job) {
     // here; see DCHECKs in SaveJob().  (See http://crbug.com/90868.)
     SaveJob(job);
     return;  // Nothing to do.
+  } else if (job.session->Succeeded() &&
+             !job.config_params.ready_task.is_null()) {
+    // If this was a configuration job with a ready task, invoke it now that
+    // we finished successfully.
+    job.config_params.ready_task.Run();
   }
 
   SDVLOG(2) << "Updating the next polling time after SyncMain";
@@ -909,11 +968,15 @@ void SyncScheduler::HandleContinuationError(
   wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
                                         length));
   if (old_job.purpose == SyncSessionJob::CONFIGURATION) {
+    SDVLOG(2) << "Configuration did not succeed, scheduling retry.";
+    // Config params should always get set.
+    DCHECK(!old_job.config_params.ready_task.is_null());
     SyncSession* old = old_job.session.get();
     SyncSession* s(new SyncSession(session_context_, this,
         old->source(), old->routing_info(), old->workers()));
     SyncSessionJob job(old_job.purpose, TimeTicks::Now() + length,
-                        make_linked_ptr(s), false, FROM_HERE);
+                       make_linked_ptr(s), false, old_job.config_params,
+                       FROM_HERE);
     wait_interval_->pending_configure_job.reset(new SyncSessionJob(job));
   } else {
     // We are not in configuration mode. So wait_interval's pending job
@@ -1035,6 +1098,7 @@ void SyncScheduler::PollTimerCallback() {
   SyncSessionJob job(SyncSessionJob::POLL, TimeTicks::Now(),
                      make_linked_ptr(s),
                      false,
+                     ConfigurationParams(),
                      FROM_HERE);
 
   ScheduleSyncSessionJob(job);
diff --git a/sync/engine/sync_scheduler.h b/sync/engine/sync_scheduler.h
index e9892d2..0ca2f5d 100644
--- a/sync/engine/sync_scheduler.h
+++ b/sync/engine/sync_scheduler.h
@@ -36,6 +36,32 @@ namespace syncer {
 
 struct ServerConnectionEvent;
 
+struct ConfigurationParams {
+  enum KeystoreKeyStatus {
+    KEYSTORE_KEY_UNNECESSARY,
+    KEYSTORE_KEY_NEEDED
+  };
+  ConfigurationParams();
+  ConfigurationParams(
+      const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource& source,
+      const syncer::ModelTypeSet& types_to_download,
+      const syncer::ModelSafeRoutingInfo& routing_info,
+      KeystoreKeyStatus keystore_key_status,
+      const base::Closure& ready_task);
+  ~ConfigurationParams();
+
+  // Source for the configuration.
+  sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source;
+  // The types that should be downloaded.
+  syncer::ModelTypeSet types_to_download;
+  // The new routing info (superset of types to be downloaded).
+  ModelSafeRoutingInfo routing_info;
+  // Whether we need to perform a GetKey command.
+  KeystoreKeyStatus keystore_key_status;
+  // Callback to invoke on configuration completion.
+  base::Closure ready_task;
+};
+
 class SyncScheduler : public sessions::SyncSession::Delegate {
  public:
   enum Mode {
@@ -62,10 +88,15 @@ class SyncScheduler : public sessions::SyncSession::Delegate {
 
   // Start the scheduler with the given mode.  If the scheduler is
   // already started, switch to the given mode, although some
-  // scheduled tasks from the old mode may still run.  If non-NULL,
-  // |callback| will be invoked when the mode has been changed to
-  // |mode|.  Takes ownership of |callback|.
-  void Start(Mode mode, const base::Closure& callback);
+  // scheduled tasks from the old mode may still run.
+  virtual void Start(Mode mode);
+
+  // Schedules the configuration task specified by |params|. Returns true if
+  // the configuration task executed immediately, false if it had to be
+  // scheduled for a later attempt. |params.ready_task| is invoked whenever the
+  // configuration task executes.
+  // Note: must already be in CONFIGURATION mode.
+  virtual bool ScheduleConfiguration(const ConfigurationParams& params);
 
   // Request that any running syncer task stop as soon as possible and
   // cancel all scheduled tasks. This function can be called from any thread,
@@ -78,21 +109,13 @@ class SyncScheduler : public sessions::SyncSession::Delegate {
   // The meat and potatoes. Both of these methods will post a delayed task
   // to attempt the actual nudge (see ScheduleNudgeImpl).
   void ScheduleNudgeAsync(const base::TimeDelta& delay, NudgeSource source,
-                     syncer::ModelTypeSet types,
-                     const tracked_objects::Location& nudge_location);
+                          syncer::ModelTypeSet types,
+                          const tracked_objects::Location& nudge_location);
   void ScheduleNudgeWithPayloadsAsync(
       const base::TimeDelta& delay, NudgeSource source,
       const syncer::ModelTypePayloadMap& types_with_payloads,
       const tracked_objects::Location& nudge_location);
 
-  // Schedule a configuration cycle. May execute immediately or at a later time
-  // (depending on backoff/throttle state).
-  // Note: The source argument of this function must come from the subset of
-  // GetUpdatesCallerInfo values related to configurations.
-  void ScheduleConfiguration(
-      syncer::ModelTypeSet types,
-      sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source);
-
   void CleanupDisabledTypes();
 
   // Change status of notifications in the SyncSessionContext.
@@ -157,6 +180,7 @@ class SyncScheduler : public sessions::SyncSession::Delegate {
     SyncSessionJob();
     SyncSessionJob(SyncSessionJobPurpose purpose, base::TimeTicks start,
         linked_ptr<sessions::SyncSession> session, bool is_canary_job,
+        const ConfigurationParams& config_params,
         const tracked_objects::Location& nudge_location);
     ~SyncSessionJob();
     static const char* GetPurposeString(SyncSessionJobPurpose purpose);
@@ -165,6 +189,7 @@ class SyncScheduler : public sessions::SyncSession::Delegate {
     base::TimeTicks scheduled_start;
     linked_ptr<sessions::SyncSession> session;
     bool is_canary_job;
+    ConfigurationParams config_params;
 
     // This is the location the job came from.  Used for debugging.
     // In case of multiple nudges getting coalesced this stores the
diff --git a/sync/engine/sync_scheduler_unittest.cc b/sync/engine/sync_scheduler_unittest.cc
index 65b6bd3..6c3e3eb 100644
--- a/sync/engine/sync_scheduler_unittest.cc
+++ b/sync/engine/sync_scheduler_unittest.cc
@@ -12,6 +12,7 @@
 #include "sync/engine/syncer.h"
 #include "sync/engine/throttled_data_type_tracker.h"
 #include "sync/sessions/test_util.h"
+#include "sync/test/callback_counter.h"
 #include "sync/test/engine/fake_model_worker.h"
 #include "sync/test/engine/mock_connection_manager.h"
 #include "sync/test/engine/test_directory_setter_upper.h"
@@ -27,6 +28,7 @@ using testing::DoAll;
 using testing::Eq;
 using testing::Invoke;
 using testing::Mock;
+using testing::Not;
 using testing::Return;
 using testing::WithArg;
 
@@ -68,6 +70,14 @@ void PumpLoop() {
   RunLoop();
 }
 
+ModelSafeRoutingInfo TypesToRoutingInfo(const ModelTypeSet& types) {
+  ModelSafeRoutingInfo routes;
+  for (ModelTypeSet::Iterator iter = types.First(); iter.Good(); iter.Inc()) {
+    routes[iter.Get()] = GROUP_PASSIVE;
+  }
+  return routes;
+}
+
 // Convenient to use in tests wishing to analyze SyncShare calls over time.
 static const size_t kMinNumSamples = 5;
 class SyncSchedulerTest : public testing::Test {
@@ -151,7 +161,7 @@ class SyncSchedulerTest : public testing::Test {
   }
 
   void StartSyncScheduler(SyncScheduler::Mode mode) {
-    scheduler()->Start(mode, base::Closure());
+    scheduler()->Start(mode);
   }
 
   // This stops the scheduler synchronously.
@@ -290,14 +300,23 @@ TEST_F(SyncSchedulerTest, Config) {
   SyncShareRecords records;
   const ModelTypeSet model_types(syncer::BOOKMARKS);
 
-  EXPECT_CALL(*syncer(), SyncShare(_,_,_))
+  EXPECT_CALL(*syncer(),
+              SyncShare(_,_,_))
+      .WillOnce(Invoke(sessions::test_util::SimulateSuccess))
       .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
                       WithArg<0>(RecordSyncShare(&records))));
 
   StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
 
-  scheduler()->ScheduleConfiguration(
-      model_types, GetUpdatesCallerInfo::RECONFIGURATION);
+  CallbackCounter counter;
+  ConfigurationParams params(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      model_types,
+      TypesToRoutingInfo(model_types),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter)));
+  ASSERT_TRUE(scheduler()->ScheduleConfiguration(params));
+  ASSERT_EQ(1, counter.times_called());
 
   ASSERT_EQ(1U, records.snapshots.size());
   EXPECT_TRUE(CompareModelTypeSetToModelTypePayloadMap(model_types,
@@ -314,7 +333,9 @@ TEST_F(SyncSchedulerTest, ConfigWithBackingOff) {
   SyncShareRecords records;
   const ModelTypeSet model_types(syncer::BOOKMARKS);
 
-  EXPECT_CALL(*syncer(), SyncShare(_,_,_))
+  EXPECT_CALL(*syncer(),
+              SyncShare(_,_,_))
+      .WillOnce(Invoke(sessions::test_util::SimulateSuccess))
       .WillOnce(DoAll(Invoke(sessions::test_util::SimulateCommitFailed),
                       WithArg<0>(RecordSyncShare(&records))))
       .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
@@ -323,61 +344,27 @@ TEST_F(SyncSchedulerTest, ConfigWithBackingOff) {
   StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
 
   ASSERT_EQ(0U, records.snapshots.size());
-  scheduler()->ScheduleConfiguration(
-      model_types, GetUpdatesCallerInfo::RECONFIGURATION);
+  CallbackCounter counter;
+  ConfigurationParams params(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      model_types,
+      TypesToRoutingInfo(model_types),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter)));
+  ASSERT_FALSE(scheduler()->ScheduleConfiguration(params));
+  ASSERT_EQ(0, counter.times_called());
 
   ASSERT_EQ(1U, records.snapshots.size());
   RunLoop();
 
   ASSERT_EQ(2U, records.snapshots.size());
+  ASSERT_EQ(1, counter.times_called());
   EXPECT_TRUE(CompareModelTypeSetToModelTypePayloadMap(model_types,
       records.snapshots[1].source().types));
   EXPECT_EQ(GetUpdatesCallerInfo::RECONFIGURATION,
             records.snapshots[1].source().updates_source);
 }
 
-// Issue 2 config commands. Second one right after the first has failed
-// and make sure LATEST is executed.
-TEST_F(SyncSchedulerTest, MultipleConfigWithBackingOff) {
-  const ModelTypeSet
-      model_types1(syncer::BOOKMARKS),
-      model_types2(syncer::AUTOFILL);
-  UseMockDelayProvider();
-  EXPECT_CALL(*delay(), GetDelay(_))
-      .WillRepeatedly(Return(TimeDelta::FromMilliseconds(30)));
-  SyncShareRecords records;
-
-  EXPECT_CALL(*syncer(), SyncShare(_,_,_))
-      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateCommitFailed),
-                      WithArg<0>(RecordSyncShare(&records))))
-      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateCommitFailed),
-                      WithArg<0>(RecordSyncShare(&records))))
-      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
-                      WithArg<0>(RecordSyncShare(&records))));
-
-  StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
-
-  ASSERT_EQ(0U, records.snapshots.size());
-  scheduler()->ScheduleConfiguration(
-      model_types1, GetUpdatesCallerInfo::RECONFIGURATION);
-
-  ASSERT_EQ(1U, records.snapshots.size());
-  scheduler()->ScheduleConfiguration(
-      model_types2, GetUpdatesCallerInfo::RECONFIGURATION);
-
-  // A canary job gets posted when we go into exponential backoff.
-  RunLoop();
-
-  ASSERT_EQ(2U, records.snapshots.size());
-  RunLoop();
-
-  ASSERT_EQ(3U, records.snapshots.size());
-  EXPECT_TRUE(CompareModelTypeSetToModelTypePayloadMap(model_types2,
-      records.snapshots[2].source().types));
-  EXPECT_EQ(GetUpdatesCallerInfo::RECONFIGURATION,
-            records.snapshots[2].source().updates_source);
-}
-
 // Issue a nudge when the config has failed. Make sure both the config and
 // nudge are executed.
 TEST_F(SyncSchedulerTest, NudgeWithConfigWithBackingOff) {
@@ -400,19 +387,28 @@ TEST_F(SyncSchedulerTest, NudgeWithConfigWithBackingOff) {
   StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
 
   ASSERT_EQ(0U, records.snapshots.size());
-  scheduler()->ScheduleConfiguration(
-      model_types, GetUpdatesCallerInfo::RECONFIGURATION);
-
+  CallbackCounter counter;
+  ConfigurationParams params(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      model_types,
+      TypesToRoutingInfo(model_types),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter)));
+  ASSERT_FALSE(scheduler()->ScheduleConfiguration(params));
+  ASSERT_EQ(0, counter.times_called());
   ASSERT_EQ(1U, records.snapshots.size());
+
   scheduler()->ScheduleNudgeAsync(
       zero(), NUDGE_SOURCE_LOCAL, model_types, FROM_HERE);
   RunLoop();
-
   ASSERT_EQ(2U, records.snapshots.size());
+  ASSERT_EQ(0, counter.times_called());
+
   RunLoop();
+  ASSERT_EQ(3U, records.snapshots.size());
+  ASSERT_EQ(1, counter.times_called());
 
   // Now change the mode so nudge can execute.
-  ASSERT_EQ(3U, records.snapshots.size());
   StartSyncScheduler(SyncScheduler::NORMAL_MODE);
 
   ASSERT_EQ(4U, records.snapshots.size());
@@ -717,13 +713,19 @@ TEST_F(SyncSchedulerTest, HasMoreToSyncThenFails) {
   EXPECT_TRUE(RunAndGetBackoff());
 }
 
-// Test that no syncing occurs when throttled.
+// Test that no syncing occurs when throttled (although CleanupDisabledTypes
+// is allowed).
 TEST_F(SyncSchedulerTest, ThrottlingDoesThrottle) {
   const ModelTypeSet types(syncer::BOOKMARKS);
   TimeDelta poll(TimeDelta::FromMilliseconds(5));
   TimeDelta throttle(TimeDelta::FromMinutes(10));
   scheduler()->OnReceivedLongPollIntervalUpdate(poll);
-  EXPECT_CALL(*syncer(), SyncShare(_,_,_))
+
+  EXPECT_CALL(*syncer(),
+              SyncShare(_, CLEANUP_DISABLED_TYPES, CLEANUP_DISABLED_TYPES))
+      .WillOnce(Invoke(sessions::test_util::SimulateSuccess));
+  EXPECT_CALL(*syncer(), SyncShare(_,Not(CLEANUP_DISABLED_TYPES),
+                                     Not(CLEANUP_DISABLED_TYPES)))
       .WillOnce(WithArg<0>(sessions::test_util::SimulateThrottled(throttle)))
       .WillRepeatedly(AddFailureAndQuitLoopNow());
 
@@ -735,8 +737,15 @@ TEST_F(SyncSchedulerTest, ThrottlingDoesThrottle) {
 
   StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
 
-  scheduler()->ScheduleConfiguration(
-      types, GetUpdatesCallerInfo::RECONFIGURATION);
+  CallbackCounter counter;
+  ConfigurationParams params(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      types,
+      TypesToRoutingInfo(types),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter)));
+  ASSERT_FALSE(scheduler()->ScheduleConfiguration(params));
+  ASSERT_EQ(0, counter.times_called());
 }
 
 TEST_F(SyncSchedulerTest, ThrottlingExpires) {
@@ -769,6 +778,7 @@ TEST_F(SyncSchedulerTest, ConfigurationMode) {
   SyncShareRecords records;
   scheduler()->OnReceivedLongPollIntervalUpdate(poll);
   EXPECT_CALL(*syncer(), SyncShare(_,_,_))
+      .WillOnce(Invoke(sessions::test_util::SimulateSuccess))
       .WillOnce(DoAll(Invoke(sessions::test_util::SimulateSuccess),
                       WithArg<0>(RecordSyncShare(&records))));
 
@@ -782,8 +792,15 @@ TEST_F(SyncSchedulerTest, ConfigurationMode) {
 
   const ModelTypeSet config_types(syncer::BOOKMARKS);
 
-  scheduler()->ScheduleConfiguration(
-      config_types, GetUpdatesCallerInfo::RECONFIGURATION);
+  CallbackCounter counter;
+  ConfigurationParams params(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      config_types,
+      TypesToRoutingInfo(config_types),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter)));
+  ASSERT_TRUE(scheduler()->ScheduleConfiguration(params));
+  ASSERT_EQ(1, counter.times_called());
 
   ASSERT_EQ(1U, records.snapshots.size());
   EXPECT_TRUE(CompareModelTypeSetToModelTypePayloadMap(config_types,
@@ -852,7 +869,7 @@ TEST_F(SyncSchedulerTest, BackoffDropsJobs) {
   scheduler()->OnReceivedLongPollIntervalUpdate(poll);
   UseMockDelayProvider();
 
-  EXPECT_CALL(*syncer(), SyncShare(_,_,_)).Times(1)
+  EXPECT_CALL(*syncer(), SyncShare(_,_,_))
       .WillRepeatedly(DoAll(Invoke(sessions::test_util::SimulateCommitFailed),
                             RecordSyncShareMultiple(&r, 1U)));
   EXPECT_CALL(*delay(), GetDelay(_)).
@@ -894,9 +911,15 @@ TEST_F(SyncSchedulerTest, BackoffDropsJobs) {
 
   StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
 
-  scheduler()->CleanupDisabledTypes();
-  scheduler()->ScheduleConfiguration(
-      types, GetUpdatesCallerInfo::RECONFIGURATION);
+  CallbackCounter counter;
+  ConfigurationParams params(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      types,
+      TypesToRoutingInfo(types),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter)));
+  ASSERT_FALSE(scheduler()->ScheduleConfiguration(params));
+  ASSERT_EQ(0, counter.times_called());
 
   StartSyncScheduler(SyncScheduler::NORMAL_MODE);
 
@@ -1055,27 +1078,49 @@ TEST_F(SyncSchedulerTest, SyncerSteps) {
   StopSyncScheduler();
   Mock::VerifyAndClearExpectations(syncer());
 
+  // Configuration (always includes a cleanup disabled types).
+  EXPECT_CALL(*syncer(),
+              SyncShare(_, CLEANUP_DISABLED_TYPES, CLEANUP_DISABLED_TYPES))
+      .WillOnce(Invoke(sessions::test_util::SimulateSuccess));
   EXPECT_CALL(*syncer(), SyncShare(_, DOWNLOAD_UPDATES, APPLY_UPDATES))
       .WillOnce(Invoke(sessions::test_util::SimulateSuccess));
   StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
 
-  scheduler()->ScheduleConfiguration(
-      ModelTypeSet(), GetUpdatesCallerInfo::RECONFIGURATION);
-
+  syncer::ModelTypeSet model_types(syncer::BOOKMARKS);
+  CallbackCounter counter;
+  ConfigurationParams params(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      model_types,
+      TypesToRoutingInfo(model_types),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter)));
+  ASSERT_TRUE(scheduler()->ScheduleConfiguration(params));
+  ASSERT_EQ(1, counter.times_called());
+  // Runs directly so no need to pump the loop.
   StopSyncScheduler();
   Mock::VerifyAndClearExpectations(syncer());
 
-  // Cleanup disabled types.
+  // Cleanup disabled types. Because no types are being configured, we just
+  // perform the cleanup.
   EXPECT_CALL(*syncer(),
-              SyncShare(_, CLEANUP_DISABLED_TYPES, CLEANUP_DISABLED_TYPES))
-      .WillOnce(Invoke(sessions::test_util::SimulateSuccess));
-  StartSyncScheduler(SyncScheduler::NORMAL_MODE);
-
-  scheduler()->CleanupDisabledTypes();
+              SyncShare(_, CLEANUP_DISABLED_TYPES, CLEANUP_DISABLED_TYPES)).
+      WillOnce(Invoke(sessions::test_util::SimulateSuccess));
+  StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
 
+  CallbackCounter counter2;
+  ConfigurationParams params2(
+      GetUpdatesCallerInfo::RECONFIGURATION,
+      ModelTypeSet(),
+      ModelSafeRoutingInfo(),
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY,
+      base::Bind(&CallbackCounter::Callback, base::Unretained(&counter2)));
+  ASSERT_TRUE(scheduler()->ScheduleConfiguration(params2));
+  ASSERT_EQ(1, counter2.times_called());
   StopSyncScheduler();
   Mock::VerifyAndClearExpectations(syncer());
 
+  StartSyncScheduler(SyncScheduler::NORMAL_MODE);
+
   // Poll.
   EXPECT_CALL(*syncer(), SyncShare(_, SYNCER_BEGIN, SYNCER_END))
       .Times(AtLeast(1))
diff --git a/sync/engine/sync_scheduler_whitebox_unittest.cc b/sync/engine/sync_scheduler_whitebox_unittest.cc
index 20e1277..760083f 100644
--- a/sync/engine/sync_scheduler_whitebox_unittest.cc
+++ b/sync/engine/sync_scheduler_whitebox_unittest.cc
@@ -107,6 +107,7 @@ class SyncSchedulerWhiteboxTest : public testing::Test {
     SyncScheduler::SyncSessionJob job(purpose, TimeTicks::Now(),
          make_linked_ptr(s),
          false,
+         ConfigurationParams(),
          FROM_HERE);
     return DecideOnJob(job);
   }
@@ -160,6 +161,7 @@ TEST_F(SyncSchedulerWhiteboxTest, SaveNudgeWhileTypeThrottled) {
                                     TimeTicks::Now(),
                                     make_linked_ptr(s),
                                     false,
+                                    ConfigurationParams(),
                                     FROM_HERE);
 
   SyncScheduler::JobProcessDecision decision = DecideOnJob(job);
diff --git a/sync/internal_api/public/sync_manager.h b/sync/internal_api/public/sync_manager.h
index 75e55aa..d31888c6 100644
--- a/sync/internal_api/public/sync_manager.h
+++ b/sync/internal_api/public/sync_manager.h
@@ -31,6 +31,7 @@ struct Experiments;
 class ExtensionsActivityMonitor;
 class JsBackend;
 class JsEventHandler;
+class SyncScheduler;
 
 namespace sessions {
 class SyncSessionSnapshot;
@@ -401,6 +402,16 @@ class SyncManager {
   // Returns the set of types for which we have stored some sync data.
   syncer::ModelTypeSet InitialSyncEndedTypes();
 
+  // Returns those types within |types| that have an empty progress marker
+  // token.
+  syncer::ModelTypeSet GetTypesWithEmptyProgressMarkerToken(
+      syncer::ModelTypeSet types);
+
+  // Purge from the directory those types with non-empty progress markers
+  // but without initial synced ended set.
+  // Returns false if an error occurred, true otherwise.
+  bool PurgePartiallySyncedTypes();
+
   // Update tokens that we're using in Sync. Email must stay the same.
   void UpdateCredentials(const SyncCredentials& credentials);
 
@@ -428,20 +439,20 @@ class SyncManager {
   // error to call this when we don't have pending keys.
   void SetDecryptionPassphrase(const std::string& passphrase);
 
-  // Puts the SyncScheduler into a mode where no normal nudge or poll traffic
-  // will occur, but calls to RequestConfig will be supported.  If |callback|
-  // is provided, it will be invoked (from the internal SyncScheduler) when
-  // the thread has changed to configuration mode.
-  void StartConfigurationMode(const base::Closure& callback);
-
-  // Switches the mode of operation to CONFIGURATION_MODE and
-  // schedules a config task to fetch updates for |types|.
-  void RequestConfig(const syncer::ModelSafeRoutingInfo& routing_info,
-                     const syncer::ModelTypeSet& types,
-                     syncer::ConfigureReason reason);
-
-  void RequestCleanupDisabledTypes(
-      const syncer::ModelSafeRoutingInfo& routing_info);
+  // Switches the mode of operation to CONFIGURATION_MODE and performs
+  // any configuration tasks needed as determined by the params. Once complete,
+  // syncer will remain in CONFIGURATION_MODE until StartSyncingNormally is
+  // called.
+  // |ready_task| is invoked when the configuration completes.
+  // |retry_task| is invoked if the configuration job could not immediately
+  //              execute. |ready_task| will still be called when it eventually
+  //              does finish.
+  void ConfigureSyncer(
+      ConfigureReason reason,
+      const syncer::ModelTypeSet& types_to_config,
+      const syncer::ModelSafeRoutingInfo& new_routing_info,
+      const base::Closure& ready_task,
+      const base::Closure& retry_task);
 
   // Adds a listener to be notified of sync events.
   // NOTE: It is OK (in fact, it's probably a good idea) to call this before
@@ -538,11 +549,17 @@ class SyncManager {
   static const FilePath::CharType kSyncDatabaseFilename[];
 
  private:
+  friend class SyncManagerTest;
   FRIEND_TEST_ALL_PREFIXES(SyncManagerTest, NudgeDelayTest);
 
   // For unit tests.
   base::TimeDelta GetNudgeDelayTimeDelta(const syncer::ModelType& model_type);
 
+  // Set the internal scheduler for testing purposes.
+  // TODO(sync): Use dependency injection instead. crbug.com/133061
+  void SetSyncSchedulerForTest(
+      scoped_ptr<syncer::SyncScheduler> scheduler);
+
   base::ThreadChecker thread_checker_;
 
   // An opaque pointer to the nested private class.
@@ -553,10 +570,6 @@ class SyncManager {
 
 bool InitialSyncEndedForTypes(syncer::ModelTypeSet types, UserShare* share);
 
-syncer::ModelTypeSet GetTypesWithEmptyProgressMarkerToken(
-    syncer::ModelTypeSet types,
-    syncer::UserShare* share);
-
 const char* ConnectionStatusToString(ConnectionStatus status);
 
 // Returns the string representation of a PassphraseRequiredReason value.
diff --git a/sync/internal_api/sync_manager.cc b/sync/internal_api/sync_manager.cc
index c8a293e..e8ba9dd 100644
--- a/sync/internal_api/sync_manager.cc
+++ b/sync/internal_api/sync_manager.cc
@@ -197,6 +197,11 @@ class SyncManager::SyncInternal
   // went wrong.
   bool SignIn(const SyncCredentials& credentials);
 
+  // Purge from the directory those types with non-empty progress markers
+  // but without initial synced ended set.
+  // Returns false if an error occurred, true otherwise.
+  bool PurgePartiallySyncedTypes();
+
   // Update tokens that we're using in Sync. Email must stay the same.
   void UpdateCredentials(const SyncCredentials& credentials);
 
@@ -357,6 +362,21 @@ class SyncManager::SyncInternal
   // Called only by our NetworkChangeNotifier.
   virtual void OnIPAddressChanged() OVERRIDE;
 
+  ModelTypeSet GetTypesWithEmptyProgressMarkerToken(ModelTypeSet types) {
+    DCHECK(initialized_);
+    syncer::ModelTypeSet result;
+    for (syncer::ModelTypeSet::Iterator i = types.First();
+         i.Good(); i.Inc()) {
+      sync_pb::DataTypeProgressMarker marker;
+      directory()->GetDownloadProgress(i.Get(), &marker);
+
+      if (marker.token().empty())
+        result.Put(i.Get());
+
+    }
+    return result;
+  }
+
   syncer::ModelTypeSet InitialSyncEndedTypes() {
     DCHECK(initialized_);
     return directory()->initial_sync_ended_types();
@@ -376,6 +396,8 @@ class SyncManager::SyncInternal
       const std::string& name, const JsArgList& args,
       const WeakHandle<JsReplyHandler>& reply_handler) OVERRIDE;
 
+  void SetSyncSchedulerForTest(scoped_ptr<SyncScheduler> scheduler);
+
  private:
   struct NotificationInfo {
     int total_count;
@@ -749,6 +771,15 @@ syncer::ModelTypeSet SyncManager::InitialSyncEndedTypes() {
   return data_->InitialSyncEndedTypes();
 }
 
+syncer::ModelTypeSet SyncManager::GetTypesWithEmptyProgressMarkerToken(
+    syncer::ModelTypeSet types) {
+  return data_->GetTypesWithEmptyProgressMarkerToken(types);
+}
+
+bool SyncManager::PurgePartiallySyncedTypes() {
+  return data_->PurgePartiallySyncedTypes();
+}
+
 void SyncManager::StartSyncingNormally(
     const syncer::ModelSafeRoutingInfo& routing_info) {
   DCHECK(thread_checker_.CalledOnValidThread());
@@ -793,40 +824,39 @@ bool SyncManager::IsUsingExplicitPassphrase() {
   return data_ && data_->IsUsingExplicitPassphrase();
 }
 
-void SyncManager::RequestCleanupDisabledTypes(
-    const syncer::ModelSafeRoutingInfo& routing_info) {
+void SyncManager::ConfigureSyncer(
+    ConfigureReason reason,
+    const syncer::ModelTypeSet& types_to_config,
+    const syncer::ModelSafeRoutingInfo& new_routing_info,
+    const base::Closure& ready_task,
+    const base::Closure& retry_task) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  if (data_->scheduler()) {
-    data_->session_context()->set_routing_info(routing_info);
-    data_->scheduler()->CleanupDisabledTypes();
-  }
-}
+  DCHECK(!ready_task.is_null());
+  DCHECK(!retry_task.is_null());
 
-void SyncManager::RequestConfig(
-    const syncer::ModelSafeRoutingInfo& routing_info,
-    const ModelTypeSet& types, ConfigureReason reason) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  if (!data_->scheduler()) {
-    LOG(INFO)
-        << "SyncManager::RequestConfig: bailing out because scheduler is "
-        << "null";
-    return;
-  }
-  StartConfigurationMode(base::Closure());
-  data_->session_context()->set_routing_info(routing_info);
-  data_->scheduler()->ScheduleConfiguration(types, GetSourceFromReason(reason));
-}
+  // TODO(zea): set this based on whether cryptographer has keystore
+  // encryption key or not (requires opening a transaction). crbug.com/129665.
+  ConfigurationParams::KeystoreKeyStatus keystore_key_status =
+      ConfigurationParams::KEYSTORE_KEY_UNNECESSARY;
+
+  ConfigurationParams params(GetSourceFromReason(reason),
+                             types_to_config,
+                             new_routing_info,
+                             keystore_key_status,
+                             ready_task);
 
-void SyncManager::StartConfigurationMode(const base::Closure& callback) {
-  DCHECK(thread_checker_.CalledOnValidThread());
   if (!data_->scheduler()) {
     LOG(INFO)
-        << "SyncManager::StartConfigurationMode: could not start "
-        << "configuration mode because because scheduler is null";
+        << "SyncManager::ConfigureSyncer: could not configure because "
+        << "scheduler is null";
+    params.ready_task.Run();
     return;
   }
-  data_->scheduler()->Start(
-      syncer::SyncScheduler::CONFIGURATION_MODE, callback);
+
+  data_->scheduler()->Start(syncer::SyncScheduler::CONFIGURATION_MODE);
+  if (!data_->scheduler()->ScheduleConfiguration(params))
+    retry_task.Run();
+
 }
 
 bool SyncManager::SyncInternal::Init(
@@ -923,16 +953,29 @@ bool SyncManager::SyncInternal::Init(
     scheduler_.reset(new SyncScheduler(name_, session_context(), new Syncer()));
   }
 
-  bool signed_in = SignIn(credentials);
+  bool success = SignIn(credentials);
 
-  if (signed_in) {
+  if (success) {
     if (scheduler()) {
-      scheduler()->Start(
-          syncer::SyncScheduler::CONFIGURATION_MODE, base::Closure());
+      scheduler()->Start(syncer::SyncScheduler::CONFIGURATION_MODE);
     }
 
     initialized_ = true;
 
+    // Unapplied datatypes (those that do not have initial sync ended set) get
+    // re-downloaded during any configuration. But, it's possible for a datatype
+    // to have a progress marker but not have initial sync ended yet, making
+    // it a candidate for migration. This is a problem, as the DataTypeManager
+    // does not support a migration while it's already in the middle of a
+    // configuration. As a result, any partially synced datatype can stall the
+    // DTM, waiting for the configuration to complete, which it never will due
+    // to the migration error. In addition, a partially synced nigori will
+    // trigger the migration logic before the backend is initialized, resulting
+    // in crashes. We therefore detect and purge any partially synced types as
+    // part of initialization.
+    if (!PurgePartiallySyncedTypes())
+      success = false;
+
     // Cryptographer should only be accessed while holding a
     // transaction.  Grabbing the user share for the transaction
     // checks the initialization state, so this must come after
@@ -950,14 +993,14 @@ bool SyncManager::SyncInternal::Init(
   FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
                     OnInitializationComplete(
                         MakeWeakHandle(weak_ptr_factory_.GetWeakPtr()),
-                        signed_in));
+                        success));
 
-  if (!signed_in && testing_mode_ == NON_TEST)
+  if (!success && testing_mode_ == NON_TEST)
     return false;
 
   sync_notifier_->AddObserver(this);
 
-  return signed_in;
+  return success;
 }
 
 void SyncManager::SyncInternal::UpdateCryptographerAndNigori(
@@ -1102,9 +1145,13 @@ void SyncManager::SyncInternal::NotifyCryptographerState(
 void SyncManager::SyncInternal::StartSyncingNormally(
     const syncer::ModelSafeRoutingInfo& routing_info) {
   // Start the sync scheduler.
-  if (scheduler()) { // NULL during certain unittests.
+  if (scheduler()) {  // NULL during certain unittests.
+    // TODO(sync): We always want the newest set of routes when we switch back
+    // to normal mode. Figure out how to enforce set_routing_info is always
+    // appropriately set and that it's only modified when switching to normal
+    // mode.
     session_context()->set_routing_info(routing_info);
-    scheduler()->Start(SyncScheduler::NORMAL_MODE, base::Closure());
+    scheduler()->Start(SyncScheduler::NORMAL_MODE);
   }
 }
 
@@ -1158,6 +1205,20 @@ bool SyncManager::SyncInternal::SignIn(const SyncCredentials& credentials) {
   return true;
 }
 
+bool SyncManager::SyncInternal::PurgePartiallySyncedTypes() {
+  syncer::ModelTypeSet partially_synced_types =
+      syncer::ModelTypeSet::All();
+  partially_synced_types.RemoveAll(InitialSyncEndedTypes());
+  partially_synced_types.RemoveAll(GetTypesWithEmptyProgressMarkerToken(
+      syncer::ModelTypeSet::All()));
+
+  UMA_HISTOGRAM_COUNTS("Sync.PartiallySyncedTypes",
+                       partially_synced_types.Size());
+  if (partially_synced_types.Empty())
+    return true;
+  return directory()->PurgeEntriesWithTypeIn(partially_synced_types);
+}
+
 void SyncManager::SyncInternal::UpdateCredentials(
     const SyncCredentials& credentials) {
   DCHECK(thread_checker_.CalledOnValidThread());
@@ -2340,6 +2401,11 @@ void SyncManager::SyncInternal::RemoveObserver(
   observers_.RemoveObserver(observer);
 }
 
+void SyncManager::SyncInternal::SetSyncSchedulerForTest(
+    scoped_ptr<SyncScheduler> sync_scheduler) {
+  scheduler_ = sync_scheduler.Pass();
+}
+
 SyncStatus SyncManager::GetDetailedStatus() const {
   return data_->GetStatus();
 }
@@ -2370,6 +2436,10 @@ TimeDelta SyncManager::GetNudgeDelayTimeDelta(
   return data_->GetNudgeDelayTimeDelta(model_type);
 }
 
+void SyncManager::SetSyncSchedulerForTest(scoped_ptr<SyncScheduler> scheduler) {
+  data_->SetSyncSchedulerForTest(scheduler.Pass());
+}
+
 syncer::ModelTypeSet SyncManager::GetEncryptedDataTypesForTest() const {
   ReadTransaction trans(FROM_HERE, GetUserShare());
   return GetEncryptedTypes(&trans);
@@ -2459,20 +2529,4 @@ bool InitialSyncEndedForTypes(syncer::ModelTypeSet types,
   return true;
 }
 
-syncer::ModelTypeSet GetTypesWithEmptyProgressMarkerToken(
-    syncer::ModelTypeSet types,
-    syncer::UserShare* share) {
-  syncer::ModelTypeSet result;
-  for (syncer::ModelTypeSet::Iterator i = types.First();
-       i.Good(); i.Inc()) {
-    sync_pb::DataTypeProgressMarker marker;
-    share->directory->GetDownloadProgress(i.Get(), &marker);
-
-    if (marker.token().empty())
-      result.Put(i.Get());
-
-  }
-  return result;
-}
-
 }  // namespace syncer
diff --git a/sync/internal_api/syncapi_unittest.cc b/sync/internal_api/syncapi_unittest.cc
index 6fff8c5..b22b585 100644
--- a/sync/internal_api/syncapi_unittest.cc
+++ b/sync/internal_api/syncapi_unittest.cc
@@ -24,6 +24,7 @@
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
 #include "sync/internal_api/public/base/model_type_test_util.h"
+#include "sync/engine/sync_scheduler.h"
 #include "sync/internal_api/public/change_record.h"
 #include "sync/internal_api/public/engine/model_safe_worker.h"
 #include "sync/internal_api/public/engine/polling_constants.h"
@@ -57,6 +58,7 @@
 #include "sync/syncable/nigori_util.h"
 #include "sync/syncable/syncable_id.h"
 #include "sync/syncable/write_transaction.h"
+#include "sync/test/callback_counter.h"
 #include "sync/test/fake_encryptor.h"
 #include "sync/test/fake_extensions_activity_monitor.h"
 #include "sync/util/cryptographer.h"
@@ -70,8 +72,10 @@ using base::ExpectDictStringValue;
 using testing::_;
 using testing::AnyNumber;
 using testing::AtLeast;
+using testing::DoAll;
 using testing::InSequence;
 using testing::Invoke;
+using testing::Return;
 using testing::SaveArg;
 using testing::StrictMock;
 
@@ -906,6 +910,10 @@ class SyncManagerTest : public testing::Test,
     return true;
   }
 
+  void SetScheduler(scoped_ptr<SyncScheduler> scheduler) {
+    sync_manager_.SetSyncSchedulerForTest(scheduler.Pass());
+  }
+
  private:
   // Needed by |sync_manager_|.
   MessageLoop message_loop_;
@@ -2485,4 +2493,120 @@ TEST_F(SyncManagerTest, SetPreviouslyEncryptedSpecifics) {
   }
 }
 
-}  // namespace syncer
+class MockSyncScheduler : public SyncScheduler {
+ public:
+  MockSyncScheduler() : SyncScheduler("name", NULL, NULL) {}
+  virtual ~MockSyncScheduler() {}
+
+  MOCK_METHOD1(Start, void(SyncScheduler::Mode));
+  MOCK_METHOD1(ScheduleConfiguration, bool(const ConfigurationParams&));
+};
+
+// Test that the configuration params are properly created and sent to
+// ScheduleConfigure. No callback should be invoked.
+TEST_F(SyncManagerTest, BasicConfiguration) {
+  ConfigureReason reason = CONFIGURE_REASON_RECONFIGURATION;
+  syncer::ModelTypeSet types_to_download(syncer::BOOKMARKS,
+                                         syncer::PREFERENCES);
+  syncer::ModelSafeRoutingInfo new_routing_info;
+  GetModelSafeRoutingInfo(&new_routing_info);
+
+  scoped_ptr<MockSyncScheduler> scheduler(new MockSyncScheduler());
+  ConfigurationParams params;
+  EXPECT_CALL(*scheduler, Start(SyncScheduler::CONFIGURATION_MODE));
+  EXPECT_CALL(*scheduler, ScheduleConfiguration(_)).
+      WillOnce(DoAll(SaveArg<0>(&params), Return(true)));
+  SetScheduler(scheduler.PassAs<SyncScheduler>());
+
+  CallbackCounter ready_task_counter, retry_task_counter;
+  sync_manager_.ConfigureSyncer(
+      reason,
+      types_to_download,
+      new_routing_info,
+      base::Bind(&CallbackCounter::Callback,
+                 base::Unretained(&ready_task_counter)),
+      base::Bind(&CallbackCounter::Callback,
+                 base::Unretained(&retry_task_counter)));
+  EXPECT_EQ(0, ready_task_counter.times_called());
+  EXPECT_EQ(0, retry_task_counter.times_called());
+  EXPECT_EQ(sync_pb::GetUpdatesCallerInfo::RECONFIGURATION,
+            params.source);
+  EXPECT_TRUE(types_to_download.Equals(params.types_to_download));
+  EXPECT_EQ(new_routing_info, params.routing_info);
+}
+
+// Test that the retry callback is invoked on configuration failure.
+TEST_F(SyncManagerTest, ConfigurationRetry) {
+  ConfigureReason reason = CONFIGURE_REASON_RECONFIGURATION;
+  syncer::ModelTypeSet types_to_download(syncer::BOOKMARKS,
+                                         syncer::PREFERENCES);
+  syncer::ModelSafeRoutingInfo new_routing_info;
+  GetModelSafeRoutingInfo(&new_routing_info);
+
+  scoped_ptr<MockSyncScheduler> scheduler(new MockSyncScheduler());
+  ConfigurationParams params;
+  EXPECT_CALL(*scheduler, Start(SyncScheduler::CONFIGURATION_MODE));
+  EXPECT_CALL(*scheduler, ScheduleConfiguration(_)).
+      WillOnce(DoAll(SaveArg<0>(&params), Return(false)));
+  SetScheduler(scheduler.PassAs<SyncScheduler>());
+
+  CallbackCounter ready_task_counter, retry_task_counter;
+  sync_manager_.ConfigureSyncer(
+      reason,
+      types_to_download,
+      new_routing_info,
+      base::Bind(&CallbackCounter::Callback,
+                 base::Unretained(&ready_task_counter)),
+      base::Bind(&CallbackCounter::Callback,
+                 base::Unretained(&retry_task_counter)));
+  EXPECT_EQ(0, ready_task_counter.times_called());
+  EXPECT_EQ(1, retry_task_counter.times_called());
+  EXPECT_EQ(sync_pb::GetUpdatesCallerInfo::RECONFIGURATION,
+            params.source);
+  EXPECT_TRUE(types_to_download.Equals(params.types_to_download));
+  EXPECT_EQ(new_routing_info, params.routing_info);
+}
+
+// Test that PurgePartiallySyncedTypes purges only those types that don't
+// have empty progress marker and don't have initial sync ended set.
+TEST_F(SyncManagerTest, PurgePartiallySyncedTypes) {
+  UserShare* share = sync_manager_.GetUserShare();
+
+  // Set Nigori and Bookmarks to be partial types.
+  sync_pb::DataTypeProgressMarker nigori_marker;
+  nigori_marker.set_data_type_id(
+      GetSpecificsFieldNumberFromModelType(NIGORI));
+  nigori_marker.set_token("token");
+  sync_pb::DataTypeProgressMarker bookmark_marker;
+  bookmark_marker.set_data_type_id(
+      GetSpecificsFieldNumberFromModelType(BOOKMARKS));
+  bookmark_marker.set_token("token");
+  share->directory->SetDownloadProgress(NIGORI, nigori_marker);
+  share->directory->SetDownloadProgress(BOOKMARKS, bookmark_marker);
+
+  // Set Preferences to be a full type.
+  sync_pb::DataTypeProgressMarker pref_marker;
+  pref_marker.set_data_type_id(
+      GetSpecificsFieldNumberFromModelType(PREFERENCES));
+  pref_marker.set_token("token");
+  share->directory->SetDownloadProgress(PREFERENCES, pref_marker);
+  share->directory->set_initial_sync_ended_for_type(PREFERENCES, true);
+
+  ModelTypeSet partial_types =
+      sync_manager_.GetTypesWithEmptyProgressMarkerToken(ModelTypeSet::All());
+  EXPECT_FALSE(partial_types.Has(NIGORI));
+  EXPECT_FALSE(partial_types.Has(BOOKMARKS));
+  EXPECT_FALSE(partial_types.Has(PREFERENCES));
+
+  EXPECT_TRUE(sync_manager_.PurgePartiallySyncedTypes());
+
+  // Ensure only bookmarks and nigori lost their progress marker. Preferences
+  // should still have it.
+  partial_types =
+      sync_manager_.GetTypesWithEmptyProgressMarkerToken(ModelTypeSet::All());
+  EXPECT_TRUE(partial_types.Has(NIGORI));
+  EXPECT_TRUE(partial_types.Has(BOOKMARKS));
+  EXPECT_FALSE(partial_types.Has(PREFERENCES));
+}
+
+}  // namespace
diff --git a/sync/sync.gyp b/sync/sync.gyp
index 04ba6b1..b1554f5 100644
--- a/sync/sync.gyp
+++ b/sync/sync.gyp
@@ -391,6 +391,7 @@
         'sessions/test_util.h',
         'syncable/syncable_mock.cc',
         'syncable/syncable_mock.h',
+        'test/callback_counter.h',
         'test/engine/fake_model_worker.cc',
         'test/engine/fake_model_worker.h',
         'test/engine/mock_connection_manager.cc',
diff --git a/sync/syncable/directory.cc b/sync/syncable/directory.cc
index da5449d..a63538c 100644
--- a/sync/syncable/directory.cc
+++ b/sync/syncable/directory.cc
@@ -577,9 +577,9 @@ bool Directory::VacuumAfterSaveChanges(const SaveChangesSnapshot& snapshot) {
   return true;
 }
 
-void Directory::PurgeEntriesWithTypeIn(ModelTypeSet types) {
+bool Directory::PurgeEntriesWithTypeIn(ModelTypeSet types) {
   if (types.Empty())
-    return;
+    return true;
 
   {
     WriteTransaction trans(FROM_HERE, PURGE_ENTRIES, this);
@@ -597,7 +597,7 @@ void Directory::PurgeEntriesWithTypeIn(ModelTypeSet types) {
         if ((IsRealDataType(local_type) && types.Has(local_type)) ||
             (IsRealDataType(server_type) && types.Has(server_type))) {
           if (!UnlinkEntryFromOrder(*it, &trans, &lock, DATA_TYPE_PURGE))
-            return;
+            return false;
 
           int64 handle = (*it)->ref(META_HANDLE);
           kernel_->metahandles_to_purge->insert(handle);
@@ -630,6 +630,7 @@ void Directory::PurgeEntriesWithTypeIn(ModelTypeSet types) {
       }
     }
   }
+  return true;
 }
 
 void Directory::HandleSaveChangesFailure(const SaveChangesSnapshot& snapshot) {
diff --git a/sync/syncable/directory.h b/sync/syncable/directory.h
index 340b404..2385f03 100644
--- a/sync/syncable/directory.h
+++ b/sync/syncable/directory.h
@@ -421,7 +421,8 @@ class Directory {
   // entries, which means something different in the syncable namespace.
   // WARNING! This can be real slow, as it iterates over all entries.
   // WARNING! Performs synchronous I/O.
-  virtual void PurgeEntriesWithTypeIn(ModelTypeSet types);
+  // Returns: true on success, false if an error was encountered.
+  virtual bool PurgeEntriesWithTypeIn(ModelTypeSet types);
 
  private:
   // A helper that implements the logic of checking tree invariants.
diff --git a/sync/syncable/syncable_mock.h b/sync/syncable/syncable_mock.h
index 24d435f..7b5327f 100644
--- a/sync/syncable/syncable_mock.h
+++ b/sync/syncable/syncable_mock.h
@@ -29,7 +29,7 @@ class MockDirectory : public Directory {
   MOCK_METHOD1(GetEntryByClientTag,
                syncable::EntryKernel*(const std::string&));
 
-  MOCK_METHOD1(PurgeEntriesWithTypeIn, void(syncer::ModelTypeSet));
+  MOCK_METHOD1(PurgeEntriesWithTypeIn, bool(syncer::ModelTypeSet));
 
  private:
   syncer::FakeEncryptor encryptor_;
diff --git a/sync/test/callback_counter.h b/sync/test/callback_counter.h
new file mode 100644
index 0000000..16b2acb
--- /dev/null
+++ b/sync/test/callback_counter.h
@@ -0,0 +1,28 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef SYNC_TEST_CALLBACK_COUNTER_H_
+#define SYNC_TEST_CALLBACK_COUNTER_H_
+
+namespace syncer {
+
+// Helper class to track how many times a callback is triggered.
+class CallbackCounter {
+ public:
+  CallbackCounter() { Reset(); }
+  ~CallbackCounter() {}
+
+  void Reset() { times_called_ = 0; }
+  void Callback() { ++times_called_; }
+  int times_called() const { return times_called_; }
+
+ private:
+  int times_called_;
+
+  DISALLOW_COPY_AND_ASSIGN(CallbackCounter);
+};
+
+}  // namespace syncer
+
+#endif  // SYNC_TEST_CALLBACK_COUNTER_H_