diff options
Diffstat (limited to 'services/sync/modules/bridged_engine.sys.mjs')
| -rw-r--r-- | services/sync/modules/bridged_engine.sys.mjs | 499 |
1 files changed, 499 insertions, 0 deletions
diff --git a/services/sync/modules/bridged_engine.sys.mjs b/services/sync/modules/bridged_engine.sys.mjs new file mode 100644 index 0000000000..45e5f685cd --- /dev/null +++ b/services/sync/modules/bridged_engine.sys.mjs @@ -0,0 +1,499 @@ +/* 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 file has all the machinery for hooking up bridged engines implemented + * in Rust. It's the JavaScript side of the Golden Gate bridge that connects + * Desktop Sync to a Rust `BridgedEngine`, via the `mozIBridgedSyncEngine` + * XPCOM interface. + * + * Creating a bridged engine only takes a few lines of code, since most of the + * hard work is done on the Rust side. On the JS side, you'll need to subclass + * `BridgedEngine` (instead of `SyncEngine`), supply a `mozIBridgedSyncEngine` + * for your subclass to wrap, and optionally implement and override the tracker. + */ + +import { SyncEngine, Tracker } from "resource://services-sync/engines.sys.mjs"; +import { RawCryptoWrapper } from "resource://services-sync/record.sys.mjs"; + +const lazy = {}; + +ChromeUtils.defineESModuleGetters(lazy, { + Log: "resource://gre/modules/Log.sys.mjs", + PlacesUtils: "resource://gre/modules/PlacesUtils.sys.mjs", +}); + +/** + * A stub store that converts between raw decrypted incoming records and + * envelopes. Since the interface we need is so minimal, this class doesn't + * inherit from the base `Store` implementation...it would take more code to + * override all those behaviors! + * + * This class isn't meant to be subclassed, because bridged engines shouldn't + * override their store classes in `_storeObj`. + */ +class BridgedStore { + constructor(name, engine) { + if (!engine) { + throw new Error("Store must be associated with an Engine instance."); + } + this.engine = engine; + this._log = lazy.Log.repository.getLogger(`Sync.Engine.${name}.Store`); + this._batchChunkSize = 500; + } + + async applyIncomingBatch(records, countTelemetry) { + for (let chunk of lazy.PlacesUtils.chunkArray( + records, + this._batchChunkSize + )) { + let incomingEnvelopesAsJSON = chunk.map(record => + JSON.stringify(record.toIncomingBso()) + ); + this._log.trace("incoming envelopes", incomingEnvelopesAsJSON); + await this.engine._bridge.storeIncoming(incomingEnvelopesAsJSON); + } + // Array of failed records. + return []; + } + + async wipe() { + await this.engine._bridge.wipe(); + } +} + +/** + * A wrapper class to convert between BSOs on the JS side, and envelopes on the + * Rust side. This class intentionally subclasses `RawCryptoWrapper`, because we + * don't want the stringification and parsing machinery in `CryptoWrapper`. + * + * This class isn't meant to be subclassed, because bridged engines shouldn't + * override their record classes in `_recordObj`. + */ +class BridgedRecord extends RawCryptoWrapper { + /** + * Creates an outgoing record from a BSO returned by a bridged engine. + * + * @param {String} collection The collection name. + * @param {Object} bso The outgoing bso (ie, a sync15::bso::OutgoingBso) returned from + * `mozIBridgedSyncEngine::apply`. + * @return {BridgedRecord} A Sync record ready to encrypt and upload. + */ + static fromOutgoingBso(collection, bso) { + // The BSO has already been JSON serialized coming out of Rust, so the + // envelope has been flattened. + if (typeof bso.id != "string") { + throw new TypeError("Outgoing BSO missing ID"); + } + if (typeof bso.payload != "string") { + throw new TypeError("Outgoing BSO missing payload"); + } + let record = new BridgedRecord(collection, bso.id); + record.cleartext = bso.payload; + return record; + } + + transformBeforeEncrypt(cleartext) { + if (typeof cleartext != "string") { + throw new TypeError("Outgoing bridged engine records must be strings"); + } + return cleartext; + } + + transformAfterDecrypt(cleartext) { + if (typeof cleartext != "string") { + throw new TypeError("Incoming bridged engine records must be strings"); + } + return cleartext; + } + + /* + * Converts this incoming record into an envelope to pass to a bridged engine. + * This object must be kept in sync with `sync15::IncomingBso`. + * + * @return {Object} The incoming envelope, to pass to + * `mozIBridgedSyncEngine::storeIncoming`. + */ + toIncomingBso() { + return { + id: this.data.id, + modified: this.data.modified, + payload: this.cleartext, + }; + } +} + +class BridgeError extends Error { + constructor(code, message) { + super(message); + this.name = "BridgeError"; + // TODO: We may want to use a different name for this, since errors with + // a `result` property are treated specially by telemetry, discarding the + // message...but, unlike other `nserror`s, the message is actually useful, + // and we still want to capture it. + this.result = code; + } +} + +class InterruptedError extends Error { + constructor(message) { + super(message); + this.name = "InterruptedError"; + } +} + +/** + * Adapts a `Log.sys.mjs` logger to a `mozIServicesLogSink`. This class is copied + * from `SyncedBookmarksMirror.jsm`. + */ +export class LogAdapter { + constructor(log) { + this.log = log; + } + + get maxLevel() { + let level = this.log.level; + if (level <= lazy.Log.Level.All) { + return Ci.mozIServicesLogSink.LEVEL_TRACE; + } + if (level <= lazy.Log.Level.Info) { + return Ci.mozIServicesLogSink.LEVEL_DEBUG; + } + if (level <= lazy.Log.Level.Warn) { + return Ci.mozIServicesLogSink.LEVEL_WARN; + } + if (level <= lazy.Log.Level.Error) { + return Ci.mozIServicesLogSink.LEVEL_ERROR; + } + return Ci.mozIServicesLogSink.LEVEL_OFF; + } + + trace(message) { + this.log.trace(message); + } + + debug(message) { + this.log.debug(message); + } + + warn(message) { + this.log.warn(message); + } + + error(message) { + this.log.error(message); + } +} + +// This converts the XPCOM-defined, callback-based mozIBridgedSyncEngine to +// a promise-based implementation. +export class BridgeWrapperXPCOM { + constructor(component) { + this.comp = component; + } + + // A few sync, non-callback based attributes. + get storageVersion() { + return this.comp.storageVersion; + } + + get allowSkippedRecord() { + return this.comp.allowSkippedRecord; + } + + get logger() { + return this.comp.logger; + } + + // And the async functions we promisify. + // Note this is `lastSync` via uniffi but `getLastSync` via xpcom + lastSync() { + return BridgeWrapperXPCOM.#promisify(this.comp.getLastSync); + } + + setLastSync(lastSyncMillis) { + return BridgeWrapperXPCOM.#promisify(this.comp.setLastSync, lastSyncMillis); + } + + getSyncId() { + return BridgeWrapperXPCOM.#promisify(this.comp.getSyncId); + } + + resetSyncId() { + return BridgeWrapperXPCOM.#promisify(this.comp.resetSyncId); + } + + ensureCurrentSyncId(newSyncId) { + return BridgeWrapperXPCOM.#promisify( + this.comp.ensureCurrentSyncId, + newSyncId + ); + } + + syncStarted() { + return BridgeWrapperXPCOM.#promisify(this.comp.syncStarted); + } + + storeIncoming(incomingEnvelopesAsJSON) { + return BridgeWrapperXPCOM.#promisify( + this.comp.storeIncoming, + incomingEnvelopesAsJSON + ); + } + + apply() { + return BridgeWrapperXPCOM.#promisify(this.comp.apply); + } + + setUploaded(newTimestampMillis, uploadedIds) { + return BridgeWrapperXPCOM.#promisify( + this.comp.setUploaded, + newTimestampMillis, + uploadedIds + ); + } + + syncFinished() { + return BridgeWrapperXPCOM.#promisify(this.comp.syncFinished); + } + + reset() { + return BridgeWrapperXPCOM.#promisify(this.comp.reset); + } + + wipe() { + return BridgeWrapperXPCOM.#promisify(this.comp.wipe); + } + + // Converts a XPCOM bridged function that takes a callback into one that returns a + // promise. + static #promisify(func, ...params) { + return new Promise((resolve, reject) => { + func(...params, { + // This object implicitly implements all three callback interfaces + // (`mozIBridgedSyncEngine{Apply, Result}Callback`), because they have + // the same methods. The only difference is the type of the argument + // passed to `handleSuccess`, which doesn't matter in JS. + handleSuccess: resolve, + handleError(code, message) { + reject(transformError(code, message)); + }, + }); + }); + } +} + +/** + * A base class used to plug a Rust engine into Sync, and have it work like any + * other engine. The constructor takes a bridge as its first argument, which is + * a "bridged sync engine", as defined by UniFFI in the application-services + * crate. + * For backwards compatibility, this can also be an instance of an XPCOM + * component class that implements `mozIBridgedSyncEngine`, wrapped in + * a `BridgeWrapperXPCOM` wrapper. + * (Note that at time of writing, the above is slightly aspirational; the + * actual definition of the UniFFI shared bridged engine is still in flux.) + * + * This class inherits from `SyncEngine`, which has a lot of machinery that we + * don't need, but that's fairly easy to override. It would be harder to + * reimplement the machinery that we _do_ need here. However, because of that, + * this class has lots of methods that do nothing, or return empty data. The + * docs above each method explain what it's overriding, and why. + * + * This class is designed to be subclassed, but the only part that your engine + * may want to override is `_trackerObj`. Even then, using the default (no-op) + * tracker is fine, because the shape of the `Tracker` interface may not make + * sense for all engines. + */ +export function BridgedEngine(name, service) { + SyncEngine.call(this, name, service); +} + +BridgedEngine.prototype = { + /** + * The Rust implemented bridge. Must be set by the engine which subclasses us. + */ + _bridge: null, + /** + * The tracker class for this engine. Subclasses may want to override this + * with their own tracker, though using the default `Tracker` is fine. + */ + _trackerObj: Tracker, + + /** Returns the record class for all bridged engines. */ + get _recordObj() { + return BridgedRecord; + }, + + set _recordObj(obj) { + throw new TypeError("Don't override the record class for bridged engines"); + }, + + /** Returns the store class for all bridged engines. */ + get _storeObj() { + return BridgedStore; + }, + + set _storeObj(obj) { + throw new TypeError("Don't override the store class for bridged engines"); + }, + + /** Returns the storage version for this engine. */ + get version() { + return this._bridge.storageVersion; + }, + + // Legacy engines allow sync to proceed if some records are too large to + // upload (eg, a payload that's bigger than the server's published limits). + // If this returns true, we will just skip the record without even attempting + // to upload. If this is false, we'll abort the entire batch. + // If the engine allows this, it will need to detect this scenario by noticing + // the ID is not in the 'success' records reported to `setUploaded`. + // (Note that this is not to be confused with the fact server's can currently + // reject records as part of a POST - but we hope to remove this ability from + // the server API. Note also that this is not bullet-proof - if the count of + // records is high, it's possible that we will have committed a previous + // batch before we hit the relevant limits, so things might have been written. + // We hope to fix this by ensuring batch limits are such that this is + // impossible) + get allowSkippedRecord() { + return this._bridge.allowSkippedRecord; + }, + + /** + * Returns the sync ID for this engine. This is exposed for tests, but + * Sync code always calls `resetSyncID()` and `ensureCurrentSyncID()`, + * not this. + * + * @returns {String?} The sync ID, or `null` if one isn't set. + */ + async getSyncID() { + // Note that all methods on an XPCOM class instance are automatically bound, + // so we don't need to write `this._bridge.getSyncId.bind(this._bridge)`. + let syncID = await this._bridge.getSyncId(); + return syncID; + }, + + async resetSyncID() { + await this._deleteServerCollection(); + let newSyncID = await this.resetLocalSyncID(); + return newSyncID; + }, + + async resetLocalSyncID() { + let newSyncID = await this._bridge.resetSyncId(); + return newSyncID; + }, + + async ensureCurrentSyncID(newSyncID) { + let assignedSyncID = await this._bridge.ensureCurrentSyncId(newSyncID); + return assignedSyncID; + }, + + async getLastSync() { + // The bridge defines lastSync as integer ms, but sync itself wants to work + // in a float seconds with 2 decimal places. + let lastSyncMS = await this._bridge.lastSync(); + return Math.round(lastSyncMS / 10) / 100; + }, + + async setLastSync(lastSyncSeconds) { + await this._bridge.setLastSync(Math.round(lastSyncSeconds * 1000)); + }, + + /** + * Returns the initial changeset for the sync. Bridged engines handle + * reconciliation internally, so we don't know what changed until after we've + * stored and applied all incoming records. So we return an empty changeset + * here, and replace it with the real one in `_processIncoming`. + */ + async pullChanges() { + return {}; + }, + + async trackRemainingChanges() { + await this._bridge.syncFinished(); + }, + + /** + * Marks a record for a hard-`DELETE` at the end of the sync. The base method + * also removes it from the tracker, but we don't use the tracker for that, + * so we override the method to just mark. + */ + _deleteId(id) { + this._noteDeletedId(id); + }, + + /** + * Always stage incoming records, bypassing the base engine's reconciliation + * machinery. + */ + async _reconcile() { + return true; + }, + + async _syncStartup() { + await super._syncStartup(); + await this._bridge.syncStarted(); + }, + + async _processIncoming(newitems) { + await super._processIncoming(newitems); + + let outgoingBsosAsJSON = await this._bridge.apply(); + let changeset = {}; + for (let bsoAsJSON of outgoingBsosAsJSON) { + this._log.trace("outgoing bso", bsoAsJSON); + let record = BridgedRecord.fromOutgoingBso( + this.name, + JSON.parse(bsoAsJSON) + ); + changeset[record.id] = { + synced: false, + record, + }; + } + this._modified.replace(changeset); + }, + + /** + * Notify the bridged engine that we've successfully uploaded a batch, so + * that it can update its local state. For example, if the engine uses a + * mirror and a temp table for outgoing records, it can write the uploaded + * records from the outgoing table back to the mirror. + */ + async _onRecordsWritten(succeeded, failed, serverModifiedTime) { + // JS uses seconds but Rust uses milliseconds so we'll need to convert + let serverModifiedMS = Math.round(serverModifiedTime * 1000); + await this._bridge.setUploaded(Math.floor(serverModifiedMS), succeeded); + }, + + async _createTombstone() { + throw new Error("Bridged engines don't support weak uploads"); + }, + + async _createRecord(id) { + let change = this._modified.changes[id]; + if (!change) { + throw new TypeError("Can't create record for unchanged item"); + } + return change.record; + }, + + async _resetClient() { + await super._resetClient(); + await this._bridge.reset(); + }, +}; +Object.setPrototypeOf(BridgedEngine.prototype, SyncEngine.prototype); + +function transformError(code, message) { + switch (code) { + case Cr.NS_ERROR_ABORT: + return new InterruptedError(message); + + default: + return new BridgeError(code, message); + } +} |