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;
+
+  // Configure jobs are saved only when backing off or throttling. So we
+  // expose the pointer here.
+  scoped_ptr<SyncSessionJob> pending_configure_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)),
@@ skipping 14 matching lines @@
   // thinks there is no valid connection as determined by this method, it
   // will drop out of *all* forward progress sync loops (it won't poll and it
   // will queue up Talk notifications but not actually call SyncShare) until
   // some external action causes a ServerConnectionManager to broadcast that
   // a valid connection has been re-established.
   if (HttpResponse::CONNECTION_UNAVAILABLE == code ||
       HttpResponse::SYNC_AUTH_ERROR == code) {
     server_connection_ok_ = false;
   } else if (HttpResponse::SERVER_CONNECTION_OK == code) {
     server_connection_ok_ = true;
+
+    // run the pending nudges if any.
+    if (mode_ == NORMAL_MODE && pending_nudge_.get()) {
+      pending_nudge_.reset();
+      syncable::ModelTypePayloadMap map;
+      ScheduleNudgeImpl(TimeDelta::FromSeconds(0),
+                        saved_source_, map, false,
+                        FROM_HERE);
+    } else if (wait_interval_.get() &&
+               wait_interval_->pending_configure_job.get()) {
+      DoCanaryJob();
+
+      // Note this means we will restart the wait interval if this job fails.
+      // Which is probably ok since we just got connection.
+      wait_interval_.reset();
+    }
+
   }
 }
 
 void SyncerThread::Start(Mode mode, ModeChangeCallback* callback) {
   if (!thread_.IsRunning()) {
     if (!thread_.Start()) {
       NOTREACHED() << "Unable to start SyncerThread.";
       return;
     }
     WatchConnectionManager();
@@ skipping 24 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 && pending_nudge_.get()) {
+    pending_nudge_.reset();
+    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.

[tim (not reviewing), 2011/04/08 20:57:41]

if the timer ran out.. is that still valid now that we have is_canary?

[lipalani1, 2011/04/08 23:45:59]

Done.

+      return CONTINUE;
+    }
+  } else {
+    // This is a config job.
+    // If our timer ran out then continue.
+    if (job.is_canary_job) {

[tim (not reviewing), 2011/04/08 20:57:41]

I want to use ternary form here.
e.g. return is_canary_job ? CONTINUE : SAVE

I think it's healthier overall for the this file if this function / group of
functions deciding on jobs is as crisp and succinct as it can be right now.

I'd argue this for the above too, since we could collapse 11 lines into three
(see my earlier code review response). And put the comments together, above the
code. It's a bit intimidating to read this function as it's sprawling out...

[lipalani1, 2011/04/08 23:45:59]

Done.

+      return CONTINUE;
+    } else {
+      return SAVE;
+    }
+  }
+}
+
+SyncerThread::JobProcessDecision SyncerThread::ProcessJob(
+    const SyncSessionJob& job) {
+  if (job.purpose == CLEAR_USER_DATA) {
+    return CONTINUE;
+  }
+
+  if (wait_interval_.get()) {
+    return DecideOnJobWhileBackingOff(job);
+  }
+
+  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;
+  }
+
+  return job.purpose == NUDGE ? SAVE : DROP;
+}
+
+void SyncerThread::CoalescePendingNudge(const SyncSessionJob& job) {

[tim (not reviewing), 2011/04/08 20:57:41]

i still see places in the code where Coalesce is called directly.

this also does more than coalesce, it actually initializes pending_nudge_.  

[lipalani1, 2011/04/08 23:45:59]

Renamed it to updatependingjob.
On 2011/04/08 20:57:41, timsteele wrote:
Done.

+  DCHECK(job.purpose != CONFIGURATION);
+  if (pending_nudge_.get() == NULL) {
+    SyncSession* s = job.session.get();
+    scoped_ptr<SyncSession> session(new SyncSession(s->context(),
+        s->delegate(), s->source(), s->routing_info(), s->workers()));
+
+    SyncSessionJob new_job = {NUDGE, job.scheduled_start,
+        make_linked_ptr(session.release()), false, job.nudge_location};
+    pending_nudge_.reset(new SyncSessionJob(new_job));
+
+    return;
+  }
+
+  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 SyncerThread::ShouldRunJob(const SyncSessionJob& job) {
+  JobProcessDecision decision = ProcessJob(job);
+  if (decision == DROP) {
     return false;
   }
 
-  // Freshness condition.
-  if (purpose == NUDGE &&
-      (scheduled_start < last_sync_session_end_time_)) {
-    return false;
+  if (decision == CONTINUE) {
+    return true;

[tim (not reviewing), 2011/04/08 20:57:41]

I still think we should combine these ifs as I suggested earlier.  There is no
4th state.  I want to keep this as crisp as possible.

This function becomes just

if (decision != SAVE)
  return decision == CONTINUE
else
  SaveJob()

which is close to being inline-able

[lipalani1, 2011/04/08 23:45:59]

Done.

   }
 
-  return server_connection_ok_;
+  DCHECK(job.purpose == NUDGE || job.purpose == CONFIGURATION);
+  DCHECK_EQ(decision, SAVE);
+  SaveJob(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:
-      return GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION;
-    case NUDGE_SOURCE_UNKNOWN:
-      return GetUpdatesCallerInfo::UNKNOWN;
-    default:
-      NOTREACHED();
-      return GetUpdatesCallerInfo::UNKNOWN;
+void SyncerThread::SaveJob(const SyncSessionJob& job) {
+  DCHECK(job.purpose != CLEAR_USER_DATA);
+  if (job.purpose == NUDGE || job.purpose == POLL) {
+    CoalescePendingNudge(job);
+  } else {
+    DCHECK(wait_interval_.get());
+    DCHECK(mode_ == CONFIGURATION_MODE);
+
+    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_configure_job.reset(new SyncSessionJob(new_job));
   }
 }
 
 // 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 12 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;
+  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)) {
-    LOG(WARNING) << "Dropping nudge at DoSyncSessionJob, source = "
-        << job.session->source().updates_source;
+  if (!ShouldRunJob(job)) {
     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();
   }
 
   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();
+    if (IsSyncingCurrentlySilenced()) {
+      DCHECK_NE(job.purpose, CLEAR_USER_DATA);
+      SaveJob(job);
+    }
   }
   VLOG(1) << "SyncerThread: Done SyncShare looping.";
   FinishSyncSessionJob(job);
 }
 
 void SyncerThread::UpdateCarryoverSessionState(const SyncSessionJob& old_job) {
   if (old_job.purpose == CONFIGURATION) {
     // Whatever types were part of a configuration task will have had updates
     // downloaded.  For that reason, we make sure they get recorded in the
     // event that they get disabled at a later time.
@@ skipping 65 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 == CONFIGURATION) {
+      ScheduleConfigImpl(old_job.session->routing_info(),
+          old_job.session->workers(),
+          GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION));
+    } else  {
+      // For all other purposes(nudge and poll) we schedule a retry nudge.
+      ScheduleNudgeImpl(TimeDelta::FromSeconds(0),
+                        GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION),
+                        old_job.session->source().types, false, FROM_HERE);
+    }
   }
 }
 
 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_configure_job.reset(new SyncSessionJob(job));
+  } else {
+    // We are not in configuration mode. So wait_interval's pending job
+    // should be null.
+    DCHECK(wait_interval_->pending_configure_job.get() == NULL);
+
+    // No matter what type of job it is.(nudge or poll, it cannot be config or
+    // or clear user data) we are going to treat

[tim (not reviewing), 2011/04/08 20:57:41]

text formatting not proper here.
also why can't it be clear user data

Why do we want to coalesce now whereas before we'd always replace?

[lipalani1, 2011/04/08 23:45:59]

added a todo for clearuserdata.
On 2011/04/08 20:57:41, timsteele wrote:
Done.

+    // it as nudge when doing exponential back off retries.
+    CoalescePendingNudge(old_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;

[tim (not reviewing), 2011/04/08 20:57:41]

this should not be required anymore.

[lipalani1, 2011/04/08 23:45:59]

This is the only place it gets cleared out. We want the semantics of allowing
one nudge per backoff interval. And it gets cleared when the backoff interval
runs out which is here.
On 2011/04/08 20:57:41, timsteele wrote:

-  SyncSessionJob copy = *pending_nudge_;
+
+  // We should have one of 2 things. Otherwise we shouldnt be running the timer.
+  DCHECK(wait_interval_->pending_configure_job.get() || pending_nudge_.get());
+  if (mode_ == CONFIGURATION_MODE) {
+    if (wait_interval_->pending_configure_job.get()) {
+      ScheduleCanaryJob(wait_interval_->pending_configure_job.get());
+      wait_interval_->pending_configure_job.reset();
+    }
+  } else {
+    if (pending_nudge_.get()) {
+      ScheduleCanaryJob(pending_nudge_.get());
+    }
+  }
+}
+
+void SyncerThread::ScheduleCanaryJob(SyncSessionJob* job) {

[tim (not reviewing), 2011/04/08 20:57:41]

this function is exactly what DoCanaryJob did before so I think that name is
better suited than Schedule. Seems like it'd be cleaner to get rid of the two
functions and just keep one called DoCanaryJob.  
It would have the following:

 SyncSessionJob* job = NULL;
 if (mode_ == CONFIGURATION_MODE && wait_interval_.get() &&
     wait_interval_->pending_configure_job.get()) {
   job = wait_interval_->pending_configure_job.get();
 } else if (pending_nudge_.get()) {
   job = pending_nudge_.get();
 }

 if (!job)
   return;

 SyncSessionJob copy = *job;
 copy.is_canary_job = true;
 DoSyncSessionJob(copy);

[lipalani1, 2011/04/08 23:45:59]

Sorry this was part of the unfinished refactoring I mentioned. Actually this
function is needed by took your feedback and renamed it to
ExecuteJobByMakingACopy(hope the name is ok). This is called from few more
places(like when switching from config mode to normal mode etc). look at the
latest review.
On 2011/04/08 20:57:41, timsteele wrote:

+  DCHECK(job);
+  SyncSessionJob copy = *job;
+  copy.is_canary_job = true;
   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