Home | History | Annotate | Download | only in internal_api 
      1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
      2 // Use of this source code is governed by a BSD-style license that can be
      3 // found in the LICENSE file.
      4 
      5 #include "sync/internal_api/sync_manager_impl.h"
      6 
      7 #include <string>
      8 
      9 #include "base/base64.h"
     10 #include "base/bind.h"
     11 #include "base/callback.h"
     12 #include "base/compiler_specific.h"
     13 #include "base/json/json_writer.h"
     14 #include "base/memory/ref_counted.h"
     15 #include "base/metrics/histogram.h"
     16 #include "base/observer_list.h"
     17 #include "base/strings/string_number_conversions.h"
     18 #include "base/thread_task_runner_handle.h"
     19 #include "base/values.h"
     20 #include "sync/engine/sync_scheduler.h"
     21 #include "sync/engine/syncer_types.h"
     22 #include "sync/internal_api/change_reorder_buffer.h"
     23 #include "sync/internal_api/public/base/cancelation_signal.h"
     24 #include "sync/internal_api/public/base/invalidation_interface.h"
     25 #include "sync/internal_api/public/base/model_type.h"
     26 #include "sync/internal_api/public/base_node.h"
     27 #include "sync/internal_api/public/configure_reason.h"
     28 #include "sync/internal_api/public/engine/polling_constants.h"
     29 #include "sync/internal_api/public/http_post_provider_factory.h"
     30 #include "sync/internal_api/public/internal_components_factory.h"
     31 #include "sync/internal_api/public/read_node.h"
     32 #include "sync/internal_api/public/read_transaction.h"
     33 #include "sync/internal_api/public/sync_context.h"
     34 #include "sync/internal_api/public/sync_context_proxy.h"
     35 #include "sync/internal_api/public/user_share.h"
     36 #include "sync/internal_api/public/util/experiments.h"
     37 #include "sync/internal_api/public/write_node.h"
     38 #include "sync/internal_api/public/write_transaction.h"
     39 #include "sync/internal_api/sync_context_proxy_impl.h"
     40 #include "sync/internal_api/syncapi_internal.h"
     41 #include "sync/internal_api/syncapi_server_connection_manager.h"
     42 #include "sync/protocol/proto_value_conversions.h"
     43 #include "sync/protocol/sync.pb.h"
     44 #include "sync/sessions/directory_type_debug_info_emitter.h"
     45 #include "sync/syncable/directory.h"
     46 #include "sync/syncable/entry.h"
     47 #include "sync/syncable/in_memory_directory_backing_store.h"
     48 #include "sync/syncable/on_disk_directory_backing_store.h"
     49 
     50 using base::TimeDelta;
     51 using sync_pb::GetUpdatesCallerInfo;
     52 
     53 class GURL;
     54 
     55 namespace syncer {
     56 
     57 using sessions::SyncSessionContext;
     58 using syncable::ImmutableWriteTransactionInfo;
     59 using syncable::SPECIFICS;
     60 using syncable::UNIQUE_POSITION;
     61 
     62 namespace {
     63 
     64 GetUpdatesCallerInfo::GetUpdatesSource GetSourceFromReason(
     65     ConfigureReason reason) {
     66   switch (reason) {
     67     case CONFIGURE_REASON_RECONFIGURATION:
     68       return GetUpdatesCallerInfo::RECONFIGURATION;
     69     case CONFIGURE_REASON_MIGRATION:
     70       return GetUpdatesCallerInfo::MIGRATION;
     71     case CONFIGURE_REASON_NEW_CLIENT:
     72       return GetUpdatesCallerInfo::NEW_CLIENT;
     73     case CONFIGURE_REASON_NEWLY_ENABLED_DATA_TYPE:
     74     case CONFIGURE_REASON_CRYPTO:
     75       return GetUpdatesCallerInfo::NEWLY_SUPPORTED_DATATYPE;
     76     case CONFIGURE_REASON_PROGRAMMATIC:
     77       return GetUpdatesCallerInfo::PROGRAMMATIC;
     78     default:
     79       NOTREACHED();
     80   }
     81   return GetUpdatesCallerInfo::UNKNOWN;
     82 }
     83 
     84 }  // namespace
     85 
     86 SyncManagerImpl::SyncManagerImpl(const std::string& name)
     87     : name_(name),
     88       change_delegate_(NULL),
     89       initialized_(false),
     90       observing_network_connectivity_changes_(false),
     91       report_unrecoverable_error_function_(NULL),
     92       weak_ptr_factory_(this) {
     93   // Pre-fill |notification_info_map_|.
     94   for (int i = FIRST_REAL_MODEL_TYPE; i < MODEL_TYPE_COUNT; ++i) {
     95     notification_info_map_.insert(
     96         std::make_pair(ModelTypeFromInt(i), NotificationInfo()));
     97   }
     98 }
     99 
    100 SyncManagerImpl::~SyncManagerImpl() {
    101   DCHECK(thread_checker_.CalledOnValidThread());
    102   CHECK(!initialized_);
    103 }
    104 
    105 SyncManagerImpl::NotificationInfo::NotificationInfo() : total_count(0) {}
    106 SyncManagerImpl::NotificationInfo::~NotificationInfo() {}
    107 
    108 base::DictionaryValue* SyncManagerImpl::NotificationInfo::ToValue() const {
    109   base::DictionaryValue* value = new base::DictionaryValue();
    110   value->SetInteger("totalCount", total_count);
    111   value->SetString("payload", payload);
    112   return value;
    113 }
    114 
    115 bool SyncManagerImpl::VisiblePositionsDiffer(
    116     const syncable::EntryKernelMutation& mutation) const {
    117   const syncable::EntryKernel& a = mutation.original;
    118   const syncable::EntryKernel& b = mutation.mutated;
    119   if (!b.ShouldMaintainPosition())
    120     return false;
    121   if (!a.ref(UNIQUE_POSITION).Equals(b.ref(UNIQUE_POSITION)))
    122     return true;
    123   if (a.ref(syncable::PARENT_ID) != b.ref(syncable::PARENT_ID))
    124     return true;
    125   return false;
    126 }
    127 
    128 bool SyncManagerImpl::VisiblePropertiesDiffer(
    129     const syncable::EntryKernelMutation& mutation,
    130     Cryptographer* cryptographer) const {
    131   const syncable::EntryKernel& a = mutation.original;
    132   const syncable::EntryKernel& b = mutation.mutated;
    133   const sync_pb::EntitySpecifics& a_specifics = a.ref(SPECIFICS);
    134   const sync_pb::EntitySpecifics& b_specifics = b.ref(SPECIFICS);
    135   DCHECK_EQ(GetModelTypeFromSpecifics(a_specifics),
    136             GetModelTypeFromSpecifics(b_specifics));
    137   ModelType model_type = GetModelTypeFromSpecifics(b_specifics);
    138   // Suppress updates to items that aren't tracked by any browser model.
    139   if (model_type < FIRST_REAL_MODEL_TYPE ||
    140       !a.ref(syncable::UNIQUE_SERVER_TAG).empty()) {
    141     return false;
    142   }
    143   if (a.ref(syncable::IS_DIR) != b.ref(syncable::IS_DIR))
    144     return true;
    145   if (!AreSpecificsEqual(cryptographer,
    146                          a.ref(syncable::SPECIFICS),
    147                          b.ref(syncable::SPECIFICS))) {
    148     return true;
    149   }
    150   if (!AreAttachmentMetadataEqual(a.ref(syncable::ATTACHMENT_METADATA),
    151                                   b.ref(syncable::ATTACHMENT_METADATA))) {
    152     return true;
    153   }
    154   // We only care if the name has changed if neither specifics is encrypted
    155   // (encrypted nodes blow away the NON_UNIQUE_NAME).
    156   if (!a_specifics.has_encrypted() && !b_specifics.has_encrypted() &&
    157       a.ref(syncable::NON_UNIQUE_NAME) != b.ref(syncable::NON_UNIQUE_NAME))
    158     return true;
    159   if (VisiblePositionsDiffer(mutation))
    160     return true;
    161   return false;
    162 }
    163 
    164 ModelTypeSet SyncManagerImpl::InitialSyncEndedTypes() {
    165   return directory()->InitialSyncEndedTypes();
    166 }
    167 
    168 ModelTypeSet SyncManagerImpl::GetTypesWithEmptyProgressMarkerToken(
    169     ModelTypeSet types) {
    170   ModelTypeSet result;
    171   for (ModelTypeSet::Iterator i = types.First(); i.Good(); i.Inc()) {
    172     sync_pb::DataTypeProgressMarker marker;
    173     directory()->GetDownloadProgress(i.Get(), &marker);
    174 
    175     if (marker.token().empty())
    176       result.Put(i.Get());
    177   }
    178   return result;
    179 }
    180 
    181 void SyncManagerImpl::ConfigureSyncer(
    182     ConfigureReason reason,
    183     ModelTypeSet to_download,
    184     ModelTypeSet to_purge,
    185     ModelTypeSet to_journal,
    186     ModelTypeSet to_unapply,
    187     const ModelSafeRoutingInfo& new_routing_info,
    188     const base::Closure& ready_task,
    189     const base::Closure& retry_task) {
    190   DCHECK(thread_checker_.CalledOnValidThread());
    191   DCHECK(!ready_task.is_null());
    192   DCHECK(!retry_task.is_null());
    193   DCHECK(initialized_);
    194 
    195   DVLOG(1) << "Configuring -"
    196            << "\n\t" << "current types: "
    197            << ModelTypeSetToString(GetRoutingInfoTypes(new_routing_info))
    198            << "\n\t" << "types to download: "
    199            << ModelTypeSetToString(to_download)
    200            << "\n\t" << "types to purge: "
    201            << ModelTypeSetToString(to_purge)
    202            << "\n\t" << "types to journal: "
    203            << ModelTypeSetToString(to_journal)
    204            << "\n\t" << "types to unapply: "
    205            << ModelTypeSetToString(to_unapply);
    206   if (!PurgeDisabledTypes(to_purge,
    207                           to_journal,
    208                           to_unapply)) {
    209     // We failed to cleanup the types. Invoke the ready task without actually
    210     // configuring any types. The caller should detect this as a configuration
    211     // failure and act appropriately.
    212     ready_task.Run();
    213     return;
    214   }
    215 
    216   ConfigurationParams params(GetSourceFromReason(reason),
    217                              to_download,
    218                              new_routing_info,
    219                              ready_task,
    220                              retry_task);
    221 
    222   scheduler_->Start(SyncScheduler::CONFIGURATION_MODE);
    223   scheduler_->ScheduleConfiguration(params);
    224 }
    225 
    226 void SyncManagerImpl::Init(InitArgs* args) {
    227   CHECK(!initialized_);
    228   DCHECK(thread_checker_.CalledOnValidThread());
    229   DCHECK(args->post_factory.get());
    230   DCHECK(!args->credentials.email.empty());
    231   DCHECK(!args->credentials.sync_token.empty());
    232   DCHECK(!args->credentials.scope_set.empty());
    233   DCHECK(args->cancelation_signal);
    234   DVLOG(1) << "SyncManager starting Init...";
    235 
    236   weak_handle_this_ = MakeWeakHandle(weak_ptr_factory_.GetWeakPtr());
    237 
    238   change_delegate_ = args->change_delegate;
    239 
    240   AddObserver(&js_sync_manager_observer_);
    241   SetJsEventHandler(args->event_handler);
    242 
    243   AddObserver(&debug_info_event_listener_);
    244 
    245   database_path_ = args->database_location.Append(
    246       syncable::Directory::kSyncDatabaseFilename);
    247   unrecoverable_error_handler_ = args->unrecoverable_error_handler.Pass();
    248   report_unrecoverable_error_function_ =
    249       args->report_unrecoverable_error_function;
    250 
    251   allstatus_.SetHasKeystoreKey(
    252       !args->restored_keystore_key_for_bootstrapping.empty());
    253   sync_encryption_handler_.reset(new SyncEncryptionHandlerImpl(
    254       &share_,
    255       args->encryptor,
    256       args->restored_key_for_bootstrapping,
    257       args->restored_keystore_key_for_bootstrapping));
    258   sync_encryption_handler_->AddObserver(this);
    259   sync_encryption_handler_->AddObserver(&debug_info_event_listener_);
    260   sync_encryption_handler_->AddObserver(&js_sync_encryption_handler_observer_);
    261 
    262   base::FilePath absolute_db_path = database_path_;
    263   DCHECK(absolute_db_path.IsAbsolute());
    264 
    265   scoped_ptr<syncable::DirectoryBackingStore> backing_store =
    266       args->internal_components_factory->BuildDirectoryBackingStore(
    267           InternalComponentsFactory::STORAGE_ON_DISK,
    268           args->credentials.email, absolute_db_path).Pass();
    269 
    270   DCHECK(backing_store.get());
    271   share_.directory.reset(
    272       new syncable::Directory(
    273           backing_store.release(),
    274           unrecoverable_error_handler_.get(),
    275           report_unrecoverable_error_function_,
    276           sync_encryption_handler_.get(),
    277           sync_encryption_handler_->GetCryptographerUnsafe()));
    278   share_.sync_credentials = args->credentials;
    279 
    280   // UserShare is accessible to a lot of code that doesn't need access to the
    281   // sync token so clear sync_token from the UserShare.
    282   share_.sync_credentials.sync_token = "";
    283 
    284   const std::string& username = args->credentials.email;
    285   DVLOG(1) << "Username: " << username;
    286   if (!OpenDirectory(username)) {
    287     NotifyInitializationFailure();
    288     LOG(ERROR) << "Sync manager initialization failed!";
    289     return;
    290   }
    291 
    292   connection_manager_.reset(new SyncAPIServerConnectionManager(
    293       args->service_url.host() + args->service_url.path(),
    294       args->service_url.EffectiveIntPort(),
    295       args->service_url.SchemeIsSecure(),
    296       args->post_factory.release(),
    297       args->cancelation_signal));
    298   connection_manager_->set_client_id(directory()->cache_guid());
    299   connection_manager_->AddListener(this);
    300 
    301   std::string sync_id = directory()->cache_guid();
    302 
    303   DVLOG(1) << "Setting sync client ID: " << sync_id;
    304   allstatus_.SetSyncId(sync_id);
    305   DVLOG(1) << "Setting invalidator client ID: " << args->invalidator_client_id;
    306   allstatus_.SetInvalidatorClientId(args->invalidator_client_id);
    307 
    308   model_type_registry_.reset(
    309       new ModelTypeRegistry(args->workers, directory(), this));
    310   sync_encryption_handler_->AddObserver(model_type_registry_.get());
    311 
    312   // Bind the SyncContext WeakPtr to this thread.  This helps us crash earlier
    313   // if the pointer is misused in debug mode.
    314   base::WeakPtr<SyncContext> weak_core = model_type_registry_->AsWeakPtr();
    315   weak_core.get();
    316 
    317   sync_context_proxy_.reset(
    318       new SyncContextProxyImpl(base::ThreadTaskRunnerHandle::Get(), weak_core));
    319 
    320   // Build a SyncSessionContext and store the worker in it.
    321   DVLOG(1) << "Sync is bringing up SyncSessionContext.";
    322   std::vector<SyncEngineEventListener*> listeners;
    323   listeners.push_back(&allstatus_);
    324   listeners.push_back(this);
    325   session_context_ =
    326       args->internal_components_factory->BuildContext(
    327                                              connection_manager_.get(),
    328                                              directory(),
    329                                              args->extensions_activity,
    330                                              listeners,
    331                                              &debug_info_event_listener_,
    332                                              model_type_registry_.get(),
    333                                              args->invalidator_client_id)
    334           .Pass();
    335   session_context_->set_account_name(args->credentials.email);
    336   scheduler_ = args->internal_components_factory->BuildScheduler(
    337       name_, session_context_.get(), args->cancelation_signal).Pass();
    338 
    339   scheduler_->Start(SyncScheduler::CONFIGURATION_MODE);
    340 
    341   initialized_ = true;
    342 
    343   net::NetworkChangeNotifier::AddIPAddressObserver(this);
    344   net::NetworkChangeNotifier::AddConnectionTypeObserver(this);
    345   observing_network_connectivity_changes_ = true;
    346 
    347   UpdateCredentials(args->credentials);
    348 
    349   NotifyInitializationSuccess();
    350 }
    351 
    352 void SyncManagerImpl::NotifyInitializationSuccess() {
    353   FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
    354                     OnInitializationComplete(
    355                         MakeWeakHandle(weak_ptr_factory_.GetWeakPtr()),
    356                         MakeWeakHandle(debug_info_event_listener_.GetWeakPtr()),
    357                         true, InitialSyncEndedTypes()));
    358 }
    359 
    360 void SyncManagerImpl::NotifyInitializationFailure() {
    361   FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
    362                     OnInitializationComplete(
    363                         MakeWeakHandle(weak_ptr_factory_.GetWeakPtr()),
    364                         MakeWeakHandle(debug_info_event_listener_.GetWeakPtr()),
    365                         false, ModelTypeSet()));
    366 }
    367 
    368 void SyncManagerImpl::OnPassphraseRequired(
    369     PassphraseRequiredReason reason,
    370     const sync_pb::EncryptedData& pending_keys) {
    371   // Does nothing.
    372 }
    373 
    374 void SyncManagerImpl::OnPassphraseAccepted() {
    375   // Does nothing.
    376 }
    377 
    378 void SyncManagerImpl::OnBootstrapTokenUpdated(
    379     const std::string& bootstrap_token,
    380     BootstrapTokenType type) {
    381   if (type == KEYSTORE_BOOTSTRAP_TOKEN)
    382     allstatus_.SetHasKeystoreKey(true);
    383 }
    384 
    385 void SyncManagerImpl::OnEncryptedTypesChanged(ModelTypeSet encrypted_types,
    386                                               bool encrypt_everything) {
    387   allstatus_.SetEncryptedTypes(encrypted_types);
    388 }
    389 
    390 void SyncManagerImpl::OnEncryptionComplete() {
    391   // Does nothing.
    392 }
    393 
    394 void SyncManagerImpl::OnCryptographerStateChanged(
    395     Cryptographer* cryptographer) {
    396   allstatus_.SetCryptographerReady(cryptographer->is_ready());
    397   allstatus_.SetCryptoHasPendingKeys(cryptographer->has_pending_keys());
    398   allstatus_.SetKeystoreMigrationTime(
    399       sync_encryption_handler_->migration_time());
    400 }
    401 
    402 void SyncManagerImpl::OnPassphraseTypeChanged(
    403     PassphraseType type,
    404     base::Time explicit_passphrase_time) {
    405   allstatus_.SetPassphraseType(type);
    406   allstatus_.SetKeystoreMigrationTime(
    407       sync_encryption_handler_->migration_time());
    408 }
    409 
    410 void SyncManagerImpl::StartSyncingNormally(
    411     const ModelSafeRoutingInfo& routing_info) {
    412   // Start the sync scheduler.
    413   // TODO(sync): We always want the newest set of routes when we switch back
    414   // to normal mode. Figure out how to enforce set_routing_info is always
    415   // appropriately set and that it's only modified when switching to normal
    416   // mode.
    417   DCHECK(thread_checker_.CalledOnValidThread());
    418   session_context_->SetRoutingInfo(routing_info);
    419   scheduler_->Start(SyncScheduler::NORMAL_MODE);
    420 }
    421 
    422 syncable::Directory* SyncManagerImpl::directory() {
    423   return share_.directory.get();
    424 }
    425 
    426 const SyncScheduler* SyncManagerImpl::scheduler() const {
    427   return scheduler_.get();
    428 }
    429 
    430 bool SyncManagerImpl::GetHasInvalidAuthTokenForTest() const {
    431   return connection_manager_->HasInvalidAuthToken();
    432 }
    433 
    434 bool SyncManagerImpl::OpenDirectory(const std::string& username) {
    435   DCHECK(!initialized_) << "Should only happen once";
    436 
    437   // Set before Open().
    438   change_observer_ = MakeWeakHandle(js_mutation_event_observer_.AsWeakPtr());
    439   WeakHandle<syncable::TransactionObserver> transaction_observer(
    440       MakeWeakHandle(js_mutation_event_observer_.AsWeakPtr()));
    441 
    442   syncable::DirOpenResult open_result = syncable::NOT_INITIALIZED;
    443   open_result = directory()->Open(username, this, transaction_observer);
    444   if (open_result != syncable::OPENED) {
    445     LOG(ERROR) << "Could not open share for:" << username;
    446     return false;
    447   }
    448 
    449   // Unapplied datatypes (those that do not have initial sync ended set) get
    450   // re-downloaded during any configuration. But, it's possible for a datatype
    451   // to have a progress marker but not have initial sync ended yet, making
    452   // it a candidate for migration. This is a problem, as the DataTypeManager
    453   // does not support a migration while it's already in the middle of a
    454   // configuration. As a result, any partially synced datatype can stall the
    455   // DTM, waiting for the configuration to complete, which it never will due
    456   // to the migration error. In addition, a partially synced nigori will
    457   // trigger the migration logic before the backend is initialized, resulting
    458   // in crashes. We therefore detect and purge any partially synced types as
    459   // part of initialization.
    460   if (!PurgePartiallySyncedTypes())
    461     return false;
    462 
    463   return true;
    464 }
    465 
    466 bool SyncManagerImpl::PurgePartiallySyncedTypes() {
    467   ModelTypeSet partially_synced_types = ModelTypeSet::All();
    468   partially_synced_types.RemoveAll(InitialSyncEndedTypes());
    469   partially_synced_types.RemoveAll(GetTypesWithEmptyProgressMarkerToken(
    470       ModelTypeSet::All()));
    471 
    472   DVLOG(1) << "Purging partially synced types "
    473            << ModelTypeSetToString(partially_synced_types);
    474   UMA_HISTOGRAM_COUNTS("Sync.PartiallySyncedTypes",
    475                        partially_synced_types.Size());
    476   if (partially_synced_types.Empty())
    477     return true;
    478   return directory()->PurgeEntriesWithTypeIn(partially_synced_types,
    479                                              ModelTypeSet(),
    480                                              ModelTypeSet());
    481 }
    482 
    483 bool SyncManagerImpl::PurgeDisabledTypes(
    484     ModelTypeSet to_purge,
    485     ModelTypeSet to_journal,
    486     ModelTypeSet to_unapply) {
    487   if (to_purge.Empty())
    488     return true;
    489   DVLOG(1) << "Purging disabled types " << ModelTypeSetToString(to_purge);
    490   DCHECK(to_purge.HasAll(to_journal));
    491   DCHECK(to_purge.HasAll(to_unapply));
    492   return directory()->PurgeEntriesWithTypeIn(to_purge, to_journal, to_unapply);
    493 }
    494 
    495 void SyncManagerImpl::UpdateCredentials(const SyncCredentials& credentials) {
    496   DCHECK(thread_checker_.CalledOnValidThread());
    497   DCHECK(initialized_);
    498   DCHECK(!credentials.email.empty());
    499   DCHECK(!credentials.sync_token.empty());
    500   DCHECK(!credentials.scope_set.empty());
    501 
    502   observing_network_connectivity_changes_ = true;
    503   if (!connection_manager_->SetAuthToken(credentials.sync_token))
    504     return;  // Auth token is known to be invalid, so exit early.
    505 
    506   scheduler_->OnCredentialsUpdated();
    507 
    508   // TODO(zea): pass the credential age to the debug info event listener.
    509 }
    510 
    511 void SyncManagerImpl::AddObserver(SyncManager::Observer* observer) {
    512   DCHECK(thread_checker_.CalledOnValidThread());
    513   observers_.AddObserver(observer);
    514 }
    515 
    516 void SyncManagerImpl::RemoveObserver(SyncManager::Observer* observer) {
    517   DCHECK(thread_checker_.CalledOnValidThread());
    518   observers_.RemoveObserver(observer);
    519 }
    520 
    521 void SyncManagerImpl::ShutdownOnSyncThread(ShutdownReason reason) {
    522   DCHECK(thread_checker_.CalledOnValidThread());
    523 
    524   // Prevent any in-flight method calls from running.  Also
    525   // invalidates |weak_handle_this_| and |change_observer_|.
    526   weak_ptr_factory_.InvalidateWeakPtrs();
    527   js_mutation_event_observer_.InvalidateWeakPtrs();
    528 
    529   scheduler_.reset();
    530   session_context_.reset();
    531 
    532   if (model_type_registry_)
    533     sync_encryption_handler_->RemoveObserver(model_type_registry_.get());
    534 
    535   model_type_registry_.reset();
    536 
    537   if (sync_encryption_handler_) {
    538     sync_encryption_handler_->RemoveObserver(&debug_info_event_listener_);
    539     sync_encryption_handler_->RemoveObserver(this);
    540   }
    541 
    542   SetJsEventHandler(WeakHandle<JsEventHandler>());
    543   RemoveObserver(&js_sync_manager_observer_);
    544 
    545   RemoveObserver(&debug_info_event_listener_);
    546 
    547   // |connection_manager_| may end up being NULL here in tests (in synchronous
    548   // initialization mode).
    549   //
    550   // TODO(akalin): Fix this behavior.
    551   if (connection_manager_)
    552     connection_manager_->RemoveListener(this);
    553   connection_manager_.reset();
    554 
    555   net::NetworkChangeNotifier::RemoveIPAddressObserver(this);
    556   net::NetworkChangeNotifier::RemoveConnectionTypeObserver(this);
    557   observing_network_connectivity_changes_ = false;
    558 
    559   if (initialized_ && directory()) {
    560     directory()->SaveChanges();
    561   }
    562 
    563   share_.directory.reset();
    564 
    565   change_delegate_ = NULL;
    566 
    567   initialized_ = false;
    568 
    569   // We reset these here, since only now we know they will not be
    570   // accessed from other threads (since we shut down everything).
    571   change_observer_.Reset();
    572   weak_handle_this_.Reset();
    573 }
    574 
    575 void SyncManagerImpl::OnIPAddressChanged() {
    576   if (!observing_network_connectivity_changes_) {
    577     DVLOG(1) << "IP address change dropped.";
    578     return;
    579   }
    580   DVLOG(1) << "IP address change detected.";
    581   OnNetworkConnectivityChangedImpl();
    582 }
    583 
    584 void SyncManagerImpl::OnConnectionTypeChanged(
    585   net::NetworkChangeNotifier::ConnectionType) {
    586   if (!observing_network_connectivity_changes_) {
    587     DVLOG(1) << "Connection type change dropped.";
    588     return;
    589   }
    590   DVLOG(1) << "Connection type change detected.";
    591   OnNetworkConnectivityChangedImpl();
    592 }
    593 
    594 void SyncManagerImpl::OnNetworkConnectivityChangedImpl() {
    595   DCHECK(thread_checker_.CalledOnValidThread());
    596   scheduler_->OnConnectionStatusChange();
    597 }
    598 
    599 void SyncManagerImpl::OnServerConnectionEvent(
    600     const ServerConnectionEvent& event) {
    601   DCHECK(thread_checker_.CalledOnValidThread());
    602   if (event.connection_code ==
    603       HttpResponse::SERVER_CONNECTION_OK) {
    604     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
    605                       OnConnectionStatusChange(CONNECTION_OK));
    606   }
    607 
    608   if (event.connection_code == HttpResponse::SYNC_AUTH_ERROR) {
    609     observing_network_connectivity_changes_ = false;
    610     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
    611                       OnConnectionStatusChange(CONNECTION_AUTH_ERROR));
    612   }
    613 
    614   if (event.connection_code == HttpResponse::SYNC_SERVER_ERROR) {
    615     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
    616                       OnConnectionStatusChange(CONNECTION_SERVER_ERROR));
    617   }
    618 }
    619 
    620 void SyncManagerImpl::HandleTransactionCompleteChangeEvent(
    621     ModelTypeSet models_with_changes) {
    622   // This notification happens immediately after the transaction mutex is
    623   // released. This allows work to be performed without blocking other threads
    624   // from acquiring a transaction.
    625   if (!change_delegate_)
    626     return;
    627 
    628   // Call commit.
    629   for (ModelTypeSet::Iterator it = models_with_changes.First();
    630        it.Good(); it.Inc()) {
    631     change_delegate_->OnChangesComplete(it.Get());
    632     change_observer_.Call(
    633         FROM_HERE,
    634         &SyncManager::ChangeObserver::OnChangesComplete,
    635         it.Get());
    636   }
    637 }
    638 
    639 ModelTypeSet
    640 SyncManagerImpl::HandleTransactionEndingChangeEvent(
    641     const ImmutableWriteTransactionInfo& write_transaction_info,
    642     syncable::BaseTransaction* trans) {
    643   // This notification happens immediately before a syncable WriteTransaction
    644   // falls out of scope. It happens while the channel mutex is still held,
    645   // and while the transaction mutex is held, so it cannot be re-entrant.
    646   if (!change_delegate_ || change_records_.empty())
    647     return ModelTypeSet();
    648 
    649   // This will continue the WriteTransaction using a read only wrapper.
    650   // This is the last chance for read to occur in the WriteTransaction
    651   // that's closing. This special ReadTransaction will not close the
    652   // underlying transaction.
    653   ReadTransaction read_trans(GetUserShare(), trans);
    654 
    655   ModelTypeSet models_with_changes;
    656   for (ChangeRecordMap::const_iterator it = change_records_.begin();
    657       it != change_records_.end(); ++it) {
    658     DCHECK(!it->second.Get().empty());
    659     ModelType type = ModelTypeFromInt(it->first);
    660     change_delegate_->
    661         OnChangesApplied(type, trans->directory()->GetTransactionVersion(type),
    662                          &read_trans, it->second);
    663     change_observer_.Call(FROM_HERE,
    664         &SyncManager::ChangeObserver::OnChangesApplied,
    665         type, write_transaction_info.Get().id, it->second);
    666     models_with_changes.Put(type);
    667   }
    668   change_records_.clear();
    669   return models_with_changes;
    670 }
    671 
    672 void SyncManagerImpl::HandleCalculateChangesChangeEventFromSyncApi(
    673     const ImmutableWriteTransactionInfo& write_transaction_info,
    674     syncable::BaseTransaction* trans,
    675     std::vector<int64>* entries_changed) {
    676   // We have been notified about a user action changing a sync model.
    677   LOG_IF(WARNING, !change_records_.empty()) <<
    678       "CALCULATE_CHANGES called with unapplied old changes.";
    679 
    680   // The mutated model type, or UNSPECIFIED if nothing was mutated.
    681   ModelTypeSet mutated_model_types;
    682 
    683   const syncable::ImmutableEntryKernelMutationMap& mutations =
    684       write_transaction_info.Get().mutations;
    685   for (syncable::EntryKernelMutationMap::const_iterator it =
    686            mutations.Get().begin(); it != mutations.Get().end(); ++it) {
    687     if (!it->second.mutated.ref(syncable::IS_UNSYNCED)) {
    688       continue;
    689     }
    690 
    691     ModelType model_type =
    692         GetModelTypeFromSpecifics(it->second.mutated.ref(SPECIFICS));
    693     if (model_type < FIRST_REAL_MODEL_TYPE) {
    694       NOTREACHED() << "Permanent or underspecified item changed via syncapi.";
    695       continue;
    696     }
    697 
    698     // Found real mutation.
    699     if (model_type != UNSPECIFIED) {
    700       mutated_model_types.Put(model_type);
    701       entries_changed->push_back(it->second.mutated.ref(syncable::META_HANDLE));
    702     }
    703   }
    704 
    705   // Nudge if necessary.
    706   if (!mutated_model_types.Empty()) {
    707     if (weak_handle_this_.IsInitialized()) {
    708       weak_handle_this_.Call(FROM_HERE,
    709                              &SyncManagerImpl::RequestNudgeForDataTypes,
    710                              FROM_HERE,
    711                              mutated_model_types);
    712     } else {
    713       NOTREACHED();
    714     }
    715   }
    716 }
    717 
    718 void SyncManagerImpl::SetExtraChangeRecordData(int64 id,
    719     ModelType type, ChangeReorderBuffer* buffer,
    720     Cryptographer* cryptographer, const syncable::EntryKernel& original,
    721     bool existed_before, bool exists_now) {
    722   // If this is a deletion and the datatype was encrypted, we need to decrypt it
    723   // and attach it to the buffer.
    724   if (!exists_now && existed_before) {
    725     sync_pb::EntitySpecifics original_specifics(original.ref(SPECIFICS));
    726     if (type == PASSWORDS) {
    727       // Passwords must use their own legacy ExtraPasswordChangeRecordData.
    728       scoped_ptr<sync_pb::PasswordSpecificsData> data(
    729           DecryptPasswordSpecifics(original_specifics, cryptographer));
    730       if (!data) {
    731         NOTREACHED();
    732         return;
    733       }
    734       buffer->SetExtraDataForId(id, new ExtraPasswordChangeRecordData(*data));
    735     } else if (original_specifics.has_encrypted()) {
    736       // All other datatypes can just create a new unencrypted specifics and
    737       // attach it.
    738       const sync_pb::EncryptedData& encrypted = original_specifics.encrypted();
    739       if (!cryptographer->Decrypt(encrypted, &original_specifics)) {
    740         NOTREACHED();
    741         return;
    742       }
    743     }
    744     buffer->SetSpecificsForId(id, original_specifics);
    745   }
    746 }
    747 
    748 void SyncManagerImpl::HandleCalculateChangesChangeEventFromSyncer(
    749     const ImmutableWriteTransactionInfo& write_transaction_info,
    750     syncable::BaseTransaction* trans,
    751     std::vector<int64>* entries_changed) {
    752   // We only expect one notification per sync step, so change_buffers_ should
    753   // contain no pending entries.
    754   LOG_IF(WARNING, !change_records_.empty()) <<
    755       "CALCULATE_CHANGES called with unapplied old changes.";
    756 
    757   ChangeReorderBuffer change_buffers[MODEL_TYPE_COUNT];
    758 
    759   Cryptographer* crypto = directory()->GetCryptographer(trans);
    760   const syncable::ImmutableEntryKernelMutationMap& mutations =
    761       write_transaction_info.Get().mutations;
    762   for (syncable::EntryKernelMutationMap::const_iterator it =
    763            mutations.Get().begin(); it != mutations.Get().end(); ++it) {
    764     bool existed_before = !it->second.original.ref(syncable::IS_DEL);
    765     bool exists_now = !it->second.mutated.ref(syncable::IS_DEL);
    766 
    767     // Omit items that aren't associated with a model.
    768     ModelType type =
    769         GetModelTypeFromSpecifics(it->second.mutated.ref(SPECIFICS));
    770     if (type < FIRST_REAL_MODEL_TYPE)
    771       continue;
    772 
    773     int64 handle = it->first;
    774     if (exists_now && !existed_before)
    775       change_buffers[type].PushAddedItem(handle);
    776     else if (!exists_now && existed_before)
    777       change_buffers[type].PushDeletedItem(handle);
    778     else if (exists_now && existed_before &&
    779              VisiblePropertiesDiffer(it->second, crypto)) {
    780       change_buffers[type].PushUpdatedItem(handle);
    781     }
    782 
    783     SetExtraChangeRecordData(handle, type, &change_buffers[type], crypto,
    784                              it->second.original, existed_before, exists_now);
    785   }
    786 
    787   ReadTransaction read_trans(GetUserShare(), trans);
    788   for (int i = FIRST_REAL_MODEL_TYPE; i < MODEL_TYPE_COUNT; ++i) {
    789     if (!change_buffers[i].IsEmpty()) {
    790       if (change_buffers[i].GetAllChangesInTreeOrder(&read_trans,
    791                                                      &(change_records_[i]))) {
    792         for (size_t j = 0; j < change_records_[i].Get().size(); ++j)
    793           entries_changed->push_back((change_records_[i].Get())[j].id);
    794       }
    795       if (change_records_[i].Get().empty())
    796         change_records_.erase(i);
    797     }
    798   }
    799 }
    800 
    801 void SyncManagerImpl::RequestNudgeForDataTypes(
    802     const tracked_objects::Location& nudge_location,
    803     ModelTypeSet types) {
    804   debug_info_event_listener_.OnNudgeFromDatatype(types.First().Get());
    805 
    806   scheduler_->ScheduleLocalNudge(types, nudge_location);
    807 }
    808 
    809 void SyncManagerImpl::NudgeForInitialDownload(syncer::ModelType type) {
    810   DCHECK(thread_checker_.CalledOnValidThread());
    811   scheduler_->ScheduleInitialSyncNudge(type);
    812 }
    813 
    814 void SyncManagerImpl::NudgeForCommit(syncer::ModelType type) {
    815   DCHECK(thread_checker_.CalledOnValidThread());
    816   RequestNudgeForDataTypes(FROM_HERE, ModelTypeSet(type));
    817 }
    818 
    819 void SyncManagerImpl::NudgeForRefresh(syncer::ModelType type) {
    820   DCHECK(thread_checker_.CalledOnValidThread());
    821   RefreshTypes(ModelTypeSet(type));
    822 }
    823 
    824 void SyncManagerImpl::OnSyncCycleEvent(const SyncCycleEvent& event) {
    825   DCHECK(thread_checker_.CalledOnValidThread());
    826   // Only send an event if this is due to a cycle ending and this cycle
    827   // concludes a canonical "sync" process; that is, based on what is known
    828   // locally we are "all happy" and up-to-date.  There may be new changes on
    829   // the server, but we'll get them on a subsequent sync.
    830   //
    831   // Notifications are sent at the end of every sync cycle, regardless of
    832   // whether we should sync again.
    833   if (event.what_happened == SyncCycleEvent::SYNC_CYCLE_ENDED) {
    834     if (!initialized_) {
    835       DVLOG(1) << "OnSyncCycleCompleted not sent because sync api is not "
    836                << "initialized";
    837       return;
    838     }
    839 
    840     DVLOG(1) << "Sending OnSyncCycleCompleted";
    841     FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
    842                       OnSyncCycleCompleted(event.snapshot));
    843   }
    844 }
    845 
    846 void SyncManagerImpl::OnActionableError(const SyncProtocolError& error) {
    847   FOR_EACH_OBSERVER(
    848       SyncManager::Observer, observers_,
    849       OnActionableError(error));
    850 }
    851 
    852 void SyncManagerImpl::OnRetryTimeChanged(base::Time) {}
    853 
    854 void SyncManagerImpl::OnThrottledTypesChanged(ModelTypeSet) {}
    855 
    856 void SyncManagerImpl::OnMigrationRequested(ModelTypeSet types) {
    857   FOR_EACH_OBSERVER(
    858       SyncManager::Observer, observers_,
    859       OnMigrationRequested(types));
    860 }
    861 
    862 void SyncManagerImpl::OnProtocolEvent(const ProtocolEvent& event) {
    863   protocol_event_buffer_.RecordProtocolEvent(event);
    864   FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
    865                     OnProtocolEvent(event));
    866 }
    867 
    868 void SyncManagerImpl::SetJsEventHandler(
    869     const WeakHandle<JsEventHandler>& event_handler) {
    870   js_sync_manager_observer_.SetJsEventHandler(event_handler);
    871   js_mutation_event_observer_.SetJsEventHandler(event_handler);
    872   js_sync_encryption_handler_observer_.SetJsEventHandler(event_handler);
    873 }
    874 
    875 scoped_ptr<base::ListValue> SyncManagerImpl::GetAllNodesForType(
    876     syncer::ModelType type) {
    877   DirectoryTypeDebugInfoEmitterMap* emitter_map =
    878       model_type_registry_->directory_type_debug_info_emitter_map();
    879   DirectoryTypeDebugInfoEmitterMap::iterator it = emitter_map->find(type);
    880 
    881   if (it == emitter_map->end()) {
    882     // This can happen in some cases.  The UI thread makes requests of us
    883     // when it doesn't really know which types are enabled or disabled.
    884     DLOG(WARNING) << "Asked to return debug info for invalid type "
    885                   << ModelTypeToString(type);
    886     return scoped_ptr<base::ListValue>(new base::ListValue());
    887   }
    888 
    889   return it->second->GetAllNodes();
    890 }
    891 
    892 void SyncManagerImpl::SetInvalidatorEnabled(bool invalidator_enabled) {
    893   DCHECK(thread_checker_.CalledOnValidThread());
    894 
    895   DVLOG(1) << "Invalidator enabled state is now: " << invalidator_enabled;
    896   allstatus_.SetNotificationsEnabled(invalidator_enabled);
    897   scheduler_->SetNotificationsEnabled(invalidator_enabled);
    898 }
    899 
    900 void SyncManagerImpl::OnIncomingInvalidation(
    901     syncer::ModelType type,
    902     scoped_ptr<InvalidationInterface> invalidation) {
    903   DCHECK(thread_checker_.CalledOnValidThread());
    904 
    905   scheduler_->ScheduleInvalidationNudge(
    906       type,
    907       invalidation.Pass(),
    908       FROM_HERE);
    909 }
    910 
    911 void SyncManagerImpl::RefreshTypes(ModelTypeSet types) {
    912   DCHECK(thread_checker_.CalledOnValidThread());
    913   if (types.Empty()) {
    914     LOG(WARNING) << "Sync received refresh request with no types specified.";
    915   } else {
    916     scheduler_->ScheduleLocalRefreshRequest(
    917         types, FROM_HERE);
    918   }
    919 }
    920 
    921 SyncStatus SyncManagerImpl::GetDetailedStatus() const {
    922   return allstatus_.status();
    923 }
    924 
    925 void SyncManagerImpl::SaveChanges() {
    926   directory()->SaveChanges();
    927 }
    928 
    929 UserShare* SyncManagerImpl::GetUserShare() {
    930   DCHECK(initialized_);
    931   return &share_;
    932 }
    933 
    934 syncer::SyncContextProxy* SyncManagerImpl::GetSyncContextProxy() {
    935   DCHECK(initialized_);
    936   return sync_context_proxy_.get();
    937 }
    938 
    939 const std::string SyncManagerImpl::cache_guid() {
    940   DCHECK(initialized_);
    941   return directory()->cache_guid();
    942 }
    943 
    944 bool SyncManagerImpl::ReceivedExperiment(Experiments* experiments) {
    945   ReadTransaction trans(FROM_HERE, GetUserShare());
    946   ReadNode nigori_node(&trans);
    947   if (nigori_node.InitTypeRoot(NIGORI) != BaseNode::INIT_OK) {
    948     DVLOG(1) << "Couldn't find Nigori node.";
    949     return false;
    950   }
    951   bool found_experiment = false;
    952 
    953   ReadNode favicon_sync_node(&trans);
    954   if (favicon_sync_node.InitByClientTagLookup(
    955           syncer::EXPERIMENTS,
    956           syncer::kFaviconSyncTag) == BaseNode::INIT_OK) {
    957     experiments->favicon_sync_limit =
    958         favicon_sync_node.GetExperimentsSpecifics().favicon_sync().
    959             favicon_sync_limit();
    960     found_experiment = true;
    961   }
    962 
    963   ReadNode pre_commit_update_avoidance_node(&trans);
    964   if (pre_commit_update_avoidance_node.InitByClientTagLookup(
    965           syncer::EXPERIMENTS,
    966           syncer::kPreCommitUpdateAvoidanceTag) == BaseNode::INIT_OK) {
    967     session_context_->set_server_enabled_pre_commit_update_avoidance(
    968         pre_commit_update_avoidance_node.GetExperimentsSpecifics().
    969             pre_commit_update_avoidance().enabled());
    970     // We don't bother setting found_experiment.  The frontend doesn't need to
    971     // know about this.
    972   }
    973 
    974   ReadNode gcm_channel_node(&trans);
    975   if (gcm_channel_node.InitByClientTagLookup(
    976           syncer::EXPERIMENTS,
    977           syncer::kGCMChannelTag) == BaseNode::INIT_OK &&
    978       gcm_channel_node.GetExperimentsSpecifics().gcm_channel().has_enabled()) {
    979     experiments->gcm_channel_state =
    980         (gcm_channel_node.GetExperimentsSpecifics().gcm_channel().enabled() ?
    981          syncer::Experiments::ENABLED : syncer::Experiments::SUPPRESSED);
    982     found_experiment = true;
    983   }
    984 
    985   ReadNode enhanced_bookmarks_node(&trans);
    986   if (enhanced_bookmarks_node.InitByClientTagLookup(
    987           syncer::EXPERIMENTS, syncer::kEnhancedBookmarksTag) ==
    988           BaseNode::INIT_OK &&
    989       enhanced_bookmarks_node.GetExperimentsSpecifics()
    990           .has_enhanced_bookmarks()) {
    991     const sync_pb::EnhancedBookmarksFlags& enhanced_bookmarks =
    992         enhanced_bookmarks_node.GetExperimentsSpecifics().enhanced_bookmarks();
    993     if (enhanced_bookmarks.has_enabled())
    994       experiments->enhanced_bookmarks_enabled = enhanced_bookmarks.enabled();
    995     if (enhanced_bookmarks.has_extension_id()) {
    996       experiments->enhanced_bookmarks_ext_id =
    997           enhanced_bookmarks.extension_id();
    998     }
    999     found_experiment = true;
   1000   }
   1001 
   1002   ReadNode gcm_invalidations_node(&trans);
   1003   if (gcm_invalidations_node.InitByClientTagLookup(
   1004           syncer::EXPERIMENTS, syncer::kGCMInvalidationsTag) ==
   1005       BaseNode::INIT_OK) {
   1006     const sync_pb::GcmInvalidationsFlags& gcm_invalidations =
   1007         gcm_invalidations_node.GetExperimentsSpecifics().gcm_invalidations();
   1008     if (gcm_invalidations.has_enabled()) {
   1009       experiments->gcm_invalidations_enabled = gcm_invalidations.enabled();
   1010       found_experiment = true;
   1011     }
   1012   }
   1013 
   1014   return found_experiment;
   1015 }
   1016 
   1017 bool SyncManagerImpl::HasUnsyncedItems() {
   1018   ReadTransaction trans(FROM_HERE, GetUserShare());
   1019   return (trans.GetWrappedTrans()->directory()->unsynced_entity_count() != 0);
   1020 }
   1021 
   1022 SyncEncryptionHandler* SyncManagerImpl::GetEncryptionHandler() {
   1023   return sync_encryption_handler_.get();
   1024 }
   1025 
   1026 ScopedVector<syncer::ProtocolEvent>
   1027     SyncManagerImpl::GetBufferedProtocolEvents() {
   1028   return protocol_event_buffer_.GetBufferedProtocolEvents();
   1029 }
   1030 
   1031 void SyncManagerImpl::RegisterDirectoryTypeDebugInfoObserver(
   1032     syncer::TypeDebugInfoObserver* observer) {
   1033   model_type_registry_->RegisterDirectoryTypeDebugInfoObserver(observer);
   1034 }
   1035 
   1036 void SyncManagerImpl::UnregisterDirectoryTypeDebugInfoObserver(
   1037     syncer::TypeDebugInfoObserver* observer) {
   1038   model_type_registry_->UnregisterDirectoryTypeDebugInfoObserver(observer);
   1039 }
   1040 
   1041 bool SyncManagerImpl::HasDirectoryTypeDebugInfoObserver(
   1042     syncer::TypeDebugInfoObserver* observer) {
   1043   return model_type_registry_->HasDirectoryTypeDebugInfoObserver(observer);
   1044 }
   1045 
   1046 void SyncManagerImpl::RequestEmitDebugInfo() {
   1047   model_type_registry_->RequestEmitDebugInfo();
   1048 }
   1049 
   1050 }  // namespace syncer
   1051