sync: Refactor job ownership in SyncScheduler

This change separates the tracking of what work needs to be done from
the decision of when to do it.  Prior to this change, SyncSessionJobs
were owned either by Closures posted to the sync thread's message loop,
or held temporarily in unscheduled_nudge_storage_, a member of the
SyncScheduler.  Following this change, there can be only two jobs in
existence, and they will be referenced only by the scoped_ptr members of
SyncScheduler named pending_nudge_job_ and pending_configure_job_.

This change, along with some previous changes to the way we schedule
tasks, makes it possible to simplify the job "saving" logic.  Jobs with
purpose == NUDGE are saved by assigning them to pending_nudge_job_ or
coalescing them with the existing pending_nudge_job_.  Jobs with purpose
== CONFIGURE are never coalesced, and can be saved in the
pending_configure_job_ member.  These changes allow us to make
SyncSessionJob::Clone() obsolete.

The logic in ScheduleNudgeImpl() has been updated to take advantage of
the fact that the pending job is much easier to find now.  It should now
be much better at coalescing its sources.  In other words, there will be
less scenarios where it can drop notification hints.  However, there are
still some cases in DecideOnJob() that may induce it to drop hints
unnecessarily.

The scheduling logic has been modified, too.  We've removed support for
the nudge while in an exponential backoff interval.  This makes it
possible to track the next wakeup time using a single timer, since the
wakeup event will be one of:
- The end of a throttled interval
- An end-of-backoff-interval retry
- A scheduled nudge
and these scenarios are now mutually exclusive.

BUG=175024

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

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

7 files changed, 198 insertions, 478 deletions

diff --git a/sync/engine/sync_scheduler_impl.cc b/sync/engine/sync_scheduler_impl.cc
index 7fd9ccd..b03b3c8 100644
--- a/sync/engine/sync_scheduler_impl.cc
+++ b/sync/engine/sync_scheduler_impl.cc
@@ -83,13 +83,10 @@ ConfigurationParams::ConfigurationParams(
 ConfigurationParams::~ConfigurationParams() {}
 
 SyncSchedulerImpl::WaitInterval::WaitInterval()
-    : mode(UNKNOWN),
-      had_nudge(false),
-      pending_configure_job(NULL) {}
+    : mode(UNKNOWN) {}
 
 SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
-    : mode(mode), had_nudge(false), length(length),
-      pending_configure_job(NULL) {}
+    : mode(mode), length(length) {}
 
 SyncSchedulerImpl::WaitInterval::~WaitInterval() {}
 
@@ -160,7 +157,6 @@ SyncSchedulerImpl::SyncSchedulerImpl(const std::string& name,
       weak_handle_this_(MakeWeakHandle(
           weak_ptr_factory_for_weak_handle_.GetWeakPtr())),
       name_(name),
-      sync_loop_(MessageLoop::current()),
       started_(false),
       syncer_short_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultShortPollIntervalSeconds)),
@@ -169,23 +165,19 @@ SyncSchedulerImpl::SyncSchedulerImpl(const std::string& name,
       sessions_commit_delay_(
           TimeDelta::FromSeconds(kDefaultSessionsCommitDelaySeconds)),
       mode_(NORMAL_MODE),
-      // Start with assuming everything is fine with the connection.
-      // At the end of the sync cycle we would have the correct status.
-      pending_nudge_(NULL),
       delay_provider_(delay_provider),
       syncer_(syncer),
       session_context_(context),
       no_scheduling_allowed_(false) {
-  DCHECK(sync_loop_);
 }
 
 SyncSchedulerImpl::~SyncSchedulerImpl() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   StopImpl(base::Closure());
 }
 
 void SyncSchedulerImpl::OnCredentialsUpdated() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
 
   if (HttpResponse::SYNC_AUTH_ERROR ==
       session_context_->connection_manager()->server_status()) {
@@ -214,14 +206,11 @@ void SyncSchedulerImpl::OnServerConnectionErrorFixed() {
   // call DoCanaryJob to achieve this, and note that nothing -- not even a
   // canary job -- can bypass a THROTTLED WaitInterval. The only thing that
   // has the authority to do that is the Unthrottle timer.
-  scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
-  if (!pending.get())
-    return;
-  DoCanaryJob(pending.Pass());
+  TryCanaryJob();
 }
 
 void SyncSchedulerImpl::Start(Mode mode) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   std::string thread_name = MessageLoop::current()->thread_name();
   if (thread_name.empty())
     thread_name = "<Main thread>";
@@ -238,25 +227,15 @@ void SyncSchedulerImpl::Start(Mode mode) {
   mode_ = mode;
   AdjustPolling(NULL);  // Will kick start poll timer if needed.
 
-  if (old_mode != mode_) {
-    // We just changed our mode. See if there are any pending jobs that we could
-    // execute in the new mode.
-    if (mode_ == NORMAL_MODE) {
-      // It is illegal to switch to NORMAL_MODE if a previous CONFIGURATION job
-      // has not yet completed.
-      DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
-    }
-
-    scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
-    if (pending.get()) {
-      SDVLOG(2) << "Executing pending job. Good luck!";
-      DoSyncSessionJob(pending.Pass(), NORMAL_PRIORITY);
-    }
+  if (old_mode != mode_ && mode_ == NORMAL_MODE && pending_nudge_job_) {
+    // 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);
   }
 }
 
 void SyncSchedulerImpl::SendInitialSnapshot() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   scoped_ptr<SyncSession> dummy(new SyncSession(
           session_context_, this, SyncSourceInfo()));
   SyncEngineEvent event(SyncEngineEvent::STATUS_CHANGED);
@@ -286,7 +265,7 @@ void BuildModelSafeParams(
 
 bool SyncSchedulerImpl::ScheduleConfiguration(
     const ConfigurationParams& params) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
   DCHECK_EQ(CONFIGURATION_MODE, mode_);
   DCHECK(!params.ready_task.is_null());
@@ -295,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(!wait_interval_.get() || !wait_interval_->pending_configure_job);
+  DCHECK(!pending_configure_job_);
 
   ModelSafeRoutingInfo restricted_routes;
   BuildModelSafeParams(params.types_to_download,
@@ -306,7 +285,7 @@ bool SyncSchedulerImpl::ScheduleConfiguration(
   // Only reconfigure if we have types to download.
   if (!params.types_to_download.Empty()) {
     DCHECK(!restricted_routes.empty());
-    scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
+    pending_configure_job_.reset(new SyncSessionJob(
         SyncSessionJob::CONFIGURATION,
         TimeTicks::Now(),
         SyncSourceInfo(params.source,
@@ -314,13 +293,15 @@ bool SyncSchedulerImpl::ScheduleConfiguration(
                            restricted_routes,
                            std::string())),
         params));
-    bool succeeded = DoSyncSessionJob(job.Pass(), NORMAL_PRIORITY);
+    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(wait_interval_.get() && wait_interval_->pending_configure_job);
+      DCHECK(pending_configure_job_);
       return false;
+    } else {
+      DCHECK(!pending_configure_job_);
     }
   } else {
     SDVLOG(2) << "No change in routing info, calling ready task directly.";
@@ -333,13 +314,12 @@ bool SyncSchedulerImpl::ScheduleConfiguration(
 SyncSchedulerImpl::JobProcessDecision
 SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob& job,
                                              JobPriority priority) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   DCHECK(wait_interval_.get());
   DCHECK_NE(job.purpose(), SyncSessionJob::POLL);
 
   SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
             << WaitInterval::GetModeString(wait_interval_->mode)
-            << (wait_interval_->had_nudge ? " (had nudge)" : "")
             << ((priority == CANARY_PRIORITY) ? " (canary)" : "");
 
   // If we save a job while in a WaitInterval, there is a well-defined moment
@@ -358,11 +338,9 @@ SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob& job,
     if (mode_ == CONFIGURATION_MODE)
       return SAVE;
 
-    // If we already had one nudge then just drop this nudge. We will retry
-    // later when the timer runs out.
     if (priority == NORMAL_PRIORITY)
-      return wait_interval_->had_nudge ? DROP : CONTINUE;
-    else // We are here because timer ran out. So retry.
+      return DROP;
+    else // Either backoff has ended, or we have permission to bypass it.
       return CONTINUE;
   }
   return (priority == CANARY_PRIORITY) ? CONTINUE : SAVE;
@@ -371,7 +349,7 @@ SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob& job,
 SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
     const SyncSessionJob& job,
     JobPriority priority) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
 
   // POLL jobs do not call this function.
   DCHECK(job.purpose() == SyncSessionJob::NUDGE ||
@@ -449,69 +427,11 @@ SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
   return job.purpose() == SyncSessionJob::NUDGE ? SAVE : DROP;
 }
 
-void SyncSchedulerImpl::HandleSaveJobDecision(scoped_ptr<SyncSessionJob> job) {
-  const bool is_nudge = job->purpose() == SyncSessionJob::NUDGE;
-  if (is_nudge && pending_nudge_) {
-    SDVLOG(2) << "Coalescing a pending nudge";
-    // TODO(tim): This basically means we never use the more-careful coalescing
-    // logic in ScheduleNudgeImpl that takes the min of the two nudge start
-    // times, because we're calling this function first.  Pull this out
-    // into a function to coalesce + set start times and reuse.
-    pending_nudge_->CoalesceSources(job->source_info());
-    return;
-  }
-
-  scoped_ptr<SyncSessionJob> job_to_save = job->Clone();
-  if (wait_interval_.get() && !wait_interval_->pending_configure_job) {
-    // This job should be made the new canary.
-    if (is_nudge) {
-      pending_nudge_ = job_to_save.get();
-    } else {
-      SDVLOG(2) << "Saving a configuration job";
-      DCHECK_EQ(job->purpose(), SyncSessionJob::CONFIGURATION);
-      DCHECK(!wait_interval_->pending_configure_job);
-      DCHECK_EQ(mode_, CONFIGURATION_MODE);
-      DCHECK(!job->config_params().ready_task.is_null());
-      // The only nudge that could exist is a scheduled canary nudge.
-      DCHECK(!unscheduled_nudge_storage_.get());
-      if (pending_nudge_) {
-        // Pre-empt the nudge canary and abandon the old nudge (owned by task).
-        unscheduled_nudge_storage_ = pending_nudge_->Clone();
-        pending_nudge_ = unscheduled_nudge_storage_.get();
-      }
-      wait_interval_->pending_configure_job = job_to_save.get();
-    }
-    TimeDelta length =
-        wait_interval_->timer.desired_run_time() - TimeTicks::Now();
-    wait_interval_->length = length < TimeDelta::FromSeconds(0) ?
-        TimeDelta::FromSeconds(0) : length;
-    RestartWaiting(job_to_save.Pass());
-    return;
-  }
-
-  // Note that today there are no cases where we SAVE a CONFIGURATION job
-  // when we're not in a WaitInterval. See bug 147736.
-  DCHECK(is_nudge);
-  // There may or may not be a pending_configure_job. Either way this nudge
-  // is unschedulable.
-  pending_nudge_ = job_to_save.get();
-  unscheduled_nudge_storage_ = job_to_save.Pass();
-}
-
-// Functor for std::find_if to search by ModelSafeGroup.
-struct ModelSafeWorkerGroupIs {
-  explicit ModelSafeWorkerGroupIs(ModelSafeGroup group) : group(group) {}
-  bool operator()(ModelSafeWorker* w) {
-    return group == w->GetModelSafeGroup();
-  }
-  ModelSafeGroup group;
-};
-
 void SyncSchedulerImpl::ScheduleNudgeAsync(
     const TimeDelta& desired_delay,
     NudgeSource source, ModelTypeSet types,
     const tracked_objects::Location& nudge_location) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   SDVLOG_LOC(nudge_location, 2)
       << "Nudge scheduled with delay "
       << desired_delay.InMilliseconds() << " ms, "
@@ -530,7 +450,7 @@ void SyncSchedulerImpl::ScheduleNudgeWithStatesAsync(
     const TimeDelta& desired_delay,
     NudgeSource source, const ModelTypeInvalidationMap& invalidation_map,
     const tracked_objects::Location& nudge_location) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   SDVLOG_LOC(nudge_location, 2)
       << "Nudge scheduled with delay "
       << desired_delay.InMilliseconds() << " ms, "
@@ -552,7 +472,7 @@ void SyncSchedulerImpl::ScheduleNudgeImpl(
     GetUpdatesCallerInfo::GetUpdatesSource source,
     const ModelTypeInvalidationMap& invalidation_map,
     const tracked_objects::Location& nudge_location) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   DCHECK(!invalidation_map.empty()) << "Nudge scheduled for no types!";
 
   if (no_scheduling_allowed_) {
@@ -586,51 +506,43 @@ void SyncSchedulerImpl::ScheduleNudgeImpl(
             << SyncSessionJob::GetPurposeString(job->purpose())
             << " in mode " << GetModeString(mode_)
             << ": " << GetDecisionString(decision);
-  if (decision != CONTINUE) {
-    // End of the line, though we may save the job for later.
-    if (decision == SAVE) {
-      HandleSaveJobDecision(job.Pass());
-    } else {
-      DCHECK_EQ(decision, DROP);
-    }
+  if (decision == DROP) {
     return;
   }
 
-  if (pending_nudge_) {
-    SDVLOG(2) << "Rescheduling pending nudge";
-    pending_nudge_->CoalesceSources(job->source_info());
-    // Choose the start time as the earliest of the 2. Note that this means
-    // if a nudge arrives with delay (e.g. kDefaultSessionsCommitDelaySeconds)
-    // but a nudge is already scheduled to go out, we'll send the (tab) commit
-    // without waiting.
-    pending_nudge_->set_scheduled_start(
-        std::min(job->scheduled_start(), pending_nudge_->scheduled_start()));
-    // Abandon the old task by cloning and replacing the session.
-    // It's possible that by "rescheduling" we're actually taking a job that
-    // was previously unscheduled and giving it wings, so take care to reset
-    // unscheduled nudge storage.
-    job = pending_nudge_->Clone();
-    pending_nudge_ = NULL;
-    unscheduled_nudge_storage_.reset();
-    // It's also possible we took a canary job, since we allow one nudge
-    // per backoff interval.
-    DCHECK(!wait_interval_ || !wait_interval_->had_nudge);
+  // 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 (decision == SAVE) {
+    return;
   }
 
-  TimeDelta run_delay = job->scheduled_start() - TimeTicks::Now();
+  TimeDelta run_delay =
+      pending_nudge_job_->scheduled_start() - TimeTicks::Now();
   if (run_delay < TimeDelta::FromMilliseconds(0))
     run_delay = TimeDelta::FromMilliseconds(0);
   SDVLOG_LOC(nudge_location, 2)
       << "Scheduling a nudge with "
       << run_delay.InMilliseconds() << " ms delay";
 
-  pending_nudge_ = job.get();
-  PostDelayedTask(nudge_location, "DoSyncSessionJob",
-      base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoSyncSessionJob),
-                                    weak_ptr_factory_.GetWeakPtr(),
-                                    base::Passed(&job),
-                                    NORMAL_PRIORITY),
-      run_delay);
+  if (started_) {
+    pending_wakeup_timer_.Start(
+        nudge_location,
+        run_delay,
+        base::Bind(&SyncSchedulerImpl::DoNudgeSyncSessionJob,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   NORMAL_PRIORITY));
+  }
 }
 
 const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
@@ -651,27 +563,9 @@ const char* SyncSchedulerImpl::GetDecisionString(
   return "";
 }
 
-void SyncSchedulerImpl::PostDelayedTask(
-    const tracked_objects::Location& from_here,
-    const char* name, const base::Closure& task, base::TimeDelta delay) {
-  SDVLOG_LOC(from_here, 3) << "Posting " << name << " task with "
-                           << delay.InMilliseconds() << " ms delay";
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  if (!started_) {
-    SDVLOG(1) << "Not posting task as scheduler is stopped.";
-    return;
-  }
-  // This cancels the previous task, if one existed.
-  pending_wakeup_.Reset(task);
-  sync_loop_->PostDelayedTask(from_here, pending_wakeup_.callback(), delay);
-}
-
-bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job,
-                                         JobPriority priority) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  if (job->purpose() == SyncSessionJob::NUDGE) {
-    pending_nudge_ = NULL;
-  }
+bool SyncSchedulerImpl::DoSyncSessionJobImpl(scoped_ptr<SyncSessionJob> job,
+                                             JobPriority priority) {
+  DCHECK(CalledOnValidThread());
 
   base::AutoReset<bool> protector(&no_scheduling_allowed_, true);
   JobProcessDecision decision = DecideOnJob(*job, priority);
@@ -682,7 +576,11 @@ bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job,
             << ": " << GetDecisionString(decision);
   if (decision != CONTINUE) {
     if (decision == SAVE) {
-      HandleSaveJobDecision(job.Pass());
+      if (job->purpose() == SyncSessionJob::CONFIGURATION) {
+        pending_configure_job_ = job.Pass();
+      } else {
+        pending_nudge_job_ = job.Pass();
+      }
     } else {
       DCHECK_EQ(decision, DROP);
     }
@@ -707,15 +605,14 @@ bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job,
     // 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).
-    scoped_ptr<SyncSessionJob> clone = job->Clone();
-    if (clone->purpose() == SyncSessionJob::NUDGE)
-      pending_nudge_ = clone.get();
-    else if (clone->purpose() == SyncSessionJob::CONFIGURATION)
-      wait_interval_->pending_configure_job = clone.get();
+    if (job->purpose() == SyncSessionJob::NUDGE)
+      pending_nudge_job_ = job.Pass();
+    else if (job->purpose() == SyncSessionJob::CONFIGURATION)
+      pending_configure_job_ = job.Pass();
     else
       NOTREACHED();
 
-    RestartWaiting(clone.Pass());
+    RestartWaiting();
     return success;
   }
 
@@ -725,6 +622,14 @@ bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job,
   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);
+}
+
 bool SyncSchedulerImpl::ShouldPoll() {
   if (wait_interval_.get()) {
     SDVLOG(2) << "Not running poll in wait interval.";
@@ -747,8 +652,15 @@ bool SyncSchedulerImpl::ShouldPoll() {
   return true;
 }
 
-void SyncSchedulerImpl::DoPollSyncSessionJob(scoped_ptr<SyncSessionJob> job) {
-  DCHECK_EQ(job->purpose(), SyncSessionJob::POLL);
+void SyncSchedulerImpl::DoPollSyncSessionJob() {
+  ModelSafeRoutingInfo r;
+  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);
 
@@ -766,17 +678,16 @@ void SyncSchedulerImpl::DoPollSyncSessionJob(scoped_ptr<SyncSessionJob> job) {
   FinishSyncSessionJob(job.get(), premature_exit, &session);
 
   if (IsSyncingCurrentlySilenced()) {
-    // This will start the countdown to unthrottle.  Other kinds of jobs would
-    // schedule themselves as the post-unthrottle canary.  A poll job is not
-    // that urgent, so it does not get to be the canary.  We still need to start
-    // the timer regardless.  Otherwise there could be no one to clear the
-    // WaitInterval when the throttling expires.
-    RestartWaiting(scoped_ptr<SyncSessionJob>());
+    // 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.
+    RestartWaiting();
   }
 }
 
 void SyncSchedulerImpl::UpdateNudgeTimeRecords(const SyncSourceInfo& info) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
 
   // We are interested in recording time between local nudges for datatypes.
   // TODO(tim): Consider tracking LOCAL_NOTIFICATION as well.
@@ -803,7 +714,7 @@ void SyncSchedulerImpl::UpdateNudgeTimeRecords(const SyncSourceInfo& info) {
 bool SyncSchedulerImpl::FinishSyncSessionJob(SyncSessionJob* job,
                                              bool exited_prematurely,
                                              SyncSession* session) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
 
   // Let job know that we're through syncing (calling SyncShare) at this point.
   bool succeeded = false;
@@ -831,7 +742,7 @@ bool SyncSchedulerImpl::FinishSyncSessionJob(SyncSessionJob* job,
 void SyncSchedulerImpl::ScheduleNextSync(
     scoped_ptr<SyncSessionJob> finished_job,
     SyncSession* session) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   DCHECK(finished_job->purpose() == SyncSessionJob::CONFIGURATION
          || finished_job->purpose() == SyncSessionJob::NUDGE);
 
@@ -841,34 +752,12 @@ void SyncSchedulerImpl::ScheduleNextSync(
   // we should be able to detect such errors and only retry when we detect
   // transient errors.
 
-  if (IsBackingOff() && wait_interval_->timer.IsRunning() &&
-      mode_ == NORMAL_MODE) {
-    // When in normal mode, we allow up to one nudge per backoff interval.  It
-    // appears that this was our nudge for this interval, and it failed.
-    //
-    // Note: This does not prevent us from running canary jobs.  For example,
-    // an IP address change might still result in another nudge being executed
-    // during this backoff interval.
-    SDVLOG(2) << "A nudge during backoff failed, creating new pending nudge.";
-    DCHECK_EQ(SyncSessionJob::NUDGE, finished_job->purpose());
-    DCHECK(!wait_interval_->had_nudge);
-
-    wait_interval_->had_nudge = true;
-    DCHECK(!pending_nudge_);
-
-    scoped_ptr<SyncSessionJob> new_job = finished_job->Clone();
-    pending_nudge_ = new_job.get();
-    RestartWaiting(new_job.Pass());
-  } else {
-    // Either this is the first failure or a consecutive failure after our
-    // backoff timer expired.  We handle it the same way in either case.
-    SDVLOG(2) << "Non-'backoff nudge' SyncShare job failed";
-    HandleContinuationError(finished_job.Pass(), session);
-  }
+  SDVLOG(2) << "SyncShare job failed; will start or update backoff";
+  HandleContinuationError(finished_job.Pass(), session);
 }
 
 void SyncSchedulerImpl::AdjustPolling(const SyncSessionJob* old_job) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
 
   TimeDelta poll  = (!session_context_->notifications_enabled()) ?
       syncer_short_poll_interval_seconds_ :
@@ -888,28 +777,28 @@ void SyncSchedulerImpl::AdjustPolling(const SyncSessionJob* old_job) {
                     &SyncSchedulerImpl::PollTimerCallback);
 }
 
-void SyncSchedulerImpl::RestartWaiting(scoped_ptr<SyncSessionJob> job) {
+void SyncSchedulerImpl::RestartWaiting() {
   CHECK(wait_interval_.get());
-  wait_interval_->timer.Stop();
   DCHECK(wait_interval_->length >= TimeDelta::FromSeconds(0));
   if (wait_interval_->mode == WaitInterval::THROTTLED) {
-    pending_wakeup_.Reset(base::Bind(&SyncSchedulerImpl::Unthrottle,
-                                     weak_ptr_factory_.GetWeakPtr(),
-                                     base::Passed(&job)));
-
+    pending_wakeup_timer_.Start(
+        FROM_HERE,
+        wait_interval_->length,
+        base::Bind(&SyncSchedulerImpl::Unthrottle,
+                   weak_ptr_factory_.GetWeakPtr()));
   } else {
-    pending_wakeup_.Reset(base::Bind(&SyncSchedulerImpl::DoCanaryJob,
-                                     weak_ptr_factory_.GetWeakPtr(),
-                                     base::Passed(&job)));
+    pending_wakeup_timer_.Start(
+        FROM_HERE,
+        wait_interval_->length,
+        base::Bind(&SyncSchedulerImpl::TryCanaryJob,
+                   weak_ptr_factory_.GetWeakPtr()));
   }
-  wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
-                              pending_wakeup_.callback());
 }
 
 void SyncSchedulerImpl::HandleContinuationError(
     scoped_ptr<SyncSessionJob> old_job,
     SyncSession* session) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
 
   TimeDelta length = delay_provider_->GetDelay(
       IsBackingOff() ? wait_interval_->length :
@@ -921,26 +810,24 @@ void SyncSchedulerImpl::HandleContinuationError(
             << " job. The time delta(ms) is "
             << length.InMilliseconds();
 
-  // This will reset the had_nudge variable as well.
   wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
                                         length));
   NotifyRetryTime(base::Time::Now() + length);
-  scoped_ptr<SyncSessionJob> new_job(old_job->Clone());
-  new_job->set_scheduled_start(TimeTicks::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());
-    wait_interval_->pending_configure_job = new_job.get();
+    DCHECK(!pending_configure_job_);
+    pending_configure_job_ = old_job.Pass();
   } else {
-    // We are not in configuration mode. So wait_interval's pending job
-    // should be null.
-    DCHECK(wait_interval_->pending_configure_job == NULL);
-    DCHECK(!pending_nudge_);
-    pending_nudge_ = new_job.get();
+    // 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(new_job.Pass());
+  RestartWaiting();
 }
 
 void SyncSchedulerImpl::RequestStop(const base::Closure& callback) {
@@ -953,7 +840,7 @@ void SyncSchedulerImpl::RequestStop(const base::Closure& callback) {
 }
 
 void SyncSchedulerImpl::StopImpl(const base::Closure& callback) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   SDVLOG(2) << "StopImpl called";
 
   // Kill any in-flight method calls.
@@ -961,9 +848,9 @@ void SyncSchedulerImpl::StopImpl(const base::Closure& callback) {
   wait_interval_.reset();
   NotifyRetryTime(base::Time());
   poll_timer_.Stop();
-  pending_nudge_ = NULL;
-  unscheduled_nudge_storage_.reset();
-  pending_wakeup_.Cancel();
+  pending_wakeup_timer_.Stop();
+  pending_nudge_job_.reset();
+  pending_configure_job_.reset();
   if (started_) {
     started_ = false;
   }
@@ -971,48 +858,24 @@ void SyncSchedulerImpl::StopImpl(const base::Closure& callback) {
     callback.Run();
 }
 
-void SyncSchedulerImpl::DoCanaryJob(scoped_ptr<SyncSessionJob> to_be_canary) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SDVLOG(2) << "Do canary job";
-
-  // This is the only place where we invoke DoSyncSessionJob with canary
-  // privileges.  Everyone else should use NORMAL_PRIORITY.
-  DoSyncSessionJob(to_be_canary.Pass(), CANARY_PRIORITY);
-}
+// This is the only place where we invoke DoSyncSessionJob with canary
+// privileges.  Everyone else should use NORMAL_PRIORITY.
+void SyncSchedulerImpl::TryCanaryJob() {
+  DCHECK(CalledOnValidThread());
 
-scoped_ptr<SyncSessionJob> SyncSchedulerImpl::TakePendingJobForCurrentMode() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  // If we find a scheduled pending_ job, abandon the old one and return a
-  // a clone.  If unscheduled, just hand over ownership.
-  scoped_ptr<SyncSessionJob> candidate;
-  if (mode_ == CONFIGURATION_MODE && wait_interval_.get()
-      && wait_interval_->pending_configure_job) {
-    SDVLOG(2) << "Found pending configure job";
-    candidate =
-        wait_interval_->pending_configure_job->Clone().Pass();
-    wait_interval_->pending_configure_job = candidate.get();
-  } else if (mode_ == NORMAL_MODE && pending_nudge_) {
-    SDVLOG(2) << "Found pending nudge job";
-    candidate = pending_nudge_->Clone();
-    pending_nudge_ = candidate.get();
-    unscheduled_nudge_storage_.reset();
+  if (mode_ == CONFIGURATION_MODE && pending_configure_job_) {
+    SDVLOG(2) << "Found pending configure job; will run as canary";
+    DoConfigurationSyncSessionJob(CANARY_PRIORITY);
+  } else if (mode_ == NORMAL_MODE && pending_nudge_job_) {
+    SDVLOG(2) << "Found pending nudge job; will run as canary";
+    DoNudgeSyncSessionJob(CANARY_PRIORITY);
+  } else {
+    SDVLOG(2) << "Found no work to do; will not run a canary";
   }
-  // If we took a job and there's a wait interval, we took the pending canary.
-  if (candidate && wait_interval_)
-    wait_interval_->timer.Stop();
-  return candidate.Pass();
 }
 
 void SyncSchedulerImpl::PollTimerCallback() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  ModelSafeRoutingInfo r;
-  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()));
+  DCHECK(CalledOnValidThread());
   if (no_scheduling_allowed_) {
     // The no_scheduling_allowed_ flag is set by a function-scoped AutoReset in
     // functions that are called only on the sync thread.  This function is also
@@ -1024,16 +887,12 @@ void SyncSchedulerImpl::PollTimerCallback() {
     return;
   }
 
-  DoPollSyncSessionJob(job.Pass());
+  DoPollSyncSessionJob();
 }
 
-void SyncSchedulerImpl::Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+void SyncSchedulerImpl::Unthrottle() {
+  DCHECK(CalledOnValidThread());
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
-  DCHECK(!to_be_canary.get() || pending_nudge_ == to_be_canary.get() ||
-         wait_interval_->pending_configure_job == to_be_canary.get());
-  SDVLOG(2) << "Unthrottled " << (to_be_canary.get() ? "with " : "without ")
-           << "canary.";
 
   // We're no longer throttled, so clear the wait interval.
   wait_interval_.reset();
@@ -1044,15 +903,11 @@ void SyncSchedulerImpl::Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary) {
   // was just created (e.g via ScheduleNudgeImpl). The main implication is
   // that we're careful to update routing info (etc) with such potentially
   // stale canary jobs.
-  if (to_be_canary.get()) {
-    DoCanaryJob(to_be_canary.Pass());
-  } else {
-    DCHECK(!unscheduled_nudge_storage_.get());
-  }
+  TryCanaryJob();
 }
 
 void SyncSchedulerImpl::Notify(SyncEngineEvent::EventCause cause) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   session_context_->NotifyListeners(SyncEngineEvent(cause));
 }
 
@@ -1063,45 +918,45 @@ void SyncSchedulerImpl::NotifyRetryTime(base::Time retry_time) {
 }
 
 bool SyncSchedulerImpl::IsBackingOff() const {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   return wait_interval_.get() && wait_interval_->mode ==
       WaitInterval::EXPONENTIAL_BACKOFF;
 }
 
 void SyncSchedulerImpl::OnSilencedUntil(
     const base::TimeTicks& silenced_until) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   wait_interval_.reset(new WaitInterval(WaitInterval::THROTTLED,
                                         silenced_until - TimeTicks::Now()));
   NotifyRetryTime(base::Time::Now() + wait_interval_->length);
 }
 
 bool SyncSchedulerImpl::IsSyncingCurrentlySilenced() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   return wait_interval_.get() && wait_interval_->mode ==
       WaitInterval::THROTTLED;
 }
 
 void SyncSchedulerImpl::OnReceivedShortPollIntervalUpdate(
     const base::TimeDelta& new_interval) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   syncer_short_poll_interval_seconds_ = new_interval;
 }
 
 void SyncSchedulerImpl::OnReceivedLongPollIntervalUpdate(
     const base::TimeDelta& new_interval) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   syncer_long_poll_interval_seconds_ = new_interval;
 }
 
 void SyncSchedulerImpl::OnReceivedSessionsCommitDelay(
     const base::TimeDelta& new_delay) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   sessions_commit_delay_ = new_delay;
 }
 
 void SyncSchedulerImpl::OnShouldStopSyncingPermanently() {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   SDVLOG(2) << "OnShouldStopSyncingPermanently";
   syncer_->RequestEarlyExit();  // Thread-safe.
   Notify(SyncEngineEvent::STOP_SYNCING_PERMANENTLY);
@@ -1109,7 +964,7 @@ void SyncSchedulerImpl::OnShouldStopSyncingPermanently() {
 
 void SyncSchedulerImpl::OnActionableError(
     const sessions::SyncSessionSnapshot& snap) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   SDVLOG(2) << "OnActionableError";
   SyncEngineEvent event(SyncEngineEvent::ACTIONABLE_ERROR);
   event.snapshot = snap;
@@ -1118,7 +973,7 @@ void SyncSchedulerImpl::OnActionableError(
 
 void SyncSchedulerImpl::OnSyncProtocolError(
     const sessions::SyncSessionSnapshot& snapshot) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   if (ShouldRequestEarlyExit(
           snapshot.model_neutral_state().sync_protocol_error)) {
     SDVLOG(2) << "Sync Scheduler requesting early exit.";
@@ -1129,12 +984,12 @@ void SyncSchedulerImpl::OnSyncProtocolError(
 }
 
 void SyncSchedulerImpl::SetNotificationsEnabled(bool notifications_enabled) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   session_context_->set_notifications_enabled(notifications_enabled);
 }
 
 base::TimeDelta SyncSchedulerImpl::GetSessionsCommitDelay() const {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  DCHECK(CalledOnValidThread());
   return sessions_commit_delay_;
 }
 
diff --git a/sync/engine/sync_scheduler_impl.h b/sync/engine/sync_scheduler_impl.h
index f3c4d5a..dfdcc80 100644
--- a/sync/engine/sync_scheduler_impl.h
+++ b/sync/engine/sync_scheduler_impl.h
@@ -15,7 +15,7 @@
 #include "base/memory/linked_ptr.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
-#include "base/observer_list.h"
+#include "base/threading/non_thread_safe.h"
 #include "base/time.h"
 #include "base/timer.h"
 #include "sync/base/sync_export.h"
@@ -34,7 +34,9 @@ namespace syncer {
 
 class BackoffDelayProvider;
 
-class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
+class SYNC_EXPORT_PRIVATE SyncSchedulerImpl
+    : public SyncScheduler,
+      public base::NonThreadSafe {
  public:
   // |name| is a display string to identify the syncer thread.  Takes
   // |ownership of |syncer| and |delay_provider|.
@@ -107,8 +109,6 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest,
                            SaveNudgeWhileTypeThrottled);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest, ContinueNudge);
-  FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest, DropPoll);
-  FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest, ContinuePoll);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest, ContinueConfiguration);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest,
                            SaveConfigurationWhileThrottled);
@@ -116,10 +116,7 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
                            SaveNudgeWhileThrottled);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest,
                            ContinueCanaryJobConfig);
-  FRIEND_TEST_ALL_PREFIXES(SyncSchedulerWhiteboxTest,
-      ContinueNudgeWhileExponentialBackOff);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerTest, TransientPollFailure);
-  FRIEND_TEST_ALL_PREFIXES(SyncSchedulerTest, GetInitialBackoffDelay);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerTest,
                            ServerConnectionChangeDuringBackoff);
   FRIEND_TEST_ALL_PREFIXES(SyncSchedulerTest,
@@ -129,9 +126,8 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
     enum Mode {
       // Uninitialized state, should not be set in practice.
       UNKNOWN = -1,
-      // A wait interval whose duration has been affected by exponential
-      // backoff.
-      // EXPONENTIAL_BACKOFF intervals are nudge-rate limited to 1 per interval.
+      // We enter a series of increasingly longer WaitIntervals if we experience
+      // repeated transient failures.  We retry at the end of each interval.
       EXPONENTIAL_BACKOFF,
       // A server-initiated throttled interval.  We do not allow any syncing
       // during such an interval.
@@ -144,38 +140,28 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
     static const char* GetModeString(Mode mode);
 
     Mode mode;
-
-    // This bool is set to true if we have observed a nudge during this
-    // interval and mode == EXPONENTIAL_BACKOFF.
-    bool had_nudge;
     base::TimeDelta length;
-    base::OneShotTimer<SyncSchedulerImpl> timer;
-
-    // Configure jobs are saved only when backing off or throttling. So we
-    // expose the pointer here (does not own, similar to pending_nudge).
-    SyncSessionJob* pending_configure_job;
   };
 
   static const char* GetModeString(Mode mode);
 
   static const char* GetDecisionString(JobProcessDecision decision);
 
-  // Helper to cancel any existing delayed task and replace it with a new one.
-  // It will not post any tasks if the scheduler is in a "stopped" state.
-  void PostDelayedTask(const tracked_objects::Location& from_here,
-                       const char* name,
-                       const base::Closure& task,
-                       base::TimeDelta delay);
+  // Invoke the syncer to perform a non-POLL job.
+  bool DoSyncSessionJobImpl(scoped_ptr<SyncSessionJob> job,
+                            JobPriority priority);
 
-  // Invoke the Syncer to perform a non-poll job.
-  bool DoSyncSessionJob(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();
 
   // Invoke the Syncer to perform a poll job.
-  void DoPollSyncSessionJob(scoped_ptr<SyncSessionJob> 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
@@ -193,7 +179,7 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
   void AdjustPolling(const SyncSessionJob* old_job);
 
   // Helper to restart waiting with |wait_interval_|'s timer.
-  void RestartWaiting(scoped_ptr<SyncSessionJob> job);
+  void RestartWaiting();
 
   // Helper to ScheduleNextSync in case of consecutive sync errors.
   void HandleContinuationError(scoped_ptr<SyncSessionJob> old_job,
@@ -203,10 +189,6 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
   JobProcessDecision DecideOnJob(const SyncSessionJob& job,
                                  JobPriority priority);
 
-  // If DecideOnJob decides that |job| should be SAVEd, this function will
-  // carry out the task of actually "saving" (or coalescing) the job.
-  void HandleSaveJobDecision(scoped_ptr<SyncSessionJob> job);
-
   // Decide on whether to CONTINUE, SAVE or DROP the job when we are in
   // backoff mode.
   JobProcessDecision DecideWhileInWaitInterval(const SyncSessionJob& job,
@@ -235,22 +217,12 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
   // Helper to signal listeners about changed retry time
   void NotifyRetryTime(base::Time retry_time);
 
-  // Callback to change backoff state. |to_be_canary| in both cases is the job
-  // that should be granted canary privileges. Note: it is possible that the
-  // job that gets scheduled when this callback is scheduled is different from
-  // the job that will actually get executed, because other jobs may have been
-  // scheduled while we were waiting for the callback.
-  void DoCanaryJob(scoped_ptr<SyncSessionJob> to_be_canary);
-  void Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary);
-
-  // Returns a pending job that has potential to run given the state of the
-  // scheduler, if it exists. Useful whenever an event occurs that may
-  // change conditions that permit a job to run, such as re-establishing
-  // network connection, auth refresh, mode changes etc. Note that the returned
-  // job may have been scheduled to run at a later time, or may have been
-  // unscheduled.  In the former case, this will result in abandoning the old
-  // job and effectively cancelling it.
-  scoped_ptr<SyncSessionJob> TakePendingJobForCurrentMode();
+  // Looks for pending work and, if it finds any, run this work at "canary"
+  // priority.
+  void TryCanaryJob();
+
+  // Transitions out of the THROTTLED WaitInterval then calls TryCanaryJob().
+  void Unthrottle();
 
   // Called when the root cause of the current connection error is fixed.
   void OnServerConnectionErrorFixed();
@@ -283,10 +255,6 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
   // Used for logging.
   const std::string name_;
 
-  // The message loop this object is on.  Almost all methods have to
-  // be called on this thread.
-  base::MessageLoop* const sync_loop_;
-
   // Set in Start(), unset in Stop().
   bool started_;
 
@@ -304,27 +272,21 @@ class SYNC_EXPORT_PRIVATE SyncSchedulerImpl : public SyncScheduler {
   // The mode of operation.
   Mode mode_;
 
-  // Tracks (does not own) in-flight nudges (scheduled or unscheduled),
-  // so we can coalesce. NULL if there is no pending nudge.
-  SyncSessionJob* pending_nudge_;
-
-  // There are certain situations where we want to remember a nudge, but
-  // there is no well defined moment in time in the future when that nudge
-  // should run, e.g. if it requires a mode switch or updated auth credentials.
-  // This member will own NUDGE jobs in those cases, until an external event
-  // (mode switch or fixed auth) occurs to trigger a retry.  Should be treated
-  // as opaque / not interacted with (i.e. we could build a wrapper to
-  // hide the type, but that's probably overkill).
-  scoped_ptr<SyncSessionJob> unscheduled_nudge_storage_;
-
   // Current wait state.  Null if we're not in backoff and not throttled.
   scoped_ptr<WaitInterval> wait_interval_;
 
   scoped_ptr<BackoffDelayProvider> delay_provider_;
 
-  // We allow at most one PostedTask to be pending at one time.  This is it.
-  // We will cancel this task before starting a new one.
-  base::CancelableClosure pending_wakeup_;
+  // 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_;
+
+  // 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_;
 
   // 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 406be4e..444f5eb 100644
--- a/sync/engine/sync_scheduler_unittest.cc
+++ b/sync/engine/sync_scheduler_unittest.cc
@@ -72,6 +72,14 @@ void PumpLoop() {
   RunLoop();
 }
 
+void PumpLoopFor(base::TimeDelta time) {
+  // Allow the loop to run for the specified amount of time.
+  MessageLoop::current()->PostDelayedTask(FROM_HERE,
+                                          base::Bind(&QuitLoopNow),
+                                          time);
+  RunLoop();
+}
+
 ModelSafeRoutingInfo TypesToRoutingInfo(ModelTypeSet types) {
   ModelSafeRoutingInfo routes;
   for (ModelTypeSet::Iterator iter = types.First(); iter.Good(); iter.Inc()) {
@@ -975,21 +983,20 @@ TEST_F(SyncSchedulerTest, BackoffDropsJobs) {
   EXPECT_EQ(GetUpdatesCallerInfo::LOCAL,
             r.snapshots[0].source().updates_source);
 
-  EXPECT_CALL(*syncer(), SyncShare(_,_,_)).Times(1)
-      .WillOnce(DoAll(Invoke(sessions::test_util::SimulateCommitFailed),
-                      RecordSyncShare(&r)));
+  // Wait a while (10x poll interval) so a few poll jobs will be attempted.
+  PumpLoopFor(poll * 10);
 
-  // We schedule a nudge with enough delay (10X poll interval) that at least
-  // one or two polls would have taken place.  The nudge should succeed.
+  // Try (and fail) to schedule a nudge.
   scheduler()->ScheduleNudgeAsync(
-      poll * 10, NUDGE_SOURCE_LOCAL, types, FROM_HERE);
-  RunLoop();
+      base::TimeDelta::FromMilliseconds(1),
+      NUDGE_SOURCE_LOCAL,
+      types,
+      FROM_HERE);
 
   Mock::VerifyAndClearExpectations(syncer());
   Mock::VerifyAndClearExpectations(delay());
-  ASSERT_EQ(2U, r.snapshots.size());
-  EXPECT_EQ(GetUpdatesCallerInfo::LOCAL,
-            r.snapshots[1].source().updates_source);
+
+  ASSERT_EQ(1U, r.snapshots.size());
 
   EXPECT_CALL(*delay(), GetDelay(_)).Times(0);
 
diff --git a/sync/engine/sync_scheduler_whitebox_unittest.cc b/sync/engine/sync_scheduler_whitebox_unittest.cc
index 9aceada..7a550fa 100644
--- a/sync/engine/sync_scheduler_whitebox_unittest.cc
+++ b/sync/engine/sync_scheduler_whitebox_unittest.cc
@@ -88,10 +88,6 @@ class SyncSchedulerWhiteboxTest : public testing::Test {
        TimeDelta::FromSeconds(1)));
   }
 
-  void SetWaitIntervalHadNudge(bool had_nudge) {
-    scheduler_->wait_interval_->had_nudge = had_nudge;
-  }
-
   SyncSchedulerImpl::JobProcessDecision DecideOnJob(
       const SyncSessionJob& job,
       SyncSchedulerImpl::JobPriority priority) {
@@ -233,22 +229,10 @@ TEST_F(SyncSchedulerWhiteboxTest, SaveNudgeWhileThrottled) {
   EXPECT_EQ(decision, SyncSchedulerImpl::SAVE);
 }
 
-TEST_F(SyncSchedulerWhiteboxTest, ContinueNudgeWhileExponentialBackOff) {
-  InitializeSyncerOnNormalMode();
-  SetMode(SyncScheduler::NORMAL_MODE);
-  SetWaitIntervalToExponentialBackoff();
-
-  SyncSchedulerImpl::JobProcessDecision decision = CreateAndDecideJob(
-      SyncSessionJob::NUDGE);
-
-  EXPECT_EQ(decision, SyncSchedulerImpl::CONTINUE);
-}
-
 TEST_F(SyncSchedulerWhiteboxTest, DropNudgeWhileExponentialBackOff) {
   InitializeSyncerOnNormalMode();
   SetMode(SyncScheduler::NORMAL_MODE);
   SetWaitIntervalToExponentialBackoff();
-  SetWaitIntervalHadNudge(true);
 
   SyncSchedulerImpl::JobProcessDecision decision = CreateAndDecideJob(
       SyncSessionJob::NUDGE);
diff --git a/sync/engine/sync_session_job.cc b/sync/engine/sync_session_job.cc
index 0195f80..84b491e 100644
--- a/sync/engine/sync_session_job.cc
+++ b/sync/engine/sync_session_job.cc
@@ -18,8 +18,7 @@ SyncSessionJob::SyncSessionJob(
     : purpose_(purpose),
       source_info_(source_info),
       scheduled_start_(start),
-      config_params_(config_params),
-      finished_(NOT_FINISHED) {
+      config_params_(config_params) {
 }
 
 #define ENUM_CASE(x) case x: return #x; break;
@@ -36,15 +35,9 @@ const char* SyncSessionJob::GetPurposeString(SyncSessionJob::Purpose purpose) {
 #undef ENUM_CASE
 
 bool SyncSessionJob::Finish(bool early_exit, sessions::SyncSession* session) {
-  DCHECK_EQ(finished_, NOT_FINISHED);
-  // Did we run through all SyncerSteps from start_step() to end_step()
-  // until the SyncSession returned !HasMoreToSync()?
-  // Note: if not, it's possible the scheduler hasn't started with
-  // SyncShare yet, it's possible there is still more to sync in the session,
-  // and it's also possible the job quit part way through due to a premature
-  // exit condition (such as shutdown).
-  finished_ = early_exit ? EARLY_EXIT : FINISHED;
-
+  // 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;
 
@@ -75,12 +68,6 @@ bool SyncSessionJob::Finish(bool early_exit, sessions::SyncSession* session) {
   return true;
 }
 
-scoped_ptr<SyncSessionJob> SyncSessionJob::Clone() const {
-  return scoped_ptr<SyncSessionJob>(new SyncSessionJob(
-      purpose_, scheduled_start_, source_info_,
-      config_params_));
-}
-
 void SyncSessionJob::CoalesceSources(const sessions::SyncSourceInfo& source) {
   CoalesceStates(source.types, &source_info_.types);
   source_info_.updates_source = source.updates_source;
diff --git a/sync/engine/sync_session_job.h b/sync/engine/sync_session_job.h
index feb7405..7db96f3 100644
--- a/sync/engine/sync_session_job.h
+++ b/sync/engine/sync_session_job.h
@@ -37,10 +37,6 @@ class SYNC_EXPORT_PRIVATE SyncSessionJob {
                  const ConfigurationParams& config_params);
   ~SyncSessionJob();
 
-  // Returns a new clone of the job, with a cloned SyncSession ready to be
-  // retried / rescheduled.
-  scoped_ptr<SyncSessionJob> Clone() const;
-
   // Overwrite the sync update source with the most recent and merge the
   // type/state map.
   void CoalesceSources(const sessions::SyncSourceInfo& source);
@@ -75,14 +71,6 @@ class SYNC_EXPORT_PRIVATE SyncSessionJob {
   ConfigurationParams config_params() const;
 
  private:
-  // A SyncSessionJob can be in one of these three states, controlled by the
-  // Finish() function, see method comments.
-  enum FinishedState {
-    NOT_FINISHED,  // Finish has not been called.
-    EARLY_EXIT,    // Finish was called but the job was "preempted",
-    FINISHED       // Indicates a "clean" finish operation.
-  };
-
   const Purpose purpose_;
   sessions::SyncSourceInfo source_info_;
 
@@ -92,11 +80,6 @@ class SYNC_EXPORT_PRIVATE SyncSessionJob {
   // Succeeded() behavior may be arguments to subclass in the future.
   const ConfigurationParams config_params_;
 
-  // Set to true if Finish() was called, false otherwise. True implies that
-  // a SyncShare operation took place with |session_| and it cycled through
-  // all requisite steps given |purpose_| without being preempted.
-  FinishedState finished_;
-
   DISALLOW_COPY_AND_ASSIGN(SyncSessionJob);
 };
 
diff --git a/sync/engine/sync_session_job_unittest.cc b/sync/engine/sync_session_job_unittest.cc
index 4f3a67c..8847833 100644
--- a/sync/engine/sync_session_job_unittest.cc
+++ b/sync/engine/sync_session_job_unittest.cc
@@ -120,64 +120,6 @@ class SyncSessionJobTest : public testing::Test {
   bool config_params_callback_invoked_;
 };
 
-TEST_F(SyncSessionJobTest, Clone) {
-  SyncSessionJob job1(SyncSessionJob::NUDGE, TimeTicks::Now(),
-      MakeSourceInfo(), ConfigurationParams());
-
-  scoped_ptr<SyncSession> session1 = MakeSession().Pass();
-  sessions::test_util::SimulateSuccess(session1.get(),
-                                       job1.start_step(),
-                                       job1.end_step());
-  job1.Finish(false, session1.get());
-  ModelSafeRoutingInfo new_routes;
-  new_routes[AUTOFILL] = GROUP_PASSIVE;
-  context()->set_routing_info(new_routes);
-  scoped_ptr<SyncSessionJob> clone1 = job1.Clone();
-
-  ExpectClones(&job1, clone1.get());
-
-  context()->set_routing_info(routes());
-  scoped_ptr<SyncSession> session2 = MakeSession().Pass();
-  sessions::test_util::SimulateSuccess(session2.get(),
-                                       clone1->start_step(),
-                                       clone1->end_step());
-  clone1->Finish(false, session2.get());
-  scoped_ptr<SyncSessionJob> clone2 = clone1->Clone();
-
-  ExpectClones(clone1.get(), clone2.get());
-
-  clone1.reset();
-  ExpectClones(&job1, clone2.get());
-}
-
-TEST_F(SyncSessionJobTest, CloneAfterEarlyExit) {
-  scoped_ptr<SyncSession> session = MakeSession().Pass();
-  SyncSessionJob job1(SyncSessionJob::NUDGE, TimeTicks::Now(),
-      MakeSourceInfo(), ConfigurationParams());
-  job1.Finish(true, session.get());
-  scoped_ptr<SyncSessionJob> job2 = job1.Clone();
-  ExpectClones(&job1, job2.get());
-}
-
-// Tests interaction between Finish and sync cycle success / failure.
-TEST_F(SyncSessionJobTest, Finish) {
-  SyncSessionJob job1(SyncSessionJob::NUDGE, TimeTicks::Now(),
-      MakeSourceInfo(), ConfigurationParams());
-
-  scoped_ptr<SyncSession> session1 = MakeSession().Pass();
-  sessions::test_util::SimulateSuccess(session1.get(),
-                                       job1.start_step(),
-                                       job1.end_step());
-  EXPECT_TRUE(job1.Finish(false /* early_exit */, session1.get()));
-
-  scoped_ptr<SyncSessionJob> job2 = job1.Clone();
-  scoped_ptr<SyncSession> session2 = MakeSession().Pass();
-  sessions::test_util::SimulateConnectionFailure(session2.get(),
-                                                 job2->start_step(),
-                                                 job2->end_step());
-  EXPECT_FALSE(job2->Finish(false, session2.get()));
-}
-
 TEST_F(SyncSessionJobTest, FinishCallsReadyTask) {
   ConfigurationParams params;
   params.ready_task = base::Bind(