diff --git a/components/autofill/src/sync/engine.rs b/components/autofill/src/sync/engine.rs index 832dd6e528c..fa565755093 100644 --- a/components/autofill/src/sync/engine.rs +++ b/components/autofill/src/sync/engine.rs @@ -105,10 +105,7 @@ impl SyncEngine for ConfigSyncEngine { Ok(()) } - fn prepare_for_sync( - &self, - _get_client_data: &dyn Fn() -> sync15::ClientData, - ) -> anyhow::Result<()> { + fn sync_started(&self) -> anyhow::Result<()> { let db = &self.store.db.lock().unwrap(); let signal = db.begin_interrupt_scope()?; crate::db::schema::create_empty_sync_temp_tables(&db.writer)?; diff --git a/components/logins/src/error.rs b/components/logins/src/error.rs index bc183effee3..7d8c27d2f68 100644 --- a/components/logins/src/error.rs +++ b/components/logins/src/error.rs @@ -196,8 +196,9 @@ impl GetErrorHandling for Error { } // The bridged sync engine (`sync::bridge`) deals in `anyhow::Result`, as that's -// what the `sync15` BridgedEngine traits use. This lets UniFFI map those errors -// onto our public error type when the bridge methods are exposed via the UDL. +// what the `sync15` `SyncEngine`/`BridgedEngineWrapper` use. This lets UniFFI map +// those errors onto our public error type when the bridge methods are exposed +// via the UDL. impl From for LoginsApiError { fn from(value: anyhow::Error) -> Self { LoginsApiError::UnexpectedLoginsApiError { diff --git a/components/logins/src/logins.udl b/components/logins/src/logins.udl index 4d6cb02ba33..bc9cafc625f 100644 --- a/components/logins/src/logins.udl +++ b/components/logins/src/logins.udl @@ -324,8 +324,9 @@ interface LoginStore { void shutdown(); }; -/// The Desktop-facing bridged sync engine. The canonical docs are in -/// https://searchfox.org/mozilla-central/source/services/interfaces/mozIBridgedSyncEngine.idl +/// The Desktop-facing bridged sync engine - a thin wrapper over the +/// `sync15::engine::SyncEngine` implemented by this component (see +/// `sync15::engine::BridgedEngineWrapper`). /// It's only actually used on Desktop, but it's fine to expose this everywhere. /// NOTE: all timestamps here are milliseconds. interface LoginsBridgedEngine { @@ -333,7 +334,7 @@ interface LoginsBridgedEngine { i64 last_sync(); [Throws=LoginsApiError] - void set_last_sync(i64 last_sync); + void reset_last_sync(); [Throws=LoginsApiError] string? sync_id(); @@ -351,7 +352,7 @@ interface LoginsBridgedEngine { void store_incoming(sequence incoming_envelopes_as_json); [Throws=LoginsApiError] - sequence apply(); + sequence apply(i64 server_modified_millis); [Throws=LoginsApiError] void set_uploaded(i64 new_timestamp, sequence uploaded_ids); diff --git a/components/logins/src/sync/bridge.rs b/components/logins/src/sync/bridge.rs index 1fc246691b2..5f0fb7a524d 100644 --- a/components/logins/src/sync/bridge.rs +++ b/components/logins/src/sync/bridge.rs @@ -6,8 +6,6 @@ use crate::sync::engine::LoginsSyncEngine; use crate::LoginStore; use anyhow::Result; use std::sync::Arc; -use sync15::engine::BridgedEngineAdaptor; -use sync15::ServerTimestamp; impl LoginStore { /// Returns a bridged sync engine for Desktop for this store. @@ -18,55 +16,15 @@ impl LoginStore { /// `LoginsApiError` through `From`. pub fn bridged_engine(self: Arc) -> Result> { let engine = LoginsSyncEngine::new(self)?; - let bridged_engine = LoginsBridgedEngineAdaptor { engine }; - Ok(Arc::new(LoginsBridgedEngine::new(Box::new(bridged_engine)))) - } -} - -/// `LoginsSyncEngine` only implements the internal `sync15::SyncEngine` trait, -/// which is what the mobile (Android/iOS) sync manager drives. Desktop's Sync -/// framework instead speaks the `mozIBridgedSyncEngine` interface, whose Rust -/// shape is `sync15::BridgedEngine`. This adaptor wraps our `SyncEngine` and, -/// via the blanket `impl BridgedEngine for A`, gives -/// us a `BridgedEngine` for free. The adaptor exists only because these two -/// sync-engine traits still live side by side; it can go away if they're ever -/// unified. -struct LoginsBridgedEngineAdaptor { - engine: LoginsSyncEngine, -} - -/// see sync15/src/engine/bridged_engine.rs for required functions for the trait -impl BridgedEngineAdaptor for LoginsBridgedEngineAdaptor { - fn last_sync(&self) -> Result { - // `get_last_sync` takes the `&LoginDb` to avoid deadlocking when called - // mid-sync (while the lock is already held). The bridge methods are - // always called outside a sync transaction, so we can lock here. - let db = self.engine.store.lock_db()?; - Ok(self - .engine - .get_last_sync(&db)? - .unwrap_or_default() - .as_millis()) - } - - fn set_last_sync(&self, last_sync_millis: i64) -> Result<()> { - let db = self.engine.store.lock_db()?; - self.engine - .set_last_sync(&db, ServerTimestamp::from_millis(last_sync_millis))?; - Ok(()) - } - - fn engine(&self) -> &dyn sync15::engine::SyncEngine { - &self.engine + Ok(Arc::new(LoginsBridgedEngine::new(Box::new(engine)))) } } // The UniFFI-exposed `LoginsBridgedEngine` (a thin newtype around // `sync15::engine::BridgedEngineWrapper`) is generated by this macro, which -// removes the facade + BSO marshalling boilerplate that used to live here. -// logins' `set_uploaded` UDL row is `sequence`, so the id element type -// is `String`. See services/interfaces/mozIBridgedSyncEngine.idl for the contract. -sync15::uniffi_bridged_engine!(LoginsBridgedEngine, String); +// removes the facade + BSO marshalling boilerplate. The wrapper drives our +// `LoginsSyncEngine`'s `SyncEngine` impl directly. +sync15::uniffi_bridged_engine!(LoginsBridgedEngine); #[cfg(not(feature = "keydb"))] #[cfg(test)] @@ -89,7 +47,7 @@ mod tests { // Fresh DB: never synced. assert_eq!(bridge.last_sync().unwrap(), 0); - bridge.set_last_sync(3).unwrap(); + bridge.set_uploaded(3, vec![]).unwrap(); assert_eq!(bridge.last_sync().unwrap(), 3); assert!(bridge.sync_id().unwrap().is_none()); @@ -98,14 +56,16 @@ mod tests { assert_eq!(bridge.sync_id().unwrap(), Some("some_guid".to_string())); // changing the sync ID should reset the timestamp assert_eq!(bridge.last_sync().unwrap(), 0); - bridge.set_last_sync(3).unwrap(); + // Advance the engine-owned last_sync + bridge.set_uploaded(3, vec![]).unwrap(); bridge.reset_sync_id().unwrap(); // should now be a random guid. assert_ne!(bridge.sync_id().unwrap(), Some("some_guid".to_string())); // should have reset the last sync timestamp. assert_eq!(bridge.last_sync().unwrap(), 0); - bridge.set_last_sync(3).unwrap(); + // Advance the engine-owned last_sync + bridge.set_uploaded(3, vec![]).unwrap(); // `reset` clears the guid and the timestamp bridge.reset().unwrap(); @@ -162,7 +122,7 @@ mod tests { // Applying stores the remote record locally and returns the local-only // login for upload. - let outgoing = bridge.apply().expect("should apply"); + let outgoing = bridge.apply(0).expect("should apply"); let changes: HashMap = outgoing .into_iter() .map(|s| { diff --git a/components/logins/src/sync/engine.rs b/components/logins/src/sync/engine.rs index 2e807fca864..711e57e6cee 100644 --- a/components/logins/src/sync/engine.rs +++ b/components/logins/src/sync/engine.rs @@ -450,7 +450,15 @@ impl SyncEngine for LoginsSyncEngine { telem: &mut telemetry::Engine, ) -> anyhow::Result> { let inbound = self.staged.lock().unwrap().drain(..).collect(); - Ok(self.do_apply_incoming(inbound, timestamp, telem)?) + let outgoing = self.do_apply_incoming(inbound, timestamp, telem)?; + // The engine owns its last-sync timestamp but during a sync, that + // value is known differently in desktop v mobile. Record a + // timestamp if we are given one. + if timestamp != ServerTimestamp(0) { + let db = self.store.lock_db()?; + self.set_last_sync(&db, timestamp)?; + } + Ok(outgoing) } fn set_uploaded(&self, new_timestamp: ServerTimestamp, ids: Vec) -> anyhow::Result<()> { @@ -460,6 +468,21 @@ impl SyncEngine for LoginsSyncEngine { )?) } + // For the Desktop bridge which makes the collection requests. + fn last_sync(&self) -> anyhow::Result> { + let db = self.store.lock_db()?; + Ok(self.get_last_sync(&db)?) + } + + // Force a full re-download next sync without a full reset. Desktop's bridged + // engine base calls this for every engine, so logins must implement it + // rather than fall back to the no-op default. + fn reset_last_sync(&self) -> anyhow::Result<()> { + let db = self.store.lock_db()?; + self.set_last_sync(&db, ServerTimestamp(0))?; + Ok(()) + } + fn get_collection_request( &self, server_timestamp: ServerTimestamp, diff --git a/components/sync15/src/client/sync.rs b/components/sync15/src/client/sync.rs index f680afaa272..cb7ff5668b9 100644 --- a/components/sync15/src/client/sync.rs +++ b/components/sync15/src/client/sync.rs @@ -37,8 +37,9 @@ pub fn synchronize_with_clients_engine( } }; + engine.sync_started()?; if let Some(clients) = clients { - engine.prepare_for_sync(&|| clients.get_client_data())?; + engine.set_clients(&|| clients.get_client_data())?; } interruptee.err_if_interrupted()?; // We assume an "engine" manages exactly one "collection" with the engine's name. diff --git a/components/sync15/src/client_types.rs b/components/sync15/src/client_types.rs index 2966a190e2a..54f21d40835 100644 --- a/components/sync15/src/client_types.rs +++ b/components/sync15/src/client_types.rs @@ -9,8 +9,9 @@ use crate::DeviceType; use serde::{Deserialize, Serialize}; use std::collections::HashMap; -/// Argument to Store::prepare_for_sync. See comment there for more info. Only -/// really intended to be used by tabs engine. +/// Argument to `SyncEngine::set_clients` - a leaky abstraction of fxa/sync +/// device ids. These are "short term" IDs in that they don't survive reauth +/// etc, so used for "recent" things like open tabs. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct ClientData { pub local_client_id: String, diff --git a/components/sync15/src/engine/bridged_engine.rs b/components/sync15/src/engine/bridged_engine.rs index b0280706c09..85e431443c7 100644 --- a/components/sync15/src/engine/bridged_engine.rs +++ b/components/sync15/src/engine/bridged_engine.rs @@ -7,133 +7,62 @@ use crate::{ServerTimestamp, telemetry}; use anyhow::Result; use crate::Guid; -use crate::bso::{IncomingBso, OutgoingBso}; +use crate::bso::IncomingBso; use super::{CollSyncIds, EngineSyncAssociation, SyncEngine}; -/// A BridgedEngine acts as a bridge between application-services, rust -/// implemented sync engines and sync engines as defined by Desktop Firefox. +/// Adapts a [`SyncEngine`] to the method set that Desktop Firefox's JS Sync +/// framework drives (historically the `mozIBridgedSyncEngine` shape). Desktop +/// owns the fetch loop, so unlike the native Rust sync client it reads and +/// writes the engine's last-sync time explicitly and manages sync IDs as opaque +/// strings; this wrapper translates those calls onto the `SyncEngine` trait, and +/// handles the JSON `String` <-> BSO marshalling that crosses the UniFFI +/// boundary. /// -/// [Desktop Firefox has an abstract implementation of a Sync -/// Engine](https://searchfox.org/mozilla-central/source/services/sync/modules/engines.js) -/// with a number of functions each engine is expected to override. Engines -/// implemented in Rust use a different shape (specifically, the -/// [SyncEngine](crate::SyncEngine) trait), so this BridgedEngine trait adapts -/// between the 2. -pub trait BridgedEngine: Send + Sync { - /// Returns the last sync time, in milliseconds, for this engine's - /// collection. This is called before each sync, to determine the lower - /// bound for new records to fetch from the server. - fn last_sync(&self) -> Result; - - /// Sets the last sync time, in milliseconds. This is called throughout - /// the sync, to fast-forward the stored last sync time to match the - /// timestamp on the uploaded records. - fn set_last_sync(&self, last_sync_millis: i64) -> Result<()>; - - /// Returns the sync ID for this engine's collection. This is only used in - /// tests. - fn sync_id(&self) -> Result>; - - /// Resets the sync ID for this engine's collection, returning the new ID. - /// As a side effect, implementations should reset all local Sync state, - /// as in `reset`. - /// (Note that bridged engines never maintain the "global" guid - that's all managed - /// by the bridged_engine consumer (ie, desktop). bridged_engines only care about - /// the per-collection one.) - fn reset_sync_id(&self) -> Result; - - /// Ensures that the locally stored sync ID for this engine's collection - /// matches the `new_sync_id` from the server. If the two don't match, - /// implementations should reset all local Sync state, as in `reset`. - /// This method returns the assigned sync ID, which can be either the - /// `new_sync_id`, or a different one if the engine wants to force other - /// devices to reset their Sync state for this collection the next time they - /// sync. - fn ensure_current_sync_id(&self, new_sync_id: &str) -> Result; - - /// Tells the tabs engine about recent FxA devices. A bit of a leaky abstraction as it only - /// makes sense for tabs. - /// The arg is a json serialized `ClientData` struct. - fn prepare_for_sync(&self, _client_data: &str) -> Result<()> { - Ok(()) - } - - /// Indicates that the engine is about to start syncing. This is called - /// once per sync, and always before `store_incoming`. - fn sync_started(&self) -> Result<()>; - - /// Stages a batch of incoming Sync records. This is called multiple - /// times per sync, once for each batch. Implementations can use the - /// signal to check if the operation was aborted, and cancel any - /// pending work. - fn store_incoming(&self, incoming_records: Vec) -> Result<()>; - - /// Applies all staged records, reconciling changes on both sides and - /// resolving conflicts. Returns a list of records to upload. - fn apply(&self) -> Result; - - /// Indicates that the given record IDs were uploaded successfully to the - /// server. This is called multiple times per sync, once for each batch - /// upload. - fn set_uploaded(&self, server_modified_millis: i64, ids: &[Guid]) -> Result<()>; - - /// Indicates that all records have been uploaded. At this point, any record - /// IDs marked for upload that haven't been passed to `set_uploaded`, can be - /// assumed to have failed: for example, because the server rejected a record - /// with an invalid TTL or sort index. - fn sync_finished(&self) -> Result<()>; - - /// Resets all local Sync state, including any change flags, mirrors, and - /// the last sync time, such that the next sync is treated as a first sync - /// with all new local data. Does not erase any local user data. - fn reset(&self) -> Result<()>; - - /// Erases all local user data for this collection, and any Sync metadata. - /// This method is destructive, and unused for most collections. - fn wipe(&self) -> Result<()>; +/// Consuming crates expose a thin newtype around this via the +/// [`uniffi_bridged_engine!`] macro rather than hand-writing the facade. +/// +/// All methods return [`anyhow::Result`], which each crate maps onto its own +/// UniFFI error type via an `impl From`. +pub struct BridgedEngineWrapper { + inner: Box, } -// This is an adaptor trait - the idea is that engines can implement this -// trait along with SyncEngine and get a BridgedEngine for free. It's temporary -// so we can land this trait without needing to update desktop. -// Longer term, we should remove both this trait and BridgedEngine entirely, sucking up -// the breaking change for desktop. The main blocker to this is moving desktop away -// from the explicit timestamp handling and moving closer to the `get_collection_request` -// model. -pub trait BridgedEngineAdaptor: Send + Sync { - // These are the main mismatches between the 2 engines - fn last_sync(&self) -> Result; - fn set_last_sync(&self, last_sync_millis: i64) -> Result<()>; - fn sync_started(&self) -> Result<()> { - Ok(()) +impl BridgedEngineWrapper { + pub fn new(inner: Box) -> Self { + Self { inner } } - fn engine(&self) -> &dyn SyncEngine; -} - -impl BridgedEngine for A { - fn last_sync(&self) -> Result { - self.last_sync() + /// The last sync time, in milliseconds. Desktop reads this to build the + /// collection URL for fetching incoming records. There is deliberately no + /// setter: the engine owns its last-sync time and advances it itself in + /// `apply`/`set_uploaded`. + pub fn last_sync(&self) -> Result { + Ok(self.inner.last_sync()?.unwrap_or_default().as_millis()) } - fn set_last_sync(&self, last_sync_millis: i64) -> Result<()> { - self.set_last_sync(last_sync_millis) + /// Force a full re-download next sync by resetting the engine-owned + /// `last_sync` timestamp - lighter than a full reset. + pub fn reset_last_sync(&self) -> Result<()> { + self.inner.reset_last_sync() } - fn sync_id(&self) -> Result> { - Ok(match self.engine().get_sync_assoc()? { + /// The per-collection sync ID, derived from the engine's sync association. + /// (Bridged engines never maintain the "global" guid - that's all managed by + /// the consumer, ie, Desktop. They only care about the per-collection one.) + pub fn sync_id(&self) -> Result> { + Ok(match self.inner.get_sync_assoc()? { EngineSyncAssociation::Disconnected => None, EngineSyncAssociation::Connected(c) => Some(c.coll.into()), }) } - fn reset_sync_id(&self) -> Result { - // Note that bridged engines never maintain the "global" guid - that's all managed - // by desktop. bridged_engines only care about the per-collection one. + /// Resets the sync ID for this collection, returning the new ID. As a side + /// effect this resets all local Sync state, as in `reset`. + pub fn reset_sync_id(&self) -> Result { let global = Guid::empty(); let coll = Guid::random(); - self.engine() + self.inner .reset(&EngineSyncAssociation::Connected(CollSyncIds { global, coll: coll.clone(), @@ -141,9 +70,10 @@ impl BridgedEngine for A { Ok(coll.to_string()) } - fn ensure_current_sync_id(&self, sync_id: &str) -> Result { - let engine = self.engine(); - let assoc = engine.get_sync_assoc()?; + /// Ensures the locally stored sync ID matches `sync_id`; resets local Sync + /// state on a mismatch. Returns the assigned sync ID. + pub fn ensure_current_sync_id(&self, sync_id: &str) -> Result { + let assoc = self.inner.get_sync_assoc()?; if matches!(assoc, EngineSyncAssociation::Connected(c) if c.coll == sync_id) { debug!("ensure_current_sync_id is current"); } else { @@ -151,145 +81,17 @@ impl BridgedEngine for A { global: Guid::empty(), coll: sync_id.into(), }; - engine.reset(&EngineSyncAssociation::Connected(new_coll_ids))?; + self.inner + .reset(&EngineSyncAssociation::Connected(new_coll_ids))?; } Ok(sync_id.to_string()) } - fn prepare_for_sync(&self, client_data: &str) -> Result<()> { + pub fn set_clients(&self, client_data: &str) -> Result<()> { // unwrap here is unfortunate, but can hopefully go away if we can // start using the ClientData type instead of the string. - self.engine() - .prepare_for_sync(&|| serde_json::from_str::(client_data).unwrap()) - } - - fn sync_started(&self) -> Result<()> { - A::sync_started(self) - } - - fn store_incoming(&self, incoming_records: Vec) -> Result<()> { - let engine = self.engine(); - let mut telem = telemetry::Engine::new(engine.collection_name()); - engine.stage_incoming(incoming_records, &mut telem) - } - - fn apply(&self) -> Result { - let engine = self.engine(); - let mut telem = telemetry::Engine::new(engine.collection_name()); - // Desktop tells a bridged engine to apply the records without telling it - // the server timestamp, and once applied, explicitly calls `set_last_sync()` - // with that timestamp. So this adaptor needs to call apply with an invalid - // timestamp, and hope that later call with the correct timestamp does come. - // This isn't ideal as it means the timestamp is updated in a different transaction, - // but nothing too bad should happen if it doesn't - we'll just end up applying - // the same records again next sync. - let records = engine.apply(ServerTimestamp::from_millis(0), &mut telem)?; - Ok(ApplyResults { - records, - num_reconciled: telem - .get_incoming() - .as_ref() - .map(|i| i.get_reconciled() as usize), - }) - } - - fn set_uploaded(&self, millis: i64, ids: &[Guid]) -> Result<()> { - self.engine() - .set_uploaded(ServerTimestamp::from_millis(millis), ids.to_vec()) - } - - fn sync_finished(&self) -> Result<()> { - self.engine().sync_finished() - } - - fn reset(&self) -> Result<()> { - self.engine().reset(&EngineSyncAssociation::Disconnected) - } - - fn wipe(&self) -> Result<()> { - self.engine().wipe() - } -} - -// TODO: We should see if we can remove this to reduce the number of types engines need to deal -// with. num_reconciled is only used for telemetry on desktop. -#[derive(Debug, Default)] -pub struct ApplyResults { - /// List of records - pub records: Vec, - /// The number of incoming records whose contents were merged because they - /// changed on both sides. None indicates we aren't reporting this - /// information. - pub num_reconciled: Option, -} - -impl ApplyResults { - pub fn new(records: Vec, num_reconciled: impl Into>) -> Self { - Self { - records, - num_reconciled: num_reconciled.into(), - } - } -} - -// Shorthand for engines that don't care. -impl From> for ApplyResults { - fn from(records: Vec) -> Self { - Self { - records, - num_reconciled: None, - } - } -} - -/// Wraps a `Box` and centralizes the work every consuming -/// crate's UniFFI-facing bridged engine needs to do: the JSON `String` <-> BSO -/// marshalling that crosses the FFI boundary, and 1:1 delegation to the wrapped -/// engine. Rather than each crate hand-writing this (it was ~100 identical lines -/// per crate), they expose a thin newtype around this via the -/// [`uniffi_bridged_engine!`] macro. -/// -/// All methods return [`anyhow::Result`], which each crate maps onto its own -/// UniFFI error type via an `impl From`. -/// -/// Note on the longer-term direction: this type, along with [`BridgedEngine`], -/// [`BridgedEngineAdaptor`] and [`ApplyResults`], only exists because we still -/// have two sync-engine traits. Once Desktop moves off explicit timestamp -/// handling to the `get_collection_request` model (see #2841) we can remove -/// `BridgedEngine` entirely, have Desktop consume [`SyncEngine`] directly, and -/// this wrapper collapses into a thin `SyncEngine` -> FFI shim (or goes away). -/// See the note in `engine/mod.rs` for the migration sequencing. -pub struct BridgedEngineWrapper { - inner: Box, -} - -impl BridgedEngineWrapper { - pub fn new(inner: Box) -> Self { - Self { inner } - } - - pub fn last_sync(&self) -> Result { - self.inner.last_sync() - } - - pub fn set_last_sync(&self, last_sync: i64) -> Result<()> { - self.inner.set_last_sync(last_sync) - } - - pub fn sync_id(&self) -> Result> { - self.inner.sync_id() - } - - pub fn reset_sync_id(&self) -> Result { - self.inner.reset_sync_id() - } - - pub fn ensure_current_sync_id(&self, sync_id: &str) -> Result { - self.inner.ensure_current_sync_id(sync_id) - } - - pub fn prepare_for_sync(&self, client_data: &str) -> Result<()> { - self.inner.prepare_for_sync(client_data) + self.inner + .set_clients(&|| serde_json::from_str::(client_data).unwrap()) } pub fn sync_started(&self) -> Result<()> { @@ -303,31 +105,37 @@ impl BridgedEngineWrapper { for inc in incoming { bsos.push(serde_json::from_str::(&inc)?); } - self.inner.store_incoming(bsos) + let mut telem = telemetry::Engine::new(self.inner.collection_name()); + self.inner.stage_incoming(bsos, &mut telem) } /// Apply staged records and encode the outgoing `OutgoingBso`s back into /// JSON for UniFFI. - pub fn apply(&self) -> Result> { - let apply_results = self.inner.apply()?; - let mut outgoing = Vec::with_capacity(apply_results.records.len()); - for e in apply_results.records { + /// + /// `server_modified_millis` is the collection's server last-modified time, + /// passed explicitly by Desktop (which has just stored it as the last sync + /// time before calling us). It's forwarded to [`SyncEngine::apply`] exactly + /// as the native Rust client does, so reconciliation sees the real + /// timestamp. + pub fn apply(&self, server_modified_millis: i64) -> Result> { + let mut telem = telemetry::Engine::new(self.inner.collection_name()); + let records = self.inner.apply( + ServerTimestamp::from_millis(server_modified_millis), + &mut telem, + )?; + let mut outgoing = Vec::with_capacity(records.len()); + for e in records { outgoing.push(serde_json::to_string(&e)?); } Ok(outgoing) } - /// Accepts anything that turns into a [`Guid`], which reconciles the - /// per-crate id representation: logins hands us `Vec`, while - /// tabs and webext-storage hand us `Vec`. Both `String` - /// and `Guid` implement `Into`. - pub fn set_uploaded>( - &self, - server_modified_millis: i64, - ids: Vec, - ) -> Result<()> { - let guids: Vec = ids.into_iter().map(Into::into).collect(); - self.inner.set_uploaded(server_modified_millis, &guids) + /// The uploaded ids always cross the UniFFI boundary as plain strings; we + /// convert them to [`Guid`] for the engine here. + pub fn set_uploaded(&self, server_modified_millis: i64, ids: Vec) -> Result<()> { + let guids: Vec = ids.into_iter().map(Guid::from).collect(); + self.inner + .set_uploaded(ServerTimestamp::from_millis(server_modified_millis), guids) } pub fn sync_finished(&self) -> Result<()> { @@ -335,7 +143,7 @@ impl BridgedEngineWrapper { } pub fn reset(&self) -> Result<()> { - self.inner.reset() + self.inner.reset(&EngineSyncAssociation::Disconnected) } pub fn wipe(&self) -> Result<()> { @@ -349,28 +157,29 @@ impl BridgedEngineWrapper { /// /// Usage (invoke in the module the crate's UDL `interface` resolves against): /// ```ignore -/// sync15::uniffi_bridged_engine!(LoginsBridgedEngine, String); -/// sync15::uniffi_bridged_engine!(TabsBridgedEngine, sync_guid::Guid); +/// sync15::uniffi_bridged_engine!(LoginsBridgedEngine); +/// sync15::uniffi_bridged_engine!(TabsBridgedEngine); /// ``` /// -/// `$guid` is the element type the crate's UDL lowers `set_uploaded`'s ids to -/// (`String` for logins' `sequence`, `sync_guid::Guid` for the tabs and -/// webext-storage custom-type sequences). The generated methods return +/// All bridged engines expose the same interface; `set_uploaded` takes ids as a +/// plain `sequence` in every crate's UDL. The generated methods return /// `anyhow::Result`, which the crate's UDL `[Throws=...]` maps to its error type /// via the existing `impl From`. /// -/// The macro always emits `prepare_for_sync`; a crate whose UDL doesn't declare -/// it (logins) simply leaves that inherent method unbound, which is harmless. +/// The macro always emits `set_clients`; a crate whose UDL doesn't declare it +/// (logins, webext-storage) simply leaves that inherent method unbound, which is +/// harmless. #[macro_export] macro_rules! uniffi_bridged_engine { - ($name:ident, $guid:ty) => { + ($name:ident) => { // This is what UniFFI exposes; it does nothing other than delegate to - // the shared `BridgedEngineWrapper`. See - // services/interfaces/mozIBridgedSyncEngine.idl for the Desktop contract. + // the shared `BridgedEngineWrapper`, which adapts our `SyncEngine`. pub struct $name($crate::engine::BridgedEngineWrapper); impl $name { - pub fn new(inner: ::std::boxed::Box) -> Self { + pub fn new( + inner: ::std::boxed::Box, + ) -> Self { Self($crate::engine::BridgedEngineWrapper::new(inner)) } @@ -378,8 +187,8 @@ macro_rules! uniffi_bridged_engine { self.0.last_sync() } - pub fn set_last_sync(&self, last_sync: i64) -> ::anyhow::Result<()> { - self.0.set_last_sync(last_sync) + pub fn reset_last_sync(&self) -> ::anyhow::Result<()> { + self.0.reset_last_sync() } pub fn sync_id(&self) -> ::anyhow::Result> { @@ -394,8 +203,8 @@ macro_rules! uniffi_bridged_engine { self.0.ensure_current_sync_id(sync_id) } - pub fn prepare_for_sync(&self, client_data: &str) -> ::anyhow::Result<()> { - self.0.prepare_for_sync(client_data) + pub fn set_clients(&self, client_data: &str) -> ::anyhow::Result<()> { + self.0.set_clients(client_data) } pub fn sync_started(&self) -> ::anyhow::Result<()> { @@ -406,14 +215,14 @@ macro_rules! uniffi_bridged_engine { self.0.store_incoming(incoming) } - pub fn apply(&self) -> ::anyhow::Result> { - self.0.apply() + pub fn apply(&self, server_modified_millis: i64) -> ::anyhow::Result> { + self.0.apply(server_modified_millis) } pub fn set_uploaded( &self, server_modified_millis: i64, - ids: Vec<$guid>, + ids: Vec, ) -> ::anyhow::Result<()> { self.0.set_uploaded(server_modified_millis, ids) } @@ -436,63 +245,61 @@ macro_rules! uniffi_bridged_engine { #[cfg(test)] mod wrapper_tests { use super::*; + use crate::CollectionName; use crate::bso::OutgoingBso; + use crate::engine::CollectionRequest; use std::sync::Mutex; - // A minimal BridgedEngine that records the guids passed to `set_uploaded`, - // so we can lock in the `Into` reconciliation for both `String` and - // `Guid` element types. + // A minimal SyncEngine that records the guids passed to `set_uploaded`, so + // we can confirm the wrapper converts the incoming string ids to `Guid` and + // drives a `SyncEngine`. #[derive(Default)] struct RecordingEngine { uploaded: Mutex>, } - impl BridgedEngine for RecordingEngine { - fn last_sync(&self) -> Result { - Ok(0) + impl SyncEngine for RecordingEngine { + fn collection_name(&self) -> CollectionName { + "test".into() } - fn set_last_sync(&self, _: i64) -> Result<()> { + fn stage_incoming( + &self, + _inbound: Vec, + _telem: &mut telemetry::Engine, + ) -> Result<()> { Ok(()) } - fn sync_id(&self) -> Result> { - Ok(None) - } - fn reset_sync_id(&self) -> Result { - Ok(String::new()) + fn apply( + &self, + _timestamp: ServerTimestamp, + _telem: &mut telemetry::Engine, + ) -> Result> { + Ok(vec![]) } - fn ensure_current_sync_id(&self, id: &str) -> Result { - Ok(id.to_string()) - } - fn sync_started(&self) -> Result<()> { - Ok(()) - } - fn store_incoming(&self, _: Vec) -> Result<()> { + fn set_uploaded(&self, _new_timestamp: ServerTimestamp, ids: Vec) -> Result<()> { + self.uploaded.lock().unwrap().extend(ids); Ok(()) } - fn apply(&self) -> Result { - Ok(Vec::::new().into()) - } - fn set_uploaded(&self, _millis: i64, ids: &[Guid]) -> Result<()> { - self.uploaded.lock().unwrap().extend_from_slice(ids); - Ok(()) - } - fn sync_finished(&self) -> Result<()> { - Ok(()) + fn get_collection_request( + &self, + _server_timestamp: ServerTimestamp, + ) -> Result> { + Ok(None) } - fn reset(&self) -> Result<()> { - Ok(()) + fn get_sync_assoc(&self) -> Result { + Ok(EngineSyncAssociation::Disconnected) } - fn wipe(&self) -> Result<()> { + fn reset(&self, _assoc: &EngineSyncAssociation) -> Result<()> { Ok(()) } } #[test] - fn set_uploaded_accepts_strings_and_guids() { + fn set_uploaded_converts_string_ids() { let wrapper = BridgedEngineWrapper::new(Box::new(RecordingEngine::default())); - // logins-style: Vec - wrapper.set_uploaded(1, vec!["aaaa".to_string()]).unwrap(); - // tabs/webext-style: Vec - wrapper.set_uploaded(2, vec![Guid::new("bbbb")]).unwrap(); + // Every crate now hands us string ids; the wrapper turns them into `Guid`. + wrapper + .set_uploaded(1, vec!["aaaa".to_string(), "bbbb".to_string()]) + .unwrap(); } } diff --git a/components/sync15/src/engine/mod.rs b/components/sync15/src/engine/mod.rs index 6be400677e6..e097d1ae205 100644 --- a/components/sync15/src/engine/mod.rs +++ b/components/sync15/src/engine/mod.rs @@ -12,40 +12,17 @@ //! encryption/decryption - that is the responsbility of the "sync client", as //! implemented in the [client] module (or in some cases, implemented externally) //! -//! There are currently 2 types of engine: -//! * Code which implements the [crate::engine::sync_engine::SyncEngine] -//! trait. These are the "original" Rust engines, designed to be used with -//! the [crate::client](sync client) -//! * Code which implements the [crate::engine::bridged_engine::BridgedEngine] -//! trait. These engines are a "bridge" between the Desktop JS Sync world and -//! this rust code. -//! -//! While these engines end up doing the same thing, the difference is due to -//! implementation differences between the Desktop Sync client and the Rust -//! client. -//! -//! We intend merging these engines - the first step will be to merge the -//! types and payload management used by these traits, then to combine the -//! requirements into a single trait that captures both use-cases. -//! -//! Steps so far, and what's left: -//! * [bridged_engine::BridgedEngineAdaptor] lets a crate implement only -//! [SyncEngine] (plus a tiny adaptor) and get a [bridged_engine::BridgedEngine] -//! for free. -//! * [bridged_engine::BridgedEngineWrapper] + the `uniffi_bridged_engine!` macro -//! remove the per-crate UniFFI facade boilerplate (the JSON<->BSO marshalling -//! and method delegation). -//! * Still to do (#2841): remove `BridgedEngine`/`BridgedEngineAdaptor`/`ApplyResults` -//! entirely and have Desktop consume [SyncEngine] directly. This is blocked on a -//! coordinated mozilla-central change: Desktop must move off explicit timestamp -//! handling (`last_sync`/`set_last_sync`) to the `get_collection_request` model, -//! and the per-crate UDL `interface *BridgedEngine` blocks (the Desktop-visible -//! contract consumed via mozIBridgedSyncEngine) must be updated in lockstep. +//! [SyncEngine](crate::engine::sync_engine::SyncEngine) is a trait which works +//! on desktop and mobile. Engines implement it once and are driven two ways: +//! * On mobile, via the [sync manager](crate::sync_manager). Engines manage +//! their own last-sync time internally. +//! * On Desktop, by the JS Sync framework via `BridgedEngineWrapper` and the +//! `uniffi_bridged_engine!` macro. mod bridged_engine; mod request; mod sync_engine; -pub use bridged_engine::{ApplyResults, BridgedEngine, BridgedEngineAdaptor, BridgedEngineWrapper}; +pub use bridged_engine::BridgedEngineWrapper; #[cfg(feature = "sync-client")] pub(crate) use request::CollectionPost; diff --git a/components/sync15/src/engine/sync_engine.rs b/components/sync15/src/engine/sync_engine.rs index 83b50b866e1..a54d26dd8ad 100644 --- a/components/sync15/src/engine/sync_engine.rs +++ b/components/sync15/src/engine/sync_engine.rs @@ -115,7 +115,7 @@ impl TryFrom<&str> for SyncEngineId { /// record into memory at once (ie, we should try and better reflect the upload batch model at /// this level) /// -/// Sync Engines should not assume they live for exactly one sync, so `prepare_for_sync()` should +/// Sync Engines should not assume they live for exactly one sync, so `sync_started()` should /// clean up any state, including staged records, from previous syncs. /// /// Different engines will produce errors of different types. To accommodate @@ -123,20 +123,14 @@ impl TryFrom<&str> for SyncEngineId { pub trait SyncEngine { fn collection_name(&self) -> CollectionName; - /// Prepares the engine for syncing. The tabs engine currently uses this to - /// store the current list of clients, which it uses to look up device names - /// and types. - /// - /// Note that this method is only called by `sync_multiple`, and only if a - /// command processor is registered. In particular, `prepare_for_sync` will - /// not be called if the store is synced using `sync::synchronize` or - /// `sync_multiple::sync_multiple`. It _will_ be called if the store is - /// synced via the Sync Manager. - /// - /// TODO(issue #2590): This is pretty cludgey and will be hard to extend for - /// any case other than the tabs case. We should find another way to support - /// tabs... - fn prepare_for_sync(&self, _get_client_data: &dyn Fn() -> ClientData) -> Result<()> { + /// Indicates that a sync is starting. + fn sync_started(&self) -> Result<()> { + Ok(()) + } + + /// Supplies the engine with the current set of Sync clients (ie, other + /// devices connected to the account). Might be called at any time. + fn set_clients(&self, _get_client_data: &dyn Fn() -> ClientData) -> Result<()> { Ok(()) } @@ -232,6 +226,19 @@ pub trait SyncEngine { fn wipe(&self) -> Result<()> { unimplemented!("The engine does not implement wipe, no wipe should be requested") } + + /// A couple of desktop specific "bridged engine" helpers, where the + /// last-modified timestamps for collections are handled differently; + /// who does the `get_collection_request()` etc impacts the owner of the + /// timestamp. + /// Engines should do both or neither, longer term it should be absorbed. + fn last_sync(&self) -> Result> { + unimplemented!("This engine is not used as a bridged engine"); + } + + fn reset_last_sync(&self) -> Result<()> { + unimplemented!("This engine is not used as a bridged engine"); + } } #[cfg(test)] diff --git a/components/tabs/src/lib.rs b/components/tabs/src/lib.rs index 84477b538a7..3dd19e0d2fd 100644 --- a/components/tabs/src/lib.rs +++ b/components/tabs/src/lib.rs @@ -22,14 +22,6 @@ uniffi::custom_type!(Timestamp, i64, { uniffi::include_scaffolding!("tabs"); -// Our UDL uses a `Guid` type. -use sync_guid::Guid as TabsGuid; -uniffi::custom_type!(TabsGuid, String, { - remote, - try_lift: |val| Ok(TabsGuid::new(val.as_str())), - lower: |obj| obj.into(), -}); - pub use crate::storage::{ ClientRemoteTabs, LocalTabsInfo, RemoteTabRecord, TabGroup, TabsDeviceType, Window, WindowType, }; diff --git a/components/tabs/src/sync/bridge.rs b/components/tabs/src/sync/bridge.rs index 1f8a5b96775..ebe3998c139 100644 --- a/components/tabs/src/sync/bridge.rs +++ b/components/tabs/src/sync/bridge.rs @@ -4,50 +4,21 @@ use crate::sync::engine::TabsEngine; use crate::TabsStore; -use anyhow::Result; use std::sync::Arc; -use sync15::engine::BridgedEngineAdaptor; -use sync15::ServerTimestamp; impl TabsStore { // Returns a bridged sync engine for Desktop for this store. pub fn bridged_engine(self: Arc) -> Arc { let engine = TabsEngine::new(self); - let bridged_engine = TabsBridgedEngineAdaptor { engine }; - Arc::new(TabsBridgedEngine::new(Box::new(bridged_engine))) - } -} - -/// A bridged engine implements all the methods needed to make the -/// `storage.sync` store work with Desktop's Sync implementation. -/// Conceptually it's very similar to our SyncEngine and there's a BridgedEngineAdaptor -/// trait we can implement to get a `BridgedEngine` from a `SyncEngine`, so that's -/// what we do. See also #2841, which will finally unify them completely. -struct TabsBridgedEngineAdaptor { - engine: TabsEngine, -} - -impl BridgedEngineAdaptor for TabsBridgedEngineAdaptor { - fn last_sync(&self) -> Result { - Ok(self.engine.get_last_sync()?.unwrap_or_default().as_millis()) - } - - fn set_last_sync(&self, last_sync_millis: i64) -> Result<()> { - self.engine - .set_last_sync(ServerTimestamp::from_millis(last_sync_millis)) - } - - fn engine(&self) -> &dyn sync15::engine::SyncEngine { - &self.engine + Arc::new(TabsBridgedEngine::new(Box::new(engine))) } } // The UniFFI-exposed `TabsBridgedEngine` (a thin newtype around // `sync15::engine::BridgedEngineWrapper`) is generated by this macro, which -// removes the facade + BSO marshalling boilerplate that used to live here. -// tabs' `set_uploaded` UDL row is `sequence` (a custom type over -// `sync_guid::Guid`), so the id element type is `sync_guid::Guid`. -sync15::uniffi_bridged_engine!(TabsBridgedEngine, sync_guid::Guid); +// removes the facade + BSO marshalling boilerplate. The wrapper drives our +// `TabsEngine`'s `SyncEngine` impl directly. +sync15::uniffi_bridged_engine!(TabsBridgedEngine); #[cfg(test)] mod tests { @@ -114,7 +85,7 @@ mod tests { ]), }; bridge - .prepare_for_sync(&serde_json::to_string(&client_data).unwrap()) + .set_clients(&serde_json::to_string(&client_data).unwrap()) .expect("should work"); let records = vec![ @@ -178,7 +149,8 @@ mod tests { bridge.store_incoming(incoming).expect("should store"); - let out = bridge.apply().expect("should apply"); + // Incoming records above are `modified: 0` + let out = bridge.apply(0).expect("should apply"); assert_eq!(out.len(), 1); let ours = serde_json::from_str::(&out[0]).unwrap(); @@ -209,7 +181,7 @@ mod tests { // Should not error or panic assert_eq!(bridge.last_sync().unwrap(), 0); - bridge.set_last_sync(3).unwrap(); + bridge.set_uploaded(3, vec![]).unwrap(); assert_eq!(bridge.last_sync().unwrap(), 3); assert!(bridge.sync_id().unwrap().is_none()); @@ -218,14 +190,16 @@ mod tests { assert_eq!(bridge.sync_id().unwrap(), Some("some_guid".to_string())); // changing the sync ID should reset the timestamp assert_eq!(bridge.last_sync().unwrap(), 0); - bridge.set_last_sync(3).unwrap(); + // set_uploaded advances the engine-owned last_sync (there's no external setter). + bridge.set_uploaded(3, vec![]).unwrap(); bridge.reset_sync_id().unwrap(); // should now be a random guid. assert_ne!(bridge.sync_id().unwrap(), Some("some_guid".to_string())); // should have reset the last sync timestamp. assert_eq!(bridge.last_sync().unwrap(), 0); - bridge.set_last_sync(3).unwrap(); + // set_uploaded advances the engine-owned last_sync (there's no external setter). + bridge.set_uploaded(3, vec![]).unwrap(); // `reset` clears the guid and the timestamp bridge.reset().unwrap(); diff --git a/components/tabs/src/sync/engine.rs b/components/tabs/src/sync/engine.rs index 5428cb6c4ef..cc7d7c4dc1c 100644 --- a/components/tabs/src/sync/engine.rs +++ b/components/tabs/src/sync/engine.rs @@ -84,26 +84,32 @@ impl TabsEngine { } } - pub fn set_last_sync(&self, last_sync: ServerTimestamp) -> Result<()> { + // Internally owned last-sync - written internally from `apply`, + // `set_uploaded` and `reset` but otherwise private. + fn set_last_sync(&self, last_sync: ServerTimestamp) -> Result<()> { let mut storage = self.store.storage.lock().unwrap(); debug!("Updating last sync to {}", last_sync); let last_sync_millis = last_sync.as_millis(); Ok(storage.put_meta(schema::LAST_SYNC_META_KEY, &last_sync_millis)?) } +} + +impl SyncEngine for TabsEngine { + fn collection_name(&self) -> CollectionName { + "tabs".into() + } - pub fn get_last_sync(&self) -> Result> { + fn last_sync(&self) -> Result> { let mut storage = self.store.storage.lock().unwrap(); let millis = storage.get_meta::(schema::LAST_SYNC_META_KEY)?; Ok(millis.map(ServerTimestamp)) } -} -impl SyncEngine for TabsEngine { - fn collection_name(&self) -> CollectionName { - "tabs".into() + fn reset_last_sync(&self) -> Result<()> { + self.set_last_sync(ServerTimestamp(0)) } - fn prepare_for_sync(&self, get_client_data: &dyn Fn() -> ClientData) -> Result<()> { + fn set_clients(&self, get_client_data: &dyn Fn() -> ClientData) -> Result<()> { let mut storage = self.store.storage.lock().unwrap(); // We only know the client list at sync time, but need to return tabs potentially // at any time -- so we store the clients in the meta table to be able to properly @@ -232,7 +238,7 @@ impl SyncEngine for TabsEngine { &self, server_timestamp: ServerTimestamp, ) -> Result> { - let since = self.get_last_sync()?.unwrap_or_default(); + let since = self.last_sync()?.unwrap_or_default(); Ok(if since == server_timestamp { None } else { @@ -340,7 +346,7 @@ pub mod test { ]), }; engine - .prepare_for_sync(&|| client_data.clone()) + .set_clients(&|| client_data.clone()) .expect("should work"); let records = vec![ @@ -439,7 +445,7 @@ pub mod test { )]), }; engine - .prepare_for_sync(&|| client_data.clone()) + .set_clients(&|| client_data.clone()) .expect("should work"); let records = vec![json!({ @@ -539,7 +545,7 @@ pub mod test { assert_eq!( engine - .get_last_sync() + .last_sync() .expect("should work") .expect("should have a value"), ServerTimestamp::from_millis(123), diff --git a/components/tabs/src/tabs.udl b/components/tabs/src/tabs.udl index 44e6c19f7dd..48f957107df 100644 --- a/components/tabs/src/tabs.udl +++ b/components/tabs/src/tabs.udl @@ -1,6 +1,3 @@ -[Custom] -typedef string TabsGuid; - // a local timestamp, a custom type to avoid using the int directly. [Custom] typedef i64 Timestamp; @@ -84,7 +81,9 @@ dictionary PendingCommand { Timestamp? time_sent; }; -/// Note the canonical docs for this are in https://searchfox.org/mozilla-central/source/services/interfaces/mozIBridgedSyncEngine.idl +/// The Desktop-facing bridged sync engine - a thin wrapper over the +/// `sync15::engine::SyncEngine` implemented by this component (see +/// `sync15::engine::BridgedEngineWrapper`). /// It's only actually used in desktop, but it's fine to expose this everywhere. /// NOTE: all timestamps here are milliseconds. interface TabsBridgedEngine { @@ -98,7 +97,7 @@ interface TabsBridgedEngine { i64 last_sync(); [Throws=TabsApiError] - void set_last_sync(i64 last_sync); + void reset_last_sync(); [Throws=TabsApiError] string? sync_id(); @@ -110,7 +109,7 @@ interface TabsBridgedEngine { string ensure_current_sync_id([ByRef]string new_sync_id); [Throws=TabsApiError] - void prepare_for_sync([ByRef]string client_data); + void set_clients([ByRef]string client_data); [Throws=TabsApiError] void sync_started(); @@ -119,10 +118,10 @@ interface TabsBridgedEngine { void store_incoming(sequence incoming_envelopes_as_json); [Throws=TabsApiError] - sequence apply(); + sequence apply(i64 server_modified_millis); [Throws=TabsApiError] - void set_uploaded(i64 new_timestamp, sequence uploaded_ids); + void set_uploaded(i64 new_timestamp, sequence uploaded_ids); [Throws=TabsApiError] void sync_finished(); diff --git a/components/webext-storage/src/lib.rs b/components/webext-storage/src/lib.rs index 5db3d59a428..8d410130ba6 100644 --- a/components/webext-storage/src/lib.rs +++ b/components/webext-storage/src/lib.rs @@ -39,10 +39,3 @@ uniffi::custom_type!(JsonValue, String, { lower: |obj| obj.to_string(), }); -// Our UDL uses a `Guid` type. -use sync_guid::Guid; -uniffi::custom_type!(Guid, String, { - remote, - try_lift: |val| Ok(Guid::new(val.as_str())), - lower: |obj| obj.into() -}); diff --git a/components/webext-storage/src/sync/bridge.rs b/components/webext-storage/src/sync/bridge.rs index 502e24d9539..94113d3b67f 100644 --- a/components/webext-storage/src/sync/bridge.rs +++ b/components/webext-storage/src/sync/bridge.rs @@ -5,10 +5,15 @@ use anyhow::Result; use rusqlite::Transaction; use std::sync::{Arc, Weak}; -use sync15::bso::IncomingBso; -use sync15::engine::{ApplyResults, BridgedEngine as Sync15BridgedEngine}; +use sync15::bso::{IncomingBso, OutgoingBso}; +use sync15::engine::{CollSyncIds, CollectionRequest, EngineSyncAssociation, SyncEngine}; +use sync15::{telemetry, CollectionName, ServerTimestamp}; use sync_guid::Guid as SyncGuid; +// The collection name Desktop's Sync framework uses for `storage.sync`. Only +// used for telemetry labelling here (Desktop builds the collection URL itself). +const COLLECTION_NAME: &str = "extension-storage"; + 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}; @@ -21,28 +26,19 @@ const SYNC_ID_META_KEY: &str = "sync_id"; impl WebExtStorageStore { // Returns a bridged sync engine for this store. pub fn bridged_engine(self: Arc) -> Arc { - let engine = Box::new(BridgedEngine::new(&self.db)); + let engine = Box::new(WebExtSyncEngine::new(&self.db)); Arc::new(WebExtStorageBridgedEngine::new(engine)) } } -/// 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 { +pub struct WebExtSyncEngine { db: Weak, } -impl BridgedEngine { +impl WebExtSyncEngine { /// Creates a bridged engine for syncing. pub fn new(db: &Arc) -> Self { - BridgedEngine { + WebExtSyncEngine { db: Arc::downgrade(db), } } @@ -63,57 +59,42 @@ impl BridgedEngine { } } -impl Sync15BridgedEngine for BridgedEngine { - fn last_sync(&self) -> Result { - let shared_db = self.thread_safe_storage_db()?; - let db = shared_db.lock(); - let conn = db.get_connection()?; - Ok(get_meta(conn, LAST_SYNC_META_KEY)?.unwrap_or(0)) +impl SyncEngine for WebExtSyncEngine { + fn collection_name(&self) -> CollectionName { + COLLECTION_NAME.into() } - fn set_last_sync(&self, last_sync_millis: i64) -> Result<()> { + // Read-only view of the engine-owned last-sync time, for the Desktop bridge. + // It's written only internally, in `apply`/`set_uploaded`. + fn last_sync(&self) -> Result> { let shared_db = self.thread_safe_storage_db()?; let db = shared_db.lock(); let conn = db.get_connection()?; - put_meta(conn, LAST_SYNC_META_KEY, &last_sync_millis)?; - Ok(()) + Ok(get_meta::(conn, LAST_SYNC_META_KEY)?.map(ServerTimestamp)) } - fn sync_id(&self) -> Result> { - let shared_db = self.thread_safe_storage_db()?; - let db = shared_db.lock(); - let conn = db.get_connection()?; - Ok(get_meta(conn, SYNC_ID_META_KEY)?) - } - - fn reset_sync_id(&self) -> Result { + fn reset_last_sync(&self) -> Result<()> { let shared_db = self.thread_safe_storage_db()?; let db = shared_db.lock(); let conn = db.get_connection()?; let tx = conn.unchecked_transaction()?; - let new_id = SyncGuid::random().to_string(); - self.do_reset(&tx)?; - put_meta(&tx, SYNC_ID_META_KEY, &new_id)?; + delete_meta(&tx, LAST_SYNC_META_KEY)?; tx.commit()?; - Ok(new_id) + Ok(()) } - fn ensure_current_sync_id(&self, sync_id: &str) -> Result { + fn get_sync_assoc(&self) -> Result { let shared_db = self.thread_safe_storage_db()?; let db = shared_db.lock(); let conn = db.get_connection()?; - let current: Option = get_meta(conn, SYNC_ID_META_KEY)?; - Ok(match current { - Some(current) if current == sync_id => current, - _ => { - let conn = db.get_connection()?; - let tx = conn.unchecked_transaction()?; - self.do_reset(&tx)?; - let result = sync_id.to_string(); - put_meta(&tx, SYNC_ID_META_KEY, &result)?; - tx.commit()?; - result - } + // Bridged engines never maintain the "global" guid - that's all managed + // by the consumer (Desktop); they only care about the per-collection one. + Ok(match get_meta::(conn, SYNC_ID_META_KEY)? { + Some(coll) => EngineSyncAssociation::Connected(CollSyncIds { + global: SyncGuid::empty(), + coll: coll.into(), + }), + None => EngineSyncAssociation::Disconnected, }) } @@ -125,7 +106,11 @@ impl Sync15BridgedEngine for BridgedEngine { Ok(()) } - fn store_incoming(&self, incoming_bsos: Vec) -> Result<()> { + fn stage_incoming( + &self, + incoming_bsos: Vec, + _telem: &mut telemetry::Engine, + ) -> Result<()> { let shared_db = self.thread_safe_storage_db()?; let db = shared_db.lock(); let signal = db.begin_interrupt_scope()?; @@ -140,7 +125,11 @@ impl Sync15BridgedEngine for BridgedEngine { Ok(()) } - fn apply(&self) -> Result { + fn apply( + &self, + timestamp: ServerTimestamp, + _telem: &mut telemetry::Engine, + ) -> Result> { let shared_db = self.thread_safe_storage_db()?; let db = shared_db.lock(); let signal = db.begin_interrupt_scope()?; @@ -153,18 +142,28 @@ impl Sync15BridgedEngine for BridgedEngine { .collect(); apply_actions(&tx, actions, &signal)?; stage_outgoing(&tx)?; + // The engine owns its last-sync time: record the collection timestamp we + // just synced to, so it advances without any external `set_last_sync`. + // (Timestamp is zero only in an upload-only path, which must not move it.) + if timestamp != ServerTimestamp(0) { + put_meta(&tx, LAST_SYNC_META_KEY, ×tamp.as_millis())?; + } tx.commit()?; - Ok(get_outgoing(conn, &signal)?.into()) + Ok(get_outgoing(conn, &signal)?) } - fn set_uploaded(&self, _server_modified_millis: i64, ids: &[SyncGuid]) -> Result<()> { + fn set_uploaded(&self, new_timestamp: ServerTimestamp, ids: Vec) -> Result<()> { let shared_db = self.thread_safe_storage_db()?; let db = shared_db.lock(); let conn = db.get_connection()?; let signal = db.begin_interrupt_scope()?; let tx = conn.unchecked_transaction()?; - record_uploaded(&tx, ids, &signal)?; + record_uploaded(&tx, &ids, &signal)?; + // Advance the engine-owned last-sync time to the post-upload timestamp. + if new_timestamp != ServerTimestamp(0) { + put_meta(&tx, LAST_SYNC_META_KEY, &new_timestamp.as_millis())?; + } tx.commit()?; Ok(()) @@ -178,13 +177,41 @@ impl Sync15BridgedEngine for BridgedEngine { Ok(()) } - fn reset(&self) -> Result<()> { + fn get_collection_request( + &self, + server_timestamp: ServerTimestamp, + ) -> Result> { + let shared_db = self.thread_safe_storage_db()?; + let db = shared_db.lock(); + let conn = db.get_connection()?; + let since = ServerTimestamp(get_meta::(conn, LAST_SYNC_META_KEY)?.unwrap_or(0)); + Ok(if since == server_timestamp { + None + } else { + Some( + CollectionRequest::new(COLLECTION_NAME.into()) + .full() + .newer_than(since), + ) + }) + } + + fn reset(&self, assoc: &EngineSyncAssociation) -> Result<()> { let shared_db = self.thread_safe_storage_db()?; let db = shared_db.lock(); let conn = db.get_connection()?; let tx = conn.unchecked_transaction()?; self.do_reset(&tx)?; - delete_meta(&tx, SYNC_ID_META_KEY)?; + // A `Disconnected` reset clears the sync ID; a `Connected` one adopts the + // (per-collection) ID. `do_reset` already cleared the last sync time. + match assoc { + EngineSyncAssociation::Disconnected => { + delete_meta(&tx, SYNC_ID_META_KEY)?; + } + EngineSyncAssociation::Connected(ids) => { + put_meta(&tx, SYNC_ID_META_KEY, &ids.coll.to_string())?; + } + } tx.commit()?; Ok(()) } @@ -205,12 +232,11 @@ impl Sync15BridgedEngine for BridgedEngine { // The UniFFI-exposed `WebExtStorageBridgedEngine` (a thin newtype around // `sync15::engine::BridgedEngineWrapper`) is generated by this macro, which -// removes the facade + BSO marshalling boilerplate that used to live here. The -// wrapped engine is the `BridgedEngine` defined above (webext-storage is -// Desktop-only and implements `BridgedEngine` directly rather than `SyncEngine`). -// Its `set_uploaded` UDL row is `sequence` (a custom type over -// `sync_guid::Guid`), so the id element type is `sync_guid::Guid`. -sync15::uniffi_bridged_engine!(WebExtStorageBridgedEngine, sync_guid::Guid); +// removes the facade + BSO marshalling boilerplate. The wrapper drives the +// `SyncEngine` impl on the `BridgedEngine` defined above (webext-storage is +// Desktop-only, but implements the one unified `SyncEngine` trait like everyone +// else). +sync15::uniffi_bridged_engine!(WebExtStorageBridgedEngine); impl From for crate::error::Error { fn from(value: anyhow::Error) -> Self { @@ -223,7 +249,16 @@ mod tests { use super::*; use crate::db::test::new_mem_thread_safe_storage_db; use crate::db::StorageDb; - use sync15::engine::BridgedEngine; + use sync15::engine::BridgedEngineWrapper; + + // The sync-ID and reset semantics that used to live on the old + // `BridgedEngine` trait now live on `BridgedEngineWrapper` (which drives our + // `SyncEngine`), so we exercise them the same way Desktop does - through the + // wrapper. Each engine holds a `Weak` to the shared db, so callers keep the + // strong `Arc` alive and inspect DB state through it directly. + fn wrapper(db: &Arc) -> BridgedEngineWrapper { + BridgedEngineWrapper::new(Box::new(WebExtSyncEngine::new(db))) + } fn query_count(db: &StorageDb, table: &str) -> u32 { let conn = db.get_connection().expect("should retrieve connection"); @@ -234,11 +269,10 @@ mod tests { } // Sets up mock data for the tests here. - fn setup_mock_data(engine: &super::BridgedEngine) -> Result<()> { + fn setup_mock_data(db: &Arc) -> Result<()> { { - let shared = engine.thread_safe_storage_db()?; - let db = shared.lock(); - let conn = db.get_connection().expect("should retrieve connection"); + let shared = db.lock(); + let conn = shared.get_connection().expect("should retrieve connection"); conn.execute( "INSERT INTO storage_sync_data (ext_id, data, sync_change_counter) VALUES ('ext-a', 'invalid-json', 2)", @@ -250,24 +284,27 @@ mod tests { [], )?; } - engine.set_last_sync(1)?; + // Seed a last-sync time directly - there's no public setter for it. + { + let shared = db.lock(); + let conn = shared.get_connection().expect("should retrieve connection"); + put_meta(conn, LAST_SYNC_META_KEY, &1i64)?; + } - let shared = engine.thread_safe_storage_db()?; - let db = shared.lock(); + let shared = db.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); + assert_eq!(query_count(&shared, "storage_sync_data"), 1); + assert_eq!(query_count(&shared, "storage_sync_mirror"), 1); + assert_eq!(query_count(&shared, "meta"), 1); Ok(()) } // Assuming a DB setup with setup_mock_data, assert it was correctly reset. - fn assert_reset(engine: &super::BridgedEngine) -> Result<()> { + fn assert_reset(db: &Arc) -> Result<()> { // A reset never wipes data... - let shared = engine.thread_safe_storage_db()?; - let db = shared.lock(); - let conn = db.get_connection().expect("should retrieve connection"); - assert_eq!(query_count(&db, "storage_sync_data"), 1); + let shared = db.lock(); + let conn = shared.get_connection().expect("should retrieve connection"); + assert_eq!(query_count(&shared, "storage_sync_data"), 1); // But did reset the change counter. let cc = conn.query_row_and_then( @@ -277,25 +314,24 @@ mod tests { )?; assert_eq!(cc, 1); // But did wipe the mirror... - assert_eq!(query_count(&db, "storage_sync_mirror"), 0); + assert_eq!(query_count(&shared, "storage_sync_mirror"), 0); // And the last_sync should have been wiped. assert!(get_meta::(conn, 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(); - let conn = db.get_connection().expect("should retrieve connection"); - assert_eq!(query_count(&db, "storage_sync_data"), 1); + fn assert_not_reset(db: &Arc) -> Result<()> { + let shared = db.lock(); + let conn = shared.get_connection().expect("should retrieve connection"); + assert_eq!(query_count(&shared, "storage_sync_data"), 1); let cc = conn.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); + assert_eq!(query_count(&shared, "storage_sync_mirror"), 1); // And the last_sync should remain. assert!(get_meta::(conn, LAST_SYNC_META_KEY)?.is_some()); Ok(()) @@ -304,15 +340,11 @@ mod tests { #[test] fn test_wipe() -> Result<()> { let strong = new_mem_thread_safe_storage_db(); - let engine = super::BridgedEngine::new(&strong); - - setup_mock_data(&engine)?; + setup_mock_data(&strong)?; - engine.wipe()?; - - let shared = engine.thread_safe_storage_db()?; - let db = shared.lock(); + wrapper(&strong).wipe()?; + let db = strong.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); @@ -321,18 +353,16 @@ mod tests { #[test] fn test_reset() -> Result<()> { - let strong = &new_mem_thread_safe_storage_db(); - let engine = super::BridgedEngine::new(strong); - - setup_mock_data(&engine)?; + let strong = new_mem_thread_safe_storage_db(); + setup_mock_data(&strong)?; { let db = strong.lock(); let conn = db.get_connection()?; put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?; } - engine.reset()?; - assert_reset(&engine)?; + wrapper(&strong).reset()?; + assert_reset(&strong)?; { let db = strong.lock(); @@ -347,83 +377,72 @@ mod tests { #[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(&strong)?; - setup_mock_data(&engine)?; - - assert_eq!(engine.sync_id()?, None); + assert_eq!(wrapper(&strong).sync_id()?, None); // We don't have a sync ID - so setting one should reset. - engine.ensure_current_sync_id("new-id")?; + wrapper(&strong).ensure_current_sync_id("new-id")?; // should have cause a reset. - assert_reset(&engine)?; + assert_reset(&strong)?; 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)?; + setup_mock_data(&strong)?; { - let storage_db = &engine.thread_safe_storage_db()?; - let db = storage_db.lock(); + let db = strong.lock(); let conn = db.get_connection()?; put_meta(conn, SYNC_ID_META_KEY, &"old-id".to_string())?; } - assert_not_reset(&engine)?; - assert_eq!(engine.sync_id()?, Some("old-id".to_string())); + assert_not_reset(&strong)?; + assert_eq!(wrapper(&strong).sync_id()?, Some("old-id".to_string())); - engine.ensure_current_sync_id("new-id")?; + wrapper(&strong).ensure_current_sync_id("new-id")?; // should have cause a reset. - assert_reset(&engine)?; + assert_reset(&strong)?; // should have the new id. - assert_eq!(engine.sync_id()?, Some("new-id".to_string())); + assert_eq!(wrapper(&strong).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)?; + setup_mock_data(&strong)?; + assert_not_reset(&strong)?; { - let storage_db = &engine.thread_safe_storage_db()?; - let db = storage_db.lock(); + let db = strong.lock(); let conn = db.get_connection()?; put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?; } - engine.ensure_current_sync_id("sync-id")?; + wrapper(&strong).ensure_current_sync_id("sync-id")?; // should not have reset. - assert_not_reset(&engine)?; + assert_not_reset(&strong)?; 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)?; + setup_mock_data(&strong)?; { - let storage_db = &engine.thread_safe_storage_db()?; - let db = storage_db.lock(); + let db = strong.lock(); let conn = db.get_connection()?; put_meta(conn, 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()?; + assert_eq!(wrapper(&strong).sync_id()?, Some("sync-id".to_string())); + let new_id = wrapper(&strong).reset_sync_id()?; // should have cause a reset. - assert_reset(&engine)?; - assert_eq!(engine.sync_id()?, Some(new_id)); + assert_reset(&strong)?; + assert_eq!(wrapper(&strong).sync_id()?, Some(new_id)); Ok(()) } } diff --git a/components/webext-storage/src/webext-storage.udl b/components/webext-storage/src/webext-storage.udl index 4925008abf1..991fa46085b 100644 --- a/components/webext-storage/src/webext-storage.udl +++ b/components/webext-storage/src/webext-storage.udl @@ -5,9 +5,6 @@ [Custom] typedef string JsonValue; -[Custom] -typedef string Guid; - namespace webextstorage { }; @@ -79,7 +76,7 @@ interface WebExtStorageBridgedEngine { i64 last_sync(); [Throws=WebExtStorageApiError] - void set_last_sync(i64 last_sync); + void reset_last_sync(); [Throws=WebExtStorageApiError] string? sync_id(); @@ -90,9 +87,6 @@ interface WebExtStorageBridgedEngine { [Throws=WebExtStorageApiError] string ensure_current_sync_id([ByRef]string new_sync_id); - [Throws=WebExtStorageApiError] - void prepare_for_sync([ByRef]string client_data); - [Throws=WebExtStorageApiError] void sync_started(); @@ -100,10 +94,10 @@ interface WebExtStorageBridgedEngine { void store_incoming(sequence incoming); [Throws=WebExtStorageApiError] - sequence apply(); + sequence apply(i64 server_modified_millis); [Throws=WebExtStorageApiError] - void set_uploaded(i64 server_modified_millis, sequence guids); + void set_uploaded(i64 server_modified_millis, sequence guids); [Throws=WebExtStorageApiError] void sync_finished(); diff --git a/examples/tabs-sync/src/tabs-sync.rs b/examples/tabs-sync/src/tabs-sync.rs index c5fddc63c1b..a4f6decfd73 100644 --- a/examples/tabs-sync/src/tabs-sync.rs +++ b/examples/tabs-sync/src/tabs-sync.rs @@ -63,7 +63,7 @@ fn do_sync( // Since we are syncing without the sync manager, there's no // command processor, therefore no clients engine, and in - // consequence `TabsStore::prepare_for_sync` is never called + // consequence `TabsEngine::set_clients` is never called // which means our `local_id` will never be set. // Do it here. *engine.local_id.write().unwrap() = local_id;