sync: Finish the SyncScheduler refactor

This change removes SyncSessionJob entirely.  The first step towards
this goal was to refactor all functions that depended on SyncSessionJob.
All these functions have either been inlined or modified to take
different parameters instead.  DoSyncSessionJob was split into
two separate functions, one for configure jobs and one for nudge jobs,
which removes the need for SyncSessionJob's "purpose" member.

The SyncScheduler's pending_configure_job_ has been replaced with
a ConfigParams member, since that was the only part of the job still
being referenced.  The pending_nudge_job_ has been replaced with
scheduled_nudge_time_ (which is similar to the old scheduled_start_
member of SyncSessionJob) and a new object called a NudgeTracker.

The NudgeTracker inherits the SyncSessionJob's responsibilities with
respect to coalescing nudge sources.  The plan is to extend this class
to support more sophisticated nudge coalescing in the future.  For now
it tries to emulate the old SyncSessionJob behaviour as closely as
possible.

Some of the refactoring does change behaviour.  In particular, the
decision-making logic has been updated to fix issues 179515 and 155296.

BUG=175024,179515,155296

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

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

13 files changed, 414 insertions, 1038 deletions

diff --git a/sync/engine/sync_scheduler_impl.cc b/sync/engine/sync_scheduler_impl.cc
index b03b3c8..0fd7ac0 100644
--- a/sync/engine/sync_scheduler_impl.cc
+++ b/sync/engine/sync_scheduler_impl.cc
@@ -225,9 +225,9 @@ void SyncSchedulerImpl::Start(Mode mode) {
   DCHECK(syncer_.get());
   Mode old_mode = mode_;
   mode_ = mode;
-  AdjustPolling(NULL);  // Will kick start poll timer if needed.
+  AdjustPolling(UPDATE_INTERVAL);  // Will kick start poll timer if needed.
 
-  if (old_mode != mode_ && mode_ == NORMAL_MODE && pending_nudge_job_) {
+  if (old_mode != mode_ && mode_ == NORMAL_MODE && !nudge_tracker_.IsEmpty()) {
     // We just got back to normal mode.  Let's try to run the work that was
     // queued up while we were configuring.
     DoNudgeSyncSessionJob(NORMAL_PRIORITY);
@@ -274,7 +274,7 @@ bool SyncSchedulerImpl::ScheduleConfiguration(
 
   // Only one configuration is allowed at a time. Verify we're not waiting
   // for a pending configure job.
-  DCHECK(!pending_configure_job_);
+  DCHECK(!pending_configure_params_);
 
   ModelSafeRoutingInfo restricted_routes;
   BuildModelSafeParams(params.types_to_download,
@@ -284,25 +284,17 @@ bool SyncSchedulerImpl::ScheduleConfiguration(
 
   // Only reconfigure if we have types to download.
   if (!params.types_to_download.Empty()) {
-    DCHECK(!restricted_routes.empty());
-    pending_configure_job_.reset(new SyncSessionJob(
-        SyncSessionJob::CONFIGURATION,
-        TimeTicks::Now(),
-        SyncSourceInfo(params.source,
-                       ModelSafeRoutingInfoToInvalidationMap(
-                           restricted_routes,
-                           std::string())),
-        params));
+    pending_configure_params_.reset(new ConfigurationParams(params));
     bool succeeded = DoConfigurationSyncSessionJob(NORMAL_PRIORITY);
 
     // If we failed, the job would have been saved as the pending configure
     // job and a wait interval would have been set.
     if (!succeeded) {
-      DCHECK(pending_configure_job_);
-      return false;
+      DCHECK(pending_configure_params_);
     } else {
-      DCHECK(!pending_configure_job_);
+      DCHECK(!pending_configure_params_);
     }
+    return succeeded;
   } else {
     SDVLOG(2) << "No change in routing info, calling ready task directly.";
     params.ready_task.Run();
@@ -311,120 +303,55 @@ bool SyncSchedulerImpl::ScheduleConfiguration(
   return true;
 }
 
-SyncSchedulerImpl::JobProcessDecision
-SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob& job,
-                                             JobPriority priority) {
+bool SyncSchedulerImpl::CanRunJobNow(JobPriority priority) {
   DCHECK(CalledOnValidThread());
-  DCHECK(wait_interval_.get());
-  DCHECK_NE(job.purpose(), SyncSessionJob::POLL);
-
-  SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
-            << WaitInterval::GetModeString(wait_interval_->mode)
-            << ((priority == CANARY_PRIORITY) ? " (canary)" : "");
-
-  // If we save a job while in a WaitInterval, there is a well-defined moment
-  // in time in the future when it makes sense for that SAVE-worthy job to try
-  // running again -- the end of the WaitInterval.
-  DCHECK(job.purpose() == SyncSessionJob::NUDGE ||
-         job.purpose() == SyncSessionJob::CONFIGURATION);
-
-  // If throttled, there's a clock ticking to unthrottle. We want to get
-  // on the same train.
-  if (wait_interval_->mode == WaitInterval::THROTTLED)
-    return SAVE;
+  if (wait_interval_ && wait_interval_->mode == WaitInterval::THROTTLED) {
+    SDVLOG(1) << "Unable to run a job because we're throttled.";
+    return false;
+  }
 
-  DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
-  if (job.purpose() == SyncSessionJob::NUDGE) {
-    if (mode_ == CONFIGURATION_MODE)
-      return SAVE;
+  if (wait_interval_
+      && wait_interval_->mode == WaitInterval::EXPONENTIAL_BACKOFF
+      && priority != CANARY_PRIORITY) {
+    SDVLOG(1) << "Unable to run a job because we're backing off.";
+    return false;
+  }
 
-    if (priority == NORMAL_PRIORITY)
-      return DROP;
-    else // Either backoff has ended, or we have permission to bypass it.
-      return CONTINUE;
+  if (session_context_->connection_manager()->HasInvalidAuthToken()) {
+    SDVLOG(1) << "Unable to run a job because we have no valid auth token.";
+    return false;
   }
-  return (priority == CANARY_PRIORITY) ? CONTINUE : SAVE;
+
+  return true;
 }
 
-SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
-    const SyncSessionJob& job,
-    JobPriority priority) {
+bool SyncSchedulerImpl::CanRunNudgeJobNow(JobPriority priority) {
   DCHECK(CalledOnValidThread());
 
-  // POLL jobs do not call this function.
-  DCHECK(job.purpose() == SyncSessionJob::NUDGE ||
-         job.purpose() == SyncSessionJob::CONFIGURATION);
+  if (!CanRunJobNow(priority)) {
+    SDVLOG(1) << "Unable to run a nudge job right now";
+    return false;
+  }
 
-  // See if our type is throttled.
+  // If all types are throttled, do not continue.  Today, we don't treat a
+  // per-datatype "unthrottle" event as something that should force a canary
+  // job. For this reason, there's no good time to reschedule this job to run
+  // -- we'll lazily wait for an independent event to trigger a sync.
   ModelTypeSet throttled_types =
       session_context_->throttled_data_type_tracker()->GetThrottledTypes();
-  if (job.purpose() == SyncSessionJob::NUDGE &&
-      job.source_info().updates_source == GetUpdatesCallerInfo::LOCAL) {
-    ModelTypeSet requested_types;
-    for (ModelTypeInvalidationMap::const_iterator i =
-         job.source_info().types.begin(); i != job.source_info().types.end();
-         ++i) {
-      requested_types.Put(i->first);
-    }
-
-    // If all types are throttled, do not CONTINUE.  Today, we don't treat
-    // a per-datatype "unthrottle" event as something that should force a
-    // canary job. For this reason, there's no good time to reschedule this job
-    // to run -- we'll lazily wait for an independent event to trigger a sync.
-    // Note that there may already be such an event if we're in a WaitInterval,
-    // so we can retry it then.
-    if (!requested_types.Empty() && throttled_types.HasAll(requested_types))
-      return DROP;  // TODO(tim): Don't drop. http://crbug.com/177659
+  if (!nudge_tracker_.GetLocallyModifiedTypes().Empty() &&
+      throttled_types.HasAll(nudge_tracker_.GetLocallyModifiedTypes())) {
+    // TODO(sync): Throttled types should be pruned from the sources list.
+    SDVLOG(1) << "Not running a nudge because we're fully datatype throttled.";
+    return false;
   }
 
-  if (wait_interval_.get())
-    return DecideWhileInWaitInterval(job, priority);
-
   if (mode_ == CONFIGURATION_MODE) {
-    if (job.purpose() == SyncSessionJob::NUDGE)
-      return SAVE;  // Running requires a mode switch.
-    else  // Implies job.purpose() == SyncSessionJob::CONFIGURATION.
-      return CONTINUE;
+    SDVLOG(1) << "Not running nudge because we're in configuration mode.";
+    return false;
   }
 
-  // We are in normal mode.
-  DCHECK_EQ(mode_, NORMAL_MODE);
-  DCHECK_NE(job.purpose(), SyncSessionJob::CONFIGURATION);
-
-  // Note about some subtle scheduling semantics.
-  //
-  // It's possible at this point that |job| is known to be unnecessary, and
-  // dropping it would be perfectly safe and correct. Consider
-  //
-  // 1) |job| is a NUDGE (for any combination of types) with a
-  //    |scheduled_start| time that is less than the time that the last
-  //    successful all-datatype NUDGE completed, and it has a NOTIFICATION
-  //    GetUpdatesCallerInfo value yet offers no new notification hint.
-  //
-  // 2) |job| is a NUDGE with a |scheduled_start| time that is less than
-  //    the time that the last successful matching-datatype NUDGE completed,
-  //    and payloads (hints) are identical to that last successful NUDGE.
-  //
-  //  We avoid cases 1 and 2 by externally synchronizing NUDGE requests --
-  //  scheduling a NUDGE requires command of the sync thread, which is
-  //  impossible* from outside of SyncScheduler if a NUDGE is taking place.
-  //  And if you have command of the sync thread when scheduling a NUDGE and a
-  //  previous NUDGE exists, they will be coalesced and the stale job will be
-  //  cancelled via the session-equality check in DoSyncSessionJob.
-  //
-  //  * It's not strictly "impossible", but it would be reentrant and hence
-  //  illegal. e.g. scheduling a job and re-entering the SyncScheduler is NOT a
-  //  legal side effect of any of the work being done as part of a sync cycle.
-  //  See |no_scheduling_allowed_| for details.
-
-  // Decision now rests on state of auth tokens.
-  if (!session_context_->connection_manager()->HasInvalidAuthToken())
-    return CONTINUE;
-
-  SDVLOG(2) << "No valid auth token. Using that to decide on job.";
-  // Running the job would require updated auth, so we can't honour
-  // job.scheduled_start().
-  return job.purpose() == SyncSessionJob::NUDGE ? SAVE : DROP;
+  return true;
 }
 
 void SyncSchedulerImpl::ScheduleNudgeAsync(
@@ -496,53 +423,38 @@ void SyncSchedulerImpl::ScheduleNudgeImpl(
   SyncSourceInfo info(source, invalidation_map);
   UpdateNudgeTimeRecords(info);
 
-  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
-      SyncSessionJob::NUDGE,
-      TimeTicks::Now() + delay,
-      info,
-      ConfigurationParams()));
-  JobProcessDecision decision = DecideOnJob(*job, NORMAL_PRIORITY);
-  SDVLOG(2) << "Should run "
-            << SyncSessionJob::GetPurposeString(job->purpose())
-            << " in mode " << GetModeString(mode_)
-            << ": " << GetDecisionString(decision);
-  if (decision == DROP) {
+  // Coalesce the new nudge information with any existing information.
+  nudge_tracker_.CoalesceSources(info);
+
+  if (!CanRunNudgeJobNow(NORMAL_PRIORITY))
     return;
-  }
 
-  // Try to coalesce in both SAVE and CONTINUE cases.
-  if (pending_nudge_job_) {
-    pending_nudge_job_->CoalesceSources(job->source_info());
-    if (decision == CONTINUE) {
-      // Only update the scheduled_start if we're going to reschedule.
-      pending_nudge_job_->set_scheduled_start(
-          std::min(job->scheduled_start(),
-                   pending_nudge_job_->scheduled_start()));
-    }
-  } else {
-    pending_nudge_job_ = job.Pass();
+  if (!started_) {
+    SDVLOG_LOC(nudge_location, 2)
+        << "Schedule not started; not running a nudge.";
+    return;
   }
 
-  if (decision == SAVE) {
+  TimeTicks incoming_run_time = TimeTicks::Now() + delay;
+  if (!scheduled_nudge_time_.is_null() &&
+    (scheduled_nudge_time_ < incoming_run_time)) {
+    // Old job arrives sooner than this one.  Don't reschedule it.
     return;
   }
 
-  TimeDelta run_delay =
-      pending_nudge_job_->scheduled_start() - TimeTicks::Now();
-  if (run_delay < TimeDelta::FromMilliseconds(0))
-    run_delay = TimeDelta::FromMilliseconds(0);
+  // Either there is no existing nudge in flight or the incoming nudge should be
+  // made to arrive first (preempt) the existing nudge.  We reschedule in either
+  // case.
   SDVLOG_LOC(nudge_location, 2)
       << "Scheduling a nudge with "
-      << run_delay.InMilliseconds() << " ms delay";
-
-  if (started_) {
-    pending_wakeup_timer_.Start(
-        nudge_location,
-        run_delay,
-        base::Bind(&SyncSchedulerImpl::DoNudgeSyncSessionJob,
-                   weak_ptr_factory_.GetWeakPtr(),
-                   NORMAL_PRIORITY));
-  }
+      << delay.InMilliseconds() << " ms delay";
+  scheduled_nudge_time_ = incoming_run_time;
+  pending_wakeup_timer_.Start(
+      nudge_location,
+      delay,
+      base::Bind(&SyncSchedulerImpl::DoNudgeSyncSessionJob,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 NORMAL_PRIORITY));
 }
 
 const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
@@ -553,103 +465,88 @@ const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
   return "";
 }
 
-const char* SyncSchedulerImpl::GetDecisionString(
-    SyncSchedulerImpl::JobProcessDecision mode) {
-  switch (mode) {
-    ENUM_CASE(CONTINUE);
-    ENUM_CASE(SAVE);
-    ENUM_CASE(DROP);
-  }
-  return "";
-}
-
-bool SyncSchedulerImpl::DoSyncSessionJobImpl(scoped_ptr<SyncSessionJob> job,
-                                             JobPriority priority) {
+void SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {
   DCHECK(CalledOnValidThread());
 
-  base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
-  JobProcessDecision decision = DecideOnJob(*job, priority);
-  SDVLOG(2) << "Should run "
-            << SyncSessionJob::GetPurposeString(job->purpose())
-            << " in mode " << GetModeString(mode_)
-            << " with source " << job->source_info().updates_source
-            << ": " << GetDecisionString(decision);
-  if (decision != CONTINUE) {
-    if (decision == SAVE) {
-      if (job->purpose() == SyncSessionJob::CONFIGURATION) {
-        pending_configure_job_ = job.Pass();
-      } else {
-        pending_nudge_job_ = job.Pass();
-      }
-    } else {
-      DCHECK_EQ(decision, DROP);
-    }
-    return false;
-  }
+  if (!CanRunNudgeJobNow(priority))
+    return;
 
-  DVLOG(2) << "Creating sync session with routes "
-           << ModelSafeRoutingInfoToString(session_context_->routing_info())
-           << "and purpose " << job->purpose();
-  SyncSession session(session_context_, this, job->source_info());
-  bool premature_exit = !syncer_->SyncShare(&session,
-                                            job->start_step(),
-                                            job->end_step());
-  SDVLOG(2) << "Done SyncShare, returned: " << premature_exit;
+  DVLOG(2) << "Will run normal mode sync cycle with routing info "
+           << ModelSafeRoutingInfoToString(session_context_->routing_info());
+  SyncSession session(session_context_, this, nudge_tracker_.source_info());
+  bool premature_exit = !syncer_->SyncShare(&session, SYNCER_BEGIN, SYNCER_END);
+  AdjustPolling(FORCE_RESET);
 
-  bool success = FinishSyncSessionJob(job.get(),
-                                      premature_exit,
-                                      &session);
+  bool success = !premature_exit
+      && !sessions::HasSyncerError(
+          session.status_controller().model_neutral_state());
 
-  if (IsSyncingCurrentlySilenced()) {
-    SDVLOG(2) << "We are currently throttled; scheduling Unthrottle.";
-    // If we're here, it's because |job| was silenced until a server specified
-    // time. (Note, it had to be |job|, because DecideOnJob would not permit
-    // any job through while in WaitInterval::THROTTLED).
-    if (job->purpose() == SyncSessionJob::NUDGE)
-      pending_nudge_job_ = job.Pass();
-    else if (job->purpose() == SyncSessionJob::CONFIGURATION)
-      pending_configure_job_ = job.Pass();
-    else
-      NOTREACHED();
+  if (success) {
+    // That cycle took care of any outstanding work we had.
+    SDVLOG(2) << "Nudge succeeded.";
+    nudge_tracker_.Reset();
+    scheduled_nudge_time_ = base::TimeTicks();
 
-    RestartWaiting();
-    return success;
+    // If we're here, then we successfully reached the server.  End all backoff.
+    wait_interval_.reset();
+    NotifyRetryTime(base::Time());
+    return;
+  } else {
+    HandleFailure(session.status_controller().model_neutral_state());
   }
-
-  if (!success)
-    ScheduleNextSync(job.Pass(), &session);
-
-  return success;
-}
-
-void SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {
-  DoSyncSessionJobImpl(pending_nudge_job_.Pass(), priority);
 }
 
 bool SyncSchedulerImpl::DoConfigurationSyncSessionJob(JobPriority priority) {
-  return DoSyncSessionJobImpl(pending_configure_job_.Pass(), priority);
-}
+  DCHECK(CalledOnValidThread());
+  DCHECK_EQ(mode_, CONFIGURATION_MODE);
 
-bool SyncSchedulerImpl::ShouldPoll() {
-  if (wait_interval_.get()) {
-    SDVLOG(2) << "Not running poll in wait interval.";
+  if (!CanRunJobNow(priority)) {
+    SDVLOG(2) << "Unable to run configure job right now.";
     return false;
   }
 
-  if (mode_ == CONFIGURATION_MODE) {
-    SDVLOG(2) << "Not running poll in configuration mode.";
-    return false;
-  }
+  SDVLOG(2) << "Will run configure SyncShare with routes "
+           << ModelSafeRoutingInfoToString(session_context_->routing_info());
+  SyncSourceInfo source_info(pending_configure_params_->source,
+                             ModelSafeRoutingInfoToInvalidationMap(
+                                 session_context_->routing_info(),
+                                 std::string()));
+  SyncSession session(session_context_, this, source_info);
+  bool premature_exit = !syncer_->SyncShare(&session,
+                                            DOWNLOAD_UPDATES,
+                                            APPLY_UPDATES);
+  AdjustPolling(FORCE_RESET);
 
-  // TODO(rlarocque): Refactor decision-making logic common to all types
-  // of jobs into a shared function.
+  bool success = !premature_exit
+      && !sessions::HasSyncerError(
+          session.status_controller().model_neutral_state());
 
-  if (session_context_->connection_manager()->HasInvalidAuthToken()) {
-    SDVLOG(2) << "Not running poll because auth token is invalid.";
+  if (success) {
+    SDVLOG(2) << "Configure succeeded.";
+    pending_configure_params_->ready_task.Run();
+    pending_configure_params_.reset();
+
+    // If we're here, then we successfully reached the server.  End all backoff.
+    wait_interval_.reset();
+    NotifyRetryTime(base::Time());
+    return true;
+  } else {
+    HandleFailure(session.status_controller().model_neutral_state());
     return false;
   }
+}
 
-  return true;
+void SyncSchedulerImpl::HandleFailure(
+    const sessions::ModelNeutralState& model_neutral_state) {
+  if (IsSyncingCurrentlySilenced()) {
+    SDVLOG(2) << "Was throttled during previous sync cycle.";
+    RestartWaiting();
+  } else {
+    UpdateExponentialBackoff(model_neutral_state);
+    SDVLOG(2) << "Sync cycle failed.  Will back off for "
+        << wait_interval_->length.InMilliseconds() << "ms.";
+    RestartWaiting();
+  }
 }
 
 void SyncSchedulerImpl::DoPollSyncSessionJob() {
@@ -657,31 +554,29 @@ void SyncSchedulerImpl::DoPollSyncSessionJob() {
   ModelTypeInvalidationMap invalidation_map =
       ModelSafeRoutingInfoToInvalidationMap(r, std::string());
   SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, invalidation_map);
-  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(SyncSessionJob::POLL,
-                                                    TimeTicks::Now(),
-                                                    info,
-                                                    ConfigurationParams()));
-
   base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
 
-  if (!ShouldPoll())
+  if (!CanRunJobNow(NORMAL_PRIORITY)) {
+    SDVLOG(2) << "Unable to run a poll job right now.";
+    return;
+  }
+
+  if (mode_ != NORMAL_MODE) {
+    SDVLOG(2) << "Not running poll job in configure mode.";
     return;
+  }
 
-  DVLOG(2) << "Polling with routes "
+  SDVLOG(2) << "Polling with routes "
            << ModelSafeRoutingInfoToString(session_context_->routing_info());
-  SyncSession session(session_context_, this, job->source_info());
-  bool premature_exit = !syncer_->SyncShare(&session,
-                                            job->start_step(),
-                                            job->end_step());
-  SDVLOG(2) << "Done SyncShare, returned: " << premature_exit;
+  SyncSession session(session_context_, this, info);
+  syncer_->SyncShare(&session, SYNCER_BEGIN, SYNCER_END);
 
-  FinishSyncSessionJob(job.get(), premature_exit, &session);
+  AdjustPolling(UPDATE_INTERVAL);
 
   if (IsSyncingCurrentlySilenced()) {
-    // Normally we would only call RestartWaiting() if we had a
-    // pending_nudge_job_ or pending_configure_job_ set.  In this case, it's
-    // possible that neither is set.  We create the wait interval anyway because
-    // we need it to make sure we get unthrottled on time.
+    SDVLOG(2) << "Poll request got us throttled.";
+    // The OnSilencedUntil() call set up the WaitInterval for us.  All we need
+    // to do is start the timer.
     RestartWaiting();
   }
 }
@@ -711,52 +606,7 @@ void SyncSchedulerImpl::UpdateNudgeTimeRecords(const SyncSourceInfo& info) {
   }
 }
 
-bool SyncSchedulerImpl::FinishSyncSessionJob(SyncSessionJob* job,
-                                             bool exited_prematurely,
-                                             SyncSession* session) {
-  DCHECK(CalledOnValidThread());
-
-  // Let job know that we're through syncing (calling SyncShare) at this point.
-  bool succeeded = false;
-  {
-    base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
-    succeeded = job->Finish(exited_prematurely, session);
-  }
-
-  SDVLOG(2) << "Updating the next polling time after SyncMain";
-
-  AdjustPolling(job);
-
-  if (succeeded) {
-    // No job currently supported by the scheduler could succeed without
-    // successfully reaching the server.  Therefore, if we make it here, it is
-    // appropriate to reset the backoff interval.
-    wait_interval_.reset();
-    NotifyRetryTime(base::Time());
-    SDVLOG(2) << "Job succeeded so not scheduling more jobs";
-  }
-
-  return succeeded;
-}
-
-void SyncSchedulerImpl::ScheduleNextSync(
-    scoped_ptr<SyncSessionJob> finished_job,
-    SyncSession* session) {
-  DCHECK(CalledOnValidThread());
-  DCHECK(finished_job->purpose() == SyncSessionJob::CONFIGURATION
-         || finished_job->purpose() == SyncSessionJob::NUDGE);
-
-  // TODO(rlarocque): There's no reason why we should blindly backoff and retry
-  // if we don't succeed.  Some types of errors are not likely to disappear on
-  // their own.  With the return values now available in the old_job.session,
-  // we should be able to detect such errors and only retry when we detect
-  // transient errors.
-
-  SDVLOG(2) << "SyncShare job failed; will start or update backoff";
-  HandleContinuationError(finished_job.Pass(), session);
-}
-
-void SyncSchedulerImpl::AdjustPolling(const SyncSessionJob* old_job) {
+void SyncSchedulerImpl::AdjustPolling(PollAdjustType type) {
   DCHECK(CalledOnValidThread());
 
   TimeDelta poll  = (!session_context_->notifications_enabled()) ?
@@ -765,7 +615,7 @@ void SyncSchedulerImpl::AdjustPolling(const SyncSessionJob* old_job) {
   bool rate_changed = !poll_timer_.IsRunning() ||
                        poll != poll_timer_.GetCurrentDelay();
 
-  if (old_job && old_job->purpose() != SyncSessionJob::POLL && !rate_changed)
+  if (type == FORCE_RESET && !rate_changed)
     poll_timer_.Reset();
 
   if (!rate_changed)
@@ -780,6 +630,9 @@ void SyncSchedulerImpl::AdjustPolling(const SyncSessionJob* old_job) {
 void SyncSchedulerImpl::RestartWaiting() {
   CHECK(wait_interval_.get());
   DCHECK(wait_interval_->length >= TimeDelta::FromSeconds(0));
+  NotifyRetryTime(base::Time::Now() + wait_interval_->length);
+  SDVLOG(2) << "Starting WaitInterval timer of length "
+      << wait_interval_->length.InMilliseconds() << "ms.";
   if (wait_interval_->mode == WaitInterval::THROTTLED) {
     pending_wakeup_timer_.Start(
         FROM_HERE,
@@ -795,39 +648,15 @@ void SyncSchedulerImpl::RestartWaiting() {
   }
 }
 
-void SyncSchedulerImpl::HandleContinuationError(
-    scoped_ptr<SyncSessionJob> old_job,
-    SyncSession* session) {
+void SyncSchedulerImpl::UpdateExponentialBackoff(
+    const sessions::ModelNeutralState& model_neutral_state) {
   DCHECK(CalledOnValidThread());
 
   TimeDelta length = delay_provider_->GetDelay(
       IsBackingOff() ? wait_interval_->length :
-          delay_provider_->GetInitialDelay(
-              session->status_controller().model_neutral_state()));
-
-  SDVLOG(2) << "In handle continuation error with "
-            << SyncSessionJob::GetPurposeString(old_job->purpose())
-            << " job. The time delta(ms) is "
-            << length.InMilliseconds();
-
+          delay_provider_->GetInitialDelay(model_neutral_state));
   wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
                                         length));
-  NotifyRetryTime(base::Time::Now() + length);
-  old_job->set_scheduled_start(TimeTicks::Now() + 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());
-    DCHECK(!pending_configure_job_);
-    pending_configure_job_ = old_job.Pass();
-  } else {
-    // We're not in configure mode so we should not have a configure job.
-    DCHECK(!pending_configure_job_);
-    DCHECK(!pending_nudge_job_);
-    pending_nudge_job_ = old_job.Pass();
-  }
-
-  RestartWaiting();
 }
 
 void SyncSchedulerImpl::RequestStop(const base::Closure& callback) {
@@ -849,11 +678,9 @@ void SyncSchedulerImpl::StopImpl(const base::Closure& callback) {
   NotifyRetryTime(base::Time());
   poll_timer_.Stop();
   pending_wakeup_timer_.Stop();
-  pending_nudge_job_.reset();
-  pending_configure_job_.reset();
-  if (started_) {
+  pending_configure_params_.reset();
+  if (started_)
     started_ = false;
-  }
   if (!callback.is_null())
     callback.Run();
 }
@@ -863,10 +690,10 @@ void SyncSchedulerImpl::StopImpl(const base::Closure& callback) {
 void SyncSchedulerImpl::TryCanaryJob() {
   DCHECK(CalledOnValidThread());
 
-  if (mode_ == CONFIGURATION_MODE && pending_configure_job_) {
+  if (mode_ == CONFIGURATION_MODE && pending_configure_params_) {
     SDVLOG(2) << "Found pending configure job; will run as canary";
     DoConfigurationSyncSessionJob(CANARY_PRIORITY);
-  } else if (mode_ == NORMAL_MODE && pending_nudge_job_) {
+  } else if (mode_ == NORMAL_MODE && !nudge_tracker_.IsEmpty()) {
     SDVLOG(2) << "Found pending nudge job; will run as canary";
     DoNudgeSyncSessionJob(CANARY_PRIORITY);
   } else {
diff --git a/sync/engine/sync_scheduler_impl.h b/sync/engine/sync_scheduler_impl.h
index dfdcc80..efcccd3 100644
--- a/sync/engine/sync_scheduler_impl.h
+++ b/sync/engine/sync_scheduler_impl.h
@@ -22,11 +22,11 @@
 #include "sync/engine/net/server_connection_manager.h"
 #include "sync/engine/nudge_source.h"
 #include "sync/engine/sync_scheduler.h"
-#include "sync/engine/sync_session_job.h"
 #include "sync/engine/syncer.h"
 #include "sync/internal_api/public/base/model_type_invalidation_map.h"
 #include "sync/internal_api/public/engine/polling_constants.h"
 #include "sync/internal_api/public/util/weak_handle.h"
+#include "sync/sessions/nudge_tracker.h"
 #include "sync/sessions/sync_session.h"
 #include "sync/sessions/sync_session_context.h"
 
@@ -34,6 +34,10 @@ namespace syncer {
 
 class BackoffDelayProvider;
 
+namespace sessions {
+struct ModelNeutralState;
+}
+
 class SYNC_EXPORT_PRIVATE SyncSchedulerImpl
     : public SyncScheduler,
       public base::NonThreadSafe {
@@ -83,15 +87,6 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl
       const sessions::SyncSessionSnapshot& snapshot) OVERRIDE;
 
  private:
-  enum JobProcessDecision {
-    // Indicates we should continue with the current job.
-    CONTINUE,
-    // Indicates that we should save it to be processed later.
-    SAVE,
-    // Indicates we should drop this job.
-    DROP,
-  };
-
   enum JobPriority {
     // Non-canary jobs respect exponential backoff.
     NORMAL_PRIORITY,
@@ -99,10 +94,18 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl
     CANARY_PRIORITY
   };
 
+  enum PollAdjustType {
+    // Restart the poll interval.
+    FORCE_RESET,
+    // Restart the poll interval only if its length has changed.
+    UPDATE_INTERVAL,
+  };
+
   friend class SyncSchedulerTest;
   friend class SyncSchedulerWhiteboxTest;
   friend class SyncerTest;
 
+  FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest, NoNudgesInConfigureMode);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest,
       DropNudgeWhileExponentialBackOff);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest, SaveNudge);
@@ -145,57 +148,37 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl
 
   static const char* GetModeString(Mode mode);
 
-  static const char* GetDecisionString(JobProcessDecision decision);
-
-  // Invoke the syncer to perform a non-POLL job.
-  bool DoSyncSessionJobImpl(scoped_ptr<SyncSessionJob> job,
-                            JobPriority priority);
-
   // Invoke the syncer to perform a nudge job.
   void DoNudgeSyncSessionJob(JobPriority priority);
 
   // Invoke the syncer to perform a configuration job.
   bool DoConfigurationSyncSessionJob(JobPriority priority);
 
-  // Returns whether or not it's safe to run a poll job at this time.
-  bool ShouldPoll();
+  // Helper function for Do{Nudge,Configuration}SyncSessionJob.
+  void HandleFailure(
+      const sessions::ModelNeutralState& model_neutral_state);
 
   // Invoke the Syncer to perform a poll job.
   void DoPollSyncSessionJob();
 
-  // Called after the Syncer has performed the sync represented by |job|, to
-  // reset our state.  |exited_prematurely| is true if the Syncer did not
-  // cycle from job.start_step() to job.end_step(), likely because the
-  // scheduler was forced to quit the job mid-way through.
-  bool FinishSyncSessionJob(SyncSessionJob* job,
-                            bool exited_prematurely,
-                            sessions::SyncSession* session);
-
-  // Helper to schedule retries of a failed configure or nudge job.
-  void ScheduleNextSync(scoped_ptr<SyncSessionJob> finished_job,
-                        sessions::SyncSession* session);
-
-  // Helper to configure polling intervals. Used by Start and ScheduleNextSync.
-  void AdjustPolling(const SyncSessionJob* old_job);
+  // Adjusts the poll timer to account for new poll interval, and possibly
+  // resets the poll interval, depedning on the flag's value.
+  void AdjustPolling(PollAdjustType type);
 
   // Helper to restart waiting with |wait_interval_|'s timer.
   void RestartWaiting();
 
-  // Helper to ScheduleNextSync in case of consecutive sync errors.
-  void HandleContinuationError(scoped_ptr<SyncSessionJob> old_job,
-                               sessions::SyncSession* session);
+  // Helper to adjust our wait interval when we expereince a transient failure.
+  void UpdateExponentialBackoff(
+      const sessions::ModelNeutralState& model_neutral_state);
 
-  // Decide whether we should CONTINUE, SAVE or DROP the job.
-  JobProcessDecision DecideOnJob(const SyncSessionJob& job,
-                                 JobPriority priority);
+  // Determines if we're allowed to contact the server right now.
+  bool CanRunJobNow(JobPriority priority);
 
-  // Decide on whether to CONTINUE, SAVE or DROP the job when we are in
-  // backoff mode.
-  JobProcessDecision DecideWhileInWaitInterval(const SyncSessionJob& job,
-                                               JobPriority priority);
+  // Determines if we're allowed to contact the server right now.
+  bool CanRunNudgeJobNow(JobPriority priority);
 
-  // 'Impl' here refers to real implementation of public functions, running on
-  // |thread_|.
+  // 'Impl' here refers to real implementation of public functions.
   void StopImpl(const base::Closure& callback);
 
   // If the scheduler's current state supports it, this will create a job based
@@ -230,9 +213,9 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl
   // Creates a session for a poll and performs the sync.
   void PollTimerCallback();
 
-  // Called once the first time thread_ is started to broadcast an initial
-  // session snapshot containing data like initial_sync_ended.  Important when
-  // the client starts up and does not need to perform an initial sync.
+  // Called as we are started to broadcast an initial session snapshot
+  // containing data like initial_sync_ended.  Important when the client starts
+  // up and does not need to perform an initial sync.
   void SendInitialSnapshot();
 
   // This is used for histogramming and analysis of ScheduleNudge* APIs.
@@ -280,13 +263,15 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl
   // The event that will wake us up.
   base::OneShotTimer<SyncSchedulerImpl> pending_wakeup_timer_;
 
-  // Pending configure job storage.  Note that
-  // (mode_ != CONFIGURATION_MODE) \implies !pending_configure_job_.
-  scoped_ptr<SyncSessionJob> pending_configure_job_;
+  // Storage for variables related to an in-progress configure request.  Note
+  // that (mode_ != CONFIGURATION_MODE) \implies !pending_configure_params_.
+  scoped_ptr<ConfigurationParams> pending_configure_params_;
+
+  // If we have a nudge pending to run soon, it will be listed here.
+  base::TimeTicks scheduled_nudge_time_;
 
-  // Pending nudge job storage.  These jobs can exist in CONFIGURATION_MODE, but
-  // they will be run only in NORMAL_MODE.
-  scoped_ptr<SyncSessionJob> pending_nudge_job_;
+  // Keeps track of work that the syncer needs to handle.
+  sessions::NudgeTracker nudge_tracker_;
 
   // Invoked to run through the sync cycle.
   scoped_ptr<Syncer> syncer_;
diff --git a/sync/engine/sync_scheduler_unittest.cc b/sync/engine/sync_scheduler_unittest.cc
index 444f5eb..a890117 100644
--- a/sync/engine/sync_scheduler_unittest.cc
+++ b/sync/engine/sync_scheduler_unittest.cc
@@ -724,7 +724,7 @@ TEST_F(SyncSchedulerTest, ThrottlingDoesThrottle) {
   StartSyncScheduler(SyncScheduler::NORMAL_MODE);
 
   scheduler()->ScheduleNudgeAsync(
-      zero(), NUDGE_SOURCE_LOCAL, types, FROM_HERE);
+      TimeDelta::FromMicroseconds(1), NUDGE_SOURCE_LOCAL, types, FROM_HERE);
   PumpLoop();
 
   StartSyncScheduler(SyncScheduler::CONFIGURATION_MODE);
diff --git a/sync/engine/sync_scheduler_whitebox_unittest.cc b/sync/engine/sync_scheduler_whitebox_unittest.cc
deleted file mode 100644
index 7a550fa..0000000
--- a/sync/engine/sync_scheduler_whitebox_unittest.cc
+++ /dev/null
@@ -1,258 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "base/message_loop.h"
-#include "base/time.h"
-#include "sync/engine/backoff_delay_provider.h"
-#include "sync/engine/sync_scheduler_impl.h"
-#include "sync/engine/throttled_data_type_tracker.h"
-#include "sync/internal_api/public/engine/polling_constants.h"
-#include "sync/sessions/sync_session_context.h"
-#include "sync/sessions/test_util.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"
-#include "sync/test/fake_extensions_activity_monitor.h"
-#include "testing/gmock/include/gmock/gmock.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-using base::TimeDelta;
-using base::TimeTicks;
-
-namespace syncer {
-using sessions::SyncSession;
-using sessions::SyncSessionContext;
-using sessions::SyncSourceInfo;
-using sync_pb::GetUpdatesCallerInfo;
-
-class SyncSchedulerWhiteboxTest : public testing::Test {
- public:
-  virtual void SetUp() {
-    dir_maker_.SetUp();
-    Syncer* syncer = new Syncer();
-
-    ModelSafeRoutingInfo routes;
-    routes[BOOKMARKS] = GROUP_UI;
-    routes[NIGORI] = GROUP_PASSIVE;
-
-    workers_.push_back(make_scoped_refptr(new FakeModelWorker(GROUP_UI)));
-    workers_.push_back(make_scoped_refptr(new FakeModelWorker(GROUP_PASSIVE)));
-
-    std::vector<ModelSafeWorker*> workers;
-    for (std::vector<scoped_refptr<FakeModelWorker> >::iterator it =
-         workers_.begin(); it != workers_.end(); ++it) {
-      workers.push_back(it->get());
-    }
-
-    connection_.reset(new MockConnectionManager(NULL));
-    throttled_data_type_tracker_.reset(new ThrottledDataTypeTracker(NULL));
-    context_.reset(
-        new SyncSessionContext(
-            connection_.get(), dir_maker_.directory(),
-            workers, &extensions_activity_monitor_,
-            throttled_data_type_tracker_.get(),
-            std::vector<SyncEngineEventListener*>(), NULL, NULL,
-            true,  // enable keystore encryption
-            "fake_invalidator_client_id"));
-    context_->set_notifications_enabled(true);
-    context_->set_account_name("Test");
-    scheduler_.reset(
-        new SyncSchedulerImpl("TestSyncSchedulerWhitebox",
-            BackoffDelayProvider::FromDefaults(),
-            context(),
-            syncer));
-  }
-
-  virtual void TearDown() {
-    scheduler_.reset();
-  }
-
-  void SetMode(SyncScheduler::Mode mode) {
-    scheduler_->mode_ = mode;
-  }
-
-  void ResetWaitInterval() {
-    scheduler_->wait_interval_.reset();
-  }
-
-  void SetWaitIntervalToThrottled() {
-    scheduler_->wait_interval_.reset(new SyncSchedulerImpl::WaitInterval(
-        SyncSchedulerImpl::WaitInterval::THROTTLED, TimeDelta::FromSeconds(1)));
-  }
-
-  void SetWaitIntervalToExponentialBackoff() {
-    scheduler_->wait_interval_.reset(
-       new SyncSchedulerImpl::WaitInterval(
-       SyncSchedulerImpl::WaitInterval::EXPONENTIAL_BACKOFF,
-       TimeDelta::FromSeconds(1)));
-  }
-
-  SyncSchedulerImpl::JobProcessDecision DecideOnJob(
-      const SyncSessionJob& job,
-      SyncSchedulerImpl::JobPriority priority) {
-    return scheduler_->DecideOnJob(job, priority);
-  }
-
-  void InitializeSyncerOnNormalMode() {
-    SetMode(SyncScheduler::NORMAL_MODE);
-    ResetWaitInterval();
-  }
-
-  SyncSchedulerImpl::JobProcessDecision CreateAndDecideJob(
-      SyncSessionJob::Purpose purpose) {
-    SyncSessionJob job(purpose, TimeTicks::Now(),
-                       SyncSourceInfo(), ConfigurationParams());
-    return DecideOnJob(job, SyncSchedulerImpl::NORMAL_PRIORITY);
-  }
-
-  bool ShouldPoll() {
-    return scheduler_->ShouldPoll();
-  }
-
-  SyncSessionContext* context() { return context_.get(); }
-
- private:
-  MessageLoop message_loop_;
-  scoped_ptr<MockConnectionManager> connection_;
-  scoped_ptr<SyncSessionContext> context_;
-  std::vector<scoped_refptr<FakeModelWorker> > workers_;
-  FakeExtensionsActivityMonitor extensions_activity_monitor_;
-  scoped_ptr<ThrottledDataTypeTracker> throttled_data_type_tracker_;
-  TestDirectorySetterUpper dir_maker_;
-
- protected:
-  // Declared here to ensure it is destructed before the objects it references.
-  scoped_ptr<SyncSchedulerImpl> scheduler_;
-};
-
-TEST_F(SyncSchedulerWhiteboxTest, SaveNudge) {
-  InitializeSyncerOnNormalMode();
-
-  // Now set the mode to configure.
-  SetMode(SyncScheduler::CONFIGURATION_MODE);
-
-  SyncSchedulerImpl::JobProcessDecision decision =
-      CreateAndDecideJob(SyncSessionJob::NUDGE);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::SAVE);
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, SaveNudgeWhileTypeThrottled) {
-  InitializeSyncerOnNormalMode();
-
-  const ModelTypeSet types(BOOKMARKS);
-
-  // Mark bookmarks as throttled.
-  context()->throttled_data_type_tracker()->SetUnthrottleTime(
-      types, base::TimeTicks::Now() + base::TimeDelta::FromHours(2));
-
-  const ModelTypeInvalidationMap& invalidation_map =
-      ModelTypeSetToInvalidationMap(types, std::string());
-
-  SyncSourceInfo info(GetUpdatesCallerInfo::LOCAL, invalidation_map);
-
-  // Now schedule a nudge with just bookmarks and the change is local.
-  SyncSessionJob job(SyncSessionJob::NUDGE,
-                     TimeTicks::Now(),
-                     info,
-                     ConfigurationParams());
-  SyncSchedulerImpl::JobProcessDecision decision =
-      DecideOnJob(job, SyncSchedulerImpl::NORMAL_PRIORITY);
-  // TODO(tim): This shouldn't drop. Bug 177659.
-  EXPECT_EQ(decision, SyncSchedulerImpl::DROP);
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, ContinueNudge) {
-  InitializeSyncerOnNormalMode();
-
-  SyncSchedulerImpl::JobProcessDecision decision = CreateAndDecideJob(
-      SyncSessionJob::NUDGE);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::CONTINUE);
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, ContinuePoll) {
-  InitializeSyncerOnNormalMode();
-  EXPECT_TRUE(ShouldPoll());
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, DropPollInConfigureMode) {
-  InitializeSyncerOnNormalMode();
-  SetMode(SyncScheduler::CONFIGURATION_MODE);
-  EXPECT_FALSE(ShouldPoll());
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, DropPollWhenThrottled) {
-  InitializeSyncerOnNormalMode();
-  SetWaitIntervalToThrottled();
-  EXPECT_FALSE(ShouldPoll());
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, DropPollInBackoff) {
-  InitializeSyncerOnNormalMode();
-  SetWaitIntervalToExponentialBackoff();
-  EXPECT_FALSE(ShouldPoll());
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, ContinueConfiguration) {
-  InitializeSyncerOnNormalMode();
-  SetMode(SyncScheduler::CONFIGURATION_MODE);
-
-  SyncSchedulerImpl::JobProcessDecision decision = CreateAndDecideJob(
-      SyncSessionJob::CONFIGURATION);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::CONTINUE);
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, SaveConfigurationWhileThrottled) {
-  InitializeSyncerOnNormalMode();
-  SetMode(SyncScheduler::CONFIGURATION_MODE);
-
-  SetWaitIntervalToThrottled();
-
-  SyncSchedulerImpl::JobProcessDecision decision = CreateAndDecideJob(
-      SyncSessionJob::CONFIGURATION);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::SAVE);
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, SaveNudgeWhileThrottled) {
-  InitializeSyncerOnNormalMode();
-  SetMode(SyncScheduler::CONFIGURATION_MODE);
-
-  SetWaitIntervalToThrottled();
-
-  SyncSchedulerImpl::JobProcessDecision decision = CreateAndDecideJob(
-      SyncSessionJob::NUDGE);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::SAVE);
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, DropNudgeWhileExponentialBackOff) {
-  InitializeSyncerOnNormalMode();
-  SetMode(SyncScheduler::NORMAL_MODE);
-  SetWaitIntervalToExponentialBackoff();
-
-  SyncSchedulerImpl::JobProcessDecision decision = CreateAndDecideJob(
-      SyncSessionJob::NUDGE);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::DROP);
-}
-
-TEST_F(SyncSchedulerWhiteboxTest, ContinueCanaryJobConfig) {
-  InitializeSyncerOnNormalMode();
-  SetMode(SyncScheduler::CONFIGURATION_MODE);
-  SetWaitIntervalToExponentialBackoff();
-
-  SyncSessionJob job(SyncSessionJob::CONFIGURATION,
-                     TimeTicks::Now(), SyncSourceInfo(),
-                     ConfigurationParams());
-
-  SyncSchedulerImpl::JobProcessDecision decision =
-      DecideOnJob(job, SyncSchedulerImpl::CANARY_PRIORITY);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::CONTINUE);
-}
-
-}  // namespace syncer
diff --git a/sync/engine/sync_session_job.cc b/sync/engine/sync_session_job.cc
deleted file mode 100644
index 84b491e..0000000
--- a/sync/engine/sync_session_job.cc
+++ /dev/null
@@ -1,130 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "sync/engine/sync_session_job.h"
-#include "sync/internal_api/public/sessions/model_neutral_state.h"
-
-namespace syncer {
-
-SyncSessionJob::~SyncSessionJob() {
-}
-
-SyncSessionJob::SyncSessionJob(
-    Purpose purpose,
-    base::TimeTicks start,
-    const sessions::SyncSourceInfo& source_info,
-    const ConfigurationParams& config_params)
-    : purpose_(purpose),
-      source_info_(source_info),
-      scheduled_start_(start),
-      config_params_(config_params) {
-}
-
-#define ENUM_CASE(x) case x: return #x; break;
-const char* SyncSessionJob::GetPurposeString(SyncSessionJob::Purpose purpose) {
-  switch (purpose) {
-    ENUM_CASE(UNKNOWN);
-    ENUM_CASE(POLL);
-    ENUM_CASE(NUDGE);
-    ENUM_CASE(CONFIGURATION);
-  }
-  NOTREACHED();
-  return "";
-}
-#undef ENUM_CASE
-
-bool SyncSessionJob::Finish(bool early_exit, sessions::SyncSession* session) {
-  // Did we quit part-way through due to premature exit condition, like
-  // shutdown?  Note that this branch will not be hit for other kinds
-  // of early return scenarios, like certain kinds of transient errors.
-  if (early_exit)
-    return false;
-
-  // Did we hit any errors along the way?
-  if (sessions::HasSyncerError(
-      session->status_controller().model_neutral_state())) {
-    return false;
-  }
-
-  const sessions::ModelNeutralState& state(
-      session->status_controller().model_neutral_state());
-  switch (purpose_) {
-    case POLL:
-    case NUDGE:
-      DCHECK_NE(state.last_download_updates_result, UNSET);
-      DCHECK_NE(state.commit_result, UNSET);
-      break;
-    case CONFIGURATION:
-      DCHECK_NE(state.last_download_updates_result, UNSET);
-      break;
-    case UNKNOWN:
-    default:
-      NOTREACHED();
-  }
-
-  if (!config_params_.ready_task.is_null())
-    config_params_.ready_task.Run();
-  return true;
-}
-
-void SyncSessionJob::CoalesceSources(const sessions::SyncSourceInfo& source) {
-  CoalesceStates(source.types, &source_info_.types);
-  source_info_.updates_source = source.updates_source;
-}
-
-SyncSessionJob::Purpose SyncSessionJob::purpose() const {
-  return purpose_;
-}
-
-const sessions::SyncSourceInfo& SyncSessionJob::source_info() const {
-  return source_info_;
-}
-
-base::TimeTicks SyncSessionJob::scheduled_start() const {
-  return scheduled_start_;
-}
-
-void SyncSessionJob::set_scheduled_start(base::TimeTicks start) {
-  scheduled_start_ = start;
-};
-
-ConfigurationParams SyncSessionJob::config_params() const {
-  return config_params_;
-}
-
-SyncerStep SyncSessionJob::start_step() const {
-  SyncerStep start, end;
-  GetSyncerStepsForPurpose(purpose_, &start, &end);
-  return start;
-}
-
-SyncerStep SyncSessionJob::end_step() const {
-  SyncerStep start, end;
-  GetSyncerStepsForPurpose(purpose_, &start, &end);
-  return end;
-}
-
-// static
-void SyncSessionJob::GetSyncerStepsForPurpose(Purpose purpose,
-                                              SyncerStep* start,
-                                              SyncerStep* end) {
-  switch (purpose) {
-    case SyncSessionJob::CONFIGURATION:
-      *start = DOWNLOAD_UPDATES;
-      *end = APPLY_UPDATES;
-      return;
-    case SyncSessionJob::NUDGE:
-    case SyncSessionJob::POLL:
-      *start = SYNCER_BEGIN;
-      *end = SYNCER_END;
-      return;
-    default:
-      NOTREACHED();
-      *start = SYNCER_END;
-      *end = SYNCER_END;
-      return;
-  }
-}
-
-}  // namespace syncer
diff --git a/sync/engine/sync_session_job.h b/sync/engine/sync_session_job.h
deleted file mode 100644
index 7db96f3..0000000
--- a/sync/engine/sync_session_job.h
+++ /dev/null
@@ -1,88 +0,0 @@
-// Copyright 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_ENGINE_SYNC_SESSION_JOB_H_
-#define SYNC_ENGINE_SYNC_SESSION_JOB_H_
-
-#include "base/memory/scoped_ptr.h"
-#include "base/memory/weak_ptr.h"
-#include "base/time.h"
-#include "sync/base/sync_export.h"
-#include "sync/engine/sync_scheduler.h"
-#include "sync/engine/syncer.h"
-#include "sync/sessions/sync_session.h"
-
-namespace syncer {
-
-class SYNC_EXPORT_PRIVATE SyncSessionJob {
- public:
-  enum Purpose {
-    // Uninitialized state, should never be hit in practice.
-    UNKNOWN = -1,
-    // Our poll timer schedules POLL jobs periodically based on a server
-    // assigned poll interval.
-    POLL,
-    // A nudge task can come from a variety of components needing to force
-    // a sync.  The source is inferable from |session.source()|.
-    NUDGE,
-    // Typically used for fetching updates for a subset of the enabled types
-    // during initial sync or reconfiguration.
-    CONFIGURATION,
-  };
-
-  SyncSessionJob(Purpose purpose,
-                 base::TimeTicks start,
-                 const sessions::SyncSourceInfo& source_info,
-                 const ConfigurationParams& config_params);
-  ~SyncSessionJob();
-
-  // Overwrite the sync update source with the most recent and merge the
-  // type/state map.
-  void CoalesceSources(const sessions::SyncSourceInfo& source);
-
-  // Record that the scheduler has deemed the job as finished and give it a
-  // chance to perform any remaining cleanup and/or notification completion
-  // callback invocations.
-  // |early_exit| specifies whether the job 1) cycled through all the
-  // SyncerSteps it needed, or 2) was pre-empted by the scheduler.
-  // Returns true if we completely ran the session without errors.
-  // There are many errors that could prevent a sync cycle from succeeding,
-  // such as invalid local state, inability to contact the server, inability
-  // to authenticate with the server, and server errors.  What they have in
-  // common is that the we either need to take some action and then retry the
-  // sync cycle or, in the case of transient errors, retry after some backoff
-  // timer has expired.  Most importantly, the SyncScheduler should not assume
-  // that the original action that triggered the sync cycle (ie. a nudge or a
-  // notification) has been properly serviced.
-  bool Finish(bool early_exit, sessions::SyncSession* session);
-
-  static const char* GetPurposeString(Purpose purpose);
-  static void GetSyncerStepsForPurpose(Purpose purpose,
-                                       SyncerStep* start,
-                                       SyncerStep* end);
-
-  Purpose purpose() const;
-  const sessions::SyncSourceInfo& source_info() const;
-  base::TimeTicks scheduled_start() const;
-  void set_scheduled_start(base::TimeTicks start);
-  SyncerStep start_step() const;
-  SyncerStep end_step() const;
-  ConfigurationParams config_params() const;
-
- private:
-  const Purpose purpose_;
-  sessions::SyncSourceInfo source_info_;
-
-  base::TimeTicks scheduled_start_;
-
-  // Only used for purpose_ == CONFIGURATION.  This, and different Finish() and
-  // Succeeded() behavior may be arguments to subclass in the future.
-  const ConfigurationParams config_params_;
-
-  DISALLOW_COPY_AND_ASSIGN(SyncSessionJob);
-};
-
-}  // namespace syncer
-
-#endif  // SYNC_ENGINE_SYNC_SESSION_JOB_H_
diff --git a/sync/engine/sync_session_job_unittest.cc b/sync/engine/sync_session_job_unittest.cc
deleted file mode 100644
index 8847833..0000000
--- a/sync/engine/sync_session_job_unittest.cc
+++ /dev/null
@@ -1,166 +0,0 @@
-// Copyright (c) 2012 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "sync/engine/sync_session_job.h"
-
-#include "base/bind.h"
-#include "base/bind_helpers.h"
-#include "base/time.h"
-#include "sync/internal_api/public/base/model_type_invalidation_map.h"
-#include "sync/sessions/sync_session.h"
-#include "sync/sessions/sync_session_context.h"
-#include "sync/sessions/test_util.h"
-#include "sync/test/engine/fake_model_worker.h"
-#include "testing/gmock/include/gmock/gmock.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-using base::TimeTicks;
-
-namespace syncer {
-
-using sessions::SyncSession;
-
-class MockDelegate : public SyncSession::Delegate {
- public:
-   MockDelegate() {}
-   ~MockDelegate() {}
-
-  MOCK_METHOD0(IsSyncingCurrentlySilenced, bool());
-  MOCK_METHOD1(OnReceivedShortPollIntervalUpdate, void(const base::TimeDelta&));
-  MOCK_METHOD1(OnReceivedLongPollIntervalUpdate ,void(const base::TimeDelta&));
-  MOCK_METHOD1(OnReceivedSessionsCommitDelay, void(const base::TimeDelta&));
-  MOCK_METHOD1(OnSyncProtocolError, void(const sessions::SyncSessionSnapshot&));
-  MOCK_METHOD0(OnShouldStopSyncingPermanently, void());
-  MOCK_METHOD1(OnSilencedUntil, void(const base::TimeTicks&));
-};
-
-class SyncSessionJobTest : public testing::Test {
- public:
-  SyncSessionJobTest() : config_params_callback_invoked_(false) {}
-  virtual void SetUp() {
-    routes_.clear();
-    workers_.clear();
-    config_params_callback_invoked_ = false;
-    routes_[BOOKMARKS] = GROUP_PASSIVE;
-    scoped_refptr<ModelSafeWorker> passive_worker(
-        new FakeModelWorker(GROUP_PASSIVE));
-    workers_.push_back(passive_worker);
-    std::vector<ModelSafeWorker*> workers;
-    GetWorkers(&workers);
-    context_.reset(new sessions::SyncSessionContext(
-        NULL,  // |connection_manager|
-        NULL,  // |directory|
-        workers,
-        NULL,  // |extensions_activity_monitor|
-        NULL,  // |throttled_data_type_tracker|
-        std::vector<SyncEngineEventListener*>(),
-        NULL,  // |debug_info_getter|
-        NULL,  // |traffic_recorder|
-        true,  // |enable keystore encryption|
-        "fake_invalidator_client_id"));
-    context_->set_routing_info(routes_);
-  }
-
-  scoped_ptr<SyncSession> MakeSession() {
-    sessions::SyncSourceInfo info(MakeSourceInfo());
-    std::vector<sessions::SyncSourceInfo> sources_list;
-    return scoped_ptr<SyncSession>(
-        new SyncSession(context_.get(), &delegate_, info));
-  }
-
-  sessions::SyncSourceInfo MakeSourceInfo() {
-    sessions::SyncSourceInfo info(sync_pb::GetUpdatesCallerInfo::LOCAL,
-                                  ModelTypeInvalidationMap());
-    return info;
-  }
-
-  ModelTypeSet ParamsMeaningAllEnabledTypes() {
-    ModelTypeSet request_params(BOOKMARKS, AUTOFILL);
-    return request_params;
-  }
-
-  ModelTypeSet ParamsMeaningJustOneEnabledType() {
-    return ModelTypeSet(AUTOFILL);
-  }
-
-  void GetWorkers(std::vector<ModelSafeWorker*>* out) const {
-    out->clear();
-    for (std::vector<scoped_refptr<ModelSafeWorker> >::const_iterator it =
-             workers_.begin(); it != workers_.end(); ++it) {
-      out->push_back(it->get());
-    }
-  }
-
-  void ConfigurationParamsCallback() {
-    config_params_callback_invoked_ = true;
-  }
-
-  bool config_params_callback_invoked() const {
-    return config_params_callback_invoked_;
-  }
-
-  sessions::SyncSessionContext* context() { return context_.get(); }
-  SyncSession::Delegate* delegate() { return &delegate_; }
-  const ModelSafeRoutingInfo& routes() { return routes_; }
-
-  // Checks that the two jobs are "clones" as defined by SyncSessionJob.
-  void ExpectClones(SyncSessionJob* job, SyncSessionJob* clone) {
-    EXPECT_EQ(job->purpose(), clone->purpose());
-    EXPECT_EQ(job->scheduled_start(), clone->scheduled_start());
-    EXPECT_EQ(job->start_step(), clone->start_step());
-    EXPECT_EQ(job->end_step(), clone->end_step());
-  }
-
- private:
-  scoped_ptr<sessions::SyncSessionContext> context_;
-  std::vector<scoped_refptr<ModelSafeWorker> > workers_;
-  MockDelegate delegate_;
-  ModelSafeRoutingInfo routes_;
-  bool config_params_callback_invoked_;
-};
-
-TEST_F(SyncSessionJobTest, FinishCallsReadyTask) {
-  ConfigurationParams params;
-  params.ready_task = base::Bind(
-      &SyncSessionJobTest::ConfigurationParamsCallback,
-      base::Unretained(this));
-
-  sessions::SyncSourceInfo info(
-      sync_pb::GetUpdatesCallerInfo::RECONFIGURATION,
-      ModelTypeInvalidationMap());
-  scoped_ptr<SyncSession> session(
-      new SyncSession(context(), delegate(), info));
-
-  SyncSessionJob job1(
-      SyncSessionJob::CONFIGURATION, TimeTicks::Now(), info, params);
-  sessions::test_util::SimulateSuccess(session.get(),
-                                       job1.start_step(),
-                                       job1.end_step());
-  job1.Finish(false, session.get());
-  EXPECT_TRUE(config_params_callback_invoked());
-}
-
-TEST_F(SyncSessionJobTest, CoalesceSources) {
-  ModelTypeInvalidationMap one_type =
-      ModelTypeSetToInvalidationMap(
-          ParamsMeaningJustOneEnabledType(),
-          std::string());
-  ModelTypeInvalidationMap all_types =
-      ModelTypeSetToInvalidationMap(
-          ParamsMeaningAllEnabledTypes(),
-          std::string());
-  sessions::SyncSourceInfo source_one(
-      sync_pb::GetUpdatesCallerInfo::PERIODIC, one_type);
-  sessions::SyncSourceInfo source_two(
-      sync_pb::GetUpdatesCallerInfo::LOCAL, all_types);
-
-  SyncSessionJob job(SyncSessionJob::NUDGE, TimeTicks::Now(),
-                     source_one, ConfigurationParams());
-
-  job.CoalesceSources(source_two);
-
-  EXPECT_EQ(source_two.updates_source, job.source_info().updates_source);
-}
-
-}  // namespace syncer
diff --git a/sync/engine/syncer_unittest.cc b/sync/engine/syncer_unittest.cc
index 2dcc4b7..dc53f46 100644
--- a/sync/engine/syncer_unittest.cc
+++ b/sync/engine/syncer_unittest.cc
@@ -181,24 +181,15 @@ class SyncerTest : public testing::Test,
     return new SyncSession(context_.get(), this, source_info);
   }
 
-
-  void SyncShareAsDelegate(SyncSessionJob::Purpose purpose) {
-    SyncerStep start;
-    SyncerStep end;
-    SyncSessionJob::GetSyncerStepsForPurpose(purpose, &start, &end);
-
-    session_.reset(MakeSession());
-    EXPECT_TRUE(syncer_->SyncShare(session_.get(), start, end));
-  }
-
   void SyncShareNudge() {
     session_.reset(MakeSession());
-    SyncShareAsDelegate(SyncSessionJob::NUDGE);
+    EXPECT_TRUE(syncer_->SyncShare(session_.get(), SYNCER_BEGIN, SYNCER_END));
   }
 
   void SyncShareConfigure() {
     session_.reset(MakeSession());
-    SyncShareAsDelegate(SyncSessionJob::CONFIGURATION);
+    EXPECT_TRUE(
+        syncer_->SyncShare(session_.get(), DOWNLOAD_UPDATES, APPLY_UPDATES));
   }
 
   virtual void SetUp() {
diff --git a/sync/sessions/nudge_tracker.cc b/sync/sessions/nudge_tracker.cc
new file mode 100644
index 0000000..714b67d
--- /dev/null
+++ b/sync/sessions/nudge_tracker.cc
@@ -0,0 +1,50 @@
+// Copyright (c) 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "sync/sessions/nudge_tracker.h"
+#include "sync/protocol/sync.pb.h"
+
+namespace syncer {
+namespace sessions {
+
+NudgeTracker::NudgeTracker() { }
+
+NudgeTracker::~NudgeTracker() { }
+
+void NudgeTracker::CoalesceSources(const SyncSourceInfo& source) {
+  CoalesceStates(source.types, &source_info_.types);
+  source_info_.updates_source = source.updates_source;
+}
+
+bool NudgeTracker::IsEmpty() {
+  return source_info_.types.empty();
+}
+
+void NudgeTracker::Reset() {
+  source_info_ = SyncSourceInfo();
+}
+
+// TODO(rlarocque): This function often reports incorrect results.  However, it
+// is compatible with the "classic" behaviour.  We would need to make the nudge
+// tracker stop overwriting its own information (ie. fix crbug.com/231693)
+// before we could even try to report correct results.  The main issue is that
+// an notifications and local modifications nudges may overlap with each other
+// in sych a way that we lose track of which types were or were not locally
+// modified.
+ModelTypeSet NudgeTracker::GetLocallyModifiedTypes() const {
+  ModelTypeSet locally_modified;
+
+  if (source_info_.updates_source != sync_pb::GetUpdatesCallerInfo::LOCAL) {
+    return locally_modified;
+  }
+
+  for (ModelTypeInvalidationMap::const_iterator i = source_info_.types.begin();
+       i != source_info().types.end(); ++i) {
+    locally_modified.Put(i->first);
+  }
+  return locally_modified;
+}
+
+}  // namespace sessions
+}  // namespace syncer
diff --git a/sync/sessions/nudge_tracker.h b/sync/sessions/nudge_tracker.h
new file mode 100644
index 0000000..0128ce0
--- /dev/null
+++ b/sync/sessions/nudge_tracker.h
@@ -0,0 +1,54 @@
+// Copyright 2013 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.
+//
+// A class to track the outstanding work required to bring the client back into
+// sync with the server.
+#ifndef SYNC_SESSIONS_NUDGE_TRACKER_H_
+#define SYNC_SESSIONS_NUDGE_TRACKER_H_
+
+#include <vector>
+
+#include "base/compiler_specific.h"
+#include "sync/base/sync_export.h"
+#include "sync/internal_api/public/sessions/sync_source_info.h"
+
+namespace syncer {
+namespace sessions {
+
+struct SyncSourceInfo;
+
+class SYNC_EXPORT_PRIVATE NudgeTracker {
+ public:
+  NudgeTracker();
+  ~NudgeTracker();
+
+  // Merges in the information from another nudge.
+  void CoalesceSources(const SyncSourceInfo& source);
+
+  // Returns true if there are no unserviced nudges.
+  bool IsEmpty();
+
+  // Clear all unserviced nudges.
+  void Reset();
+
+  // Returns the coalesced source info.
+  const SyncSourceInfo& source_info() const {
+    return source_info_;
+  }
+
+  // Returns the set of locally modified types, according to our tracked source
+  // infos.  The result is often wrong; see implementation comment for details.
+  ModelTypeSet GetLocallyModifiedTypes() const;
+
+ private:
+  // Merged source info for the nudge(s).
+  SyncSourceInfo source_info_;
+
+  DISALLOW_COPY_AND_ASSIGN(NudgeTracker);
+};
+
+}  // namespace sessions
+}  // namespace syncer
+
+#endif  // SYNC_SESSIONS_NUDGE_TRACKER_H_
diff --git a/sync/sessions/nudge_tracker_unittest.cc b/sync/sessions/nudge_tracker_unittest.cc
new file mode 100644
index 0000000..192a737
--- /dev/null
+++ b/sync/sessions/nudge_tracker_unittest.cc
@@ -0,0 +1,112 @@
+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "sync/sessions/nudge_tracker.h"
+
+#include "sync/internal_api/public/base/model_type_invalidation_map.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace syncer {
+
+namespace {
+
+ModelTypeSet ParamsMeaningAllEnabledTypes() {
+  ModelTypeSet request_params(BOOKMARKS, AUTOFILL);
+  return request_params;
+}
+
+ModelTypeSet ParamsMeaningJustOneEnabledType() {
+  return ModelTypeSet(AUTOFILL);
+}
+
+}  // namespace
+
+namespace sessions {
+
+TEST(NudgeTrackerTest, CoalesceSources) {
+  ModelTypeInvalidationMap one_type =
+      ModelTypeSetToInvalidationMap(
+          ParamsMeaningJustOneEnabledType(),
+          std::string());
+  ModelTypeInvalidationMap all_types =
+      ModelTypeSetToInvalidationMap(
+          ParamsMeaningAllEnabledTypes(),
+          std::string());
+  sessions::SyncSourceInfo source_one(
+      sync_pb::GetUpdatesCallerInfo::NOTIFICATION, one_type);
+  sessions::SyncSourceInfo source_two(
+      sync_pb::GetUpdatesCallerInfo::LOCAL, all_types);
+
+  NudgeTracker tracker;
+  EXPECT_TRUE(tracker.IsEmpty());
+
+  tracker.CoalesceSources(source_one);
+  EXPECT_EQ(source_one.updates_source, tracker.source_info().updates_source);
+
+  tracker.CoalesceSources(source_two);
+  EXPECT_EQ(source_two.updates_source, tracker.source_info().updates_source);
+}
+
+TEST(NudgeTrackerTest, LocallyModifiedTypes_WithInvalidationFirst) {
+  ModelTypeInvalidationMap one_type =
+      ModelTypeSetToInvalidationMap(
+          ParamsMeaningJustOneEnabledType(),
+          std::string());
+  ModelTypeInvalidationMap all_types =
+      ModelTypeSetToInvalidationMap(
+          ParamsMeaningAllEnabledTypes(),
+          std::string());
+  sessions::SyncSourceInfo source_one(
+      sync_pb::GetUpdatesCallerInfo::NOTIFICATION, all_types);
+  sessions::SyncSourceInfo source_two(
+      sync_pb::GetUpdatesCallerInfo::LOCAL, one_type);
+
+  NudgeTracker tracker;
+  EXPECT_TRUE(tracker.IsEmpty());
+  EXPECT_TRUE(tracker.GetLocallyModifiedTypes().Empty());
+
+  tracker.CoalesceSources(source_one);
+  EXPECT_TRUE(tracker.GetLocallyModifiedTypes().Empty());
+
+  tracker.CoalesceSources(source_two);
+  // TODO: This result is wrong, but that's how the code has always been.  A
+  // local invalidation for a single type should mean that we have only one
+  // locally modified source.  It should not "inherit" the list of data types
+  // from the previous source.
+  EXPECT_TRUE(tracker.GetLocallyModifiedTypes().Equals(
+          ParamsMeaningAllEnabledTypes()));
+}
+
+TEST(NudgeTrackerTest, LocallyModifiedTypes_WithInvalidationSecond) {
+  ModelTypeInvalidationMap one_type =
+      ModelTypeSetToInvalidationMap(
+          ParamsMeaningJustOneEnabledType(),
+          std::string());
+  ModelTypeInvalidationMap all_types =
+      ModelTypeSetToInvalidationMap(
+          ParamsMeaningAllEnabledTypes(),
+          std::string());
+  sessions::SyncSourceInfo source_one(
+      sync_pb::GetUpdatesCallerInfo::LOCAL, one_type);
+  sessions::SyncSourceInfo source_two(
+      sync_pb::GetUpdatesCallerInfo::NOTIFICATION, all_types);
+
+  NudgeTracker tracker;
+  EXPECT_TRUE(tracker.IsEmpty());
+  EXPECT_TRUE(tracker.GetLocallyModifiedTypes().Empty());
+
+  tracker.CoalesceSources(source_one);
+  EXPECT_TRUE(tracker.GetLocallyModifiedTypes().Equals(
+          ParamsMeaningJustOneEnabledType()));
+
+  tracker.CoalesceSources(source_two);
+
+  // TODO: This result is wrong, but that's how the code has always been.
+  // The receipt of an invalidation should have no effect on the set of
+  // locally modified types.
+  EXPECT_TRUE(tracker.GetLocallyModifiedTypes().Empty());
+}
+
+}  // namespace sessions
+}  // namespace syncer
diff --git a/sync/sync_core.gypi b/sync/sync_core.gypi
index d9eb244..9c311ee 100644
--- a/sync/sync_core.gypi
+++ b/sync/sync_core.gypi
@@ -68,8 +68,6 @@
     'engine/sync_scheduler.h',
     'engine/sync_scheduler_impl.cc',
     'engine/sync_scheduler_impl.h',
-    'engine/sync_session_job.cc',
-    'engine/sync_session_job.h',
     'engine/syncer.cc',
     'engine/syncer.h',
     'engine/syncer_command.cc',
@@ -104,6 +102,8 @@
     'protocol/sync_protocol_error.cc',
     'protocol/sync_protocol_error.h',
     'sessions/debug_info_getter.h',
+    'sessions/nudge_tracker.cc',
+    'sessions/nudge_tracker.h',
     'sessions/ordered_commit_set.cc',
     'sessions/ordered_commit_set.h',
     'sessions/status_controller.cc',
diff --git a/sync/sync_tests.gypi b/sync/sync_tests.gypi
index 30e06db..f7e2b64 100644
--- a/sync/sync_tests.gypi
+++ b/sync/sync_tests.gypi
@@ -248,9 +248,7 @@
           'engine/process_commit_response_command_unittest.cc',
           'engine/process_updates_command_unittest.cc',
           'engine/store_timestamps_command_unittest.cc',
-          'engine/sync_session_job_unittest.cc',
           'engine/sync_scheduler_unittest.cc',
-          'engine/sync_scheduler_whitebox_unittest.cc',
           'engine/syncer_proto_util_unittest.cc',
           'engine/syncer_unittest.cc',
           'engine/throttled_data_type_tracker_unittest.cc',
@@ -260,6 +258,7 @@
           'js/sync_js_controller_unittest.cc',
           'protocol/proto_enum_conversions_unittest.cc',
           'protocol/proto_value_conversions_unittest.cc',
+          'sessions/nudge_tracker_unittest.cc',
           'sessions/ordered_commit_set_unittest.cc',
           'sessions/status_controller_unittest.cc',
           'sessions/sync_session_unittest.cc',