The Tor Browser / file revision

 /* 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;

   }

 };