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;
+
+  // 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);
 }
 
 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) {
 }
 
 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), NUDGE_SOURCE_NOTIFICATION,
+                      map, FROM_HERE);
+  }
 }
 
-bool SyncerThread::ShouldRunJob(SyncSessionJobPurpose purpose,
-    const TimeTicks& scheduled_start) {
-  DCHECK_EQ(MessageLoop::current(), thread_.message_loop());
-
-  // Check wait interval.
+SyncerThread::JobProcessDecision SyncerThread::ShouldRunNudgeJob(
+  const base::TimeTicks& scheduled_start) {
   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;
+    // Means we are either in throttled or exponential back off.
+    // However it is also possible that we are in exponential back off
+    // and we are retrying.(In that case the timer ran out but wait_interval
+    // is not yet cleared.) wait_interval_.timer_running is false;
+    if (wait_interval_->mode == WaitInterval::THROTTLED) {
+      return SAVE;
+    }
 
     DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
-    if ((purpose != NUDGE) || wait_interval_->had_nudge)
-      return false;
+    if (mode_ == CONFIGURATION) {
+      return SAVE;
+    }
+    // The mode is normal. We already had a nudge. No point retrying.
+    if (wait_interval_->had_nudge) {
+      return DROP;
+    }
+    // Either this is the first nudge or we are in exponential backoff
+    // and we are trying because our timer ran out.(In either case had_nudge
+    // is false)
+    return CONTINUE;
   }
 
-  // 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;
+  // We are not in any kind of backoff.
+  if (mode_ == CONFIGURATION) {
+    return SAVE;
+  }
+  // Freshness condition
+  if (scheduled_start < last_sync_session_end_time_) {
+    return DROP;
   }
 
-  // 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)) {
+  return CONTINUE;
+}
+
+// Note: We should never be dropping a config request.
+SyncerThread::JobProcessDecision SyncerThread::ShouldRunConfigureJob() {

[tim (not reviewing), 2011/04/07 06:13:07]

wouldn't it result in a substantial amount less code to have these in one method
as before, due to similarities?  there appears to be a fair bit of duplication
both among and within the split-up methods.  it was easier on the mind to see it
all in one place too. I think one of the main improvements of the new impl was
that there was one and only one place to go look to see the rules for running /
dropping jobs... we should try and preserve that.

[lipalani1, 2011/04/07 18:35:45]

The original function did its job well in terms of code reuse and readability. 

Now the new logic is little more dense.
This arrangement uncomplicates the logic wrt to future changes. For ex: If I
make a chance in Nudge it wont cause a regression in configure or default.  
Currently the only common code is checking for throttling(in nudge and
configure. Even default does not share it. Interestingly the common code most of
the time is shared between only 2 modes and the third is different!). Even in
future if we want a certain condition to have common behavior across all the
modes it makes to think of them seperately but write a common function which all
the modes call.(Currently I dont see such a refactor possible. But if you see
one let me know)
On 2011/04/07 06:13:07, timsteele wrote:

+  if (wait_interval_.get()) {
+    if (wait_interval_->mode == WaitInterval::THROTTLED) {
+      return SAVE;
+    }
+
+    DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
+    DCHECK(mode_ == CONFIGURATION_MODE);
+    if (wait_interval_->timer.IsRunning() == true) {
+      return SAVE;
+    }
+
+    return CONTINUE;
+  }
+
+  // We are not in any kind of back off.
+  DCHECK(mode_ == CONFIGURATION_MODE);
+  return CONTINUE;
+}
+
+SyncerThread::JobProcessDecision SyncerThread::ShouldRunJobDefaultImpl() {
+  if (wait_interval_.get()) {
+    return DROP;
+  }
+
+  if (mode_ == CONFIGURATION) {
+    return DROP;
+  }
+
+  return CONTINUE;
+}
+
+SyncerThread::JobProcessDecision SyncerThread::ShouldRunJob(
+    SyncSessionJobPurpose purpose,
+    const TimeTicks& scheduled_start) {
+  if (purpose == NUDGE) {
+    return ShouldRunNudgeJob(scheduled_start);
+  } else if (purpose == CONFIGURATION) {
+    return ShouldRunConfigureJob();
+  } else {
+    return ShouldRunJobDefaultImpl();
+  }
+}
+
+bool SyncerThread::ProcessJob(const SyncSessionJob& job) {

[tim (not reviewing), 2011/04/07 06:13:07]

It's pretty confusing that we now have both Process and ShouldRun...
in every place that we now call processJob, it would be far more readable to
call ShouldRun. e.g.  while(ShouldRunJob has way less cognitive overhead than
while(ProcessJob

It also causes a lot of extra / duplicated code to create jobs in multiple
locations.  I tried quite hard to keep the job creation routine limited to
ScheduleSyncSessionJob (params).  That's why that method exists.

[lipalani1, 2011/04/07 18:35:45]

Will do.

Regarding job creation it was also done in HandleConsecutiveError case. This
code is similar to that.. Let me know what you think. The other option I see is
just storing the datatypes. So job creation still lies with
schedulesyncsessionjob. I am not passionate about one way or the other. The
reason I picked job was it had all the data like purpose and modeltypes in one
structure. If we passed in model types then we have to store the purpose as well
seperately and also the source.
On 2011/04/07 06:13:07, timsteele wrote:

+  JobProcessDecision decision = ShouldRunJob(job.purpose, job.scheduled_start);
+  if (decision == DROP) {
     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);
+  if (job.purpose == NUDGE) {
+    saved_nudge_ = true;
+  } else {
+    DCHECK(wait_interval_.get());
+    DCHECK(mode_ == CONFIGURATION_MODE);
+
+    // Save off the nudge if we had one already stored.
+    if (wait_interval_->pending_job.get()) {
+      if (wait_interval_->pending_job->purpose == NUDGE) {
+        saved_nudge_ = true;
+      }
+    }
+    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), job.nudge_location};
+    wait_interval_->pending_job.reset(new SyncSessionJob(new_job));
+  }
+  return false;
 }
 
 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:
@@ skipping 58 matching lines @@
       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) {
   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);
 
-  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), nudge_location};
+
+  session = NULL;
+  if (!ProcessJob(job)) {

[tim (not reviewing), 2011/04/07 06:13:07]

We only had to move this down because we must pass in a job, right?  
And the job is only needed when purpose != NUDGE in ProcessJob... 
We should strive to limit the places where we have to create a session and a
job.

[lipalani1, 2011/04/07 18:35:45]

The other option as I mentioned was to pass only the datatypes. 
On 2011/04/07 06:13:07, timsteele wrote:

+    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.

[tim (not reviewing), 2011/04/07 06:13:07]

if we do that then there is no point to having ScheduleSyncSessionJob.
It exists so that we only have to create a job in one place.

[lipalani1, 2011/04/07 18:35:45]

Same as above.
On 2011/04/07 06:13:07, timsteele wrote:

+  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};
   if (purpose == NUDGE) {
@@ skipping 20 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 (!ProcessJob(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 (ProcessJob(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), NUDGE_SOURCE_CONTINUATION,
+                        old_job.session->source().types, 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,
+  SyncSessionJob job = {old_job.purpose, TimeTicks::Now() + length,
                         make_linked_ptr(s), FROM_HERE};
-  pending_nudge_.reset(new SyncSessionJob(job));
+  wait_interval_->pending_job.reset(new SyncSessionJob(job));
   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_;
+  SyncSessionJob copy = *(wait_interval_->pending_job);
+  if (copy.purpose == CONFIGURATION) {
+    DCHECK(mode_ == CONFIGURATION_MODE);
+  } else {
+    pending_nudge_.reset(new SyncSessionJob(copy));
+  }
   DoSyncSessionJob(copy);
 }
 
 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