sync/engine/get_updates_processor.cc

   1 // Copyright 2014 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/engine/get_updates_processor.h"
   6
   7 #include <map>
   8
   9 #include "base/trace_event/trace_event.h"
  10 #include "sync/engine/get_updates_delegate.h"
  11 #include "sync/engine/syncer_proto_util.h"
  12 #include "sync/engine/update_handler.h"
  13 #include "sync/internal_api/public/events/get_updates_response_event.h"
  14 #include "sync/protocol/sync.pb.h"
  15 #include "sync/sessions/status_controller.h"
  16 #include "sync/sessions/sync_session.h"
  17 #include "sync/syncable/directory.h"
  18 #include "sync/syncable/nigori_handler.h"
  19 #include "sync/syncable/syncable_read_transaction.h"
  20
  21 typedef std::vector<const sync_pb::SyncEntity*> SyncEntityList;
  22 typedef std::map<syncer::ModelType, SyncEntityList> TypeSyncEntityMap;
  23
  24 namespace syncer {
  25
  26 typedef std::map<ModelType, size_t> TypeToIndexMap;
  27
  28 namespace {
  29
  30 bool ShouldRequestEncryptionKey(sessions::SyncSessionContext* context) {
  31   syncable::Directory* dir = context->directory();
  32   syncable::ReadTransaction trans(FROM_HERE, dir);
  33   syncable::NigoriHandler* nigori_handler = dir->GetNigoriHandler();
  34   return nigori_handler->NeedKeystoreKey(&trans);
  35 }
  36
  37
  38 SyncerError HandleGetEncryptionKeyResponse(
  39     const sync_pb::ClientToServerResponse& update_response,
  40     syncable::Directory* dir) {
  41   bool success = false;
  42   if (update_response.get_updates().encryption_keys_size() == 0) {
  43     LOG(ERROR) << "Failed to receive encryption key from server.";
  44     return SERVER_RESPONSE_VALIDATION_FAILED;
  45   }
  46   syncable::ReadTransaction trans(FROM_HERE, dir);
  47   syncable::NigoriHandler* nigori_handler = dir->GetNigoriHandler();
  48   success = nigori_handler->SetKeystoreKeys(
  49       update_response.get_updates().encryption_keys(),
  50       &trans);
  51
  52   DVLOG(1) << "GetUpdates returned "
  53            << update_response.get_updates().encryption_keys_size()
  54            << "encryption keys. Nigori keystore key "
  55            << (success ? "" : "not ") << "updated.";
  56   return (success ? SYNCER_OK : SERVER_RESPONSE_VALIDATION_FAILED);
  57 }
  58
  59 // Given a GetUpdates response, iterates over all the returned items and
  60 // divides them according to their type.  Outputs a map from model types to
  61 // received SyncEntities.  The output map will have entries (possibly empty)
  62 // for all types in |requested_types|.
  63 void PartitionUpdatesByType(const sync_pb::GetUpdatesResponse& gu_response,
  64                             ModelTypeSet requested_types,
  65                             TypeSyncEntityMap* updates_by_type) {
  66   int update_count = gu_response.entries().size();
  67   for (ModelTypeSet::Iterator it = requested_types.First();
  68        it.Good(); it.Inc()) {
  69     updates_by_type->insert(std::make_pair(it.Get(), SyncEntityList()));
  70   }
  71   for (int i = 0; i < update_count; ++i) {
  72     const sync_pb::SyncEntity& update = gu_response.entries(i);
  73     ModelType type = GetModelType(update);
  74     if (!IsRealDataType(type)) {
  75       NOTREACHED() << "Received update with invalid type.";
  76       continue;
  77     }
  78
  79     TypeSyncEntityMap::iterator it = updates_by_type->find(type);
  80     if (it == updates_by_type->end()) {
  81       DLOG(WARNING)
  82           << "Received update for unexpected type or the type is throttled:"
  83           << ModelTypeToString(type);
  84       continue;
  85     }
  86
  87     it->second.push_back(&update);
  88   }
  89 }
  90
  91 // Builds a map of ModelTypes to indices to progress markers in the given
  92 // |gu_response| message.  The map is returned in the |index_map| parameter.
  93 void PartitionProgressMarkersByType(
  94     const sync_pb::GetUpdatesResponse& gu_response,
  95     ModelTypeSet request_types,
  96     TypeToIndexMap* index_map) {
  97   for (int i = 0; i < gu_response.new_progress_marker_size(); ++i) {
  98     int field_number = gu_response.new_progress_marker(i).data_type_id();
  99     ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
 100     if (!IsRealDataType(model_type)) {
 101       DLOG(WARNING) << "Unknown field number " << field_number;
 102       continue;
 103     }
 104     if (!request_types.Has(model_type)) {
 105       DLOG(WARNING)
 106           << "Skipping unexpected progress marker for non-enabled type "
 107           << ModelTypeToString(model_type);
 108       continue;
 109     }
 110     index_map->insert(std::make_pair(model_type, i));
 111   }
 112 }
 113
 114 void PartitionContextMutationsByType(
 115     const sync_pb::GetUpdatesResponse& gu_response,
 116     ModelTypeSet request_types,
 117     TypeToIndexMap* index_map) {
 118   for (int i = 0; i < gu_response.context_mutations_size(); ++i) {
 119     int field_number = gu_response.context_mutations(i).data_type_id();
 120     ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
 121     if (!IsRealDataType(model_type)) {
 122       DLOG(WARNING) << "Unknown field number " << field_number;
 123       continue;
 124     }
 125     if (!request_types.Has(model_type)) {
 126       DLOG(WARNING)
 127           << "Skipping unexpected context mutation for non-enabled type "
 128           << ModelTypeToString(model_type);
 129       continue;
 130     }
 131     index_map->insert(std::make_pair(model_type, i));
 132   }
 133 }
 134
 135 // Initializes the parts of the GetUpdatesMessage that depend on shared state,
 136 // like the ShouldRequestEncryptionKey() status.  This is kept separate from the
 137 // other of the message-building functions to make the rest of the code easier
 138 // to test.
 139 void InitDownloadUpdatesContext(
 140     sessions::SyncSession* session,
 141     bool create_mobile_bookmarks_folder,
 142     sync_pb::ClientToServerMessage* message) {
 143   message->set_share(session->context()->account_name());
 144   message->set_message_contents(sync_pb::ClientToServerMessage::GET_UPDATES);
 145
 146   sync_pb::GetUpdatesMessage* get_updates = message->mutable_get_updates();
 147
 148   // We want folders for our associated types, always.  If we were to set
 149   // this to false, the server would send just the non-container items
 150   // (e.g. Bookmark URLs but not their containing folders).
 151   get_updates->set_fetch_folders(true);
 152
 153   get_updates->set_create_mobile_bookmarks_folder(
 154       create_mobile_bookmarks_folder);
 155   bool need_encryption_key = ShouldRequestEncryptionKey(session->context());
 156   get_updates->set_need_encryption_key(need_encryption_key);
 157
 158   // Set legacy GetUpdatesMessage.GetUpdatesCallerInfo information.
 159   get_updates->mutable_caller_info()->set_notifications_enabled(
 160       session->context()->notifications_enabled());
 161 }
 162
 163 }  // namespace
 164
 165 GetUpdatesProcessor::GetUpdatesProcessor(UpdateHandlerMap* update_handler_map,
 166                                          const GetUpdatesDelegate& delegate)
 167     : update_handler_map_(update_handler_map), delegate_(delegate) {}
 168
 169 GetUpdatesProcessor::~GetUpdatesProcessor() {}
 170
 171 SyncerError GetUpdatesProcessor::DownloadUpdates(
 172     ModelTypeSet* request_types,
 173     sessions::SyncSession* session,
 174     bool create_mobile_bookmarks_folder) {
 175   TRACE_EVENT0("sync", "DownloadUpdates");
 176
 177   sync_pb::ClientToServerMessage message;
 178   InitDownloadUpdatesContext(session,
 179                              create_mobile_bookmarks_folder,
 180                              &message);
 181   PrepareGetUpdates(*request_types, &message);
 182
 183   SyncerError result = ExecuteDownloadUpdates(request_types, session, &message);
 184   session->mutable_status_controller()->set_last_download_updates_result(
 185       result);
 186   return result;
 187 }
 188
 189 void GetUpdatesProcessor::PrepareGetUpdates(
 190     ModelTypeSet gu_types,
 191     sync_pb::ClientToServerMessage* message) {
 192   sync_pb::GetUpdatesMessage* get_updates = message->mutable_get_updates();
 193
 194   for (ModelTypeSet::Iterator it = gu_types.First(); it.Good(); it.Inc()) {
 195     UpdateHandlerMap::iterator handler_it = update_handler_map_->find(it.Get());
 196     DCHECK(handler_it != update_handler_map_->end())
 197         << "Failed to look up handler for " << ModelTypeToString(it.Get());
 198     sync_pb::DataTypeProgressMarker* progress_marker =
 199         get_updates->add_from_progress_marker();
 200     handler_it->second->GetDownloadProgress(progress_marker);
 201     progress_marker->clear_gc_directive();
 202
 203     sync_pb::DataTypeContext context;
 204     handler_it->second->GetDataTypeContext(&context);
 205     if (!context.context().empty())
 206       get_updates->add_client_contexts()->Swap(&context);
 207   }
 208
 209   delegate_.HelpPopulateGuMessage(get_updates);
 210 }
 211
 212 SyncerError GetUpdatesProcessor::ExecuteDownloadUpdates(
 213     ModelTypeSet* request_types,
 214     sessions::SyncSession* session,
 215     sync_pb::ClientToServerMessage* msg) {
 216   sync_pb::ClientToServerResponse update_response;
 217   sessions::StatusController* status = session->mutable_status_controller();
 218   bool need_encryption_key = ShouldRequestEncryptionKey(session->context());
 219
 220   if (session->context()->debug_info_getter()) {
 221     sync_pb::DebugInfo* debug_info = msg->mutable_debug_info();
 222     CopyClientDebugInfo(session->context()->debug_info_getter(), debug_info);
 223   }
 224
 225   session->SendProtocolEvent(
 226       *(delegate_.GetNetworkRequestEvent(base::Time::Now(), *msg)));
 227
 228   ModelTypeSet partial_failure_data_types;
 229
 230   SyncerError result = SyncerProtoUtil::PostClientToServerMessage(
 231       msg, &update_response, session, &partial_failure_data_types);
 232
 233   DVLOG(2) << SyncerProtoUtil::ClientToServerResponseDebugString(
 234       update_response);
 235
 236   if (result == SERVER_RETURN_PARTIAL_FAILURE) {
 237     request_types->RemoveAll(partial_failure_data_types);
 238   } else if (result != SYNCER_OK) {
 239     GetUpdatesResponseEvent response_event(
 240         base::Time::Now(), update_response, result);
 241     session->SendProtocolEvent(response_event);
 242
 243     // Sync authorization expires every 60 mintues, so SYNC_AUTH_ERROR will
 244     // appear every 60 minutes, and then sync services will refresh the
 245     // authorization. Therefore SYNC_AUTH_ERROR is excluded here to reduce the
 246     // ERROR messages in the log.
 247     if (result != SYNC_AUTH_ERROR) {
 248       LOG(ERROR) << "PostClientToServerMessage() failed during GetUpdates";
 249     }
 250
 251     return result;
 252   }
 253
 254   DVLOG(1) << "GetUpdates returned "
 255            << update_response.get_updates().entries_size()
 256            << " updates.";
 257
 258
 259   if (session->context()->debug_info_getter()) {
 260     // Clear debug info now that we have successfully sent it to the server.
 261     DVLOG(1) << "Clearing client debug info.";
 262     session->context()->debug_info_getter()->ClearDebugInfo();
 263   }
 264
 265   if (need_encryption_key ||
 266       update_response.get_updates().encryption_keys_size() > 0) {
 267     syncable::Directory* dir = session->context()->directory();
 268     status->set_last_get_key_result(
 269         HandleGetEncryptionKeyResponse(update_response, dir));
 270   }
 271
 272   SyncerError process_result =
 273       ProcessResponse(update_response.get_updates(), *request_types, status);
 274
 275   GetUpdatesResponseEvent response_event(
 276       base::Time::Now(), update_response, process_result);
 277   session->SendProtocolEvent(response_event);
 278
 279   DVLOG(1) << "GetUpdates result: " << process_result;
 280
 281   return process_result;
 282 }
 283
 284 SyncerError GetUpdatesProcessor::ProcessResponse(
 285     const sync_pb::GetUpdatesResponse& gu_response,
 286     ModelTypeSet request_types,
 287     sessions::StatusController* status) {
 288   status->increment_num_updates_downloaded_by(gu_response.entries_size());
 289
 290   // The changes remaining field is used to prevent the client from looping.  If
 291   // that field is being set incorrectly, we're in big trouble.
 292   if (!gu_response.has_changes_remaining()) {
 293     return SERVER_RESPONSE_VALIDATION_FAILED;
 294   }
 295
 296   syncer::SyncerError result =
 297       ProcessGetUpdatesResponse(request_types, gu_response, status);
 298   if (result != syncer::SYNCER_OK)
 299     return result;
 300
 301   if (gu_response.changes_remaining() == 0) {
 302     return SYNCER_OK;
 303   } else {
 304     return SERVER_MORE_TO_DOWNLOAD;
 305   }
 306 }
 307
 308 syncer::SyncerError GetUpdatesProcessor::ProcessGetUpdatesResponse(
 309     ModelTypeSet gu_types,
 310     const sync_pb::GetUpdatesResponse& gu_response,
 311     sessions::StatusController* status_controller) {
 312   TypeSyncEntityMap updates_by_type;
 313   PartitionUpdatesByType(gu_response, gu_types, &updates_by_type);
 314   DCHECK_EQ(gu_types.Size(), updates_by_type.size());
 315
 316   TypeToIndexMap progress_index_by_type;
 317   PartitionProgressMarkersByType(gu_response,
 318                                  gu_types,
 319                                  &progress_index_by_type);
 320   if (gu_types.Size() != progress_index_by_type.size()) {
 321     NOTREACHED() << "Missing progress markers in GetUpdates response.";
 322     return syncer::SERVER_RESPONSE_VALIDATION_FAILED;
 323   }
 324
 325   TypeToIndexMap context_by_type;
 326   PartitionContextMutationsByType(gu_response, gu_types, &context_by_type);
 327
 328   // Iterate over these maps in parallel, processing updates for each type.
 329   TypeToIndexMap::iterator progress_marker_iter =
 330       progress_index_by_type.begin();
 331   TypeSyncEntityMap::iterator updates_iter = updates_by_type.begin();
 332   for (; (progress_marker_iter != progress_index_by_type.end()
 333            && updates_iter != updates_by_type.end());
 334        ++progress_marker_iter, ++updates_iter) {
 335     DCHECK_EQ(progress_marker_iter->first, updates_iter->first);
 336     ModelType type = progress_marker_iter->first;
 337
 338     UpdateHandlerMap::iterator update_handler_iter =
 339         update_handler_map_->find(type);
 340
 341     sync_pb::DataTypeContext context;
 342     TypeToIndexMap::iterator context_iter = context_by_type.find(type);
 343     if (context_iter != context_by_type.end())
 344       context.CopyFrom(gu_response.context_mutations(context_iter->second));
 345
 346     if (update_handler_iter != update_handler_map_->end()) {
 347       syncer::SyncerError result =
 348           update_handler_iter->second->ProcessGetUpdatesResponse(
 349               gu_response.new_progress_marker(progress_marker_iter->second),
 350               context,
 351               updates_iter->second,
 352               status_controller);
 353       if (result != syncer::SYNCER_OK)
 354         return result;
 355     } else {
 356       DLOG(WARNING)
 357           << "Ignoring received updates of a type we can't handle.  "
 358           << "Type is: " << ModelTypeToString(type);
 359       continue;
 360     }
 361   }
 362   DCHECK(progress_marker_iter == progress_index_by_type.end() &&
 363          updates_iter == updates_by_type.end());
 364
 365   return syncer::SYNCER_OK;
 366 }
 367
 368 void GetUpdatesProcessor::ApplyUpdates(
 369     ModelTypeSet gu_types,
 370     sessions::StatusController* status_controller) {
 371   delegate_.ApplyUpdates(gu_types, status_controller, update_handler_map_);
 372 }
 373
 374 void GetUpdatesProcessor::CopyClientDebugInfo(
 375     sessions::DebugInfoGetter* debug_info_getter,
 376     sync_pb::DebugInfo* debug_info) {
 377   DVLOG(1) << "Copying client debug info to send.";
 378   debug_info_getter->GetDebugInfo(debug_info);
 379 }
 380
 381 }  // namespace syncer