tor-browser

test_syncengine_sync.js (58976B)

---

/* Any copyright is dedicated to the Public Domain.
* http://creativecommons.org/publicdomain/zero/1.0/ */

const { Weave } = ChromeUtils.importESModule(
 "resource://services-sync/main.sys.mjs"
);
const { WBORecord } = ChromeUtils.importESModule(
 "resource://services-sync/record.sys.mjs"
);
const { Service } = ChromeUtils.importESModule(
 "resource://services-sync/service.sys.mjs"
);
const { RotaryEngine } = ChromeUtils.importESModule(
 "resource://testing-common/services/sync/rotaryengine.sys.mjs"
);

function makeRotaryEngine() {
 return new RotaryEngine(Service);
}

async function clean(engine) {
 for (const pref of Svc.PrefBranch.getChildList("")) {
   Svc.PrefBranch.clearUserPref(pref);
 }
 Svc.PrefBranch.setStringPref("log.logger.engine.rotary", "Trace");
 Service.recordManager.clearCache();
 await engine._tracker.clearChangedIDs();
 await engine.finalize();
}

async function cleanAndGo(engine, server) {
 await clean(engine);
 await promiseStopServer(server);
}

async function promiseClean(engine, server) {
 await clean(engine);
 await promiseStopServer(server);
}

async function createServerAndConfigureClient() {
 let engine = new RotaryEngine(Service);
 let syncID = await engine.resetLocalSyncID();

 let contents = {
   meta: {
     global: { engines: { rotary: { version: engine.version, syncID } } },
   },
   crypto: {},
   rotary: {},
 };

 const USER = "foo";
 let server = new SyncServer();
 server.registerUser(USER, "password");
 server.createContents(USER, contents);
 server.start();

 await SyncTestingInfrastructure(server, USER);
 Service._updateCachedURLs();

 return [engine, server, USER];
}

/*
* Tests
*
* SyncEngine._sync() is divided into four rather independent steps:
*
* - _syncStartup()
* - _processIncoming()
* - _uploadOutgoing()
* - _syncFinish()
*
* In the spirit of unit testing, these are tested individually for
* different scenarios below.
*/

add_task(async function setup() {
 await generateNewKeys(Service.collectionKeys);
 Svc.PrefBranch.setStringPref("log.logger.engine.rotary", "Trace");
});

add_task(async function test_syncStartup_emptyOrOutdatedGlobalsResetsSync() {
 _(
   "SyncEngine._syncStartup resets sync and wipes server data if there's no or an outdated global record"
 );

 // Some server side data that's going to be wiped
 let collection = new ServerCollection();
 collection.insert(
   "flying",
   encryptPayload({ id: "flying", denomination: "LNER Class A3 4472" })
 );
 collection.insert(
   "scotsman",
   encryptPayload({ id: "scotsman", denomination: "Flying Scotsman" })
 );

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 engine._store.items = { rekolok: "Rekonstruktionslokomotive" };
 try {
   // Confirm initial environment
   const changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.rekolok, undefined);
   let metaGlobal = await Service.recordManager.get(engine.metaURL);
   Assert.equal(metaGlobal.payload.engines, undefined);
   Assert.ok(!!collection.payload("flying"));
   Assert.ok(!!collection.payload("scotsman"));

   await engine.setLastSync(Date.now() / 1000);

   // Trying to prompt a wipe -- we no longer track CryptoMeta per engine,
   // so it has nothing to check.
   await engine._syncStartup();

   // The meta/global WBO has been filled with data about the engine
   let engineData = metaGlobal.payload.engines.rotary;
   Assert.equal(engineData.version, engine.version);
   Assert.equal(engineData.syncID, await engine.getSyncID());

   // Sync was reset and server data was wiped
   Assert.equal(await engine.getLastSync(), 0);
   Assert.equal(collection.payload("flying"), undefined);
   Assert.equal(collection.payload("scotsman"), undefined);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_syncStartup_serverHasNewerVersion() {
 _("SyncEngine._syncStartup ");

 let global = new ServerWBO("global", {
   engines: { rotary: { version: 23456 } },
 });
 let server = httpd_setup({
   "/1.1/foo/storage/meta/global": global.handler(),
 });

 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 try {
   // The server has a newer version of the data and our engine can
   // handle.  That should give us an exception.
   let error;
   try {
     await engine._syncStartup();
   } catch (ex) {
     error = ex;
   }
   Assert.equal(error.failureCode, VERSION_OUT_OF_DATE);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_syncStartup_syncIDMismatchResetsClient() {
 _("SyncEngine._syncStartup resets sync if syncIDs don't match");

 let server = sync_httpd_setup({});

 await SyncTestingInfrastructure(server);

 // global record with a different syncID than our engine has
 let engine = makeRotaryEngine();
 let global = new ServerWBO("global", {
   engines: { rotary: { version: engine.version, syncID: "foobar" } },
 });
 server.registerPathHandler("/1.1/foo/storage/meta/global", global.handler());

 try {
   // Confirm initial environment
   Assert.equal(await engine.getSyncID(), "");
   const changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.rekolok, undefined);

   await engine.setLastSync(Date.now() / 1000);
   await engine._syncStartup();

   // The engine has assumed the server's syncID
   Assert.equal(await engine.getSyncID(), "foobar");

   // Sync was reset
   Assert.equal(await engine.getLastSync(), 0);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_emptyServer() {
 _("SyncEngine._processIncoming working with an empty server backend");

 let collection = new ServerCollection();
 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 try {
   // Merely ensure that this code path is run without any errors
   await engine._processIncoming();
   Assert.equal(await engine.getLastSync(), 0);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_createFromServer() {
 _("SyncEngine._processIncoming creates new records from server data");

 // Some server records that will be downloaded
 let collection = new ServerCollection();
 collection.insert(
   "flying",
   encryptPayload({ id: "flying", denomination: "LNER Class A3 4472" })
 );
 collection.insert(
   "scotsman",
   encryptPayload({ id: "scotsman", denomination: "Flying Scotsman" })
 );

 // Two pathological cases involving relative URIs gone wrong.
 let pathologicalPayload = encryptPayload({
   id: "../pathological",
   denomination: "Pathological Case",
 });
 collection.insert("../pathological", pathologicalPayload);

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
   "/1.1/foo/storage/rotary/flying": collection.wbo("flying").handler(),
   "/1.1/foo/storage/rotary/scotsman": collection.wbo("scotsman").handler(),
 });

 await SyncTestingInfrastructure(server);

 await generateNewKeys(Service.collectionKeys);

 let engine = makeRotaryEngine();
 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 try {
   // Confirm initial environment
   Assert.equal(await engine.getLastSync(), 0);
   Assert.equal(engine.lastModified, null);
   Assert.equal(engine._store.items.flying, undefined);
   Assert.equal(engine._store.items.scotsman, undefined);
   Assert.equal(engine._store.items["../pathological"], undefined);

   await engine._syncStartup();
   await engine._processIncoming();

   // Timestamps of last sync and last server modification are set.
   Assert.greater(await engine.getLastSync(), 0);
   Assert.greater(engine.lastModified, 0);

   // Local records have been created from the server data.
   Assert.equal(engine._store.items.flying, "LNER Class A3 4472");
   Assert.equal(engine._store.items.scotsman, "Flying Scotsman");
   Assert.equal(engine._store.items["../pathological"], "Pathological Case");
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_reconcile() {
 _("SyncEngine._processIncoming updates local records");

 let collection = new ServerCollection();

 // This server record is newer than the corresponding client one,
 // so it'll update its data.
 collection.insert(
   "newrecord",
   encryptPayload({ id: "newrecord", denomination: "New stuff..." })
 );

 // This server record is newer than the corresponding client one,
 // so it'll update its data.
 collection.insert(
   "newerserver",
   encryptPayload({ id: "newerserver", denomination: "New data!" })
 );

 // This server record is 2 mins older than the client counterpart
 // but identical to it, so we're expecting the client record's
 // changedID to be reset.
 collection.insert(
   "olderidentical",
   encryptPayload({
     id: "olderidentical",
     denomination: "Older but identical",
   })
 );
 collection._wbos.olderidentical.modified -= 120;

 // This item simply has different data than the corresponding client
 // record (which is unmodified), so it will update the client as well
 collection.insert(
   "updateclient",
   encryptPayload({ id: "updateclient", denomination: "Get this!" })
 );

 // This is a dupe of 'original'.
 collection.insert(
   "duplication",
   encryptPayload({ id: "duplication", denomination: "Original Entry" })
 );

 // This record is marked as deleted, so we're expecting the client
 // record to be removed.
 collection.insert(
   "nukeme",
   encryptPayload({ id: "nukeme", denomination: "Nuke me!", deleted: true })
 );

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 engine._store.items = {
   newerserver: "New data, but not as new as server!",
   olderidentical: "Older but identical",
   updateclient: "Got data?",
   original: "Original Entry",
   long_original: "Long Original Entry",
   nukeme: "Nuke me!",
 };
 // Make this record 1 min old, thus older than the one on the server
 await engine._tracker.addChangedID("newerserver", Date.now() / 1000 - 60);
 // This record has been changed 2 mins later than the one on the server
 await engine._tracker.addChangedID("olderidentical", Date.now() / 1000);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 try {
   // Confirm initial environment
   Assert.equal(engine._store.items.newrecord, undefined);
   Assert.equal(
     engine._store.items.newerserver,
     "New data, but not as new as server!"
   );
   Assert.equal(engine._store.items.olderidentical, "Older but identical");
   Assert.equal(engine._store.items.updateclient, "Got data?");
   Assert.equal(engine._store.items.nukeme, "Nuke me!");
   let changes = await engine._tracker.getChangedIDs();
   Assert.greater(changes.olderidentical, 0);

   await engine._syncStartup();
   await engine._processIncoming();

   // Timestamps of last sync and last server modification are set.
   Assert.greater(await engine.getLastSync(), 0);
   Assert.greater(engine.lastModified, 0);

   // The new record is created.
   Assert.equal(engine._store.items.newrecord, "New stuff...");

   // The 'newerserver' record is updated since the server data is newer.
   Assert.equal(engine._store.items.newerserver, "New data!");

   // The data for 'olderidentical' is identical on the server, so
   // it's no longer marked as changed anymore.
   Assert.equal(engine._store.items.olderidentical, "Older but identical");
   changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.olderidentical, undefined);

   // Updated with server data.
   Assert.equal(engine._store.items.updateclient, "Get this!");

   // The incoming ID is preferred.
   Assert.equal(engine._store.items.original, undefined);
   Assert.equal(engine._store.items.duplication, "Original Entry");
   Assert.notEqual(engine._delete.ids.indexOf("original"), -1);

   // The 'nukeme' record marked as deleted is removed.
   Assert.equal(engine._store.items.nukeme, undefined);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_reconcile_local_deleted() {
 _("Ensure local, duplicate ID is deleted on server.");

 // When a duplicate is resolved, the local ID (which is never taken) should
 // be deleted on the server.
 let [engine, server, user] = await createServerAndConfigureClient();

 let now = Date.now() / 1000 - 10;
 await engine.setLastSync(now);
 engine.lastModified = now + 1;

 let record = encryptPayload({
   id: "DUPE_INCOMING",
   denomination: "incoming",
 });
 let wbo = new ServerWBO("DUPE_INCOMING", record, now + 2);
 server.insertWBO(user, "rotary", wbo);

 record = encryptPayload({ id: "DUPE_LOCAL", denomination: "local" });
 wbo = new ServerWBO("DUPE_LOCAL", record, now - 1);
 server.insertWBO(user, "rotary", wbo);

 await engine._store.create({ id: "DUPE_LOCAL", denomination: "local" });
 Assert.ok(await engine._store.itemExists("DUPE_LOCAL"));
 Assert.equal("DUPE_LOCAL", await engine._findDupe({ id: "DUPE_INCOMING" }));

 await engine._sync();

 do_check_attribute_count(engine._store.items, 1);
 Assert.ok("DUPE_INCOMING" in engine._store.items);

 let collection = server.getCollection(user, "rotary");
 Assert.equal(1, collection.count());
 Assert.notEqual(undefined, collection.wbo("DUPE_INCOMING"));

 await cleanAndGo(engine, server);
});

add_task(async function test_processIncoming_reconcile_equivalent() {
 _("Ensure proper handling of incoming records that match local.");

 let [engine, server, user] = await createServerAndConfigureClient();

 let now = Date.now() / 1000 - 10;
 await engine.setLastSync(now);
 engine.lastModified = now + 1;

 let record = encryptPayload({ id: "entry", denomination: "denomination" });
 let wbo = new ServerWBO("entry", record, now + 2);
 server.insertWBO(user, "rotary", wbo);

 engine._store.items = { entry: "denomination" };
 Assert.ok(await engine._store.itemExists("entry"));

 await engine._sync();

 do_check_attribute_count(engine._store.items, 1);

 await cleanAndGo(engine, server);
});

add_task(
 async function test_processIncoming_reconcile_locally_deleted_dupe_new() {
   _(
     "Ensure locally deleted duplicate record newer than incoming is handled."
   );

   // This is a somewhat complicated test. It ensures that if a client receives
   // a modified record for an item that is deleted locally but with a different
   // ID that the incoming record is ignored. This is a corner case for record
   // handling, but it needs to be supported.
   let [engine, server, user] = await createServerAndConfigureClient();

   let now = Date.now() / 1000 - 10;
   await engine.setLastSync(now);
   engine.lastModified = now + 1;

   let record = encryptPayload({
     id: "DUPE_INCOMING",
     denomination: "incoming",
   });
   let wbo = new ServerWBO("DUPE_INCOMING", record, now + 2);
   server.insertWBO(user, "rotary", wbo);

   // Simulate a locally-deleted item.
   engine._store.items = {};
   await engine._tracker.addChangedID("DUPE_LOCAL", now + 3);
   Assert.equal(false, await engine._store.itemExists("DUPE_LOCAL"));
   Assert.equal(false, await engine._store.itemExists("DUPE_INCOMING"));
   Assert.equal("DUPE_LOCAL", await engine._findDupe({ id: "DUPE_INCOMING" }));

   engine.lastModified = server.getCollection(user, engine.name).timestamp;
   await engine._sync();

   // After the sync, the server's payload for the original ID should be marked
   // as deleted.
   do_check_empty(engine._store.items);
   let collection = server.getCollection(user, "rotary");
   Assert.equal(1, collection.count());
   wbo = collection.wbo("DUPE_INCOMING");
   Assert.notEqual(null, wbo);
   let payload = wbo.getCleartext();
   Assert.ok(payload.deleted);

   await cleanAndGo(engine, server);
 }
);

add_task(
 async function test_processIncoming_reconcile_locally_deleted_dupe_old() {
   _(
     "Ensure locally deleted duplicate record older than incoming is restored."
   );

   // This is similar to the above test except it tests the condition where the
   // incoming record is newer than the local deletion, therefore overriding it.

   let [engine, server, user] = await createServerAndConfigureClient();

   let now = Date.now() / 1000 - 10;
   await engine.setLastSync(now);
   engine.lastModified = now + 1;

   let record = encryptPayload({
     id: "DUPE_INCOMING",
     denomination: "incoming",
   });
   let wbo = new ServerWBO("DUPE_INCOMING", record, now + 2);
   server.insertWBO(user, "rotary", wbo);

   // Simulate a locally-deleted item.
   engine._store.items = {};
   await engine._tracker.addChangedID("DUPE_LOCAL", now + 1);
   Assert.equal(false, await engine._store.itemExists("DUPE_LOCAL"));
   Assert.equal(false, await engine._store.itemExists("DUPE_INCOMING"));
   Assert.equal("DUPE_LOCAL", await engine._findDupe({ id: "DUPE_INCOMING" }));

   await engine._sync();

   // Since the remote change is newer, the incoming item should exist locally.
   do_check_attribute_count(engine._store.items, 1);
   Assert.ok("DUPE_INCOMING" in engine._store.items);
   Assert.equal("incoming", engine._store.items.DUPE_INCOMING);

   let collection = server.getCollection(user, "rotary");
   Assert.equal(1, collection.count());
   wbo = collection.wbo("DUPE_INCOMING");
   let payload = wbo.getCleartext();
   Assert.equal("incoming", payload.denomination);

   await cleanAndGo(engine, server);
 }
);

add_task(async function test_processIncoming_reconcile_changed_dupe() {
 _("Ensure that locally changed duplicate record is handled properly.");

 let [engine, server, user] = await createServerAndConfigureClient();

 let now = Date.now() / 1000 - 10;
 await engine.setLastSync(now);
 engine.lastModified = now + 1;

 // The local record is newer than the incoming one, so it should be retained.
 let record = encryptPayload({
   id: "DUPE_INCOMING",
   denomination: "incoming",
 });
 let wbo = new ServerWBO("DUPE_INCOMING", record, now + 2);
 server.insertWBO(user, "rotary", wbo);

 await engine._store.create({ id: "DUPE_LOCAL", denomination: "local" });
 await engine._tracker.addChangedID("DUPE_LOCAL", now + 3);
 Assert.ok(await engine._store.itemExists("DUPE_LOCAL"));
 Assert.equal("DUPE_LOCAL", await engine._findDupe({ id: "DUPE_INCOMING" }));

 engine.lastModified = server.getCollection(user, engine.name).timestamp;
 await engine._sync();

 // The ID should have been changed to incoming.
 do_check_attribute_count(engine._store.items, 1);
 Assert.ok("DUPE_INCOMING" in engine._store.items);

 // On the server, the local ID should be deleted and the incoming ID should
 // have its payload set to what was in the local record.
 let collection = server.getCollection(user, "rotary");
 Assert.equal(1, collection.count());
 wbo = collection.wbo("DUPE_INCOMING");
 Assert.notEqual(undefined, wbo);
 let payload = wbo.getCleartext();
 Assert.equal("local", payload.denomination);

 await cleanAndGo(engine, server);
});

add_task(async function test_processIncoming_reconcile_changed_dupe_new() {
 _("Ensure locally changed duplicate record older than incoming is ignored.");

 // This test is similar to the above except the incoming record is younger
 // than the local record. The incoming record should be authoritative.
 let [engine, server, user] = await createServerAndConfigureClient();

 let now = Date.now() / 1000 - 10;
 await engine.setLastSync(now);
 engine.lastModified = now + 1;

 let record = encryptPayload({
   id: "DUPE_INCOMING",
   denomination: "incoming",
 });
 let wbo = new ServerWBO("DUPE_INCOMING", record, now + 2);
 server.insertWBO(user, "rotary", wbo);

 await engine._store.create({ id: "DUPE_LOCAL", denomination: "local" });
 await engine._tracker.addChangedID("DUPE_LOCAL", now + 1);
 Assert.ok(await engine._store.itemExists("DUPE_LOCAL"));
 Assert.equal("DUPE_LOCAL", await engine._findDupe({ id: "DUPE_INCOMING" }));

 engine.lastModified = server.getCollection(user, engine.name).timestamp;
 await engine._sync();

 // The ID should have been changed to incoming.
 do_check_attribute_count(engine._store.items, 1);
 Assert.ok("DUPE_INCOMING" in engine._store.items);

 // On the server, the local ID should be deleted and the incoming ID should
 // have its payload retained.
 let collection = server.getCollection(user, "rotary");
 Assert.equal(1, collection.count());
 wbo = collection.wbo("DUPE_INCOMING");
 Assert.notEqual(undefined, wbo);
 let payload = wbo.getCleartext();
 Assert.equal("incoming", payload.denomination);
 await cleanAndGo(engine, server);
});

add_task(async function test_processIncoming_resume_toFetch() {
 _(
   "toFetch and previousFailed items left over from previous syncs are fetched on the next sync, along with new items."
 );

 const LASTSYNC = Date.now() / 1000;

 // Server records that will be downloaded
 let collection = new ServerCollection();
 collection.insert(
   "flying",
   encryptPayload({ id: "flying", denomination: "LNER Class A3 4472" })
 );
 collection.insert(
   "scotsman",
   encryptPayload({ id: "scotsman", denomination: "Flying Scotsman" })
 );
 collection.insert(
   "rekolok",
   encryptPayload({ id: "rekolok", denomination: "Rekonstruktionslokomotive" })
 );
 for (let i = 0; i < 3; i++) {
   let id = "failed" + i;
   let payload = encryptPayload({ id, denomination: "Record No. " + i });
   let wbo = new ServerWBO(id, payload);
   wbo.modified = LASTSYNC - 10;
   collection.insertWBO(wbo);
 }

 collection.wbo("flying").modified = collection.wbo("scotsman").modified =
   LASTSYNC - 10;
 collection._wbos.rekolok.modified = LASTSYNC + 10;

 // Time travel 10 seconds into the future but still download the above WBOs.
 let engine = makeRotaryEngine();
 await engine.setLastSync(LASTSYNC);
 engine.toFetch = new SerializableSet(["flying", "scotsman"]);
 engine.previousFailed = new SerializableSet([
   "failed0",
   "failed1",
   "failed2",
 ]);

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };
 try {
   // Confirm initial environment
   Assert.equal(engine._store.items.flying, undefined);
   Assert.equal(engine._store.items.scotsman, undefined);
   Assert.equal(engine._store.items.rekolok, undefined);

   await engine._syncStartup();
   await engine._processIncoming();

   // Local records have been created from the server data.
   Assert.equal(engine._store.items.flying, "LNER Class A3 4472");
   Assert.equal(engine._store.items.scotsman, "Flying Scotsman");
   Assert.equal(engine._store.items.rekolok, "Rekonstruktionslokomotive");
   Assert.equal(engine._store.items.failed0, "Record No. 0");
   Assert.equal(engine._store.items.failed1, "Record No. 1");
   Assert.equal(engine._store.items.failed2, "Record No. 2");
   Assert.equal(engine.previousFailed.size, 0);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_notify_count() {
 _("Ensure that failed records are reported only once.");

 const NUMBER_OF_RECORDS = 15;

 // Engine that fails every 5 records.
 let engine = makeRotaryEngine();
 engine._store._applyIncomingBatch = engine._store.applyIncomingBatch;
 engine._store.applyIncomingBatch = async function (records, countTelemetry) {
   let sortedRecords = records.sort((a, b) => (a.id > b.id ? 1 : -1));
   let recordsToApply = [],
     recordsToFail = [];
   for (let i = 0; i < sortedRecords.length; i++) {
     (i % 5 === 0 ? recordsToFail : recordsToApply).push(sortedRecords[i]);
   }
   recordsToFail.forEach(() => {
     countTelemetry.addIncomingFailedReason("failed message");
   });
   await engine._store._applyIncomingBatch(recordsToApply, countTelemetry);

   return recordsToFail.map(record => record.id);
 };

 // Create a batch of server side records.
 let collection = new ServerCollection();
 for (var i = 0; i < NUMBER_OF_RECORDS; i++) {
   let id = "record-no-" + i.toString(10).padStart(2, "0");
   let payload = encryptPayload({ id, denomination: "Record No. " + id });
   collection.insert(id, payload);
 }

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };
 try {
   // Confirm initial environment.
   Assert.equal(await engine.getLastSync(), 0);
   Assert.equal(engine.toFetch.size, 0);
   Assert.equal(engine.previousFailed.size, 0);
   do_check_empty(engine._store.items);

   let called = 0;
   let counts;
   function onApplied(count) {
     _("Called with " + JSON.stringify(counts));
     counts = count;
     called++;
   }
   Svc.Obs.add("weave:engine:sync:applied", onApplied);

   // Do sync.
   await engine._syncStartup();
   await engine._processIncoming();

   // Confirm failures.
   do_check_attribute_count(engine._store.items, 12);
   Assert.deepEqual(
     Array.from(engine.previousFailed).sort(),
     ["record-no-00", "record-no-05", "record-no-10"].sort()
   );

   // There are newly failed records and they are reported.
   Assert.equal(called, 1);
   Assert.equal(counts.failed, 3);
   Assert.equal(counts.failedReasons[0].count, 3);
   Assert.equal(counts.failedReasons[0].name, "failed message");
   Assert.equal(counts.applied, 15);
   Assert.equal(counts.newFailed, 3);
   Assert.equal(counts.succeeded, 12);

   // Sync again, 1 of the failed items are the same, the rest didn't fail.
   await engine._processIncoming();

   // Confirming removed failures.
   do_check_attribute_count(engine._store.items, 14);
   // After failing twice the record that failed again [record-no-00]
   // should NOT be stored to try again
   Assert.deepEqual(Array.from(engine.previousFailed), []);

   Assert.equal(called, 2);
   Assert.equal(counts.failed, 1);
   Assert.equal(counts.failedReasons[0].count, 1);
   Assert.equal(counts.failedReasons[0].name, "failed message");
   Assert.equal(counts.applied, 3);
   Assert.equal(counts.newFailed, 0);
   Assert.equal(counts.succeeded, 2);

   Svc.Obs.remove("weave:engine:sync:applied", onApplied);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_previousFailed() {
 _("Ensure that failed records are retried.");

 const NUMBER_OF_RECORDS = 14;

 // Engine that alternates between failing and applying every 2 records.
 let engine = makeRotaryEngine();
 engine._store._applyIncomingBatch = engine._store.applyIncomingBatch;
 engine._store.applyIncomingBatch = async function (records, countTelemetry) {
   let sortedRecords = records.sort((a, b) => (a.id > b.id ? 1 : -1));
   let recordsToApply = [],
     recordsToFail = [];
   let chunks = Array.from(PlacesUtils.chunkArray(sortedRecords, 2));
   for (let i = 0; i < chunks.length; i++) {
     (i % 2 === 0 ? recordsToFail : recordsToApply).push(...chunks[i]);
   }
   await engine._store._applyIncomingBatch(recordsToApply, countTelemetry);
   return recordsToFail.map(record => record.id);
 };

 // Create a batch of server side records.
 let collection = new ServerCollection();
 for (var i = 0; i < NUMBER_OF_RECORDS; i++) {
   let id = "record-no-" + i.toString(10).padStart(2, "0");
   let payload = encryptPayload({ id, denomination: "Record No. " + i });
   collection.insert(id, payload);
 }

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };
 try {
   // Confirm initial environment.
   Assert.equal(await engine.getLastSync(), 0);
   Assert.equal(engine.toFetch.size, 0);
   Assert.equal(engine.previousFailed.size, 0);
   do_check_empty(engine._store.items);

   // Initial failed items in previousFailed to be reset.
   let previousFailed = new SerializableSet([
     Utils.makeGUID(),
     Utils.makeGUID(),
     Utils.makeGUID(),
   ]);
   engine.previousFailed = previousFailed;
   Assert.equal(engine.previousFailed, previousFailed);

   // Do sync.
   await engine._syncStartup();
   await engine._processIncoming();

   // Expected result: 4 sync batches with 2 failures each => 8 failures
   do_check_attribute_count(engine._store.items, 6);
   Assert.deepEqual(
     Array.from(engine.previousFailed).sort(),
     [
       "record-no-00",
       "record-no-01",
       "record-no-04",
       "record-no-05",
       "record-no-08",
       "record-no-09",
       "record-no-12",
       "record-no-13",
     ].sort()
   );

   // Sync again with the same failed items (records 0, 1, 8, 9).
   await engine._processIncoming();

   do_check_attribute_count(engine._store.items, 10);
   // A second sync with the same failed items should NOT add the same items again.
   // Items that did not fail a second time should no longer be in previousFailed.
   Assert.deepEqual(Array.from(engine.previousFailed).sort(), []);

   // Refetched items that didn't fail the second time are in engine._store.items.
   Assert.equal(engine._store.items["record-no-04"], "Record No. 4");
   Assert.equal(engine._store.items["record-no-05"], "Record No. 5");
   Assert.equal(engine._store.items["record-no-12"], "Record No. 12");
   Assert.equal(engine._store.items["record-no-13"], "Record No. 13");
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_failed_records() {
 _(
   "Ensure that failed records from _reconcile and applyIncomingBatch are refetched."
 );

 // Let's create three and a bit batches worth of server side records.
 let APPLY_BATCH_SIZE = 50;
 let collection = new ServerCollection();
 const NUMBER_OF_RECORDS = APPLY_BATCH_SIZE * 3 + 5;
 for (let i = 0; i < NUMBER_OF_RECORDS; i++) {
   let id = "record-no-" + i;
   let payload = encryptPayload({ id, denomination: "Record No. " + id });
   let wbo = new ServerWBO(id, payload);
   wbo.modified = Date.now() / 1000 + 60 * (i - APPLY_BATCH_SIZE * 3);
   collection.insertWBO(wbo);
 }

 // Engine that batches but likes to throw on a couple of records,
 // two in each batch: the even ones fail in reconcile, the odd ones
 // in applyIncoming.
 const BOGUS_RECORDS = [
   "record-no-" + 42,
   "record-no-" + 23,
   "record-no-" + (42 + APPLY_BATCH_SIZE),
   "record-no-" + (23 + APPLY_BATCH_SIZE),
   "record-no-" + (42 + APPLY_BATCH_SIZE * 2),
   "record-no-" + (23 + APPLY_BATCH_SIZE * 2),
   "record-no-" + (2 + APPLY_BATCH_SIZE * 3),
   "record-no-" + (1 + APPLY_BATCH_SIZE * 3),
 ];
 let engine = makeRotaryEngine();

 engine.__reconcile = engine._reconcile;
 engine._reconcile = async function _reconcile(record) {
   if (BOGUS_RECORDS.indexOf(record.id) % 2 == 0) {
     throw new Error("I don't like this record! Baaaaaah!");
   }
   return this.__reconcile.apply(this, arguments);
 };
 engine._store._applyIncoming = engine._store.applyIncoming;
 engine._store.applyIncoming = async function (record) {
   if (BOGUS_RECORDS.indexOf(record.id) % 2 == 1) {
     throw new Error("I don't like this record! Baaaaaah!");
   }
   return this._applyIncoming.apply(this, arguments);
 };

 // Keep track of requests made of a collection.
 let count = 0;
 let uris = [];
 function recording_handler(recordedCollection) {
   let h = recordedCollection.handler();
   return function (req, res) {
     ++count;
     uris.push(req.path + "?" + req.queryString);
     return h(req, res);
   };
 }
 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": recording_handler(collection),
 });

 await SyncTestingInfrastructure(server);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 try {
   // Confirm initial environment
   Assert.equal(await engine.getLastSync(), 0);
   Assert.equal(engine.toFetch.size, 0);
   Assert.equal(engine.previousFailed.size, 0);
   do_check_empty(engine._store.items);

   let observerSubject;
   let observerData;
   Svc.Obs.add("weave:engine:sync:applied", function onApplied(subject, data) {
     Svc.Obs.remove("weave:engine:sync:applied", onApplied);
     observerSubject = subject;
     observerData = data;
   });

   await engine._syncStartup();
   await engine._processIncoming();

   // Ensure that all records but the bogus 4 have been applied.
   do_check_attribute_count(
     engine._store.items,
     NUMBER_OF_RECORDS - BOGUS_RECORDS.length
   );

   // Ensure that the bogus records will be fetched again on the next sync.
   Assert.equal(engine.previousFailed.size, BOGUS_RECORDS.length);
   Assert.deepEqual(
     Array.from(engine.previousFailed).sort(),
     BOGUS_RECORDS.sort()
   );

   // Ensure the observer was notified
   Assert.equal(observerData, engine.name);
   Assert.equal(observerSubject.failed, BOGUS_RECORDS.length);
   Assert.equal(observerSubject.newFailed, BOGUS_RECORDS.length);

   // Testing batching of failed item fetches.
   // Try to sync again. Ensure that we split the request into chunks to avoid
   // URI length limitations.
   async function batchDownload(batchSize) {
     count = 0;
     uris = [];
     engine.guidFetchBatchSize = batchSize;
     await engine._processIncoming();
     _("Tried again. Requests: " + count + "; URIs: " + JSON.stringify(uris));
     return count;
   }

   // There are 8 bad records, so this needs 3 fetches.
   _("Test batching with ID batch size 3, normal mobile batch size.");
   Assert.equal(await batchDownload(3), 3);

   // Since there the previous batch failed again, there should be
   // no more records to fetch
   _("Test that the second time a record failed to sync, gets ignored");
   Assert.equal(await batchDownload(BOGUS_RECORDS.length), 0);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_processIncoming_decrypt_failed() {
 _("Ensure that records failing to decrypt are either replaced or refetched.");

 // Some good and some bogus records. One doesn't contain valid JSON,
 // the other will throw during decrypt.
 let collection = new ServerCollection();
 collection._wbos.flying = new ServerWBO(
   "flying",
   encryptPayload({ id: "flying", denomination: "LNER Class A3 4472" })
 );
 collection._wbos.nojson = new ServerWBO("nojson", "This is invalid JSON");
 collection._wbos.nojson2 = new ServerWBO("nojson2", "This is invalid JSON");
 collection._wbos.scotsman = new ServerWBO(
   "scotsman",
   encryptPayload({ id: "scotsman", denomination: "Flying Scotsman" })
 );
 collection._wbos.nodecrypt = new ServerWBO("nodecrypt", "Decrypt this!");
 collection._wbos.nodecrypt2 = new ServerWBO("nodecrypt2", "Decrypt this!");

 // Patch the fake crypto service to throw on the record above.
 Weave.Crypto._decrypt = Weave.Crypto.decrypt;
 Weave.Crypto.decrypt = function (ciphertext) {
   if (ciphertext == "Decrypt this!") {
     throw new Error(
       "Derp! Cipher finalized failed. Im ur crypto destroyin ur recordz."
     );
   }
   return this._decrypt.apply(this, arguments);
 };

 // Some broken records also exist locally.
 let engine = makeRotaryEngine();
 engine.enabled = true;
 engine._store.items = { nojson: "Valid JSON", nodecrypt: "Valid ciphertext" };

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };
 try {
   // Confirm initial state
   Assert.equal(engine.toFetch.size, 0);
   Assert.equal(engine.previousFailed.size, 0);

   let observerSubject;
   let observerData;
   Svc.Obs.add("weave:engine:sync:applied", function onApplied(subject, data) {
     Svc.Obs.remove("weave:engine:sync:applied", onApplied);
     observerSubject = subject;
     observerData = data;
   });

   await engine.setLastSync(collection.wbo("nojson").modified - 1);
   let ping = await sync_engine_and_validate_telem(engine, true);
   Assert.equal(ping.engines[0].incoming.applied, 2);
   Assert.equal(ping.engines[0].incoming.failed, 4);
   console.log("incoming telem: ", ping.engines[0].incoming);
   Assert.equal(
     ping.engines[0].incoming.failedReasons[0].name,
     "No ciphertext: nothing to decrypt?"
   );
   // There should be 4 of the same error
   Assert.equal(ping.engines[0].incoming.failedReasons[0].count, 4);

   Assert.equal(engine.previousFailed.size, 4);
   Assert.ok(engine.previousFailed.has("nojson"));
   Assert.ok(engine.previousFailed.has("nojson2"));
   Assert.ok(engine.previousFailed.has("nodecrypt"));
   Assert.ok(engine.previousFailed.has("nodecrypt2"));

   // Ensure the observer was notified
   Assert.equal(observerData, engine.name);
   Assert.equal(observerSubject.applied, 2);
   Assert.equal(observerSubject.failed, 4);
   Assert.equal(observerSubject.failedReasons[0].count, 4);
 } finally {
   await promiseClean(engine, server);
 }
});

add_task(async function test_uploadOutgoing_toEmptyServer() {
 _("SyncEngine._uploadOutgoing uploads new records to server");

 let collection = new ServerCollection();
 collection._wbos.flying = new ServerWBO("flying");
 collection._wbos.scotsman = new ServerWBO("scotsman");

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
   "/1.1/foo/storage/rotary/flying": collection.wbo("flying").handler(),
   "/1.1/foo/storage/rotary/scotsman": collection.wbo("scotsman").handler(),
 });

 await SyncTestingInfrastructure(server);
 await generateNewKeys(Service.collectionKeys);

 let engine = makeRotaryEngine();
 engine._store.items = {
   flying: "LNER Class A3 4472",
   scotsman: "Flying Scotsman",
 };
 // Mark one of these records as changed
 await engine._tracker.addChangedID("scotsman", 0);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 try {
   await engine.setLastSync(123); // needs to be non-zero so that tracker is queried

   // Confirm initial environment
   Assert.equal(collection.payload("flying"), undefined);
   Assert.equal(collection.payload("scotsman"), undefined);

   await engine._syncStartup();
   await engine._uploadOutgoing();

   // Ensure the marked record ('scotsman') has been uploaded and is
   // no longer marked.
   Assert.equal(collection.payload("flying"), undefined);
   Assert.ok(!!collection.payload("scotsman"));
   Assert.equal(collection.cleartext("scotsman").id, "scotsman");
   const changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.scotsman, undefined);

   // The 'flying' record wasn't marked so it wasn't uploaded
   Assert.equal(collection.payload("flying"), undefined);
 } finally {
   await cleanAndGo(engine, server);
 }
});

async function test_uploadOutgoing_max_record_payload_bytes(
 allowSkippedRecord
) {
 _(
   "SyncEngine._uploadOutgoing throws when payload is bigger than max_record_payload_bytes"
 );
 let collection = new ServerCollection();
 collection._wbos.flying = new ServerWBO("flying");
 collection._wbos.scotsman = new ServerWBO("scotsman");

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
   "/1.1/foo/storage/rotary/flying": collection.wbo("flying").handler(),
   "/1.1/foo/storage/rotary/scotsman": collection.wbo("scotsman").handler(),
 });

 await SyncTestingInfrastructure(server);
 await generateNewKeys(Service.collectionKeys);

 let engine = makeRotaryEngine();
 engine.allowSkippedRecord = allowSkippedRecord;
 engine._store.items = { flying: "a".repeat(1024 * 1024), scotsman: "abcd" };

 await engine._tracker.addChangedID("flying", 1000);
 await engine._tracker.addChangedID("scotsman", 1000);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 try {
   await engine.setLastSync(1); // needs to be non-zero so that tracker is queried

   // Confirm initial environment
   Assert.equal(collection.payload("flying"), undefined);
   Assert.equal(collection.payload("scotsman"), undefined);

   await engine._syncStartup();
   await engine._uploadOutgoing();

   if (!allowSkippedRecord) {
     do_throw("should not get here");
   }

   await engine.trackRemainingChanges();

   // Check we uploaded the other record to the server
   Assert.ok(collection.payload("scotsman"));
   // And that we won't try to upload the huge record next time.
   const changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.flying, undefined);
 } catch (e) {
   if (allowSkippedRecord) {
     do_throw("should not get here");
   }

   await engine.trackRemainingChanges();

   // Check that we will try to upload the huge record next time
   const changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.flying, 1000);
 } finally {
   // Check we didn't upload the oversized record to the server
   Assert.equal(collection.payload("flying"), undefined);
   await cleanAndGo(engine, server);
 }
}

add_task(
 async function test_uploadOutgoing_max_record_payload_bytes_disallowSkippedRecords() {
   return test_uploadOutgoing_max_record_payload_bytes(false);
 }
);

add_task(
 async function test_uploadOutgoing_max_record_payload_bytes_allowSkippedRecords() {
   return test_uploadOutgoing_max_record_payload_bytes(true);
 }
);

add_task(async function test_uploadOutgoing_failed() {
 _(
   "SyncEngine._uploadOutgoing doesn't clear the tracker of objects that failed to upload."
 );

 let collection = new ServerCollection();
 // We only define the "flying" WBO on the server, not the "scotsman"
 // and "peppercorn" ones.
 collection._wbos.flying = new ServerWBO("flying");

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 engine._store.items = {
   flying: "LNER Class A3 4472",
   scotsman: "Flying Scotsman",
   peppercorn: "Peppercorn Class",
 };
 // Mark these records as changed
 const FLYING_CHANGED = 12345;
 const SCOTSMAN_CHANGED = 23456;
 const PEPPERCORN_CHANGED = 34567;
 await engine._tracker.addChangedID("flying", FLYING_CHANGED);
 await engine._tracker.addChangedID("scotsman", SCOTSMAN_CHANGED);
 await engine._tracker.addChangedID("peppercorn", PEPPERCORN_CHANGED);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 try {
   await engine.setLastSync(123); // needs to be non-zero so that tracker is queried

   // Confirm initial environment
   Assert.equal(collection.payload("flying"), undefined);
   let changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.flying, FLYING_CHANGED);
   Assert.equal(changes.scotsman, SCOTSMAN_CHANGED);
   Assert.equal(changes.peppercorn, PEPPERCORN_CHANGED);

   engine.enabled = true;
   await sync_engine_and_validate_telem(engine, true);

   // Ensure the 'flying' record has been uploaded and is no longer marked.
   Assert.ok(!!collection.payload("flying"));
   changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.flying, undefined);

   // The 'scotsman' and 'peppercorn' records couldn't be uploaded so
   // they weren't cleared from the tracker.
   Assert.equal(changes.scotsman, SCOTSMAN_CHANGED);
   Assert.equal(changes.peppercorn, PEPPERCORN_CHANGED);
 } finally {
   await promiseClean(engine, server);
 }
});

async function createRecordFailTelemetry(allowSkippedRecord) {
 Services.prefs.setStringPref("services.sync.username", "foo");
 let collection = new ServerCollection();
 collection._wbos.flying = new ServerWBO("flying");
 collection._wbos.scotsman = new ServerWBO("scotsman");

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 engine.allowSkippedRecord = allowSkippedRecord;
 let oldCreateRecord = engine._store.createRecord;
 engine._store.createRecord = async (id, col) => {
   if (id != "flying") {
     throw new Error("oops");
   }
   return oldCreateRecord.call(engine._store, id, col);
 };
 engine._store.items = {
   flying: "LNER Class A3 4472",
   scotsman: "Flying Scotsman",
 };
 // Mark these records as changed
 const FLYING_CHANGED = 12345;
 const SCOTSMAN_CHANGED = 23456;
 await engine._tracker.addChangedID("flying", FLYING_CHANGED);
 await engine._tracker.addChangedID("scotsman", SCOTSMAN_CHANGED);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 let ping;
 try {
   await engine.setLastSync(123); // needs to be non-zero so that tracker is queried

   // Confirm initial environment
   Assert.equal(collection.payload("flying"), undefined);
   let changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.flying, FLYING_CHANGED);
   Assert.equal(changes.scotsman, SCOTSMAN_CHANGED);

   engine.enabled = true;
   ping = await sync_engine_and_validate_telem(engine, true, onErrorPing => {
     ping = onErrorPing;
   });

   if (!allowSkippedRecord) {
     do_throw("should not get here");
   }

   // Ensure the 'flying' record has been uploaded and is no longer marked.
   Assert.ok(!!collection.payload("flying"));
   changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.flying, undefined);
 } catch (err) {
   if (allowSkippedRecord) {
     do_throw("should not get here");
   }

   // Ensure the 'flying' record has not been uploaded and is still marked
   Assert.ok(!collection.payload("flying"));
   const changes = await engine._tracker.getChangedIDs();
   Assert.ok(changes.flying);
 } finally {
   // We reported in telemetry that we failed a record
   Assert.equal(ping.engines[0].outgoing[0].failed, 1);
   Assert.equal(ping.engines[0].outgoing[0].failedReasons[0].name, "oops");

   // In any case, the 'scotsman' record couldn't be created so it wasn't
   // uploaded nor it was not cleared from the tracker.
   Assert.ok(!collection.payload("scotsman"));
   const changes = await engine._tracker.getChangedIDs();
   Assert.equal(changes.scotsman, SCOTSMAN_CHANGED);

   engine._store.createRecord = oldCreateRecord;
   await promiseClean(engine, server);
 }
}

add_task(
 async function test_uploadOutgoing_createRecord_throws_reported_telemetry() {
   _(
     "SyncEngine._uploadOutgoing reports a failed record to telemetry if createRecord throws"
   );
   await createRecordFailTelemetry(true);
 }
);

add_task(
 async function test_uploadOutgoing_createRecord_throws_dontAllowSkipRecord() {
   _(
     "SyncEngine._uploadOutgoing will throw if createRecord throws and allowSkipRecord is set to false"
   );
   await createRecordFailTelemetry(false);
 }
);

add_task(async function test_uploadOutgoing_largeRecords() {
 _(
   "SyncEngine._uploadOutgoing throws on records larger than the max record payload size"
 );

 let collection = new ServerCollection();

 let engine = makeRotaryEngine();
 engine.allowSkippedRecord = false;
 engine._store.items["large-item"] = "Y".repeat(
   Service.getMaxRecordPayloadSize() * 2
 );
 await engine._tracker.addChangedID("large-item", 0);
 collection.insert("large-item");

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 try {
   await engine._syncStartup();
   let error = null;
   try {
     await engine._uploadOutgoing();
   } catch (e) {
     error = e;
   }
   ok(!!error);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_syncFinish_deleteByIds() {
 _(
   "SyncEngine._syncFinish deletes server records slated for deletion (list of record IDs)."
 );

 let collection = new ServerCollection();
 collection._wbos.flying = new ServerWBO(
   "flying",
   encryptPayload({ id: "flying", denomination: "LNER Class A3 4472" })
 );
 collection._wbos.scotsman = new ServerWBO(
   "scotsman",
   encryptPayload({ id: "scotsman", denomination: "Flying Scotsman" })
 );
 collection._wbos.rekolok = new ServerWBO(
   "rekolok",
   encryptPayload({ id: "rekolok", denomination: "Rekonstruktionslokomotive" })
 );

 let server = httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });
 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 try {
   engine._delete = { ids: ["flying", "rekolok"] };
   await engine._syncFinish();

   // The 'flying' and 'rekolok' records were deleted while the
   // 'scotsman' one wasn't.
   Assert.equal(collection.payload("flying"), undefined);
   Assert.ok(!!collection.payload("scotsman"));
   Assert.equal(collection.payload("rekolok"), undefined);

   // The deletion todo list has been reset.
   Assert.equal(engine._delete.ids, undefined);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_syncFinish_deleteLotsInBatches() {
 _(
   "SyncEngine._syncFinish deletes server records in batches of 100 (list of record IDs)."
 );

 let collection = new ServerCollection();

 // Let's count how many times the client does a DELETE request to the server
 var noOfUploads = 0;
 collection.delete = (function (orig) {
   return function () {
     noOfUploads++;
     return orig.apply(this, arguments);
   };
 })(collection.delete);

 // Create a bunch of records on the server
 let now = Date.now();
 for (var i = 0; i < 234; i++) {
   let id = "record-no-" + i;
   let payload = encryptPayload({ id, denomination: "Record No. " + i });
   let wbo = new ServerWBO(id, payload);
   wbo.modified = now / 1000 - 60 * (i + 110);
   collection.insertWBO(wbo);
 }

 let server = httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let engine = makeRotaryEngine();
 try {
   // Confirm initial environment
   Assert.equal(noOfUploads, 0);

   // Declare what we want to have deleted: all records no. 100 and
   // up and all records that are less than 200 mins old (which are
   // records 0 thru 90).
   engine._delete = { ids: [], newer: now / 1000 - 60 * 200.5 };
   for (i = 100; i < 234; i++) {
     engine._delete.ids.push("record-no-" + i);
   }

   await engine._syncFinish();

   // Ensure that the appropriate server data has been wiped while
   // preserving records 90 thru 200.
   for (i = 0; i < 234; i++) {
     let id = "record-no-" + i;
     if (i <= 90 || i >= 100) {
       Assert.equal(collection.payload(id), undefined);
     } else {
       Assert.ok(!!collection.payload(id));
     }
   }

   // The deletion was done in batches
   Assert.equal(noOfUploads, 2 + 1);

   // The deletion todo list has been reset.
   Assert.equal(engine._delete.ids, undefined);
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_sync_partialUpload() {
 _("SyncEngine.sync() keeps changedIDs that couldn't be uploaded.");

 let collection = new ServerCollection();
 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });
 let oldServerConfiguration = Service.serverConfiguration;
 Service.serverConfiguration = {
   max_post_records: 100,
 };
 await SyncTestingInfrastructure(server);
 await generateNewKeys(Service.collectionKeys);

 let engine = makeRotaryEngine();

 // Let the third upload fail completely
 var noOfUploads = 0;
 collection.post = (function (orig) {
   return function () {
     if (noOfUploads == 2) {
       throw new Error("FAIL!");
     }
     noOfUploads++;
     return orig.apply(this, arguments);
   };
 })(collection.post);

 // Create a bunch of records (and server side handlers)
 for (let i = 0; i < 234; i++) {
   let id = "record-no-" + i;
   engine._store.items[id] = "Record No. " + i;
   await engine._tracker.addChangedID(id, i);
   // Let two items in the first upload batch fail.
   if (i != 23 && i != 42) {
     collection.insert(id);
   }
 }

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 try {
   await engine.setLastSync(123); // needs to be non-zero so that tracker is queried

   engine.enabled = true;
   let error;
   try {
     await sync_engine_and_validate_telem(engine, true);
   } catch (ex) {
     error = ex;
   }

   ok(!!error);

   const changes = await engine._tracker.getChangedIDs();
   for (let i = 0; i < 234; i++) {
     let id = "record-no-" + i;
     // Ensure failed records are back in the tracker:
     // * records no. 23 and 42 were rejected by the server,
     // * records after the third batch and higher couldn't be uploaded because
     //   we failed hard on the 3rd upload.
     if (i == 23 || i == 42 || i >= 200) {
       Assert.equal(changes[id], i);
     } else {
       Assert.equal(false, id in changes);
     }
   }
 } finally {
   Service.serverConfiguration = oldServerConfiguration;
   await promiseClean(engine, server);
 }
});

add_task(async function test_canDecrypt_noCryptoKeys() {
 _(
   "SyncEngine.canDecrypt returns false if the engine fails to decrypt items on the server, e.g. due to a missing crypto key collection."
 );

 // Wipe collection keys so we can test the desired scenario.
 Service.collectionKeys.clear();

 let collection = new ServerCollection();
 collection._wbos.flying = new ServerWBO(
   "flying",
   encryptPayload({ id: "flying", denomination: "LNER Class A3 4472" })
 );

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);
 let engine = makeRotaryEngine();
 try {
   Assert.equal(false, await engine.canDecrypt());
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_canDecrypt_true() {
 _(
   "SyncEngine.canDecrypt returns true if the engine can decrypt the items on the server."
 );

 await generateNewKeys(Service.collectionKeys);

 let collection = new ServerCollection();
 collection._wbos.flying = new ServerWBO(
   "flying",
   encryptPayload({ id: "flying", denomination: "LNER Class A3 4472" })
 );

 let server = sync_httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);
 let engine = makeRotaryEngine();
 try {
   Assert.ok(await engine.canDecrypt());
 } finally {
   await cleanAndGo(engine, server);
 }
});

add_task(async function test_syncapplied_observer() {
 const NUMBER_OF_RECORDS = 10;

 let engine = makeRotaryEngine();

 // Create a batch of server side records.
 let collection = new ServerCollection();
 for (var i = 0; i < NUMBER_OF_RECORDS; i++) {
   let id = "record-no-" + i;
   let payload = encryptPayload({ id, denomination: "Record No. " + id });
   collection.insert(id, payload);
 }

 let server = httpd_setup({
   "/1.1/foo/storage/rotary": collection.handler(),
 });

 await SyncTestingInfrastructure(server);

 let syncID = await engine.resetLocalSyncID();
 let meta_global = Service.recordManager.set(
   engine.metaURL,
   new WBORecord(engine.metaURL)
 );
 meta_global.payload.engines = { rotary: { version: engine.version, syncID } };

 let numApplyCalls = 0;
 let engine_name;
 let count;
 function onApplied(subject, data) {
   numApplyCalls++;
   engine_name = data;
   count = subject;
 }

 Svc.Obs.add("weave:engine:sync:applied", onApplied);

 try {
   Service.scheduler.hasIncomingItems = false;

   // Do sync.
   await engine._syncStartup();
   await engine._processIncoming();

   do_check_attribute_count(engine._store.items, 10);

   Assert.equal(numApplyCalls, 1);
   Assert.equal(engine_name, "rotary");
   Assert.equal(count.applied, 10);

   Assert.ok(Service.scheduler.hasIncomingItems);
 } finally {
   await cleanAndGo(engine, server);
   Service.scheduler.hasIncomingItems = false;
   Svc.Obs.remove("weave:engine:sync:applied", onApplied);
 }
});