The Tor Browser: comparison services/sync/modules/engines.js

comparison: services/sync/modules/engines.js

services/sync/modules/engines.js

branch: TOR_BUG_9701
changeset 15: b8a032363ba2

equal deleted inserted replaced

--1:000000000000
+:0057b5968240
+/* 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/. */
+this.EXPORTED_SYMBOLS = [
+"EngineManager",
+"Engine",
+"SyncEngine",
+"Tracker",
+"Store"
+];
+const {classes: Cc, interfaces: Ci, results: Cr, utils: Cu} = Components;
+Cu.import("resource://services-common/async.js");
+Cu.import("resource://gre/modules/Log.jsm");
+Cu.import("resource://services-common/observers.js");
+Cu.import("resource://services-common/utils.js");
+Cu.import("resource://services-sync/constants.js");
+Cu.import("resource://services-sync/identity.js");
+Cu.import("resource://services-sync/record.js");
+Cu.import("resource://services-sync/resource.js");
+Cu.import("resource://services-sync/util.js");
+/*
+* Trackers are associated with a single engine and deal with
+* listening for changes to their particular data type.
+*
+* There are two things they keep track of:
+* 1) A score, indicating how urgently the engine wants to sync
+* 2) A list of IDs for all the changed items that need to be synced
+* and updating their 'score', indicating how urgently they
+* want to sync.
+*
+*/
+this.Tracker = function Tracker(name, engine) {
+if (!engine) {
+throw new Error("Tracker must be associated with an Engine instance.");
+}
+name = name || "Unnamed";
+this.name = this.file = name.toLowerCase();
+this.engine = engine;
+this._log = Log.repository.getLogger("Sync.Tracker." + name);
+let level = Svc.Prefs.get("log.logger.engine." + this.name, "Debug");
+this._log.level = Log.Level[level];
+this._score = 0;
+this._ignored = [];
+this.ignoreAll = false;
+this.changedIDs = {};
+this.loadChangedIDs();
+Svc.Obs.add("weave:engine:start-tracking", this);
+Svc.Obs.add("weave:engine:stop-tracking", this);
+};
+Tracker.prototype = {
+/*
+* 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: function () {
+this._score = 0;
+},
+persistChangedIDs: true,
+/**
+* Persist changedIDs to disk at a later date.
+* Optionally pass a callback to be invoked when the write has occurred.
+*/
+saveChangedIDs: function (cb) {
+if (!this.persistChangedIDs) {
+this._log.debug("Not saving changedIDs.");
+return;
+}
+Utils.namedTimer(function () {
+this._log.debug("Saving changed IDs to " + this.file);
+Utils.jsonSave("changes/" + this.file, this, this.changedIDs, cb);
+}, 1000, this, "_lazySave");
+},
+loadChangedIDs: function (cb) {
+Utils.jsonLoad("changes/" + this.file, this, function(json) {
+if (json && (typeof(json) == "object")) {
+this.changedIDs = json;
+} else {
+this._log.warn("Changed IDs file " + this.file + " contains non-object value.");
+json = null;
+}
+if (cb) {
+cb.call(this, json);
+}
+});
+},
+// 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: function (id) {
+this.unignoreID(id);
+this._ignored.push(id);
+},
+unignoreID: function (id) {
+let index = this._ignored.indexOf(id);
+if (index != -1)
+this._ignored.splice(index, 1);
+},
+addChangedID: function (id, when) {
+if (!id) {
+this._log.warn("Attempted to add undefined ID to tracker");
+return false;
+}
+if (this.ignoreAll || (id in this._ignored)) {
+return false;
+}
+// Default to the current time in seconds if no time is provided.
+if (when == null) {
+when = Math.floor(Date.now() / 1000);
+}
+// Add/update the entry if we have a newer time.
+if ((this.changedIDs[id] || -Infinity) < when) {
+this._log.trace("Adding changed ID: " + id + ", " + when);
+this.changedIDs[id] = when;
+this.saveChangedIDs(this.onSavedChangedIDs);
+}
+return true;
+},
+removeChangedID: function (id) {
+if (!id) {
+this._log.warn("Attempted to remove undefined ID to tracker");
+return false;
+}
+if (this.ignoreAll || (id in this._ignored))
+return false;
+if (this.changedIDs[id] != null) {
+this._log.trace("Removing changed ID " + id);
+delete this.changedIDs[id];
+this.saveChangedIDs();
+}
+return true;
+},
+clearChangedIDs: function () {
+this._log.trace("Clearing changed ID list");
+this.changedIDs = {};
+this.saveChangedIDs();
+},
+_isTracking: false,
+// Override these in your subclasses.
+startTracking: function () {
+},
+stopTracking: function () {
+},
+engineIsEnabled: function () {
+if (!this.engine) {
+// Can't tell -- we must be running in a test!
+return true;
+}
+return this.engine.enabled;
+},
+onEngineEnabledChanged: function (engineEnabled) {
+if (engineEnabled == this._isTracking) {
+return;
+}
+if (engineEnabled) {
+this.startTracking();
+this._isTracking = true;
+} else {
+this.stopTracking();
+this._isTracking = false;
+this.clearChangedIDs();
+}
+},
+observe: function (subject, topic, data) {
+switch (topic) {
+case "weave:engine:start-tracking":
+if (!this.engineIsEnabled()) {
+return;
+}
+this._log.trace("Got start-tracking.");
+if (!this._isTracking) {
+this.startTracking();
+this._isTracking = true;
+}
+return;
+case "weave:engine:stop-tracking":
+this._log.trace("Got stop-tracking.");
+if (this._isTracking) {
+this.stopTracking();
+this._isTracking = false;
+}
+return;
+}
+}
+};
+/**
+* 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.
+*/
+this.Store = 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.Store." + name);
+let level = Svc.Prefs.get("log.logger.engine." + this.name, "Debug");
+this._log.level = Log.Level[level];
+XPCOMUtils.defineLazyGetter(this, "_timer", function() {
+return Cc["@mozilla.org/timer;1"].createInstance(Ci.nsITimer);
+});
+}
+Store.prototype = {
+_sleep: function _sleep(delay) {
+let cb = Async.makeSyncCallback();
+this._timer.initWithCallback(cb, delay, Ci.nsITimer.TYPE_ONE_SHOT);
+Async.waitForSyncCallback(cb);
+},
+/**
+* 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
+* @return Array of record IDs which did not apply cleanly
+*/
+applyIncomingBatch: function (records) {
+let failed = [];
+for each (let record in records) {
+try {
+this.applyIncoming(record);
+} catch (ex if (ex.code == Engine.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;
+} catch (ex) {
+this._log.warn("Failed to apply incoming record " + record.id);
+this._log.warn("Encountered exception: " + Utils.exceptionStr(ex));
+failed.push(record.id);
+}
+};
+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
+*/
+applyIncoming: function (record) {
+if (record.deleted)
+this.remove(record);
+else if (!this.itemExists(record.id))
+this.create(record);
+else
+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
+*/
+create: function (record) {
+throw "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
+*/
+remove: function (record) {
+throw "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
+*/
+update: function (record) {
+throw "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
+*/
+itemExists: function (id) {
+throw "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
+*/
+createRecord: function (id, collection) {
+throw "override createRecord in a subclass";
+},
+/**
+* Change the ID of a record.
+*
+* @param  oldID
+*         string old/current record ID
+* @param  newID
+*         string new record ID
+*/
+changeItemID: function (oldID, newID) {
+throw "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.
+*/
+getAllIDs: function () {
+throw "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.
+*/
+wipe: function () {
+throw "override wipe in a subclass";
+}
+};
+this.EngineManager = function EngineManager(service) {
+this.service = service;
+this._engines = {};
+// This will be populated by Service on startup.
+this._declined = new Set();
+this._log = Log.repository.getLogger("Sync.EngineManager");
+this._log.level = Log.Level[Svc.Prefs.get("log.logger.service.engines", "Debug")];
+}
+EngineManager.prototype = {
+get: function (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;
+}
+let engine = this._engines[name];
+if (!engine) {
+this._log.debug("Could not get engine: " + name);
+if (Object.keys) {
+this._log.debug("Engines are: " + JSON.stringify(Object.keys(this._engines)));
+}
+}
+return engine;
+},
+getAll: function () {
+return [engine for ([name, engine] in Iterator(this._engines))];
+},
+/**
+* N.B., does not pay attention to the declined list.
+*/
+getEnabled: function () {
+return this.getAll().filter(function(engine) engine.enabled);
+},
+get enabledEngineNames() {
+return [e.name for each (e in this.getEnabled())];
+},
+persistDeclined: function () {
+Svc.Prefs.set("declinedEngines", [...this._declined].join(","));
+},
+/**
+* Returns an array.
+*/
+getDeclined: function () {
+return [...this._declined];
+},
+setDeclined: function (engines) {
+this._declined = new Set(engines);
+this.persistDeclined();
+},
+isDeclined: function (engineName) {
+return this._declined.has(engineName);
+},
+/**
+* Accepts a Set or an array.
+*/
+decline: function (engines) {
+for (let e of engines) {
+this._declined.add(e);
+}
+this.persistDeclined();
+},
+undecline: function (engines) {
+for (let e of engines) {
+this._declined.delete(e);
+}
+this.persistDeclined();
+},
+/**
+* Mark any non-enabled engines as declined.
+*
+* This is useful after initial customization during setup.
+*/
+declineDisabled: function () {
+for (let e of this.getAll()) {
+if (!e.enabled) {
+this._log.debug("Declining disabled engine " + e.name);
+this._declined.add(e.name);
+}
+}
+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
+*/
+register: function (engineObject) {
+if (Array.isArray(engineObject)) {
+return engineObject.map(this.register, this);
+}
+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 {
+this._engines[name] = engine;
+}
+} catch (ex) {
+this._log.error(CommonUtils.exceptionStr(ex));
+let mesg = ex.message ? ex.message : ex;
+let name = engineObject || "";
+name = name.prototype || "";
+name = name.name || "";
+let out = "Could not initialize engine '" + name + "': " + mesg;
+this._log.error(out);
+return engineObject;
+}
+},
+unregister: function (val) {
+let name = val;
+if (val instanceof Engine) {
+name = val.name;
+}
+delete this._engines[name];
+},
+clear: function () {
+for (let name in this._engines) {
+delete this._engines[name];
+}
+},
+};
+this.Engine = function Engine(name, service) {
+if (!service) {
+throw new Error("Engine 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);
+let level = Svc.Prefs.get("log.logger.engine." + this.name, "Debug");
+this._log.level = Log.Level[level];
+this._tracker; // initialize tracker to load previously changed IDs
+this._log.debug("Engine initialized");
+}
+Engine.prototype = {
+// _storeObj, and _trackerObj should to be overridden in subclasses
+_storeObj: Store,
+_trackerObj: Tracker,
+// Local 'constant'.
+// Signal to the engine that processing further records is pointless.
+eEngineAbortApplyIncoming: "error.engine.abort.applyincoming",
+get prefName() this.name,
+get enabled() {
+return Svc.Prefs.get("engine." + this.prefName, false);
+},
+set enabled(val) {
+Svc.Prefs.set("engine." + this.prefName, !!val);
+this._tracker.onEngineEnabledChanged(val);
+},
+get score() this._tracker.score,
+get _store() {
+let store = new this._storeObj(this.Name, this);
+this.__defineGetter__("_store", function() store);
+return store;
+},
+get _tracker() {
+let tracker = new this._trackerObj(this.Name, this);
+this.__defineGetter__("_tracker", function() tracker);
+return tracker;
+},
+sync: function () {
+if (!this.enabled) {
+return;
+}
+if (!this._sync) {
+throw "engine does not implement _sync method";
+}
+this._notify("sync", this.name, this._sync)();
+},
+/**
+* Get rid of any local meta-data.
+*/
+resetClient: function () {
+if (!this._resetClient) {
+throw "engine does not implement _resetClient method";
+}
+this._notify("reset-client", this.name, this._resetClient)();
+},
+_wipeClient: function () {
+this.resetClient();
+this._log.debug("Deleting all local data");
+this._tracker.ignoreAll = true;
+this._store.wipe();
+this._tracker.ignoreAll = false;
+this._tracker.clearChangedIDs();
+},
+wipeClient: function () {
+this._notify("wipe-client", this.name, this._wipeClient)();
+}
+};
+this.SyncEngine = function SyncEngine(name, service) {
+Engine.call(this, name || "SyncEngine", service);
+this.loadToFetch();
+this.loadPreviousFailed();
+}
+// 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 = {
+__proto__: Engine.prototype,
+_recordObj: CryptoWrapper,
+version: 1,
+// 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,
+mobileGUIDFetchBatchSize: DEFAULT_MOBILE_GUID_FETCH_BATCH_SIZE,
+// How many records to process in a single batch.
+applyIncomingBatchSize: DEFAULT_STORE_BATCH_SIZE,
+get storageURL() this.service.storageURL,
+get engineURL() this.storageURL + this.name,
+get cryptoKeysURL() this.storageURL + "crypto/keys",
+get metaURL() this.storageURL + "meta/global",
+get syncID() {
+// Generate a random syncID if we don't have one
+let syncID = Svc.Prefs.get(this.name + ".syncID", "");
+return syncID == "" ? this.syncID = Utils.makeGUID() : syncID;
+},
+set syncID(value) {
+Svc.Prefs.set(this.name + ".syncID", value);
+},
+/*
+* lastSync is a timestamp in server time.
+*/
+get lastSync() {
+return parseFloat(Svc.Prefs.get(this.name + ".lastSync", "0"));
+},
+set lastSync(value) {
+// Reset the pref in-case it's a number instead of a string
+Svc.Prefs.reset(this.name + ".lastSync");
+// Store the value as a string to keep floating point precision
+Svc.Prefs.set(this.name + ".lastSync", value.toString());
+},
+resetLastSync: function () {
+this._log.debug("Resetting " + this.name + " last sync time");
+Svc.Prefs.reset(this.name + ".lastSync");
+Svc.Prefs.set(this.name + ".lastSync", "0");
+this.lastSyncLocal = 0;
+},
+get toFetch() this._toFetch,
+set toFetch(val) {
+let cb = (error) => this._log.error(Utils.exceptionStr(error));
+// Coerce the array to a string for more efficient comparison.
+if (val + "" == this._toFetch) {
+return;
+}
+this._toFetch = val;
+Utils.namedTimer(function () {
+Utils.jsonSave("toFetch/" + this.name, this, val, cb);
+}, 0, this, "_toFetchDelay");
+},
+loadToFetch: function () {
+// Initialize to empty if there's no file.
+this._toFetch = [];
+Utils.jsonLoad("toFetch/" + this.name, this, function(toFetch) {
+if (toFetch) {
+this._toFetch = toFetch;
+}
+});
+},
+get previousFailed() this._previousFailed,
+set previousFailed(val) {
+let cb = (error) => this._log.error(Utils.exceptionStr(error));
+// Coerce the array to a string for more efficient comparison.
+if (val + "" == this._previousFailed) {
+return;
+}
+this._previousFailed = val;
+Utils.namedTimer(function () {
+Utils.jsonSave("failed/" + this.name, this, val, cb);
+}, 0, this, "_previousFailedDelay");
+},
+loadPreviousFailed: function () {
+// Initialize to empty if there's no file
+this._previousFailed = [];
+Utils.jsonLoad("failed/" + this.name, this, function(previousFailed) {
+if (previousFailed) {
+this._previousFailed = previousFailed;
+}
+});
+},
+/*
+* lastSyncLocal is a timestamp in local time.
+*/
+get lastSyncLocal() {
+return parseInt(Svc.Prefs.get(this.name + ".lastSyncLocal", "0"), 10);
+},
+set lastSyncLocal(value) {
+// Store as a string because pref can only store C longs as numbers.
+Svc.Prefs.set(this.name + ".lastSyncLocal", value.toString());
+},
+/*
+* Returns a mapping of IDs -> changed timestamp. Engine implementations
+* can override this method to bypass the tracker for certain or all
+* changed items.
+*/
+getChangedIDs: function () {
+return this._tracker.changedIDs;
+},
+// Create a new record using the store and add in crypto fields.
+_createRecord: function (id) {
+let record = this._store.createRecord(id, this.name);
+record.id = id;
+record.collection = this.name;
+return record;
+},
+// Any setup that needs to happen at the beginning of each sync.
+_syncStartup: function () {
+// Determine if we need to wipe on outdated versions
+let metaGlobal = this.service.recordManager.get(this.metaURL);
+let engines = metaGlobal.payload.engines || {};
+let engineData = engines[this.name] || {};
+let needsWipe = false;
+// 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]);
+// Prepare to clear the server and upload everything
+needsWipe = true;
+this.syncID = "";
+// Set the newer version and newly generated syncID
+engineData.version = this.version;
+engineData.syncID = this.syncID;
+// Put the new data back into meta/global and mark for upload
+engines[this.name] = engineData;
+metaGlobal.payload.engines = engines;
+metaGlobal.changed = true;
+}
+// Don't sync this engine if the server has newer data
+else if (engineData.version > this.version) {
+let error = new String("New data: " + [engineData.version, this.version]);
+error.failureCode = VERSION_OUT_OF_DATE;
+throw error;
+}
+// Changes to syncID mean we'll need to upload everything
+else if (engineData.syncID != this.syncID) {
+this._log.debug("Engine syncIDs: " + [engineData.syncID, this.syncID]);
+this.syncID = engineData.syncID;
+this._resetClient();
+};
+// Delete any existing data and reupload on bad version or missing meta.
+// No crypto component here...? We could regenerate per-collection keys...
+if (needsWipe) {
+this.wipeServer();
+}
+// Save objects that need to be uploaded in this._modified. We also save
+// the timestamp of this fetch in this.lastSyncLocal. 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.
+this.lastSyncLocal = Date.now();
+if (this.lastSync) {
+this._modified = this.getChangedIDs();
+} else {
+// Mark all items to be uploaded, but treat them as changed from long ago
+this._log.debug("First sync, uploading all items");
+this._modified = {};
+for (let id in this._store.getAllIDs()) {
+this._modified[id] = 0;
+}
+}
+// Clear the tracker now. If the sync fails we'll add the ones we failed
+// to upload back.
+this._tracker.clearChangedIDs();
+this._log.info(Object.keys(this._modified).length +
+" outgoing items pre-reconciliation");
+// Keep track of what to delete at the end of sync
+this._delete = {};
+},
+/**
+* A tiny abstraction to make it easier to test incoming record
+* application.
+*/
+_itemSource: function () {
+return new Collection(this.engineURL, this._recordObj, this.service);
+},
+/**
+* Process incoming records.
+* In the most awful and untestable way possible.
+* This now accepts something that makes testing vaguely less impossible.
+*/
+_processIncoming: function (newitems) {
+this._log.trace("Downloading & applying server changes");
+// Figure out how many total items to fetch this sync; do less on mobile.
+let batchSize = this.downloadLimit || Infinity;
+let isMobile = (Svc.Prefs.get("client.type") == "mobile");
+if (!newitems) {
+newitems = this._itemSource();
+}
+if (isMobile) {
+batchSize = MOBILE_BATCH_SIZE;
+}
+newitems.newer = this.lastSync;
+newitems.full  = true;
+newitems.limit = batchSize;
+// 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.
+let count = {applied: 0, failed: 0, newFailed: 0, reconciled: 0};
+let handled = [];
+let applyBatch = [];
+let failed = [];
+let failedInPreviousSync = this.previousFailed;
+let fetchBatch = Utils.arrayUnion(this.toFetch, failedInPreviousSync);
+// Reset previousFailed for each sync since previously failed items may not fail again.
+this.previousFailed = [];
+// Used (via exceptions) to allow the record handler/reconciliation/etc.
+// methods to signal that they would like processing of incoming records to
+// cease.
+let aborting = undefined;
+function doApplyBatch() {
+this._tracker.ignoreAll = true;
+try {
+failed = failed.concat(this._store.applyIncomingBatch(applyBatch));
+} catch (ex) {
+// Catch any error that escapes from applyIncomingBatch. At present
+// those will all be abort events.
+this._log.warn("Got exception " + Utils.exceptionStr(ex) +
+", aborting processIncoming.");
+aborting = ex;
+}
+this._tracker.ignoreAll = false;
+applyBatch = [];
+}
+function doApplyBatchAndPersistFailed() {
+// Apply remaining batch.
+if (applyBatch.length) {
+doApplyBatch.call(this);
+}
+// Persist failed items so we refetch them.
+if (failed.length) {
+this.previousFailed = Utils.arrayUnion(failed, this.previousFailed);
+count.failed += failed.length;
+this._log.debug("Records that failed to apply: " + failed);
+failed = [];
+}
+}
+let key = this.service.collectionKeys.keyForCollection(this.name);
+// Not binding this method to 'this' for performance reasons. It gets
+// called for every incoming record.
+let self = this;
+newitems.recordHandler = function(item) {
+if (aborting) {
+return;
+}
+// Grab a later last modified if possible
+if (self.lastModified == null || item.modified > self.lastModified)
+self.lastModified = item.modified;
+// Track the collection for the WBO.
+item.collection = self.name;
+// Remember which records were processed
+handled.push(item.id);
+try {
+try {
+item.decrypt(key);
+} catch (ex if Utils.isHMACMismatch(ex)) {
+let strategy = self.handleHMACMismatch(item, true);
+if (strategy == SyncEngine.kRecoveryStrategy.retry) {
+// You only get one retry.
+try {
+// Try decrypting again, typically because we've got new keys.
+self._log.info("Trying decrypt again...");
+key = self.service.collectionKeys.keyForCollection(self.name);
+item.decrypt(key);
+strategy = null;
+} catch (ex if Utils.isHMACMismatch(ex)) {
+strategy = self.handleHMACMismatch(item, false);
+}
+}
+switch (strategy) {
+case null:
+// Retry succeeded! No further handling.
+break;
+case SyncEngine.kRecoveryStrategy.retry:
+self._log.debug("Ignoring second retry suggestion.");
+// Fall through to error case.
+case SyncEngine.kRecoveryStrategy.error:
+self._log.warn("Error decrypting record: " + Utils.exceptionStr(ex));
+failed.push(item.id);
+return;
+case SyncEngine.kRecoveryStrategy.ignore:
+self._log.debug("Ignoring record " + item.id +
+" with bad HMAC: already handled.");
+return;
+}
+}
+} catch (ex) {
+self._log.warn("Error decrypting record: " + Utils.exceptionStr(ex));
+failed.push(item.id);
+return;
+}
+let shouldApply;
+try {
+shouldApply = self._reconcile(item);
+} catch (ex if (ex.code == Engine.prototype.eEngineAbortApplyIncoming)) {
+self._log.warn("Reconciliation failed: aborting incoming processing.");
+failed.push(item.id);
+aborting = ex.cause;
+} catch (ex) {
+self._log.warn("Failed to reconcile incoming record " + item.id);
+self._log.warn("Encountered exception: " + Utils.exceptionStr(ex));
+failed.push(item.id);
+return;
+}
+if (shouldApply) {
+count.applied++;
+applyBatch.push(item);
+} else {
+count.reconciled++;
+self._log.trace("Skipping reconciled incoming item " + item.id);
+}
+if (applyBatch.length == self.applyIncomingBatchSize) {
+doApplyBatch.call(self);
+}
+self._store._sleep(0);
+};
+// Only bother getting data from the server if there's new things
+if (this.lastModified == null || this.lastModified > this.lastSync) {
+let resp = newitems.get();
+doApplyBatchAndPersistFailed.call(this);
+if (!resp.success) {
+resp.failureCode = ENGINE_DOWNLOAD_FAIL;
+throw resp;
+}
+if (aborting) {
+throw aborting;
+}
+}
+// Mobile: check if we got the maximum that we requested; get the rest if so.
+if (handled.length == newitems.limit) {
+let guidColl = new Collection(this.engineURL, null, this.service);
+// Sort and limit so that on mobile we only get the last X records.
+guidColl.limit = this.downloadLimit;
+guidColl.newer = this.lastSync;
+// index: Orders by the sortindex descending (highest weight first).
+guidColl.sort  = "index";
+let guids = guidColl.get();
+if (!guids.success)
+throw guids;
+// Figure out which guids weren't just fetched then remove any guids that
+// were already waiting and prepend the new ones
+let extra = Utils.arraySub(guids.obj, handled);
+if (extra.length > 0) {
+fetchBatch = Utils.arrayUnion(extra, fetchBatch);
+this.toFetch = Utils.arrayUnion(extra, this.toFetch);
+}
+}
+// Fast-foward the lastSync timestamp since we have stored the
+// remaining items in toFetch.
+if (this.lastSync < this.lastModified) {
+this.lastSync = this.lastModified;
+}
+// Process any backlog of GUIDs.
+// At this point we impose an upper limit on the number of items to fetch
+// in a single request, even for desktop, to avoid hitting URI limits.
+batchSize = isMobile ? this.mobileGUIDFetchBatchSize :
+this.guidFetchBatchSize;
+while (fetchBatch.length && !aborting) {
+// Reuse the original query, but get rid of the restricting params
+// and batch remaining records.
+newitems.limit = 0;
+newitems.newer = 0;
+newitems.ids = fetchBatch.slice(0, batchSize);
+// Reuse the existing record handler set earlier
+let resp = newitems.get();
+if (!resp.success) {
+resp.failureCode = ENGINE_DOWNLOAD_FAIL;
+throw resp;
+}
+// This batch was successfully applied. Not using
+// doApplyBatchAndPersistFailed() here to avoid writing toFetch twice.
+fetchBatch = fetchBatch.slice(batchSize);
+this.toFetch = Utils.arraySub(this.toFetch, newitems.ids);
+this.previousFailed = Utils.arrayUnion(this.previousFailed, failed);
+if (failed.length) {
+count.failed += failed.length;
+this._log.debug("Records that failed to apply: " + failed);
+}
+failed = [];
+if (aborting) {
+throw aborting;
+}
+if (this.lastSync < this.lastModified) {
+this.lastSync = this.lastModified;
+}
+}
+// Apply remaining items.
+doApplyBatchAndPersistFailed.call(this);
+count.newFailed = Utils.arraySub(this.previousFailed, failedInPreviousSync).length;
+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);
+},
+/**
+* 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
+*/
+_findDupe: function (item) {
+// By default, assume there's no dupe items for the engine
+},
+_deleteId: function (id) {
+this._tracker.removeChangedID(id);
+// Remember this id to delete at the end of sync
+if (this._delete.ids == null)
+this._delete.ids = [id];
+else
+this._delete.ids.push(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.
+*/
+_reconcile: function (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   = this._store.itemExists(item.id);
+let locallyModified = item.id in this._modified;
+// TODO Handle clock drift better. Tracked in bug 721181.
+let remoteAge = AsyncResource.serverTime - item.modified;
+let localAge  = locallyModified ?
+(Date.now() / 1000 - this._modified[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;
+}
+// TODO As part of bug 720592, determine whether we should do more here.
+// In the case where the local changes are newer, it is quite possible
+// that the local client will restore data a remote client had tried to
+// delete. There might be a good reason for that delete and it might be
+// enexpected for this client to restore that data.
+this._log.trace("Incoming record is deleted but we had local changes. " +
+"Applying the youngest record.");
+return remoteIsNewer;
+}
+// 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 dupeID = this._findDupe(item);
+if (dupeID) {
+this._log.trace("Local item " + dupeID + " is a duplicate for " +
+"incoming item " + item.id);
+// The local, duplicate ID is always deleted on the server.
+this._deleteId(dupeID);
+// The current API contract does not mandate that the ID returned by
+// _findDupe() actually exists. Therefore, we have to perform this
+// check.
+existsLocally = this._store.itemExists(dupeID);
+// 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: " + dupeID + " -> " +
+item.id);
+this._store.changeItemID(dupeID, item.id);
+// If the local item was modified, we carry its metadata forward so
+// appropriate reconciling can be performed.
+if (dupeID in this._modified) {
+locallyModified = true;
+localAge = Date.now() / 1000 - this._modified[dupeID];
+remoteIsNewer = remoteAge < localAge;
+this._modified[item.id] = this._modified[dupeID];
+delete this._modified[dupeID];
+} else {
+locallyModified = false;
+localAge = null;
+}
+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.");
+delete this._modified[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 = 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.");
+delete this._modified[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);
+return remoteIsNewer;
+},
+// Upload outgoing records.
+_uploadOutgoing: function () {
+this._log.trace("Uploading local changes to server.");
+let modifiedIDs = Object.keys(this._modified);
+if (modifiedIDs.length) {
+this._log.trace("Preparing " + modifiedIDs.length +
+" outgoing records");
+// collection we'll upload
+let up = new Collection(this.engineURL, null, this.service);
+let count = 0;
+// Upload what we've got so far in the collection
+let doUpload = Utils.bind2(this, function(desc) {
+this._log.info("Uploading " + desc + " of " + modifiedIDs.length +
+" records");
+let resp = up.post();
+if (!resp.success) {
+this._log.debug("Uploading records failed: " + resp);
+resp.failureCode = ENGINE_UPLOAD_FAIL;
+throw resp;
+}
+// Update server timestamp from the upload.
+let modified = resp.headers["x-weave-timestamp"];
+if (modified > this.lastSync)
+this.lastSync = modified;
+let failed_ids = Object.keys(resp.obj.failed);
+if (failed_ids.length)
+this._log.debug("Records that will be uploaded again because "
++ "the server couldn't store them: "
++ failed_ids.join(", "));
+// Clear successfully uploaded objects.
+for each (let id in resp.obj.success) {
+delete this._modified[id];
+}
+up.clearRecords();
+});
+for each (let id in modifiedIDs) {
+try {
+let out = this._createRecord(id);
+if (this._log.level <= Log.Level.Trace)
+this._log.trace("Outgoing: " + out);
+out.encrypt(this.service.collectionKeys.keyForCollection(this.name));
+up.pushData(out);
+}
+catch(ex) {
+this._log.warn("Error creating record: " + Utils.exceptionStr(ex));
+}
+// Partial upload
+if ((++count % MAX_UPLOAD_RECORDS) == 0)
+doUpload((count - MAX_UPLOAD_RECORDS) + " - " + count + " out");
+this._store._sleep(0);
+}
+// Final upload
+if (count % MAX_UPLOAD_RECORDS > 0)
+doUpload(count >= MAX_UPLOAD_RECORDS ? "last batch" : "all");
+}
+},
+// Any cleanup necessary.
+// Save the current snapshot so as to calculate changes at next sync
+_syncFinish: function () {
+this._log.trace("Finishing up sync");
+this._tracker.resetScore();
+let doDelete = Utils.bind2(this, function(key, val) {
+let coll = new Collection(this.engineURL, this._recordObj, this.service);
+coll[key] = val;
+coll.delete();
+});
+for (let [key, val] in Iterator(this._delete)) {
+// Remove the key for future uses
+delete this._delete[key];
+// Send a simple delete for the property
+if (key != "ids" || val.length <= 100)
+doDelete(key, val);
+else {
+// For many ids, split into chunks of at most 100
+while (val.length > 0) {
+doDelete(key, val.slice(0, 100));
+val = val.slice(100);
+}
+}
+}
+},
+_syncCleanup: function () {
+if (!this._modified) {
+return;
+}
+// Mark failed WBOs as changed again so they are reuploaded next time.
+for (let [id, when] in Iterator(this._modified)) {
+this._tracker.addChangedID(id, when);
+}
+this._modified = {};
+},
+_sync: function () {
+try {
+this._syncStartup();
+Observers.notify("weave:engine:sync:status", "process-incoming");
+this._processIncoming();
+Observers.notify("weave:engine:sync:status", "upload-outgoing");
+this._uploadOutgoing();
+this._syncFinish();
+} finally {
+this._syncCleanup();
+}
+},
+canDecrypt: function () {
+// 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);
+test.recordHandler = function recordHandler(record) {
+record.decrypt(key);
+canDecrypt = true;
+}.bind(this);
+// Any failure fetching/decrypting will just result in false
+try {
+this._log.trace("Trying to decrypt a record from the server..");
+test.get();
+}
+catch(ex) {
+this._log.debug("Failed test decrypt: " + Utils.exceptionStr(ex));
+}
+return canDecrypt;
+},
+_resetClient: function () {
+this.resetLastSync();
+this.previousFailed = [];
+this.toFetch = [];
+},
+wipeServer: function () {
+let response = this.service.resource(this.engineURL).delete();
+if (response.status != 200 && response.status != 404) {
+throw response;
+}
+this._resetClient();
+},
+removeClientData: function () {
+// 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.
+*/
+handleHMACMismatch: function (item, mayRetry) {
+// By default we either try again, or bail out noisily.
+return (this.service.handleHMACEvent() && mayRetry) ?
+SyncEngine.kRecoveryStrategy.retry :
+SyncEngine.kRecoveryStrategy.error;
+}
+};