summaryrefslogtreecommitdiffstats
path: root/services/sync/modules/engines.sys.mjs
diff options
context:
space:
mode:
Diffstat (limited to 'services/sync/modules/engines.sys.mjs')
-rw-r--r--services/sync/modules/engines.sys.mjs2274
1 files changed, 2274 insertions, 0 deletions
diff --git a/services/sync/modules/engines.sys.mjs b/services/sync/modules/engines.sys.mjs
new file mode 100644
index 0000000000..0d490ac4b3
--- /dev/null
+++ b/services/sync/modules/engines.sys.mjs
@@ -0,0 +1,2274 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+import { XPCOMUtils } from "resource://gre/modules/XPCOMUtils.sys.mjs";
+
+import { JSONFile } from "resource://gre/modules/JSONFile.sys.mjs";
+import { Log } from "resource://gre/modules/Log.sys.mjs";
+
+import { Async } from "resource://services-common/async.sys.mjs";
+import { Observers } from "resource://services-common/observers.sys.mjs";
+
+import {
+ DEFAULT_DOWNLOAD_BATCH_SIZE,
+ DEFAULT_GUID_FETCH_BATCH_SIZE,
+ ENGINE_BATCH_INTERRUPTED,
+ ENGINE_DOWNLOAD_FAIL,
+ ENGINE_UPLOAD_FAIL,
+ VERSION_OUT_OF_DATE,
+ PREFS_BRANCH,
+} from "resource://services-sync/constants.sys.mjs";
+
+import {
+ Collection,
+ CryptoWrapper,
+} from "resource://services-sync/record.sys.mjs";
+import { Resource } from "resource://services-sync/resource.sys.mjs";
+import {
+ SerializableSet,
+ Svc,
+ Utils,
+} from "resource://services-sync/util.sys.mjs";
+import { SyncedRecordsTelemetry } from "resource://services-sync/telemetry.sys.mjs";
+
+const lazy = {};
+
+ChromeUtils.defineESModuleGetters(lazy, {
+ PlacesUtils: "resource://gre/modules/PlacesUtils.sys.mjs",
+});
+
+function ensureDirectory(path) {
+ return IOUtils.makeDirectory(PathUtils.parent(path), {
+ createAncestors: true,
+ });
+}
+
+/**
+ * Trackers are associated with a single engine and deal with
+ * listening for changes to their particular data type.
+ *
+ * The base `Tracker` only supports listening for changes, and bumping the score
+ * to indicate how urgently the engine wants to sync. It does not persist any
+ * data. Engines that track changes directly in the storage layer (like
+ * bookmarks, bridged engines, addresses, and credit cards) or only upload a
+ * single record (tabs and preferences) should subclass `Tracker`.
+ */
+export function Tracker(name, engine) {
+ if (!engine) {
+ throw new Error("Tracker must be associated with an Engine instance.");
+ }
+
+ name = name || "Unnamed";
+ this.name = name.toLowerCase();
+ this.engine = engine;
+
+ this._log = Log.repository.getLogger(`Sync.Engine.${name}.Tracker`);
+
+ this._score = 0;
+
+ this.asyncObserver = Async.asyncObserver(this, this._log);
+}
+
+Tracker.prototype = {
+ // New-style trackers use change sources to filter out changes made by Sync in
+ // observer notifications, so we don't want to let the engine ignore all
+ // changes during a sync.
+ get ignoreAll() {
+ return false;
+ },
+
+ // Define an empty setter so that the engine doesn't throw a `TypeError`
+ // setting a read-only property.
+ set ignoreAll(value) {},
+
+ /*
+ * Score can be called as often as desired to decide which engines to sync
+ *
+ * Valid values for score:
+ * -1: Do not sync unless the user specifically requests it (almost disabled)
+ * 0: Nothing has changed
+ * 100: Please sync me ASAP!
+ *
+ * Setting it to other values should (but doesn't currently) throw an exception
+ */
+ get score() {
+ return this._score;
+ },
+
+ set score(value) {
+ this._score = value;
+ Observers.notify("weave:engine:score:updated", this.name);
+ },
+
+ // Should be called by service everytime a sync has been done for an engine
+ resetScore() {
+ this._score = 0;
+ },
+
+ // Unsupported, and throws a more descriptive error to ensure callers aren't
+ // accidentally using persistence.
+ async getChangedIDs() {
+ throw new TypeError("This tracker doesn't store changed IDs");
+ },
+
+ // Also unsupported.
+ async addChangedID(id, when) {
+ throw new TypeError("Can't add changed ID to this tracker");
+ },
+
+ // Ditto.
+ async removeChangedID(...ids) {
+ throw new TypeError("Can't remove changed IDs from this tracker");
+ },
+
+ // This method is called at various times, so we override with a no-op
+ // instead of throwing.
+ clearChangedIDs() {},
+
+ _now() {
+ return Date.now() / 1000;
+ },
+
+ _isTracking: false,
+
+ start() {
+ if (!this.engineIsEnabled()) {
+ return;
+ }
+ this._log.trace("start().");
+ if (!this._isTracking) {
+ this.onStart();
+ this._isTracking = true;
+ }
+ },
+
+ async stop() {
+ this._log.trace("stop().");
+ if (this._isTracking) {
+ await this.asyncObserver.promiseObserversComplete();
+ this.onStop();
+ this._isTracking = false;
+ }
+ },
+
+ // Override these in your subclasses.
+ onStart() {},
+ onStop() {},
+ async observe(subject, topic, data) {},
+
+ engineIsEnabled() {
+ if (!this.engine) {
+ // Can't tell -- we must be running in a test!
+ return true;
+ }
+ return this.engine.enabled;
+ },
+
+ /**
+ * Starts or stops listening for changes depending on the associated engine's
+ * enabled state.
+ *
+ * @param {Boolean} engineEnabled Whether the engine was enabled.
+ */
+ async onEngineEnabledChanged(engineEnabled) {
+ if (engineEnabled == this._isTracking) {
+ return;
+ }
+
+ if (engineEnabled) {
+ this.start();
+ } else {
+ await this.stop();
+ this.clearChangedIDs();
+ }
+ },
+
+ async finalize() {
+ await this.stop();
+ },
+};
+
+/*
+ * A tracker that persists a list of IDs for all changed items that need to be
+ * synced. This is 🚨 _extremely deprecated_ 🚨 and only kept around for current
+ * engines. ⚠️ Please **don't use it** for new engines! ⚠️
+ *
+ * Why is this kind of external change tracking deprecated? Because it causes
+ * consistency issues due to missed notifications, interrupted syncs, and the
+ * tracker's view of what changed diverging from the data store's.
+ */
+export function LegacyTracker(name, engine) {
+ Tracker.call(this, name, engine);
+
+ this._ignored = [];
+ this.file = this.name;
+ this._storage = new JSONFile({
+ path: Utils.jsonFilePath("changes", this.file),
+ dataPostProcessor: json => this._dataPostProcessor(json),
+ beforeSave: () => this._beforeSave(),
+ });
+ this._ignoreAll = false;
+}
+
+LegacyTracker.prototype = {
+ get ignoreAll() {
+ return this._ignoreAll;
+ },
+
+ set ignoreAll(value) {
+ this._ignoreAll = value;
+ },
+
+ // Default to an empty object if the file doesn't exist.
+ _dataPostProcessor(json) {
+ return (typeof json == "object" && json) || {};
+ },
+
+ // Ensure the Weave storage directory exists before writing the file.
+ _beforeSave() {
+ return ensureDirectory(this._storage.path);
+ },
+
+ async getChangedIDs() {
+ await this._storage.load();
+ return this._storage.data;
+ },
+
+ _saveChangedIDs() {
+ this._storage.saveSoon();
+ },
+
+ // ignore/unignore specific IDs. Useful for ignoring items that are
+ // being processed, or that shouldn't be synced.
+ // But note: not persisted to disk
+
+ ignoreID(id) {
+ this.unignoreID(id);
+ this._ignored.push(id);
+ },
+
+ unignoreID(id) {
+ let index = this._ignored.indexOf(id);
+ if (index != -1) {
+ this._ignored.splice(index, 1);
+ }
+ },
+
+ async _saveChangedID(id, when) {
+ this._log.trace(`Adding changed ID: ${id}, ${JSON.stringify(when)}`);
+ const changedIDs = await this.getChangedIDs();
+ changedIDs[id] = when;
+ this._saveChangedIDs();
+ },
+
+ async addChangedID(id, when) {
+ if (!id) {
+ this._log.warn("Attempted to add undefined ID to tracker");
+ return false;
+ }
+
+ if (this.ignoreAll || this._ignored.includes(id)) {
+ return false;
+ }
+
+ // Default to the current time in seconds if no time is provided.
+ if (when == null) {
+ when = this._now();
+ }
+
+ const changedIDs = await this.getChangedIDs();
+ // Add/update the entry if we have a newer time.
+ if ((changedIDs[id] || -Infinity) < when) {
+ await this._saveChangedID(id, when);
+ }
+
+ return true;
+ },
+
+ async removeChangedID(...ids) {
+ if (!ids.length || this.ignoreAll) {
+ return false;
+ }
+ for (let id of ids) {
+ if (!id) {
+ this._log.warn("Attempted to remove undefined ID from tracker");
+ continue;
+ }
+ if (this._ignored.includes(id)) {
+ this._log.debug(`Not removing ignored ID ${id} from tracker`);
+ continue;
+ }
+ const changedIDs = await this.getChangedIDs();
+ if (changedIDs[id] != null) {
+ this._log.trace("Removing changed ID " + id);
+ delete changedIDs[id];
+ }
+ }
+ this._saveChangedIDs();
+ return true;
+ },
+
+ clearChangedIDs() {
+ this._log.trace("Clearing changed ID list");
+ this._storage.data = {};
+ this._saveChangedIDs();
+ },
+
+ async finalize() {
+ // Persist all pending tracked changes to disk, and wait for the final write
+ // to finish.
+ await super.finalize();
+ this._saveChangedIDs();
+ await this._storage.finalize();
+ },
+};
+Object.setPrototypeOf(LegacyTracker.prototype, Tracker.prototype);
+
+/**
+ * The Store serves as the interface between Sync and stored data.
+ *
+ * The name "store" is slightly a misnomer because it doesn't actually "store"
+ * anything. Instead, it serves as a gateway to something that actually does
+ * the "storing."
+ *
+ * The store is responsible for record management inside an engine. It tells
+ * Sync what items are available for Sync, converts items to and from Sync's
+ * record format, and applies records from Sync into changes on the underlying
+ * store.
+ *
+ * Store implementations require a number of functions to be implemented. These
+ * are all documented below.
+ *
+ * For stores that deal with many records or which have expensive store access
+ * routines, it is highly recommended to implement a custom applyIncomingBatch
+ * and/or applyIncoming function on top of the basic APIs.
+ */
+
+export function Store(name, engine) {
+ if (!engine) {
+ throw new Error("Store must be associated with an Engine instance.");
+ }
+
+ name = name || "Unnamed";
+ this.name = name.toLowerCase();
+ this.engine = engine;
+
+ this._log = Log.repository.getLogger(`Sync.Engine.${name}.Store`);
+
+ ChromeUtils.defineLazyGetter(this, "_timer", function () {
+ return Cc["@mozilla.org/timer;1"].createInstance(Ci.nsITimer);
+ });
+}
+
+Store.prototype = {
+ /**
+ * Apply multiple incoming records against the store.
+ *
+ * This is called with a set of incoming records to process. The function
+ * should look at each record, reconcile with the current local state, and
+ * make the local changes required to bring its state in alignment with the
+ * record.
+ *
+ * The default implementation simply iterates over all records and calls
+ * applyIncoming(). Store implementations may overwrite this function
+ * if desired.
+ *
+ * @param records Array of records to apply
+ * @param a SyncedRecordsTelemetry obj that will keep track of failed reasons
+ * @return Array of record IDs which did not apply cleanly
+ */
+ async applyIncomingBatch(records, countTelemetry) {
+ let failed = [];
+
+ await Async.yieldingForEach(records, async record => {
+ try {
+ await this.applyIncoming(record);
+ } catch (ex) {
+ if (ex.code == SyncEngine.prototype.eEngineAbortApplyIncoming) {
+ // This kind of exception should have a 'cause' attribute, which is an
+ // originating exception.
+ // ex.cause will carry its stack with it when rethrown.
+ throw ex.cause;
+ }
+ if (Async.isShutdownException(ex)) {
+ throw ex;
+ }
+ this._log.warn("Failed to apply incoming record " + record.id, ex);
+ failed.push(record.id);
+ countTelemetry.addIncomingFailedReason(ex.message);
+ }
+ });
+
+ return failed;
+ },
+
+ /**
+ * Apply a single record against the store.
+ *
+ * This takes a single record and makes the local changes required so the
+ * local state matches what's in the record.
+ *
+ * The default implementation calls one of remove(), create(), or update()
+ * depending on the state obtained from the store itself. Store
+ * implementations may overwrite this function if desired.
+ *
+ * @param record
+ * Record to apply
+ */
+ async applyIncoming(record) {
+ if (record.deleted) {
+ await this.remove(record);
+ } else if (!(await this.itemExists(record.id))) {
+ await this.create(record);
+ } else {
+ await this.update(record);
+ }
+ },
+
+ // override these in derived objects
+
+ /**
+ * Create an item in the store from a record.
+ *
+ * This is called by the default implementation of applyIncoming(). If using
+ * applyIncomingBatch(), this won't be called unless your store calls it.
+ *
+ * @param record
+ * The store record to create an item from
+ */
+ async create(record) {
+ throw new Error("override create in a subclass");
+ },
+
+ /**
+ * Remove an item in the store from a record.
+ *
+ * This is called by the default implementation of applyIncoming(). If using
+ * applyIncomingBatch(), this won't be called unless your store calls it.
+ *
+ * @param record
+ * The store record to delete an item from
+ */
+ async remove(record) {
+ throw new Error("override remove in a subclass");
+ },
+
+ /**
+ * Update an item from a record.
+ *
+ * This is called by the default implementation of applyIncoming(). If using
+ * applyIncomingBatch(), this won't be called unless your store calls it.
+ *
+ * @param record
+ * The record to use to update an item from
+ */
+ async update(record) {
+ throw new Error("override update in a subclass");
+ },
+
+ /**
+ * Determine whether a record with the specified ID exists.
+ *
+ * Takes a string record ID and returns a booleans saying whether the record
+ * exists.
+ *
+ * @param id
+ * string record ID
+ * @return boolean indicating whether record exists locally
+ */
+ async itemExists(id) {
+ throw new Error("override itemExists in a subclass");
+ },
+
+ /**
+ * Create a record from the specified ID.
+ *
+ * If the ID is known, the record should be populated with metadata from
+ * the store. If the ID is not known, the record should be created with the
+ * delete field set to true.
+ *
+ * @param id
+ * string record ID
+ * @param collection
+ * Collection to add record to. This is typically passed into the
+ * constructor for the newly-created record.
+ * @return record type for this engine
+ */
+ async createRecord(id, collection) {
+ throw new Error("override createRecord in a subclass");
+ },
+
+ /**
+ * Change the ID of a record.
+ *
+ * @param oldID
+ * string old/current record ID
+ * @param newID
+ * string new record ID
+ */
+ async changeItemID(oldID, newID) {
+ throw new Error("override changeItemID in a subclass");
+ },
+
+ /**
+ * Obtain the set of all known record IDs.
+ *
+ * @return Object with ID strings as keys and values of true. The values
+ * are ignored.
+ */
+ async getAllIDs() {
+ throw new Error("override getAllIDs in a subclass");
+ },
+
+ /**
+ * Wipe all data in the store.
+ *
+ * This function is called during remote wipes or when replacing local data
+ * with remote data.
+ *
+ * This function should delete all local data that the store is managing. It
+ * can be thought of as clearing out all state and restoring the "new
+ * browser" state.
+ */
+ async wipe() {
+ throw new Error("override wipe in a subclass");
+ },
+};
+
+export function EngineManager(service) {
+ this.service = service;
+
+ this._engines = {};
+
+ this._altEngineInfo = {};
+
+ // This will be populated by Service on startup.
+ this._declined = new Set();
+ this._log = Log.repository.getLogger("Sync.EngineManager");
+ this._log.manageLevelFromPref("services.sync.log.logger.service.engines");
+ // define the default level for all engine logs here (although each engine
+ // allows its level to be controlled via a specific, non-default pref)
+ Log.repository
+ .getLogger(`Sync.Engine`)
+ .manageLevelFromPref("services.sync.log.logger.engine");
+}
+
+EngineManager.prototype = {
+ get(name) {
+ // Return an array of engines if we have an array of names
+ if (Array.isArray(name)) {
+ let engines = [];
+ name.forEach(function (name) {
+ let engine = this.get(name);
+ if (engine) {
+ engines.push(engine);
+ }
+ }, this);
+ return engines;
+ }
+
+ return this._engines[name]; // Silently returns undefined for unknown names.
+ },
+
+ getAll() {
+ let engines = [];
+ for (let [, engine] of Object.entries(this._engines)) {
+ engines.push(engine);
+ }
+ return engines;
+ },
+
+ /**
+ * If a user has changed a pref that controls which variant of a sync engine
+ * for a given collection we use, unregister the old engine and register the
+ * new one.
+ *
+ * This is called by EngineSynchronizer before every sync.
+ */
+ async switchAlternatives() {
+ for (let [name, info] of Object.entries(this._altEngineInfo)) {
+ let prefValue = info.prefValue;
+ if (prefValue === info.lastValue) {
+ this._log.trace(
+ `No change for engine ${name} (${info.pref} is still ${prefValue})`
+ );
+ continue;
+ }
+ // Unregister the old engine, register the new one.
+ this._log.info(
+ `Switching ${name} engine ("${info.pref}" went from ${info.lastValue} => ${prefValue})`
+ );
+ try {
+ await this._removeAndFinalize(name);
+ } catch (e) {
+ this._log.warn(`Failed to remove previous ${name} engine...`, e);
+ }
+ let engineType = prefValue ? info.whenTrue : info.whenFalse;
+ try {
+ // If register throws, we'll try again next sync, but until then there
+ // won't be an engine registered for this collection.
+ await this.register(engineType);
+ info.lastValue = prefValue;
+ // Note: engineType.name is using Function.prototype.name.
+ this._log.info(`Switched the ${name} engine to use ${engineType.name}`);
+ } catch (e) {
+ this._log.warn(
+ `Switching the ${name} engine to use ${engineType.name} failed (couldn't register)`,
+ e
+ );
+ }
+ }
+ },
+
+ async registerAlternatives(name, pref, whenTrue, whenFalse) {
+ let info = { name, pref, whenTrue, whenFalse };
+
+ XPCOMUtils.defineLazyPreferenceGetter(info, "prefValue", pref, false);
+
+ let chosen = info.prefValue ? info.whenTrue : info.whenFalse;
+ info.lastValue = info.prefValue;
+ this._altEngineInfo[name] = info;
+
+ await this.register(chosen);
+ },
+
+ /**
+ * N.B., does not pay attention to the declined list.
+ */
+ getEnabled() {
+ return this.getAll()
+ .filter(engine => engine.enabled)
+ .sort((a, b) => a.syncPriority - b.syncPriority);
+ },
+
+ get enabledEngineNames() {
+ return this.getEnabled().map(e => e.name);
+ },
+
+ persistDeclined() {
+ Svc.PrefBranch.setStringPref(
+ "declinedEngines",
+ [...this._declined].join(",")
+ );
+ },
+
+ /**
+ * Returns an array.
+ */
+ getDeclined() {
+ return [...this._declined];
+ },
+
+ setDeclined(engines) {
+ this._declined = new Set(engines);
+ this.persistDeclined();
+ },
+
+ isDeclined(engineName) {
+ return this._declined.has(engineName);
+ },
+
+ /**
+ * Accepts a Set or an array.
+ */
+ decline(engines) {
+ for (let e of engines) {
+ this._declined.add(e);
+ }
+ this.persistDeclined();
+ },
+
+ undecline(engines) {
+ for (let e of engines) {
+ this._declined.delete(e);
+ }
+ this.persistDeclined();
+ },
+
+ /**
+ * Register an Engine to the service. Alternatively, give an array of engine
+ * objects to register.
+ *
+ * @param engineObject
+ * Engine object used to get an instance of the engine
+ * @return The engine object if anything failed
+ */
+ async register(engineObject) {
+ if (Array.isArray(engineObject)) {
+ for (const e of engineObject) {
+ await this.register(e);
+ }
+ return;
+ }
+
+ try {
+ let engine = new engineObject(this.service);
+ let name = engine.name;
+ if (name in this._engines) {
+ this._log.error("Engine '" + name + "' is already registered!");
+ } else {
+ if (engine.initialize) {
+ await engine.initialize();
+ }
+ this._engines[name] = engine;
+ }
+ } catch (ex) {
+ let name = engineObject || "";
+ name = name.prototype || "";
+ name = name.name || "";
+
+ this._log.error(`Could not initialize engine ${name}`, ex);
+ }
+ },
+
+ async unregister(val) {
+ let name = val;
+ if (val instanceof SyncEngine) {
+ name = val.name;
+ }
+ await this._removeAndFinalize(name);
+ delete this._altEngineInfo[name];
+ },
+
+ // Common code for disabling an engine by name, that doesn't complain if the
+ // engine doesn't exist. Doesn't touch the engine's alternative info (if any
+ // exists).
+ async _removeAndFinalize(name) {
+ if (name in this._engines) {
+ let engine = this._engines[name];
+ delete this._engines[name];
+ await engine.finalize();
+ }
+ },
+
+ async clear() {
+ for (let name in this._engines) {
+ let engine = this._engines[name];
+ delete this._engines[name];
+ await engine.finalize();
+ }
+ this._altEngineInfo = {};
+ },
+};
+
+export function SyncEngine(name, service) {
+ if (!service) {
+ throw new Error("SyncEngine must be associated with a Service instance.");
+ }
+
+ this.Name = name || "Unnamed";
+ this.name = name.toLowerCase();
+ this.service = service;
+
+ this._notify = Utils.notify("weave:engine:");
+ this._log = Log.repository.getLogger("Sync.Engine." + this.Name);
+ this._log.manageLevelFromPref(`services.sync.log.logger.engine.${this.name}`);
+
+ this._modified = this.emptyChangeset();
+ this._tracker; // initialize tracker to load previously changed IDs
+ this._log.debug("Engine constructed");
+
+ this._toFetchStorage = new JSONFile({
+ path: Utils.jsonFilePath("toFetch", this.name),
+ dataPostProcessor: json => this._metadataPostProcessor(json),
+ beforeSave: () => this._beforeSaveMetadata(),
+ });
+
+ this._previousFailedStorage = new JSONFile({
+ path: Utils.jsonFilePath("failed", this.name),
+ dataPostProcessor: json => this._metadataPostProcessor(json),
+ beforeSave: () => this._beforeSaveMetadata(),
+ });
+
+ XPCOMUtils.defineLazyPreferenceGetter(
+ this,
+ "_enabled",
+ `services.sync.engine.${this.prefName}`,
+ false
+ );
+ XPCOMUtils.defineLazyPreferenceGetter(
+ this,
+ "_syncID",
+ `services.sync.${this.name}.syncID`,
+ ""
+ );
+ XPCOMUtils.defineLazyPreferenceGetter(
+ this,
+ "_lastSync",
+ `services.sync.${this.name}.lastSync`,
+ "0",
+ null,
+ v => parseFloat(v)
+ );
+ // Async initializations can be made in the initialize() method.
+
+ this.asyncObserver = Async.asyncObserver(this, this._log);
+}
+
+// Enumeration to define approaches to handling bad records.
+// Attached to the constructor to allow use as a kind of static enumeration.
+SyncEngine.kRecoveryStrategy = {
+ ignore: "ignore",
+ retry: "retry",
+ error: "error",
+};
+
+SyncEngine.prototype = {
+ _recordObj: CryptoWrapper,
+ // _storeObj, and _trackerObj should to be overridden in subclasses
+ _storeObj: Store,
+ _trackerObj: Tracker,
+ version: 1,
+
+ // Local 'constant'.
+ // Signal to the engine that processing further records is pointless.
+ eEngineAbortApplyIncoming: "error.engine.abort.applyincoming",
+
+ // Should we keep syncing if we find a record that cannot be uploaded (ever)?
+ // If this is false, we'll throw, otherwise, we'll ignore the record and
+ // continue. This currently can only happen due to the record being larger
+ // than the record upload limit.
+ allowSkippedRecord: true,
+
+ // Which sortindex to use when retrieving records for this engine.
+ _defaultSort: undefined,
+
+ _hasSyncedThisSession: false,
+
+ _metadataPostProcessor(json) {
+ if (Array.isArray(json)) {
+ // Pre-`JSONFile` storage stored an array, but `JSONFile` defaults to
+ // an object, so we wrap the array for consistency.
+ json = { ids: json };
+ }
+ if (!json.ids) {
+ json.ids = [];
+ }
+ // The set serializes the same way as an array, but offers more efficient
+ // methods of manipulation.
+ json.ids = new SerializableSet(json.ids);
+ return json;
+ },
+
+ async _beforeSaveMetadata() {
+ await ensureDirectory(this._toFetchStorage.path);
+ await ensureDirectory(this._previousFailedStorage.path);
+ },
+
+ // A relative priority to use when computing an order
+ // for engines to be synced. Higher-priority engines
+ // (lower numbers) are synced first.
+ // It is recommended that a unique value be used for each engine,
+ // in order to guarantee a stable sequence.
+ syncPriority: 0,
+
+ // How many records to pull in a single sync. This is primarily to avoid very
+ // long first syncs against profiles with many history records.
+ downloadLimit: null,
+
+ // How many records to pull at one time when specifying IDs. This is to avoid
+ // URI length limitations.
+ guidFetchBatchSize: DEFAULT_GUID_FETCH_BATCH_SIZE,
+
+ downloadBatchSize: DEFAULT_DOWNLOAD_BATCH_SIZE,
+
+ async initialize() {
+ await this._toFetchStorage.load();
+ await this._previousFailedStorage.load();
+ Services.prefs.addObserver(
+ `${PREFS_BRANCH}engine.${this.prefName}`,
+ this.asyncObserver,
+ true
+ );
+ this._log.debug("SyncEngine initialized", this.name);
+ },
+
+ get prefName() {
+ return this.name;
+ },
+
+ get enabled() {
+ return this._enabled;
+ },
+
+ set enabled(val) {
+ if (!!val != this._enabled) {
+ Svc.PrefBranch.setBoolPref("engine." + this.prefName, !!val);
+ }
+ },
+
+ get score() {
+ return this._tracker.score;
+ },
+
+ get _store() {
+ let store = new this._storeObj(this.Name, this);
+ this.__defineGetter__("_store", () => store);
+ return store;
+ },
+
+ get _tracker() {
+ let tracker = new this._trackerObj(this.Name, this);
+ this.__defineGetter__("_tracker", () => tracker);
+ return tracker;
+ },
+
+ get storageURL() {
+ return this.service.storageURL;
+ },
+
+ get engineURL() {
+ return this.storageURL + this.name;
+ },
+
+ get cryptoKeysURL() {
+ return this.storageURL + "crypto/keys";
+ },
+
+ get metaURL() {
+ return this.storageURL + "meta/global";
+ },
+
+ startTracking() {
+ this._tracker.start();
+ },
+
+ // Returns a promise
+ stopTracking() {
+ return this._tracker.stop();
+ },
+
+ // Listens for engine enabled state changes, and updates the tracker's state.
+ // This is an async observer because the tracker waits on all its async
+ // observers to finish when it's stopped.
+ async observe(subject, topic, data) {
+ if (
+ topic == "nsPref:changed" &&
+ data == `services.sync.engine.${this.prefName}`
+ ) {
+ await this._tracker.onEngineEnabledChanged(this._enabled);
+ }
+ },
+
+ async sync() {
+ if (!this.enabled) {
+ return false;
+ }
+
+ if (!this._sync) {
+ throw new Error("engine does not implement _sync method");
+ }
+
+ return this._notify("sync", this.name, this._sync)();
+ },
+
+ // Override this method to return a new changeset type.
+ emptyChangeset() {
+ return new Changeset();
+ },
+
+ /**
+ * Returns the local sync ID for this engine, or `""` if the engine hasn't
+ * synced for the first time. This is exposed for tests.
+ *
+ * @return the current sync ID.
+ */
+ async getSyncID() {
+ return this._syncID;
+ },
+
+ /**
+ * Ensures that the local sync ID for the engine matches the sync ID for the
+ * collection on the server. A mismatch indicates that another client wiped
+ * the collection; we're syncing after a node reassignment, and another
+ * client synced before us; or the store was replaced since the last sync.
+ * In case of a mismatch, we need to reset all local Sync state and start
+ * over as a first sync.
+ *
+ * In most cases, this method should return the new sync ID as-is. However, an
+ * engine may ignore the given ID and assign a different one, if it determines
+ * that the sync ID on the server is out of date. The bookmarks engine uses
+ * this to wipe the server and other clients on the first sync after the user
+ * restores from a backup.
+ *
+ * @param newSyncID
+ * The new sync ID for the collection from `meta/global`.
+ * @return The assigned sync ID. If this doesn't match `newSyncID`, we'll
+ * replace the sync ID in `meta/global` with the assigned ID.
+ */
+ async ensureCurrentSyncID(newSyncID) {
+ let existingSyncID = this._syncID;
+ if (existingSyncID == newSyncID) {
+ return existingSyncID;
+ }
+ this._log.debug(
+ `Engine syncIDs differ (old="${existingSyncID}", new="${newSyncID}") - resetting the engine`
+ );
+ await this.resetClient();
+ Svc.PrefBranch.setStringPref(this.name + ".syncID", newSyncID);
+ Svc.PrefBranch.setStringPref(this.name + ".lastSync", "0");
+ return newSyncID;
+ },
+
+ /**
+ * Resets the local sync ID for the engine, wipes the server, and resets all
+ * local Sync state to start over as a first sync.
+ *
+ * @return the new sync ID.
+ */
+ async resetSyncID() {
+ let newSyncID = await this.resetLocalSyncID();
+ await this.wipeServer();
+ return newSyncID;
+ },
+
+ /**
+ * Resets the local sync ID for the engine, signaling that we're starting over
+ * as a first sync.
+ *
+ * @return the new sync ID.
+ */
+ async resetLocalSyncID() {
+ return this.ensureCurrentSyncID(Utils.makeGUID());
+ },
+
+ /**
+ * Allows overriding scheduler logic -- added to help reduce kinto server
+ * getting hammered because our scheduler never got tuned for it.
+ *
+ * Note: Overriding engines must take resyncs into account -- score will not
+ * be cleared.
+ */
+ shouldSkipSync(syncReason) {
+ return false;
+ },
+
+ /*
+ * lastSync is a timestamp in server time.
+ */
+ async getLastSync() {
+ return this._lastSync;
+ },
+ async setLastSync(lastSync) {
+ // Store the value as a string to keep floating point precision
+ Svc.PrefBranch.setStringPref(this.name + ".lastSync", lastSync.toString());
+ },
+ async resetLastSync() {
+ this._log.debug("Resetting " + this.name + " last sync time");
+ await this.setLastSync(0);
+ },
+
+ get hasSyncedThisSession() {
+ return this._hasSyncedThisSession;
+ },
+
+ set hasSyncedThisSession(hasSynced) {
+ this._hasSyncedThisSession = hasSynced;
+ },
+
+ get toFetch() {
+ this._toFetchStorage.ensureDataReady();
+ return this._toFetchStorage.data.ids;
+ },
+
+ set toFetch(ids) {
+ if (ids.constructor.name != "SerializableSet") {
+ throw new Error(
+ "Bug: Attempted to set toFetch to something that isn't a SerializableSet"
+ );
+ }
+ this._toFetchStorage.data = { ids };
+ this._toFetchStorage.saveSoon();
+ },
+
+ get previousFailed() {
+ this._previousFailedStorage.ensureDataReady();
+ return this._previousFailedStorage.data.ids;
+ },
+
+ set previousFailed(ids) {
+ if (ids.constructor.name != "SerializableSet") {
+ throw new Error(
+ "Bug: Attempted to set previousFailed to something that isn't a SerializableSet"
+ );
+ }
+ this._previousFailedStorage.data = { ids };
+ this._previousFailedStorage.saveSoon();
+ },
+
+ /*
+ * Returns a changeset for this sync. Engine implementations can override this
+ * method to bypass the tracker for certain or all changed items.
+ */
+ async getChangedIDs() {
+ return this._tracker.getChangedIDs();
+ },
+
+ // Create a new record using the store and add in metadata.
+ async _createRecord(id) {
+ let record = await this._store.createRecord(id, this.name);
+ record.id = id;
+ record.collection = this.name;
+ return record;
+ },
+
+ // Creates a tombstone Sync record with additional metadata.
+ _createTombstone(id) {
+ let tombstone = new this._recordObj(this.name, id);
+ tombstone.id = id;
+ tombstone.collection = this.name;
+ tombstone.deleted = true;
+ return tombstone;
+ },
+
+ // Any setup that needs to happen at the beginning of each sync.
+ async _syncStartup() {
+ // Determine if we need to wipe on outdated versions
+ let metaGlobal = await this.service.recordManager.get(this.metaURL);
+ let engines = metaGlobal.payload.engines || {};
+ let engineData = engines[this.name] || {};
+
+ // Assume missing versions are 0 and wipe the server
+ if ((engineData.version || 0) < this.version) {
+ this._log.debug("Old engine data: " + [engineData.version, this.version]);
+
+ // Clear the server and reupload everything on bad version or missing
+ // meta. Note that we don't regenerate per-collection keys here.
+ let newSyncID = await this.resetSyncID();
+
+ // Set the newer version and newly generated syncID
+ engineData.version = this.version;
+ engineData.syncID = newSyncID;
+
+ // Put the new data back into meta/global and mark for upload
+ engines[this.name] = engineData;
+ metaGlobal.payload.engines = engines;
+ metaGlobal.changed = true;
+ } else if (engineData.version > this.version) {
+ // Don't sync this engine if the server has newer data
+
+ let error = new Error("New data: " + [engineData.version, this.version]);
+ error.failureCode = VERSION_OUT_OF_DATE;
+ throw error;
+ } else {
+ // Changes to syncID mean we'll need to upload everything
+ let assignedSyncID = await this.ensureCurrentSyncID(engineData.syncID);
+ if (assignedSyncID != engineData.syncID) {
+ engineData.syncID = assignedSyncID;
+ metaGlobal.changed = true;
+ }
+ }
+
+ // Save objects that need to be uploaded in this._modified. As we
+ // successfully upload objects we remove them from this._modified. If an
+ // error occurs or any objects fail to upload, they will remain in
+ // this._modified. At the end of a sync, or after an error, we add all
+ // objects remaining in this._modified to the tracker.
+ let initialChanges = await this.pullChanges();
+ this._modified.replace(initialChanges);
+ // Clear the tracker now. If the sync fails we'll add the ones we failed
+ // to upload back.
+ this._tracker.clearChangedIDs();
+ this._tracker.resetScore();
+
+ // Keep track of what to delete at the end of sync
+ this._delete = {};
+ },
+
+ async pullChanges() {
+ let lastSync = await this.getLastSync();
+ if (lastSync) {
+ return this.pullNewChanges();
+ }
+ this._log.debug("First sync, uploading all items");
+ return this.pullAllChanges();
+ },
+
+ /**
+ * A tiny abstraction to make it easier to test incoming record
+ * application.
+ */
+ itemSource() {
+ return new Collection(this.engineURL, this._recordObj, this.service);
+ },
+
+ /**
+ * Download and apply remote records changed since the last sync. This
+ * happens in three stages.
+ *
+ * In the first stage, we fetch full records for all changed items, newest
+ * first, up to the download limit. The limit lets us make progress for large
+ * collections, where the sync is likely to be interrupted before we
+ * can fetch everything.
+ *
+ * In the second stage, we fetch the IDs of any remaining records changed
+ * since the last sync, add them to our backlog, and fast-forward our last
+ * sync time.
+ *
+ * In the third stage, we fetch and apply records for all backlogged IDs,
+ * as well as any records that failed to apply during the last sync. We
+ * request records for the IDs in chunks, to avoid exceeding URL length
+ * limits, then remove successfully applied records from the backlog, and
+ * record IDs of any records that failed to apply to retry on the next sync.
+ */
+ async _processIncoming() {
+ this._log.trace("Downloading & applying server changes");
+
+ let newitems = this.itemSource();
+ let lastSync = await this.getLastSync();
+
+ newitems.newer = lastSync;
+ newitems.full = true;
+
+ let downloadLimit = Infinity;
+ if (this.downloadLimit) {
+ // Fetch new records up to the download limit. Currently, only the history
+ // engine sets a limit, since the history collection has the highest volume
+ // of changed records between syncs. The other engines fetch all records
+ // changed since the last sync.
+ if (this._defaultSort) {
+ // A download limit with a sort order doesn't make sense: we won't know
+ // which records to backfill.
+ throw new Error("Can't specify download limit with default sort order");
+ }
+ newitems.sort = "newest";
+ downloadLimit = newitems.limit = this.downloadLimit;
+ } else if (this._defaultSort) {
+ // The bookmarks engine fetches records by sort index; other engines leave
+ // the order unspecified. We can remove `_defaultSort` entirely after bug
+ // 1305563: the sort index won't matter because we'll buffer all bookmarks
+ // before applying.
+ newitems.sort = this._defaultSort;
+ }
+
+ // applied => number of items that should be applied.
+ // failed => number of items that failed in this sync.
+ // newFailed => number of items that failed for the first time in this sync.
+ // reconciled => number of items that were reconciled.
+ // failedReasons => {name, count} of reasons a record failed
+ let countTelemetry = new SyncedRecordsTelemetry();
+ let count = countTelemetry.incomingCounts;
+ let recordsToApply = [];
+ let failedInCurrentSync = new SerializableSet();
+
+ let oldestModified = this.lastModified;
+ let downloadedIDs = new Set();
+
+ // Stage 1: Fetch new records from the server, up to the download limit.
+ if (this.lastModified == null || this.lastModified > lastSync) {
+ let { response, records } = await newitems.getBatched(
+ this.downloadBatchSize
+ );
+ if (!response.success) {
+ response.failureCode = ENGINE_DOWNLOAD_FAIL;
+ throw response;
+ }
+
+ await Async.yieldingForEach(records, async record => {
+ downloadedIDs.add(record.id);
+
+ if (record.modified < oldestModified) {
+ oldestModified = record.modified;
+ }
+
+ let { shouldApply, error } = await this._maybeReconcile(record);
+ if (error) {
+ failedInCurrentSync.add(record.id);
+ count.failed++;
+ countTelemetry.addIncomingFailedReason(error.message);
+ return;
+ }
+ if (!shouldApply) {
+ count.reconciled++;
+ return;
+ }
+ recordsToApply.push(record);
+ });
+
+ let failedToApply = await this._applyRecords(
+ recordsToApply,
+ countTelemetry
+ );
+ Utils.setAddAll(failedInCurrentSync, failedToApply);
+
+ // `applied` is a bit of a misnomer: it counts records that *should* be
+ // applied, so it also includes records that we tried to apply and failed.
+ // `recordsToApply.length - failedToApply.length` is the number of records
+ // that we *successfully* applied.
+ count.failed += failedToApply.length;
+ count.applied += recordsToApply.length;
+ }
+
+ // Stage 2: If we reached our download limit, we might still have records
+ // on the server that changed since the last sync. Fetch the IDs for the
+ // remaining records, and add them to the backlog. Note that this stage
+ // only runs for engines that set a download limit.
+ if (downloadedIDs.size == downloadLimit) {
+ let guidColl = this.itemSource();
+
+ guidColl.newer = lastSync;
+ guidColl.older = oldestModified;
+ guidColl.sort = "oldest";
+
+ let guids = await guidColl.get();
+ if (!guids.success) {
+ throw guids;
+ }
+
+ // Filtering out already downloaded IDs here isn't necessary. We only do
+ // that in case the Sync server doesn't support `older` (bug 1316110).
+ let remainingIDs = guids.obj.filter(id => !downloadedIDs.has(id));
+ if (remainingIDs.length) {
+ this.toFetch = Utils.setAddAll(this.toFetch, remainingIDs);
+ }
+ }
+
+ // Fast-foward the lastSync timestamp since we have backlogged the
+ // remaining items.
+ if (lastSync < this.lastModified) {
+ lastSync = this.lastModified;
+ await this.setLastSync(lastSync);
+ }
+
+ // Stage 3: Backfill records from the backlog, and those that failed to
+ // decrypt or apply during the last sync. We only backfill up to the
+ // download limit, to prevent a large backlog for one engine from blocking
+ // the others. We'll keep processing the backlog on subsequent engine syncs.
+ let failedInPreviousSync = this.previousFailed;
+ let idsToBackfill = Array.from(
+ Utils.setAddAll(
+ Utils.subsetOfSize(this.toFetch, downloadLimit),
+ failedInPreviousSync
+ )
+ );
+
+ // Note that we intentionally overwrite the previously failed list here.
+ // Records that fail to decrypt or apply in two consecutive syncs are likely
+ // corrupt; we remove them from the list because retrying and failing on
+ // every subsequent sync just adds noise.
+ this.previousFailed = failedInCurrentSync;
+
+ let backfilledItems = this.itemSource();
+
+ backfilledItems.sort = "newest";
+ backfilledItems.full = true;
+
+ // `getBatched` includes the list of IDs as a query parameter, so we need to fetch
+ // records in chunks to avoid exceeding URI length limits.
+ if (this.guidFetchBatchSize) {
+ for (let ids of lazy.PlacesUtils.chunkArray(
+ idsToBackfill,
+ this.guidFetchBatchSize
+ )) {
+ backfilledItems.ids = ids;
+
+ let { response, records } = await backfilledItems.getBatched(
+ this.downloadBatchSize
+ );
+ if (!response.success) {
+ response.failureCode = ENGINE_DOWNLOAD_FAIL;
+ throw response;
+ }
+
+ let backfilledRecordsToApply = [];
+ let failedInBackfill = [];
+
+ await Async.yieldingForEach(records, async record => {
+ let { shouldApply, error } = await this._maybeReconcile(record);
+ if (error) {
+ failedInBackfill.push(record.id);
+ count.failed++;
+ countTelemetry.addIncomingFailedReason(error.message);
+ return;
+ }
+ if (!shouldApply) {
+ count.reconciled++;
+ return;
+ }
+ backfilledRecordsToApply.push(record);
+ });
+
+ let failedToApply = await this._applyRecords(
+ backfilledRecordsToApply,
+ countTelemetry
+ );
+ failedInBackfill.push(...failedToApply);
+
+ count.failed += failedToApply.length;
+ count.applied += backfilledRecordsToApply.length;
+
+ this.toFetch = Utils.setDeleteAll(this.toFetch, ids);
+ this.previousFailed = Utils.setAddAll(
+ this.previousFailed,
+ failedInBackfill
+ );
+
+ if (lastSync < this.lastModified) {
+ lastSync = this.lastModified;
+ await this.setLastSync(lastSync);
+ }
+ }
+ }
+
+ count.newFailed = 0;
+ for (let item of this.previousFailed) {
+ // Anything that failed in the current sync that also failed in
+ // the previous sync means there is likely something wrong with
+ // the record, we remove it from trying again to prevent
+ // infinitely syncing corrupted records
+ if (failedInPreviousSync.has(item)) {
+ this.previousFailed.delete(item);
+ } else {
+ // otherwise it's a new failed and we count it as so
+ ++count.newFailed;
+ }
+ }
+
+ count.succeeded = Math.max(0, count.applied - count.failed);
+ this._log.info(
+ [
+ "Records:",
+ count.applied,
+ "applied,",
+ count.succeeded,
+ "successfully,",
+ count.failed,
+ "failed to apply,",
+ count.newFailed,
+ "newly failed to apply,",
+ count.reconciled,
+ "reconciled.",
+ ].join(" ")
+ );
+ Observers.notify("weave:engine:sync:applied", count, this.name);
+ },
+
+ async _maybeReconcile(item) {
+ let key = this.service.collectionKeys.keyForCollection(this.name);
+
+ // Grab a later last modified if possible
+ if (this.lastModified == null || item.modified > this.lastModified) {
+ this.lastModified = item.modified;
+ }
+
+ try {
+ try {
+ await item.decrypt(key);
+ } catch (ex) {
+ if (!Utils.isHMACMismatch(ex)) {
+ throw ex;
+ }
+ let strategy = await this.handleHMACMismatch(item, true);
+ if (strategy == SyncEngine.kRecoveryStrategy.retry) {
+ // You only get one retry.
+ try {
+ // Try decrypting again, typically because we've got new keys.
+ this._log.info("Trying decrypt again...");
+ key = this.service.collectionKeys.keyForCollection(this.name);
+ await item.decrypt(key);
+ strategy = null;
+ } catch (ex) {
+ if (!Utils.isHMACMismatch(ex)) {
+ throw ex;
+ }
+ strategy = await this.handleHMACMismatch(item, false);
+ }
+ }
+
+ switch (strategy) {
+ case null:
+ // Retry succeeded! No further handling.
+ break;
+ case SyncEngine.kRecoveryStrategy.retry:
+ this._log.debug("Ignoring second retry suggestion.");
+ // Fall through to error case.
+ case SyncEngine.kRecoveryStrategy.error:
+ this._log.warn("Error decrypting record", ex);
+ return { shouldApply: false, error: ex };
+ case SyncEngine.kRecoveryStrategy.ignore:
+ this._log.debug(
+ "Ignoring record " + item.id + " with bad HMAC: already handled."
+ );
+ return { shouldApply: false, error: null };
+ }
+ }
+ } catch (ex) {
+ if (Async.isShutdownException(ex)) {
+ throw ex;
+ }
+ this._log.warn("Error decrypting record", ex);
+ return { shouldApply: false, error: ex };
+ }
+
+ if (this._shouldDeleteRemotely(item)) {
+ this._log.trace("Deleting item from server without applying", item);
+ await this._deleteId(item.id);
+ return { shouldApply: false, error: null };
+ }
+
+ let shouldApply;
+ try {
+ shouldApply = await this._reconcile(item);
+ } catch (ex) {
+ if (ex.code == SyncEngine.prototype.eEngineAbortApplyIncoming) {
+ this._log.warn("Reconciliation failed: aborting incoming processing.");
+ throw ex.cause;
+ } else if (!Async.isShutdownException(ex)) {
+ this._log.warn("Failed to reconcile incoming record " + item.id, ex);
+ return { shouldApply: false, error: ex };
+ } else {
+ throw ex;
+ }
+ }
+
+ if (!shouldApply) {
+ this._log.trace("Skipping reconciled incoming item " + item.id);
+ }
+
+ return { shouldApply, error: null };
+ },
+
+ async _applyRecords(records, countTelemetry) {
+ this._tracker.ignoreAll = true;
+ try {
+ let failedIDs = await this._store.applyIncomingBatch(
+ records,
+ countTelemetry
+ );
+ return failedIDs;
+ } catch (ex) {
+ // Catch any error that escapes from applyIncomingBatch. At present
+ // those will all be abort events.
+ this._log.warn("Got exception, aborting processIncoming", ex);
+ throw ex;
+ } finally {
+ this._tracker.ignoreAll = false;
+ }
+ },
+
+ // Indicates whether an incoming item should be deleted from the server at
+ // the end of the sync. Engines can override this method to clean up records
+ // that shouldn't be on the server.
+ _shouldDeleteRemotely(remoteItem) {
+ return false;
+ },
+
+ /**
+ * Find a GUID of an item that is a duplicate of the incoming item but happens
+ * to have a different GUID
+ *
+ * @return GUID of the similar item; falsy otherwise
+ */
+ async _findDupe(item) {
+ // By default, assume there's no dupe items for the engine
+ },
+
+ /**
+ * Called before a remote record is discarded due to failed reconciliation.
+ * Used by bookmark sync to merge folder child orders.
+ */
+ beforeRecordDiscard(localRecord, remoteRecord, remoteIsNewer) {},
+
+ // Called when the server has a record marked as deleted, but locally we've
+ // changed it more recently than the deletion. If we return false, the
+ // record will be deleted locally. If we return true, we'll reupload the
+ // record to the server -- any extra work that's needed as part of this
+ // process should be done at this point (such as mark the record's parent
+ // for reuploading in the case of bookmarks).
+ async _shouldReviveRemotelyDeletedRecord(remoteItem) {
+ return true;
+ },
+
+ async _deleteId(id) {
+ await this._tracker.removeChangedID(id);
+ this._noteDeletedId(id);
+ },
+
+ // Marks an ID for deletion at the end of the sync.
+ _noteDeletedId(id) {
+ if (this._delete.ids == null) {
+ this._delete.ids = [id];
+ } else {
+ this._delete.ids.push(id);
+ }
+ },
+
+ async _switchItemToDupe(localDupeGUID, incomingItem) {
+ // The local, duplicate ID is always deleted on the server.
+ await this._deleteId(localDupeGUID);
+
+ // We unconditionally change the item's ID in case the engine knows of
+ // an item but doesn't expose it through itemExists. If the API
+ // contract were stronger, this could be changed.
+ this._log.debug(
+ "Switching local ID to incoming: " +
+ localDupeGUID +
+ " -> " +
+ incomingItem.id
+ );
+ return this._store.changeItemID(localDupeGUID, incomingItem.id);
+ },
+
+ /**
+ * Reconcile incoming record with local state.
+ *
+ * This function essentially determines whether to apply an incoming record.
+ *
+ * @param item
+ * Record from server to be tested for application.
+ * @return boolean
+ * Truthy if incoming record should be applied. False if not.
+ */
+ async _reconcile(item) {
+ if (this._log.level <= Log.Level.Trace) {
+ this._log.trace("Incoming: " + item);
+ }
+
+ // We start reconciling by collecting a bunch of state. We do this here
+ // because some state may change during the course of this function and we
+ // need to operate on the original values.
+ let existsLocally = await this._store.itemExists(item.id);
+ let locallyModified = this._modified.has(item.id);
+
+ // TODO Handle clock drift better. Tracked in bug 721181.
+ let remoteAge = Resource.serverTime - item.modified;
+ let localAge = locallyModified
+ ? Date.now() / 1000 - this._modified.getModifiedTimestamp(item.id)
+ : null;
+ let remoteIsNewer = remoteAge < localAge;
+
+ this._log.trace(
+ "Reconciling " +
+ item.id +
+ ". exists=" +
+ existsLocally +
+ "; modified=" +
+ locallyModified +
+ "; local age=" +
+ localAge +
+ "; incoming age=" +
+ remoteAge
+ );
+
+ // We handle deletions first so subsequent logic doesn't have to check
+ // deleted flags.
+ if (item.deleted) {
+ // If the item doesn't exist locally, there is nothing for us to do. We
+ // can't check for duplicates because the incoming record has no data
+ // which can be used for duplicate detection.
+ if (!existsLocally) {
+ this._log.trace(
+ "Ignoring incoming item because it was deleted and " +
+ "the item does not exist locally."
+ );
+ return false;
+ }
+
+ // We decide whether to process the deletion by comparing the record
+ // ages. If the item is not modified locally, the remote side wins and
+ // the deletion is processed. If it is modified locally, we take the
+ // newer record.
+ if (!locallyModified) {
+ this._log.trace(
+ "Applying incoming delete because the local item " +
+ "exists and isn't modified."
+ );
+ return true;
+ }
+ this._log.trace("Incoming record is deleted but we had local changes.");
+
+ if (remoteIsNewer) {
+ this._log.trace("Remote record is newer -- deleting local record.");
+ return true;
+ }
+ // If the local record is newer, we defer to individual engines for
+ // how to handle this. By default, we revive the record.
+ let willRevive = await this._shouldReviveRemotelyDeletedRecord(item);
+ this._log.trace("Local record is newer -- reviving? " + willRevive);
+
+ return !willRevive;
+ }
+
+ // At this point the incoming record is not for a deletion and must have
+ // data. If the incoming record does not exist locally, we check for a local
+ // duplicate existing under a different ID. The default implementation of
+ // _findDupe() is empty, so engines have to opt in to this functionality.
+ //
+ // If we find a duplicate, we change the local ID to the incoming ID and we
+ // refresh the metadata collected above. See bug 710448 for the history
+ // of this logic.
+ if (!existsLocally) {
+ let localDupeGUID = await this._findDupe(item);
+ if (localDupeGUID) {
+ this._log.trace(
+ "Local item " +
+ localDupeGUID +
+ " is a duplicate for " +
+ "incoming item " +
+ item.id
+ );
+
+ // The current API contract does not mandate that the ID returned by
+ // _findDupe() actually exists. Therefore, we have to perform this
+ // check.
+ existsLocally = await this._store.itemExists(localDupeGUID);
+
+ // If the local item was modified, we carry its metadata forward so
+ // appropriate reconciling can be performed.
+ if (this._modified.has(localDupeGUID)) {
+ locallyModified = true;
+ localAge =
+ this._tracker._now() -
+ this._modified.getModifiedTimestamp(localDupeGUID);
+ remoteIsNewer = remoteAge < localAge;
+
+ this._modified.changeID(localDupeGUID, item.id);
+ } else {
+ locallyModified = false;
+ localAge = null;
+ }
+
+ // Tell the engine to do whatever it needs to switch the items.
+ await this._switchItemToDupe(localDupeGUID, item);
+
+ this._log.debug(
+ "Local item after duplication: age=" +
+ localAge +
+ "; modified=" +
+ locallyModified +
+ "; exists=" +
+ existsLocally
+ );
+ } else {
+ this._log.trace("No duplicate found for incoming item: " + item.id);
+ }
+ }
+
+ // At this point we've performed duplicate detection. But, nothing here
+ // should depend on duplicate detection as the above should have updated
+ // state seamlessly.
+
+ if (!existsLocally) {
+ // If the item doesn't exist locally and we have no local modifications
+ // to the item (implying that it was not deleted), always apply the remote
+ // item.
+ if (!locallyModified) {
+ this._log.trace(
+ "Applying incoming because local item does not exist " +
+ "and was not deleted."
+ );
+ return true;
+ }
+
+ // If the item was modified locally but isn't present, it must have
+ // been deleted. If the incoming record is younger, we restore from
+ // that record.
+ if (remoteIsNewer) {
+ this._log.trace(
+ "Applying incoming because local item was deleted " +
+ "before the incoming item was changed."
+ );
+ this._modified.delete(item.id);
+ return true;
+ }
+
+ this._log.trace(
+ "Ignoring incoming item because the local item's " +
+ "deletion is newer."
+ );
+ return false;
+ }
+
+ // If the remote and local records are the same, there is nothing to be
+ // done, so we don't do anything. In the ideal world, this logic wouldn't
+ // be here and the engine would take a record and apply it. The reason we
+ // want to defer this logic is because it would avoid a redundant and
+ // possibly expensive dip into the storage layer to query item state.
+ // This should get addressed in the async rewrite, so we ignore it for now.
+ let localRecord = await this._createRecord(item.id);
+ let recordsEqual = Utils.deepEquals(item.cleartext, localRecord.cleartext);
+
+ // If the records are the same, we don't need to do anything. This does
+ // potentially throw away a local modification time. But, if the records
+ // are the same, does it matter?
+ if (recordsEqual) {
+ this._log.trace(
+ "Ignoring incoming item because the local item is identical."
+ );
+
+ this._modified.delete(item.id);
+ return false;
+ }
+
+ // At this point the records are different.
+
+ // If we have no local modifications, always take the server record.
+ if (!locallyModified) {
+ this._log.trace("Applying incoming record because no local conflicts.");
+ return true;
+ }
+
+ // At this point, records are different and the local record is modified.
+ // We resolve conflicts by record age, where the newest one wins. This does
+ // result in data loss and should be handled by giving the engine an
+ // opportunity to merge the records. Bug 720592 tracks this feature.
+ this._log.warn(
+ "DATA LOSS: Both local and remote changes to record: " + item.id
+ );
+ if (!remoteIsNewer) {
+ this.beforeRecordDiscard(localRecord, item, remoteIsNewer);
+ }
+ return remoteIsNewer;
+ },
+
+ // Upload outgoing records.
+ async _uploadOutgoing() {
+ this._log.trace("Uploading local changes to server.");
+
+ // collection we'll upload
+ let up = new Collection(this.engineURL, null, this.service);
+ let modifiedIDs = new Set(this._modified.ids());
+ let countTelemetry = new SyncedRecordsTelemetry();
+ let counts = countTelemetry.outgoingCounts;
+ this._log.info(`Uploading ${modifiedIDs.size} outgoing records`);
+ if (modifiedIDs.size) {
+ counts.sent = modifiedIDs.size;
+
+ let failed = [];
+ let successful = [];
+ let lastSync = await this.getLastSync();
+ let handleResponse = async (postQueue, resp, batchOngoing) => {
+ // Note: We don't want to update this.lastSync, or this._modified until
+ // the batch is complete, however we want to remember success/failure
+ // indicators for when that happens.
+ if (!resp.success) {
+ this._log.debug(`Uploading records failed: ${resp.status}`);
+ resp.failureCode =
+ resp.status == 412 ? ENGINE_BATCH_INTERRUPTED : ENGINE_UPLOAD_FAIL;
+ throw resp;
+ }
+
+ // Update server timestamp from the upload.
+ failed = failed.concat(Object.keys(resp.obj.failed));
+ successful = successful.concat(resp.obj.success);
+
+ if (batchOngoing) {
+ // Nothing to do yet
+ return;
+ }
+
+ if (failed.length && this._log.level <= Log.Level.Debug) {
+ this._log.debug(
+ "Records that will be uploaded again because " +
+ "the server couldn't store them: " +
+ failed.join(", ")
+ );
+ }
+
+ counts.failed += failed.length;
+ Object.values(failed).forEach(message => {
+ countTelemetry.addOutgoingFailedReason(message);
+ });
+
+ for (let id of successful) {
+ this._modified.delete(id);
+ }
+
+ await this._onRecordsWritten(
+ successful,
+ failed,
+ postQueue.lastModified
+ );
+
+ // Advance lastSync since we've finished the batch.
+ if (postQueue.lastModified > lastSync) {
+ lastSync = postQueue.lastModified;
+ await this.setLastSync(lastSync);
+ }
+
+ // clear for next batch
+ failed.length = 0;
+ successful.length = 0;
+ };
+
+ let postQueue = up.newPostQueue(this._log, lastSync, handleResponse);
+
+ for (let id of modifiedIDs) {
+ let out;
+ let ok = false;
+ try {
+ out = await this._createRecord(id);
+ if (this._log.level <= Log.Level.Trace) {
+ this._log.trace("Outgoing: " + out);
+ }
+ await out.encrypt(
+ this.service.collectionKeys.keyForCollection(this.name)
+ );
+ ok = true;
+ } catch (ex) {
+ this._log.warn("Error creating record", ex);
+ ++counts.failed;
+ countTelemetry.addOutgoingFailedReason(ex.message);
+ if (Async.isShutdownException(ex) || !this.allowSkippedRecord) {
+ if (!this.allowSkippedRecord) {
+ // Don't bother for shutdown errors
+ Observers.notify("weave:engine:sync:uploaded", counts, this.name);
+ }
+ throw ex;
+ }
+ }
+ if (ok) {
+ let { enqueued, error } = await postQueue.enqueue(out);
+ if (!enqueued) {
+ ++counts.failed;
+ countTelemetry.addOutgoingFailedReason(error.message);
+ if (!this.allowSkippedRecord) {
+ Observers.notify("weave:engine:sync:uploaded", counts, this.name);
+ this._log.warn(
+ `Failed to enqueue record "${id}" (aborting)`,
+ error
+ );
+ throw error;
+ }
+ this._modified.delete(id);
+ this._log.warn(
+ `Failed to enqueue record "${id}" (skipping)`,
+ error
+ );
+ }
+ }
+ await Async.promiseYield();
+ }
+ await postQueue.flush(true);
+ }
+
+ if (counts.sent || counts.failed) {
+ Observers.notify("weave:engine:sync:uploaded", counts, this.name);
+ }
+ },
+
+ async _onRecordsWritten(succeeded, failed, serverModifiedTime) {
+ // Implement this method to take specific actions against successfully
+ // uploaded records and failed records.
+ },
+
+ // Any cleanup necessary.
+ // Save the current snapshot so as to calculate changes at next sync
+ async _syncFinish() {
+ this._log.trace("Finishing up sync");
+
+ let doDelete = async (key, val) => {
+ let coll = new Collection(this.engineURL, this._recordObj, this.service);
+ coll[key] = val;
+ await coll.delete();
+ };
+
+ for (let [key, val] of Object.entries(this._delete)) {
+ // Remove the key for future uses
+ delete this._delete[key];
+
+ this._log.trace("doing post-sync deletions", { key, val });
+ // Send a simple delete for the property
+ if (key != "ids" || val.length <= 100) {
+ await doDelete(key, val);
+ } else {
+ // For many ids, split into chunks of at most 100
+ while (val.length) {
+ await doDelete(key, val.slice(0, 100));
+ val = val.slice(100);
+ }
+ }
+ }
+ this.hasSyncedThisSession = true;
+ await this._tracker.asyncObserver.promiseObserversComplete();
+ },
+
+ async _syncCleanup() {
+ try {
+ // Mark failed WBOs as changed again so they are reuploaded next time.
+ await this.trackRemainingChanges();
+ } finally {
+ this._modified.clear();
+ }
+ },
+
+ async _sync() {
+ try {
+ Async.checkAppReady();
+ await this._syncStartup();
+ Async.checkAppReady();
+ Observers.notify("weave:engine:sync:status", "process-incoming");
+ await this._processIncoming();
+ Async.checkAppReady();
+ Observers.notify("weave:engine:sync:status", "upload-outgoing");
+ try {
+ await this._uploadOutgoing();
+ Async.checkAppReady();
+ await this._syncFinish();
+ } catch (ex) {
+ if (!ex.status || ex.status != 412) {
+ throw ex;
+ }
+ // a 412 posting just means another client raced - but we don't want
+ // to treat that as a sync error - the next sync is almost certain
+ // to work.
+ this._log.warn("412 error during sync - will retry.");
+ }
+ } finally {
+ await this._syncCleanup();
+ }
+ },
+
+ async canDecrypt() {
+ // Report failure even if there's nothing to decrypt
+ let canDecrypt = false;
+
+ // Fetch the most recently uploaded record and try to decrypt it
+ let test = new Collection(this.engineURL, this._recordObj, this.service);
+ test.limit = 1;
+ test.sort = "newest";
+ test.full = true;
+
+ let key = this.service.collectionKeys.keyForCollection(this.name);
+
+ // Any failure fetching/decrypting will just result in false
+ try {
+ this._log.trace("Trying to decrypt a record from the server..");
+ let json = (await test.get()).obj[0];
+ let record = new this._recordObj();
+ record.deserialize(json);
+ await record.decrypt(key);
+ canDecrypt = true;
+ } catch (ex) {
+ if (Async.isShutdownException(ex)) {
+ throw ex;
+ }
+ this._log.debug("Failed test decrypt", ex);
+ }
+
+ return canDecrypt;
+ },
+
+ /**
+ * Deletes the collection for this engine on the server, and removes all local
+ * Sync metadata for this engine. This does *not* remove any existing data on
+ * other clients. This is called when we reset the sync ID.
+ */
+ async wipeServer() {
+ await this._deleteServerCollection();
+ await this._resetClient();
+ },
+
+ /**
+ * Deletes the collection for this engine on the server, without removing
+ * any local Sync metadata or user data. Deleting the collection will not
+ * remove any user data on other clients, but will force other clients to
+ * start over as a first sync.
+ */
+ async _deleteServerCollection() {
+ let response = await this.service.resource(this.engineURL).delete();
+ if (response.status != 200 && response.status != 404) {
+ throw response;
+ }
+ },
+
+ async removeClientData() {
+ // Implement this method in engines that store client specific data
+ // on the server.
+ },
+
+ /*
+ * Decide on (and partially effect) an error-handling strategy.
+ *
+ * Asks the Service to respond to an HMAC error, which might result in keys
+ * being downloaded. That call returns true if an action which might allow a
+ * retry to occur.
+ *
+ * If `mayRetry` is truthy, and the Service suggests a retry,
+ * handleHMACMismatch returns kRecoveryStrategy.retry. Otherwise, it returns
+ * kRecoveryStrategy.error.
+ *
+ * Subclasses of SyncEngine can override this method to allow for different
+ * behavior -- e.g., to delete and ignore erroneous entries.
+ *
+ * All return values will be part of the kRecoveryStrategy enumeration.
+ */
+ async handleHMACMismatch(item, mayRetry) {
+ // By default we either try again, or bail out noisily.
+ return (await this.service.handleHMACEvent()) && mayRetry
+ ? SyncEngine.kRecoveryStrategy.retry
+ : SyncEngine.kRecoveryStrategy.error;
+ },
+
+ /**
+ * Returns a changeset containing all items in the store. The default
+ * implementation returns a changeset with timestamps from long ago, to
+ * ensure we always use the remote version if one exists.
+ *
+ * This function is only called for the first sync. Subsequent syncs call
+ * `pullNewChanges`.
+ *
+ * @return A `Changeset` object.
+ */
+ async pullAllChanges() {
+ let changes = {};
+ let ids = await this._store.getAllIDs();
+ for (let id in ids) {
+ changes[id] = 0;
+ }
+ return changes;
+ },
+
+ /*
+ * Returns a changeset containing entries for all currently tracked items.
+ * The default implementation returns a changeset with timestamps indicating
+ * when the item was added to the tracker.
+ *
+ * @return A `Changeset` object.
+ */
+ async pullNewChanges() {
+ await this._tracker.asyncObserver.promiseObserversComplete();
+ return this.getChangedIDs();
+ },
+
+ /**
+ * Adds all remaining changeset entries back to the tracker, typically for
+ * items that failed to upload. This method is called at the end of each sync.
+ *
+ */
+ async trackRemainingChanges() {
+ for (let [id, change] of this._modified.entries()) {
+ await this._tracker.addChangedID(id, change);
+ }
+ },
+
+ /**
+ * Removes all local Sync metadata for this engine, but keeps all existing
+ * local user data.
+ */
+ async resetClient() {
+ return this._notify("reset-client", this.name, this._resetClient)();
+ },
+
+ async _resetClient() {
+ await this.resetLastSync();
+ this.hasSyncedThisSession = false;
+ this.previousFailed = new SerializableSet();
+ this.toFetch = new SerializableSet();
+ },
+
+ /**
+ * Removes all local Sync metadata and user data for this engine.
+ */
+ async wipeClient() {
+ return this._notify("wipe-client", this.name, this._wipeClient)();
+ },
+
+ async _wipeClient() {
+ await this.resetClient();
+ this._log.debug("Deleting all local data");
+ this._tracker.ignoreAll = true;
+ await this._store.wipe();
+ this._tracker.ignoreAll = false;
+ this._tracker.clearChangedIDs();
+ },
+
+ /**
+ * If one exists, initialize and return a validator for this engine (which
+ * must have a `validate(engine)` method that returns a promise to an object
+ * with a getSummary method). Otherwise return null.
+ */
+ getValidator() {
+ return null;
+ },
+
+ async finalize() {
+ Services.prefs.removeObserver(
+ `${PREFS_BRANCH}engine.${this.prefName}`,
+ this.asyncObserver
+ );
+ await this.asyncObserver.promiseObserversComplete();
+ await this._tracker.finalize();
+ await this._toFetchStorage.finalize();
+ await this._previousFailedStorage.finalize();
+ },
+
+ // Returns a new watchdog. Exposed for tests.
+ _newWatchdog() {
+ return Async.watchdog();
+ },
+};
+
+/**
+ * A changeset is created for each sync in `Engine::get{Changed, All}IDs`,
+ * and stores opaque change data for tracked IDs. The default implementation
+ * only records timestamps, though engines can extend this to store additional
+ * data for each entry.
+ */
+export class Changeset {
+ // Creates an empty changeset.
+ constructor() {
+ this.changes = {};
+ }
+
+ // Returns the last modified time, in seconds, for an entry in the changeset.
+ // `id` is guaranteed to be in the set.
+ getModifiedTimestamp(id) {
+ return this.changes[id];
+ }
+
+ // Adds a change for a tracked ID to the changeset.
+ set(id, change) {
+ this.changes[id] = change;
+ }
+
+ // Adds multiple entries to the changeset, preserving existing entries.
+ insert(changes) {
+ Object.assign(this.changes, changes);
+ }
+
+ // Overwrites the existing set of tracked changes with new entries.
+ replace(changes) {
+ this.changes = changes;
+ }
+
+ // Indicates whether an entry is in the changeset.
+ has(id) {
+ return id in this.changes;
+ }
+
+ // Deletes an entry from the changeset. Used to clean up entries for
+ // reconciled and successfully uploaded records.
+ delete(id) {
+ delete this.changes[id];
+ }
+
+ // Changes the ID of an entry in the changeset. Used when reconciling
+ // duplicates that have local changes.
+ changeID(oldID, newID) {
+ this.changes[newID] = this.changes[oldID];
+ delete this.changes[oldID];
+ }
+
+ // Returns an array of all tracked IDs in this changeset.
+ ids() {
+ return Object.keys(this.changes);
+ }
+
+ // Returns an array of `[id, change]` tuples. Used to repopulate the tracker
+ // with entries for failed uploads at the end of a sync.
+ entries() {
+ return Object.entries(this.changes);
+ }
+
+ // Returns the number of entries in this changeset.
+ count() {
+ return this.ids().length;
+ }
+
+ // Clears the changeset.
+ clear() {
+ this.changes = {};
+ }
+}