6 files changed, 805 insertions, 0 deletions
diff --git a/services/sync/golden_gate/Cargo.toml b/services/sync/golden_gate/Cargo.toml
new file mode 100644
index 0000000000..3f94e1a1e9
--- /dev/null
+++ b/services/sync/golden_gate/Cargo.toml
@@ -0,0 +1,25 @@
+[package]
+name = "golden_gate"
+description = "A bridge for wiring up Sync engines implemented in Rust"
+version = "0.1.0"
+authors = ["The Firefox Sync Developers <sync-team@mozilla.com>"]
+edition = "2018"
+license = "MPL-2.0"
+
+[dependencies]
+anyhow = "1"
+atomic_refcell = "0.1"
+cstr = "0.2"
+interrupt-support = "0.1"
+log = "0.4"
+moz_task = { path = "../../../xpcom/rust/moz_task" }
+nserror = { path = "../../../xpcom/rust/nserror" }
+nsstring = { path = "../../../xpcom/rust/nsstring" }
+serde_json = "1"
+storage_variant = { path = "../../../storage/variant" }
+sync15 = "0.1"
+xpcom = { path = "../../../xpcom/rust/xpcom" }
+
+[dependencies.thin-vec]
+version = "0.2.1"
+features = ["gecko-ffi"]
diff --git a/services/sync/golden_gate/src/error.rs b/services/sync/golden_gate/src/error.rs
new file mode 100644
index 0000000000..373d20756e
--- /dev/null
+++ b/services/sync/golden_gate/src/error.rs
@@ -0,0 +1,71 @@
+/* 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/. */
+
+use std::{error, fmt, result, str::Utf8Error};
+
+use nserror::{nsresult, NS_ERROR_INVALID_ARG, NS_ERROR_UNEXPECTED};
+use serde_json::Error as JsonError;
+
+/// A specialized `Result` type for Golden Gate.
+pub type Result<T> = result::Result<T, Error>;
+
+/// The error type for Golden Gate errors.
+#[derive(Debug)]
+pub enum Error {
+    /// A wrapped XPCOM error.
+    Nsresult(nsresult),
+
+    /// A ferry didn't run on the background task queue.
+    DidNotRun(&'static str),
+
+    /// A string contains invalid UTF-8 or JSON.
+    MalformedString(Box<dyn error::Error + Send + Sync + 'static>),
+}
+
+impl error::Error for Error {
+    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
+        match self {
+            Error::MalformedString(error) => Some(error.as_ref()),
+            _ => None,
+        }
+    }
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Error::Nsresult(result) => write!(f, "Operation failed with {}", result.error_name()),
+            Error::DidNotRun(what) => write!(f, "Failed to run `{what}` on background thread"),
+            Error::MalformedString(error) => error.fmt(f),
+        }
+    }
+}
+
+impl From<nsresult> for Error {
+    fn from(result: nsresult) -> Error {
+        Error::Nsresult(result)
+    }
+}
+
+impl From<Utf8Error> for Error {
+    fn from(error: Utf8Error) -> Error {
+        Error::MalformedString(error.into())
+    }
+}
+
+impl From<JsonError> for Error {
+    fn from(error: JsonError) -> Error {
+        Error::MalformedString(error.into())
+    }
+}
+
+impl From<Error> for nsresult {
+    fn from(error: Error) -> nsresult {
+        match error {
+            Error::DidNotRun(_) => NS_ERROR_UNEXPECTED,
+            Error::Nsresult(result) => result,
+            Error::MalformedString(_) => NS_ERROR_INVALID_ARG,
+        }
+    }
+}
diff --git a/services/sync/golden_gate/src/ferry.rs b/services/sync/golden_gate/src/ferry.rs
new file mode 100644
index 0000000000..99994811ab
--- /dev/null
+++ b/services/sync/golden_gate/src/ferry.rs
@@ -0,0 +1,74 @@
+/* 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/. */
+
+use nsstring::nsCString;
+use storage_variant::VariantType;
+use sync15::Guid;
+use xpcom::{interfaces::nsIVariant, RefPtr};
+
+/// An operation that runs on the background thread, and optionally passes a
+/// result to its callback.
+pub enum Ferry {
+    LastSync,
+    SetLastSync(i64),
+    SyncId,
+    ResetSyncId,
+    EnsureCurrentSyncId(String),
+    SyncStarted,
+    StoreIncoming(Vec<nsCString>),
+    SetUploaded(i64, Vec<Guid>),
+    SyncFinished,
+    Reset,
+    Wipe,
+}
+
+impl Ferry {
+    /// Returns the operation name for debugging and labeling the task
+    /// runnable.
+    pub fn name(&self) -> &'static str {
+        match self {
+            Ferry::LastSync => concat!(module_path!(), "getLastSync"),
+            Ferry::SetLastSync(_) => concat!(module_path!(), "setLastSync"),
+            Ferry::SyncId => concat!(module_path!(), "getSyncId"),
+            Ferry::ResetSyncId => concat!(module_path!(), "resetSyncId"),
+            Ferry::EnsureCurrentSyncId(_) => concat!(module_path!(), "ensureCurrentSyncId"),
+            Ferry::SyncStarted => concat!(module_path!(), "syncStarted"),
+            Ferry::StoreIncoming { .. } => concat!(module_path!(), "storeIncoming"),
+            Ferry::SetUploaded { .. } => concat!(module_path!(), "setUploaded"),
+            Ferry::SyncFinished => concat!(module_path!(), "syncFinished"),
+            Ferry::Reset => concat!(module_path!(), "reset"),
+            Ferry::Wipe => concat!(module_path!(), "wipe"),
+        }
+    }
+}
+
+/// The result of a ferry task, sent from the background thread back to the
+/// main thread. Results are converted to variants, and passed as arguments to
+/// `mozIBridgedSyncEngineCallback`s.
+pub enum FerryResult {
+    LastSync(i64),
+    SyncId(Option<String>),
+    AssignedSyncId(String),
+    Null,
+}
+
+impl Default for FerryResult {
+    fn default() -> Self {
+        FerryResult::Null
+    }
+}
+
+impl FerryResult {
+    /// Converts the result to an `nsIVariant` that can be passed as an
+    /// argument to `callback.handleResult()`.
+    pub fn into_variant(self) -> RefPtr<nsIVariant> {
+        match self {
+            FerryResult::LastSync(v) => v.into_variant(),
+            FerryResult::SyncId(Some(v)) => nsCString::from(v).into_variant(),
+            FerryResult::SyncId(None) => ().into_variant(),
+            FerryResult::AssignedSyncId(v) => nsCString::from(v).into_variant(),
+            FerryResult::Null => ().into_variant(),
+        }
+    }
+}
diff --git a/services/sync/golden_gate/src/lib.rs b/services/sync/golden_gate/src/lib.rs
new file mode 100644
index 0000000000..8da6524bd7
--- /dev/null
+++ b/services/sync/golden_gate/src/lib.rs
@@ -0,0 +1,119 @@
+/* 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/. */
+
+//! **Golden Gate** 🌉 is a crate for bridging Desktop Sync to our suite of
+//! Rust sync and storage components. It connects Sync's `BridgedEngine` class
+//! to the Rust `BridgedEngine` trait via the `mozIBridgedSyncEngine` XPCOM
+//! interface.
+//!
+//! Due to limitations in implementing XPCOM interfaces for generic types,
+//! Golden Gate doesn't implement `mozIBridgedSyncEngine` directly. Instead,
+//! it provides helpers, called "ferries", for passing Sync records between
+//! JavaScript and Rust. The ferries also handle threading and type
+//! conversions.
+//!
+//! Here's a step-by-step guide for adding a new Rust Sync engine to Firefox.
+//!
+//! ## Step 1: Create your (XPCOM) bridge
+//!
+//! In your consuming crate, define a type for your `mozIBridgedSyncEngine`
+//! implementation. We'll call this type the **brige**. The bridge is
+//! responsible for exposing your Sync engine to XPIDL [^1], in a way that lets
+//! JavaScript call it.
+//!
+//! For your bridge type, you'll need to implement an xpcom interface with the
+//! `#[xpcom(implement(mozIBridgedSyncEngine), nonatomic)]` attribute then
+//! define `xpcom_method!()` stubs for the `mozIBridgedSyncEngine` methods.  For
+//! more details about implementing XPCOM methods in Rust, check out the docs in
+//! `xpcom/rust/xpcom/src/method.rs`.
+//!
+//! You'll also need to add an entry for your bridge type to `components.conf`,
+//! and define C++ and Rust constructors for it, so that JavaScript code can
+//! create instances of it. Check out `NS_NewWebExtStorage` (and, in C++,
+//! `mozilla::extensions::storageapi::NewWebExtStorage`) and
+//! `NS_NewSyncedBookmarksMerger` (`mozilla::places::NewSyncedBookmarksMerger`
+//! in C++) for how to do this.
+//!
+//! [^1]: You can think of XPIDL as a souped-up C FFI, with richer types and a
+//! degree of type safety.
+//!
+//! ## Step 2: Add a background task queue to your bridge
+//!
+//! A task queue lets your engine do I/O, merging, and other syncing tasks on a
+//! background thread pool. This is important because database reads and writes
+//! can take an unpredictable amount of time. Doing these on the main thread can
+//! cause jank, and, in the worst case, lock up the browser UI for seconds at a
+//! time.
+//!
+//! The `moz_task` crate provides a `create_background_task_queue` function to
+//! do this. Once you have a queue, you can use it to call into your Rust
+//! engine. Golden Gate takes care of ferrying arguments back and forth across
+//! the thread boundary.
+//!
+//! Since it's a queue, ferries arrive in the order they're scheduled, so
+//! your engine's `store_incoming` method will always be called before `apply`,
+//! which is likewise called before `set_uploaded`. The thread manager scales
+//! the pool for you; you don't need to create or manage your own threads.
+//!
+//! ## Step 3: Create your Rust engine
+//!
+//! Next, you'll need to implement the Rust side of the bridge. This is a type
+//! that implements the `BridgedEngine` trait.
+//!
+//! Bridged engines handle storing incoming Sync records, merging changes,
+//! resolving conflicts, and fetching outgoing records for upload. Under the
+//! hood, your engine will hold either a database connection directly, or
+//! another object that does.
+//!
+//! Although outside the scope of Golden Gate, your engine will also likely
+//! expose a data storage API, for fetching, updating, and deleting items
+//! locally. Golden Gate provides the syncing layer on top of this local store.
+//!
+//! A `BridgedEngine` itself doesn't need to be `Send` or `Sync`, but the
+//! ferries require both, since they're calling into your bridge on the
+//! background task queue.
+//!
+//! In practice, this means your bridge will need to hold a thread-safe owned
+//! reference to the engine, via `Arc<Mutex<BridgedEngine>>`. In fact, this
+//! pattern is so common that Golden Gate implements `BridgedEngine` for any
+//! `Mutex<BridgedEngine>`, which automatically locks the mutex before calling
+//! into the engine.
+//!
+//! ## Step 4: Connect the bridge to the JavaScript and Rust sides
+//!
+//! On the JavaScript side, you'll need to subclass Sync's `BridgedEngine`
+//! class, and give it a handle to your XPCOM bridge. The base class has all the
+//! machinery for hooking up any `mozIBridgedSyncEngine` implementation so that
+//! Sync can drive it.
+//!
+//! On the Rust side, each `mozIBridgedSyncEngine` method should create a
+//! Golden Gate ferry, and dispatch it to the background task queue. The
+//! ferries correspond to the method names. For example, `ensureCurrentSyncId`
+//! should create a `Ferry::ensure_current_sync_id(...)`; `storeIncoming`, a
+//! `Ferry::store_incoming(...)`; and so on. This is mostly boilerplate.
+//!
+//! And that's it! Each ferry will, in turn, call into your Rust
+//! `BridgedEngine`, and send the results back to JavaScript.
+//!
+//! For an example of how all this works, including exposing a storage (not
+//! just syncing!) API to JS via XPIDL, check out `webext_storage::Bridge` for
+//! the `storage.sync` API!
+
+#[macro_use]
+extern crate cstr;
+
+pub mod error;
+mod ferry;
+pub mod log;
+pub mod task;
+
+pub use crate::log::LogSink;
+pub use error::{Error, Result};
+// Re-export items from `interrupt-support` and `sync15`, so that
+// consumers of `golden_gate` don't have to depend on them.
+pub use interrupt_support::{Interrupted, Interruptee};
+pub use sync15::bso::{IncomingBso, OutgoingBso};
+pub use sync15::engine::{ApplyResults, BridgedEngine};
+pub use sync15::Guid;
+pub use task::{ApplyTask, FerryTask};
diff --git a/services/sync/golden_gate/src/log.rs b/services/sync/golden_gate/src/log.rs
new file mode 100644
index 0000000000..de7fd0dfc3
--- /dev/null
+++ b/services/sync/golden_gate/src/log.rs
@@ -0,0 +1,161 @@
+/* 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/. */
+
+use std::fmt::{self, Write};
+
+use log::{Level, LevelFilter, Log, Metadata, Record};
+use moz_task::{Task, TaskRunnable, ThreadPtrHandle, ThreadPtrHolder};
+use nserror::nsresult;
+use nsstring::nsString;
+use xpcom::{interfaces::mozIServicesLogSink, RefPtr};
+
+pub struct LogSink {
+    pub max_level: LevelFilter,
+    logger: Option<ThreadPtrHandle<mozIServicesLogSink>>,
+}
+
+impl Default for LogSink {
+    fn default() -> Self {
+        LogSink {
+            max_level: LevelFilter::Off,
+            logger: None,
+        }
+    }
+}
+
+impl LogSink {
+    /// Creates a log sink that adapts the Rust `log` crate to the Sync
+    /// `Log.sys.mjs` logger.
+    ///
+    /// This is copied from `bookmark_sync::Logger`. It would be nice to share
+    /// these, but, for now, we've just duplicated it to make prototyping
+    /// easier.
+    #[inline]
+    pub fn new(max_level: LevelFilter, logger: ThreadPtrHandle<mozIServicesLogSink>) -> LogSink {
+        LogSink {
+            max_level,
+            logger: Some(logger),
+        }
+    }
+
+    /// Creates a log sink using the given Services `logger` as the
+    /// underlying implementation. The `logger` will always be called
+    /// asynchronously on its owning thread; it doesn't need to be
+    /// thread-safe.
+    pub fn with_logger(logger: Option<&mozIServicesLogSink>) -> Result<LogSink, nsresult> {
+        Ok(if let Some(logger) = logger {
+            // Fetch the maximum log level while we're on the main thread, so
+            // that `LogSink::enabled()` can check it while on the background
+            // thread. Otherwise, we'd need to dispatch a `LogTask` for every
+            // log message, only to discard most of them when the task calls
+            // into the logger on the main thread.
+            let mut raw_max_level = 0i16;
+            let rv = unsafe { logger.GetMaxLevel(&mut raw_max_level) };
+            let max_level = if rv.succeeded() {
+                match raw_max_level {
+                    mozIServicesLogSink::LEVEL_ERROR => LevelFilter::Error,
+                    mozIServicesLogSink::LEVEL_WARN => LevelFilter::Warn,
+                    mozIServicesLogSink::LEVEL_DEBUG => LevelFilter::Debug,
+                    mozIServicesLogSink::LEVEL_TRACE => LevelFilter::Trace,
+                    mozIServicesLogSink::LEVEL_INFO => LevelFilter::Info,
+                    _ => LevelFilter::Off,
+                }
+            } else {
+                LevelFilter::Off
+            };
+            LogSink::new(
+                max_level,
+                ThreadPtrHolder::new(cstr!("mozIServicesLogSink"), RefPtr::new(logger))?,
+            )
+        } else {
+            LogSink::default()
+        })
+    }
+
+    /// Returns a reference to the underlying `mozIServicesLogSink`.
+    pub fn logger(&self) -> Option<&mozIServicesLogSink> {
+        self.logger.as_ref().and_then(|l| l.get())
+    }
+
+    /// Logs a message to the Sync logger, if one is set. This would be better
+    /// implemented as a macro, as Dogear does, so that we can pass variadic
+    /// arguments without manually invoking `fmt_args!()` every time we want
+    /// to log a message.
+    ///
+    /// The `log` crate's macros aren't suitable here, because those log to the
+    /// global logger. However, we don't want to set the global logger in our
+    /// crate, because that will log _everything_ that uses the Rust `log` crate
+    /// to the Sync logs, including WebRender and audio logging.
+    pub fn debug(&self, args: fmt::Arguments) {
+        let meta = Metadata::builder()
+            .level(Level::Debug)
+            .target(module_path!())
+            .build();
+        if self.enabled(&meta) {
+            self.log(&Record::builder().args(args).metadata(meta).build());
+        }
+    }
+}
+
+impl Log for LogSink {
+    #[inline]
+    fn enabled(&self, meta: &Metadata) -> bool {
+        self.logger.is_some() && meta.level() <= self.max_level
+    }
+
+    fn log(&self, record: &Record) {
+        if !self.enabled(record.metadata()) {
+            return;
+        }
+        if let Some(logger) = &self.logger {
+            let mut message = nsString::new();
+            if write!(message, "{}", record.args()).is_ok() {
+                let task = LogTask {
+                    logger: logger.clone(),
+                    level: record.metadata().level(),
+                    message,
+                };
+                let _ = TaskRunnable::new("extension_storage_sync::Logger::log", Box::new(task))
+                    .and_then(|r| TaskRunnable::dispatch(r, logger.owning_thread()));
+            }
+        }
+    }
+
+    fn flush(&self) {}
+}
+
+/// Logs a message to the mirror logger. This task is created on the background
+/// thread queue, and dispatched to the main thread.
+struct LogTask {
+    logger: ThreadPtrHandle<mozIServicesLogSink>,
+    level: Level,
+    message: nsString,
+}
+
+impl Task for LogTask {
+    fn run(&self) {
+        let logger = self.logger.get().unwrap();
+        match self.level {
+            Level::Error => unsafe {
+                logger.Error(&*self.message);
+            },
+            Level::Warn => unsafe {
+                logger.Warn(&*self.message);
+            },
+            Level::Debug => unsafe {
+                logger.Debug(&*self.message);
+            },
+            Level::Trace => unsafe {
+                logger.Trace(&*self.message);
+            },
+            Level::Info => unsafe {
+                logger.Info(&*self.message);
+            },
+        }
+    }
+
+    fn done(&self) -> Result<(), nsresult> {
+        Ok(())
+    }
+}
diff --git a/services/sync/golden_gate/src/task.rs b/services/sync/golden_gate/src/task.rs
new file mode 100644
index 0000000000..8cab21830b
--- /dev/null
+++ b/services/sync/golden_gate/src/task.rs
@@ -0,0 +1,355 @@
+/* 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/. */
+
+use std::{fmt::Write, mem, result};
+
+use atomic_refcell::AtomicRefCell;
+use moz_task::{DispatchOptions, Task, TaskRunnable, ThreadPtrHandle, ThreadPtrHolder};
+use nserror::{nsresult, NS_ERROR_FAILURE};
+use nsstring::{nsACString, nsCString};
+use sync15::engine::{ApplyResults, BridgedEngine};
+use sync15::Guid;
+use thin_vec::ThinVec;
+use xpcom::{
+    interfaces::{
+        mozIBridgedSyncEngineApplyCallback, mozIBridgedSyncEngineCallback, nsIEventTarget,
+    },
+    RefPtr,
+};
+
+use crate::error::{Error, Result};
+use crate::ferry::{Ferry, FerryResult};
+
+/// A ferry task sends (or ferries) an operation to a bridged engine on a
+/// background thread or task queue, and ferries back an optional result to
+/// a callback.
+pub struct FerryTask {
+    /// We want to ensure scheduled ferries can't block finalization of the underlying
+    /// store - we want a degree of confidence that closing the database will happen when
+    /// we want even if tasks are queued up to run on another thread.
+    /// We rely on the semantics of our BridgedEngines to help here:
+    /// * A bridged engine is expected to hold a weak reference to its store.
+    /// * Our LazyStore is the only thing holding a reference to the "real" store.
+    /// Thus, when our LazyStore asks our "real" store to close, we can be confident
+    /// a close will happen (ie, we assume that the real store will be able to unwrapp
+    /// the underlying sqlite `Connection` (using `Arc::try_unwrap`) and close it.
+    /// However, note that if an operation on the bridged engine is currently running,
+    /// we will block waiting for that operation to complete, so while this isn't
+    /// guaranteed to happen immediately, it should happen "soon enough".
+    engine: Box<dyn BridgedEngine>,
+    ferry: Ferry,
+    callback: ThreadPtrHandle<mozIBridgedSyncEngineCallback>,
+    result: AtomicRefCell<anyhow::Result<FerryResult>>,
+}
+
+impl FerryTask {
+    /// Creates a task to fetch the engine's last sync time, in milliseconds.
+    #[inline]
+    pub fn for_last_sync(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::LastSync, callback)
+    }
+
+    /// Creates a task to set the engine's last sync time, in milliseconds.
+    #[inline]
+    pub fn for_set_last_sync(
+        engine: Box<dyn BridgedEngine>,
+        last_sync_millis: i64,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::SetLastSync(last_sync_millis), callback)
+    }
+
+    /// Creates a task to fetch the engine's sync ID.
+    #[inline]
+    pub fn for_sync_id(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::SyncId, callback)
+    }
+
+    /// Creates a task to reset the engine's sync ID and all its local Sync
+    /// metadata.
+    #[inline]
+    pub fn for_reset_sync_id(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::ResetSyncId, callback)
+    }
+
+    /// Creates a task to compare the bridged engine's local sync ID with
+    /// the `new_sync_id` from `meta/global`, and ferry back the final sync ID
+    /// to use.
+    #[inline]
+    pub fn for_ensure_current_sync_id(
+        engine: Box<dyn BridgedEngine>,
+        new_sync_id: &nsACString,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(
+            engine,
+            Ferry::EnsureCurrentSyncId(std::str::from_utf8(new_sync_id)?.into()),
+            callback,
+        )
+    }
+
+    /// Creates a task to signal that the engine is about to sync.
+    #[inline]
+    pub fn for_sync_started(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::SyncStarted, callback)
+    }
+
+    /// Creates a task to store incoming records.
+    pub fn for_store_incoming(
+        engine: Box<dyn BridgedEngine>,
+        incoming_envelopes_json: &[nsCString],
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(
+            engine,
+            Ferry::StoreIncoming(incoming_envelopes_json.to_vec()),
+            callback,
+        )
+    }
+
+    /// Creates a task to mark a subset of outgoing records as uploaded. This
+    /// may be called multiple times per sync, or not at all if there are no
+    /// records to upload.
+    pub fn for_set_uploaded(
+        engine: Box<dyn BridgedEngine>,
+        server_modified_millis: i64,
+        uploaded_ids: &[nsCString],
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        let uploaded_ids = uploaded_ids.iter().map(|id| Guid::from_slice(id)).collect();
+        Self::with_ferry(
+            engine,
+            Ferry::SetUploaded(server_modified_millis, uploaded_ids),
+            callback,
+        )
+    }
+
+    /// Creates a task to signal that all records have been uploaded, and
+    /// the engine has been synced. This is called even if there were no
+    /// records uploaded.
+    #[inline]
+    pub fn for_sync_finished(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::SyncFinished, callback)
+    }
+
+    /// Creates a task to reset all local Sync state for the engine, without
+    /// erasing user data.
+    #[inline]
+    pub fn for_reset(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::Reset, callback)
+    }
+
+    /// Creates a task to erase all local user data for the engine.
+    #[inline]
+    pub fn for_wipe(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        Self::with_ferry(engine, Ferry::Wipe, callback)
+    }
+
+    /// Creates a task for a ferry. The `callback` is bound to the current
+    /// thread, and will be called once, after the ferry returns from the
+    /// background thread.
+    fn with_ferry(
+        engine: Box<dyn BridgedEngine>,
+        ferry: Ferry,
+        callback: &mozIBridgedSyncEngineCallback,
+    ) -> Result<FerryTask> {
+        let name = ferry.name();
+        Ok(FerryTask {
+            engine,
+            ferry,
+            callback: ThreadPtrHolder::new(
+                cstr!("mozIBridgedSyncEngineCallback"),
+                RefPtr::new(callback),
+            )?,
+            result: AtomicRefCell::new(Err(Error::DidNotRun(name).into())),
+        })
+    }
+
+    /// Dispatches the task to the given thread `target`.
+    pub fn dispatch(self, target: &nsIEventTarget) -> Result<()> {
+        let runnable = TaskRunnable::new(self.ferry.name(), Box::new(self))?;
+        // `may_block` schedules the task on the I/O thread pool, since we
+        // expect most operations to wait on I/O.
+        TaskRunnable::dispatch_with_options(
+            runnable,
+            target,
+            DispatchOptions::default().may_block(true),
+        )?;
+        Ok(())
+    }
+
+    /// Runs the task on the background thread. This is split out into its own
+    /// method to make error handling easier.
+    fn inner_run(&self) -> anyhow::Result<FerryResult> {
+        let engine = &self.engine;
+        Ok(match &self.ferry {
+            Ferry::LastSync => FerryResult::LastSync(engine.last_sync()?),
+            Ferry::SetLastSync(last_sync_millis) => {
+                engine.set_last_sync(*last_sync_millis)?;
+                FerryResult::default()
+            }
+            Ferry::SyncId => FerryResult::SyncId(engine.sync_id()?),
+            Ferry::ResetSyncId => FerryResult::AssignedSyncId(engine.reset_sync_id()?),
+            Ferry::EnsureCurrentSyncId(new_sync_id) => {
+                FerryResult::AssignedSyncId(engine.ensure_current_sync_id(new_sync_id)?)
+            }
+            Ferry::SyncStarted => {
+                engine.sync_started()?;
+                FerryResult::default()
+            }
+            Ferry::StoreIncoming(incoming_envelopes_json) => {
+                let incoming_envelopes = incoming_envelopes_json
+                    .iter()
+                    .map(|envelope| Ok(serde_json::from_slice(envelope)?))
+                    .collect::<Result<_>>()?;
+
+                engine.store_incoming(incoming_envelopes)?;
+                FerryResult::default()
+            }
+            Ferry::SetUploaded(server_modified_millis, uploaded_ids) => {
+                engine.set_uploaded(*server_modified_millis, uploaded_ids.as_slice())?;
+                FerryResult::default()
+            }
+            Ferry::SyncFinished => {
+                engine.sync_finished()?;
+                FerryResult::default()
+            }
+            Ferry::Reset => {
+                engine.reset()?;
+                FerryResult::default()
+            }
+            Ferry::Wipe => {
+                engine.wipe()?;
+                FerryResult::default()
+            }
+        })
+    }
+}
+
+impl Task for FerryTask {
+    fn run(&self) {
+        *self.result.borrow_mut() = self.inner_run();
+    }
+
+    fn done(&self) -> result::Result<(), nsresult> {
+        let callback = self.callback.get().unwrap();
+        match mem::replace(
+            &mut *self.result.borrow_mut(),
+            Err(Error::DidNotRun(self.ferry.name()).into()),
+        ) {
+            Ok(result) => unsafe { callback.HandleSuccess(result.into_variant().coerce()) },
+            Err(err) => {
+                let mut message = nsCString::new();
+                write!(message, "{err}").unwrap();
+                unsafe { callback.HandleError(NS_ERROR_FAILURE, &*message) }
+            }
+        }
+        .to_result()
+    }
+}
+
+/// An apply task ferries incoming records to an engine on a background
+/// thread, and ferries back records to upload. It's separate from
+/// `FerryTask` because its callback type is different.
+pub struct ApplyTask {
+    engine: Box<dyn BridgedEngine>,
+    callback: ThreadPtrHandle<mozIBridgedSyncEngineApplyCallback>,
+    result: AtomicRefCell<anyhow::Result<Vec<String>>>,
+}
+
+impl ApplyTask {
+    /// Returns the task name for debugging.
+    pub fn name() -> &'static str {
+        concat!(module_path!(), "apply")
+    }
+
+    /// Runs the task on the background thread.
+    fn inner_run(&self) -> anyhow::Result<Vec<String>> {
+        let ApplyResults {
+            records: outgoing_records,
+            ..
+        } = self.engine.apply()?;
+        let outgoing_records_json = outgoing_records
+            .iter()
+            .map(|record| Ok(serde_json::to_string(record)?))
+            .collect::<Result<_>>()?;
+        Ok(outgoing_records_json)
+    }
+
+    /// Creates a task. The `callback` is bound to the current thread, and will
+    /// be called once, after the records are applied on the background thread.
+    pub fn new(
+        engine: Box<dyn BridgedEngine>,
+        callback: &mozIBridgedSyncEngineApplyCallback,
+    ) -> Result<ApplyTask> {
+        Ok(ApplyTask {
+            engine,
+            callback: ThreadPtrHolder::new(
+                cstr!("mozIBridgedSyncEngineApplyCallback"),
+                RefPtr::new(callback),
+            )?,
+            result: AtomicRefCell::new(Err(Error::DidNotRun(Self::name()).into())),
+        })
+    }
+
+    /// Dispatches the task to the given thread `target`.
+    pub fn dispatch(self, target: &nsIEventTarget) -> Result<()> {
+        let runnable = TaskRunnable::new(Self::name(), Box::new(self))?;
+        TaskRunnable::dispatch_with_options(
+            runnable,
+            target,
+            DispatchOptions::default().may_block(true),
+        )?;
+        Ok(())
+    }
+}
+
+impl Task for ApplyTask {
+    fn run(&self) {
+        *self.result.borrow_mut() = self.inner_run();
+    }
+
+    fn done(&self) -> result::Result<(), nsresult> {
+        let callback = self.callback.get().unwrap();
+        match mem::replace(
+            &mut *self.result.borrow_mut(),
+            Err(Error::DidNotRun(Self::name()).into()),
+        ) {
+            Ok(envelopes) => {
+                let result = envelopes
+                    .into_iter()
+                    .map(nsCString::from)
+                    .collect::<ThinVec<_>>();
+                unsafe { callback.HandleSuccess(&result) }
+            }
+            Err(err) => {
+                let mut message = nsCString::new();
+                write!(message, "{err}").unwrap();
+                unsafe { callback.HandleError(NS_ERROR_FAILURE, &*message) }
+            }
+        }
+        .to_result()
+    }
+}