syncable_service_based_bridge.cc 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624
  1. // Copyright 2018 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/model/syncable_service_based_bridge.h"
  5. #include <stdint.h>
  6. #include <utility>
  7. #include "base/bind.h"
  8. #include "base/callback_helpers.h"
  9. #include "base/location.h"
  10. #include "base/logging.h"
  11. #include "base/memory/raw_ptr.h"
  12. #include "base/trace_event/memory_usage_estimator.h"
  13. #include "components/sync/base/client_tag_hash.h"
  14. #include "components/sync/model/client_tag_based_model_type_processor.h"
  15. #include "components/sync/model/conflict_resolution.h"
  16. #include "components/sync/model/mutable_data_batch.h"
  17. #include "components/sync/model/sync_change.h"
  18. #include "components/sync/model/sync_error_factory.h"
  19. #include "components/sync/model/syncable_service.h"
  20. #include "components/sync/protocol/entity_specifics.pb.h"
  21. #include "components/sync/protocol/persisted_entity_data.pb.h"
  22. #include "components/sync/protocol/proto_memory_estimations.h"
  23. namespace syncer {
  24. namespace {
  25. std::unique_ptr<EntityData> ConvertPersistedToEntityData(
  26. const ClientTagHash& client_tag_hash,
  27. sync_pb::PersistedEntityData data) {
  28. DCHECK(!client_tag_hash.value().empty());
  29. auto entity_data = std::make_unique<EntityData>();
  30. entity_data->name = std::move(*data.mutable_name());
  31. entity_data->specifics = std::move(*data.mutable_specifics());
  32. entity_data->client_tag_hash = client_tag_hash;
  33. // Purposefully crash if we have client only data, as this could result in
  34. // sending password in plain text.
  35. CHECK(!entity_data->specifics.password().has_client_only_encrypted_data());
  36. return entity_data;
  37. }
  38. sync_pb::PersistedEntityData CreatePersistedFromRemoteData(
  39. const EntityData& entity_data) {
  40. sync_pb::PersistedEntityData persisted;
  41. persisted.set_name(entity_data.name);
  42. *persisted.mutable_specifics() = entity_data.specifics;
  43. return persisted;
  44. }
  45. sync_pb::PersistedEntityData CreatePersistedFromLocalData(
  46. const SyncData& sync_data) {
  47. DCHECK(sync_data.IsValid());
  48. DCHECK(!sync_data.GetTitle().empty());
  49. sync_pb::PersistedEntityData persisted;
  50. persisted.set_name(sync_data.GetTitle());
  51. *persisted.mutable_specifics() = sync_data.GetSpecifics();
  52. return persisted;
  53. }
  54. SyncChange::SyncChangeType ConvertToSyncChangeType(
  55. EntityChange::ChangeType type) {
  56. switch (type) {
  57. case EntityChange::ACTION_DELETE:
  58. return SyncChange::ACTION_DELETE;
  59. case EntityChange::ACTION_ADD:
  60. return SyncChange::ACTION_ADD;
  61. case EntityChange::ACTION_UPDATE:
  62. return SyncChange::ACTION_UPDATE;
  63. }
  64. NOTREACHED();
  65. return SyncChange::ACTION_UPDATE;
  66. }
  67. // Parses the content of |record_list| into |*in_memory_store|. The output
  68. // parameter is first for binding purposes.
  69. absl::optional<ModelError> ParseInMemoryStoreOnBackendSequence(
  70. SyncableServiceBasedBridge::InMemoryStore* in_memory_store,
  71. std::unique_ptr<ModelTypeStore::RecordList> record_list) {
  72. DCHECK(in_memory_store);
  73. DCHECK(in_memory_store->empty());
  74. DCHECK(record_list);
  75. for (const ModelTypeStore::Record& record : *record_list) {
  76. sync_pb::PersistedEntityData persisted_entity;
  77. if (!persisted_entity.ParseFromString(record.value)) {
  78. return ModelError(FROM_HERE, "Failed deserializing data.");
  79. }
  80. in_memory_store->emplace(record.id,
  81. std::move(*persisted_entity.mutable_specifics()));
  82. }
  83. return absl::nullopt;
  84. }
  85. // Object to propagate local changes to the bridge, which will ultimately
  86. // propagate them to the server.
  87. class LocalChangeProcessor : public SyncChangeProcessor {
  88. public:
  89. LocalChangeProcessor(
  90. ModelType type,
  91. const base::RepeatingCallback<void(const absl::optional<ModelError>&)>&
  92. error_callback,
  93. ModelTypeStore* store,
  94. SyncableServiceBasedBridge::InMemoryStore* in_memory_store,
  95. ModelTypeChangeProcessor* other)
  96. : type_(type),
  97. error_callback_(error_callback),
  98. store_(store),
  99. in_memory_store_(in_memory_store),
  100. other_(other) {
  101. DCHECK(store);
  102. DCHECK(other);
  103. }
  104. LocalChangeProcessor(const LocalChangeProcessor&) = delete;
  105. LocalChangeProcessor& operator=(const LocalChangeProcessor&) = delete;
  106. ~LocalChangeProcessor() override = default;
  107. absl::optional<ModelError> ProcessSyncChanges(
  108. const base::Location& from_here,
  109. const SyncChangeList& change_list) override {
  110. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  111. // Reject changes if the processor has already experienced errors.
  112. absl::optional<ModelError> processor_error = other_->GetError();
  113. if (processor_error) {
  114. return processor_error;
  115. }
  116. std::unique_ptr<ModelTypeStore::WriteBatch> batch =
  117. store_->CreateWriteBatch();
  118. for (const SyncChange& change : change_list) {
  119. switch (change.change_type()) {
  120. case SyncChange::ACTION_ADD:
  121. case SyncChange::ACTION_UPDATE: {
  122. DCHECK_EQ(type_, change.sync_data().GetDataType());
  123. DCHECK(change.sync_data().IsValid())
  124. << " from " << change.location().ToString();
  125. // Local adds and updates must have a non-unique-title.
  126. DCHECK(!change.sync_data().GetTitle().empty())
  127. << " from " << change.location().ToString();
  128. const ClientTagHash client_tag_hash =
  129. change.sync_data().GetClientTagHash();
  130. const std::string storage_key = client_tag_hash.value();
  131. DCHECK(!storage_key.empty());
  132. (*in_memory_store_)[storage_key] = change.sync_data().GetSpecifics();
  133. sync_pb::PersistedEntityData persisted_entity =
  134. CreatePersistedFromLocalData(change.sync_data());
  135. batch->WriteData(storage_key, persisted_entity.SerializeAsString());
  136. other_->Put(storage_key,
  137. ConvertPersistedToEntityData(
  138. /*client_tag_hash=*/client_tag_hash,
  139. std::move(persisted_entity)),
  140. batch->GetMetadataChangeList());
  141. break;
  142. }
  143. case SyncChange::ACTION_DELETE: {
  144. const std::string storage_key =
  145. change.sync_data().GetClientTagHash().value();
  146. DCHECK(!storage_key.empty())
  147. << " from " << change.location().ToString();
  148. if (IsActOnceDataType(type_)) {
  149. if (other_->IsEntityUnsynced(storage_key)) {
  150. // Ignore the local deletion if the entity hasn't been committed
  151. // yet, similarly to how WriteNode::Drop() does it.
  152. continue;
  153. }
  154. batch->GetMetadataChangeList()->ClearMetadata(storage_key);
  155. other_->UntrackEntityForStorageKey(storage_key);
  156. } else {
  157. other_->Delete(storage_key, batch->GetMetadataChangeList());
  158. }
  159. in_memory_store_->erase(storage_key);
  160. batch->DeleteData(storage_key);
  161. break;
  162. }
  163. }
  164. }
  165. store_->CommitWriteBatch(std::move(batch), error_callback_);
  166. return absl::nullopt;
  167. }
  168. private:
  169. const ModelType type_;
  170. const base::RepeatingCallback<void(const absl::optional<ModelError>&)>
  171. error_callback_;
  172. const raw_ptr<ModelTypeStore> store_;
  173. const raw_ptr<SyncableServiceBasedBridge::InMemoryStore> in_memory_store_;
  174. const raw_ptr<ModelTypeChangeProcessor> other_;
  175. SEQUENCE_CHECKER(sequence_checker_);
  176. };
  177. class SyncErrorFactoryImpl : public SyncErrorFactory {
  178. public:
  179. explicit SyncErrorFactoryImpl(ModelType type) : type_(type) {}
  180. SyncErrorFactoryImpl(const SyncErrorFactoryImpl&) = delete;
  181. SyncErrorFactoryImpl& operator=(const SyncErrorFactoryImpl&) = delete;
  182. ~SyncErrorFactoryImpl() override = default;
  183. SyncError CreateAndUploadError(const base::Location& location,
  184. const std::string& message) override {
  185. // Uploading is not supported, we simply return the error.
  186. return SyncError(location, SyncError::DATATYPE_ERROR, message, type_);
  187. }
  188. private:
  189. const ModelType type_;
  190. };
  191. } // namespace
  192. SyncableServiceBasedBridge::SyncableServiceBasedBridge(
  193. ModelType type,
  194. OnceModelTypeStoreFactory store_factory,
  195. std::unique_ptr<ModelTypeChangeProcessor> change_processor,
  196. SyncableService* syncable_service)
  197. : ModelTypeSyncBridge(std::move(change_processor)),
  198. type_(type),
  199. syncable_service_(syncable_service),
  200. syncable_service_started_(false) {
  201. DCHECK(syncable_service_);
  202. std::move(store_factory)
  203. .Run(type_, base::BindOnce(&SyncableServiceBasedBridge::OnStoreCreated,
  204. weak_ptr_factory_.GetWeakPtr()));
  205. }
  206. SyncableServiceBasedBridge::~SyncableServiceBasedBridge() {
  207. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  208. // Stop the syncable service to make sure instances of LocalChangeProcessor
  209. // are not continued to be used.
  210. if (syncable_service_started_) {
  211. syncable_service_->StopSyncing(type_);
  212. }
  213. }
  214. std::unique_ptr<MetadataChangeList>
  215. SyncableServiceBasedBridge::CreateMetadataChangeList() {
  216. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  217. return ModelTypeStore::WriteBatch::CreateMetadataChangeList();
  218. }
  219. absl::optional<ModelError> SyncableServiceBasedBridge::MergeSyncData(
  220. std::unique_ptr<MetadataChangeList> metadata_change_list,
  221. EntityChangeList entity_change_list) {
  222. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  223. DCHECK(store_);
  224. DCHECK(change_processor()->IsTrackingMetadata());
  225. DCHECK(!syncable_service_started_);
  226. DCHECK(in_memory_store_.empty());
  227. StoreAndConvertRemoteChanges(std::move(metadata_change_list),
  228. std::move(entity_change_list));
  229. // We ignore the output of previous call of StoreAndConvertRemoteChanges() at
  230. // this point and let StartSyncableService() read from |in_memory_store_|,
  231. // which has been updated above as part of StoreAndConvertRemoteChanges().
  232. return StartSyncableService();
  233. }
  234. absl::optional<ModelError> SyncableServiceBasedBridge::ApplySyncChanges(
  235. std::unique_ptr<MetadataChangeList> metadata_change_list,
  236. EntityChangeList entity_change_list) {
  237. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  238. DCHECK(store_);
  239. DCHECK(change_processor()->IsTrackingMetadata());
  240. DCHECK(syncable_service_started_);
  241. SyncChangeList sync_change_list = StoreAndConvertRemoteChanges(
  242. std::move(metadata_change_list), std::move(entity_change_list));
  243. if (sync_change_list.empty()) {
  244. return absl::nullopt;
  245. }
  246. return syncable_service_->ProcessSyncChanges(FROM_HERE, sync_change_list);
  247. }
  248. void SyncableServiceBasedBridge::GetData(StorageKeyList storage_keys,
  249. DataCallback callback) {
  250. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  251. DCHECK(store_);
  252. store_->ReadData(
  253. storage_keys,
  254. base::BindOnce(&SyncableServiceBasedBridge::OnReadDataForProcessor,
  255. weak_ptr_factory_.GetWeakPtr(), std::move(callback)));
  256. }
  257. void SyncableServiceBasedBridge::GetAllDataForDebugging(DataCallback callback) {
  258. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  259. DCHECK(store_);
  260. store_->ReadAllData(
  261. base::BindOnce(&SyncableServiceBasedBridge::OnReadAllDataForProcessor,
  262. weak_ptr_factory_.GetWeakPtr(), std::move(callback)));
  263. }
  264. std::string SyncableServiceBasedBridge::GetClientTag(
  265. const EntityData& entity_data) {
  266. // Not supported as per SupportsGetClientTag().
  267. NOTREACHED();
  268. return std::string();
  269. }
  270. std::string SyncableServiceBasedBridge::GetStorageKey(
  271. const EntityData& entity_data) {
  272. // Not supported as per SupportsGetStorageKey().
  273. NOTREACHED();
  274. return std::string();
  275. }
  276. bool SyncableServiceBasedBridge::SupportsGetClientTag() const {
  277. return false;
  278. }
  279. bool SyncableServiceBasedBridge::SupportsGetStorageKey() const {
  280. return false;
  281. }
  282. ConflictResolution SyncableServiceBasedBridge::ResolveConflict(
  283. const std::string& storage_key,
  284. const EntityData& remote_data) const {
  285. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  286. if (!remote_data.is_deleted()) {
  287. return ConflictResolution::kUseRemote;
  288. }
  289. // Ignore local changes for extensions/apps when server had a delete, to
  290. // avoid unwanted reinstall of an uninstalled extension.
  291. if (type_ == EXTENSIONS || type_ == APPS) {
  292. DVLOG(1) << "Resolving conflict, ignoring local changes for extension/app";
  293. return ConflictResolution::kUseRemote;
  294. }
  295. return ConflictResolution::kUseLocal;
  296. }
  297. void SyncableServiceBasedBridge::ApplyStopSyncChanges(
  298. std::unique_ptr<MetadataChangeList> delete_metadata_change_list) {
  299. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  300. DCHECK(store_);
  301. // If Sync is being stopped only temporarily (i.e. we want to keep tracking
  302. // metadata), then there's nothing to do here.
  303. if (!delete_metadata_change_list) {
  304. return;
  305. }
  306. in_memory_store_.clear();
  307. store_->DeleteAllDataAndMetadata(base::DoNothing());
  308. if (syncable_service_started_) {
  309. syncable_service_->StopSyncing(type_);
  310. syncable_service_started_ = false;
  311. }
  312. }
  313. size_t SyncableServiceBasedBridge::EstimateSyncOverheadMemoryUsage() const {
  314. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  315. return base::trace_event::EstimateMemoryUsage(in_memory_store_);
  316. }
  317. // static
  318. std::unique_ptr<SyncChangeProcessor>
  319. SyncableServiceBasedBridge::CreateLocalChangeProcessorForTesting(
  320. ModelType type,
  321. ModelTypeStore* store,
  322. InMemoryStore* in_memory_store,
  323. ModelTypeChangeProcessor* other) {
  324. return std::make_unique<LocalChangeProcessor>(
  325. type, /*error_callback=*/base::DoNothing(), store, in_memory_store,
  326. other);
  327. }
  328. void SyncableServiceBasedBridge::OnStoreCreated(
  329. const absl::optional<ModelError>& error,
  330. std::unique_ptr<ModelTypeStore> store) {
  331. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  332. if (error) {
  333. change_processor()->ReportError(*error);
  334. return;
  335. }
  336. DCHECK(store);
  337. store_ = std::move(store);
  338. auto in_memory_store = std::make_unique<InMemoryStore>();
  339. InMemoryStore* raw_in_memory_store = in_memory_store.get();
  340. store_->ReadAllDataAndPreprocess(
  341. base::BindOnce(&ParseInMemoryStoreOnBackendSequence,
  342. base::Unretained(raw_in_memory_store)),
  343. base::BindOnce(&SyncableServiceBasedBridge::OnReadAllDataForInit,
  344. weak_ptr_factory_.GetWeakPtr(),
  345. std::move(in_memory_store)));
  346. }
  347. void SyncableServiceBasedBridge::OnReadAllDataForInit(
  348. std::unique_ptr<InMemoryStore> in_memory_store,
  349. const absl::optional<ModelError>& error) {
  350. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  351. DCHECK(in_memory_store.get());
  352. DCHECK(in_memory_store_.empty());
  353. if (error) {
  354. change_processor()->ReportError(*error);
  355. return;
  356. }
  357. in_memory_store_ = std::move(*in_memory_store);
  358. store_->ReadAllMetadata(
  359. base::BindOnce(&SyncableServiceBasedBridge::OnReadAllMetadataForInit,
  360. weak_ptr_factory_.GetWeakPtr()));
  361. }
  362. void SyncableServiceBasedBridge::OnReadAllMetadataForInit(
  363. const absl::optional<ModelError>& error,
  364. std::unique_ptr<MetadataBatch> metadata_batch) {
  365. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  366. DCHECK(!syncable_service_started_);
  367. if (error) {
  368. change_processor()->ReportError(*error);
  369. return;
  370. }
  371. syncable_service_->WaitUntilReadyToSync(base::BindOnce(
  372. &SyncableServiceBasedBridge::OnSyncableServiceReady,
  373. weak_ptr_factory_.GetWeakPtr(), std::move(metadata_batch)));
  374. }
  375. void SyncableServiceBasedBridge::OnSyncableServiceReady(
  376. std::unique_ptr<MetadataBatch> metadata_batch) {
  377. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  378. DCHECK(!syncable_service_started_);
  379. // Guard against inconsistent state, and recover from it by starting from
  380. // scratch, which will cause the eventual refetching of all entities from the
  381. // server.
  382. if (!metadata_batch->GetModelTypeState().initial_sync_done() &&
  383. !in_memory_store_.empty()) {
  384. in_memory_store_.clear();
  385. store_->DeleteAllDataAndMetadata(base::DoNothing());
  386. change_processor()->ModelReadyToSync(std::make_unique<MetadataBatch>());
  387. DCHECK(!change_processor()->IsTrackingMetadata());
  388. return;
  389. }
  390. change_processor()->ModelReadyToSync(std::move(metadata_batch));
  391. // If sync was previously enabled according to the loaded metadata, then
  392. // immediately start the SyncableService to track as many local changes as
  393. // possible (regardless of whether sync actually starts or not). Otherwise,
  394. // the SyncableService will be started from MergeSyncData().
  395. if (change_processor()->IsTrackingMetadata()) {
  396. ReportErrorIfSet(StartSyncableService());
  397. }
  398. }
  399. absl::optional<ModelError> SyncableServiceBasedBridge::StartSyncableService() {
  400. DCHECK(store_);
  401. DCHECK(!syncable_service_started_);
  402. DCHECK(change_processor()->IsTrackingMetadata());
  403. // Sync enabled, so exercise MergeDataAndStartSyncing() immediately, since
  404. // this function is reached only if sync is starting already.
  405. SyncDataList initial_sync_data;
  406. initial_sync_data.reserve(in_memory_store_.size());
  407. for (const auto& [storage_key, specifics] : in_memory_store_) {
  408. // Note that client tag hash is used as storage key too.
  409. initial_sync_data.push_back(SyncData::CreateRemoteData(
  410. std::move(specifics), ClientTagHash::FromHashed(storage_key)));
  411. }
  412. auto error_callback =
  413. base::BindRepeating(&SyncableServiceBasedBridge::ReportErrorIfSet,
  414. weak_ptr_factory_.GetWeakPtr());
  415. auto local_change_processor = std::make_unique<LocalChangeProcessor>(
  416. type_, error_callback, store_.get(), &in_memory_store_,
  417. change_processor());
  418. const absl::optional<ModelError> merge_error =
  419. syncable_service_->MergeDataAndStartSyncing(
  420. type_, initial_sync_data, std::move(local_change_processor),
  421. std::make_unique<SyncErrorFactoryImpl>(type_));
  422. if (!merge_error) {
  423. syncable_service_started_ = true;
  424. }
  425. return merge_error;
  426. }
  427. SyncChangeList SyncableServiceBasedBridge::StoreAndConvertRemoteChanges(
  428. std::unique_ptr<MetadataChangeList> initial_metadata_change_list,
  429. EntityChangeList input_entity_change_list) {
  430. std::unique_ptr<ModelTypeStore::WriteBatch> batch =
  431. store_->CreateWriteBatch();
  432. batch->TakeMetadataChangesFrom(std::move(initial_metadata_change_list));
  433. SyncChangeList output_sync_change_list;
  434. output_sync_change_list.reserve(input_entity_change_list.size());
  435. for (const std::unique_ptr<EntityChange>& change : input_entity_change_list) {
  436. switch (change->type()) {
  437. case EntityChange::ACTION_DELETE: {
  438. const std::string& storage_key = change->storage_key();
  439. DCHECK_NE(0U, in_memory_store_.count(storage_key));
  440. DVLOG(1) << ModelTypeToDebugString(type_)
  441. << ": Processing deletion with storage key: " << storage_key;
  442. output_sync_change_list.emplace_back(
  443. FROM_HERE, SyncChange::ACTION_DELETE,
  444. SyncData::CreateRemoteData(in_memory_store_[storage_key],
  445. ClientTagHash::FromHashed(storage_key)));
  446. // For tombstones, there is no actual data, which means no client tag
  447. // hash either, but the processor provides the storage key.
  448. DCHECK(!storage_key.empty());
  449. batch->DeleteData(storage_key);
  450. in_memory_store_.erase(storage_key);
  451. break;
  452. }
  453. case EntityChange::ACTION_ADD:
  454. // Because we use the client tag hash as storage key, let the processor
  455. // know.
  456. change_processor()->UpdateStorageKey(
  457. change->data(),
  458. /*storage_key=*/change->data().client_tag_hash.value(),
  459. batch->GetMetadataChangeList());
  460. [[fallthrough]];
  461. case EntityChange::ACTION_UPDATE: {
  462. const std::string& storage_key = change->data().client_tag_hash.value();
  463. DVLOG(1) << ModelTypeToDebugString(type_)
  464. << ": Processing add/update with key: " << storage_key;
  465. output_sync_change_list.emplace_back(
  466. FROM_HERE, ConvertToSyncChangeType(change->type()),
  467. SyncData::CreateRemoteData(change->data().specifics,
  468. change->data().client_tag_hash));
  469. batch->WriteData(
  470. storage_key,
  471. CreatePersistedFromRemoteData(change->data()).SerializeAsString());
  472. in_memory_store_[storage_key] = change->data().specifics;
  473. break;
  474. }
  475. }
  476. }
  477. store_->CommitWriteBatch(
  478. std::move(batch),
  479. base::BindOnce(&SyncableServiceBasedBridge::ReportErrorIfSet,
  480. weak_ptr_factory_.GetWeakPtr()));
  481. return output_sync_change_list;
  482. }
  483. void SyncableServiceBasedBridge::OnReadDataForProcessor(
  484. DataCallback callback,
  485. const absl::optional<ModelError>& error,
  486. std::unique_ptr<ModelTypeStore::RecordList> record_list,
  487. std::unique_ptr<ModelTypeStore::IdList> missing_id_list) {
  488. OnReadAllDataForProcessor(std::move(callback), error, std::move(record_list));
  489. }
  490. void SyncableServiceBasedBridge::OnReadAllDataForProcessor(
  491. DataCallback callback,
  492. const absl::optional<ModelError>& error,
  493. std::unique_ptr<ModelTypeStore::RecordList> record_list) {
  494. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  495. if (error) {
  496. change_processor()->ReportError(*error);
  497. return;
  498. }
  499. auto batch = std::make_unique<MutableDataBatch>();
  500. for (const ModelTypeStore::Record& record : *record_list) {
  501. sync_pb::PersistedEntityData persisted_entity;
  502. if (record.id.empty() || !persisted_entity.ParseFromString(record.value)) {
  503. change_processor()->ReportError(
  504. {FROM_HERE, "Failed deserializing data."});
  505. return;
  506. }
  507. // Note that client tag hash is used as storage key too.
  508. batch->Put(record.id, ConvertPersistedToEntityData(
  509. ClientTagHash::FromHashed(record.id),
  510. std::move(persisted_entity)));
  511. }
  512. std::move(callback).Run(std::move(batch));
  513. }
  514. void SyncableServiceBasedBridge::ReportErrorIfSet(
  515. const absl::optional<ModelError>& error) {
  516. if (error) {
  517. change_processor()->ReportError(*error);
  518. }
  519. }
  520. } // namespace syncer