sync: Make nudge + config jobs reliable in SyncerThread2

chrome/browser/sync/engine/syncer_thread2.cc

 // Copyright (c) 2011 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 "chrome/browser/sync/engine/syncer_thread2.h"
 
 #include <algorithm>
 
 #include "base/rand_util.h"
 #include "chrome/browser/sync/engine/syncer.h"
 
 using base::TimeDelta;
 using base::TimeTicks;
 
 namespace browser_sync {
 
 using sessions::SyncSession;
 using sessions::SyncSessionSnapshot;
 using sessions::SyncSourceInfo;
 using syncable::ModelTypePayloadMap;
 using syncable::ModelTypeBitSet;
 using sync_pb::GetUpdatesCallerInfo;
 
 namespace s3 {
 
+struct SyncerThread::SyncSessionJob {
+  SyncSessionJobPurpose purpose;
+  base::TimeTicks scheduled_start;
+  linked_ptr<sessions::SyncSession> session;
+  bool is_canary_job;
+
+  // This is the location the nudge came from. used for debugging purpose.
+  // In case of multiple nudges getting coalesced this stores the first nudge
+  // that came in.
+  tracked_objects::Location nudge_location;
+};
+
 struct SyncerThread::WaitInterval {
   enum Mode {
     // A wait interval whose duration has been affected by exponential
     // backoff.
     // EXPONENTIAL_BACKOFF intervals are nudge-rate limited to 1 per interval.
     EXPONENTIAL_BACKOFF,
     // A server-initiated throttled interval.  We do not allow any syncing
     // during such an interval.
     THROTTLED,
   };
   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<SyncerThread> timer;
+  scoped_ptr<SyncSessionJob> pending_job;
   WaitInterval(Mode mode, base::TimeDelta length);
 };
 
-struct SyncerThread::SyncSessionJob {
-  SyncSessionJobPurpose purpose;
-  base::TimeTicks scheduled_start;
-  linked_ptr<sessions::SyncSession> session;
-
-  // This is the location the nudge came from. used for debugging purpose.
-  // In case of multiple nudges getting coalesced this stores the first nudge
-  // that came in.
-  tracked_objects::Location nudge_location;
-};
-
 SyncerThread::DelayProvider::DelayProvider() {}
 SyncerThread::DelayProvider::~DelayProvider() {}
 
 TimeDelta SyncerThread::DelayProvider::GetDelay(
     const base::TimeDelta& last_delay) {
   return SyncerThread::GetRecommendedDelay(last_delay);
 }
 
+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_UNKNOWN:
+      return GetUpdatesCallerInfo::UNKNOWN;
+    default:
+      NOTREACHED();
+      return GetUpdatesCallerInfo::UNKNOWN;
+  }
+}
+
 SyncerThread::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
     : mode(mode), had_nudge(false), length(length) { }
 
 SyncerThread::SyncerThread(sessions::SyncSessionContext* context,
                            Syncer* syncer)
     : thread_("SyncEngine_SyncerThread"),
       syncer_short_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultShortPollIntervalSeconds)),
       syncer_long_poll_interval_seconds_(
           TimeDelta::FromSeconds(kDefaultLongPollIntervalSeconds)),
       mode_(NORMAL_MODE),
       server_connection_ok_(false),
       delay_provider_(new DelayProvider()),
       syncer_(syncer),
-      session_context_(context) {
+      session_context_(context),
+      saved_nudge_(false),
+      saved_source_(GetUpdatesCallerInfo::UNKNOWN) {
 }
 
 SyncerThread::~SyncerThread() {
   DCHECK(!thread_.IsRunning());
 }
 
 void SyncerThread::CheckServerConnectionManagerStatus(
     HttpResponse::ServerConnectionCode code) {
   // Note, be careful when adding cases here because if the SyncerThread
   // thinks there is no valid connection as determined by this method, it
@@ skipping 43 matching lines @@
 
 void SyncerThread::StartImpl(Mode mode,
                              linked_ptr<ModeChangeCallback> callback) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
   DCHECK(!session_context_->account_name().empty());
   DCHECK(syncer_.get());
   mode_ = mode;
   AdjustPolling(NULL);  // Will kick start poll timer if needed.
   if (callback.get())
     callback->Run();
+
+  if (mode_ == NORMAL_MODE && saved_nudge_ == true) {
+    syncable::ModelTypePayloadMap map;
+    ScheduleNudgeImpl(TimeDelta::FromSeconds(0),
+                      saved_source_, map, false,
+                      FROM_HERE);
+  }
 }
 
-bool SyncerThread::ShouldRunJob(SyncSessionJobPurpose purpose,
-    const TimeTicks& scheduled_start) {
-  DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
+SyncerThread::JobProcessDecision SyncerThread::DecideOnJobWhileBackingOff(
+    const SyncSessionJob& job) {
 
-  // Check wait interval.
-  if (wait_interval_.get()) {
-    // TODO(tim): Consider different handling for CLEAR_USER_DATA (i.e. permit
-    // when throttled).
-    if (wait_interval_->mode == WaitInterval::THROTTLED)
-      return false;
+  DCHECK(wait_interval_.get());
+  DCHECK(job.purpose != CLEAR_USER_DATA);
 
-    DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
-    if ((purpose != NUDGE) || wait_interval_->had_nudge)
-      return false;
+  if (job.purpose == POLL) {
+    return DROP;
   }
 
-  // Mode / purpose contract (See 'Mode' enum in header). Don't run jobs that
-  // were intended for a normal sync if we are in configuration mode, and vice
-  // versa.
-  switch (mode_) {
-    case CONFIGURATION_MODE:
-      if (purpose != CONFIGURATION)
-        return false;
-      break;
-    case NORMAL_MODE:
-      if (purpose == CONFIGURATION)
-        return false;
-      break;
-    default:
-      NOTREACHED() << "Unknown SyncerThread Mode: " << mode_;
-      return false;
+  DCHECK(job.purpose == NUDGE || job.purpose == CONFIGURATION);
+  if (wait_interval_->mode == WaitInterval::THROTTLED) {
+    return SAVE;
   }
 
-  // Continuation NUDGE tasks have priority over POLLs because they are the
-  // only tasks that trigger exponential backoff, so this prevents them from
-  // being starved from running (e.g. due to a very, very low poll interval,
-  // such as 0ms). It's rare that this would ever matter in practice.
-  if (purpose == POLL && (pending_nudge_.get() &&
-      pending_nudge_->session->source().updates_source ==
-          GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION)) {
+  DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
+  if (job.purpose == NUDGE) {
+    if (mode_ == CONFIGURATION) {
+      return SAVE;
+    }
+
+    // If we already had one nudge then just drop this nudge. We will retry
+    // later when the timer runs out.
+    if (wait_interval_->had_nudge) {
+      return DROP;
+    } else {
+      // If this is our first nudge during backoff or if our timer ran out
+      // let us retry once more.
+      return CONTINUE;
+    }
+  } else {
+    // This is a config job.
+    // If our timer ran out then continue.

[tim (not reviewing), 2011/04/08 03:37:46]

this can go on above line.

[lipalani1, 2011/04/08 18:40:23]

Done.

+    if (job.is_canary_job) {
+      return CONTINUE;
+    } else {
+      return SAVE;
+    }
+  }
+}
+
+SyncerThread::JobProcessDecision SyncerThread::ProcessJob(

[tim (not reviewing), 2011/04/08 03:37:46]

Perhaps MakeJobDecision or DecideOnJob
Process is misleading.

[lipalani1, 2011/04/08 18:40:23]

Done.

+    const SyncSessionJob& job) {
+  if (job.purpose == CLEAR_USER_DATA) {

[tim (not reviewing), 2011/04/08 03:37:46]

this file in general follows the "no { } on single line if statement", and the
style position is to follow convention as no style is required.  So you should
audit your ifs for this.  (Or change the rest of the file technically but please
don't do that :) )

[lipalani1, 2011/04/08 18:40:23]

Left as such if the if had an else clause. Otherwise changed it.
On 2011/04/08 03:37:46, timsteele wrote:
Done.

+    return CONTINUE;
+  }
+
+  if (wait_interval_.get()) {
+    return DecideOnJobWhileBackingOff(job);

[tim (not reviewing), 2011/04/08 03:37:46]

this method could be named better e.g. not 'BackingfOff' but InWaitInterval 

[lipalani1, 2011/04/08 18:40:23]

Done.

+  }
+
+  if (mode_ == CONFIGURATION_MODE) {
+    if (job.purpose == NUDGE) {
+      return SAVE;
+    } else if (job.purpose == CONFIGURATION) {
+      return CONTINUE;
+    } else {
+      return DROP;
+    }
+  }
+
+  // We are in normal mode.
+  DCHECK(mode_ == NORMAL_MODE);
+  DCHECK(job.purpose != CONFIGURATION);
+
+  // Freshness condition
+  if (job.scheduled_start < last_sync_session_end_time_) {
+    return DROP;
+  }
+
+  if (server_connection_ok_) {
+    return CONTINUE;
+  }
+
+  job.purpose == NUDGE ? SAVE : DROP;
+}
+
+bool SyncerThread::ShouldRunJob(const SyncSessionJob& job) {
+  JobProcessDecision decision = ProcessJob(job);
+  if (decision == DROP) {

[tim (not reviewing), 2011/04/08 03:37:46]

I'd collapse these 2 if statements into

if (decision != SAVE)
  return decision == CONTINUE;

then the DCHECK_EQ below isn't even needed as it's implicit.

[lipalani1, 2011/04/08 18:40:23]

I am OK with changing it. But readability wise this looks more readable. And
also lesser regressions. 
case in point: if we introduce a 4th state then the above line has to change.
However let me know.
On 2011/04/08 03:37:46, timsteele wrote:

     return false;
   }
 
-  // Freshness condition.
-  if (purpose == NUDGE &&
-      (scheduled_start < last_sync_session_end_time_)) {
-    return false;
+  if (decision == CONTINUE) {
+    return true;
   }
 
-  return server_connection_ok_;
-}
+  DCHECK(job.purpose == NUDGE || job.purpose == CONFIGURATION);
+  DCHECK_EQ(decision, SAVE);
+  if (job.purpose == NUDGE) {
+    saved_nudge_ = true;
+    saved_source_ = job.session->source().updates_source;
+  } else {
+    DCHECK(wait_interval_.get());
+    DCHECK(mode_ == CONFIGURATION_MODE);

[tim (not reviewing), 2011/04/08 03:37:46]

how do we know this??

[lipalani1, 2011/04/08 18:40:23]

If the purpose is config then we should be in config mode. otherwise it is a
bug.
On 2011/04/08 03:37:46, timsteele wrote:

 
-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_UNKNOWN:
-      return GetUpdatesCallerInfo::UNKNOWN;
-    default:
-      NOTREACHED();
-      return GetUpdatesCallerInfo::UNKNOWN;
+    SyncSession* old = job.session.get();
+    SyncSession* s(new SyncSession(session_context_.get(), this,
+        old->source(), old->routing_info(), old->workers()));
+    SyncSessionJob new_job = {job.purpose, TimeTicks::Now(),
+                          make_linked_ptr(s), false, job.nudge_location};
+    wait_interval_->pending_job.reset(new SyncSessionJob(new_job));
   }
+  return false;
 }
 
 // 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;
 };
@@ skipping 11 matching lines @@
     NudgeSource source, const ModelTypeBitSet& types,
     const tracked_objects::Location& nudge_location) {
   if (!thread_.IsRunning()) {
     NOTREACHED();
     return;
   }
 
   ModelTypePayloadMap types_with_payloads =
       syncable::ModelTypePayloadMapFromBitSet(types, std::string());
   thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-      this, &SyncerThread::ScheduleNudgeImpl, delay, source,
-      types_with_payloads, nudge_location));
+      this, &SyncerThread::ScheduleNudgeImpl, delay,
+      GetUpdatesFromNudgeSource(source), types_with_payloads, false,
+      nudge_location));
 }
 
 void SyncerThread::ScheduleNudgeWithPayloads(const TimeDelta& delay,
     NudgeSource source, const ModelTypePayloadMap& types_with_payloads,
     const tracked_objects::Location& nudge_location) {
   if (!thread_.IsRunning()) {
     NOTREACHED();
     return;
   }
 
   thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-      this, &SyncerThread::ScheduleNudgeImpl, delay, source,
-      types_with_payloads, nudge_location));
+      this, &SyncerThread::ScheduleNudgeImpl, delay,
+      GetUpdatesFromNudgeSource(source), types_with_payloads, false,
+      nudge_location));
 }
 
 void SyncerThread::ScheduleClearUserDataImpl() {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
   SyncSession* session = new SyncSession(session_context_.get(), this,
       SyncSourceInfo(), ModelSafeRoutingInfo(),
       std::vector<ModelSafeWorker*>());
   ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), CLEAR_USER_DATA, session,
                          FROM_HERE);
 }
 
 void SyncerThread::ScheduleNudgeImpl(const TimeDelta& delay,
-    NudgeSource source, const ModelTypePayloadMap& types_with_payloads,
-    const tracked_objects::Location& nudge_location) {
+    GetUpdatesCallerInfo::GetUpdatesSource source,
+    const ModelTypePayloadMap& types_with_payloads,
+    bool is_canary_job, const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
-  TimeTicks rough_start = TimeTicks::Now() + delay;
-  if (!ShouldRunJob(NUDGE, rough_start)) {
-    LOG(WARNING) << "Dropping nudge at scheduling time, source = "
-                 << source;
-    return;
-  }
 
   // Note we currently nudge for all types regardless of the ones incurring
   // the nudge.  Doing different would throw off some syncer commands like
   // CleanupDisabledTypes.  We may want to change this in the future.
   ModelSafeRoutingInfo routes;
   std::vector<ModelSafeWorker*> workers;
   session_context_->registrar()->GetModelSafeRoutingInfo(&routes);
   session_context_->registrar()->GetWorkers(&workers);
-  SyncSourceInfo info(GetUpdatesFromNudgeSource(source),
-                      types_with_payloads);
+  SyncSourceInfo info(source, types_with_payloads);
 
-  scoped_ptr<SyncSession> session(new SyncSession(
+  SyncSession* session(new SyncSession(
       session_context_.get(), this, info, routes, workers));
 
+  SyncSessionJob job = {NUDGE, TimeTicks::Now() + delay,
+                        make_linked_ptr(session), is_canary_job,
+                        nudge_location};
+
+  session = NULL;

[tim (not reviewing), 2011/04/08 03:37:46]

what's this for?

[lipalani1, 2011/04/08 18:40:23]

So that nobody reuses session. I can put it in a scope if that makes it easier.
On 2011/04/08 03:37:46, timsteele wrote:

+  if (!ShouldRunJob(job)) {
+    return;
+  }
+
   if (pending_nudge_.get()) {
     if (IsBackingOff() && delay > TimeDelta::FromSeconds(1))
       return;
 
-    pending_nudge_->session->Coalesce(*session.get());
+    pending_nudge_->session->Coalesce(*(job.session.get()));
 
     if (!IsBackingOff()) {
       return;
     } else {
       // Re-schedule the current pending nudge.
       SyncSession* s = pending_nudge_->session.get();
-      session.reset(new SyncSession(s->context(), s->delegate(), s->source(),
-          s->routing_info(), s->workers()));
+      job.session.reset(new SyncSession(s->context(), s->delegate(),
+          s->source(), s->routing_info(), s->workers()));
       pending_nudge_.reset();
     }
   }
-  ScheduleSyncSessionJob(delay, NUDGE, session.release(), nudge_location);
+
+  //  TODO(lipalani) - pass the job itself to ScheduleSyncSessionJob.
+  ScheduleSyncSessionJob(delay, NUDGE, job.session.release(), nudge_location);
 }
 
 // Helper to extract the routing info and workers corresponding to types in
 // |types| from |registrar|.
 void GetModelSafeParamsForTypes(const ModelTypeBitSet& types,
     ModelSafeWorkerRegistrar* registrar, ModelSafeRoutingInfo* routes,
     std::vector<ModelSafeWorker*>* workers) {
   ModelSafeRoutingInfo r_tmp;
   std::vector<ModelSafeWorker*> w_tmp;
   registrar->GetModelSafeRoutingInfo(&r_tmp);
@@ skipping 27 matching lines @@
     NOTREACHED();
     return;
   }
 
   ModelSafeRoutingInfo routes;
   std::vector<ModelSafeWorker*> workers;
   GetModelSafeParamsForTypes(types, session_context_->registrar(),
                              &routes, &workers);
 
   thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-      this, &SyncerThread::ScheduleConfigImpl, routes, workers));
+      this, &SyncerThread::ScheduleConfigImpl, routes, workers,
+      GetUpdatesCallerInfo::FIRST_UPDATE));
 }
 
 void SyncerThread::ScheduleConfigImpl(const ModelSafeRoutingInfo& routing_info,
-    const std::vector<ModelSafeWorker*>& workers) {
+    const std::vector<ModelSafeWorker*>& workers,
+    const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
 
   // TODO(tim): config-specific GetUpdatesCallerInfo value?
   SyncSession* session = new SyncSession(session_context_.get(), this,
-      SyncSourceInfo(GetUpdatesCallerInfo::FIRST_UPDATE,
+      SyncSourceInfo(source,
           syncable::ModelTypePayloadMapFromRoutingInfo(
               routing_info, std::string())),
       routing_info, workers);
   ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), CONFIGURATION, session,
-                         FROM_HERE);
+    FROM_HERE);
 }
 
 void SyncerThread::ScheduleSyncSessionJob(const base::TimeDelta& delay,
     SyncSessionJobPurpose purpose, sessions::SyncSession* session,
     const tracked_objects::Location& nudge_location) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
 
   SyncSessionJob job = {purpose, TimeTicks::Now() + delay,
-                        make_linked_ptr(session), nudge_location};
+                        make_linked_ptr(session), false, nudge_location};
   if (purpose == NUDGE) {
     DCHECK(!pending_nudge_.get() || pending_nudge_->session.get() == session);
     pending_nudge_.reset(new SyncSessionJob(job));
   }
   MessageLoop::current()->PostDelayedTask(FROM_HERE, NewRunnableMethod(this,
       &SyncerThread::DoSyncSessionJob, job),
       delay.InMilliseconds());
 }
 
 void SyncerThread::SetSyncerStepsForPurpose(SyncSessionJobPurpose purpose,
@@ skipping 11 matching lines @@
     case POLL:
       *start = SYNCER_BEGIN;
       return;
     default:
       NOTREACHED();
   }
 }
 
 void SyncerThread::DoSyncSessionJob(const SyncSessionJob& job) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
-  if (!ShouldRunJob(job.purpose, job.scheduled_start)) {
+  if (!ShouldRunJob(job)) {
     LOG(WARNING) << "Dropping nudge at DoSyncSessionJob, source = "
         << job.session->source().updates_source;
     return;
   }
 
   if (job.purpose == NUDGE) {
     DCHECK(pending_nudge_.get());
     if (pending_nudge_->session != job.session)
       return;  // Another nudge must have been scheduled in in the meantime.
     pending_nudge_.reset();
+    saved_nudge_ = false;
   }
 
   SyncerStep begin(SYNCER_BEGIN);
   SyncerStep end(SYNCER_END);
   SetSyncerStepsForPurpose(job.purpose, &begin, &end);
 
   bool has_more_to_sync = true;
-  while (ShouldRunJob(job.purpose, job.scheduled_start) && has_more_to_sync) {
+  while (ShouldRunJob(job) && has_more_to_sync) {
     VLOG(1) << "SyncerThread: Calling SyncShare.";
     // Synchronously perform the sync session from this thread.
     syncer_->SyncShare(job.session.get(), begin, end);
     has_more_to_sync = job.session->HasMoreToSync();
     if (has_more_to_sync)
       job.session->ResetTransientState();
   }
   VLOG(1) << "SyncerThread: Done SyncShare looping.";
   FinishSyncSessionJob(job);
 }
@@ skipping 71 matching lines @@
     HandleConsecutiveContinuationError(old_job);
   } else if (IsBackingOff()) {
     // We weren't continuing but we're in backoff; must have been a nudge.
     DCHECK_EQ(NUDGE, old_job.purpose);
     DCHECK(!wait_interval_->had_nudge);
     wait_interval_->had_nudge = true;
     wait_interval_->timer.Reset();
   } else {
     // We weren't continuing and we aren't in backoff.  Schedule a normal
     // continuation.
-    ScheduleNudgeImpl(TimeDelta::FromSeconds(0), NUDGE_SOURCE_CONTINUATION,
-                      old_job.session->source().types, FROM_HERE);
+    if (old_job.purpose == NUDGE) {
+      ScheduleNudgeImpl(TimeDelta::FromSeconds(0),
+                        GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION),
+                        old_job.session->source().types, false, FROM_HERE);
+    } else if (old_job.purpose == CONFIGURATION) {
+      ScheduleConfigImpl(old_job.session->routing_info(),
+          old_job.session->workers(),
+          GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION));
+    } // Drop the rest.
   }
 }
 
 void SyncerThread::AdjustPolling(const SyncSessionJob* old_job) {
   DCHECK(thread_.IsRunning());
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
 
   TimeDelta poll  = (!session_context_->notifications_enabled()) ?
       syncer_short_poll_interval_seconds_ :
       syncer_long_poll_interval_seconds_;
@@ skipping 15 matching lines @@
     const SyncSessionJob& old_job) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
   DCHECK(!IsBackingOff() || !wait_interval_->timer.IsRunning());
   SyncSession* old = old_job.session.get();
   SyncSession* s(new SyncSession(session_context_.get(), this,
       old->source(), old->routing_info(), old->workers()));
   TimeDelta length = delay_provider_->GetDelay(
       IsBackingOff() ? wait_interval_->length : TimeDelta::FromSeconds(1));
   wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
                                         length));
-  SyncSessionJob job = {NUDGE, TimeTicks::Now() + length,
-                        make_linked_ptr(s), FROM_HERE};
-  pending_nudge_.reset(new SyncSessionJob(job));
+  if (old_job.purpose == CONFIGURATION) {
+    SyncSessionJob job = {old_job.purpose, TimeTicks::Now() + length,
+                        make_linked_ptr(s), false, FROM_HERE};
+    wait_interval_->pending_job.reset(new SyncSessionJob(job));
+  } else {
+    // No matter what type of job it is.(nudge or poll), we are going to treat
+    // it as nudge when doing exponential back off retries.
+    saved_nudge_ = true;
+    saved_source_ = old_job.session->source().updates_source;
+    wait_interval_->pending_job.reset();
+  }
   wait_interval_->timer.Start(length, this, &SyncerThread::DoCanaryJob);
 }
 
 // static
 TimeDelta SyncerThread::GetRecommendedDelay(const TimeDelta& last_delay) {
   if (last_delay.InSeconds() >= kMaxBackoffSeconds)
     return TimeDelta::FromSeconds(kMaxBackoffSeconds);
 
   // This calculates approx. base_delay_seconds * 2 +/- base_delay_seconds / 2
   int64 backoff_s =
@@ skipping 14 matching lines @@
   return TimeDelta::FromSeconds(backoff_s);
 }
 
 void SyncerThread::Stop() {
   syncer_->RequestEarlyExit();  // Safe to call from any thread.
   session_context_->connection_manager()->RemoveListener(this);
   thread_.Stop();
 }
 
 void SyncerThread::DoCanaryJob() {
-  DCHECK(pending_nudge_.get());
   wait_interval_->had_nudge = false;
-  SyncSessionJob copy = *pending_nudge_;
-  DoSyncSessionJob(copy);
+
+  // We should have one of 2 things. Otherwise we shouldnt be running the timer.
+  DCHECK(wait_interval_->pending_job.get() || saved_nudge_);
+  if (mode_ == CONFIGURATION_MODE) {
+    if (wait_interval_->pending_job.get()) {
+      SyncSessionJob copy = *(wait_interval_->pending_job);
+      copy.is_canary_job = true;
+      DoSyncSessionJob(copy);
+    }
+  } else {
+    if (saved_nudge_) {
+      syncable::ModelTypePayloadMap map;
+      ScheduleNudgeImpl(TimeDelta::FromSeconds(0), saved_source_, map, true,
+                        FROM_HERE);
+    }
+  }
 }
 
 void SyncerThread::PollTimerCallback() {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
   ModelSafeRoutingInfo r;
   std::vector<ModelSafeWorker*> w;
   session_context_->registrar()->GetModelSafeRoutingInfo(&r);
   session_context_->registrar()->GetWorkers(&w);
   ModelTypePayloadMap types_with_payloads =
       syncable::ModelTypePayloadMapFromRoutingInfo(r, std::string());
   SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, types_with_payloads);
   SyncSession* s = new SyncSession(session_context_.get(), this, info, r, w);
   ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), POLL, s, FROM_HERE);
 }
 
 void SyncerThread::Unthrottle() {
   DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
+  DoCanaryJob();
   wait_interval_.reset();
 }
 
 void SyncerThread::Notify(SyncEngineEvent::EventCause cause) {
   DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
   session_context_->NotifyListeners(SyncEngineEvent(cause));
 }
 
 bool SyncerThread::IsBackingOff() const {
   return wait_interval_.get() && wait_interval_->mode ==
@@ skipping 35 matching lines @@
       &SyncerThread::CheckServerConnectionManagerStatus,
       event.connection_code));
 }
 
 void SyncerThread::set_notifications_enabled(bool notifications_enabled) {
   session_context_->set_notifications_enabled(notifications_enabled);
 }
 
 }  // s3
 }  // browser_sync