get_updates_processor.cc 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366
  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. #include "components/sync/engine/get_updates_processor.h"
  5. #include <stddef.h>
  6. #include <map>
  7. #include <string>
  8. #include <utility>
  9. #include <vector>
  10. #include "base/logging.h"
  11. #include "base/trace_event/trace_event.h"
  12. #include "components/sync/engine/cycle/status_controller.h"
  13. #include "components/sync/engine/cycle/sync_cycle.h"
  14. #include "components/sync/engine/events/get_updates_response_event.h"
  15. #include "components/sync/engine/get_updates_delegate.h"
  16. #include "components/sync/engine/nigori/keystore_keys_handler.h"
  17. #include "components/sync/engine/syncer_proto_util.h"
  18. #include "components/sync/engine/update_handler.h"
  19. #include "components/sync/protocol/data_type_progress_marker.pb.h"
  20. #include "components/sync/protocol/sync.pb.h"
  21. #include "components/sync/protocol/sync_entity.pb.h"
  22. #include "third_party/protobuf/src/google/protobuf/repeated_field.h"
  23. namespace syncer {
  24. namespace {
  25. using SyncEntityList = std::vector<const sync_pb::SyncEntity*>;
  26. using TypeSyncEntityMap = std::map<ModelType, SyncEntityList>;
  27. using TypeToIndexMap = std::map<ModelType, size_t>;
  28. bool ShouldRequestEncryptionKey(SyncCycleContext* context) {
  29. return context->model_type_registry()
  30. ->keystore_keys_handler()
  31. ->NeedKeystoreKey();
  32. }
  33. std::vector<std::vector<uint8_t>> ExtractKeystoreKeys(
  34. const sync_pb::ClientToServerResponse& update_response) {
  35. const google::protobuf::RepeatedPtrField<std::string>& encryption_keys =
  36. update_response.get_updates().encryption_keys();
  37. std::vector<std::vector<uint8_t>> keystore_keys;
  38. keystore_keys.reserve(encryption_keys.size());
  39. for (const std::string& key : encryption_keys)
  40. keystore_keys.emplace_back(key.begin(), key.end());
  41. return keystore_keys;
  42. }
  43. SyncerError HandleGetEncryptionKeyResponse(
  44. const sync_pb::ClientToServerResponse& update_response,
  45. SyncCycleContext* context) {
  46. bool success = false;
  47. if (update_response.get_updates().encryption_keys_size() == 0) {
  48. LOG(ERROR) << "Failed to receive encryption key from server.";
  49. return SyncerError(SyncerError::SERVER_RESPONSE_VALIDATION_FAILED);
  50. }
  51. std::vector<std::vector<uint8_t>> keystore_keys =
  52. ExtractKeystoreKeys(update_response);
  53. success =
  54. context->model_type_registry()->keystore_keys_handler()->SetKeystoreKeys(
  55. keystore_keys);
  56. DVLOG(1) << "GetUpdates returned "
  57. << update_response.get_updates().encryption_keys_size()
  58. << "encryption keys. Nigori keystore key " << (success ? "" : "not ")
  59. << "updated.";
  60. return (success
  61. ? SyncerError(SyncerError::SYNCER_OK)
  62. : SyncerError(SyncerError::SERVER_RESPONSE_VALIDATION_FAILED));
  63. }
  64. // Given a GetUpdates response, iterates over all the returned items and
  65. // divides them according to their type. Outputs a map from model types to
  66. // received SyncEntities. The output map will have entries (possibly empty)
  67. // for all types in |requested_types|.
  68. void PartitionUpdatesByType(const sync_pb::GetUpdatesResponse& gu_response,
  69. ModelTypeSet requested_types,
  70. TypeSyncEntityMap* updates_by_type) {
  71. for (ModelType type : requested_types) {
  72. updates_by_type->insert(std::make_pair(type, SyncEntityList()));
  73. }
  74. for (const sync_pb::SyncEntity& update : gu_response.entries()) {
  75. ModelType type = GetModelTypeFromSpecifics(update.specifics());
  76. if (!IsRealDataType(type)) {
  77. NOTREACHED() << "Received update with invalid type.";
  78. continue;
  79. }
  80. auto it = updates_by_type->find(type);
  81. if (it == updates_by_type->end()) {
  82. DLOG(WARNING) << "Received update for unexpected type, or the type is "
  83. "throttled or failed with partial failure:"
  84. << ModelTypeToDebugString(type);
  85. continue;
  86. }
  87. it->second.push_back(&update);
  88. }
  89. }
  90. // Builds a map of ModelTypes to indices to progress markers in the given
  91. // |gu_response| message. The map is returned in the |index_map| parameter.
  92. void PartitionProgressMarkersByType(
  93. const sync_pb::GetUpdatesResponse& gu_response,
  94. const ModelTypeSet& request_types,
  95. TypeToIndexMap* index_map) {
  96. for (int i = 0; i < gu_response.new_progress_marker_size(); ++i) {
  97. int field_number = gu_response.new_progress_marker(i).data_type_id();
  98. ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
  99. if (!IsRealDataType(model_type)) {
  100. DLOG(WARNING) << "Unknown field number " << field_number;
  101. continue;
  102. }
  103. if (!request_types.Has(model_type)) {
  104. DLOG(WARNING)
  105. << "Skipping unexpected progress marker for non-enabled type "
  106. << ModelTypeToDebugString(model_type);
  107. continue;
  108. }
  109. index_map->insert(std::make_pair(model_type, i));
  110. }
  111. }
  112. void PartitionContextMutationsByType(
  113. const sync_pb::GetUpdatesResponse& gu_response,
  114. const ModelTypeSet& request_types,
  115. TypeToIndexMap* index_map) {
  116. for (int i = 0; i < gu_response.context_mutations_size(); ++i) {
  117. int field_number = gu_response.context_mutations(i).data_type_id();
  118. ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
  119. if (!IsRealDataType(model_type)) {
  120. DLOG(WARNING) << "Unknown field number " << field_number;
  121. continue;
  122. }
  123. if (!request_types.Has(model_type)) {
  124. DLOG(WARNING)
  125. << "Skipping unexpected context mutation for non-enabled type "
  126. << ModelTypeToDebugString(model_type);
  127. continue;
  128. }
  129. index_map->insert(std::make_pair(model_type, i));
  130. }
  131. }
  132. // Initializes the parts of the GetUpdatesMessage that depend on shared state,
  133. // like the ShouldRequestEncryptionKey() status. This is kept separate from the
  134. // other of the message-building functions to make the rest of the code easier
  135. // to test.
  136. void InitDownloadUpdatesContext(SyncCycle* cycle,
  137. sync_pb::ClientToServerMessage* message) {
  138. message->set_share(cycle->context()->account_name());
  139. message->set_message_contents(sync_pb::ClientToServerMessage::GET_UPDATES);
  140. sync_pb::GetUpdatesMessage* get_updates = message->mutable_get_updates();
  141. // We want folders for our associated types, always. If we were to set
  142. // this to false, the server would send just the non-container items
  143. // (e.g. Bookmark URLs but not their containing folders).
  144. get_updates->set_fetch_folders(true);
  145. // This is a deprecated field that should be cleaned up after server's
  146. // behavior is updated.
  147. get_updates->set_create_mobile_bookmarks_folder(true);
  148. bool need_encryption_key = ShouldRequestEncryptionKey(cycle->context());
  149. get_updates->set_need_encryption_key(need_encryption_key);
  150. get_updates->mutable_caller_info()->set_notifications_enabled(
  151. cycle->context()->notifications_enabled());
  152. }
  153. } // namespace
  154. GetUpdatesProcessor::GetUpdatesProcessor(UpdateHandlerMap* update_handler_map,
  155. const GetUpdatesDelegate& delegate)
  156. : update_handler_map_(update_handler_map), delegate_(delegate) {}
  157. GetUpdatesProcessor::~GetUpdatesProcessor() = default;
  158. SyncerError GetUpdatesProcessor::DownloadUpdates(ModelTypeSet* request_types,
  159. SyncCycle* cycle) {
  160. TRACE_EVENT0("sync", "DownloadUpdates");
  161. sync_pb::ClientToServerMessage message;
  162. InitDownloadUpdatesContext(cycle, &message);
  163. PrepareGetUpdates(*request_types, &message);
  164. SyncerError result = ExecuteDownloadUpdates(request_types, cycle, &message);
  165. cycle->mutable_status_controller()->set_last_download_updates_result(result);
  166. return result;
  167. }
  168. void GetUpdatesProcessor::PrepareGetUpdates(
  169. const ModelTypeSet& gu_types,
  170. sync_pb::ClientToServerMessage* message) {
  171. sync_pb::GetUpdatesMessage* get_updates = message->mutable_get_updates();
  172. for (ModelType type : gu_types) {
  173. auto handler_it = update_handler_map_->find(type);
  174. DCHECK(handler_it != update_handler_map_->end())
  175. << "Failed to look up handler for " << ModelTypeToDebugString(type);
  176. sync_pb::DataTypeProgressMarker* progress_marker =
  177. get_updates->add_from_progress_marker();
  178. *progress_marker = handler_it->second->GetDownloadProgress();
  179. progress_marker->clear_gc_directive();
  180. sync_pb::DataTypeContext context = handler_it->second->GetDataTypeContext();
  181. if (!context.context().empty())
  182. *get_updates->add_client_contexts() = std::move(context);
  183. }
  184. delegate_.HelpPopulateGuMessage(get_updates);
  185. }
  186. SyncerError GetUpdatesProcessor::ExecuteDownloadUpdates(
  187. ModelTypeSet* request_types,
  188. SyncCycle* cycle,
  189. sync_pb::ClientToServerMessage* msg) {
  190. sync_pb::ClientToServerResponse update_response;
  191. StatusController* status = cycle->mutable_status_controller();
  192. bool need_encryption_key = ShouldRequestEncryptionKey(cycle->context());
  193. if (cycle->context()->debug_info_getter()) {
  194. *msg->mutable_debug_info() =
  195. cycle->context()->debug_info_getter()->GetDebugInfo();
  196. }
  197. SyncerProtoUtil::AddRequiredFieldsToClientToServerMessage(cycle, msg);
  198. cycle->SendProtocolEvent(
  199. *(delegate_.GetNetworkRequestEvent(base::Time::Now(), *msg)));
  200. ModelTypeSet partial_failure_data_types;
  201. SyncerError result = SyncerProtoUtil::PostClientToServerMessage(
  202. *msg, &update_response, cycle, &partial_failure_data_types);
  203. DVLOG(2) << SyncerProtoUtil::ClientToServerResponseDebugString(
  204. update_response);
  205. if (!partial_failure_data_types.Empty()) {
  206. request_types->RemoveAll(partial_failure_data_types);
  207. }
  208. if (result.value() != SyncerError::SYNCER_OK) {
  209. GetUpdatesResponseEvent response_event(base::Time::Now(), update_response,
  210. result);
  211. cycle->SendProtocolEvent(response_event);
  212. // Sync authorization expires every 60 mintues, so SYNC_AUTH_ERROR will
  213. // appear every 60 minutes, and then sync services will refresh the
  214. // authorization. Therefore SYNC_AUTH_ERROR is excluded here to reduce the
  215. // ERROR messages in the log.
  216. if (result.value() != SyncerError::SYNC_AUTH_ERROR) {
  217. LOG(ERROR) << "PostClientToServerMessage() failed during GetUpdates "
  218. "with error "
  219. << result.value();
  220. }
  221. return result;
  222. }
  223. DVLOG(1) << "GetUpdates returned "
  224. << update_response.get_updates().entries_size() << " updates.";
  225. if (cycle->context()->debug_info_getter()) {
  226. // Clear debug info now that we have successfully sent it to the server.
  227. DVLOG(1) << "Clearing client debug info.";
  228. cycle->context()->debug_info_getter()->ClearDebugInfo();
  229. }
  230. if (need_encryption_key ||
  231. update_response.get_updates().encryption_keys_size() > 0) {
  232. status->set_last_get_key_result(
  233. HandleGetEncryptionKeyResponse(update_response, cycle->context()));
  234. }
  235. SyncerError process_result =
  236. ProcessResponse(update_response.get_updates(), *request_types, status);
  237. GetUpdatesResponseEvent response_event(base::Time::Now(), update_response,
  238. process_result);
  239. cycle->SendProtocolEvent(response_event);
  240. DVLOG(1) << "GetUpdates result: " << process_result.ToString();
  241. return process_result;
  242. }
  243. SyncerError GetUpdatesProcessor::ProcessResponse(
  244. const sync_pb::GetUpdatesResponse& gu_response,
  245. const ModelTypeSet& gu_types,
  246. StatusController* status_controller) {
  247. status_controller->increment_num_updates_downloaded_by(
  248. gu_response.entries_size());
  249. // The changes remaining field is used to prevent the client from looping. If
  250. // that field is being set incorrectly, we're in big trouble.
  251. if (!gu_response.has_changes_remaining()) {
  252. return SyncerError(SyncerError::SERVER_RESPONSE_VALIDATION_FAILED);
  253. }
  254. TypeSyncEntityMap updates_by_type;
  255. PartitionUpdatesByType(gu_response, gu_types, &updates_by_type);
  256. DCHECK_EQ(gu_types.Size(), updates_by_type.size());
  257. TypeToIndexMap progress_index_by_type;
  258. PartitionProgressMarkersByType(gu_response, gu_types,
  259. &progress_index_by_type);
  260. if (gu_types.Size() != progress_index_by_type.size()) {
  261. NOTREACHED() << "Missing progress markers in GetUpdates response.";
  262. return SyncerError(SyncerError::SERVER_RESPONSE_VALIDATION_FAILED);
  263. }
  264. TypeToIndexMap context_by_type;
  265. PartitionContextMutationsByType(gu_response, gu_types, &context_by_type);
  266. // Iterate over these maps in parallel, processing updates for each type.
  267. auto progress_marker_iter = progress_index_by_type.begin();
  268. auto updates_iter = updates_by_type.begin();
  269. for (; (progress_marker_iter != progress_index_by_type.end() &&
  270. updates_iter != updates_by_type.end());
  271. ++progress_marker_iter, ++updates_iter) {
  272. DCHECK_EQ(progress_marker_iter->first, updates_iter->first);
  273. ModelType type = progress_marker_iter->first;
  274. auto update_handler_iter = update_handler_map_->find(type);
  275. sync_pb::DataTypeContext context;
  276. auto context_iter = context_by_type.find(type);
  277. if (context_iter != context_by_type.end())
  278. context.CopyFrom(gu_response.context_mutations(context_iter->second));
  279. if (update_handler_iter != update_handler_map_->end()) {
  280. update_handler_iter->second->ProcessGetUpdatesResponse(
  281. gu_response.new_progress_marker(progress_marker_iter->second),
  282. context, updates_iter->second, status_controller);
  283. } else {
  284. DLOG(WARNING) << "Ignoring received updates of a type we can't handle. "
  285. << "Type is: " << ModelTypeToDebugString(type);
  286. continue;
  287. }
  288. }
  289. DCHECK(progress_marker_iter == progress_index_by_type.end() &&
  290. updates_iter == updates_by_type.end());
  291. return gu_response.changes_remaining() == 0
  292. ? SyncerError(SyncerError::SYNCER_OK)
  293. : SyncerError(SyncerError::SERVER_MORE_TO_DOWNLOAD);
  294. }
  295. void GetUpdatesProcessor::ApplyUpdates(const ModelTypeSet& gu_types,
  296. StatusController* status_controller) {
  297. for (const auto& [type, update_handler] : *update_handler_map_) {
  298. if (gu_types.Has(type)) {
  299. update_handler->ApplyUpdates(status_controller);
  300. }
  301. }
  302. }
  303. } // namespace syncer