Adding upstream version 115.7.0esr.upstream/115.7.0esrupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

13 files changed, 4472 insertions, 0 deletions

diff --git a/third_party/rust/webext-storage/src/api.rs b/third_party/rust/webext-storage/src/api.rs
new file mode 100644
index 0000000000..fe838a075d
--- /dev/null
+++ b/third_party/rust/webext-storage/src/api.rs
@@ -0,0 +1,745 @@
+/* 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 crate::error::*;
+use rusqlite::{Connection, Transaction};
+use serde::{ser::SerializeMap, Serialize, Serializer};
+
+use serde_json::{Map, Value as JsonValue};
+use sql_support::{self, ConnExt};
+
+// These constants are defined by the chrome.storage.sync spec. We export them
+// publicly from this module, then from the crate, so they wind up in the
+// clients.
+// Note the limits for `chrome.storage.sync` and `chrome.storage.local` are
+// different, and these are from `.sync` - we'll have work to do if we end up
+// wanting this to be used for `.local` too!
+pub const SYNC_QUOTA_BYTES: usize = 102_400;
+pub const SYNC_QUOTA_BYTES_PER_ITEM: usize = 8_192;
+pub const SYNC_MAX_ITEMS: usize = 512;
+// Note there are also constants for "operations per minute" etc, which aren't
+// enforced here.
+
+type JsonMap = Map<String, JsonValue>;
+
+enum StorageChangeOp {
+    Clear,
+    Set(JsonValue),
+    SetWithoutQuota(JsonValue),
+}
+
+fn get_from_db(conn: &Connection, ext_id: &str) -> Result<Option<JsonMap>> {
+    Ok(
+        match conn.try_query_one::<String, _>(
+            "SELECT data FROM storage_sync_data
+             WHERE ext_id = :ext_id",
+            &[(":ext_id", &ext_id)],
+            true,
+        )? {
+            Some(s) => match serde_json::from_str(&s)? {
+                JsonValue::Object(m) => Some(m),
+                // we could panic here as it's theoretically impossible, but we
+                // might as well treat it as not existing...
+                _ => None,
+            },
+            None => None,
+        },
+    )
+}
+
+fn save_to_db(tx: &Transaction<'_>, ext_id: &str, val: &StorageChangeOp) -> Result<()> {
+    // This function also handles removals. Either an empty map or explicit null
+    // is a removal. If there's a mirror record for this extension ID, then we
+    // must leave a tombstone behind for syncing.
+    let is_delete = match val {
+        StorageChangeOp::Clear => true,
+        StorageChangeOp::Set(JsonValue::Object(v)) => v.is_empty(),
+        StorageChangeOp::SetWithoutQuota(JsonValue::Object(v)) => v.is_empty(),
+        _ => false,
+    };
+    if is_delete {
+        let in_mirror = tx
+            .try_query_one(
+                "SELECT EXISTS(SELECT 1 FROM storage_sync_mirror WHERE ext_id = :ext_id);",
+                rusqlite::named_params! {
+                    ":ext_id": ext_id,
+                },
+                true,
+            )?
+            .unwrap_or_default();
+        if in_mirror {
+            log::trace!("saving data for '{}': leaving a tombstone", ext_id);
+            tx.execute_cached(
+                "
+                INSERT INTO storage_sync_data(ext_id, data, sync_change_counter)
+                VALUES (:ext_id, NULL, 1)
+                ON CONFLICT (ext_id) DO UPDATE
+                SET data = NULL, sync_change_counter = sync_change_counter + 1",
+                rusqlite::named_params! {
+                    ":ext_id": ext_id,
+                },
+            )?;
+        } else {
+            log::trace!("saving data for '{}': removing the row", ext_id);
+            tx.execute_cached(
+                "
+                DELETE FROM storage_sync_data WHERE ext_id = :ext_id",
+                rusqlite::named_params! {
+                    ":ext_id": ext_id,
+                },
+            )?;
+        }
+    } else {
+        // Convert to bytes so we can enforce the quota if necessary.
+        let sval = match val {
+            StorageChangeOp::Set(v) => {
+                let sv = v.to_string();
+                if sv.len() > SYNC_QUOTA_BYTES {
+                    return Err(ErrorKind::QuotaError(QuotaReason::TotalBytes).into());
+                }
+                sv
+            }
+            StorageChangeOp::SetWithoutQuota(v) => v.to_string(),
+            StorageChangeOp::Clear => unreachable!(),
+        };
+
+        log::trace!("saving data for '{}': writing", ext_id);
+        tx.execute_cached(
+            "INSERT INTO storage_sync_data(ext_id, data, sync_change_counter)
+                VALUES (:ext_id, :data, 1)
+                ON CONFLICT (ext_id) DO UPDATE
+                set data=:data, sync_change_counter = sync_change_counter + 1",
+            rusqlite::named_params! {
+                ":ext_id": ext_id,
+                ":data": &sval,
+            },
+        )?;
+    }
+    Ok(())
+}
+
+fn remove_from_db(tx: &Transaction<'_>, ext_id: &str) -> Result<()> {
+    save_to_db(tx, ext_id, &StorageChangeOp::Clear)
+}
+
+// This is a "helper struct" for the callback part of the chrome.storage spec,
+// but shaped in a way to make it more convenient from the rust side of the
+// world.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct StorageValueChange {
+    #[serde(skip_serializing)]
+    pub key: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub old_value: Option<JsonValue>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub new_value: Option<JsonValue>,
+}
+
+// This is, largely, a helper so that this serializes correctly as per the
+// chrome.storage.sync spec. If not for custom serialization it should just
+// be a plain vec
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct StorageChanges {
+    changes: Vec<StorageValueChange>,
+}
+
+impl StorageChanges {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn with_capacity(n: usize) -> Self {
+        Self {
+            changes: Vec::with_capacity(n),
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.changes.is_empty()
+    }
+
+    pub fn push(&mut self, change: StorageValueChange) {
+        self.changes.push(change)
+    }
+}
+
+// and it serializes as a map.
+impl Serialize for StorageChanges {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let mut map = serializer.serialize_map(Some(self.changes.len()))?;
+        for change in &self.changes {
+            map.serialize_entry(&change.key, change)?;
+        }
+        map.end()
+    }
+}
+
+// A helper to determine the size of a key/value combination from the
+// perspective of quota and getBytesInUse().
+pub fn get_quota_size_of(key: &str, v: &JsonValue) -> usize {
+    // Reading the chrome docs literally re the quota, the length of the key
+    // is just the string len, but the value is the json val, as bytes.
+    key.len() + v.to_string().len()
+}
+
+/// The implementation of `storage[.sync].set()`. On success this returns the
+/// StorageChanges defined by the chrome API - it's assumed the caller will
+/// arrange to deliver this to observers as defined in that API.
+pub fn set(tx: &Transaction<'_>, ext_id: &str, val: JsonValue) -> Result<StorageChanges> {
+    let val_map = match val {
+        JsonValue::Object(m) => m,
+        // Not clear what the error semantics should be yet. For now, pretend an empty map.
+        _ => Map::new(),
+    };
+
+    let mut current = get_from_db(tx, ext_id)?.unwrap_or_default();
+
+    let mut changes = StorageChanges::with_capacity(val_map.len());
+
+    // iterate over the value we are adding/updating.
+    for (k, v) in val_map.into_iter() {
+        let old_value = current.remove(&k);
+        if current.len() >= SYNC_MAX_ITEMS {
+            return Err(ErrorKind::QuotaError(QuotaReason::MaxItems).into());
+        }
+        // Reading the chrome docs literally re the quota, the length of the key
+        // is just the string len, but the value is the json val, as bytes
+        if get_quota_size_of(&k, &v) > SYNC_QUOTA_BYTES_PER_ITEM {
+            return Err(ErrorKind::QuotaError(QuotaReason::ItemBytes).into());
+        }
+        let change = StorageValueChange {
+            key: k.clone(),
+            old_value,
+            new_value: Some(v.clone()),
+        };
+        changes.push(change);
+        current.insert(k, v);
+    }
+
+    save_to_db(
+        tx,
+        ext_id,
+        &StorageChangeOp::Set(JsonValue::Object(current)),
+    )?;
+    Ok(changes)
+}
+
+// A helper which takes a param indicating what keys should be returned and
+// converts that to a vec of real strings. Also returns "default" values to
+// be used if no item exists for that key.
+fn get_keys(keys: JsonValue) -> Vec<(String, Option<JsonValue>)> {
+    match keys {
+        JsonValue::String(s) => vec![(s, None)],
+        JsonValue::Array(keys) => {
+            // because nothing with json is ever simple, each key may not be
+            // a string. We ignore any which aren't.
+            keys.iter()
+                .filter_map(|v| v.as_str().map(|s| (s.to_string(), None)))
+                .collect()
+        }
+        JsonValue::Object(m) => m.into_iter().map(|(k, d)| (k, Some(d))).collect(),
+        _ => vec![],
+    }
+}
+
+/// The implementation of `storage[.sync].get()` - on success this always
+/// returns a Json object.
+pub fn get(conn: &Connection, ext_id: &str, keys: JsonValue) -> Result<JsonValue> {
+    // key is optional, or string or array of string or object keys
+    let maybe_existing = get_from_db(conn, ext_id)?;
+    let mut existing = match (maybe_existing, keys.is_object()) {
+        (None, true) => return Ok(keys),
+        (None, false) => return Ok(JsonValue::Object(Map::new())),
+        (Some(v), _) => v,
+    };
+    // take the quick path for null, where we just return the entire object.
+    if keys.is_null() {
+        return Ok(JsonValue::Object(existing));
+    }
+    // OK, so we need to build a list of keys to get.
+    let keys_and_defaults = get_keys(keys);
+    let mut result = Map::with_capacity(keys_and_defaults.len());
+    for (key, maybe_default) in keys_and_defaults {
+        if let Some(v) = existing.remove(&key) {
+            result.insert(key, v);
+        } else if let Some(def) = maybe_default {
+            result.insert(key, def);
+        }
+        // else |keys| is a string/array instead of an object with defaults.
+        // Don't include keys without default values.
+    }
+    Ok(JsonValue::Object(result))
+}
+
+/// The implementation of `storage[.sync].remove()`. On success this returns the
+/// StorageChanges defined by the chrome API - it's assumed the caller will
+/// arrange to deliver this to observers as defined in that API.
+pub fn remove(tx: &Transaction<'_>, ext_id: &str, keys: JsonValue) -> Result<StorageChanges> {
+    let mut existing = match get_from_db(tx, ext_id)? {
+        None => return Ok(StorageChanges::new()),
+        Some(v) => v,
+    };
+
+    // Note: get_keys parses strings, arrays and objects, but remove()
+    // is expected to only be passed a string or array of strings.
+    let keys_and_defs = get_keys(keys);
+
+    let mut result = StorageChanges::with_capacity(keys_and_defs.len());
+    for (key, _) in keys_and_defs {
+        if let Some(v) = existing.remove(&key) {
+            result.push(StorageValueChange {
+                key,
+                old_value: Some(v),
+                new_value: None,
+            });
+        }
+    }
+    if !result.is_empty() {
+        save_to_db(
+            tx,
+            ext_id,
+            &StorageChangeOp::SetWithoutQuota(JsonValue::Object(existing)),
+        )?;
+    }
+    Ok(result)
+}
+
+/// The implementation of `storage[.sync].clear()`. On success this returns the
+/// StorageChanges defined by the chrome API - it's assumed the caller will
+/// arrange to deliver this to observers as defined in that API.
+pub fn clear(tx: &Transaction<'_>, ext_id: &str) -> Result<StorageChanges> {
+    let existing = match get_from_db(tx, ext_id)? {
+        None => return Ok(StorageChanges::new()),
+        Some(v) => v,
+    };
+    let mut result = StorageChanges::with_capacity(existing.len());
+    for (key, val) in existing.into_iter() {
+        result.push(StorageValueChange {
+            key: key.to_string(),
+            new_value: None,
+            old_value: Some(val),
+        });
+    }
+    remove_from_db(tx, ext_id)?;
+    Ok(result)
+}
+
+/// The implementation of `storage[.sync].getBytesInUse()`.
+pub fn get_bytes_in_use(conn: &Connection, ext_id: &str, keys: JsonValue) -> Result<usize> {
+    let maybe_existing = get_from_db(conn, ext_id)?;
+    let existing = match maybe_existing {
+        None => return Ok(0),
+        Some(v) => v,
+    };
+    // Make an array of all the keys we we are going to count.
+    let keys: Vec<&str> = match &keys {
+        JsonValue::Null => existing.keys().map(|v| v.as_str()).collect(),
+        JsonValue::String(name) => vec![name.as_str()],
+        JsonValue::Array(names) => names.iter().filter_map(|v| v.as_str()).collect(),
+        // in the spirit of json-based APIs, silently ignore strange things.
+        _ => return Ok(0),
+    };
+    // We must use the same way of counting as our quota enforcement.
+    let mut size = 0;
+    for key in keys.into_iter() {
+        if let Some(v) = existing.get(key) {
+            size += get_quota_size_of(key, v);
+        }
+    }
+    Ok(size)
+}
+
+/// Information about the usage of a single extension.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct UsageInfo {
+    /// The extension id.
+    pub ext_id: String,
+    /// The number of keys the extension uses.
+    pub num_keys: usize,
+    /// The number of bytes used by the extension. This result is somewhat rough
+    /// -- it doesn't bother counting the size of the extension ID, or data in
+    /// the mirror, and favors returning the exact number of bytes used by the
+    /// column (that is, the size of the JSON object) rather than replicating
+    /// the `get_bytes_in_use` return value for all keys.
+    pub num_bytes: usize,
+}
+
+/// Exposes information about per-collection usage for the purpose of telemetry.
+/// (Doesn't map to an actual `chrome.storage.sync` API).
+pub fn usage(db: &Connection) -> Result<Vec<UsageInfo>> {
+    type JsonObject = Map<String, JsonValue>;
+    let sql = "
+        SELECT ext_id, data
+        FROM storage_sync_data
+        WHERE data IS NOT NULL
+        -- for tests and determinism
+        ORDER BY ext_id
+    ";
+    db.query_rows_into(sql, [], |row| {
+        let ext_id: String = row.get("ext_id")?;
+        let data: String = row.get("data")?;
+        let num_bytes = data.len();
+        let num_keys = serde_json::from_str::<JsonObject>(&data)?.len();
+        Ok(UsageInfo {
+            ext_id,
+            num_keys,
+            num_bytes,
+        })
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db::test::new_mem_db;
+    use serde_json::json;
+
+    #[test]
+    fn test_serialize_storage_changes() -> Result<()> {
+        let c = StorageChanges {
+            changes: vec![StorageValueChange {
+                key: "key".to_string(),
+                old_value: Some(json!("old")),
+                new_value: None,
+            }],
+        };
+        assert_eq!(serde_json::to_string(&c)?, r#"{"key":{"oldValue":"old"}}"#);
+        let c = StorageChanges {
+            changes: vec![StorageValueChange {
+                key: "key".to_string(),
+                old_value: None,
+                new_value: Some(json!({"foo": "bar"})),
+            }],
+        };
+        assert_eq!(
+            serde_json::to_string(&c)?,
+            r#"{"key":{"newValue":{"foo":"bar"}}}"#
+        );
+        Ok(())
+    }
+
+    fn make_changes(changes: &[(&str, Option<JsonValue>, Option<JsonValue>)]) -> StorageChanges {
+        let mut r = StorageChanges::with_capacity(changes.len());
+        for (name, old_value, new_value) in changes {
+            r.push(StorageValueChange {
+                key: (*name).to_string(),
+                old_value: old_value.clone(),
+                new_value: new_value.clone(),
+            });
+        }
+        r
+    }
+
+    #[test]
+    fn test_simple() -> Result<()> {
+        let ext_id = "x";
+        let mut db = new_mem_db();
+        let tx = db.transaction()?;
+
+        // an empty store.
+        for q in vec![JsonValue::Null, json!("foo"), json!(["foo"])].into_iter() {
+            assert_eq!(get(&tx, ext_id, q)?, json!({}));
+        }
+
+        // Default values in an empty store.
+        for q in vec![json!({ "foo": null }), json!({"foo": "default"})].into_iter() {
+            assert_eq!(get(&tx, ext_id, q.clone())?, q.clone());
+        }
+
+        // Single item in the store.
+        set(&tx, ext_id, json!({"foo": "bar" }))?;
+        for q in vec![
+            JsonValue::Null,
+            json!("foo"),
+            json!(["foo"]),
+            json!({ "foo": null }),
+            json!({"foo": "default"}),
+        ]
+        .into_iter()
+        {
+            assert_eq!(get(&tx, ext_id, q)?, json!({"foo": "bar" }));
+        }
+
+        // Default values in a non-empty store.
+        for q in vec![
+            json!({ "non_existing_key": null }),
+            json!({"non_existing_key": 0}),
+            json!({"non_existing_key": false}),
+            json!({"non_existing_key": "default"}),
+            json!({"non_existing_key": ["array"]}),
+            json!({"non_existing_key": {"objectkey": "value"}}),
+        ]
+        .into_iter()
+        {
+            assert_eq!(get(&tx, ext_id, q.clone())?, q.clone());
+        }
+
+        // more complex stuff, including changes checking.
+        assert_eq!(
+            set(&tx, ext_id, json!({"foo": "new", "other": "also new" }))?,
+            make_changes(&[
+                ("foo", Some(json!("bar")), Some(json!("new"))),
+                ("other", None, Some(json!("also new")))
+            ])
+        );
+        assert_eq!(
+            get(&tx, ext_id, JsonValue::Null)?,
+            json!({"foo": "new", "other": "also new"})
+        );
+        assert_eq!(get(&tx, ext_id, json!("foo"))?, json!({"foo": "new"}));
+        assert_eq!(
+            get(&tx, ext_id, json!(["foo", "other"]))?,
+            json!({"foo": "new", "other": "also new"})
+        );
+        assert_eq!(
+            get(&tx, ext_id, json!({"foo": null, "default": "yo"}))?,
+            json!({"foo": "new", "default": "yo"})
+        );
+
+        assert_eq!(
+            remove(&tx, ext_id, json!("foo"))?,
+            make_changes(&[("foo", Some(json!("new")), None)]),
+        );
+
+        assert_eq!(
+            set(&tx, ext_id, json!({"foo": {"sub-object": "sub-value"}}))?,
+            make_changes(&[("foo", None, Some(json!({"sub-object": "sub-value"}))),])
+        );
+
+        // XXX - other variants.
+
+        assert_eq!(
+            clear(&tx, ext_id)?,
+            make_changes(&[
+                ("foo", Some(json!({"sub-object": "sub-value"})), None),
+                ("other", Some(json!("also new")), None),
+            ]),
+        );
+        assert_eq!(get(&tx, ext_id, JsonValue::Null)?, json!({}));
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_check_get_impl() -> Result<()> {
+        // This is a port of checkGetImpl in test_ext_storage.js in Desktop.
+        let ext_id = "x";
+        let mut db = new_mem_db();
+        let tx = db.transaction()?;
+
+        let prop = "test-prop";
+        let value = "test-value";
+
+        set(&tx, ext_id, json!({ prop: value }))?;
+
+        // this is the checkGetImpl part!
+        let mut data = get(&tx, ext_id, json!(null))?;
+        assert_eq!(value, json!(data[prop]), "null getter worked for {}", prop);
+
+        data = get(&tx, ext_id, json!(prop))?;
+        assert_eq!(
+            value,
+            json!(data[prop]),
+            "string getter worked for {}",
+            prop
+        );
+        assert_eq!(
+            data.as_object().unwrap().len(),
+            1,
+            "string getter should return an object with a single property"
+        );
+
+        data = get(&tx, ext_id, json!([prop]))?;
+        assert_eq!(value, json!(data[prop]), "array getter worked for {}", prop);
+        assert_eq!(
+            data.as_object().unwrap().len(),
+            1,
+            "array getter with a single key should return an object with a single property"
+        );
+
+        // checkGetImpl() uses `{ [prop]: undefined }` - but json!() can't do that :(
+        // Hopefully it's just testing a simple object, so we use `{ prop: null }`
+        data = get(&tx, ext_id, json!({ prop: null }))?;
+        assert_eq!(
+            value,
+            json!(data[prop]),
+            "object getter worked for {}",
+            prop
+        );
+        assert_eq!(
+            data.as_object().unwrap().len(),
+            1,
+            "object getter with a single key should return an object with a single property"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_bug_1621162() -> Result<()> {
+        // apparently Firefox, unlike Chrome, will not optimize the changes.
+        // See bug 1621162 for more!
+        let mut db = new_mem_db();
+        let tx = db.transaction()?;
+        let ext_id = "xyz";
+
+        set(&tx, ext_id, json!({"foo": "bar" }))?;
+
+        assert_eq!(
+            set(&tx, ext_id, json!({"foo": "bar" }))?,
+            make_changes(&[("foo", Some(json!("bar")), Some(json!("bar")))]),
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_quota_maxitems() -> Result<()> {
+        let mut db = new_mem_db();
+        let tx = db.transaction()?;
+        let ext_id = "xyz";
+        for i in 1..SYNC_MAX_ITEMS + 1 {
+            set(
+                &tx,
+                ext_id,
+                json!({ format!("key-{}", i): format!("value-{}", i) }),
+            )?;
+        }
+        let e = set(&tx, ext_id, json!({"another": "another"})).unwrap_err();
+        match e.kind() {
+            ErrorKind::QuotaError(QuotaReason::MaxItems) => {}
+            _ => panic!("unexpected error type"),
+        };
+        Ok(())
+    }
+
+    #[test]
+    fn test_quota_bytesperitem() -> Result<()> {
+        let mut db = new_mem_db();
+        let tx = db.transaction()?;
+        let ext_id = "xyz";
+        // A string 5 bytes less than the max. This should be counted as being
+        // 3 bytes less than the max as the quotes are counted. Plus the length
+        // of the key (no quotes) means we should come in 2 bytes under.
+        let val = "x".repeat(SYNC_QUOTA_BYTES_PER_ITEM - 5);
+
+        // Key length doesn't push it over.
+        set(&tx, ext_id, json!({ "x": val }))?;
+        assert_eq!(
+            get_bytes_in_use(&tx, ext_id, json!("x"))?,
+            SYNC_QUOTA_BYTES_PER_ITEM - 2
+        );
+
+        // Key length does push it over.
+        let e = set(&tx, ext_id, json!({ "xxxx": val })).unwrap_err();
+        match e.kind() {
+            ErrorKind::QuotaError(QuotaReason::ItemBytes) => {}
+            _ => panic!("unexpected error type"),
+        };
+        Ok(())
+    }
+
+    #[test]
+    fn test_quota_bytes() -> Result<()> {
+        let mut db = new_mem_db();
+        let tx = db.transaction()?;
+        let ext_id = "xyz";
+        let val = "x".repeat(SYNC_QUOTA_BYTES + 1);
+
+        // Init an over quota db with a single key.
+        save_to_db(
+            &tx,
+            ext_id,
+            &StorageChangeOp::SetWithoutQuota(json!({ "x": val })),
+        )?;
+
+        // Adding more data fails.
+        let e = set(&tx, ext_id, json!({ "y": "newvalue" })).unwrap_err();
+        match e.kind() {
+            ErrorKind::QuotaError(QuotaReason::TotalBytes) => {}
+            _ => panic!("unexpected error type"),
+        };
+
+        // Remove data does not fails.
+        remove(&tx, ext_id, json!["x"])?;
+
+        // Restore the over quota data.
+        save_to_db(
+            &tx,
+            ext_id,
+            &StorageChangeOp::SetWithoutQuota(json!({ "y": val })),
+        )?;
+
+        // Overwrite with less data does not fail.
+        set(&tx, ext_id, json!({ "y": "lessdata" }))?;
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_get_bytes_in_use() -> Result<()> {
+        let mut db = new_mem_db();
+        let tx = db.transaction()?;
+        let ext_id = "xyz";
+
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!(null))?, 0);
+
+        set(&tx, ext_id, json!({ "a": "a" }))?; // should be 4
+        set(&tx, ext_id, json!({ "b": "bb" }))?; // should be 5
+        set(&tx, ext_id, json!({ "c": "ccc" }))?; // should be 6
+        set(&tx, ext_id, json!({ "n": 999_999 }))?; // should be 7
+
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!("x"))?, 0);
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!("a"))?, 4);
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!("b"))?, 5);
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!("c"))?, 6);
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!("n"))?, 7);
+
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!(["a"]))?, 4);
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!(["a", "x"]))?, 4);
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!(["a", "b"]))?, 9);
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!(["a", "c"]))?, 10);
+
+        assert_eq!(
+            get_bytes_in_use(&tx, ext_id, json!(["a", "b", "c", "n"]))?,
+            22
+        );
+        assert_eq!(get_bytes_in_use(&tx, ext_id, json!(null))?, 22);
+        Ok(())
+    }
+
+    #[test]
+    fn test_usage() {
+        let mut db = new_mem_db();
+        let tx = db.transaction().unwrap();
+        // '{"a":"a","b":"bb","c":"ccc","n":999999}': 39 bytes
+        set(&tx, "xyz", json!({ "a": "a" })).unwrap();
+        set(&tx, "xyz", json!({ "b": "bb" })).unwrap();
+        set(&tx, "xyz", json!({ "c": "ccc" })).unwrap();
+        set(&tx, "xyz", json!({ "n": 999_999 })).unwrap();
+
+        // '{"a":"a"}': 9 bytes
+        set(&tx, "abc", json!({ "a": "a" })).unwrap();
+
+        tx.commit().unwrap();
+
+        let usage = usage(&db).unwrap();
+        let expect = [
+            UsageInfo {
+                ext_id: "abc".to_string(),
+                num_keys: 1,
+                num_bytes: 9,
+            },
+            UsageInfo {
+                ext_id: "xyz".to_string(),
+                num_keys: 4,
+                num_bytes: 39,
+            },
+        ];
+        assert_eq!(&usage, &expect);
+    }
+}
diff --git a/third_party/rust/webext-storage/src/db.rs b/third_party/rust/webext-storage/src/db.rs
new file mode 100644
index 0000000000..c88dc99959
--- /dev/null
+++ b/third_party/rust/webext-storage/src/db.rs
@@ -0,0 +1,301 @@
+/* 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 crate::error::*;
+use crate::schema;
+use interrupt_support::{SqlInterruptHandle, SqlInterruptScope};
+use parking_lot::Mutex;
+use rusqlite::types::{FromSql, ToSql};
+use rusqlite::Connection;
+use rusqlite::OpenFlags;
+use sql_support::open_database::open_database_with_flags;
+use sql_support::ConnExt;
+use std::ops::{Deref, DerefMut};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use url::Url;
+
+/// A `StorageDb` wraps a read-write SQLite connection, and handles schema
+/// migrations and recovering from database file corruption. It can be used
+/// anywhere a `rusqlite::Connection` is expected, thanks to its `Deref{Mut}`
+/// implementations.
+///
+/// We only support a single writer connection - so that's the only thing we
+/// store. It's still a bit overkill, but there's only so many yaks in a day.
+pub struct StorageDb {
+    writer: Connection,
+    interrupt_handle: Arc<SqlInterruptHandle>,
+}
+impl StorageDb {
+    /// Create a new, or fetch an already open, StorageDb backed by a file on disk.
+    pub fn new(db_path: impl AsRef<Path>) -> Result<Self> {
+        let db_path = normalize_path(db_path)?;
+        Self::new_named(db_path)
+    }
+
+    /// Create a new, or fetch an already open, memory-based StorageDb. You must
+    /// provide a name, but you are still able to have a single writer and many
+    /// reader connections to the same memory DB open.
+    #[cfg(test)]
+    pub fn new_memory(db_path: &str) -> Result<Self> {
+        let name = PathBuf::from(format!("file:{}?mode=memory&cache=shared", db_path));
+        Self::new_named(name)
+    }
+
+    fn new_named(db_path: PathBuf) -> Result<Self> {
+        // We always create the read-write connection for an initial open so
+        // we can create the schema and/or do version upgrades.
+        let flags = OpenFlags::SQLITE_OPEN_NO_MUTEX
+            | OpenFlags::SQLITE_OPEN_URI
+            | OpenFlags::SQLITE_OPEN_CREATE
+            | OpenFlags::SQLITE_OPEN_READ_WRITE;
+
+        let conn = open_database_with_flags(db_path, flags, &schema::WebExtMigrationLogin)?;
+        Ok(Self {
+            interrupt_handle: Arc::new(SqlInterruptHandle::new(&conn)),
+            writer: conn,
+        })
+    }
+
+    pub fn interrupt_handle(&self) -> Arc<SqlInterruptHandle> {
+        Arc::clone(&self.interrupt_handle)
+    }
+
+    #[allow(dead_code)]
+    pub fn begin_interrupt_scope(&self) -> Result<SqlInterruptScope> {
+        Ok(self.interrupt_handle.begin_interrupt_scope()?)
+    }
+
+    /// Closes the database connection. If there are any unfinalized prepared
+    /// statements on the connection, `close` will fail and the `StorageDb` will
+    /// remain open and the connection will be leaked - we used to return the
+    /// underlying connection so the caller can retry but (a) that's very tricky
+    /// in an Arc<Mutex<>> world and (b) we never actually took advantage of
+    /// that retry capability.
+    pub fn close(self) -> Result<()> {
+        self.writer.close().map_err(|(writer, err)| {
+            // In rusqlite 0.28.0 and earlier, if we just let `writer` drop,
+            // the close would panic on failure.
+            // Later rusqlite versions will not panic, but this behavior doesn't
+            // hurt there.
+            std::mem::forget(writer);
+            err.into()
+        })
+    }
+}
+
+impl Deref for StorageDb {
+    type Target = Connection;
+
+    fn deref(&self) -> &Self::Target {
+        &self.writer
+    }
+}
+
+impl DerefMut for StorageDb {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.writer
+    }
+}
+
+// We almost exclusively use this ThreadSafeStorageDb
+pub struct ThreadSafeStorageDb {
+    db: Mutex<StorageDb>,
+    // This "outer" interrupt_handle not protected by the mutex means
+    // consumers can interrupt us when the mutex is held - which it always will
+    // be if we are doing anything interruptable!
+    interrupt_handle: Arc<SqlInterruptHandle>,
+}
+
+impl ThreadSafeStorageDb {
+    pub fn new(db: StorageDb) -> Self {
+        Self {
+            interrupt_handle: db.interrupt_handle(),
+            db: Mutex::new(db),
+        }
+    }
+
+    pub fn interrupt_handle(&self) -> Arc<SqlInterruptHandle> {
+        Arc::clone(&self.interrupt_handle)
+    }
+
+    pub fn begin_interrupt_scope(&self) -> Result<SqlInterruptScope> {
+        Ok(self.interrupt_handle.begin_interrupt_scope()?)
+    }
+
+    pub fn into_inner(self) -> StorageDb {
+        self.db.into_inner()
+    }
+}
+
+// Deref to a Mutex<StorageDb>, which is how we will use ThreadSafeStorageDb most of the time
+impl Deref for ThreadSafeStorageDb {
+    type Target = Mutex<StorageDb>;
+
+    #[inline]
+    fn deref(&self) -> &Mutex<StorageDb> {
+        &self.db
+    }
+}
+
+// Also implement AsRef<SqlInterruptHandle> so that we can interrupt this at shutdown
+impl AsRef<SqlInterruptHandle> for ThreadSafeStorageDb {
+    fn as_ref(&self) -> &SqlInterruptHandle {
+        &self.interrupt_handle
+    }
+}
+
+pub(crate) mod sql_fns {
+    use rusqlite::{functions::Context, Result};
+    use sync_guid::Guid as SyncGuid;
+
+    #[inline(never)]
+    pub fn generate_guid(_ctx: &Context<'_>) -> Result<SyncGuid> {
+        Ok(SyncGuid::random())
+    }
+}
+
+// These should be somewhere else...
+pub fn put_meta(db: &Connection, key: &str, value: &dyn ToSql) -> Result<()> {
+    db.conn().execute_cached(
+        "REPLACE INTO meta (key, value) VALUES (:key, :value)",
+        rusqlite::named_params! { ":key": key, ":value": value },
+    )?;
+    Ok(())
+}
+
+pub fn get_meta<T: FromSql>(db: &Connection, key: &str) -> Result<Option<T>> {
+    let res = db.conn().try_query_one(
+        "SELECT value FROM meta WHERE key = :key",
+        &[(":key", &key)],
+        true,
+    )?;
+    Ok(res)
+}
+
+pub fn delete_meta(db: &Connection, key: &str) -> Result<()> {
+    db.conn()
+        .execute_cached("DELETE FROM meta WHERE key = :key", &[(":key", &key)])?;
+    Ok(())
+}
+
+// Utilities for working with paths.
+// (From places_utils - ideally these would be shared, but the use of
+// ErrorKind values makes that non-trivial.
+
+/// `Path` is basically just a `str` with no validation, and so in practice it
+/// could contain a file URL. Rusqlite takes advantage of this a bit, and says
+/// `AsRef<Path>` but really means "anything sqlite can take as an argument".
+///
+/// Swift loves using file urls (the only support it has for file manipulation
+/// is through file urls), so it's handy to support them if possible.
+fn unurl_path(p: impl AsRef<Path>) -> PathBuf {
+    p.as_ref()
+        .to_str()
+        .and_then(|s| Url::parse(s).ok())
+        .and_then(|u| {
+            if u.scheme() == "file" {
+                u.to_file_path().ok()
+            } else {
+                None
+            }
+        })
+        .unwrap_or_else(|| p.as_ref().to_owned())
+}
+
+/// If `p` is a file URL, return it, otherwise try and make it one.
+///
+/// Errors if `p` is a relative non-url path, or if it's a URL path
+/// that's isn't a `file:` URL.
+#[allow(dead_code)]
+pub fn ensure_url_path(p: impl AsRef<Path>) -> Result<Url> {
+    if let Some(u) = p.as_ref().to_str().and_then(|s| Url::parse(s).ok()) {
+        if u.scheme() == "file" {
+            Ok(u)
+        } else {
+            Err(ErrorKind::IllegalDatabasePath(p.as_ref().to_owned()).into())
+        }
+    } else {
+        let p = p.as_ref();
+        let u = Url::from_file_path(p).map_err(|_| ErrorKind::IllegalDatabasePath(p.to_owned()))?;
+        Ok(u)
+    }
+}
+
+/// As best as possible, convert `p` into an absolute path, resolving
+/// all symlinks along the way.
+///
+/// If `p` is a file url, it's converted to a path before this.
+fn normalize_path(p: impl AsRef<Path>) -> Result<PathBuf> {
+    let path = unurl_path(p);
+    if let Ok(canonical) = path.canonicalize() {
+        return Ok(canonical);
+    }
+    // It probably doesn't exist yet. This is an error, although it seems to
+    // work on some systems.
+    //
+    // We resolve this by trying to canonicalize the parent directory, and
+    // appending the requested file name onto that. If we can't canonicalize
+    // the parent, we return an error.
+    //
+    // Also, we return errors if the path ends in "..", if there is no
+    // parent directory, etc.
+    let file_name = path
+        .file_name()
+        .ok_or_else(|| ErrorKind::IllegalDatabasePath(path.clone()))?;
+
+    let parent = path
+        .parent()
+        .ok_or_else(|| ErrorKind::IllegalDatabasePath(path.clone()))?;
+
+    let mut canonical = parent.canonicalize()?;
+    canonical.push(file_name);
+    Ok(canonical)
+}
+
+// Helpers for tests
+#[cfg(test)]
+pub mod test {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    // A helper for our tests to get their own memory Api.
+    static ATOMIC_COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+    pub fn new_mem_db() -> StorageDb {
+        let _ = env_logger::try_init();
+        let counter = ATOMIC_COUNTER.fetch_add(1, Ordering::Relaxed);
+        StorageDb::new_memory(&format!("test-api-{}", counter)).expect("should get an API")
+    }
+
+    pub fn new_mem_thread_safe_storage_db() -> Arc<ThreadSafeStorageDb> {
+        Arc::new(ThreadSafeStorageDb::new(new_mem_db()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::test::*;
+    use super::*;
+
+    // Sanity check that we can create a database.
+    #[test]
+    fn test_open() {
+        new_mem_db();
+        // XXX - should we check anything else? Seems a bit pointless, but if
+        // we move the meta functions away from here then it's better than
+        // nothing.
+    }
+
+    #[test]
+    fn test_meta() -> Result<()> {
+        let writer = new_mem_db();
+        assert_eq!(get_meta::<String>(&writer, "foo")?, None);
+        put_meta(&writer, "foo", &"bar".to_string())?;
+        assert_eq!(get_meta(&writer, "foo")?, Some("bar".to_string()));
+        delete_meta(&writer, "foo")?;
+        assert_eq!(get_meta::<String>(&writer, "foo")?, None);
+        Ok(())
+    }
+}
diff --git a/third_party/rust/webext-storage/src/error.rs b/third_party/rust/webext-storage/src/error.rs
new file mode 100644
index 0000000000..1a2e650cd0
--- /dev/null
+++ b/third_party/rust/webext-storage/src/error.rs
@@ -0,0 +1,72 @@
+/* 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 interrupt_support::Interrupted;
+
+#[derive(Debug)]
+pub enum QuotaReason {
+    TotalBytes,
+    ItemBytes,
+    MaxItems,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ErrorKind {
+    #[error("Quota exceeded: {0:?}")]
+    QuotaError(QuotaReason),
+
+    #[error("Error parsing JSON data: {0}")]
+    JsonError(#[from] serde_json::Error),
+
+    #[error("Error executing SQL: {0}")]
+    SqlError(#[from] rusqlite::Error),
+
+    #[error("A connection of this type is already open")]
+    ConnectionAlreadyOpen,
+
+    #[error("An invalid connection type was specified")]
+    InvalidConnectionType,
+
+    #[error("IO error: {0}")]
+    IoError(#[from] std::io::Error),
+
+    #[error("Operation interrupted")]
+    InterruptedError(#[from] Interrupted),
+
+    #[error("Tried to close connection on wrong StorageApi instance")]
+    WrongApiForClose,
+
+    // This will happen if you provide something absurd like
+    // "/" or "" as your database path. For more subtley broken paths,
+    // we'll likely return an IoError.
+    #[error("Illegal database path: {0:?}")]
+    IllegalDatabasePath(std::path::PathBuf),
+
+    #[error("UTF8 Error: {0}")]
+    Utf8Error(#[from] std::str::Utf8Error),
+
+    #[error("Error opening database: {0}")]
+    OpenDatabaseError(#[from] sql_support::open_database::Error),
+
+    // When trying to close a connection but we aren't the exclusive owner of the containing Arc<>
+    #[error("Other shared references to this connection are alive")]
+    OtherConnectionReferencesExist,
+
+    #[error("The storage database has been closed")]
+    DatabaseConnectionClosed,
+
+    #[error("Sync Error: {0}")]
+    SyncError(String),
+}
+
+error_support::define_error! {
+    ErrorKind {
+        (JsonError, serde_json::Error),
+        (SqlError, rusqlite::Error),
+        (IoError, std::io::Error),
+        (InterruptedError, Interrupted),
+        (Utf8Error, std::str::Utf8Error),
+        (OpenDatabaseError, sql_support::open_database::Error),
+    }
+}
diff --git a/third_party/rust/webext-storage/src/ffi.rs b/third_party/rust/webext-storage/src/ffi.rs
new file mode 100644
index 0000000000..dc4af8ed6e
--- /dev/null
+++ b/third_party/rust/webext-storage/src/ffi.rs
@@ -0,0 +1,48 @@
+/* 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 module contains glue for the FFI, including error codes and a
+/// `From<Error>` implementation for `ffi_support::ExternError`. These must be
+/// implemented in this crate, and not `webext_storage_ffi`, because of Rust's
+/// orphan rule for implementing traits (`webext_storage_ffi` can't implement
+/// a trait in `ffi_support` for a type in `webext_storage`).
+use ffi_support::{ErrorCode, ExternError};
+
+use crate::error::{Error, ErrorKind, QuotaReason};
+
+mod error_codes {
+    /// An unexpected error occurred which likely cannot be meaningfully handled
+    /// by the application.
+    pub const UNEXPECTED: i32 = 1;
+
+    /// The application passed an invalid JSON string for a storage key or value.
+    pub const INVALID_JSON: i32 = 2;
+
+    /// The total number of bytes stored in the database for this extension,
+    /// counting all key-value pairs serialized to JSON, exceeds the allowed limit.
+    pub const QUOTA_TOTAL_BYTES_EXCEEDED: i32 = 32;
+
+    /// A single key-value pair exceeds the allowed byte limit when serialized
+    /// to JSON.
+    pub const QUOTA_ITEM_BYTES_EXCEEDED: i32 = 32 + 1;
+
+    /// The total number of key-value pairs stored for this extension exceeded the
+    /// allowed limit.
+    pub const QUOTA_MAX_ITEMS_EXCEEDED: i32 = 32 + 2;
+}
+
+impl From<Error> for ExternError {
+    fn from(err: Error) -> ExternError {
+        let code = ErrorCode::new(match err.kind() {
+            ErrorKind::JsonError(_) => error_codes::INVALID_JSON,
+            ErrorKind::QuotaError(QuotaReason::TotalBytes) => {
+                error_codes::QUOTA_TOTAL_BYTES_EXCEEDED
+            }
+            ErrorKind::QuotaError(QuotaReason::ItemBytes) => error_codes::QUOTA_ITEM_BYTES_EXCEEDED,
+            ErrorKind::QuotaError(QuotaReason::MaxItems) => error_codes::QUOTA_MAX_ITEMS_EXCEEDED,
+            _ => error_codes::UNEXPECTED,
+        });
+        ExternError::new_error(code, err.to_string())
+    }
+}
diff --git a/third_party/rust/webext-storage/src/lib.rs b/third_party/rust/webext-storage/src/lib.rs
new file mode 100644
index 0000000000..2e2554781d
--- /dev/null
+++ b/third_party/rust/webext-storage/src/lib.rs
@@ -0,0 +1,26 @@
+/* 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/. */
+
+#![allow(unknown_lints)]
+#![warn(rust_2018_idioms)]
+
+mod api;
+mod db;
+pub mod error;
+mod ffi;
+mod migration;
+mod schema;
+pub mod store;
+mod sync;
+
+pub use migration::MigrationInfo;
+
+// We publish some constants from non-public modules.
+pub use sync::STORAGE_VERSION;
+
+pub use api::SYNC_MAX_ITEMS;
+pub use api::SYNC_QUOTA_BYTES;
+pub use api::SYNC_QUOTA_BYTES_PER_ITEM;
+
+pub use api::UsageInfo;
diff --git a/third_party/rust/webext-storage/src/migration.rs b/third_party/rust/webext-storage/src/migration.rs
new file mode 100644
index 0000000000..27f230ef47
--- /dev/null
+++ b/third_party/rust/webext-storage/src/migration.rs
@@ -0,0 +1,454 @@
+/* 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 crate::error::*;
+use rusqlite::{named_params, Connection, OpenFlags, Transaction};
+use serde_json::{Map, Value};
+use sql_support::ConnExt;
+use std::collections::HashSet;
+use std::path::Path;
+
+// Simple migration from the "old" kinto-with-sqlite-backing implementation
+// to ours.
+// Could almost be trivially done in JS using the regular public API if not
+// for:
+// * We don't want to enforce the same quotas when migrating.
+// * We'd rather do the entire migration in a single transaction for perf
+//   reasons.
+
+// The sqlite database we migrate from has a very simple structure:
+// * table collection_data with columns collection_name, record_id and record
+// * `collection_name` is a string of form "default/{extension_id}"
+// * `record_id` is `key-{key}`
+// * `record` is a string with json, of form: {
+//     id: {the record id repeated},
+//     key: {the key},
+//     data: {the actual data},
+//     _status: {sync status},
+//     last_modified: {timestamp},
+//  }
+// So the info we need is stored somewhat redundantly.
+// Further:
+// * There's a special collection_name "default/storage-sync-crypto" that
+//   we don't want to migrate. Its record_id is 'keys' and its json has no
+//   `data`
+
+// Note we don't enforce a quota - we migrate everything - even if this means
+// it's too big for the server to store. This is a policy decision - it's better
+// to not lose data than to try and work out what data can be disposed of, as
+// the addon has the ability to determine this.
+
+// Our error strategy is "ignore read errors, propagate write errors" under the
+// assumption that the former tends to mean a damaged DB or file-system and is
+// unlikely to work if we try later (eg, replacing the disk isn't likely to
+// uncorrupt the DB), where the latter is likely to be disk-space or file-system
+// error, but retry might work (eg, replacing the disk then trying again might
+// make the writes work)
+
+// The struct we read from the DB.
+struct LegacyRow {
+    col_name: String, // collection_name column
+    record: String,   // record column
+}
+
+impl LegacyRow {
+    // Parse the 2 columns from the DB into the data we need to insert into
+    // our target database.
+    fn parse(&self) -> Option<Parsed<'_>> {
+        if self.col_name.len() < 8 {
+            log::trace!("collection_name of '{}' is too short", self.col_name);
+            return None;
+        }
+        if &self.col_name[..8] != "default/" {
+            log::trace!("collection_name of '{}' isn't ours", self.col_name);
+            return None;
+        }
+        let ext_id = &self.col_name[8..];
+        let mut record_map = match serde_json::from_str(&self.record) {
+            Ok(Value::Object(m)) => m,
+            Ok(o) => {
+                log::info!("skipping non-json-object 'record' column");
+                log::trace!("record value is json, but not an object: {}", o);
+                return None;
+            }
+            Err(e) => {
+                log::info!("skipping non-json 'record' column");
+                log::trace!("record value isn't json: {}", e);
+                return None;
+            }
+        };
+
+        let key = match record_map.remove("key") {
+            Some(Value::String(s)) if !s.is_empty() => s,
+            Some(o) => {
+                log::trace!("key is json but not a string: {}", o);
+                return None;
+            }
+            _ => {
+                log::trace!("key doesn't exist in the map");
+                return None;
+            }
+        };
+        let data = match record_map.remove("data") {
+            Some(d) => d,
+            _ => {
+                log::trace!("data doesn't exist in the map");
+                return None;
+            }
+        };
+        Some(Parsed { ext_id, key, data })
+    }
+}
+
+// The info we parse from the raw DB strings.
+struct Parsed<'a> {
+    ext_id: &'a str,
+    key: String,
+    data: serde_json::Value,
+}
+
+pub fn migrate(tx: &Transaction<'_>, filename: &Path) -> Result<MigrationInfo> {
+    // We do the grouping manually, collecting string values as we go.
+    let mut last_ext_id = "".to_string();
+    let mut curr_values: Vec<(String, serde_json::Value)> = Vec::new();
+    let (rows, mut mi) = read_rows(filename);
+    for row in rows {
+        log::trace!("processing '{}' - '{}'", row.col_name, row.record);
+        let parsed = match row.parse() {
+            Some(p) => p,
+            None => continue,
+        };
+        // Do our "grouping"
+        if parsed.ext_id != last_ext_id {
+            if !last_ext_id.is_empty() && !curr_values.is_empty() {
+                // a different extension id - write what we have to the DB.
+                let entries = do_insert(tx, &last_ext_id, curr_values)?;
+                mi.extensions_successful += 1;
+                mi.entries_successful += entries;
+            }
+            last_ext_id = parsed.ext_id.to_string();
+            curr_values = Vec::new();
+        }
+        // no 'else' here - must also enter this block on ext_id change.
+        if parsed.ext_id == last_ext_id {
+            curr_values.push((parsed.key.to_string(), parsed.data));
+            log::trace!(
+                "extension {} now has {} keys",
+                parsed.ext_id,
+                curr_values.len()
+            );
+        }
+    }
+    // and the last one
+    if !last_ext_id.is_empty() && !curr_values.is_empty() {
+        // a different extension id - write what we have to the DB.
+        let entries = do_insert(tx, &last_ext_id, curr_values)?;
+        mi.extensions_successful += 1;
+        mi.entries_successful += entries;
+    }
+    log::info!("migrated {} extensions: {:?}", mi.extensions_successful, mi);
+    Ok(mi)
+}
+
+fn read_rows(filename: &Path) -> (Vec<LegacyRow>, MigrationInfo) {
+    let flags = OpenFlags::SQLITE_OPEN_NO_MUTEX | OpenFlags::SQLITE_OPEN_READ_ONLY;
+    let src_conn = match Connection::open_with_flags(filename, flags) {
+        Ok(conn) => conn,
+        Err(e) => {
+            log::warn!("Failed to open the source DB: {}", e);
+            return (Vec::new(), MigrationInfo::open_failure());
+        }
+    };
+    // Failure to prepare the statement probably just means the source DB is
+    // damaged.
+    let mut stmt = match src_conn.prepare(
+        "SELECT collection_name, record FROM collection_data
+         WHERE collection_name != 'default/storage-sync-crypto'
+         ORDER BY collection_name",
+    ) {
+        Ok(stmt) => stmt,
+        Err(e) => {
+            log::warn!("Failed to prepare the statement: {}", e);
+            return (Vec::new(), MigrationInfo::open_failure());
+        }
+    };
+    let rows = match stmt.query_and_then([], |row| -> Result<LegacyRow> {
+        Ok(LegacyRow {
+            col_name: row.get(0)?,
+            record: row.get(1)?,
+        })
+    }) {
+        Ok(r) => r,
+        Err(e) => {
+            log::warn!("Failed to read any rows from the source DB: {}", e);
+            return (Vec::new(), MigrationInfo::open_failure());
+        }
+    };
+    let all_rows: Vec<Result<LegacyRow>> = rows.collect();
+    let entries = all_rows.len();
+    let successful_rows: Vec<LegacyRow> = all_rows.into_iter().filter_map(Result::ok).collect();
+    let distinct_extensions: HashSet<_> = successful_rows.iter().map(|c| &c.col_name).collect();
+
+    let mi = MigrationInfo {
+        entries,
+        extensions: distinct_extensions.len(),
+        // Populated later.
+        extensions_successful: 0,
+        entries_successful: 0,
+        open_failure: false,
+    };
+
+    (successful_rows, mi)
+}
+
+/// Insert the extension and values. If there are multiple values with the same
+/// key (which shouldn't be possible but who knows, database corruption causes
+/// strange things), chooses an arbitrary one. Returns the number of entries
+/// inserted, which could be different from `vals.len()` if multiple entries in
+/// `vals` have the same key.
+fn do_insert(tx: &Transaction<'_>, ext_id: &str, vals: Vec<(String, Value)>) -> Result<usize> {
+    let mut map = Map::with_capacity(vals.len());
+    for (key, val) in vals {
+        map.insert(key, val);
+    }
+    let num_entries = map.len();
+    tx.execute_cached(
+        "INSERT OR REPLACE INTO storage_sync_data(ext_id, data, sync_change_counter)
+         VALUES (:ext_id, :data, 1)",
+        rusqlite::named_params! {
+            ":ext_id": &ext_id,
+            ":data": &Value::Object(map),
+        },
+    )?;
+    Ok(num_entries)
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct MigrationInfo {
+    /// The number of entries (rows in the original table) we attempted to
+    /// migrate. Zero if there was some error in computing this number.
+    ///
+    /// Note that for the original table, a single row stores a single
+    /// preference for one extension. That is, if you view the set of
+    /// preferences for a given extension as a HashMap (as we do), it would be a
+    /// single entry/key-value-pair in the map.
+    pub entries: usize,
+    /// The number of records we successfully migrated (equal to `entries` for
+    /// entirely successful migrations).
+    pub entries_successful: usize,
+    /// The number of extensions (distinct extension ids) in the original
+    /// table.
+    pub extensions: usize,
+    /// The number of extensions we successfully migrated
+    pub extensions_successful: usize,
+    /// True iff we failed to open the source DB at all.
+    pub open_failure: bool,
+}
+
+impl MigrationInfo {
+    /// Returns a MigrationInfo indicating that we failed to read any rows due
+    /// to some error case (e.g. the database open failed, or some other very
+    /// early read error).
+    fn open_failure() -> Self {
+        Self {
+            open_failure: true,
+            ..Self::default()
+        }
+    }
+
+    const META_KEY: &'static str = "migration_info";
+
+    /// Store `self` in the provided database under `Self::META_KEY`.
+    pub(crate) fn store(&self, conn: &Connection) -> Result<()> {
+        let json = serde_json::to_string(self)?;
+        conn.execute(
+            "INSERT OR REPLACE INTO meta(key, value) VALUES (:k, :v)",
+            named_params! {
+                ":k": Self::META_KEY,
+                ":v": &json
+            },
+        )?;
+        Ok(())
+    }
+
+    /// Get the MigrationInfo stored under `Self::META_KEY` (if any) out of the
+    /// DB, and delete it.
+    pub(crate) fn take(tx: &Transaction<'_>) -> Result<Option<Self>> {
+        let s = tx.try_query_one::<String, _>(
+            "SELECT value FROM meta WHERE key = :k",
+            named_params! {
+                ":k": Self::META_KEY,
+            },
+            false,
+        )?;
+        tx.execute(
+            "DELETE FROM meta WHERE key = :k",
+            named_params! {
+                ":k": Self::META_KEY,
+            },
+        )?;
+        if let Some(s) = s {
+            match serde_json::from_str(&s) {
+                Ok(v) => Ok(Some(v)),
+                Err(e) => {
+                    // Force test failure, but just log an error otherwise so that
+                    // we commit the transaction that wil.
+                    debug_assert!(false, "Failed to read migration JSON: {:?}", e);
+                    error_support::report_error!(
+                        "webext-storage-migration-json",
+                        "Failed to read migration JSON: {}",
+                        e
+                    );
+                    Ok(None)
+                }
+            }
+        } else {
+            Ok(None)
+        }
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::api;
+    use crate::db::{test::new_mem_db, StorageDb};
+    use serde_json::json;
+    use tempfile::tempdir;
+
+    fn init_source_db(path: impl AsRef<Path>, f: impl FnOnce(&Connection)) {
+        let flags = OpenFlags::SQLITE_OPEN_NO_MUTEX
+            | OpenFlags::SQLITE_OPEN_CREATE
+            | OpenFlags::SQLITE_OPEN_READ_WRITE;
+        let mut conn = Connection::open_with_flags(path, flags).expect("open should work");
+        let tx = conn.transaction().expect("should be able to get a tx");
+        tx.execute_batch(
+            "CREATE TABLE collection_data (
+                collection_name TEXT,
+                record_id TEXT,
+                record TEXT
+            );",
+        )
+        .expect("create should work");
+        f(&tx);
+        tx.commit().expect("should commit");
+        conn.close().expect("close should work");
+    }
+
+    // Create a test database, populate it via the callback, migrate it, and
+    // return a connection to the new, migrated DB for further checking.
+    fn do_migrate<F>(expect_mi: MigrationInfo, f: F) -> StorageDb
+    where
+        F: FnOnce(&Connection),
+    {
+        let tmpdir = tempdir().unwrap();
+        let path = tmpdir.path().join("source.db");
+        init_source_db(path, f);
+
+        // now migrate
+        let mut db = new_mem_db();
+        let tx = db.transaction().expect("tx should work");
+
+        let mi = migrate(&tx, &tmpdir.path().join("source.db")).expect("migrate should work");
+        tx.commit().expect("should work");
+        assert_eq!(mi, expect_mi);
+        db
+    }
+
+    fn assert_has(c: &Connection, ext_id: &str, expect: Value) {
+        assert_eq!(
+            api::get(c, ext_id, json!(null)).expect("should get"),
+            expect
+        );
+    }
+
+    const HAPPY_PATH_SQL: &str = r#"
+        INSERT INTO collection_data(collection_name, record)
+        VALUES
+            ('default/{e7fefcf3-b39c-4f17-5215-ebfe120a7031}', '{"id":"key-userWelcomed","key":"userWelcomed","data":1570659224457,"_status":"synced","last_modified":1579755940527}'),
+            ('default/{e7fefcf3-b39c-4f17-5215-ebfe120a7031}', '{"id":"key-isWho","key":"isWho","data":"4ec8109f","_status":"synced","last_modified":1579755940497}'),
+            ('default/storage-sync-crypto', '{"id":"keys","keys":{"default":["rQ=","lR="],"collections":{"extension@redux.devtools":["Bd=","ju="]}}}'),
+            ('default/https-everywhere@eff.org', '{"id":"key-userRules","key":"userRules","data":[],"_status":"synced","last_modified":1570079920045}'),
+            ('default/https-everywhere@eff.org', '{"id":"key-ruleActiveStates","key":"ruleActiveStates","data":{},"_status":"synced","last_modified":1570079919993}'),
+            ('default/https-everywhere@eff.org', '{"id":"key-migration_5F_version","key":"migration_version","data":2,"_status":"synced","last_modified":1570079919966}')
+    "#;
+    const HAPPY_PATH_MIGRATION_INFO: MigrationInfo = MigrationInfo {
+        entries: 5,
+        entries_successful: 5,
+        extensions: 2,
+        extensions_successful: 2,
+        open_failure: false,
+    };
+
+    #[allow(clippy::unreadable_literal)]
+    #[test]
+    fn test_happy_paths() {
+        // some real data.
+        let conn = do_migrate(HAPPY_PATH_MIGRATION_INFO, |c| {
+            c.execute_batch(HAPPY_PATH_SQL).expect("should populate")
+        });
+
+        assert_has(
+            &conn,
+            "{e7fefcf3-b39c-4f17-5215-ebfe120a7031}",
+            json!({"userWelcomed": 1570659224457u64, "isWho": "4ec8109f"}),
+        );
+        assert_has(
+            &conn,
+            "https-everywhere@eff.org",
+            json!({"userRules": [], "ruleActiveStates": {}, "migration_version": 2}),
+        );
+    }
+
+    #[test]
+    fn test_sad_paths() {
+        do_migrate(
+            MigrationInfo {
+                entries: 10,
+                entries_successful: 0,
+                extensions: 6,
+                extensions_successful: 0,
+                open_failure: false,
+            },
+            |c| {
+                c.execute_batch(
+                    r#"INSERT INTO collection_data(collection_name, record)
+                    VALUES
+                    ('default/test', '{"key":2,"data":1}'), -- key not a string
+                    ('default/test', '{"key":"","data":1}'), -- key empty string
+                    ('default/test', '{"xey":"k","data":1}'), -- key missing
+                    ('default/test', '{"key":"k","xata":1}'), -- data missing
+                    ('default/test', '{"key":"k","data":1'), -- invalid json
+                    ('xx/test', '{"key":"k","data":1}'), -- bad key format
+                    ('default', '{"key":"k","data":1}'), -- bad key format 2
+                    ('default/', '{"key":"k","data":1}'), -- bad key format 3
+                    ('defaultx/test', '{"key":"k","data":1}'), -- bad key format 4
+                    ('', '') -- empty strings
+                    "#,
+                )
+                .expect("should populate");
+            },
+        );
+    }
+
+    #[test]
+    fn test_migration_info_storage() {
+        let tmpdir = tempdir().unwrap();
+        let path = tmpdir.path().join("source.db");
+        init_source_db(&path, |c| {
+            c.execute_batch(HAPPY_PATH_SQL).expect("should populate")
+        });
+
+        // now migrate
+        let db = crate::store::test::new_mem_store();
+        db.migrate(&path).expect("migration should work");
+        let mi = db
+            .take_migration_info()
+            .expect("take failed with info present");
+        assert_eq!(mi, Some(HAPPY_PATH_MIGRATION_INFO));
+        let mi2 = db
+            .take_migration_info()
+            .expect("take failed with info missing");
+        assert_eq!(mi2, None);
+    }
+}
diff --git a/third_party/rust/webext-storage/src/schema.rs b/third_party/rust/webext-storage/src/schema.rs
new file mode 100644
index 0000000000..59efcf495a
--- /dev/null
+++ b/third_party/rust/webext-storage/src/schema.rs
@@ -0,0 +1,213 @@
+/* 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 crate::db::sql_fns;
+use crate::error::Result;
+use rusqlite::{Connection, Transaction};
+use sql_support::open_database::{
+    ConnectionInitializer as MigrationLogic, Error as MigrationError, Result as MigrationResult,
+};
+
+const CREATE_SCHEMA_SQL: &str = include_str!("../sql/create_schema.sql");
+const CREATE_SYNC_TEMP_TABLES_SQL: &str = include_str!("../sql/create_sync_temp_tables.sql");
+
+pub struct WebExtMigrationLogin;
+
+impl MigrationLogic for WebExtMigrationLogin {
+    const NAME: &'static str = "webext storage db";
+    const END_VERSION: u32 = 2;
+
+    fn prepare(&self, conn: &Connection, _db_empty: bool) -> MigrationResult<()> {
+        let initial_pragmas = "
+            -- We don't care about temp tables being persisted to disk.
+            PRAGMA temp_store = 2;
+            -- we unconditionally want write-ahead-logging mode
+            PRAGMA journal_mode=WAL;
+            -- foreign keys seem worth enforcing!
+            PRAGMA foreign_keys = ON;
+        ";
+        conn.execute_batch(initial_pragmas)?;
+        define_functions(conn)?;
+        conn.set_prepared_statement_cache_capacity(128);
+        Ok(())
+    }
+
+    fn init(&self, db: &Transaction<'_>) -> MigrationResult<()> {
+        log::debug!("Creating schema");
+        db.execute_batch(CREATE_SCHEMA_SQL)?;
+        Ok(())
+    }
+
+    fn upgrade_from(&self, db: &Transaction<'_>, version: u32) -> MigrationResult<()> {
+        match version {
+            1 => upgrade_from_1(db),
+            _ => Err(MigrationError::IncompatibleVersion(version)),
+        }
+    }
+}
+
+fn define_functions(c: &Connection) -> MigrationResult<()> {
+    use rusqlite::functions::FunctionFlags;
+    c.create_scalar_function(
+        "generate_guid",
+        0,
+        FunctionFlags::SQLITE_UTF8,
+        sql_fns::generate_guid,
+    )?;
+    Ok(())
+}
+
+fn upgrade_from_1(db: &Connection) -> MigrationResult<()> {
+    // We changed a not null constraint
+    db.execute_batch("ALTER TABLE storage_sync_mirror RENAME TO old_mirror;")?;
+    // just re-run the full schema commands to recreate the able.
+    db.execute_batch(CREATE_SCHEMA_SQL)?;
+    db.execute_batch(
+        "INSERT OR IGNORE INTO storage_sync_mirror(guid, ext_id, data)
+         SELECT guid, ext_id, data FROM old_mirror;",
+    )?;
+    db.execute_batch("DROP TABLE old_mirror;")?;
+    db.execute_batch("PRAGMA user_version = 2;")?;
+    Ok(())
+}
+
+// Note that we expect this to be called before and after a sync - before to
+// ensure we are syncing with a clean state, after to be good memory citizens
+// given the temp tables are in memory.
+pub fn create_empty_sync_temp_tables(db: &Connection) -> Result<()> {
+    log::debug!("Initializing sync temp tables");
+    db.execute_batch(CREATE_SYNC_TEMP_TABLES_SQL)?;
+    Ok(())
+}
+
+#[cfg(test)]
+pub mod test {
+    use prettytable::{Cell, Row};
+    use rusqlite::Result as RusqliteResult;
+    use rusqlite::{types::Value, Connection};
+
+    // To help debugging tests etc.
+    #[allow(unused)]
+    pub fn print_table(conn: &Connection, table_name: &str) -> RusqliteResult<()> {
+        let mut stmt = conn.prepare(&format!("SELECT * FROM {}", table_name))?;
+        let mut rows = stmt.query([])?;
+        let mut table = prettytable::Table::new();
+        let mut titles = Row::empty();
+        for col in rows.as_ref().expect("must have statement").columns() {
+            titles.add_cell(Cell::new(col.name()));
+        }
+        table.set_titles(titles);
+        while let Some(sql_row) = rows.next()? {
+            let mut table_row = Row::empty();
+            for i in 0..sql_row.as_ref().column_count() {
+                let val = match sql_row.get::<_, Value>(i)? {
+                    Value::Null => "null".to_string(),
+                    Value::Integer(i) => i.to_string(),
+                    Value::Real(f) => f.to_string(),
+                    Value::Text(s) => s,
+                    Value::Blob(b) => format!("<blob with {} bytes>", b.len()),
+                };
+                table_row.add_cell(Cell::new(&val));
+            }
+            table.add_row(table_row);
+        }
+        table.printstd();
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db::test::new_mem_db;
+    use rusqlite::Error;
+    use sql_support::open_database::test_utils::MigratedDatabaseFile;
+    use sql_support::ConnExt;
+
+    const CREATE_SCHEMA_V1_SQL: &str = include_str!("../sql/tests/create_schema_v1.sql");
+
+    #[test]
+    fn test_create_schema_twice() {
+        let db = new_mem_db();
+        db.execute_batch(CREATE_SCHEMA_SQL)
+            .expect("should allow running twice");
+    }
+
+    #[test]
+    fn test_create_empty_sync_temp_tables_twice() {
+        let db = new_mem_db();
+        create_empty_sync_temp_tables(&db).expect("should work first time");
+        // insert something into our new temp table and check it's there.
+        db.execute_batch(
+            "INSERT INTO temp.storage_sync_staging
+                            (guid, ext_id) VALUES
+                            ('guid', 'ext_id');",
+        )
+        .expect("should work once");
+        let count = db
+            .query_row_and_then(
+                "SELECT COUNT(*) FROM temp.storage_sync_staging;",
+                [],
+                |row| row.get::<_, u32>(0),
+            )
+            .expect("query should work");
+        assert_eq!(count, 1, "should be one row");
+
+        // re-execute
+        create_empty_sync_temp_tables(&db).expect("should second first time");
+        // and it should have deleted existing data.
+        let count = db
+            .query_row_and_then(
+                "SELECT COUNT(*) FROM temp.storage_sync_staging;",
+                [],
+                |row| row.get::<_, u32>(0),
+            )
+            .expect("query should work");
+        assert_eq!(count, 0, "should be no rows");
+    }
+
+    #[test]
+    fn test_all_upgrades() -> Result<()> {
+        let db_file = MigratedDatabaseFile::new(WebExtMigrationLogin, CREATE_SCHEMA_V1_SQL);
+        db_file.run_all_upgrades();
+        let db = db_file.open();
+
+        let get_id_data = |guid: &str| -> Result<(Option<String>, Option<String>)> {
+            let (ext_id, data) = db
+                .try_query_row::<_, Error, _, _>(
+                    "SELECT ext_id, data FROM storage_sync_mirror WHERE guid = :guid",
+                    &[(":guid", &guid.to_string())],
+                    |row| Ok((row.get(0)?, row.get(1)?)),
+                    true,
+                )?
+                .expect("row should exist.");
+            Ok((ext_id, data))
+        };
+        assert_eq!(
+            get_id_data("guid-1")?,
+            (Some("ext-id-1".to_string()), Some("data-1".to_string()))
+        );
+        assert_eq!(
+            get_id_data("guid-2")?,
+            (Some("ext-id-2".to_string()), Some("data-2".to_string()))
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_upgrade_2() -> Result<()> {
+        let _ = env_logger::try_init();
+
+        let db_file = MigratedDatabaseFile::new(WebExtMigrationLogin, CREATE_SCHEMA_V1_SQL);
+        db_file.upgrade_to(2);
+        let db = db_file.open();
+
+        // Should be able to insert a new with a NULL ext_id
+        db.execute_batch(
+            "INSERT INTO storage_sync_mirror(guid, ext_id, data)
+             VALUES ('guid-3', NULL, NULL);",
+        )?;
+        Ok(())
+    }
+}
diff --git a/third_party/rust/webext-storage/src/store.rs b/third_party/rust/webext-storage/src/store.rs
new file mode 100644
index 0000000000..043ec5a11c
--- /dev/null
+++ b/third_party/rust/webext-storage/src/store.rs
@@ -0,0 +1,218 @@
+/* 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 crate::api::{self, StorageChanges};
+use crate::db::{StorageDb, ThreadSafeStorageDb};
+use crate::error::*;
+use crate::migration::{migrate, MigrationInfo};
+use crate::sync;
+use std::path::Path;
+use std::sync::Arc;
+
+use interrupt_support::SqlInterruptHandle;
+use serde_json::Value as JsonValue;
+
+/// A store is used to access `storage.sync` data. It manages an underlying
+/// database connection, and exposes methods for reading and writing storage
+/// items scoped to an extension ID. Each item is a JSON object, with one or
+/// more string keys, and values of any type that can serialize to JSON.
+///
+/// An application should create only one store, and manage the instance as a
+/// singleton. While this isn't enforced, if you make multiple stores pointing
+/// to the same database file, you are going to have a bad time: each store will
+/// create its own database connection, using up extra memory and CPU cycles,
+/// and causing write contention. For this reason, you should only call
+/// `Store::new()` (or `webext_store_new()`, from the FFI) once.
+///
+/// Note that our Db implementation is behind an Arc<> because we share that
+/// connection with our sync engines - ie, these engines also hold an Arc<>
+/// around the same object.
+pub struct Store {
+    db: Arc<ThreadSafeStorageDb>,
+}
+
+impl Store {
+    /// Creates a store backed by a database at `db_path`. The path can be a
+    /// file path or `file:` URI.
+    pub fn new(db_path: impl AsRef<Path>) -> Result<Self> {
+        let db = StorageDb::new(db_path)?;
+        Ok(Self {
+            db: Arc::new(ThreadSafeStorageDb::new(db)),
+        })
+    }
+
+    /// Creates a store backed by an in-memory database.
+    #[cfg(test)]
+    pub fn new_memory(db_path: &str) -> Result<Self> {
+        let db = StorageDb::new_memory(db_path)?;
+        Ok(Self {
+            db: Arc::new(ThreadSafeStorageDb::new(db)),
+        })
+    }
+
+    /// Returns an interrupt handle for this store.
+    pub fn interrupt_handle(&self) -> Arc<SqlInterruptHandle> {
+        self.db.interrupt_handle()
+    }
+
+    /// Sets one or more JSON key-value pairs for an extension ID. Returns a
+    /// list of changes, with existing and new values for each key in `val`.
+    pub fn set(&self, ext_id: &str, val: JsonValue) -> Result<StorageChanges> {
+        let db = self.db.lock();
+        let tx = db.unchecked_transaction()?;
+        let result = api::set(&tx, ext_id, val)?;
+        tx.commit()?;
+        Ok(result)
+    }
+
+    /// Returns information about per-extension usage
+    pub fn usage(&self) -> Result<Vec<crate::UsageInfo>> {
+        let db = self.db.lock();
+        api::usage(&db)
+    }
+
+    /// Returns the values for one or more keys `keys` can be:
+    ///
+    /// - `null`, in which case all key-value pairs for the extension are
+    ///   returned, or an empty object if the extension doesn't have any
+    ///   stored data.
+    /// - A single string key, in which case an object with only that key
+    ///   and its value is returned, or an empty object if the key doesn't
+    //    exist.
+    /// - An array of string keys, in which case an object with only those
+    ///   keys and their values is returned. Any keys that don't exist will be
+    ///   omitted.
+    /// - An object where the property names are keys, and each value is the
+    ///   default value to return if the key doesn't exist.
+    ///
+    /// This method always returns an object (that is, a
+    /// `serde_json::Value::Object`).
+    pub fn get(&self, ext_id: &str, keys: JsonValue) -> Result<JsonValue> {
+        // Don't care about transactions here.
+        let db = self.db.lock();
+        api::get(&db, ext_id, keys)
+    }
+
+    /// Deletes the values for one or more keys. As with `get`, `keys` can be
+    /// either a single string key, or an array of string keys. Returns a list
+    /// of changes, where each change contains the old value for each deleted
+    /// key.
+    pub fn remove(&self, ext_id: &str, keys: JsonValue) -> Result<StorageChanges> {
+        let db = self.db.lock();
+        let tx = db.unchecked_transaction()?;
+        let result = api::remove(&tx, ext_id, keys)?;
+        tx.commit()?;
+        Ok(result)
+    }
+
+    /// Deletes all key-value pairs for the extension. As with `remove`, returns
+    /// a list of changes, where each change contains the old value for each
+    /// deleted key.
+    pub fn clear(&self, ext_id: &str) -> Result<StorageChanges> {
+        let db = self.db.lock();
+        let tx = db.unchecked_transaction()?;
+        let result = api::clear(&tx, ext_id)?;
+        tx.commit()?;
+        Ok(result)
+    }
+
+    /// Returns the bytes in use for the specified items (which can be null,
+    /// a string, or an array)
+    pub fn get_bytes_in_use(&self, ext_id: &str, keys: JsonValue) -> Result<usize> {
+        let db = self.db.lock();
+        api::get_bytes_in_use(&db, ext_id, keys)
+    }
+
+    /// Returns a bridged sync engine for Desktop for this store.
+    pub fn bridged_engine(&self) -> sync::BridgedEngine {
+        sync::BridgedEngine::new(&self.db)
+    }
+
+    /// Closes the store and its database connection. See the docs for
+    /// `StorageDb::close` for more details on when this can fail.
+    pub fn close(self) -> Result<()> {
+        // Even though this consumes `self`, the fact we use an Arc<> means
+        // we can't guarantee we can actually consume the inner DB - so do
+        // the best we can.
+        let shared: ThreadSafeStorageDb = match Arc::try_unwrap(self.db) {
+            Ok(shared) => shared,
+            _ => {
+                // The only way this is possible is if the sync engine has an operation
+                // running - but that shouldn't be possible in practice because desktop
+                // uses a single "task queue" such that the close operation can't possibly
+                // be running concurrently with any sync or storage tasks.
+
+                // If this *could* get hit, rusqlite will attempt to close the DB connection
+                // as it is dropped, and if that close fails, then rusqlite 0.28.0 and earlier
+                // would panic - but even that only happens if prepared statements are
+                // not finalized, which ruqlite also does.
+
+                // tl;dr - this should be impossible. If it was possible, rusqlite might panic,
+                // but we've never seen it panic in practice other places we don't close
+                // connections, and the next rusqlite version will not panic anyway.
+                // So this-is-fine.jpg
+                log::warn!("Attempting to close a store while other DB references exist.");
+                return Err(ErrorKind::OtherConnectionReferencesExist.into());
+            }
+        };
+        // consume the mutex and get back the inner.
+        let db = shared.into_inner();
+        db.close()
+    }
+
+    /// Gets the changes which the current sync applied. Should be used
+    /// immediately after the bridged engine is told to apply incoming changes,
+    /// and can be used to notify observers of the StorageArea of the changes
+    /// that were applied.
+    /// The result is a Vec of already JSON stringified changes.
+    pub fn get_synced_changes(&self) -> Result<Vec<sync::SyncedExtensionChange>> {
+        let db = self.db.lock();
+        sync::get_synced_changes(&db)
+    }
+
+    /// Migrates data from a database in the format of the "old" kinto
+    /// implementation. Information about how the migration went is stored in
+    /// the database, and can be read using `Self::take_migration_info`.
+    ///
+    /// Note that `filename` isn't normalized or canonicalized.
+    pub fn migrate(&self, filename: impl AsRef<Path>) -> Result<()> {
+        let db = self.db.lock();
+        let tx = db.unchecked_transaction()?;
+        let result = migrate(&tx, filename.as_ref())?;
+        tx.commit()?;
+        // Failing to store this information should not cause migration failure.
+        if let Err(e) = result.store(&db) {
+            debug_assert!(false, "Migration error: {:?}", e);
+            log::warn!("Failed to record migration telmetry: {}", e);
+        }
+        Ok(())
+    }
+
+    /// Read-and-delete (e.g. `take` in rust parlance, see Option::take)
+    /// operation for any MigrationInfo stored in this database.
+    pub fn take_migration_info(&self) -> Result<Option<MigrationInfo>> {
+        let db = self.db.lock();
+        let tx = db.unchecked_transaction()?;
+        let result = MigrationInfo::take(&tx)?;
+        tx.commit()?;
+        Ok(result)
+    }
+}
+
+#[cfg(test)]
+pub mod test {
+    use super::*;
+    #[test]
+    fn test_send() {
+        fn ensure_send<T: Send>() {}
+        // Compile will fail if not send.
+        ensure_send::<Store>();
+    }
+
+    pub fn new_mem_store() -> Store {
+        Store {
+            db: Arc::new(ThreadSafeStorageDb::new(crate::db::test::new_mem_db())),
+        }
+    }
+}
diff --git a/third_party/rust/webext-storage/src/sync/bridge.rs b/third_party/rust/webext-storage/src/sync/bridge.rs
new file mode 100644
index 0000000000..760a0d0e10
--- /dev/null
+++ b/third_party/rust/webext-storage/src/sync/bridge.rs
@@ -0,0 +1,388 @@
+/* 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 anyhow::Result;
+use rusqlite::Transaction;
+use std::sync::{Arc, Weak};
+use sync15::bso::IncomingBso;
+use sync15::engine::ApplyResults;
+use sync_guid::Guid as SyncGuid;
+
+use crate::db::{delete_meta, get_meta, put_meta, ThreadSafeStorageDb};
+use crate::schema;
+use crate::sync::incoming::{apply_actions, get_incoming, plan_incoming, stage_incoming};
+use crate::sync::outgoing::{get_outgoing, record_uploaded, stage_outgoing};
+
+const LAST_SYNC_META_KEY: &str = "last_sync_time";
+const SYNC_ID_META_KEY: &str = "sync_id";
+
+/// A bridged engine implements all the methods needed to make the
+/// `storage.sync` store work with Desktop's Sync implementation.
+/// Conceptually, it's similar to `sync15::Store`, which we
+/// should eventually rename and unify with this trait (#2841).
+///
+/// Unlike most of our other implementation which hold a strong reference
+/// to the store, this engine keeps a weak reference in an attempt to keep
+/// the desktop semantics as close as possible to what they were when the
+/// engines all took lifetime params to ensure they don't outlive the store.
+pub struct BridgedEngine {
+    db: Weak<ThreadSafeStorageDb>,
+}
+
+impl BridgedEngine {
+    /// Creates a bridged engine for syncing.
+    pub fn new(db: &Arc<ThreadSafeStorageDb>) -> Self {
+        BridgedEngine {
+            db: Arc::downgrade(db),
+        }
+    }
+
+    fn do_reset(&self, tx: &Transaction<'_>) -> Result<()> {
+        tx.execute_batch(
+            "DELETE FROM storage_sync_mirror;
+             UPDATE storage_sync_data SET sync_change_counter = 1;",
+        )?;
+        delete_meta(tx, LAST_SYNC_META_KEY)?;
+        Ok(())
+    }
+
+    fn thread_safe_storage_db(&self) -> Result<Arc<ThreadSafeStorageDb>> {
+        self.db
+            .upgrade()
+            .ok_or_else(|| crate::error::ErrorKind::DatabaseConnectionClosed.into())
+    }
+}
+
+impl sync15::engine::BridgedEngine for BridgedEngine {
+    fn last_sync(&self) -> Result<i64> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        Ok(get_meta(&db, LAST_SYNC_META_KEY)?.unwrap_or(0))
+    }
+
+    fn set_last_sync(&self, last_sync_millis: i64) -> Result<()> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        put_meta(&db, LAST_SYNC_META_KEY, &last_sync_millis)?;
+        Ok(())
+    }
+
+    fn sync_id(&self) -> Result<Option<String>> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        Ok(get_meta(&db, SYNC_ID_META_KEY)?)
+    }
+
+    fn reset_sync_id(&self) -> Result<String> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        let tx = db.unchecked_transaction()?;
+        let new_id = SyncGuid::random().to_string();
+        self.do_reset(&tx)?;
+        put_meta(&tx, SYNC_ID_META_KEY, &new_id)?;
+        tx.commit()?;
+        Ok(new_id)
+    }
+
+    fn ensure_current_sync_id(&self, sync_id: &str) -> Result<String> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        let current: Option<String> = get_meta(&db, SYNC_ID_META_KEY)?;
+        Ok(match current {
+            Some(current) if current == sync_id => current,
+            _ => {
+                let tx = db.unchecked_transaction()?;
+                self.do_reset(&tx)?;
+                let result = sync_id.to_string();
+                put_meta(&tx, SYNC_ID_META_KEY, &result)?;
+                tx.commit()?;
+                result
+            }
+        })
+    }
+
+    fn sync_started(&self) -> Result<()> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        schema::create_empty_sync_temp_tables(&db)?;
+        Ok(())
+    }
+
+    fn store_incoming(&self, incoming_bsos: Vec<IncomingBso>) -> Result<()> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        let signal = db.begin_interrupt_scope()?;
+        let tx = db.unchecked_transaction()?;
+        let incoming_content: Vec<_> = incoming_bsos
+            .into_iter()
+            .map(IncomingBso::into_content::<super::WebextRecord>)
+            .collect();
+        stage_incoming(&tx, &incoming_content, &signal)?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    fn apply(&self) -> Result<ApplyResults> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        let signal = db.begin_interrupt_scope()?;
+
+        let tx = db.unchecked_transaction()?;
+        let incoming = get_incoming(&tx)?;
+        let actions = incoming
+            .into_iter()
+            .map(|(item, state)| (item, plan_incoming(state)))
+            .collect();
+        apply_actions(&tx, actions, &signal)?;
+        stage_outgoing(&tx)?;
+        tx.commit()?;
+
+        Ok(get_outgoing(&db, &signal)?.into())
+    }
+
+    fn set_uploaded(&self, _server_modified_millis: i64, ids: &[SyncGuid]) -> Result<()> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        let signal = db.begin_interrupt_scope()?;
+        let tx = db.unchecked_transaction()?;
+        record_uploaded(&tx, ids, &signal)?;
+        tx.commit()?;
+
+        Ok(())
+    }
+
+    fn sync_finished(&self) -> Result<()> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        schema::create_empty_sync_temp_tables(&db)?;
+        Ok(())
+    }
+
+    fn reset(&self) -> Result<()> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        let tx = db.unchecked_transaction()?;
+        self.do_reset(&tx)?;
+        delete_meta(&tx, SYNC_ID_META_KEY)?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    fn wipe(&self) -> Result<()> {
+        let shared_db = self.thread_safe_storage_db()?;
+        let db = shared_db.lock();
+        let tx = db.unchecked_transaction()?;
+        // We assume the meta table is only used by sync.
+        tx.execute_batch(
+            "DELETE FROM storage_sync_data; DELETE FROM storage_sync_mirror; DELETE FROM meta;",
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+}
+
+impl From<anyhow::Error> for crate::error::Error {
+    fn from(value: anyhow::Error) -> Self {
+        crate::error::ErrorKind::SyncError(value.to_string()).into()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db::test::new_mem_thread_safe_storage_db;
+    use crate::db::StorageDb;
+    use sync15::engine::BridgedEngine;
+
+    fn query_count(conn: &StorageDb, table: &str) -> u32 {
+        conn.query_row_and_then(&format!("SELECT COUNT(*) FROM {};", table), [], |row| {
+            row.get::<_, u32>(0)
+        })
+        .expect("should work")
+    }
+
+    // Sets up mock data for the tests here.
+    fn setup_mock_data(engine: &super::BridgedEngine) -> Result<()> {
+        {
+            let shared = engine.thread_safe_storage_db()?;
+            let db = shared.lock();
+            db.execute(
+                "INSERT INTO storage_sync_data (ext_id, data, sync_change_counter)
+                    VALUES ('ext-a', 'invalid-json', 2)",
+                [],
+            )?;
+            db.execute(
+                "INSERT INTO storage_sync_mirror (guid, ext_id, data)
+                    VALUES ('guid', 'ext-a', '3')",
+                [],
+            )?;
+        }
+        engine.set_last_sync(1)?;
+
+        let shared = engine.thread_safe_storage_db()?;
+        let db = shared.lock();
+        // and assert we wrote what we think we did.
+        assert_eq!(query_count(&db, "storage_sync_data"), 1);
+        assert_eq!(query_count(&db, "storage_sync_mirror"), 1);
+        assert_eq!(query_count(&db, "meta"), 1);
+        Ok(())
+    }
+
+    // Assuming a DB setup with setup_mock_data, assert it was correctly reset.
+    fn assert_reset(engine: &super::BridgedEngine) -> Result<()> {
+        // A reset never wipes data...
+        let shared = engine.thread_safe_storage_db()?;
+        let db = shared.lock();
+        assert_eq!(query_count(&db, "storage_sync_data"), 1);
+
+        // But did reset the change counter.
+        let cc = db.query_row_and_then(
+            "SELECT sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext-a';",
+            [],
+            |row| row.get::<_, u32>(0),
+        )?;
+        assert_eq!(cc, 1);
+        // But did wipe the mirror...
+        assert_eq!(query_count(&db, "storage_sync_mirror"), 0);
+        // And the last_sync should have been wiped.
+        assert!(get_meta::<i64>(&db, LAST_SYNC_META_KEY)?.is_none());
+        Ok(())
+    }
+
+    // Assuming a DB setup with setup_mock_data, assert it has not been reset.
+    fn assert_not_reset(engine: &super::BridgedEngine) -> Result<()> {
+        let shared = engine.thread_safe_storage_db()?;
+        let db = shared.lock();
+        assert_eq!(query_count(&db, "storage_sync_data"), 1);
+        let cc = db.query_row_and_then(
+            "SELECT sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext-a';",
+            [],
+            |row| row.get::<_, u32>(0),
+        )?;
+        assert_eq!(cc, 2);
+        assert_eq!(query_count(&db, "storage_sync_mirror"), 1);
+        // And the last_sync should remain.
+        assert!(get_meta::<i64>(&db, LAST_SYNC_META_KEY)?.is_some());
+        Ok(())
+    }
+
+    #[test]
+    fn test_wipe() -> Result<()> {
+        let strong = new_mem_thread_safe_storage_db();
+        let engine = super::BridgedEngine::new(&strong);
+
+        setup_mock_data(&engine)?;
+
+        engine.wipe()?;
+
+        let shared = engine.thread_safe_storage_db()?;
+        let db = shared.lock();
+
+        assert_eq!(query_count(&db, "storage_sync_data"), 0);
+        assert_eq!(query_count(&db, "storage_sync_mirror"), 0);
+        assert_eq!(query_count(&db, "meta"), 0);
+        Ok(())
+    }
+
+    #[test]
+    fn test_reset() -> Result<()> {
+        let strong = new_mem_thread_safe_storage_db();
+        let engine = super::BridgedEngine::new(&strong);
+
+        setup_mock_data(&engine)?;
+        put_meta(
+            &engine.thread_safe_storage_db()?.lock(),
+            SYNC_ID_META_KEY,
+            &"sync-id".to_string(),
+        )?;
+
+        engine.reset()?;
+        assert_reset(&engine)?;
+        // Only an explicit reset kills the sync-id, so check that here.
+        assert_eq!(
+            get_meta::<String>(&engine.thread_safe_storage_db()?.lock(), SYNC_ID_META_KEY)?,
+            None
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_ensure_missing_sync_id() -> Result<()> {
+        let strong = new_mem_thread_safe_storage_db();
+        let engine = super::BridgedEngine::new(&strong);
+
+        setup_mock_data(&engine)?;
+
+        assert_eq!(engine.sync_id()?, None);
+        // We don't have a sync ID - so setting one should reset.
+        engine.ensure_current_sync_id("new-id")?;
+        // should have cause a reset.
+        assert_reset(&engine)?;
+        Ok(())
+    }
+
+    #[test]
+    fn test_ensure_new_sync_id() -> Result<()> {
+        let strong = new_mem_thread_safe_storage_db();
+        let engine = super::BridgedEngine::new(&strong);
+
+        setup_mock_data(&engine)?;
+
+        put_meta(
+            &engine.thread_safe_storage_db()?.lock(),
+            SYNC_ID_META_KEY,
+            &"old-id".to_string(),
+        )?;
+        assert_not_reset(&engine)?;
+        assert_eq!(engine.sync_id()?, Some("old-id".to_string()));
+
+        engine.ensure_current_sync_id("new-id")?;
+        // should have cause a reset.
+        assert_reset(&engine)?;
+        // should have the new id.
+        assert_eq!(engine.sync_id()?, Some("new-id".to_string()));
+        Ok(())
+    }
+
+    #[test]
+    fn test_ensure_same_sync_id() -> Result<()> {
+        let strong = new_mem_thread_safe_storage_db();
+        let engine = super::BridgedEngine::new(&strong);
+
+        setup_mock_data(&engine)?;
+        assert_not_reset(&engine)?;
+
+        put_meta(
+            &engine.thread_safe_storage_db()?.lock(),
+            SYNC_ID_META_KEY,
+            &"sync-id".to_string(),
+        )?;
+
+        engine.ensure_current_sync_id("sync-id")?;
+        // should not have reset.
+        assert_not_reset(&engine)?;
+        Ok(())
+    }
+
+    #[test]
+    fn test_reset_sync_id() -> Result<()> {
+        let strong = new_mem_thread_safe_storage_db();
+        let engine = super::BridgedEngine::new(&strong);
+
+        setup_mock_data(&engine)?;
+        put_meta(
+            &engine.thread_safe_storage_db()?.lock(),
+            SYNC_ID_META_KEY,
+            &"sync-id".to_string(),
+        )?;
+
+        assert_eq!(engine.sync_id()?, Some("sync-id".to_string()));
+        let new_id = engine.reset_sync_id()?;
+        // should have cause a reset.
+        assert_reset(&engine)?;
+        assert_eq!(engine.sync_id()?, Some(new_id));
+        Ok(())
+    }
+}
diff --git a/third_party/rust/webext-storage/src/sync/incoming.rs b/third_party/rust/webext-storage/src/sync/incoming.rs
new file mode 100644
index 0000000000..5d00fe8de3
--- /dev/null
+++ b/third_party/rust/webext-storage/src/sync/incoming.rs
@@ -0,0 +1,863 @@
+/* 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/. */
+
+// The "incoming" part of syncing - handling the incoming rows, staging them,
+// working out a plan for them, updating the local data and mirror, etc.
+
+use interrupt_support::Interruptee;
+use rusqlite::{Connection, Row, Transaction};
+use sql_support::ConnExt;
+use sync15::bso::{IncomingContent, IncomingKind};
+use sync_guid::Guid as SyncGuid;
+
+use crate::api::{StorageChanges, StorageValueChange};
+use crate::error::*;
+
+use super::{merge, remove_matching_keys, JsonMap, WebextRecord};
+
+/// The state data can be in. Could be represented as Option<JsonMap>, but this
+/// is clearer and independent of how the data is stored.
+#[derive(Debug, PartialEq, Eq)]
+pub enum DataState {
+    /// The data was deleted.
+    Deleted,
+    /// Data exists, as stored in the map.
+    Exists(JsonMap),
+}
+
+// A little helper to create a StorageChanges object when we are creating
+// a new value with multiple keys that doesn't exist locally.
+fn changes_for_new_incoming(new: &JsonMap) -> StorageChanges {
+    let mut result = StorageChanges::with_capacity(new.len());
+    for (key, val) in new.iter() {
+        result.push(StorageValueChange {
+            key: key.clone(),
+            old_value: None,
+            new_value: Some(val.clone()),
+        });
+    }
+    result
+}
+
+// This module deals exclusively with the Map inside a JsonValue::Object().
+// This helper reads such a Map from a SQL row, ignoring anything which is
+// either invalid JSON or a different JSON type.
+fn json_map_from_row(row: &Row<'_>, col: &str) -> Result<DataState> {
+    let s = row.get::<_, Option<String>>(col)?;
+    Ok(match s {
+        None => DataState::Deleted,
+        Some(s) => match serde_json::from_str(&s) {
+            Ok(serde_json::Value::Object(m)) => DataState::Exists(m),
+            _ => {
+                // We don't want invalid json or wrong types to kill syncing.
+                // It should be impossible as we never write anything which
+                // could cause it, but we can't really log the bad data as there
+                // might be PII. Logging just a message without any additional
+                // clues is going to be unhelpfully noisy, so, silently None.
+                // XXX - Maybe record telemetry?
+                DataState::Deleted
+            }
+        },
+    })
+}
+
+/// The first thing we do with incoming items is to "stage" them in a temp table.
+/// The actual processing is done via this table.
+pub fn stage_incoming(
+    tx: &Transaction<'_>,
+    incoming_records: &[IncomingContent<WebextRecord>],
+    signal: &dyn Interruptee,
+) -> Result<()> {
+    sql_support::each_sized_chunk(
+        incoming_records,
+        // We bind 3 params per chunk.
+        sql_support::default_max_variable_number() / 3,
+        |chunk, _| -> Result<()> {
+            let mut params = Vec::with_capacity(chunk.len() * 3);
+            for record in chunk {
+                signal.err_if_interrupted()?;
+                match &record.kind {
+                    IncomingKind::Content(r) => {
+                        params.push(Some(record.envelope.id.to_string()));
+                        params.push(Some(r.ext_id.to_string()));
+                        params.push(Some(r.data.clone()));
+                    }
+                    IncomingKind::Tombstone => {
+                        params.push(Some(record.envelope.id.to_string()));
+                        params.push(None);
+                        params.push(None);
+                    }
+                    IncomingKind::Malformed => {
+                        log::error!("Ignoring incoming malformed record: {}", record.envelope.id);
+                    }
+                }
+            }
+            // we might have skipped records
+            let actual_len = params.len() / 3;
+            if actual_len != 0 {
+                let sql = format!(
+                    "INSERT OR REPLACE INTO temp.storage_sync_staging
+                    (guid, ext_id, data)
+                    VALUES {}",
+                    sql_support::repeat_multi_values(actual_len, 3)
+                );
+                tx.execute(&sql, rusqlite::params_from_iter(params))?;
+            }
+            Ok(())
+        },
+    )?;
+    Ok(())
+}
+
+/// The "state" we find ourselves in when considering an incoming/staging
+/// record. This "state" is the input to calculating the IncomingAction and
+/// carries all the data we need to make the required local changes.
+#[derive(Debug, PartialEq, Eq)]
+pub enum IncomingState {
+    /// There's an incoming item, but data for that extension doesn't exist
+    /// either in our local data store or in the local mirror. IOW, this is the
+    /// very first time we've seen this extension.
+    IncomingOnlyData { ext_id: String, data: JsonMap },
+
+    /// An incoming tombstone that doesn't exist locally. Because tombstones
+    /// don't carry the ext-id, it means it's not in our mirror. We are just
+    /// going to ignore it, but we track the state for consistency.
+    IncomingOnlyTombstone,
+
+    /// There's an incoming item and we have data for the same extension in
+    /// our local store - but not in our mirror. This should be relatively
+    /// uncommon as it means:
+    /// * Some other profile has recently installed an extension and synced.
+    /// * This profile has recently installed the same extension.
+    /// * This is the first sync for this profile since both those events
+    ///   happened.
+    HasLocal {
+        ext_id: String,
+        incoming: DataState,
+        local: DataState,
+    },
+    /// There's an incoming item and there's an item for the same extension in
+    /// the mirror. The addon probably doesn't exist locally, or if it does,
+    /// the last time we synced we synced the deletion of all data.
+    NotLocal {
+        ext_id: String,
+        incoming: DataState,
+        mirror: DataState,
+    },
+    /// This will be the most common "incoming" case - there's data incoming,
+    /// in the mirror and in the local store for an addon.
+    Everywhere {
+        ext_id: String,
+        incoming: DataState,
+        mirror: DataState,
+        local: DataState,
+    },
+}
+
+/// Get the items we need to process from the staging table. Return details about
+/// the item and the state of that item, ready for processing.
+pub fn get_incoming(conn: &Connection) -> Result<Vec<(SyncGuid, IncomingState)>> {
+    let sql = "
+        SELECT
+            s.guid as guid,
+            l.ext_id as l_ext_id,
+            m.ext_id as m_ext_id,
+            s.ext_id as s_ext_id,
+            s.data as s_data, m.data as m_data, l.data as l_data,
+            l.sync_change_counter
+        FROM temp.storage_sync_staging s
+        LEFT JOIN storage_sync_mirror m ON m.guid = s.guid
+        LEFT JOIN storage_sync_data l on l.ext_id IN (m.ext_id, s.ext_id);";
+
+    fn from_row(row: &Row<'_>) -> Result<(SyncGuid, IncomingState)> {
+        let guid = row.get("guid")?;
+        // This is complicated because the staging row doesn't hold the ext_id.
+        // However, both the local table and the mirror do.
+        let mirror_ext_id: Option<String> = row.get("m_ext_id")?;
+        let local_ext_id: Option<String> = row.get("l_ext_id")?;
+        let staged_ext_id: Option<String> = row.get("s_ext_id")?;
+        let incoming = json_map_from_row(row, "s_data")?;
+
+        // We find the state by examining which tables the ext-id exists in,
+        // using whether that column is null as a proxy for that.
+        let state = match (local_ext_id, mirror_ext_id) {
+            (None, None) => {
+                match staged_ext_id {
+                    Some(ext_id) => {
+                        let data = match incoming {
+                            // incoming record with missing data that's not a
+                            // tombstone shouldn't happen, but we can cope by
+                            // pretending it was an empty json map.
+                            DataState::Deleted => JsonMap::new(),
+                            DataState::Exists(data) => data,
+                        };
+                        IncomingState::IncomingOnlyData { ext_id, data }
+                    }
+                    None => IncomingState::IncomingOnlyTombstone,
+                }
+            }
+            (Some(ext_id), None) => IncomingState::HasLocal {
+                ext_id,
+                incoming,
+                local: json_map_from_row(row, "l_data")?,
+            },
+            (None, Some(ext_id)) => IncomingState::NotLocal {
+                ext_id,
+                incoming,
+                mirror: json_map_from_row(row, "m_data")?,
+            },
+            (Some(ext_id), Some(_)) => IncomingState::Everywhere {
+                ext_id,
+                incoming,
+                mirror: json_map_from_row(row, "m_data")?,
+                local: json_map_from_row(row, "l_data")?,
+            },
+        };
+        Ok((guid, state))
+    }
+
+    conn.conn().query_rows_and_then(sql, [], from_row)
+}
+
+/// This is the set of actions we know how to take *locally* for incoming
+/// records. Which one depends on the IncomingState.
+/// Every state which updates also records the set of changes we should notify
+#[derive(Debug, PartialEq, Eq)]
+pub enum IncomingAction {
+    /// We should locally delete the data for this record
+    DeleteLocally {
+        ext_id: String,
+        changes: StorageChanges,
+    },
+    /// We will take the remote.
+    TakeRemote {
+        ext_id: String,
+        data: JsonMap,
+        changes: StorageChanges,
+    },
+    /// We merged this data - this is what we came up with.
+    Merge {
+        ext_id: String,
+        data: JsonMap,
+        changes: StorageChanges,
+    },
+    /// Entry exists locally and it's the same as the incoming record.
+    Same { ext_id: String },
+    /// Incoming tombstone for an item we've never seen.
+    Nothing,
+}
+
+/// Takes the state of an item and returns the action we should take for it.
+pub fn plan_incoming(s: IncomingState) -> IncomingAction {
+    match s {
+        IncomingState::Everywhere {
+            ext_id,
+            incoming,
+            local,
+            mirror,
+        } => {
+            // All records exist - but do they all have data?
+            match (incoming, local, mirror) {
+                (
+                    DataState::Exists(incoming_data),
+                    DataState::Exists(local_data),
+                    DataState::Exists(mirror_data),
+                ) => {
+                    // all records have data - 3-way merge.
+                    merge(ext_id, incoming_data, local_data, Some(mirror_data))
+                }
+                (
+                    DataState::Exists(incoming_data),
+                    DataState::Exists(local_data),
+                    DataState::Deleted,
+                ) => {
+                    // No parent, so first time seeing this remotely - 2-way merge
+                    merge(ext_id, incoming_data, local_data, None)
+                }
+                (DataState::Exists(incoming_data), DataState::Deleted, _) => {
+                    // Incoming data, removed locally. Server wins.
+                    IncomingAction::TakeRemote {
+                        ext_id,
+                        changes: changes_for_new_incoming(&incoming_data),
+                        data: incoming_data,
+                    }
+                }
+                (DataState::Deleted, DataState::Exists(local_data), DataState::Exists(mirror)) => {
+                    // Deleted remotely.
+                    // Treat this as a delete of every key that we
+                    // know was present at the time.
+                    let (result, changes) = remove_matching_keys(local_data, &mirror);
+                    if result.is_empty() {
+                        // If there were no more keys left, we can
+                        // delete our version too.
+                        IncomingAction::DeleteLocally { ext_id, changes }
+                    } else {
+                        IncomingAction::Merge {
+                            ext_id,
+                            data: result,
+                            changes,
+                        }
+                    }
+                }
+                (DataState::Deleted, DataState::Exists(local_data), DataState::Deleted) => {
+                    // Perhaps another client created and then deleted
+                    // the whole object for this extension since the
+                    // last time we synced.
+                    // Treat this as a delete of every key that we
+                    // knew was present. Unfortunately, we don't know
+                    // any keys that were present, so we delete no keys.
+                    IncomingAction::Merge {
+                        ext_id,
+                        data: local_data,
+                        changes: StorageChanges::new(),
+                    }
+                }
+                (DataState::Deleted, DataState::Deleted, _) => {
+                    // We agree with the remote (regardless of what we
+                    // have mirrored).
+                    IncomingAction::Same { ext_id }
+                }
+            }
+        }
+        IncomingState::HasLocal {
+            ext_id,
+            incoming,
+            local,
+        } => {
+            // So we have a local record and an incoming/staging record, but *not* a
+            // mirror record. This means some other device has synced this for
+            // the first time and we are yet to do the same.
+            match (incoming, local) {
+                (DataState::Exists(incoming_data), DataState::Exists(local_data)) => {
+                    // This means the extension exists locally and remotely
+                    // but this is the first time we've synced it. That's no problem, it's
+                    // just a 2-way merge...
+                    merge(ext_id, incoming_data, local_data, None)
+                }
+                (DataState::Deleted, DataState::Exists(local_data)) => {
+                    // We've data locally, but there's an incoming deletion.
+                    // We would normally remove keys that we knew were
+                    // present on the server, but we don't know what
+                    // was on the server, so we don't remove anything.
+                    IncomingAction::Merge {
+                        ext_id,
+                        data: local_data,
+                        changes: StorageChanges::new(),
+                    }
+                }
+                (DataState::Exists(incoming_data), DataState::Deleted) => {
+                    // No data locally, but some is incoming - take it.
+                    IncomingAction::TakeRemote {
+                        ext_id,
+                        changes: changes_for_new_incoming(&incoming_data),
+                        data: incoming_data,
+                    }
+                }
+                (DataState::Deleted, DataState::Deleted) => {
+                    // Nothing anywhere - odd, but OK.
+                    IncomingAction::Same { ext_id }
+                }
+            }
+        }
+        IncomingState::NotLocal {
+            ext_id, incoming, ..
+        } => {
+            // No local data but there's mirror and an incoming record.
+            // This means a local deletion is being replaced by, or just re-doing
+            // the incoming record.
+            match incoming {
+                DataState::Exists(data) => IncomingAction::TakeRemote {
+                    ext_id,
+                    changes: changes_for_new_incoming(&data),
+                    data,
+                },
+                DataState::Deleted => IncomingAction::Same { ext_id },
+            }
+        }
+        IncomingState::IncomingOnlyData { ext_id, data } => {
+            // Only the staging record exists and it's not a tombstone.
+            // This means it's the first time we've ever seen it. No
+            // conflict possible, just take the remote.
+            IncomingAction::TakeRemote {
+                ext_id,
+                changes: changes_for_new_incoming(&data),
+                data,
+            }
+        }
+        IncomingState::IncomingOnlyTombstone => {
+            // Only the staging record exists and it is a tombstone - nothing to do.
+            IncomingAction::Nothing
+        }
+    }
+}
+
+fn insert_changes(tx: &Transaction<'_>, ext_id: &str, changes: &StorageChanges) -> Result<()> {
+    tx.execute_cached(
+        "INSERT INTO temp.storage_sync_applied (ext_id, changes)
+            VALUES (:ext_id, :changes)",
+        rusqlite::named_params! {
+            ":ext_id": ext_id,
+            ":changes": &serde_json::to_string(&changes)?,
+        },
+    )?;
+    Ok(())
+}
+
+// Apply the actions necessary to fully process the incoming items.
+pub fn apply_actions(
+    tx: &Transaction<'_>,
+    actions: Vec<(SyncGuid, IncomingAction)>,
+    signal: &dyn Interruptee,
+) -> Result<()> {
+    for (item, action) in actions {
+        signal.err_if_interrupted()?;
+
+        log::trace!("action for '{:?}': {:?}", item, action);
+        match action {
+            IncomingAction::DeleteLocally { ext_id, changes } => {
+                // Can just nuke it entirely.
+                tx.execute_cached(
+                    "DELETE FROM storage_sync_data WHERE ext_id = :ext_id",
+                    &[(":ext_id", &ext_id)],
+                )?;
+                insert_changes(tx, &ext_id, &changes)?;
+            }
+            // We want to update the local record with 'data' and after this update the item no longer is considered dirty.
+            IncomingAction::TakeRemote {
+                ext_id,
+                data,
+                changes,
+            } => {
+                tx.execute_cached(
+                    "INSERT OR REPLACE INTO storage_sync_data(ext_id, data, sync_change_counter)
+                        VALUES (:ext_id, :data, 0)",
+                    rusqlite::named_params! {
+                        ":ext_id": ext_id,
+                        ":data": serde_json::Value::Object(data),
+                    },
+                )?;
+                insert_changes(tx, &ext_id, &changes)?;
+            }
+
+            // We merged this data, so need to update locally but still consider
+            // it dirty because the merged data must be uploaded.
+            IncomingAction::Merge {
+                ext_id,
+                data,
+                changes,
+            } => {
+                tx.execute_cached(
+                    "UPDATE storage_sync_data SET data = :data, sync_change_counter = sync_change_counter + 1 WHERE ext_id = :ext_id",
+                    rusqlite::named_params! {
+                        ":ext_id": ext_id,
+                        ":data": serde_json::Value::Object(data),
+                    },
+                )?;
+                insert_changes(tx, &ext_id, &changes)?;
+            }
+
+            // Both local and remote ended up the same - only need to nuke the
+            // change counter.
+            IncomingAction::Same { ext_id } => {
+                tx.execute_cached(
+                    "UPDATE storage_sync_data SET sync_change_counter = 0 WHERE ext_id = :ext_id",
+                    &[(":ext_id", &ext_id)],
+                )?;
+                // no changes to write
+            }
+            // Literally nothing to do!
+            IncomingAction::Nothing => {}
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test::new_syncable_mem_db;
+    use super::*;
+    use crate::api;
+    use interrupt_support::NeverInterrupts;
+    use serde_json::{json, Value};
+    use sync15::bso::IncomingBso;
+
+    // select simple int
+    fn ssi(conn: &Connection, stmt: &str) -> u32 {
+        conn.try_query_one(stmt, [], true)
+            .expect("must work")
+            .unwrap_or_default()
+    }
+
+    fn array_to_incoming(mut array: Value) -> Vec<IncomingContent<WebextRecord>> {
+        let jv = array.as_array_mut().expect("you must pass a json array");
+        let mut result = Vec::with_capacity(jv.len());
+        for elt in jv {
+            result.push(IncomingBso::from_test_content(elt.take()).into_content());
+        }
+        result
+    }
+
+    // Can't find a way to import these from crate::sync::tests...
+    macro_rules! map {
+        ($($map:tt)+) => {
+            json!($($map)+).as_object().unwrap().clone()
+        };
+    }
+    macro_rules! change {
+        ($key:literal, None, None) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: None,
+                new_value: None,
+            };
+        };
+        ($key:literal, $old:tt, None) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: Some(json!($old)),
+                new_value: None,
+            }
+        };
+        ($key:literal, None, $new:tt) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: None,
+                new_value: Some(json!($new)),
+            };
+        };
+        ($key:literal, $old:tt, $new:tt) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: Some(json!($old)),
+                new_value: Some(json!($new)),
+            }
+        };
+    }
+    macro_rules! changes {
+        ( $( $change:expr ),* ) => {
+            {
+                let mut changes = StorageChanges::new();
+                $(
+                    changes.push($change);
+                )*
+                changes
+            }
+        };
+    }
+
+    #[test]
+    fn test_incoming_populates_staging() -> Result<()> {
+        let mut db = new_syncable_mem_db();
+        let tx = db.transaction()?;
+
+        let incoming = json! {[
+            {
+                "id": "guidAAAAAAAA",
+                "extId": "ext1@example.com",
+                "data": json!({"foo": "bar"}).to_string(),
+            }
+        ]};
+
+        stage_incoming(&tx, &array_to_incoming(incoming), &NeverInterrupts)?;
+        // check staging table
+        assert_eq!(
+            ssi(&tx, "SELECT count(*) FROM temp.storage_sync_staging"),
+            1
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_fetch_incoming_state() -> Result<()> {
+        let mut db = new_syncable_mem_db();
+        let tx = db.transaction()?;
+
+        // Start with an item just in staging.
+        tx.execute(
+            r#"
+            INSERT INTO temp.storage_sync_staging (guid, ext_id, data)
+            VALUES ('guid', 'ext_id', '{"foo":"bar"}')
+        "#,
+            [],
+        )?;
+
+        let incoming = get_incoming(&tx)?;
+        assert_eq!(incoming.len(), 1);
+        assert_eq!(incoming[0].0, SyncGuid::new("guid"),);
+        assert_eq!(
+            incoming[0].1,
+            IncomingState::IncomingOnlyData {
+                ext_id: "ext_id".to_string(),
+                data: map!({"foo": "bar"}),
+            }
+        );
+
+        // Add the same item to the mirror.
+        tx.execute(
+            r#"
+            INSERT INTO storage_sync_mirror (guid, ext_id, data)
+            VALUES ('guid', 'ext_id', '{"foo":"new"}')
+        "#,
+            [],
+        )?;
+        let incoming = get_incoming(&tx)?;
+        assert_eq!(incoming.len(), 1);
+        assert_eq!(
+            incoming[0].1,
+            IncomingState::NotLocal {
+                ext_id: "ext_id".to_string(),
+                incoming: DataState::Exists(map!({"foo": "bar"})),
+                mirror: DataState::Exists(map!({"foo": "new"})),
+            }
+        );
+
+        // and finally the data itself - might as use the API here!
+        api::set(&tx, "ext_id", json!({"foo": "local"}))?;
+        let incoming = get_incoming(&tx)?;
+        assert_eq!(incoming.len(), 1);
+        assert_eq!(
+            incoming[0].1,
+            IncomingState::Everywhere {
+                ext_id: "ext_id".to_string(),
+                incoming: DataState::Exists(map!({"foo": "bar"})),
+                local: DataState::Exists(map!({"foo": "local"})),
+                mirror: DataState::Exists(map!({"foo": "new"})),
+            }
+        );
+        Ok(())
+    }
+
+    // Like test_fetch_incoming_state, but check NULLs are handled correctly.
+    #[test]
+    fn test_fetch_incoming_state_nulls() -> Result<()> {
+        let mut db = new_syncable_mem_db();
+        let tx = db.transaction()?;
+
+        // Start with a tombstone just in staging.
+        tx.execute(
+            r#"
+            INSERT INTO temp.storage_sync_staging (guid, ext_id, data)
+            VALUES ('guid', NULL, NULL)
+        "#,
+            [],
+        )?;
+
+        let incoming = get_incoming(&tx)?;
+        assert_eq!(incoming.len(), 1);
+        assert_eq!(incoming[0].1, IncomingState::IncomingOnlyTombstone,);
+
+        // Add the same item to the mirror (can't store an ext_id for a
+        // tombstone in the mirror as incoming tombstones never have it)
+        tx.execute(
+            r#"
+            INSERT INTO storage_sync_mirror (guid, ext_id, data)
+            VALUES ('guid', NULL, NULL)
+        "#,
+            [],
+        )?;
+        let incoming = get_incoming(&tx)?;
+        assert_eq!(incoming.len(), 1);
+        assert_eq!(incoming[0].1, IncomingState::IncomingOnlyTombstone);
+
+        tx.execute(
+            r#"
+            INSERT INTO storage_sync_data (ext_id, data)
+            VALUES ('ext_id', NULL)
+        "#,
+            [],
+        )?;
+        let incoming = get_incoming(&tx)?;
+        assert_eq!(incoming.len(), 1);
+        assert_eq!(
+            incoming[0].1,
+            // IncomingOnly* seems a little odd, but it is because we can't
+            // tie the tombstones together due to the lack of any ext-id/guid
+            // mapping in this case.
+            IncomingState::IncomingOnlyTombstone
+        );
+        Ok(())
+    }
+
+    // apply_action tests.
+    #[derive(Debug, PartialEq)]
+    struct LocalItem {
+        data: DataState,
+        sync_change_counter: i32,
+    }
+
+    fn get_local_item(conn: &Connection) -> Option<LocalItem> {
+        conn.try_query_row::<_, Error, _, _>(
+            "SELECT data, sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext_id'",
+            [],
+            |row| {
+                let data = json_map_from_row(row, "data")?;
+                let sync_change_counter = row.get::<_, i32>(1)?;
+                Ok(LocalItem {
+                    data,
+                    sync_change_counter,
+                })
+            },
+            true,
+        )
+        .expect("query should work")
+    }
+
+    fn get_applied_item_changes(conn: &Connection) -> Option<StorageChanges> {
+        // no custom deserialize for storagechanges and we only need it for
+        // tests, so do it manually.
+        conn.try_query_row::<_, Error, _, _>(
+            "SELECT changes FROM temp.storage_sync_applied WHERE ext_id = 'ext_id'",
+            [],
+            |row| Ok(serde_json::from_str(&row.get::<_, String>("changes")?)?),
+            true,
+        )
+        .expect("query should work")
+        .map(|val: serde_json::Value| {
+            let ob = val.as_object().expect("should be an object of items");
+            let mut result = StorageChanges::with_capacity(ob.len());
+            for (key, val) in ob.into_iter() {
+                let details = val.as_object().expect("elts should be objects");
+                result.push(StorageValueChange {
+                    key: key.to_string(),
+                    old_value: details.get("oldValue").cloned(),
+                    new_value: details.get("newValue").cloned(),
+                });
+            }
+            result
+        })
+    }
+
+    fn do_apply_action(tx: &Transaction<'_>, action: IncomingAction) {
+        let guid = SyncGuid::new("guid");
+        apply_actions(tx, vec![(guid, action)], &NeverInterrupts).expect("should apply");
+    }
+
+    #[test]
+    fn test_apply_actions() -> Result<()> {
+        let mut db = new_syncable_mem_db();
+
+        // DeleteLocally - row should be entirely removed.
+        let tx = db.transaction().expect("transaction should work");
+        api::set(&tx, "ext_id", json!({"foo": "local"}))?;
+        assert_eq!(
+            api::get(&tx, "ext_id", json!(null))?,
+            json!({"foo": "local"})
+        );
+        let changes = changes![change!("foo", "local", None)];
+        do_apply_action(
+            &tx,
+            IncomingAction::DeleteLocally {
+                ext_id: "ext_id".to_string(),
+                changes: changes.clone(),
+            },
+        );
+        assert_eq!(api::get(&tx, "ext_id", json!(null))?, json!({}));
+        // and there should not be a local record at all.
+        assert!(get_local_item(&tx).is_none());
+        assert_eq!(get_applied_item_changes(&tx), Some(changes));
+        tx.rollback()?;
+
+        // TakeRemote - replace local data with remote and marked as not dirty.
+        let tx = db.transaction().expect("transaction should work");
+        api::set(&tx, "ext_id", json!({"foo": "local"}))?;
+        assert_eq!(
+            api::get(&tx, "ext_id", json!(null))?,
+            json!({"foo": "local"})
+        );
+        // data should exist locally with a change recorded.
+        assert_eq!(
+            get_local_item(&tx),
+            Some(LocalItem {
+                data: DataState::Exists(map!({"foo": "local"})),
+                sync_change_counter: 1
+            })
+        );
+        let changes = changes![change!("foo", "local", "remote")];
+        do_apply_action(
+            &tx,
+            IncomingAction::TakeRemote {
+                ext_id: "ext_id".to_string(),
+                data: map!({"foo": "remote"}),
+                changes: changes.clone(),
+            },
+        );
+        // data should exist locally with the remote data and not be dirty.
+        assert_eq!(
+            get_local_item(&tx),
+            Some(LocalItem {
+                data: DataState::Exists(map!({"foo": "remote"})),
+                sync_change_counter: 0
+            })
+        );
+        assert_eq!(get_applied_item_changes(&tx), Some(changes));
+        tx.rollback()?;
+
+        // Merge - like ::TakeRemote, but data remains dirty.
+        let tx = db.transaction().expect("transaction should work");
+        api::set(&tx, "ext_id", json!({"foo": "local"}))?;
+        assert_eq!(
+            api::get(&tx, "ext_id", json!(null))?,
+            json!({"foo": "local"})
+        );
+        // data should exist locally with a change recorded.
+        assert_eq!(
+            get_local_item(&tx),
+            Some(LocalItem {
+                data: DataState::Exists(map!({"foo": "local"})),
+                sync_change_counter: 1
+            })
+        );
+        let changes = changes![change!("foo", "local", "remote")];
+        do_apply_action(
+            &tx,
+            IncomingAction::Merge {
+                ext_id: "ext_id".to_string(),
+                data: map!({"foo": "remote"}),
+                changes: changes.clone(),
+            },
+        );
+        assert_eq!(
+            get_local_item(&tx),
+            Some(LocalItem {
+                data: DataState::Exists(map!({"foo": "remote"})),
+                sync_change_counter: 2
+            })
+        );
+        assert_eq!(get_applied_item_changes(&tx), Some(changes));
+        tx.rollback()?;
+
+        // Same - data stays the same but is marked not dirty.
+        let tx = db.transaction().expect("transaction should work");
+        api::set(&tx, "ext_id", json!({"foo": "local"}))?;
+        assert_eq!(
+            api::get(&tx, "ext_id", json!(null))?,
+            json!({"foo": "local"})
+        );
+        // data should exist locally with a change recorded.
+        assert_eq!(
+            get_local_item(&tx),
+            Some(LocalItem {
+                data: DataState::Exists(map!({"foo": "local"})),
+                sync_change_counter: 1
+            })
+        );
+        do_apply_action(
+            &tx,
+            IncomingAction::Same {
+                ext_id: "ext_id".to_string(),
+            },
+        );
+        assert_eq!(
+            get_local_item(&tx),
+            Some(LocalItem {
+                data: DataState::Exists(map!({"foo": "local"})),
+                sync_change_counter: 0
+            })
+        );
+        assert_eq!(get_applied_item_changes(&tx), None);
+        tx.rollback()?;
+
+        Ok(())
+    }
+}
diff --git a/third_party/rust/webext-storage/src/sync/mod.rs b/third_party/rust/webext-storage/src/sync/mod.rs
new file mode 100644
index 0000000000..3144049dca
--- /dev/null
+++ b/third_party/rust/webext-storage/src/sync/mod.rs
@@ -0,0 +1,429 @@
+/* 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/. */
+
+mod bridge;
+mod incoming;
+mod outgoing;
+
+#[cfg(test)]
+mod sync_tests;
+
+use crate::api::{StorageChanges, StorageValueChange};
+use crate::db::StorageDb;
+use crate::error::*;
+use serde::Deserialize;
+use serde_derive::*;
+use sql_support::ConnExt;
+use sync_guid::Guid as SyncGuid;
+
+pub use bridge::BridgedEngine;
+use incoming::IncomingAction;
+
+type JsonMap = serde_json::Map<String, serde_json::Value>;
+
+pub const STORAGE_VERSION: usize = 1;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WebextRecord {
+    #[serde(rename = "id")]
+    guid: SyncGuid,
+    #[serde(rename = "extId")]
+    ext_id: String,
+    data: String,
+}
+
+// Perform a 2-way or 3-way merge, where the incoming value wins on confict.
+fn merge(
+    ext_id: String,
+    mut other: JsonMap,
+    mut ours: JsonMap,
+    parent: Option<JsonMap>,
+) -> IncomingAction {
+    if other == ours {
+        return IncomingAction::Same { ext_id };
+    }
+    let old_incoming = other.clone();
+    // worst case is keys in each are unique.
+    let mut changes = StorageChanges::with_capacity(other.len() + ours.len());
+    if let Some(parent) = parent {
+        // Perform 3-way merge. First, for every key in parent,
+        // compare the parent value with the incoming value to compute
+        // an implicit "diff".
+        for (key, parent_value) in parent.into_iter() {
+            if let Some(incoming_value) = other.remove(&key) {
+                if incoming_value != parent_value {
+                    log::trace!(
+                        "merge: key {} was updated in incoming - copying value locally",
+                        key
+                    );
+                    let old_value = ours.remove(&key);
+                    let new_value = Some(incoming_value.clone());
+                    if old_value != new_value {
+                        changes.push(StorageValueChange {
+                            key: key.clone(),
+                            old_value,
+                            new_value,
+                        });
+                    }
+                    ours.insert(key, incoming_value);
+                }
+            } else {
+                // Key was not present in incoming value.
+                // Another client must have deleted it.
+                log::trace!(
+                    "merge: key {} no longer present in incoming - removing it locally",
+                    key
+                );
+                if let Some(old_value) = ours.remove(&key) {
+                    changes.push(StorageValueChange {
+                        key,
+                        old_value: Some(old_value),
+                        new_value: None,
+                    });
+                }
+            }
+        }
+
+        // Then, go through every remaining key in incoming. These are
+        // the ones where a corresponding key does not exist in
+        // parent, so it is a new key, and we need to add it.
+        for (key, incoming_value) in other.into_iter() {
+            log::trace!(
+                "merge: key {} doesn't occur in parent - copying from incoming",
+                key
+            );
+            changes.push(StorageValueChange {
+                key: key.clone(),
+                old_value: None,
+                new_value: Some(incoming_value.clone()),
+            });
+            ours.insert(key, incoming_value);
+        }
+    } else {
+        // No parent. Server wins. Overwrite every key in ours with
+        // the corresponding value in other.
+        log::trace!("merge: no parent - copying all keys from incoming");
+        for (key, incoming_value) in other.into_iter() {
+            let old_value = ours.remove(&key);
+            let new_value = Some(incoming_value.clone());
+            if old_value != new_value {
+                changes.push(StorageValueChange {
+                    key: key.clone(),
+                    old_value,
+                    new_value,
+                });
+            }
+            ours.insert(key, incoming_value);
+        }
+    }
+
+    if ours == old_incoming {
+        IncomingAction::TakeRemote {
+            ext_id,
+            data: old_incoming,
+            changes,
+        }
+    } else {
+        IncomingAction::Merge {
+            ext_id,
+            data: ours,
+            changes,
+        }
+    }
+}
+
+fn remove_matching_keys(mut ours: JsonMap, keys_to_remove: &JsonMap) -> (JsonMap, StorageChanges) {
+    let mut changes = StorageChanges::with_capacity(keys_to_remove.len());
+    for key in keys_to_remove.keys() {
+        if let Some(old_value) = ours.remove(key) {
+            changes.push(StorageValueChange {
+                key: key.clone(),
+                old_value: Some(old_value),
+                new_value: None,
+            });
+        }
+    }
+    (ours, changes)
+}
+
+/// Holds a JSON-serialized map of all synced changes for an extension.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SyncedExtensionChange {
+    /// The extension ID.
+    pub ext_id: String,
+    /// The contents of a `StorageChanges` struct, in JSON format. We don't
+    /// deserialize these because they need to be passed back to the browser
+    /// as strings anyway.
+    pub changes: String,
+}
+
+// Fetches the applied changes we stashed in the storage_sync_applied table.
+pub fn get_synced_changes(db: &StorageDb) -> Result<Vec<SyncedExtensionChange>> {
+    let signal = db.begin_interrupt_scope()?;
+    let sql = "SELECT ext_id, changes FROM temp.storage_sync_applied";
+    db.conn().query_rows_and_then(sql, [], |row| -> Result<_> {
+        signal.err_if_interrupted()?;
+        Ok(SyncedExtensionChange {
+            ext_id: row.get("ext_id")?,
+            changes: row.get("changes")?,
+        })
+    })
+}
+
+// Helpers for tests
+#[cfg(test)]
+pub mod test {
+    use crate::db::{test::new_mem_db, StorageDb};
+    use crate::schema::create_empty_sync_temp_tables;
+
+    pub fn new_syncable_mem_db() -> StorageDb {
+        let _ = env_logger::try_init();
+        let db = new_mem_db();
+        create_empty_sync_temp_tables(&db).expect("should work");
+        db
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::test::new_syncable_mem_db;
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn test_serde_record_ser() {
+        assert_eq!(
+            serde_json::to_string(&WebextRecord {
+                guid: "guid".into(),
+                ext_id: "ext_id".to_string(),
+                data: "data".to_string()
+            })
+            .unwrap(),
+            r#"{"id":"guid","extId":"ext_id","data":"data"}"#
+        );
+    }
+
+    // a macro for these tests - constructs a serde_json::Value::Object
+    macro_rules! map {
+        ($($map:tt)+) => {
+            json!($($map)+).as_object().unwrap().clone()
+        };
+    }
+
+    macro_rules! change {
+        ($key:literal, None, None) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: None,
+                new_value: None,
+            };
+        };
+        ($key:literal, $old:tt, None) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: Some(json!($old)),
+                new_value: None,
+            }
+        };
+        ($key:literal, None, $new:tt) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: None,
+                new_value: Some(json!($new)),
+            }
+        };
+        ($key:literal, $old:tt, $new:tt) => {
+            StorageValueChange {
+                key: $key.to_string(),
+                old_value: Some(json!($old)),
+                new_value: Some(json!($new)),
+            }
+        };
+    }
+    macro_rules! changes {
+        ( ) => {
+            StorageChanges::new()
+        };
+        ( $( $change:expr ),* ) => {
+            {
+                let mut changes = StorageChanges::new();
+                $(
+                    changes.push($change);
+                )*
+                changes
+            }
+        };
+    }
+
+    #[test]
+    fn test_3way_merging() {
+        // No conflict - identical local and remote.
+        assert_eq!(
+            merge(
+                "ext-id".to_string(),
+                map!({"one": "one", "two": "two"}),
+                map!({"two": "two", "one": "one"}),
+                Some(map!({"parent_only": "parent"})),
+            ),
+            IncomingAction::Same {
+                ext_id: "ext-id".to_string()
+            }
+        );
+        assert_eq!(
+            merge(
+                "ext-id".to_string(),
+                map!({"other_only": "other", "common": "common"}),
+                map!({"ours_only": "ours", "common": "common"}),
+                Some(map!({"parent_only": "parent", "common": "old_common"})),
+            ),
+            IncomingAction::Merge {
+                ext_id: "ext-id".to_string(),
+                data: map!({"other_only": "other", "ours_only": "ours", "common": "common"}),
+                changes: changes![change!("other_only", None, "other")],
+            }
+        );
+        // Simple conflict - parent value is neither local nor incoming. incoming wins.
+        assert_eq!(
+            merge(
+                "ext-id".to_string(),
+                map!({"other_only": "other", "common": "incoming"}),
+                map!({"ours_only": "ours", "common": "local"}),
+                Some(map!({"parent_only": "parent", "common": "parent"})),
+            ),
+            IncomingAction::Merge {
+                ext_id: "ext-id".to_string(),
+                data: map!({"other_only": "other", "ours_only": "ours", "common": "incoming"}),
+                changes: changes![
+                    change!("common", "local", "incoming"),
+                    change!("other_only", None, "other")
+                ],
+            }
+        );
+        // Local change, no conflict.
+        assert_eq!(
+            merge(
+                "ext-id".to_string(),
+                map!({"other_only": "other", "common": "old_value"}),
+                map!({"ours_only": "ours", "common": "new_value"}),
+                Some(map!({"parent_only": "parent", "common": "old_value"})),
+            ),
+            IncomingAction::Merge {
+                ext_id: "ext-id".to_string(),
+                data: map!({"other_only": "other", "ours_only": "ours", "common": "new_value"}),
+                changes: changes![change!("other_only", None, "other")],
+            }
+        );
+        // Field was removed remotely.
+        assert_eq!(
+            merge(
+                "ext-id".to_string(),
+                map!({"other_only": "other"}),
+                map!({"common": "old_value"}),
+                Some(map!({"common": "old_value"})),
+            ),
+            IncomingAction::TakeRemote {
+                ext_id: "ext-id".to_string(),
+                data: map!({"other_only": "other"}),
+                changes: changes![
+                    change!("common", "old_value", None),
+                    change!("other_only", None, "other")
+                ],
+            }
+        );
+        // Field was removed remotely but we added another one.
+        assert_eq!(
+            merge(
+                "ext-id".to_string(),
+                map!({"other_only": "other"}),
+                map!({"common": "old_value", "new_key": "new_value"}),
+                Some(map!({"common": "old_value"})),
+            ),
+            IncomingAction::Merge {
+                ext_id: "ext-id".to_string(),
+                data: map!({"other_only": "other", "new_key": "new_value"}),
+                changes: changes![
+                    change!("common", "old_value", None),
+                    change!("other_only", None, "other")
+                ],
+            }
+        );
+        // Field was removed both remotely and locally.
+        assert_eq!(
+            merge(
+                "ext-id".to_string(),
+                map!({}),
+                map!({"new_key": "new_value"}),
+                Some(map!({"common": "old_value"})),
+            ),
+            IncomingAction::Merge {
+                ext_id: "ext-id".to_string(),
+                data: map!({"new_key": "new_value"}),
+                changes: changes![],
+            }
+        );
+    }
+
+    #[test]
+    fn test_remove_matching_keys() {
+        assert_eq!(
+            remove_matching_keys(
+                map!({"key1": "value1", "key2": "value2"}),
+                &map!({"key1": "ignored", "key3": "ignored"})
+            ),
+            (
+                map!({"key2": "value2"}),
+                changes![change!("key1", "value1", None)]
+            )
+        );
+    }
+
+    #[test]
+    fn test_get_synced_changes() -> Result<()> {
+        let db = new_syncable_mem_db();
+        db.execute_batch(&format!(
+            r#"INSERT INTO temp.storage_sync_applied (ext_id, changes)
+                VALUES
+                ('an-extension', '{change1}'),
+                ('ext"id', '{change2}')
+            "#,
+            change1 = serde_json::to_string(&changes![change!("key1", "old-val", None)])?,
+            change2 = serde_json::to_string(&changes![change!("key-for-second", None, "new-val")])?
+        ))?;
+        let changes = get_synced_changes(&db)?;
+        assert_eq!(changes[0].ext_id, "an-extension");
+        // sanity check it's valid!
+        let c1: JsonMap =
+            serde_json::from_str(&changes[0].changes).expect("changes must be an object");
+        assert_eq!(
+            c1.get("key1")
+                .expect("must exist")
+                .as_object()
+                .expect("must be an object")
+                .get("oldValue"),
+            Some(&json!("old-val"))
+        );
+
+        // phew - do it again to check the string got escaped.
+        assert_eq!(
+            changes[1],
+            SyncedExtensionChange {
+                ext_id: "ext\"id".into(),
+                changes: r#"{"key-for-second":{"newValue":"new-val"}}"#.into(),
+            }
+        );
+        assert_eq!(changes[1].ext_id, "ext\"id");
+        let c2: JsonMap =
+            serde_json::from_str(&changes[1].changes).expect("changes must be an object");
+        assert_eq!(
+            c2.get("key-for-second")
+                .expect("must exist")
+                .as_object()
+                .expect("must be an object")
+                .get("newValue"),
+            Some(&json!("new-val"))
+        );
+        Ok(())
+    }
+}
diff --git a/third_party/rust/webext-storage/src/sync/outgoing.rs b/third_party/rust/webext-storage/src/sync/outgoing.rs
new file mode 100644
index 0000000000..c6c9dd64e5
--- /dev/null
+++ b/third_party/rust/webext-storage/src/sync/outgoing.rs
@@ -0,0 +1,186 @@
+/* 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/. */
+
+// The "outgoing" part of syncing - building the payloads to upload and
+// managing the sync state of the local DB.
+
+use interrupt_support::Interruptee;
+use rusqlite::{Connection, Row, Transaction};
+use sql_support::ConnExt;
+use sync15::bso::OutgoingBso;
+use sync_guid::Guid as SyncGuid;
+
+use crate::error::*;
+
+use super::WebextRecord;
+
+fn outgoing_from_row(row: &Row<'_>) -> Result<OutgoingBso> {
+    let guid: SyncGuid = row.get("guid")?;
+    let ext_id: String = row.get("ext_id")?;
+    let raw_data: Option<String> = row.get("data")?;
+    Ok(match raw_data {
+        Some(raw_data) => {
+            let record = WebextRecord {
+                guid,
+                ext_id,
+                data: raw_data,
+            };
+            OutgoingBso::from_content_with_id(record)?
+        }
+        None => OutgoingBso::new_tombstone(guid.into()),
+    })
+}
+
+/// Stages info about what should be uploaded in a temp table. This should be
+/// called in the same transaction as `apply_actions`. record_uploaded() can be
+/// called after the upload is complete and the data in the temp table will be
+/// used to update the local store.
+pub fn stage_outgoing(tx: &Transaction<'_>) -> Result<()> {
+    let sql = "
+        -- Stage outgoing items. The item may not yet have a GUID (ie, it might
+        -- not already be in either the mirror nor the incoming staging table),
+        -- so we generate one if it doesn't exist.
+        INSERT INTO storage_sync_outgoing_staging
+        (guid, ext_id, data, sync_change_counter)
+        SELECT coalesce(m.guid, s.guid, generate_guid()),
+        l.ext_id, l.data, l.sync_change_counter
+        FROM storage_sync_data l
+        -- left joins as one or both may not exist.
+        LEFT JOIN storage_sync_mirror m ON m.ext_id = l.ext_id
+        LEFT JOIN storage_sync_staging s ON s.ext_id = l.ext_id
+        WHERE sync_change_counter > 0;
+
+        -- At this point, we've merged in all new records, so copy incoming
+        -- staging into the mirror so that it matches what's on the server.
+        INSERT OR REPLACE INTO storage_sync_mirror (guid, ext_id, data)
+        SELECT guid, ext_id, data FROM temp.storage_sync_staging;
+
+        -- And copy any incoming records that we aren't reuploading into the
+        -- local table. We'll copy the outgoing ones into the mirror and local
+        -- after we upload them.
+        INSERT OR REPLACE INTO storage_sync_data (ext_id, data, sync_change_counter)
+        SELECT ext_id, data, 0
+        FROM storage_sync_staging s
+        WHERE ext_id IS NOT NULL
+        AND NOT EXISTS(SELECT 1 FROM storage_sync_outgoing_staging o
+                       WHERE o.guid = s.guid);";
+    tx.execute_batch(sql)?;
+    Ok(())
+}
+
+/// Returns a vec of the outgoing records which should be uploaded.
+pub fn get_outgoing(conn: &Connection, signal: &dyn Interruptee) -> Result<Vec<OutgoingBso>> {
+    let sql = "SELECT guid, ext_id, data FROM storage_sync_outgoing_staging";
+    let elts = conn
+        .conn()
+        .query_rows_and_then(sql, [], |row| -> Result<_> {
+            signal.err_if_interrupted()?;
+            outgoing_from_row(row)
+        })?;
+
+    log::debug!("get_outgoing found {} items", elts.len());
+    Ok(elts.into_iter().collect())
+}
+
+/// Record the fact that items were uploaded. This updates the state of the
+/// local DB to reflect the state of the server we just updated.
+/// Note that this call is almost certainly going to be made in a *different*
+/// transaction than the transaction used in `stage_outgoing()`, and it will
+/// be called once per batch upload.
+pub fn record_uploaded(
+    tx: &Transaction<'_>,
+    items: &[SyncGuid],
+    signal: &dyn Interruptee,
+) -> Result<()> {
+    log::debug!(
+        "record_uploaded recording that {} items were uploaded",
+        items.len()
+    );
+
+    // Updating the `was_uploaded` column fires the `record_uploaded` trigger,
+    // which updates the local change counter and writes the uploaded record
+    // data back to the mirror.
+    sql_support::each_chunk(items, |chunk, _| -> Result<()> {
+        signal.err_if_interrupted()?;
+        let sql = format!(
+            "UPDATE storage_sync_outgoing_staging SET
+                 was_uploaded = 1
+             WHERE guid IN ({})",
+            sql_support::repeat_sql_vars(chunk.len()),
+        );
+        tx.execute(&sql, rusqlite::params_from_iter(chunk))?;
+        Ok(())
+    })?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test::new_syncable_mem_db;
+    use super::*;
+    use interrupt_support::NeverInterrupts;
+
+    #[test]
+    fn test_simple() -> Result<()> {
+        let mut db = new_syncable_mem_db();
+        let tx = db.transaction()?;
+
+        tx.execute_batch(
+            r#"
+            INSERT INTO storage_sync_data (ext_id, data, sync_change_counter)
+            VALUES
+                ('ext_no_changes', '{"foo":"bar"}', 0),
+                ('ext_with_changes', '{"foo":"bar"}', 1);
+        "#,
+        )?;
+
+        stage_outgoing(&tx)?;
+        let changes = get_outgoing(&tx, &NeverInterrupts)?;
+        assert_eq!(changes.len(), 1);
+        let record: serde_json::Value = serde_json::from_str(&changes[0].payload).unwrap();
+        let ext_id = record.get("extId").unwrap().as_str().unwrap();
+
+        assert_eq!(ext_id, "ext_with_changes");
+
+        record_uploaded(
+            &tx,
+            changes
+                .into_iter()
+                .map(|p| p.envelope.id)
+                .collect::<Vec<SyncGuid>>()
+                .as_slice(),
+            &NeverInterrupts,
+        )?;
+
+        let counter: i32 = tx.conn().query_one(
+            "SELECT sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext_with_changes'",
+        )?;
+        assert_eq!(counter, 0);
+        Ok(())
+    }
+
+    #[test]
+    fn test_payload_serialization() {
+        let record = WebextRecord {
+            guid: SyncGuid::new("guid"),
+            ext_id: "ext-id".to_string(),
+            data: "{}".to_string(),
+        };
+
+        let outgoing = OutgoingBso::from_content_with_id(record).unwrap();
+
+        // The envelope should have our ID.
+        assert_eq!(outgoing.envelope.id, "guid");
+
+        let outgoing_payload =
+            serde_json::from_str::<serde_json::Value>(&outgoing.payload).unwrap();
+        let outgoing_map = outgoing_payload.as_object().unwrap();
+
+        assert!(outgoing_map.contains_key("id"));
+        assert!(outgoing_map.contains_key("data"));
+        assert!(outgoing_map.contains_key("extId"));
+        assert_eq!(outgoing_map.len(), 3);
+    }
+}
diff --git a/third_party/rust/webext-storage/src/sync/sync_tests.rs b/third_party/rust/webext-storage/src/sync/sync_tests.rs
new file mode 100644
index 0000000000..6940be2e59
--- /dev/null
+++ b/third_party/rust/webext-storage/src/sync/sync_tests.rs
@@ -0,0 +1,529 @@
+/* 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 tries to simulate "full syncs" - ie, from local state changing, to
+// fetching incoming items, generating items to upload, then updating the local
+// state (including the mirror) as a result.
+
+use crate::api::{clear, get, set};
+use crate::error::*;
+use crate::schema::create_empty_sync_temp_tables;
+use crate::sync::incoming::{apply_actions, get_incoming, plan_incoming, stage_incoming};
+use crate::sync::outgoing::{get_outgoing, record_uploaded, stage_outgoing};
+use crate::sync::test::new_syncable_mem_db;
+use crate::sync::WebextRecord;
+use interrupt_support::NeverInterrupts;
+use rusqlite::{Connection, Row, Transaction};
+use serde_json::json;
+use sql_support::ConnExt;
+use sync15::bso::{IncomingBso, IncomingContent, OutgoingBso};
+use sync_guid::Guid;
+
+// Here we try and simulate everything done by a "full sync", just minus the
+// engine. Returns the records we uploaded.
+fn do_sync(
+    tx: &Transaction<'_>,
+    incoming_payloads: &[IncomingContent<WebextRecord>],
+) -> Result<Vec<OutgoingBso>> {
+    log::info!("test do_sync() starting");
+    // First we stage the incoming in the temp tables.
+    stage_incoming(tx, incoming_payloads, &NeverInterrupts)?;
+    // Then we process them getting a Vec of (item, state), which we turn into
+    // a Vec of (item, action)
+    let actions = get_incoming(tx)?
+        .into_iter()
+        .map(|(item, state)| (item, plan_incoming(state)))
+        .collect();
+    log::debug!("do_sync applying {:?}", actions);
+    apply_actions(tx, actions, &NeverInterrupts)?;
+    // So we've done incoming - do outgoing.
+    stage_outgoing(tx)?;
+    let outgoing = get_outgoing(tx, &NeverInterrupts)?;
+    log::debug!("do_sync has outgoing {:?}", outgoing);
+    record_uploaded(
+        tx,
+        outgoing
+            .iter()
+            .map(|p| p.envelope.id.clone())
+            .collect::<Vec<Guid>>()
+            .as_slice(),
+        &NeverInterrupts,
+    )?;
+    create_empty_sync_temp_tables(tx)?;
+    log::info!("test do_sync() complete");
+    Ok(outgoing)
+}
+
+// Check *both* the mirror and local API have ended up with the specified data.
+fn check_finished_with(conn: &Connection, ext_id: &str, val: serde_json::Value) -> Result<()> {
+    let local = get(conn, ext_id, serde_json::Value::Null)?;
+    assert_eq!(local, val);
+    let guid = get_mirror_guid(conn, ext_id)?;
+    let mirror = get_mirror_data(conn, &guid);
+    assert_eq!(mirror, DbData::Data(val.to_string()));
+    // and there should be zero items with a change counter.
+    let count = conn.query_row_and_then(
+        "SELECT COUNT(*) FROM storage_sync_data WHERE sync_change_counter != 0;",
+        [],
+        |row| row.get::<_, u32>(0),
+    )?;
+    assert_eq!(count, 0);
+    Ok(())
+}
+
+fn get_mirror_guid(conn: &Connection, extid: &str) -> Result<Guid> {
+    let guid = conn.query_row_and_then(
+        "SELECT m.guid FROM storage_sync_mirror m WHERE m.ext_id = ?;",
+        [extid],
+        |row| row.get::<_, Guid>(0),
+    )?;
+    Ok(guid)
+}
+
+#[derive(Debug, PartialEq)]
+enum DbData {
+    NoRow,
+    NullRow,
+    Data(String),
+}
+
+impl DbData {
+    fn has_data(&self) -> bool {
+        matches!(self, DbData::Data(_))
+    }
+}
+
+fn _get(conn: &Connection, id_name: &str, expected_extid: &str, table: &str) -> DbData {
+    let sql = format!("SELECT {} as id, data FROM {}", id_name, table);
+
+    fn from_row(row: &Row<'_>) -> Result<(String, Option<String>)> {
+        Ok((row.get("id")?, row.get("data")?))
+    }
+    let mut items = conn
+        .conn()
+        .query_rows_and_then(&sql, [], from_row)
+        .expect("should work");
+    if items.is_empty() {
+        DbData::NoRow
+    } else {
+        let item = items.pop().expect("it exists");
+        assert_eq!(Guid::new(&item.0), expected_extid);
+        match item.1 {
+            None => DbData::NullRow,
+            Some(v) => DbData::Data(v),
+        }
+    }
+}
+
+fn get_mirror_data(conn: &Connection, expected_extid: &str) -> DbData {
+    _get(conn, "guid", expected_extid, "storage_sync_mirror")
+}
+
+fn get_local_data(conn: &Connection, expected_extid: &str) -> DbData {
+    _get(conn, "ext_id", expected_extid, "storage_sync_data")
+}
+
+fn make_incoming(
+    guid: &Guid,
+    ext_id: &str,
+    data: &serde_json::Value,
+) -> IncomingContent<WebextRecord> {
+    let content = json!({"id": guid, "extId": ext_id, "data": data.to_string()});
+    IncomingBso::from_test_content(content).into_content()
+}
+
+fn make_incoming_tombstone(guid: &Guid) -> IncomingContent<WebextRecord> {
+    IncomingBso::new_test_tombstone(guid.clone()).into_content()
+}
+
+#[test]
+fn test_simple_outgoing_sync() -> Result<()> {
+    // So we are starting with an empty local store and empty server store.
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    set(&tx, "ext-id", data.clone())?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    check_finished_with(&tx, "ext-id", data)?;
+    Ok(())
+}
+
+#[test]
+fn test_simple_incoming_sync() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    let bridge_record = make_incoming(&Guid::new("guid"), "ext-id", &data);
+    assert_eq!(do_sync(&tx, &[bridge_record])?.len(), 0);
+    let key1_from_api = get(&tx, "ext-id", json!("key1"))?;
+    assert_eq!(key1_from_api, json!({"key1": "key1-value"}));
+    check_finished_with(&tx, "ext-id", data)?;
+    Ok(())
+}
+
+#[test]
+fn test_outgoing_tombstone() -> Result<()> {
+    // Tombstones are only kept when the mirror has that record - so first
+    // test that, then arrange for the mirror to have the record.
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    set(&tx, "ext-id", data.clone())?;
+    assert_eq!(
+        get_local_data(&tx, "ext-id"),
+        DbData::Data(data.to_string())
+    );
+    // hasn't synced yet, so clearing shouldn't write a tombstone.
+    clear(&tx, "ext-id")?;
+    assert_eq!(get_local_data(&tx, "ext-id"), DbData::NoRow);
+    // now set data again and sync and *then* remove.
+    set(&tx, "ext-id", data)?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    assert!(get_local_data(&tx, "ext-id").has_data());
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    assert!(get_mirror_data(&tx, &guid).has_data());
+    clear(&tx, "ext-id")?;
+    assert_eq!(get_local_data(&tx, "ext-id"), DbData::NullRow);
+    // then after syncing, the tombstone will be in the mirror but the local row
+    // has been removed.
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    assert_eq!(get_local_data(&tx, "ext-id"), DbData::NoRow);
+    assert_eq!(get_mirror_data(&tx, &guid), DbData::NullRow);
+    Ok(())
+}
+
+#[test]
+fn test_incoming_tombstone_exists() -> Result<()> {
+    // An incoming tombstone for a record we've previously synced (and thus
+    // have data for)
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    set(&tx, "ext-id", data.clone())?;
+    assert_eq!(
+        get_local_data(&tx, "ext-id"),
+        DbData::Data(data.to_string())
+    );
+    // sync to get data in our mirror.
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    assert!(get_local_data(&tx, "ext-id").has_data());
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    assert!(get_mirror_data(&tx, &guid).has_data());
+    // Now an incoming tombstone for it.
+    let tombstone = make_incoming_tombstone(&guid);
+    assert_eq!(
+        do_sync(&tx, &[tombstone])?.len(),
+        0,
+        "expect no outgoing records"
+    );
+    assert_eq!(get_local_data(&tx, "ext-id"), DbData::NoRow);
+    assert_eq!(get_mirror_data(&tx, &guid), DbData::NullRow);
+    Ok(())
+}
+
+#[test]
+fn test_incoming_tombstone_not_exists() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    // An incoming tombstone for something that's not anywhere locally.
+    let guid = Guid::new("guid");
+    let tombstone = make_incoming_tombstone(&guid);
+    assert_eq!(
+        do_sync(&tx, &[tombstone])?.len(),
+        0,
+        "expect no outgoing records"
+    );
+    // But we still keep the tombstone in the mirror.
+    assert_eq!(get_local_data(&tx, "ext-id"), DbData::NoRow);
+    assert_eq!(get_mirror_data(&tx, &guid), DbData::NullRow);
+    Ok(())
+}
+
+#[test]
+fn test_reconciled() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value"});
+    set(&tx, "ext-id", data)?;
+    // Incoming payload with the same data
+    let record = make_incoming(&Guid::new("guid"), "ext-id", &json!({"key1": "key1-value"}));
+    // Should be no outgoing records as we reconciled.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    check_finished_with(&tx, "ext-id", json!({"key1": "key1-value"}))?;
+    Ok(())
+}
+
+/// Tests that we handle things correctly if we get a payload that is
+/// identical to what is in the mirrored table.
+#[test]
+fn test_reconcile_with_null_payload() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value"});
+    set(&tx, "ext-id", data.clone())?;
+    // We try to push this change on the next sync.
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    assert_eq!(get_mirror_data(&tx, &guid), DbData::Data(data.to_string()));
+    // Incoming payload with the same data.
+    // This could happen if, for example, another client changed the
+    // key and then put it back the way it was.
+    let record = make_incoming(&guid, "ext-id", &data);
+    // Should be no outgoing records as we reconciled.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    check_finished_with(&tx, "ext-id", data)?;
+    Ok(())
+}
+
+#[test]
+fn test_accept_incoming_when_local_is_deleted() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    // We only record an extension as deleted locally if it has been
+    // uploaded before being deleted.
+    let data = json!({"key1": "key1-value"});
+    set(&tx, "ext-id", data)?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    clear(&tx, "ext-id")?;
+    // Incoming payload without 'key1'. Because we previously uploaded
+    // key1, this means another client deleted it.
+    let record = make_incoming(&guid, "ext-id", &json!({"key2": "key2-value"}));
+
+    // We completely accept the incoming record.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    check_finished_with(&tx, "ext-id", json!({"key2": "key2-value"}))?;
+    Ok(())
+}
+
+#[test]
+fn test_accept_incoming_when_local_is_deleted_no_mirror() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value"});
+    set(&tx, "ext-id", data)?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    clear(&tx, "ext-id")?;
+
+    // Use a random guid so that we don't find the mirrored data.
+    // This test is somewhat bad because deduping might obviate
+    // the need for it.
+    let record = make_incoming(&Guid::new("guid"), "ext-id", &json!({"key2": "key2-value"}));
+
+    // We completely accept the incoming record.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    check_finished_with(&tx, "ext-id", json!({"key2": "key2-value"}))?;
+    Ok(())
+}
+
+#[test]
+fn test_accept_deleted_key_mirrored() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    set(&tx, "ext-id", data)?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    // Incoming payload without 'key1'. Because we previously uploaded
+    // key1, this means another client deleted it.
+    let record = make_incoming(&guid, "ext-id", &json!({"key2": "key2-value"}));
+    // We completely accept the incoming record.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    check_finished_with(&tx, "ext-id", json!({"key2": "key2-value"}))?;
+    Ok(())
+}
+
+#[test]
+fn test_merged_no_mirror() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value"});
+    set(&tx, "ext-id", data)?;
+    // Incoming payload without 'key1' and some data for 'key2'.
+    // Because we never uploaded 'key1', we merge our local values
+    // with the remote.
+    let record = make_incoming(&Guid::new("guid"), "ext-id", &json!({"key2": "key2-value"}));
+    assert_eq!(do_sync(&tx, &[record])?.len(), 1);
+    check_finished_with(
+        &tx,
+        "ext-id",
+        json!({"key1": "key1-value", "key2": "key2-value"}),
+    )?;
+    Ok(())
+}
+
+#[test]
+fn test_merged_incoming() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let old_data = json!({"key1": "key1-value", "key2": "key2-value", "doomed_key": "deletable"});
+    set(&tx, "ext-id", old_data)?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    // We update 'key1' locally.
+    let local_data = json!({"key1": "key1-new", "key2": "key2-value", "doomed_key": "deletable"});
+    set(&tx, "ext-id", local_data)?;
+    // Incoming payload where another client set 'key2' and removed
+    // the 'doomed_key'.
+    // Because we never uploaded our data, we'll merge our
+    // key1 in, but otherwise keep the server's changes.
+    let record = make_incoming(
+        &guid,
+        "ext-id",
+        &json!({"key1": "key1-value", "key2": "key2-incoming"}),
+    );
+    // We should send our 'key1'
+    assert_eq!(do_sync(&tx, &[record])?.len(), 1);
+    check_finished_with(
+        &tx,
+        "ext-id",
+        json!({"key1": "key1-new", "key2": "key2-incoming"}),
+    )?;
+    Ok(())
+}
+
+#[test]
+fn test_merged_with_null_payload() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let old_data = json!({"key1": "key1-value"});
+    set(&tx, "ext-id", old_data.clone())?;
+    // Push this change remotely.
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    assert_eq!(
+        get_mirror_data(&tx, &guid),
+        DbData::Data(old_data.to_string())
+    );
+    let local_data = json!({"key1": "key1-new", "key2": "key2-value"});
+    set(&tx, "ext-id", local_data.clone())?;
+    // Incoming payload with the same old data.
+    let record = make_incoming(&guid, "ext-id", &old_data);
+    // Three-way-merge will not detect any change in key1, so we
+    // should keep our entire new value.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 1);
+    check_finished_with(&tx, "ext-id", local_data)?;
+    Ok(())
+}
+
+#[test]
+fn test_deleted_mirrored_object_accept() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    set(&tx, "ext-id", data)?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    // Incoming payload with data deleted.
+    // We synchronize this deletion by deleting the keys we think
+    // were on the server.
+    let record = make_incoming_tombstone(&guid);
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    assert_eq!(get_local_data(&tx, "ext-id"), DbData::NoRow);
+    assert_eq!(get_mirror_data(&tx, &guid), DbData::NullRow);
+    Ok(())
+}
+
+#[test]
+fn test_deleted_mirrored_object_merged() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    set(&tx, "ext-id", json!({"key1": "key1-value"}))?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    set(
+        &tx,
+        "ext-id",
+        json!({"key1": "key1-new", "key2": "key2-value"}),
+    )?;
+    // Incoming payload with data deleted.
+    // We synchronize this deletion by deleting the keys we think
+    // were on the server.
+    let record = make_incoming_tombstone(&guid);
+    // This overrides the change to 'key1', but we still upload 'key2'.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 1);
+    check_finished_with(&tx, "ext-id", json!({"key2": "key2-value"}))?;
+    Ok(())
+}
+
+/// Like the above test, but with a mirrored tombstone.
+#[test]
+fn test_deleted_mirrored_tombstone_merged() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    // Sync some data so we can get the guid for this extension.
+    set(&tx, "ext-id", json!({"key1": "key1-value"}))?;
+    assert_eq!(do_sync(&tx, &[])?.len(), 1);
+    let guid = get_mirror_guid(&tx, "ext-id")?;
+    // Sync a delete for this data so we have a tombstone in the mirror.
+    let record = make_incoming_tombstone(&guid);
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    assert_eq!(get_mirror_data(&tx, &guid), DbData::NullRow);
+
+    // Set some data and sync it simultaneously with another incoming delete.
+    set(&tx, "ext-id", json!({"key2": "key2-value"}))?;
+    let record = make_incoming_tombstone(&guid);
+    // We cannot delete any matching keys because there are no
+    // matching keys. Instead we push our data.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 1);
+    check_finished_with(&tx, "ext-id", json!({"key2": "key2-value"}))?;
+    Ok(())
+}
+
+#[test]
+fn test_deleted_not_mirrored_object_merged() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    set(&tx, "ext-id", data)?;
+    // Incoming payload with data deleted.
+    let record = make_incoming_tombstone(&Guid::new("guid"));
+    // We normally delete the keys we think were on the server, but
+    // here we have no information about what was on the server, so we
+    // don't delete anything. We merge in all undeleted keys.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 1);
+    check_finished_with(
+        &tx,
+        "ext-id",
+        json!({"key1": "key1-value", "key2": "key2-value"}),
+    )?;
+    Ok(())
+}
+
+#[test]
+fn test_conflicting_incoming() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let data = json!({"key1": "key1-value", "key2": "key2-value"});
+    set(&tx, "ext-id", data)?;
+    // Incoming payload without 'key1' and conflicting for 'key2'.
+    // Because we never uploaded either of our keys, we'll merge our
+    // key1 in, but the server key2 wins.
+    let record = make_incoming(
+        &Guid::new("guid"),
+        "ext-id",
+        &json!({"key2": "key2-incoming"}),
+    );
+    // We should send our 'key1'
+    assert_eq!(do_sync(&tx, &[record])?.len(), 1);
+    check_finished_with(
+        &tx,
+        "ext-id",
+        json!({"key1": "key1-value", "key2": "key2-incoming"}),
+    )?;
+    Ok(())
+}
+
+#[test]
+fn test_invalid_incoming() -> Result<()> {
+    let mut db = new_syncable_mem_db();
+    let tx = db.transaction()?;
+    let json = json!({"id": "id", "payload": json!("").to_string()});
+    let bso = serde_json::from_value::<IncomingBso>(json).unwrap();
+    let record = bso.into_content();
+
+    // Should do nothing.
+    assert_eq!(do_sync(&tx, &[record])?.len(), 0);
+    Ok(())
+}