[Sync] Sync types never recovers from throttle

components/sync/engine_impl/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 "components/sync/engine_impl/sync_scheduler_impl.h"
 
 #include <algorithm>
 #include <cstring>
 #include <utility>
 
@@ skipping 131 matching lines @@
       mode_(CONFIGURATION_MODE),
       delay_provider_(delay_provider),
       syncer_(syncer),
       cycle_context_(context),
       next_sync_cycle_job_priority_(NORMAL_PRIORITY),
       ignore_auth_credentials_(ignore_auth_credentials),
       weak_ptr_factory_(this) {}
 
 SyncSchedulerImpl::~SyncSchedulerImpl() {
   DCHECK(CalledOnValidThread());
+
   Stop();
 }
 
 void SyncSchedulerImpl::OnCredentialsUpdated() {
   DCHECK(CalledOnValidThread());
 
   if (HttpResponse::SYNC_AUTH_ERROR ==
       cycle_context_->connection_manager()->server_status()) {
     OnServerConnectionErrorFixed();
   }
 }
 
 void SyncSchedulerImpl::OnConnectionStatusChange() {
+  DCHECK(CalledOnValidThread());
+
   if (HttpResponse::CONNECTION_UNAVAILABLE ==
       cycle_context_->connection_manager()->server_status()) {
     // Optimistically assume that the connection is fixed and try
     // connecting.
     OnServerConnectionErrorFixed();
   }
 }
 
 void SyncSchedulerImpl::OnServerConnectionErrorFixed() {
   // 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 TryCanaryJob 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.
   TryCanaryJob();
 }
 
 void SyncSchedulerImpl::Start(Mode mode, base::Time last_poll_time) {
   DCHECK(CalledOnValidThread());
+
   std::string thread_name = base::PlatformThread::GetName();
   if (thread_name.empty())
     thread_name = "<Main thread>";
   SDVLOG(2) << "Start called from thread " << thread_name << " with mode "
             << GetModeString(mode);
   if (!started_) {
     started_ = true;
     SendInitialSnapshot();
   }
 
@@ skipping 32 matching lines @@
 ModelTypeSet SyncSchedulerImpl::GetEnabledAndUnblockedTypes() {
   ModelTypeSet enabled_types = cycle_context_->GetEnabledTypes();
   ModelTypeSet enabled_protocol_types =
       Intersection(ProtocolTypes(), enabled_types);
   ModelTypeSet blocked_types = nudge_tracker_.GetBlockedTypes();
   return Difference(enabled_protocol_types, blocked_types);
 }
 
 void SyncSchedulerImpl::SendInitialSnapshot() {
   DCHECK(CalledOnValidThread());
+
   std::unique_ptr<SyncCycle> dummy(SyncCycle::Build(cycle_context_, this));
   SyncCycleEvent event(SyncCycleEvent::STATUS_CHANGED);
   event.snapshot = dummy->TakeSnapshot();
   for (auto& observer : *cycle_context_->listeners())
     observer.OnSyncCycleEvent(event);
 }
 
 void SyncSchedulerImpl::ScheduleConfiguration(
     const ConfigurationParams& params) {
   DCHECK(CalledOnValidThread());
@@ skipping 16 matching lines @@
     params.ready_task.Run();
   }
 }
 
 void SyncSchedulerImpl::ScheduleClearServerData(const ClearParams& params) {
   DCHECK(CalledOnValidThread());
   DCHECK_EQ(CLEAR_SERVER_DATA_MODE, mode_);
   DCHECK(!pending_configure_params_);
   DCHECK(!params.report_success_task.is_null());
   CHECK(started_) << "Scheduler must be running to clear.";
+
   pending_clear_params_ = base::MakeUnique<ClearParams>(params);
   TrySyncCycleJob();
 }
 
 bool SyncSchedulerImpl::CanRunJobNow(JobPriority priority) {
   DCHECK(CalledOnValidThread());
+
   if (IsCurrentlyThrottled()) {
     SDVLOG(1) << "Unable to run a job because we're throttled.";
     return false;
   }
 
   if (IsBackingOff() && priority != CANARY_PRIORITY) {
     SDVLOG(1) << "Unable to run a job because we're backing off.";
     return false;
   }
 
@@ skipping 91 matching lines @@
         << "Dropping nudge, scheduler is not running.";
     return;
   }
 
   SDVLOG_LOC(nudge_location, 2) << "In ScheduleNudgeImpl with delay "
                                 << delay.InMilliseconds() << " ms";
 
   if (!CanRunNudgeJobNow(NORMAL_PRIORITY))
     return;
 
-  TimeTicks incoming_run_time = TimeTicks::Now() + delay;
-  if (pending_wakeup_timer_.IsRunning() &&
-      (pending_wakeup_timer_.desired_run_time() < incoming_run_time)) {
+  if (!IsEarlierThanCurrentPendingJob(delay)) {
     // Old job arrives sooner than this one.  Don't reschedule it.
     return;
   }
 
   // Either there is no existing nudge in flight or the incoming nudge should be
   // made to arrive first (preempt) the existing nudge.  We reschedule in either
   // case.
   SDVLOG_LOC(nudge_location, 2) << "Scheduling a nudge with "
                                 << delay.InMilliseconds() << " ms delay";
   pending_wakeup_timer_.Start(
       nudge_location, delay, base::Bind(&SyncSchedulerImpl::PerformDelayedNudge,
                                         weak_ptr_factory_.GetWeakPtr()));
 }
 
 const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
   switch (mode) {
     ENUM_CASE(CONFIGURATION_MODE);
     ENUM_CASE(CLEAR_SERVER_DATA_MODE);
     ENUM_CASE(NORMAL_MODE);
   }
   return "";
 }
 
 void SyncSchedulerImpl::SetDefaultNudgeDelay(TimeDelta delay_ms) {
   DCHECK(CalledOnValidThread());
+
   nudge_tracker_.SetDefaultNudgeDelay(delay_ms);
 }
 
 void SyncSchedulerImpl::DoNudgeSyncCycleJob(JobPriority priority) {
   DCHECK(CalledOnValidThread());
   DCHECK(CanRunNudgeJobNow(priority));
 
   DVLOG(2) << "Will run normal mode sync cycle with types "
            << ModelTypeSetToString(GetEnabledAndUnblockedTypes());
   std::unique_ptr<SyncCycle> cycle(SyncCycle::Build(cycle_context_, this));
@@ skipping 92 matching lines @@
     SDVLOG(2) << "Sync cycle failed.  Will back off for "
               << wait_interval_->length.InMilliseconds() << "ms.";
   } else {
     // Increase our backoff interval and schedule another retry.
     TimeDelta length = delay_provider_->GetDelay(wait_interval_->length);
     wait_interval_ = base::MakeUnique<WaitInterval>(
         WaitInterval::EXPONENTIAL_BACKOFF, length);
     SDVLOG(2) << "Sync cycle failed.  Will back off for "
               << wait_interval_->length.InMilliseconds() << "ms.";
   }
-  RestartWaiting();
 }
 
 void SyncSchedulerImpl::DoPollSyncCycleJob() {
   SDVLOG(2) << "Polling with types "
             << ModelTypeSetToString(GetEnabledAndUnblockedTypes());
   std::unique_ptr<SyncCycle> cycle(SyncCycle::Build(cycle_context_, this));
   bool success =
       syncer_->PollSyncShare(GetEnabledAndUnblockedTypes(), cycle.get());
 
   // Only restart the timer if the poll succeeded. Otherwise rely on normal
   // failure handling to retry with backoff.
   if (success) {
     AdjustPolling(FORCE_RESET);
     HandleSuccess();
   } else {
     HandleFailure(cycle->status_controller().model_neutral_state());
   }
 }
 
 void SyncSchedulerImpl::UpdateNudgeTimeRecords(ModelTypeSet types) {
   DCHECK(CalledOnValidThread());
+
   TimeTicks now = TimeTicks::Now();
   // Update timing information for how often datatypes are triggering nudges.
   for (ModelTypeSet::Iterator iter = types.First(); iter.Good(); iter.Inc()) {
     TimeTicks previous = last_local_nudges_by_model_type_[iter.Get()];
     last_local_nudges_by_model_type_[iter.Get()] = 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.Get());
 #undef PER_DATA_TYPE_MACRO
   }
 }
 
 TimeDelta SyncSchedulerImpl::GetPollInterval() {
   return (!cycle_context_->notifications_enabled() ||
           !cycle_context_->ShouldFetchUpdatesBeforeCommit())
              ? syncer_short_poll_interval_seconds_
              : syncer_long_poll_interval_seconds_;
 }
 
 void SyncSchedulerImpl::AdjustPolling(PollAdjustType type) {
   DCHECK(CalledOnValidThread());
+
   if (!started_)
     return;
 
   TimeDelta poll_interval = GetPollInterval();
   TimeDelta poll_delay = poll_interval;
   const TimeTicks now = TimeTicks::Now();
 
   if (type == UPDATE_INTERVAL) {
     if (!last_poll_reset_.is_null()) {
       // Override the delay based on the last successful poll time (if it was
@@ skipping 23 matching lines @@
   SDVLOG(1) << "Updating polling delay to " << poll_delay.InMinutes()
             << " minutes.";
 
   // Adjust poll rate. Start will reset the timer if it was already running.
   poll_timer_.Start(FROM_HERE, poll_delay, this,
                     &SyncSchedulerImpl::PollTimerCallback);
 }
 
 void SyncSchedulerImpl::RestartWaiting() {
   if (wait_interval_.get()) {
-    // Global throttling or backoff
+    // Global throttling or backoff.
+    if (!IsEarlierThanCurrentPendingJob(wait_interval_->length)) {
+      // Since RestartWaiting() is called in TrySyncCycleJobImpl(), we should

[skym, 2017/05/01 17:59:14]

I don't understand what this comment is saying, or why we should be checking
IsEarlierThanCurrentPendingJob for global blocking reasons.

[Gang Wu, 2017/05/01 23:01:41]

Done.

+      // not overwrite existing the scheduled job, otherwise the unblock jobs
+      // will be keep push away and never be run.
+      return;
+    }
     NotifyRetryTime(base::Time::Now() + wait_interval_->length);
     SDVLOG(2) << "Starting WaitInterval timer of length "
               << wait_interval_->length.InMilliseconds() << "ms.";
     if (wait_interval_->mode == WaitInterval::THROTTLED) {
       pending_wakeup_timer_.Start(FROM_HERE, wait_interval_->length,
                                   base::Bind(&SyncSchedulerImpl::Unthrottle,
                                              weak_ptr_factory_.GetWeakPtr()));
     } else {
       pending_wakeup_timer_.Start(
           FROM_HERE, wait_interval_->length,
           base::Bind(&SyncSchedulerImpl::ExponentialBackoffRetry,
                      weak_ptr_factory_.GetWeakPtr()));
     }
   } else if (nudge_tracker_.IsAnyTypeBlocked()) {
     // Per-datatype throttled or backed off.
     TimeDelta time_until_next_unblock =
         nudge_tracker_.GetTimeUntilNextUnblock();
+    if (!IsEarlierThanCurrentPendingJob(time_until_next_unblock)) {
+      return;
+    }
     pending_wakeup_timer_.Start(FROM_HERE, time_until_next_unblock,
                                 base::Bind(&SyncSchedulerImpl::OnTypesUnblocked,
                                            weak_ptr_factory_.GetWeakPtr()));
   }
 }
 
 void SyncSchedulerImpl::Stop() {
   DCHECK(CalledOnValidThread());
   SDVLOG(2) << "Stop called";
 
@@ skipping 19 matching lines @@
 
 void SyncSchedulerImpl::TrySyncCycleJob() {
   // Post call to TrySyncCycleJobImpl on current thread. Later request for
   // access token will be here.
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE, base::Bind(&SyncSchedulerImpl::TrySyncCycleJobImpl,
                             weak_ptr_factory_.GetWeakPtr()));
 }
 
 void SyncSchedulerImpl::TrySyncCycleJobImpl() {
+  DCHECK(CalledOnValidThread());
+
   JobPriority priority = next_sync_cycle_job_priority_;
   next_sync_cycle_job_priority_ = NORMAL_PRIORITY;
 
   nudge_tracker_.SetSyncCycleStartTime(TimeTicks::Now());
 
-  DCHECK(CalledOnValidThread());
   if (mode_ == CONFIGURATION_MODE) {
     if (pending_configure_params_) {
       SDVLOG(2) << "Found pending configure job";
       DoConfigurationSyncCycleJob(priority);
     }
   } else if (mode_ == CLEAR_SERVER_DATA_MODE) {
     if (pending_clear_params_) {
       DoClearServerDataSyncCycleJob(priority);
     }
   } else if (CanRunNudgeJobNow(priority)) {
     if (nudge_tracker_.IsSyncRequired()) {
       SDVLOG(2) << "Found pending nudge job";
       DoNudgeSyncCycleJob(priority);
     } else if (((TimeTicks::Now() - last_poll_reset_) >= GetPollInterval())) {
       SDVLOG(2) << "Found pending poll";
       DoPollSyncCycleJob();
     }
   } else {
     // We must be in an error state. Transitioning out of each of these
     // error states should trigger a canary job.
     DCHECK(IsCurrentlyThrottled() || IsBackingOff() ||
            cycle_context_->connection_manager()->HasInvalidAuthToken());
   }
 
-  if (IsBackingOff() && !pending_wakeup_timer_.IsRunning()) {
-    // If we succeeded, our wait interval would have been cleared.  If it hasn't
-    // been cleared, then we should increase our backoff interval and schedule
-    // another retry.
-    TimeDelta length = delay_provider_->GetDelay(wait_interval_->length);
-    wait_interval_ = base::MakeUnique<WaitInterval>(
-        WaitInterval::EXPONENTIAL_BACKOFF, length);
-    SDVLOG(2) << "Sync cycle failed.  Will back off for "
-              << wait_interval_->length.InMilliseconds() << "ms.";
-    RestartWaiting();
-  }
+  RestartWaiting();

[skym, 2017/05/01 17:59:14]

We should explain somewhere what the overall strategy here is. How we ensure we
call RestartWaiting() at the correct times and don't drop various throttling
scenarios like we used to. Maybe this explanation belongs in the .h file
definition of RestartWaiting()?

[Gang Wu, 2017/05/01 23:01:41]

Done.

 }
 
 void SyncSchedulerImpl::PollTimerCallback() {
   DCHECK(CalledOnValidThread());
   CHECK(!syncer_->IsSyncing());
 
   TrySyncCycleJob();
 }
 
 void SyncSchedulerImpl::RetryTimerCallback() {
@@ skipping 12 matching lines @@
   // We treat this as a 'canary' in the sense that it was originally scheduled
   // to run some time ago, failed, and we now want to retry, versus a job that
   // was just created (e.g via ScheduleNudgeImpl). The main implication is
   // that we're careful to update routing info (etc) with such potentially
   // stale canary jobs.
   TryCanaryJob();
 }
 
 void SyncSchedulerImpl::OnTypesUnblocked() {
   DCHECK(CalledOnValidThread());
+
   nudge_tracker_.UpdateTypeThrottlingAndBackoffState();
   NotifyBlockedTypesChanged(nudge_tracker_.GetBlockedTypes());
 
-  RestartWaiting();
-
   // Maybe this is a good time to run a nudge job.  Let's try it.
+  // If not a good time, reschedule a new run.
   if (nudge_tracker_.IsSyncRequired() && CanRunNudgeJobNow(NORMAL_PRIORITY))
     TrySyncCycleJob();
+  else
+    RestartWaiting();
 }
 
 void SyncSchedulerImpl::PerformDelayedNudge() {
   // Circumstances may have changed since we scheduled this delayed nudge.
   // We must check to see if it's OK to run the job before we do so.
   if (CanRunNudgeJobNow(NORMAL_PRIORITY))
     TrySyncCycleJob();
 
   // We're not responsible for setting up any retries here.  The functions that
   // first put us into a state that prevents successful sync cycles (eg. global
@@ skipping 26 matching lines @@
   }
 
   for (auto& observer : *cycle_context_->listeners()) {
     observer.OnThrottledTypesChanged(throttled_types);
     observer.OnBackedOffTypesChanged(backed_off_types);
   }
 }
 
 bool SyncSchedulerImpl::IsBackingOff() const {
   DCHECK(CalledOnValidThread());
+
   return wait_interval_.get() &&
          wait_interval_->mode == WaitInterval::EXPONENTIAL_BACKOFF;
 }
 
 void SyncSchedulerImpl::OnThrottled(const TimeDelta& throttle_duration) {
   DCHECK(CalledOnValidThread());
+
   wait_interval_ = base::MakeUnique<WaitInterval>(WaitInterval::THROTTLED,
                                                   throttle_duration);
   NotifyRetryTime(base::Time::Now() + wait_interval_->length);
 
   for (auto& observer : *cycle_context_->listeners()) {
     observer.OnThrottledTypesChanged(ModelTypeSet::All());
   }
 }
 
 void SyncSchedulerImpl::OnTypesThrottled(ModelTypeSet types,
                                          const TimeDelta& throttle_duration) {
+  DCHECK(CalledOnValidThread());
+
   TimeTicks now = TimeTicks::Now();
 
   SDVLOG(1) << "Throttling " << ModelTypeSetToString(types) << " for "
             << throttle_duration.InMinutes() << " minutes.";
 
   nudge_tracker_.SetTypesThrottledUntil(types, throttle_duration, now);
-  RestartWaiting();
   NotifyBlockedTypesChanged(nudge_tracker_.GetBlockedTypes());
 }
 
 void SyncSchedulerImpl::OnTypesBackedOff(ModelTypeSet types) {
+  DCHECK(CalledOnValidThread());
+
   TimeTicks now = TimeTicks::Now();
 
   for (ModelTypeSet::Iterator type = types.First(); type.Good(); type.Inc()) {
     TimeDelta last_backoff_time =
         TimeDelta::FromSeconds(kInitialBackoffRetrySeconds);
     if (nudge_tracker_.GetTypeBlockingMode(type.Get()) ==
         WaitInterval::EXPONENTIAL_BACKOFF_RETRYING) {
       last_backoff_time = nudge_tracker_.GetTypeLastBackoffInterval(type.Get());
     }
 
     TimeDelta length = delay_provider_->GetDelay(last_backoff_time);
     nudge_tracker_.SetTypeBackedOff(type.Get(), length, now);
     SDVLOG(1) << "Backing off " << ModelTypeToString(type.Get()) << " for "
               << length.InSeconds() << " second.";
   }
-  RestartWaiting();
   NotifyBlockedTypesChanged(nudge_tracker_.GetBlockedTypes());
 }
 
 bool SyncSchedulerImpl::IsCurrentlyThrottled() {
   DCHECK(CalledOnValidThread());
+
   return wait_interval_.get() &&
          wait_interval_->mode == WaitInterval::THROTTLED;
 }
 
 void SyncSchedulerImpl::OnReceivedShortPollIntervalUpdate(
     const TimeDelta& new_interval) {
   DCHECK(CalledOnValidThread());
+
   if (new_interval == syncer_short_poll_interval_seconds_)
     return;
   SDVLOG(1) << "Updating short poll interval to " << new_interval.InMinutes()
             << " minutes.";
   syncer_short_poll_interval_seconds_ = new_interval;
   AdjustPolling(UPDATE_INTERVAL);
 }
 
 void SyncSchedulerImpl::OnReceivedLongPollIntervalUpdate(
     const TimeDelta& new_interval) {
   DCHECK(CalledOnValidThread());
+
   if (new_interval == syncer_long_poll_interval_seconds_)
     return;
   SDVLOG(1) << "Updating long poll interval to " << new_interval.InMinutes()
             << " minutes.";
   syncer_long_poll_interval_seconds_ = new_interval;
   AdjustPolling(UPDATE_INTERVAL);
 }
 
 void SyncSchedulerImpl::OnReceivedCustomNudgeDelays(
     const std::map<ModelType, TimeDelta>& nudge_delays) {
   DCHECK(CalledOnValidThread());
+
   nudge_tracker_.OnReceivedCustomNudgeDelays(nudge_delays);
 }
 
 void SyncSchedulerImpl::OnReceivedClientInvalidationHintBufferSize(int size) {
+  DCHECK(CalledOnValidThread());
+
   if (size > 0)
     nudge_tracker_.SetHintBufferSize(size);
   else
     NOTREACHED() << "Hint buffer size should be > 0.";
 }
 
 void SyncSchedulerImpl::OnSyncProtocolError(
     const SyncProtocolError& sync_protocol_error) {
   DCHECK(CalledOnValidThread());
+
   if (ShouldRequestEarlyExit(sync_protocol_error)) {
     SDVLOG(2) << "Sync Scheduler requesting early exit.";
     Stop();
   }
   if (IsActionableError(sync_protocol_error)) {
     SDVLOG(2) << "OnActionableError";
     for (auto& observer : *cycle_context_->listeners())
       observer.OnActionableError(sync_protocol_error);
   }
 }
 
 void SyncSchedulerImpl::OnReceivedGuRetryDelay(const TimeDelta& delay) {
+  DCHECK(CalledOnValidThread());
+
   nudge_tracker_.SetNextRetryTime(TimeTicks::Now() + delay);
   retry_timer_.Start(FROM_HERE, delay, this,
                      &SyncSchedulerImpl::RetryTimerCallback);
 }
 
 void SyncSchedulerImpl::OnReceivedMigrationRequest(ModelTypeSet types) {
+  DCHECK(CalledOnValidThread());
+
   for (auto& observer : *cycle_context_->listeners())
     observer.OnMigrationRequested(types);
 }
 
 void SyncSchedulerImpl::SetNotificationsEnabled(bool notifications_enabled) {
   DCHECK(CalledOnValidThread());
+
   cycle_context_->set_notifications_enabled(notifications_enabled);
   if (notifications_enabled)
     nudge_tracker_.OnInvalidationsEnabled();
   else
     nudge_tracker_.OnInvalidationsDisabled();
 }
 
+bool SyncSchedulerImpl::IsEarlierThanCurrentPendingJob(const TimeDelta& delay) {
+  TimeTicks incoming_run_time = TimeTicks::Now() + delay;
+  if (pending_wakeup_timer_.IsRunning() &&
+      (pending_wakeup_timer_.desired_run_time() < incoming_run_time)) {
+    // Old job arrives sooner than this one.
+    return false;
+  }
+  return true;
+}
+
 #undef SDVLOG_LOC
 
 #undef SDVLOG
 
 #undef SLOG
 
 #undef ENUM_CASE
 
 }  // namespace syncer