[Sync] Make BackendMigrator not wait for full sync cycles

chrome/browser/sync/profile_sync_service_harness.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/profile_sync_service_harness.h"
 
 #include <stddef.h>
 #include <algorithm>
 #include <iterator>
 #include <ostream>
 #include <set>
 #include <sstream>
 #include <vector>
 
+#include "base/base64.h"
+#include "base/compiler_specific.h"
 #include "base/json/json_writer.h"
 #include "base/logging.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop.h"
 #include "base/task.h"
 #include "base/tracked.h"
 #include "chrome/browser/profiles/profile.h"
 #include "chrome/browser/sync/sessions/session_state.h"
 #include "chrome/browser/sync/signin_manager.h"
 #include "chrome/browser/sync/sync_ui_util.h"
@@ skipping 75 matching lines @@
       wait_state_(INITIAL_WAIT_STATE),
       profile_(profile),
       service_(NULL),
       timestamp_match_partner_(NULL),
       username_(username),
       password_(password),
       profile_debug_name_(profile->GetDebugName()) {
   if (IsSyncAlreadySetup()) {
     service_ = profile_->GetProfileSyncService();
     service_->AddObserver(this);
+    ignore_result(TryListeningToMigrationEvents());
     wait_state_ = FULLY_SYNCED;
   }
 }
 
 ProfileSyncServiceHarness::~ProfileSyncServiceHarness() {}
 
 // static
 ProfileSyncServiceHarness* ProfileSyncServiceHarness::CreateAndAttach(
     Profile* profile) {
   if (!profile->HasProfileSyncService()) {
@@ skipping 53 matching lines @@
                << " seconds.";
     return false;
   }
 
   // Choose the datatypes to be synced. If all datatypes are to be synced,
   // set sync_everything to true; otherwise, set it to false.
   bool sync_everything = (synced_datatypes.size() ==
       (syncable::MODEL_TYPE_COUNT - syncable::FIRST_REAL_MODEL_TYPE));
   service()->OnUserChoseDatatypes(sync_everything, synced_datatypes);
 
+  // Subscribe sync client to notifications from the backend migrator
+  // (possible only after choosing data types).
+  if (!TryListeningToMigrationEvents()) {
+    NOTREACHED();
+    return false;
+  }
+
   // Make sure that a partner client hasn't already set an explicit passphrase.
   if (wait_state_ == SET_PASSPHRASE_FAILED) {
     LOG(ERROR) << "A passphrase is required for decryption. Sync cannot proceed"
                   " until SetPassphrase is called.";
     return false;
   }
 
   // Wait for initial sync cycle to be completed.
   DCHECK_EQ(wait_state_, WAITING_FOR_INITIAL_SYNC);
   if (!AwaitStatusChangeWithTimeout(kLiveSyncOperationTimeoutMs,
@@ skipping 10 matching lines @@
                   " until SetPassphrase is called.";
     return false;
   }
 
   // Indicate to the browser that sync setup is complete.
   service()->SetSyncSetupCompleted();
 
   return true;
 }
 
+bool ProfileSyncServiceHarness::TryListeningToMigrationEvents() {
+  browser_sync::BackendMigrator* migrator =
+      service_->GetBackendMigratorForTest();
+  if (migrator && !migrator->HasMigrationObserver(this)) {
+    migrator->AddMigrationObserver(this);
+    return true;
+  }
+  return false;
+}
+
 void ProfileSyncServiceHarness::SignalStateCompleteWithNextState(
     WaitState next_state) {
   wait_state_ = next_state;
   SignalStateComplete();
 }
 
 void ProfileSyncServiceHarness::SignalStateComplete() {
   MessageLoop::current()->Quit();
 }
 
@@ skipping 114 matching lines @@
       VLOG(1) << GetClientInfoString(
                      "WAITING_FOR_EXPONENTIAL_BACKOFF_VERIFICATION");
       const browser_sync::sessions::SyncSessionSnapshot *snap =
           GetLastSessionSnapshot();
       CHECK(snap);
       retry_verifier_.VerifyRetryInterval(*snap);
       if (retry_verifier_.done())
         SignalStateCompleteWithNextState(WAITING_FOR_NOTHING);
       break;
     }
+    case WAITING_FOR_MIGRATION_TO_START: {
+      VLOG(1) << GetClientInfoString("WAITING_FOR_MIGRATION_TO_START");
+      if (HasPendingBackendMigration()) {
+        SignalStateCompleteWithNextState(WAITING_FOR_MIGRATION_TO_FINISH);
+      }
+      break;
+    }
+    case WAITING_FOR_MIGRATION_TO_FINISH: {
+      VLOG(1) << GetClientInfoString("WAITING_FOR_MIGRATION_TO_FINISH");
+      if (!HasPendingBackendMigration()) {
+        SignalStateCompleteWithNextState(WAITING_FOR_NOTHING);
+      }
+      break;
+    }
     case SERVER_UNREACHABLE: {
       VLOG(1) << GetClientInfoString("SERVER_UNREACHABLE");
       if (GetStatus().server_reachable) {
         // The client was offline due to the network being disabled, but is now
         // back online. Wait for the pending sync cycle to complete.
         SignalStateCompleteWithNextState(WAITING_FOR_SYNC_TO_FINISH);
       }
       break;
     }
     case SET_PASSPHRASE_FAILED: {
@@ skipping 23 matching lines @@
       // classes using the the UI message loop.  Defer to their handling.
       break;
   }
   return original_wait_state != wait_state_;
 }
 
 void ProfileSyncServiceHarness::OnStateChanged() {
   RunStateChangeMachine();
 }
 
+void ProfileSyncServiceHarness::OnMigrationStateChange() {
+  // Update migration state.
+  if (HasPendingBackendMigration()) {
+    // Merge current pending migration types into
+    // |pending_migration_types_|.
+    syncable::ModelTypeSet new_pending_migration_types =
+        service()->GetBackendMigratorForTest()->
+        GetPendingMigrationTypesForTest();
+    syncable::ModelTypeSet temp;
+    std::set_union(pending_migration_types_.begin(),
+                   pending_migration_types_.end(),
+                   new_pending_migration_types.begin(),
+                   new_pending_migration_types.end(),
+                   std::inserter(temp, temp.end()));
+    std::swap(pending_migration_types_, temp);
+    VLOG(1) << profile_debug_name_ << ": new pending migration types "
+            << syncable::ModelTypeSetToString(pending_migration_types_);
+  } else {
+    // Merge just-finished pending migration types into
+    // |migration_types_|.
+    syncable::ModelTypeSet temp;
+    std::set_union(pending_migration_types_.begin(),
+                   pending_migration_types_.end(),
+                   migrated_types_.begin(),
+                   migrated_types_.end(),
+                   std::inserter(temp, temp.end()));
+    std::swap(migrated_types_, temp);
+    pending_migration_types_.clear();
+    VLOG(1) << profile_debug_name_ << ": new migrated types "
+            << syncable::ModelTypeSetToString(migrated_types_);
+  }
+  RunStateChangeMachine();
+}
+
 bool ProfileSyncServiceHarness::AwaitPassphraseRequired() {
   VLOG(1) << GetClientInfoString("AwaitPassphraseRequired");
   if (wait_state_ == SYNC_DISABLED) {
     LOG(ERROR) << "Sync disabled for " << profile_debug_name_ << ".";
     return false;
   }
 
   if (service()->IsPassphraseRequired()) {
     // It's already true that a passphrase is required; don't wait.
     return true;
@@ skipping 62 matching lines @@
   if (wait_state_ == SYNC_DISABLED) {
     LOG(ERROR) << "Sync disabled for " << profile_debug_name_ << ".";
     return false;
   }
 
   if (IsSynced()) {
     // Client is already synced; don't wait.
     return true;
   }
 
-  return AwaitSyncCycleCompletionHelper(reason);
-}
-
-bool ProfileSyncServiceHarness::AwaitNextSyncCycleCompletion(
-    const std::string& reason) {
-  VLOG(1) << GetClientInfoString("AwaitNextSyncCycleCompletion");
-  return AwaitSyncCycleCompletionHelper(reason);
-}
-
-bool ProfileSyncServiceHarness::AwaitSyncCycleCompletionHelper(
-    const std::string& reason) {
   if (wait_state_ == SERVER_UNREACHABLE) {
     // Client was offline; wait for it to go online, and then wait for sync.
     AwaitStatusChangeWithTimeout(kLiveSyncOperationTimeoutMs, reason);
     DCHECK_EQ(wait_state_, WAITING_FOR_SYNC_TO_FINISH);
     return AwaitStatusChangeWithTimeout(kLiveSyncOperationTimeoutMs, reason);
   }
 
   DCHECK(service()->sync_initialized());
   wait_state_ = WAITING_FOR_SYNC_TO_FINISH;
   AwaitStatusChangeWithTimeout(kLiveSyncOperationTimeoutMs, reason);
@@ skipping 22 matching lines @@
   const browser_sync::sessions::SyncSessionSnapshot *snap =
       GetLastSessionSnapshot();
   CHECK(snap);
   retry_verifier_.Initialize(*snap);
   wait_state_ = WAITING_FOR_EXPONENTIAL_BACKOFF_VERIFICATION;
   AwaitStatusChangeWithTimeout(kExponentialBackoffVerificationTimeoutMs,
       "Verify Exponential backoff");
   return (retry_verifier_.Succeeded());
 }
 
+bool ProfileSyncServiceHarness::AwaitMigration(
+    const syncable::ModelTypeSet& expected_migrated_types) {
+  VLOG(1) << GetClientInfoString("AwaitMigration");
+  VLOG(1) << profile_debug_name_ << ": waiting until migration is done for "
+          << syncable::ModelTypeSetToString(expected_migrated_types);
+  while (true) {
+    bool migration_finished =
+        std::includes(migrated_types_.begin(), migrated_types_.end(),
+                      expected_migrated_types.begin(),
+                      expected_migrated_types.end());
+    VLOG(1) << "Migrated types "
+            << syncable::ModelTypeSetToString(migrated_types_)
+            << (migration_finished ? " contains " : " does not contain ")
+            << syncable::ModelTypeSetToString(expected_migrated_types);
+    if (migration_finished) {
+      return true;
+    }
+
+    if (HasPendingBackendMigration()) {
+      wait_state_ = WAITING_FOR_MIGRATION_TO_FINISH;
+    } else {
+      wait_state_ = WAITING_FOR_MIGRATION_TO_START;
+      AwaitStatusChangeWithTimeout(kLiveSyncOperationTimeoutMs,
+                                   "Wait for migration to start");
+      if (wait_state_ != WAITING_FOR_MIGRATION_TO_FINISH) {
+        VLOG(1) << profile_debug_name_
+                << ": wait state = " << wait_state_
+                << " after migration start is not "
+                << "WAITING_FOR_MIGRATION_TO_FINISH";
+        return false;
+      }
+    }
+    AwaitStatusChangeWithTimeout(kLiveSyncOperationTimeoutMs,
+                                 "Wait for migration to finish");
+    if (wait_state_ != WAITING_FOR_NOTHING) {
+      VLOG(1) << profile_debug_name_
+              << ": wait state = " << wait_state_
+              << " after migration finish is not WAITING_FOR_NOTHING";
+      return false;
+    }
+    if (!AwaitSyncCycleCompletion(
+            "Config sync cycle after migration cycle")) {
+      return false;
+    }
+  }
+}
+
 bool ProfileSyncServiceHarness::AwaitMutualSyncCycleCompletion(
     ProfileSyncServiceHarness* partner) {
   VLOG(1) << GetClientInfoString("AwaitMutualSyncCycleCompletion");
   if (!AwaitSyncCycleCompletion("Sync cycle completion on active client."))
     return false;
   return partner->WaitUntilTimestampMatches(this,
       "Sync cycle completion on passive client.");
 }
 
 bool ProfileSyncServiceHarness::AwaitGroupSyncCycleCompletion(
@@ skipping 89 matching lines @@
   const SyncSessionSnapshot* snap = GetLastSessionSnapshot();
   // TODO(rsimha): Remove additional checks of snap->has_more_to_sync and
   // snap->unsynced_count once http://crbug.com/48989 is fixed.
   bool is_synced = snap &&
       snap->num_blocking_conflicting_updates == 0 &&
       ServiceIsPushingChanges() &&
       GetStatus().notifications_enabled &&
       !service()->HasUnsyncedItems() &&
       !snap->has_more_to_sync &&
       snap->unsynced_count == 0 &&
-      !service()->HasPendingBackendMigration() &&
+      !HasPendingBackendMigration() &&
       service()->passphrase_required_reason() !=
       sync_api::REASON_SET_PASSPHRASE_FAILED;
   VLOG(1) << GetClientInfoString(
       is_synced ? "IsSynced: true" : "IsSynced: false");
   return is_synced;
 }
 
+bool ProfileSyncServiceHarness::HasPendingBackendMigration() {
+  browser_sync::BackendMigrator* migrator =
+      service()->GetBackendMigratorForTest();
+  return migrator && migrator->state() != browser_sync::BackendMigrator::IDLE;
+}
+
 bool ProfileSyncServiceHarness::MatchesOtherClient(
     ProfileSyncServiceHarness* partner) {
-  if (!IsSynced())
+  // TODO(akalin): Shouldn't this belong with the intersection check?
+  // Otherwise, this function isn't symmetric.
+  if (!IsSynced()) {
+    VLOG(2) << profile_debug_name_ << ": not synced, assuming doesn't match";
     return false;
+  }
 
   // Only look for a match if we have at least one enabled datatype in
   // common with the partner client.
   syncable::ModelTypeSet types, other_types, intersection_types;
   service()->GetPreferredDataTypes(&types);
   partner->service()->GetPreferredDataTypes(&other_types);
   std::set_intersection(types.begin(), types.end(), other_types.begin(),
                         other_types.end(),
                         inserter(intersection_types,
                                  intersection_types.begin()));
+
+  VLOG(2) << profile_debug_name_ << ", " << partner->profile_debug_name_
+          << ": common types are "
+          << syncable::ModelTypeSetToString(intersection_types);
+
+  if (!intersection_types.empty() && !partner->IsSynced()) {
+    VLOG(2) << "non-empty common types and "
+            << partner->profile_debug_name_ << " isn't synced";
+    return false;
+  }
+
   for (syncable::ModelTypeSet::iterator i = intersection_types.begin();
-       i != intersection_types.end();
-       ++i) {
-    if (!partner->IsSynced() ||
-        partner->GetUpdatedTimestamp(*i) != GetUpdatedTimestamp(*i)) {
+       i != intersection_types.end(); ++i) {
+    const std::string timestamp = GetUpdatedTimestamp(*i);
+    const std::string partner_timestamp = partner->GetUpdatedTimestamp(*i);
+    if (timestamp != partner_timestamp) {
+      if (VLOG_IS_ON(2)) {
+        std::string timestamp_base64, partner_timestamp_base64;
+        if (!base::Base64Encode(timestamp, &timestamp_base64)) {
+          NOTREACHED();
+        }
+        if (!base::Base64Encode(
+                partner_timestamp, &partner_timestamp_base64)) {
+          NOTREACHED();
+        }
+        VLOG(2) << syncable::ModelTypeToString(*i) << ": "
+                << profile_debug_name_ << " timestamp = "
+                << timestamp_base64 << ", "
+                << partner->profile_debug_name_
+                << " partner timestamp = "
+                << partner_timestamp_base64;
+      }
       return false;
     }
   }
   return true;
 }
 
 const SyncSessionSnapshot*
     ProfileSyncServiceHarness::GetLastSessionSnapshot() const {
   DCHECK(service_ != NULL) << "Sync service has not yet been set up.";
   if (service_->sync_initialized()) {
@@ skipping 146 matching lines @@
          << ", num_updates_downloaded : "
          << snap->syncer_status.num_updates_downloaded_total
          << ", passphrase_required_reason: "
          << sync_api::PassphraseRequiredReasonToString(
              service()->passphrase_required_reason())
          << ", notifications_enabled: "
          << status.notifications_enabled
          << ", service_is_pushing_changes: "
          << ServiceIsPushingChanges()
          << ", has_pending_backend_migration: "
-         << service()->HasPendingBackendMigration();
+         << HasPendingBackendMigration();
     } else {
       os << "Sync session snapshot not available";
     }
   } else {
     os << "Sync service not available";
   }
   return os.str();
 }
 
 bool ProfileSyncServiceHarness::EnableEncryptionForType(
@@ skipping 57 matching lines @@
   return (synced_types.count(type) != 0);
 }
 
 std::string ProfileSyncServiceHarness::GetServiceStatus() {
   DictionaryValue value;
   sync_ui_util::ConstructAboutInformation(service_, &value);
   std::string service_status;
   base::JSONWriter::Write(&value, true, &service_status);
   return service_status;
 }