sync: make scheduling logic and job ownership more obvious.

sync/engine/sync_scheduler_impl.cc

 // 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_scheduler_impl.h"
 
 #include <algorithm>
 #include <cstring>
 
 #include "base/auto_reset.h"
 #include "base/bind.h"
+#include "base/bind_helpers.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "sync/engine/backoff_delay_provider.h"
 #include "sync/engine/syncer.h"
 #include "sync/engine/throttled_data_type_tracker.h"
 #include "sync/protocol/proto_enum_conversions.h"
 #include "sync/protocol/sync.pb.h"
 #include "sync/util/data_type_histogram.h"
@@ skipping 54 matching lines @@
     : source(source),
       types_to_download(types_to_download),
       routing_info(routing_info),
       ready_task(ready_task) {
   DCHECK(!ready_task.is_null());
 }
 ConfigurationParams::~ConfigurationParams() {}
 
 SyncSchedulerImpl::WaitInterval::WaitInterval()
     : mode(UNKNOWN),
-      had_nudge(false) {
-}
+      had_nudge(false),
+      pending_configure_job(NULL) {}
+
+SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
+    : mode(mode), had_nudge(false), length(length),
+      pending_configure_job(NULL) {}
 
 SyncSchedulerImpl::WaitInterval::~WaitInterval() {}
 
 #define ENUM_CASE(x) case x: return #x; break;
 
 const char* SyncSchedulerImpl::WaitInterval::GetModeString(Mode mode) {
   switch (mode) {
     ENUM_CASE(UNKNOWN);
     ENUM_CASE(EXPONENTIAL_BACKOFF);
     ENUM_CASE(THROTTLED);
   }
   NOTREACHED();
   return "";
 }
 
-SyncSchedulerImpl::SyncSessionJob::SyncSessionJob()
-    : purpose(UNKNOWN),
-      is_canary_job(false) {
-}
-
-SyncSchedulerImpl::SyncSessionJob::~SyncSessionJob() {}
-
-SyncSchedulerImpl::SyncSessionJob::SyncSessionJob(SyncSessionJobPurpose purpose,
-    base::TimeTicks start,
-    linked_ptr<sessions::SyncSession> session,
-    bool is_canary_job,
-    const ConfigurationParams& config_params,
-    const tracked_objects::Location& from_here)
-    : purpose(purpose),
-      scheduled_start(start),
-      session(session),
-      is_canary_job(is_canary_job),
-      config_params(config_params),
-      from_here(from_here) {
-}
-
-const char* SyncSchedulerImpl::SyncSessionJob::GetPurposeString(
-    SyncSchedulerImpl::SyncSessionJob::SyncSessionJobPurpose purpose) {
-  switch (purpose) {
-    ENUM_CASE(UNKNOWN);
-    ENUM_CASE(POLL);
-    ENUM_CASE(NUDGE);
-    ENUM_CASE(CONFIGURATION);
-  }
-  NOTREACHED();
-  return "";
-}
-
 GetUpdatesCallerInfo::GetUpdatesSource GetUpdatesFromNudgeSource(
     NudgeSource source) {
   switch (source) {
     case NUDGE_SOURCE_NOTIFICATION:
       return GetUpdatesCallerInfo::NOTIFICATION;
     case NUDGE_SOURCE_LOCAL:
       return GetUpdatesCallerInfo::LOCAL;
     case NUDGE_SOURCE_CONTINUATION:
       return GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION;
     case NUDGE_SOURCE_LOCAL_REFRESH:
       return GetUpdatesCallerInfo::DATATYPE_REFRESH;
     case NUDGE_SOURCE_UNKNOWN:
       return GetUpdatesCallerInfo::UNKNOWN;
     default:
       NOTREACHED();
       return GetUpdatesCallerInfo::UNKNOWN;
   }
 }
 
-SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
-    : mode(mode), had_nudge(false), length(length) { }
-
 // Helper macros to log with the syncer thread name; useful when there
 // are multiple syncer threads involved.
 
 #define SLOG(severity) LOG(severity) << name_ << ": "
 
 #define SDVLOG(verbose_level) DVLOG(verbose_level) << name_ << ": "
 
 #define SDVLOG_LOC(from_here, verbose_level)             \
   DVLOG_LOC(from_here, verbose_level) << name_ << ": "
 
@@ skipping 30 matching lines @@
       syncer_short_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultShortPollIntervalSeconds)),
       syncer_long_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultLongPollIntervalSeconds)),
       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.
       connection_code_(HttpResponse::SERVER_CONNECTION_OK),
+      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_);
   StopImpl(base::Closure());
@@ skipping 16 matching lines @@
 void SyncSchedulerImpl::OnConnectionStatusChange() {
   if (HttpResponse::CONNECTION_UNAVAILABLE  == connection_code_) {
     // Optimistically assume that the connection is fixed and try
     // connecting.
     OnServerConnectionErrorFixed();
   }
 }
 
 void SyncSchedulerImpl::OnServerConnectionErrorFixed() {
   connection_code_ = HttpResponse::SERVER_CONNECTION_OK;
+  // There could be a pending nudge or configuration job in several cases:
+  //
+  // 1. We're in exponential backoff.
+  // 2. We're silenced / throttled.
+  // 3. A nudge was saved previously due to not having a valid auth token.
+  // 4. A nudge was scheduled + saved while in configuration mode.
+  //
+  // In all cases except (2), we want to retry contacting the server. We
+  // 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;
+
   PostTask(FROM_HERE, "DoCanaryJob",
            base::Bind(&SyncSchedulerImpl::DoCanaryJob,
-                      weak_ptr_factory_.GetWeakPtr()));
-
+                      weak_ptr_factory_.GetWeakPtr(),
+                      base::Passed(&pending)));
 }
 
 void SyncSchedulerImpl::UpdateServerConnectionManagerStatus(
     HttpResponse::ServerConnectionCode code) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "New server connection code: "
             << HttpResponse::GetServerConnectionCodeString(code);
 
   connection_code_ = code;
 }
@@ skipping 15 matching lines @@
   Mode old_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.get());
+      DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
     }
 
-    DoPendingJobIfPossible(false);
+    scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
+    if (pending.get()) {
+      // TODO(tim): We should be able to remove this...
+      scoped_ptr<SyncSession> session(CreateSyncSession(
+          pending->session()->source()));
+      // Also the routing info might have been changed since we cached the
+      // pending nudge. Update it by coalescing to the latest.
+      pending->mutable_session()->Coalesce(*(session));
+      SDVLOG(2) << "Executing pending job. Good luck!";
+      DoSyncSessionJob(pending.Pass());
+    }
   }
 }
 
 void SyncSchedulerImpl::SendInitialSnapshot() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   scoped_ptr<SyncSession> dummy(new SyncSession(session_context_, this,
       SyncSourceInfo(), ModelSafeRoutingInfo(),
       std::vector<ModelSafeWorker*>()));
   SyncEngineEvent event(SyncEngineEvent::STATUS_CHANGED);
   event.snapshot = dummy->TakeSnapshot();
@@ skipping 23 matching lines @@
 bool SyncSchedulerImpl::ScheduleConfiguration(
     const ConfigurationParams& params) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(IsConfigRelatedUpdateSourceValue(params.source));
   DCHECK_EQ(CONFIGURATION_MODE, mode_);
   DCHECK(!params.ready_task.is_null());
   SDVLOG(2) << "Reconfiguring syncer.";
 
   // Only one configuration is allowed at a time. Verify we're not waiting
   // for a pending configure job.
-  DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job.get());
+  DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
 
   ModelSafeRoutingInfo restricted_routes;
   BuildModelSafeParams(params.types_to_download,
                        params.routing_info,
                        &restricted_routes);
   session_context_->set_routing_info(params.routing_info);
 
   // Only reconfigure if we have types to download.
   if (!params.types_to_download.Empty()) {
     DCHECK(!restricted_routes.empty());
-    linked_ptr<SyncSession> session(new SyncSession(
+    scoped_ptr<SyncSession> session(new SyncSession(
         session_context_,
         this,
         SyncSourceInfo(params.source,
                        ModelSafeRoutingInfoToInvalidationMap(
                            restricted_routes,
                            std::string())),
         restricted_routes,
         session_context_->workers()));
-    SyncSessionJob job(SyncSessionJob::CONFIGURATION,
-                       TimeTicks::Now(),
-                       session,
-                       false,
-                       params,
-                       FROM_HERE);
-    DoSyncSessionJob(job);
+    scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
+        SyncSessionJob::CONFIGURATION,
+        TimeTicks::Now(),
+        session.Pass(),
+        params,
+        FROM_HERE));
+    bool succeeded = DoSyncSessionJob(job.Pass());
 
     // If we failed, the job would have been saved as the pending configure
     // job and a wait interval would have been set.
-    if (!session->Succeeded()) {
-      DCHECK(wait_interval_.get() &&
-             wait_interval_->pending_configure_job.get());
+    if (!succeeded) {
+      DCHECK(wait_interval_.get() && wait_interval_->pending_configure_job);
       return false;
     }
   } else {
     SDVLOG(2) << "No change in routing info, calling ready task directly.";
     params.ready_task.Run();
   }
 
   return true;
 }
 
 SyncSchedulerImpl::JobProcessDecision
-SyncSchedulerImpl::DecideWhileInWaitInterval(
-    const SyncSessionJob& job) {
+SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob* job) {

[akalin, 2012/10/17 17:39:17]

why does this need a pointer?

[tim (not reviewing), 2012/10/17 20:33:32]

Every callsite has a pointer, so it seems simpler / cleaner?

[akalin, 2012/10/17 22:55:45]

I don't know...const references are the usual style, unless the pointer value is
nullable or otherwise used.  I was scanning the function wondering how it was
used that made it necessary for a pointer instead of a reference. :/

   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(wait_interval_.get());
 
   SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
             << WaitInterval::GetModeString(wait_interval_->mode)
             << (wait_interval_->had_nudge ? " (had nudge)" : "")
-            << (job.is_canary_job ? " (canary)" : "");
+            << (job->is_canary() ? " (canary)" : "");
 
-  if (job.purpose == SyncSessionJob::POLL)
+  if (job->purpose() == SyncSessionJob::POLL)
     return DROP;
 
-  DCHECK(job.purpose == SyncSessionJob::NUDGE ||
-         job.purpose == SyncSessionJob::CONFIGURATION);
+  // 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;
 
   DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
-  if (job.purpose == SyncSessionJob::NUDGE) {
+  if (job->purpose() == SyncSessionJob::NUDGE) {
     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 (!job.is_canary_job)
+    if (!job->is_canary())
       return wait_interval_->had_nudge ? DROP : CONTINUE;
     else // We are here because timer ran out. So retry.
       return CONTINUE;
   }
-  return job.is_canary_job ? CONTINUE : SAVE;
+  return job->is_canary() ? CONTINUE : SAVE;
 }
 
 SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
-    const SyncSessionJob& job) {
+    const SyncSessionJob* job) {

[akalin, 2012/10/17 17:39:17]

why does this need a pointer?

[tim (not reviewing), 2012/10/17 20:33:32]

Every callsite has a pointer, so it seems simpler / cleaner?

[akalin, 2012/10/17 22:55:45]

same.

   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   // See if our type is throttled.
   ModelTypeSet throttled_types =
       session_context_->throttled_data_type_tracker()->GetThrottledTypes();
-  if (job.purpose == SyncSessionJob::NUDGE &&
-      job.session->source().updates_source == GetUpdatesCallerInfo::LOCAL) {
+  if (job->purpose() == SyncSessionJob::NUDGE &&
+      job->session()->source().updates_source == GetUpdatesCallerInfo::LOCAL) {
     ModelTypeSet requested_types;
     for (ModelTypeInvalidationMap::const_iterator i =
-         job.session->source().types.begin();
-         i != job.session->source().types.end();
+         job->session()->source().types.begin();
+         i != job->session()->source().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 SAVE;
   }
 
   if (wait_interval_.get())
     return DecideWhileInWaitInterval(job);
 
   if (mode_ == CONFIGURATION_MODE) {
-    if (job.purpose == SyncSessionJob::NUDGE)
-      return SAVE;
-    else if (job.purpose == SyncSessionJob::CONFIGURATION)
+    if (job->purpose() == SyncSessionJob::NUDGE)
+      return SAVE; // Running requires a mode switch.
+    else if (job->purpose() == SyncSessionJob::CONFIGURATION)
       return CONTINUE;
     else
       return DROP;
   }
 
   // We are in normal mode.
   DCHECK_EQ(mode_, NORMAL_MODE);
-  DCHECK_NE(job.purpose, SyncSessionJob::CONFIGURATION);
+  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 POLL with a |scheduled_start| time that is less than
   //    the time that the last successful all-datatype NUDGE completed.
   //
   // 2) |job| is a NUDGE (for any combination of types) with a
@@ skipping 22 matching lines @@
   //  * 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.";
-  return job.purpose == SyncSessionJob::NUDGE ? SAVE : DROP;
+  // Running the job would require updated auth, so we can't honour
+  // job.scheduled_start().
+  return job->purpose() == SyncSessionJob::NUDGE ? SAVE : DROP;
 }
 
-void SyncSchedulerImpl::InitOrCoalescePendingJob(const SyncSessionJob& job) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  DCHECK(job.purpose != SyncSessionJob::CONFIGURATION);
-  if (pending_nudge_.get() == NULL) {
-    SDVLOG(2) << "Creating a pending nudge job";
-    SyncSession* s = job.session.get();
-
-    // Get a fresh session with similar configuration as before (resets
-    // StatusController).
-    scoped_ptr<SyncSession> session(new SyncSession(s->context(),
-        s->delegate(), s->source(), s->routing_info(), s->workers()));
-
-    SyncSessionJob new_job(SyncSessionJob::NUDGE, job.scheduled_start,
-        make_linked_ptr(session.release()), false,
-        ConfigurationParams(), job.from_here);
-    pending_nudge_.reset(new SyncSessionJob(new_job));
-    return;
-  }
-
-  SDVLOG(2) << "Coalescing a pending nudge";
-  pending_nudge_->session->Coalesce(*(job.session.get()));
-  pending_nudge_->scheduled_start = job.scheduled_start;
-
-  // Unfortunately the nudge location cannot be modified. So it stores the
-  // location of the first caller.
-}
-
-bool SyncSchedulerImpl::ShouldRunJob(const SyncSessionJob& job) {
+bool SyncSchedulerImpl::ShouldRunJobSaveIfNecessary(SyncSessionJob* job) {

[akalin, 2012/10/17 17:39:17]

the name for this is unclear -- "ShouldRunJob" implies that it's just checks a
condition, but the "SaveIfNecessary" part seems more important.  Perhaps just
rename to "SaveIfNecessary" and comment re. the return value?

[tim (not reviewing), 2012/10/17 20:33:32]

SaveIfNecessary works, although there's also the case where it returns false and
doesn't save.  The return value is really there to tell the caller what to do.

[akalin, 2012/10/17 22:55:45]

I see.  See my comment below.

   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(started_);
 
   JobProcessDecision decision = DecideOnJob(job);
   SDVLOG(2) << "Should run "
-            << SyncSessionJob::GetPurposeString(job.purpose)
-            << " job in mode " << GetModeString(mode_)
+            << SyncSessionJob::GetPurposeString(job->purpose())
+            << " job " << job->session()
+            << " in mode " << GetModeString(mode_)
             << ": " << GetDecisionString(decision);
-  if (decision != SAVE)
-    return decision == CONTINUE;
+  if (decision == DROP)
+    return false;
+  if (decision == CONTINUE)
+    return true;
+  DCHECK_EQ(decision, SAVE);
 
-  DCHECK(job.purpose == SyncSessionJob::NUDGE || job.purpose ==
-      SyncSessionJob::CONFIGURATION);
+  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

[rlarocque, 2012/10/08 19:18:05]

After some more thought, I'm not sure this ever mattered.  This can only happen
when we decided to save the job, which implies we're in a WaitInterval or
Configure mode.  In either case, this job's scheduled start time is meaningless.
 We'll either run it when we switch out of configure mode, or on the next
attempt to break out of the WaitInterval.

Based on your other comment, I'm guessing you already knew this.  However, it
didn't register for me until now.

+    // 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_->mutable_session()->Coalesce(*(job->session()));
+    return false;
+  }
 
-  SaveJob(job);
+  scoped_ptr<SyncSessionJob> job_to_save = job->CloneAndAbandon();
+  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_->CloneAndAbandon();
+        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 false;
+  }
+
+  // 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();
   return false;
 }
 
-void SyncSchedulerImpl::SaveJob(const SyncSessionJob& job) {
-  DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  if (job.purpose == SyncSessionJob::NUDGE) {
-    SDVLOG(2) << "Saving a nudge job";
-    InitOrCoalescePendingJob(job);
-  } else if (job.purpose == SyncSessionJob::CONFIGURATION){
-    SDVLOG(2) << "Saving a configuration job";
-    DCHECK(wait_interval_.get());
-    DCHECK(mode_ == CONFIGURATION_MODE);
-
-    // Config params should always get set.
-    DCHECK(!job.config_params.ready_task.is_null());
-    SyncSession* old = job.session.get();
-    SyncSession* s(new SyncSession(session_context_, this, old->source(),
-                                   old->routing_info(), old->workers()));
-    SyncSessionJob new_job(job.purpose,
-                           TimeTicks::Now(),
-                           make_linked_ptr(s),
-                           false,
-                           job.config_params,
-                           job.from_here);
-    wait_interval_->pending_configure_job.reset(new SyncSessionJob(new_job));
-  } // drop the rest.
-  // TODO(sync): Is it okay to drop the rest?  It's weird that
-  // SaveJob() only does what it says sometimes.  (See
-  // http://crbug.com/90868.)
-}
-
 // 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& delay,
+    const TimeDelta& desired_delay,
     NudgeSource source, ModelTypeSet types,
     const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG_LOC(nudge_location, 2)
-      << "Nudge scheduled with delay " << delay.InMilliseconds() << " ms, "
+      << "Nudge scheduled with delay "
+      << desired_delay.InMilliseconds() << " ms, "
       << "source " << GetNudgeSourceString(source) << ", "
       << "types " << ModelTypeSetToString(types);
 
   ModelTypeInvalidationMap invalidation_map =
       ModelTypeSetToInvalidationMap(types, std::string());
-  SyncSchedulerImpl::ScheduleNudgeImpl(delay,
+  SyncSchedulerImpl::ScheduleNudgeImpl(desired_delay,
                                        GetUpdatesFromNudgeSource(source),
                                        invalidation_map,
-                                       false,
                                        nudge_location);
 }
 
 void SyncSchedulerImpl::ScheduleNudgeWithStatesAsync(
-    const TimeDelta& delay,
+    const TimeDelta& desired_delay,
     NudgeSource source, const ModelTypeInvalidationMap& invalidation_map,
     const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG_LOC(nudge_location, 2)
-      << "Nudge scheduled with delay " << delay.InMilliseconds() << " ms, "
+      << "Nudge scheduled with delay "
+      << desired_delay.InMilliseconds() << " ms, "
       << "source " << GetNudgeSourceString(source) << ", "
       << "payloads "
       << ModelTypeInvalidationMapToString(invalidation_map);
 
-  SyncSchedulerImpl::ScheduleNudgeImpl(delay,
+  SyncSchedulerImpl::ScheduleNudgeImpl(desired_delay,
                                        GetUpdatesFromNudgeSource(source),
                                        invalidation_map,
-                                       false,
                                        nudge_location);
 }
 
 void SyncSchedulerImpl::ScheduleNudgeImpl(
     const TimeDelta& delay,
     GetUpdatesCallerInfo::GetUpdatesSource source,
     const ModelTypeInvalidationMap& invalidation_map,
-    bool is_canary_job, const tracked_objects::Location& nudge_location) {
+    const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK(!invalidation_map.empty()) << "Nudge scheduled for no types!";
 
   SDVLOG_LOC(nudge_location, 2)
       << "In ScheduleNudgeImpl with delay "
       << delay.InMilliseconds() << " ms, "
       << "source " << GetUpdatesSourceString(source) << ", "
       << "payloads "
-      << ModelTypeInvalidationMapToString(invalidation_map)
-      << (is_canary_job ? " (canary)" : "");
+      << ModelTypeInvalidationMapToString(invalidation_map);
 
   SyncSourceInfo info(source, invalidation_map);
   UpdateNudgeTimeRecords(info);
 
-  SyncSession* session(CreateSyncSession(info));
-  SyncSessionJob job(SyncSessionJob::NUDGE, TimeTicks::Now() + delay,
-                     make_linked_ptr(session), is_canary_job,
-                     ConfigurationParams(), nudge_location);
+  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
+      SyncSessionJob::NUDGE,
+      TimeTicks::Now() + delay,
+      CreateSyncSession(info).Pass(),
+      ConfigurationParams(),
+      nudge_location));
 
-  session = NULL;
-  if (!ShouldRunJob(job))
+  if (!ShouldRunJobSaveIfNecessary(job.get()))

[akalin, 2012/10/17 17:39:17]

Naively, I'd expect to be able to split this call up like this:

If (!ShouldRunJob(*job))
  return;

// Abandons job and save a clone.
Save(job);


Anything wrong with that?

[tim (not reviewing), 2012/10/17 20:33:32]

Yes.  !ShouldRunJob doesn't mean you can forget about the job, it means you
shouldn't run it right now.  

What could be done is to call DecideOnJob and call a save function if it needs
to be saved.  I was explicitly trying to avoid having a 'SaveJob' function
floating around like before, because we wound up wanting to call it all over the
place and added some special cases to it that led to comments like "it's weird
that save doesn't do what it says sometimes".  The only things that really need
the save behavior are the two pre-scheduling cases, and in this case I think
inlining the saving code right after the call to DecideOnJob in ShouldRun is
clearest, and is least likely to be abused because the method is pretty targeted
at that use case.  

But if that leaves you unconvinced I'm not opposed.  

[akalin, 2012/10/17 22:55:45]

I see.  See comment below...

     return;
 
-  if (pending_nudge_.get()) {
-    if (IsBackingOff() && delay > TimeDelta::FromSeconds(1)) {
-      SDVLOG(2) << "Dropping the nudge because we are in backoff";
-      return;
-    }
-
-    SDVLOG(2) << "Coalescing pending nudge";
-    pending_nudge_->session->Coalesce(*(job.session.get()));
-
+  if (pending_nudge_) {
     SDVLOG(2) << "Rescheduling pending nudge";
-    SyncSession* s = pending_nudge_->session.get();
-    job.session.reset(new SyncSession(s->context(), s->delegate(),
-        s->source(), s->routing_info(), s->workers()));
-
-    // Choose the start time as the earliest of the 2.
-    job.scheduled_start = std::min(job.scheduled_start,
-                                   pending_nudge_->scheduled_start);
-    pending_nudge_.reset();
+    pending_nudge_->mutable_session()->Coalesce(*(job->session()));
+    // 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_->CloneAndAbandon();
+    unscheduled_nudge_storage_.reset();
+    pending_nudge_ = NULL;
   }
 
   // TODO(zea): Consider adding separate throttling/backoff for datatype
   // refresh requests.
-  ScheduleSyncSessionJob(job);
+  ScheduleSyncSessionJob(job.Pass());

[akalin, 2012/10/17 17:39:17]

Isn't the job abandoned at this point?  Do we still need to schedule it?

[tim (not reviewing), 2012/10/17 20:33:32]

Err, did you mean to put this comment elsewhere? This is actually _the_ main
place jobs get scheduled?

[akalin, 2012/10/17 22:55:45]

Right, I was misunderstanding the flow.

So it looks like we create a new job, we pass it to ShouldRunJobSaveIfNecessary;
if it returns false, we exit early (and destroy the job, which may have been
cloned and stashed somewhere).  If it returns true, we continue to schedule the
job, passing ownership to ScheduleSyncSessionJob.

What about making ShouldRunJobSaveIfNecessary take ownership of job, then have
it return a scoped_ptr<Job> that may be null -- if it's not null, that means we
should continue to schedule the job.  If it's null, we may or may not have
saved/cloned the job somewhere, but we're not going to run it now.

Maybe it can then be renamed to GetJobToRun()?  WDYT?

I guess it'd look something lke:

scoped_ptr<SyncSessionJob> new_job(...);

scoped_ptr<SyncSessionJob> job_to_run(GetJobToRun(new_job));

if (pending_nudge_) {
  // coalesce into job_to_run
}

ScheduleSyncSessionJob(job_to_run.Pass());

 }
 
 const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
   switch (mode) {
     ENUM_CASE(CONFIGURATION_MODE);
     ENUM_CASE(NORMAL_MODE);
   }
   return "";
 }
 
 const char* SyncSchedulerImpl::GetDecisionString(
     SyncSchedulerImpl::JobProcessDecision mode) {
   switch (mode) {
     ENUM_CASE(CONTINUE);
     ENUM_CASE(SAVE);
     ENUM_CASE(DROP);
   }
   return "";
 }
 
-// static
-void SyncSchedulerImpl::SetSyncerStepsForPurpose(
-    SyncSessionJob::SyncSessionJobPurpose 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;
-  }
-}
-
 void SyncSchedulerImpl::PostTask(
     const tracked_objects::Location& from_here,
     const char* name, const base::Closure& task) {
   SDVLOG_LOC(from_here, 3) << "Posting " << name << " task";
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   if (!started_) {
     SDVLOG(1) << "Not posting task as scheduler is stopped.";
     return;
   }
   sync_loop_->PostTask(from_here, task);
 }
 
 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;
   }
   sync_loop_->PostDelayedTask(from_here, task, delay);
 }
 
-void SyncSchedulerImpl::ScheduleSyncSessionJob(const SyncSessionJob& job) {
+void SyncSchedulerImpl::ScheduleSyncSessionJob(
+    scoped_ptr<SyncSessionJob> job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   if (no_scheduling_allowed_) {
     NOTREACHED() << "Illegal to schedule job while session in progress.";
     return;
   }
 
-  TimeDelta delay = job.scheduled_start - TimeTicks::Now();
+  TimeDelta delay = job->scheduled_start() - TimeTicks::Now();
+  tracked_objects::Location loc(job->from_location());
   if (delay < TimeDelta::FromMilliseconds(0))
     delay = TimeDelta::FromMilliseconds(0);
-  SDVLOG_LOC(job.from_here, 2)
+  SDVLOG_LOC(loc, 2)
       << "In ScheduleSyncSessionJob with "
-      << SyncSessionJob::GetPurposeString(job.purpose)
+      << SyncSessionJob::GetPurposeString(job->purpose())
       << " job and " << delay.InMilliseconds() << " ms delay";
 
-  DCHECK(job.purpose == SyncSessionJob::NUDGE ||
-         job.purpose == SyncSessionJob::POLL);
-  if (job.purpose == SyncSessionJob::NUDGE) {
-    SDVLOG_LOC(job.from_here, 2) << "Resetting pending_nudge";
-    DCHECK(!pending_nudge_.get() || pending_nudge_->session.get() ==
-           job.session);
-    pending_nudge_.reset(new SyncSessionJob(job));
+  DCHECK(job->purpose() == SyncSessionJob::NUDGE ||
+         job->purpose() == SyncSessionJob::POLL);
+  if (job->purpose() == SyncSessionJob::NUDGE) {
+    SDVLOG_LOC(loc, 2) << "Resetting pending_nudge to ";
+    DCHECK(!pending_nudge_ || pending_nudge_->session() ==
+           job->session());
+    pending_nudge_ = job.get();
   }
-  PostDelayedTask(job.from_here, "DoSyncSessionJob",
-                  base::Bind(&SyncSchedulerImpl::DoSyncSessionJob,
-                             weak_ptr_factory_.GetWeakPtr(),
-                             job),
-                  delay);
+
+  PostDelayedTask(loc, "DoSyncSessionJob",
+      base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoSyncSessionJob),
+                                    weak_ptr_factory_.GetWeakPtr(),
+                                    base::Passed(&job)),
+      delay);
 }
 
-void SyncSchedulerImpl::DoSyncSessionJob(const SyncSessionJob& job) {
+bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
+  if (job->purpose() == SyncSessionJob::NUDGE) {
+    if (pending_nudge_ == NULL ||
+        pending_nudge_->session() != job->session()) {
+      // |job| is abandoned.
+      SDVLOG(2) << "Dropping a nudge in "
+                << "DoSyncSessionJob because another nudge was scheduled";
+      return false;
+    }
+    pending_nudge_ = NULL;
+
+    // Rebase the session with the latest model safe table and use it to purge
+    // and update any disabled or modified entries in the job.
+    job->mutable_session()->RebaseRoutingInfoWithLatest(
+        session_context_->routing_info(), session_context_->workers());
+  }
 
   AutoReset<bool> protector(&no_scheduling_allowed_, true);
-  if (!ShouldRunJob(job)) {
+  GetUpdatesCallerInfo::GetUpdatesSource source(
+      job->session()->source().updates_source);
+  if (!ShouldRunJobSaveIfNecessary(job.get())) {
     SLOG(WARNING)
         << "Not executing "
-        << SyncSessionJob::GetPurposeString(job.purpose) << " job from "
-        << GetUpdatesSourceString(job.session->source().updates_source);
-    return;
+        << SyncSessionJob::GetPurposeString(job->purpose()) << " job from "
+        << GetUpdatesSourceString(source);
+    return false;
   }
 
-  if (job.purpose == SyncSessionJob::NUDGE) {
-    if (pending_nudge_.get() == NULL ||
-        pending_nudge_->session != job.session) {
-      SDVLOG(2) << "Dropping a nudge in "
-                << "DoSyncSessionJob because another nudge was scheduled";
-      return;  // Another nudge must have been scheduled in in the meantime.
-    }
-    pending_nudge_.reset();
-
-    // Create the session with the latest model safe table and use it to purge
-    // and update any disabled or modified entries in the job.
-    scoped_ptr<SyncSession> session(CreateSyncSession(job.session->source()));
-
-    job.session->RebaseRoutingInfoWithLatest(*session);
-  }
   SDVLOG(2) << "DoSyncSessionJob with "
-            << SyncSessionJob::GetPurposeString(job.purpose) << " job";
-
-  SyncerStep begin(SYNCER_END);
-  SyncerStep end(SYNCER_END);
-  SetSyncerStepsForPurpose(job.purpose, &begin, &end);
+            << SyncSessionJob::GetPurposeString(job->purpose()) << " job";
 
   bool has_more_to_sync = true;
-  while (ShouldRunJob(job) && has_more_to_sync) {
+  bool premature_exit = false;
+  while (DecideOnJob(job.get()) == CONTINUE && has_more_to_sync) {
     SDVLOG(2) << "Calling SyncShare.";
     // Synchronously perform the sync session from this thread.
-    syncer_->SyncShare(job.session.get(), begin, end);
-    has_more_to_sync = job.session->HasMoreToSync();
+    premature_exit = !syncer_->SyncShare(job->mutable_session(),
+                                         job->start_step(),
+                                         job->end_step());
+
+    has_more_to_sync = job->session()->HasMoreToSync();
     if (has_more_to_sync)
-      job.session->PrepareForAnotherSyncCycle();
+      job->mutable_session()->PrepareForAnotherSyncCycle();
   }
   SDVLOG(2) << "Done SyncShare looping.";
 
-  FinishSyncSessionJob(job);
+  return FinishSyncSessionJob(job.Pass(), premature_exit);
 }
 
 void SyncSchedulerImpl::UpdateNudgeTimeRecords(const SyncSourceInfo& info) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   // We are interested in recording time between local nudges for datatypes.
   // TODO(tim): Consider tracking LOCAL_NOTIFICATION as well.
   if (info.updates_source != GetUpdatesCallerInfo::LOCAL)
     return;
 
   base::TimeTicks now = TimeTicks::Now();
   // Update timing information for how often datatypes are triggering nudges.
   for (ModelTypeInvalidationMap::const_iterator iter = info.types.begin();
        iter != info.types.end();
        ++iter) {
     base::TimeTicks previous = last_local_nudges_by_model_type_[iter->first];
     last_local_nudges_by_model_type_[iter->first] = now;
     if (previous.is_null())
       continue;
 
 #define PER_DATA_TYPE_MACRO(type_str) \
     SYNC_FREQ_HISTOGRAM("Sync.Freq" type_str, now - previous);
-      SYNC_DATA_TYPE_HISTOGRAM(iter->first);
+    SYNC_DATA_TYPE_HISTOGRAM(iter->first);
 #undef PER_DATA_TYPE_MACRO
   }
 }
 
-void SyncSchedulerImpl::FinishSyncSessionJob(const SyncSessionJob& job) {
+bool SyncSchedulerImpl::FinishSyncSessionJob(scoped_ptr<SyncSessionJob> job,
+                                             bool exited_prematurely) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   // Now update the status of the connection from SCM. We need this to decide
-  // whether we need to save/run future jobs. The notifications from SCM are not
-  // reliable.
+  // whether we need to save/run future jobs. The notifications from SCM are
+  // not reliable.
   //
   // TODO(rlarocque): crbug.com/110954
   // We should get rid of the notifications and it is probably not needed to
-  // maintain this status variable in 2 places. We should query it directly from
-  // SCM when needed.
+  // maintain this status variable in 2 places. We should query it directly
+  // from SCM when needed.
   ServerConnectionManager* scm = session_context_->connection_manager();
   UpdateServerConnectionManagerStatus(scm->server_status());
 
-  if (IsSyncingCurrentlySilenced()) {
-    SDVLOG(2) << "We are currently throttled; not scheduling the next sync.";
-    // TODO(sync): Investigate whether we need to check job.purpose
-    // here; see DCHECKs in SaveJob().  (See http://crbug.com/90868.)
-    SaveJob(job);
-    return;  // Nothing to do.
-  } else if (job.session->Succeeded() &&
-             !job.config_params.ready_task.is_null()) {
-    // If this was a configuration job with a ready task, invoke it now that
-    // we finished successfully.
+  // Let job know that we're through syncing (calling SyncShare) at this point.
+  bool succeeded = false;
+  {
     AutoReset<bool> protector(&no_scheduling_allowed_, true);
-    job.config_params.ready_task.Run();
+    succeeded = job->Finish(exited_prematurely);
   }
 
   SDVLOG(2) << "Updating the next polling time after SyncMain";
-  ScheduleNextSync(job);
+  ScheduleNextSync(job.Pass(), succeeded);
+  return succeeded;
 }
 
-void SyncSchedulerImpl::ScheduleNextSync(const SyncSessionJob& old_job) {
+void SyncSchedulerImpl::ScheduleNextSync(
+    scoped_ptr<SyncSessionJob> finished_job, bool succeeded) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  DCHECK(!old_job.session->HasMoreToSync());
+  DCHECK(!finished_job->session()->HasMoreToSync());
 
-  AdjustPolling(&old_job);
+  AdjustPolling(finished_job.get());
 
-  if (old_job.session->Succeeded()) {
+  if (succeeded) {
     // Only reset backoff if we actually reached the server.
-    if (old_job.session->SuccessfullyReachedServer())
+    // It's possible that we reached the server on one attempt, then had an
+    // error on the next (or didn't perform some of the server-communicating
+    // commands). We want to verify that, for all commands attempted, we
+    // successfully spoke with the server. Therefore, we verify no errors
+    // and at least one SYNCER_OK.
+    if (finished_job->session()->DidReachServer())
       wait_interval_.reset();
     SDVLOG(2) << "Job succeeded so not scheduling more jobs";
     return;
   }
 
-  if (old_job.purpose == SyncSessionJob::POLL) {
+  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).
+    scoped_ptr<SyncSessionJob> clone = finished_job->Clone();
+    if (clone->purpose() == SyncSessionJob::NUDGE)
+      pending_nudge_ = clone.get();
+    else if (clone->purpose() == SyncSessionJob::CONFIGURATION)
+      wait_interval_->pending_configure_job = clone.get();
+    else
+      clone.reset();  // Unthrottling is enough, no need to force a canary.
+
+    RestartWaiting(clone.Pass());
+    return;
+  }
+
+  if (finished_job->purpose() == SyncSessionJob::POLL) {
     return; // We don't retry POLL jobs.
   }
 
   // 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
+  // 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.
 
   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
+    // 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";
-
-    DCHECK_EQ(SyncSessionJob::NUDGE, old_job.purpose);
+    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;
-    InitOrCoalescePendingJob(old_job);
-    RestartWaiting();
+    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(old_job);
+    HandleContinuationError(finished_job.get());
   }
 }
 
 void SyncSchedulerImpl::AdjustPolling(const SyncSessionJob* old_job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
 
   TimeDelta poll  = (!session_context_->notifications_enabled()) ?
       syncer_short_poll_interval_seconds_ :
       syncer_long_poll_interval_seconds_;
   bool rate_changed = !poll_timer_.IsRunning() ||
                        poll != poll_timer_.GetCurrentDelay();
 
-  if (old_job && old_job->purpose != SyncSessionJob::POLL && !rate_changed)
+  if (old_job && old_job->purpose() != SyncSessionJob::POLL && !rate_changed)
     poll_timer_.Reset();
 
   if (!rate_changed)
     return;
 
   // Adjust poll rate.
   poll_timer_.Stop();
   poll_timer_.Start(FROM_HERE, poll, this,
                     &SyncSchedulerImpl::PollTimerCallback);
 }
 
-void SyncSchedulerImpl::RestartWaiting() {
+void SyncSchedulerImpl::RestartWaiting(scoped_ptr<SyncSessionJob> job) {
   CHECK(wait_interval_.get());
   wait_interval_->timer.Stop();
-  wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
-                              this, &SyncSchedulerImpl::DoCanaryJob);
+  DCHECK(wait_interval_->length >= TimeDelta::FromSeconds(0));
+  if (wait_interval_->mode == WaitInterval::THROTTLED) {
+    wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
+                                base::Bind(&SyncSchedulerImpl::Unthrottle,
+                                    weak_ptr_factory_.GetWeakPtr(),
+                                    base::Passed(&job)));
+  } else {
+    wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
+                                base::Bind(&SyncSchedulerImpl::DoCanaryJob,
+                                    weak_ptr_factory_.GetWeakPtr(),
+                                    base::Passed(&job)));
+  }
 }
 
 void SyncSchedulerImpl::HandleContinuationError(
-    const SyncSessionJob& old_job) {
+    const SyncSessionJob* old_job) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   if (DCHECK_IS_ON()) {
     if (IsBackingOff()) {
-      DCHECK(wait_interval_->timer.IsRunning() || old_job.is_canary_job);
+      DCHECK(wait_interval_->timer.IsRunning() || old_job->is_canary());
     }
   }
 
   TimeDelta length = delay_provider_->GetDelay(
       IsBackingOff() ? wait_interval_->length :
           delay_provider_->GetInitialDelay(
-              old_job.session->status_controller().model_neutral_state()));
+              old_job->session()->status_controller().model_neutral_state()));
 
   SDVLOG(2) << "In handle continuation error with "
-            << SyncSessionJob::GetPurposeString(old_job.purpose)
+            << SyncSessionJob::GetPurposeString(old_job->purpose())
             << " 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));
-  if (old_job.purpose == SyncSessionJob::CONFIGURATION) {
+  scoped_ptr<SyncSessionJob> new_job(old_job->CloneFromLocation(FROM_HERE));
+  new_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());
-    SyncSession* old = old_job.session.get();
-    SyncSession* s(new SyncSession(session_context_, this,
-        old->source(), old->routing_info(), old->workers()));
-    SyncSessionJob job(old_job.purpose, TimeTicks::Now() + length,
-                       make_linked_ptr(s), false, old_job.config_params,
-                       FROM_HERE);
-    wait_interval_->pending_configure_job.reset(new SyncSessionJob(job));
+    DCHECK(!old_job->config_params().ready_task.is_null());
+    wait_interval_->pending_configure_job = new_job.get();
   } else {
     // We are not in configuration mode. So wait_interval's pending job
     // should be null.
-    DCHECK(wait_interval_->pending_configure_job.get() == NULL);
+    DCHECK(wait_interval_->pending_configure_job == NULL);
+    DCHECK(!pending_nudge_);
+    pending_nudge_ = new_job.get();
+  }
 
-    // TODO(lipalani) - handle clear user data.
-    InitOrCoalescePendingJob(old_job);
-  }
-  RestartWaiting();
+  RestartWaiting(new_job.Pass());
 }
 
 void SyncSchedulerImpl::RequestStop(const base::Closure& callback) {
   syncer_->RequestEarlyExit();  // Safe to call from any thread.
   DCHECK(weak_handle_this_.IsInitialized());
   SDVLOG(3) << "Posting StopImpl";
   weak_handle_this_.Call(FROM_HERE,
                          &SyncSchedulerImpl::StopImpl,
                          callback);
 }
 
 void SyncSchedulerImpl::StopImpl(const base::Closure& callback) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "StopImpl called";
 
   // Kill any in-flight method calls.
   weak_ptr_factory_.InvalidateWeakPtrs();
   wait_interval_.reset();
   poll_timer_.Stop();
   if (started_) {
     started_ = false;
   }
   if (!callback.is_null())
     callback.Run();
 }
 
-void SyncSchedulerImpl::DoCanaryJob() {
+void SyncSchedulerImpl::DoCanaryJob(scoped_ptr<SyncSessionJob> to_be_canary) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   SDVLOG(2) << "Do canary job";
-  DoPendingJobIfPossible(true);
+
+  // Only set canary privileges here, when we are about to run the job. This
+  // avoids confusion in managing canary bits during scheduling, when you
+  // consider that mode switches (e.g., to config) can "pre-empt" a NUDGE that
+  // was scheduled as canary, and send it to an "unscheduled" state.
+  to_be_canary->GrantCanaryPrivilege();
+
+  if (to_be_canary->purpose() == SyncSessionJob::NUDGE) {
+    // TODO(tim): We should be able to remove this...
+    scoped_ptr<SyncSession> temp = CreateSyncSession(
+        to_be_canary->session()->source()).Pass();
+    // The routing info might have been changed since we cached the
+    // pending nudge. Update it by coalescing to the latest.
+    to_be_canary->mutable_session()->Coalesce(*(temp));
+  }
+  DoSyncSessionJob(to_be_canary.Pass());
 }
 
-void SyncSchedulerImpl::DoPendingJobIfPossible(bool is_canary_job) {
+scoped_ptr<SyncSessionJob> SyncSchedulerImpl::TakePendingJobForCurrentMode() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
-  SyncSessionJob* job_to_execute = NULL;
+  // 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.get()) {
+      && wait_interval_->pending_configure_job) {
     SDVLOG(2) << "Found pending configure job";
-    job_to_execute = wait_interval_->pending_configure_job.get();
-  } else if (mode_ == NORMAL_MODE && pending_nudge_.get()) {
+    candidate =
+        wait_interval_->pending_configure_job->CloneAndAbandon().Pass();
+    wait_interval_->pending_configure_job = candidate.get();
+  } else if (mode_ == NORMAL_MODE && pending_nudge_) {
     SDVLOG(2) << "Found pending nudge job";
-
-    scoped_ptr<SyncSession> session(CreateSyncSession(
-        pending_nudge_->session->source()));
-
-    // Also the routing info might have been changed since we cached the
-    // pending nudge. Update it by coalescing to the latest.
-    pending_nudge_->session->Coalesce(*(session.get()));
-    // The pending nudge would be cleared in the DoSyncSessionJob function.
-    job_to_execute = pending_nudge_.get();
+    candidate = pending_nudge_->CloneAndAbandon();
+    pending_nudge_ = candidate.get();
+    unscheduled_nudge_storage_.reset();
   }
-
-  if (job_to_execute != NULL) {
-    SDVLOG(2) << "Executing pending job";
-    SyncSessionJob copy = *job_to_execute;
-    copy.is_canary_job = is_canary_job;
-    DoSyncSessionJob(copy);
-  }
+  return candidate.Pass();
 }
 
-SyncSession* SyncSchedulerImpl::CreateSyncSession(
+scoped_ptr<SyncSession> SyncSchedulerImpl::CreateSyncSession(
     const SyncSourceInfo& source) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DVLOG(2) << "Creating sync session with routes "
            << ModelSafeRoutingInfoToString(session_context_->routing_info());
 
   SyncSourceInfo info(source);
-  SyncSession* session(new SyncSession(session_context_, this, info,
+  return scoped_ptr<SyncSession>(new SyncSession(session_context_, this, info,
       session_context_->routing_info(), session_context_->workers()));
-
-  return session;
 }
 
 void SyncSchedulerImpl::PollTimerCallback() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   ModelSafeRoutingInfo r;
   ModelTypeInvalidationMap invalidation_map =
       ModelSafeRoutingInfoToInvalidationMap(r, std::string());
   SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, invalidation_map);
-  SyncSession* s = CreateSyncSession(info);
-
-  SyncSessionJob job(SyncSessionJob::POLL, TimeTicks::Now(),
-                     make_linked_ptr(s),
-                     false,
-                     ConfigurationParams(),
-                     FROM_HERE);
-
-  ScheduleSyncSessionJob(job);
+  scoped_ptr<SyncSession> s(CreateSyncSession(info));
+  scoped_ptr<SyncSessionJob> job(new SyncSessionJob(SyncSessionJob::POLL,
+                                                    TimeTicks::Now(),
+                                                    s.Pass(),
+                                                    ConfigurationParams(),
+                                                    FROM_HERE));
+  ScheduleSyncSessionJob(job.Pass());
 }
 
-void SyncSchedulerImpl::Unthrottle() {
+void SyncSchedulerImpl::Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
-  SDVLOG(2) << "Unthrottled.";
-  DoCanaryJob();
+  SDVLOG(2) << "Unthrottled " << (to_be_canary.get() ? "with " : "without ")
+            << "canary.";
+  if (to_be_canary.get())
+    DoCanaryJob(to_be_canary.Pass());
+
+  // TODO(tim): The way DecideOnJob works today, canary privileges aren't
+  // enough to bypass a THROTTLED wait interval, which would suggest we need
+  // to reset before DoCanaryJob (though trusting canary in DecideOnJob is
+  // probably the "right" thing to do). Bug 154216.
   wait_interval_.reset();
 }
 
 void SyncSchedulerImpl::Notify(SyncEngineEvent::EventCause cause) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   session_context_->NotifyListeners(SyncEngineEvent(cause));
 }
 
 bool SyncSchedulerImpl::IsBackingOff() const {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   return wait_interval_.get() && wait_interval_->mode ==
       WaitInterval::EXPONENTIAL_BACKOFF;
 }
 
 void SyncSchedulerImpl::OnSilencedUntil(
     const base::TimeTicks& silenced_until) {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   wait_interval_.reset(new WaitInterval(WaitInterval::THROTTLED,
                                         silenced_until - TimeTicks::Now()));
-  wait_interval_->timer.Start(FROM_HERE, wait_interval_->length, this,
-      &SyncSchedulerImpl::Unthrottle);
 }
 
 bool SyncSchedulerImpl::IsSyncingCurrentlySilenced() {
   DCHECK_EQ(MessageLoop::current(), sync_loop_);
   return wait_interval_.get() && wait_interval_->mode ==
       WaitInterval::THROTTLED;
 }
 
 void SyncSchedulerImpl::OnReceivedShortPollIntervalUpdate(
     const base::TimeDelta& new_interval) {
@@ skipping 53 matching lines @@
 
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // namespace syncer