Call batch/incremental fetch with more strict conditions

chrome/browser/sync_file_system/drive_file_sync_service.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 "chrome/browser/sync_file_system/drive_file_sync_service.h"
 
 #include <algorithm>
 #include <string>
 #include <utility>
 
@@ skipping 58 matching lines @@
     return;
   }
 
   if (!file_util::Delete(file_path, true))
     LOG(ERROR) << "Leaked temporary file for Sync FileSystem: "
                << file_path.value();
 }
 
 void EmptyStatusCallback(fileapi::SyncStatusCode code) {}
 
-void MarkFetchingChangesCompleted(bool* is_fetching_changes) {
-  *is_fetching_changes = false;
-}
-
 void DidRemoveOrigin(const GURL& origin, fileapi::SyncStatusCode status) {
   // TODO(calvinlo): Disable syncing if status not ok (http://crbug.com/171611).
   DCHECK_EQ(fileapi::SYNC_STATUS_OK, status);
   LOG(WARNING) << "Remove origin failed for: " << origin.spec()
                << " status=" << status;
 }
 
 fileapi::FileChange CreateFileChange(bool is_deleted) {
   if (is_deleted) {
     return fileapi::FileChange(fileapi::FileChange::FILE_CHANGE_DELETE,
@@ skipping 11 matching lines @@
  public:
   explicit TaskToken(const base::WeakPtr<DriveFileSyncService>& sync_service)
       : sync_service_(sync_service),
         task_type_(TASK_TYPE_NONE) {
   }
 
   void ResetTask(const tracked_objects::Location& location) {
     location_ = location;
     task_type_ = TASK_TYPE_NONE;
     description_.clear();
-    if (!completion_callback_.is_null())
-      completion_callback_.Run();
-    completion_callback_.Reset();
   }
 
   void UpdateTask(const tracked_objects::Location& location,
                   TaskType task_type,
                   const std::string& description) {
     location_ = location;
     task_type_ = task_type;
     description_ = description;
 
     DVLOG(2) << "Token updated: " << description_
              << " " << location_.ToString();
   }
 
   const tracked_objects::Location& location() const { return location_; }
   TaskType task_type() const { return task_type_; }
   const std::string& description() const { return description_; }
   std::string done_description() const { return description_ + " done"; }
 
-  void set_completion_callback(const base::Closure& callback) {
-    completion_callback_ = callback;
-  }
-
-  const base::Closure& completion_callback() {
-    return completion_callback_;
-  }
-
   ~TaskToken() {
     // All task on DriveFileSyncService must hold TaskToken instance to ensure
     // no other tasks are running. Also, as soon as a task finishes to work,
     // it must return the token to DriveFileSyncService.
     // Destroying a token with valid |sync_service_| indicates the token was
     // dropped by a task without returning.
     if (sync_service_) {
       LOG(ERROR) << "Unexpected TaskToken deletion from: "
                  << location_.ToString() << " while: " << description_;
     }
     DCHECK(!sync_service_);
   }
 
  private:
   base::WeakPtr<DriveFileSyncService> sync_service_;
   tracked_objects::Location location_;
   TaskType task_type_;
   std::string description_;
-  base::Closure completion_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(TaskToken);
 };
 
 void DriveFileSyncService::OnInvalidatorStateChange(
     syncer::InvalidatorState state) {
   SetPushNotificationEnabled(state);
 }
 
 void DriveFileSyncService::SetPushNotificationEnabled(
     syncer::InvalidatorState state) {
   push_notification_enabled_ = (state == syncer::INVALIDATIONS_ENABLED);
   if (!push_notification_enabled_)
     return;
 
   // Push notifications are enabled so reset polling timer.
   UpdatePollingDelay(kPollingDelaySecondsWithNotification);
 }
 
 void DriveFileSyncService::OnIncomingInvalidation(
     const syncer::ObjectIdInvalidationMap& invalidation_map) {
   DCHECK(push_notification_enabled_);
   DCHECK_EQ(1U, invalidation_map.size());
   const invalidation::ObjectId object_id(
       ipc::invalidation::ObjectSource::COSMO_CHANGELOG,
       kDriveInvalidationObjectId);
   DCHECK_EQ(1U, invalidation_map.count(object_id));
 
-  FetchChangesForIncrementalSync();
+  may_have_unfetched_changes_ = true;
+  MaybeStartFetchChanges();
 }
 
 struct DriveFileSyncService::ProcessRemoteChangeParam {
   scoped_ptr<TaskToken> token;
   RemoteChangeProcessor* processor;
   RemoteChange remote_change;
   fileapi::SyncFileCallback callback;
 
   DriveMetadata drive_metadata;
   bool metadata_updated;
@@ skipping 80 matching lines @@
 
 DriveFileSyncService::DriveFileSyncService(Profile* profile)
     : profile_(profile),
       last_operation_status_(fileapi::SYNC_STATUS_OK),
       state_(REMOTE_SERVICE_OK),
       sync_enabled_(true),
       largest_fetched_changestamp_(0),
       push_notification_registered_(false),
       push_notification_enabled_(false),
       polling_delay_seconds_(kMinimumPollingDelaySeconds),
-      is_fetching_changes_(false),
+      may_have_unfetched_changes_(true),
       weak_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
   temporary_file_dir_ =
       profile->GetPath().Append(kSyncFileSystemDir).Append(kTempDirName);
   token_.reset(new TaskToken(AsWeakPtr()));
 
   sync_client_.reset(new DriveFileSyncClient(profile));
   sync_client_->AddObserver(this);
 
   metadata_store_.reset(new DriveMetadataStore(
       profile->GetPath().Append(kSyncFileSystemDir),
@@ skipping 107 matching lines @@
     return;
   }
 
   OriginToChangesMap::iterator found = origin_to_changes_map_.find(origin);
   if (found != origin_to_changes_map_.end()) {
     for (PathToChangeMap::iterator itr = found->second.begin();
          itr != found->second.end(); ++itr)
       pending_changes_.erase(itr->second.position_in_queue);
     origin_to_changes_map_.erase(found);
   }
+  pending_batch_sync_origins_.erase(origin);
 
   metadata_store_->RemoveOrigin(origin, base::Bind(
       &DriveFileSyncService::DidRemoveOriginOnMetadataStore,
       AsWeakPtr(), base::Passed(&token), callback));
 }
 
 void DriveFileSyncService::ProcessRemoteChange(
     RemoteChangeProcessor* processor,
     const fileapi::SyncFileCallback& callback) {
   scoped_ptr<TaskToken> token(
@@ skipping 241 matching lines @@
   if (state_ == REMOTE_SERVICE_OK)
     return;
   DVLOG(1) << "OnAuthenticated";
   state_ = REMOTE_SERVICE_OK;
   if (GetCurrentState() != REMOTE_SERVICE_OK)
     return;
   FOR_EACH_OBSERVER(
       Observer, service_observers_,
       OnRemoteServiceStateUpdated(GetCurrentState(), "Authenticated"));
   UpdatePollingDelay(kMinimumPollingDelaySeconds);
-  SchedulePolling();
+
+  may_have_unfetched_changes_ = true;
+  MaybeStartFetchChanges();
 }
 
 void DriveFileSyncService::OnNetworkConnected() {
   if (state_ == REMOTE_SERVICE_OK)
     return;
   DVLOG(1) << "OnNetworkConnected";
   state_ = REMOTE_SERVICE_OK;
   if (GetCurrentState() != REMOTE_SERVICE_OK)
     return;
   FOR_EACH_OBSERVER(
       Observer, service_observers_,
       OnRemoteServiceStateUpdated(GetCurrentState(), "Network connected"));
   UpdatePollingDelay(kMinimumPollingDelaySeconds);
-  SchedulePolling();
+
+  may_have_unfetched_changes_ = true;
+  MaybeStartFetchChanges();
 }
 
 // Called by CreateForTesting.
 DriveFileSyncService::DriveFileSyncService(
     Profile* profile,
     const base::FilePath& base_dir,
     scoped_ptr<DriveFileSyncClient> sync_client,
     scoped_ptr<DriveMetadataStore> metadata_store)
     : profile_(profile),
       last_operation_status_(fileapi::SYNC_STATUS_OK),
       state_(REMOTE_SERVICE_OK),
       sync_enabled_(true),
       largest_fetched_changestamp_(0),
       push_notification_registered_(false),
       push_notification_enabled_(false),
       polling_delay_seconds_(-1),
-      is_fetching_changes_(false),
+      may_have_unfetched_changes_(false),
       weak_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
   DCHECK(profile);
   temporary_file_dir_ = base_dir.Append(kTempDirName);
 
   token_.reset(new TaskToken(AsWeakPtr()));
   sync_client_ = sync_client.Pass();
   metadata_store_ = metadata_store.Pass();
 
   base::MessageLoopProxy::current()->PostTask(
       FROM_HERE,
@@ skipping 16 matching lines @@
   return token_.Pass();
 }
 
 void DriveFileSyncService::NotifyTaskDone(fileapi::SyncStatusCode status,
                                           scoped_ptr<TaskToken> token) {
   DCHECK(token);
   last_operation_status_ = status;
   token_ = token.Pass();
   TRACE_EVENT_ASYNC_END0("Sync FileSystem", "GetToken", this);
 
-
   if (token_->task_type() != TASK_TYPE_NONE) {
     DVLOG(2) << "NotifyTaskDone: " << token_->description()
              << ": finished with status=" << status
              << " (" << SyncStatusCodeToString(status) << ")"
              << " " << token_->location().ToString();
 
     RemoteServiceState old_state = GetCurrentState();
     UpdateServiceState();
 
     // Reset the polling delay. This will adjust the polling timer
     // based on the current service state.
     UpdatePollingDelay(polling_delay_seconds_);
 
     // Notify remote sync service state if the state has been changed.
     if (!token_->description().empty() || old_state != GetCurrentState()) {
       FOR_EACH_OBSERVER(
           Observer, service_observers_,
           OnRemoteServiceStateUpdated(GetCurrentState(),
                                       token_->done_description()));
     }
   }
 
-  if (!token_->completion_callback().is_null())
-    token_->completion_callback().Run();
-
   token_->ResetTask(FROM_HERE);
   if (!pending_tasks_.empty()) {
     base::Closure closure = pending_tasks_.front();
     pending_tasks_.pop_front();
     closure.Run();
     return;
   }
 
-  SchedulePolling();
-
-  if (GetCurrentState() != REMOTE_SERVICE_OK &&
-      GetCurrentState() != REMOTE_SERVICE_TEMPORARY_UNAVAILABLE)
+  if (GetCurrentState() == REMOTE_SERVICE_DISABLED)
     return;
 
-  // If the state has become OK or TEMPORARY_UNAVAILABLE and we have any
-  // pending batch sync origins, restart batch sync for them.
-  if (!pending_batch_sync_origins_.empty()) {
-    GURL origin = *pending_batch_sync_origins_.begin();
-    pending_batch_sync_origins_.erase(pending_batch_sync_origins_.begin());
-    std::string resource_id = metadata_store_->GetResourceIdForOrigin(origin);
-    StartBatchSyncForOrigin(origin, resource_id);
-    return;
-  }
+  MaybeStartFetchChanges();
+
+  SchedulePolling();
 
   // Notify observer of the update of |pending_changes_|.
   FOR_EACH_OBSERVER(Observer, service_observers_,
                     OnRemoteChangeQueueUpdated(pending_changes_.size()));
 }
 
 void DriveFileSyncService::UpdateServiceState() {
   switch (last_operation_status_) {
     // Possible regular operation errors.
     case fileapi::SYNC_STATUS_OK:
@@ skipping 161 matching lines @@
     return;
   }
   RegisterOriginForTrackingChanges(origin, callback);
 }
 
 void DriveFileSyncService::StartBatchSyncForOrigin(
     const GURL& origin,
     const std::string& resource_id) {
   scoped_ptr<TaskToken> token(
       GetToken(FROM_HERE, TASK_TYPE_DRIVE, "Retrieving largest changestamp"));
-  if (!token) {
-    pending_batch_sync_origins_.insert(origin);
-    return;
-  }
+  DCHECK(token);
+  DCHECK(GetCurrentState() == REMOTE_SERVICE_OK || may_have_unfetched_changes_);
+
+  DVLOG(1) << "Start batch sync for:" << origin.spec();
 
   sync_client_->GetLargestChangeStamp(
       base::Bind(&DriveFileSyncService::DidGetLargestChangeStampForBatchSync,
                  AsWeakPtr(), base::Passed(&token), origin, resource_id));
+
+  may_have_unfetched_changes_ = false;
 }
 
 void DriveFileSyncService::DidGetDirectoryForOrigin(
     scoped_ptr<TaskToken> token,
     const GURL& origin,
     const fileapi::SyncStatusCallback& callback,
     google_apis::GDataErrorCode error,
     const std::string& resource_id) {
   if (error != google_apis::HTTP_SUCCESS &&
       error != google_apis::HTTP_CREATED) {
@@ skipping 69 matching lines @@
             AsWeakPtr(), base::Passed(&token), origin, largest_changestamp));
     return;
   }
 
   // Move |origin| to the incremental sync origin set if the origin has no file.
   if (metadata_store_->IsBatchSyncOrigin(origin) &&
       !ContainsKey(origin_to_changes_map_, origin)) {
     metadata_store_->MoveBatchSyncOriginToIncremental(origin);
   }
 
+  // If this was the last batch sync origin and push_notification is enabled
+  // (indicates that we may have longer polling cycle), trigger the first
+  // incremental sync on next task cycle.
+  if (pending_batch_sync_origins_.empty() &&
+      push_notification_enabled_) {
+    may_have_unfetched_changes_ = true;
+  }
+
   NotifyTaskDone(fileapi::SYNC_STATUS_OK, token.Pass());
 }
 
 void DriveFileSyncService::DidRemoveOriginOnMetadataStore(
     scoped_ptr<TaskToken> token,
     const fileapi::SyncStatusCallback& callback,
     fileapi::SyncStatusCode status) {
   NotifyTaskDone(status, token.Pass());
   callback.Run(status);
 }
@@ skipping 745 matching lines @@
   if (found_url == origin_to_changes_map_.end())
     return false;
   const PathToChangeMap& path_to_change = found_url->second;
   PathToChangeMap::const_iterator found_path = path_to_change.find(url.path());
   if (found_path == path_to_change.end())
     return false;
   *change = found_path->second;
   return true;
 }
 
+void DriveFileSyncService::MaybeStartFetchChanges() {
+  if (!token_ || GetCurrentState() == REMOTE_SERVICE_DISABLED) {
+    // If another task is already running or the service is disabled
+    // just return here.
+    // Note that token_ should be already non-null if this is called
+    // from NotifyTaskDone().
+    return;
+  }
+
+  // If we have pending_batch_sync_origins, try starting the batch sync.
+  if (!pending_batch_sync_origins_.empty()) {
+    if (GetCurrentState() == REMOTE_SERVICE_OK ||
+        may_have_unfetched_changes_) {
+      GURL origin = *pending_batch_sync_origins_.begin();
+      pending_batch_sync_origins_.erase(pending_batch_sync_origins_.begin());
+      std::string resource_id = metadata_store_->GetResourceIdForOrigin(origin);
+      StartBatchSyncForOrigin(origin, resource_id);
+    }
+    return;
+  }
+
+  if (may_have_unfetched_changes_ &&
+      !metadata_store_->incremental_sync_origins().empty() &&
+      pending_changes_.empty()) {
+    FetchChangesForIncrementalSync();
+  }
+}
+
 void DriveFileSyncService::FetchChangesForIncrementalSync() {
   scoped_ptr<TaskToken> token(GetToken(FROM_HERE, TASK_TYPE_DRIVE,
                                        "Fetching remote change list"));
-  // If we got |token| successfully, |is_fetching_changes_| should be false.
-  // |is_fetching_changes_| is true only when the FetchChanges sequence is
-  // running or is in |pending_queue_|. In both case, the token is owned by
-  // the FetchChanges sequence.
-  DCHECK(!token || !is_fetching_changes_);
-
-  if (!sync_enabled_ ||
-      is_fetching_changes_ ||
-      !pending_batch_sync_origins_.empty() ||
-      metadata_store_->incremental_sync_origins().empty() ||
-      !pending_changes_.empty()) {
-    if (token) {
-      token->ResetTask(FROM_HERE);
-      NotifyTaskDone(last_operation_status_, token.Pass());
-    }
-    return;
-  }
-
-  is_fetching_changes_ = true;
-
-  if (!token) {
-    pending_tasks_.push_back(base::Bind(
-        &DriveFileSyncService::FetchChangesForIncrementalSync, AsWeakPtr()));
-    return;
-  }
-
-  token->set_completion_callback(
-      base::Bind(&MarkFetchingChangesCompleted, &is_fetching_changes_));
-
-  if (metadata_store_->incremental_sync_origins().empty()) {
-    token->ResetTask(FROM_HERE);
-    NotifyTaskDone(fileapi::SYNC_STATUS_OK, token.Pass());
-    return;
-  }
+  DCHECK(token);
+  DCHECK(may_have_unfetched_changes_);
+  DCHECK(pending_batch_sync_origins_.empty());
+  DCHECK(!metadata_store_->incremental_sync_origins().empty());
+  DCHECK(pending_changes_.empty());
 
   DVLOG(1) << "FetchChangesForIncrementalSync (start_changestamp:"
            << (largest_fetched_changestamp_ + 1) << ")";
 
   sync_client_->ListChanges(
       largest_fetched_changestamp_ + 1,
       base::Bind(&DriveFileSyncService::DidFetchChangesForIncrementalSync,
                  AsWeakPtr(), base::Passed(&token), false));
+
+  may_have_unfetched_changes_ = false;
 }
 
 void DriveFileSyncService::DidFetchChangesForIncrementalSync(
     scoped_ptr<TaskToken> token,
     bool has_new_changes,
     google_apis::GDataErrorCode error,
     scoped_ptr<google_apis::ResourceList> changes) {
   if (error != google_apis::HTTP_SUCCESS) {
     NotifyTaskDone(GDataErrorCodeToSyncStatusCodeWrapper(error), token.Pass());
     return;
@@ skipping 83 matching lines @@
   ids.insert(invalidation::ObjectId(
       ipc::invalidation::ObjectSource::COSMO_CHANGELOG,
       kDriveInvalidationObjectId));
   profile_sync_service->UpdateRegisteredInvalidationIds(this, ids);
   push_notification_registered_ = true;
   SetPushNotificationEnabled(profile_sync_service->GetInvalidatorState());
 }
 
 void DriveFileSyncService::SchedulePolling() {
   if (polling_timer_.IsRunning() ||
-      polling_delay_seconds_ < 0)
-    return;
-
-  if (state_ != REMOTE_SERVICE_OK &&
-      state_ != REMOTE_SERVICE_TEMPORARY_UNAVAILABLE)
+      polling_delay_seconds_ < 0 ||
+      GetCurrentState() == REMOTE_SERVICE_DISABLED)
     return;
 
   DVLOG(1) << "Polling scheduled"
            << " (delay:" << polling_delay_seconds_ << "s)";
 
   polling_timer_.Start(
       FROM_HERE, base::TimeDelta::FromSeconds(polling_delay_seconds_),
-      base::Bind(&DriveFileSyncService::FetchChangesForIncrementalSync,
-                 AsWeakPtr()));
+      base::Bind(&DriveFileSyncService::OnPollingTimerFired, AsWeakPtr()));
+}
+
+void DriveFileSyncService::OnPollingTimerFired() {
+  may_have_unfetched_changes_ = true;
+  MaybeStartFetchChanges();
 }
 
 void DriveFileSyncService::UpdatePollingDelay(int64 new_delay_sec) {
   // polling_delay_seconds_ made negative to disable polling for testing.
   if (polling_delay_seconds_ < 0)
     return;
 
   if (state_ == REMOTE_SERVICE_TEMPORARY_UNAVAILABLE) {
     // If the service state is TEMPORARY_UNAVAILABLE, poll the service
     // with a modest duration (but more frequently than
@@ skipping 21 matching lines @@
 fileapi::SyncStatusCode
 DriveFileSyncService::GDataErrorCodeToSyncStatusCodeWrapper(
     google_apis::GDataErrorCode error) const {
   fileapi::SyncStatusCode status = GDataErrorCodeToSyncStatusCode(error);
   if (status != fileapi::SYNC_STATUS_OK && !sync_client_->IsAuthenticated())
     return fileapi::SYNC_STATUS_AUTHENTICATION_FAILED;
   return status;
 }
 
 }  // namespace sync_file_system