webext_storage/sync/
bridge.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
4
5use anyhow::Result;
6use rusqlite::Transaction;
7use std::sync::{Arc, Weak};
8use sync15::bso::IncomingBso;
9use sync15::engine::{ApplyResults, BridgedEngine as Sync15BridgedEngine};
10use sync_guid::Guid as SyncGuid;
11
12use crate::db::{delete_meta, get_meta, put_meta, ThreadSafeStorageDb};
13use crate::schema;
14use crate::sync::incoming::{apply_actions, get_incoming, plan_incoming, stage_incoming};
15use crate::sync::outgoing::{get_outgoing, record_uploaded, stage_outgoing};
16use crate::WebExtStorageStore;
17
18const LAST_SYNC_META_KEY: &str = "last_sync_time";
19const SYNC_ID_META_KEY: &str = "sync_id";
20
21impl WebExtStorageStore {
22    // Returns a bridged sync engine for this store.
23    pub fn bridged_engine(self: Arc<Self>) -> Arc<WebExtStorageBridgedEngine> {
24        let engine = Box::new(BridgedEngine::new(&self.db));
25        Arc::new(WebExtStorageBridgedEngine::new(engine))
26    }
27}
28
29/// A bridged engine implements all the methods needed to make the
30/// `storage.sync` store work with Desktop's Sync implementation.
31/// Conceptually, it's similar to `sync15::Store`, which we
32/// should eventually rename and unify with this trait (#2841).
33///
34/// Unlike most of our other implementation which hold a strong reference
35/// to the store, this engine keeps a weak reference in an attempt to keep
36/// the desktop semantics as close as possible to what they were when the
37/// engines all took lifetime params to ensure they don't outlive the store.
38pub struct BridgedEngine {
39    db: Weak<ThreadSafeStorageDb>,
40}
41
42impl BridgedEngine {
43    /// Creates a bridged engine for syncing.
44    pub fn new(db: &Arc<ThreadSafeStorageDb>) -> Self {
45        BridgedEngine {
46            db: Arc::downgrade(db),
47        }
48    }
49
50    fn do_reset(&self, tx: &Transaction<'_>) -> Result<()> {
51        tx.execute_batch(
52            "DELETE FROM storage_sync_mirror;
53             UPDATE storage_sync_data SET sync_change_counter = 1;",
54        )?;
55        delete_meta(tx, LAST_SYNC_META_KEY)?;
56        Ok(())
57    }
58
59    fn thread_safe_storage_db(&self) -> Result<Arc<ThreadSafeStorageDb>> {
60        self.db
61            .upgrade()
62            .ok_or_else(|| crate::error::Error::DatabaseConnectionClosed.into())
63    }
64}
65
66impl Sync15BridgedEngine for BridgedEngine {
67    fn last_sync(&self) -> Result<i64> {
68        let shared_db = self.thread_safe_storage_db()?;
69        let db = shared_db.lock();
70        let conn = db.get_connection()?;
71        Ok(get_meta(conn, LAST_SYNC_META_KEY)?.unwrap_or(0))
72    }
73
74    fn set_last_sync(&self, last_sync_millis: i64) -> Result<()> {
75        let shared_db = self.thread_safe_storage_db()?;
76        let db = shared_db.lock();
77        let conn = db.get_connection()?;
78        put_meta(conn, LAST_SYNC_META_KEY, &last_sync_millis)?;
79        Ok(())
80    }
81
82    fn sync_id(&self) -> Result<Option<String>> {
83        let shared_db = self.thread_safe_storage_db()?;
84        let db = shared_db.lock();
85        let conn = db.get_connection()?;
86        Ok(get_meta(conn, SYNC_ID_META_KEY)?)
87    }
88
89    fn reset_sync_id(&self) -> Result<String> {
90        let shared_db = self.thread_safe_storage_db()?;
91        let db = shared_db.lock();
92        let conn = db.get_connection()?;
93        let tx = conn.unchecked_transaction()?;
94        let new_id = SyncGuid::random().to_string();
95        self.do_reset(&tx)?;
96        put_meta(&tx, SYNC_ID_META_KEY, &new_id)?;
97        tx.commit()?;
98        Ok(new_id)
99    }
100
101    fn ensure_current_sync_id(&self, sync_id: &str) -> Result<String> {
102        let shared_db = self.thread_safe_storage_db()?;
103        let db = shared_db.lock();
104        let conn = db.get_connection()?;
105        let current: Option<String> = get_meta(conn, SYNC_ID_META_KEY)?;
106        Ok(match current {
107            Some(current) if current == sync_id => current,
108            _ => {
109                let conn = db.get_connection()?;
110                let tx = conn.unchecked_transaction()?;
111                self.do_reset(&tx)?;
112                let result = sync_id.to_string();
113                put_meta(&tx, SYNC_ID_META_KEY, &result)?;
114                tx.commit()?;
115                result
116            }
117        })
118    }
119
120    fn sync_started(&self) -> Result<()> {
121        let shared_db = self.thread_safe_storage_db()?;
122        let db = shared_db.lock();
123        let conn = db.get_connection()?;
124        schema::create_empty_sync_temp_tables(conn)?;
125        Ok(())
126    }
127
128    fn store_incoming(&self, incoming_bsos: Vec<IncomingBso>) -> Result<()> {
129        let shared_db = self.thread_safe_storage_db()?;
130        let db = shared_db.lock();
131        let signal = db.begin_interrupt_scope()?;
132        let conn = db.get_connection()?;
133        let tx = conn.unchecked_transaction()?;
134        let incoming_content: Vec<_> = incoming_bsos
135            .into_iter()
136            .map(IncomingBso::into_content::<super::WebextRecord>)
137            .collect();
138        stage_incoming(&tx, &incoming_content, &signal)?;
139        tx.commit()?;
140        Ok(())
141    }
142
143    fn apply(&self) -> Result<ApplyResults> {
144        let shared_db = self.thread_safe_storage_db()?;
145        let db = shared_db.lock();
146        let signal = db.begin_interrupt_scope()?;
147        let conn = db.get_connection()?;
148        let tx = conn.unchecked_transaction()?;
149        let incoming = get_incoming(&tx)?;
150        let actions = incoming
151            .into_iter()
152            .map(|(item, state)| (item, plan_incoming(state)))
153            .collect();
154        apply_actions(&tx, actions, &signal)?;
155        stage_outgoing(&tx)?;
156        tx.commit()?;
157
158        Ok(get_outgoing(conn, &signal)?.into())
159    }
160
161    fn set_uploaded(&self, _server_modified_millis: i64, ids: &[SyncGuid]) -> Result<()> {
162        let shared_db = self.thread_safe_storage_db()?;
163        let db = shared_db.lock();
164        let conn = db.get_connection()?;
165        let signal = db.begin_interrupt_scope()?;
166        let tx = conn.unchecked_transaction()?;
167        record_uploaded(&tx, ids, &signal)?;
168        tx.commit()?;
169
170        Ok(())
171    }
172
173    fn sync_finished(&self) -> Result<()> {
174        let shared_db = self.thread_safe_storage_db()?;
175        let db = shared_db.lock();
176        let conn = db.get_connection()?;
177        schema::create_empty_sync_temp_tables(conn)?;
178        Ok(())
179    }
180
181    fn reset(&self) -> Result<()> {
182        let shared_db = self.thread_safe_storage_db()?;
183        let db = shared_db.lock();
184        let conn = db.get_connection()?;
185        let tx = conn.unchecked_transaction()?;
186        self.do_reset(&tx)?;
187        delete_meta(&tx, SYNC_ID_META_KEY)?;
188        tx.commit()?;
189        Ok(())
190    }
191
192    fn wipe(&self) -> Result<()> {
193        let shared_db = self.thread_safe_storage_db()?;
194        let db = shared_db.lock();
195        let conn = db.get_connection()?;
196        let tx = conn.unchecked_transaction()?;
197        // We assume the meta table is only used by sync.
198        tx.execute_batch(
199            "DELETE FROM storage_sync_data; DELETE FROM storage_sync_mirror; DELETE FROM meta;",
200        )?;
201        tx.commit()?;
202        Ok(())
203    }
204}
205
206// The UniFFI-exposed `WebExtStorageBridgedEngine` (a thin newtype around
207// `sync15::engine::BridgedEngineWrapper`) is generated by this macro, which
208// removes the facade + BSO marshalling boilerplate that used to live here. The
209// wrapped engine is the `BridgedEngine` defined above (webext-storage is
210// Desktop-only and implements `BridgedEngine` directly rather than `SyncEngine`).
211// Its `set_uploaded` UDL row is `sequence<Guid>` (a custom type over
212// `sync_guid::Guid`), so the id element type is `sync_guid::Guid`.
213sync15::uniffi_bridged_engine!(WebExtStorageBridgedEngine, sync_guid::Guid);
214
215impl From<anyhow::Error> for crate::error::Error {
216    fn from(value: anyhow::Error) -> Self {
217        crate::error::Error::SyncError(value.to_string())
218    }
219}
220
221#[cfg(test)]
222mod tests {
223    use super::*;
224    use crate::db::test::new_mem_thread_safe_storage_db;
225    use crate::db::StorageDb;
226    use sync15::engine::BridgedEngine;
227
228    fn query_count(db: &StorageDb, table: &str) -> u32 {
229        let conn = db.get_connection().expect("should retrieve connection");
230        conn.query_row_and_then(&format!("SELECT COUNT(*) FROM {};", table), [], |row| {
231            row.get::<_, u32>(0)
232        })
233        .expect("should work")
234    }
235
236    // Sets up mock data for the tests here.
237    fn setup_mock_data(engine: &super::BridgedEngine) -> Result<()> {
238        {
239            let shared = engine.thread_safe_storage_db()?;
240            let db = shared.lock();
241            let conn = db.get_connection().expect("should retrieve connection");
242            conn.execute(
243                "INSERT INTO storage_sync_data (ext_id, data, sync_change_counter)
244                    VALUES ('ext-a', 'invalid-json', 2)",
245                [],
246            )?;
247            conn.execute(
248                "INSERT INTO storage_sync_mirror (guid, ext_id, data)
249                    VALUES ('guid', 'ext-a', '3')",
250                [],
251            )?;
252        }
253        engine.set_last_sync(1)?;
254
255        let shared = engine.thread_safe_storage_db()?;
256        let db = shared.lock();
257        // and assert we wrote what we think we did.
258        assert_eq!(query_count(&db, "storage_sync_data"), 1);
259        assert_eq!(query_count(&db, "storage_sync_mirror"), 1);
260        assert_eq!(query_count(&db, "meta"), 1);
261        Ok(())
262    }
263
264    // Assuming a DB setup with setup_mock_data, assert it was correctly reset.
265    fn assert_reset(engine: &super::BridgedEngine) -> Result<()> {
266        // A reset never wipes data...
267        let shared = engine.thread_safe_storage_db()?;
268        let db = shared.lock();
269        let conn = db.get_connection().expect("should retrieve connection");
270        assert_eq!(query_count(&db, "storage_sync_data"), 1);
271
272        // But did reset the change counter.
273        let cc = conn.query_row_and_then(
274            "SELECT sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext-a';",
275            [],
276            |row| row.get::<_, u32>(0),
277        )?;
278        assert_eq!(cc, 1);
279        // But did wipe the mirror...
280        assert_eq!(query_count(&db, "storage_sync_mirror"), 0);
281        // And the last_sync should have been wiped.
282        assert!(get_meta::<i64>(conn, LAST_SYNC_META_KEY)?.is_none());
283        Ok(())
284    }
285
286    // Assuming a DB setup with setup_mock_data, assert it has not been reset.
287    fn assert_not_reset(engine: &super::BridgedEngine) -> Result<()> {
288        let shared = engine.thread_safe_storage_db()?;
289        let db = shared.lock();
290        let conn = db.get_connection().expect("should retrieve connection");
291        assert_eq!(query_count(&db, "storage_sync_data"), 1);
292        let cc = conn.query_row_and_then(
293            "SELECT sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext-a';",
294            [],
295            |row| row.get::<_, u32>(0),
296        )?;
297        assert_eq!(cc, 2);
298        assert_eq!(query_count(&db, "storage_sync_mirror"), 1);
299        // And the last_sync should remain.
300        assert!(get_meta::<i64>(conn, LAST_SYNC_META_KEY)?.is_some());
301        Ok(())
302    }
303
304    #[test]
305    fn test_wipe() -> Result<()> {
306        let strong = new_mem_thread_safe_storage_db();
307        let engine = super::BridgedEngine::new(&strong);
308
309        setup_mock_data(&engine)?;
310
311        engine.wipe()?;
312
313        let shared = engine.thread_safe_storage_db()?;
314        let db = shared.lock();
315
316        assert_eq!(query_count(&db, "storage_sync_data"), 0);
317        assert_eq!(query_count(&db, "storage_sync_mirror"), 0);
318        assert_eq!(query_count(&db, "meta"), 0);
319        Ok(())
320    }
321
322    #[test]
323    fn test_reset() -> Result<()> {
324        let strong = &new_mem_thread_safe_storage_db();
325        let engine = super::BridgedEngine::new(strong);
326
327        setup_mock_data(&engine)?;
328        {
329            let db = strong.lock();
330            let conn = db.get_connection()?;
331            put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?;
332        }
333
334        engine.reset()?;
335        assert_reset(&engine)?;
336
337        {
338            let db = strong.lock();
339            let conn = db.get_connection()?;
340            // Only an explicit reset kills the sync-id, so check that here.
341            assert_eq!(get_meta::<String>(conn, SYNC_ID_META_KEY)?, None);
342        }
343
344        Ok(())
345    }
346
347    #[test]
348    fn test_ensure_missing_sync_id() -> Result<()> {
349        let strong = new_mem_thread_safe_storage_db();
350        let engine = super::BridgedEngine::new(&strong);
351
352        setup_mock_data(&engine)?;
353
354        assert_eq!(engine.sync_id()?, None);
355        // We don't have a sync ID - so setting one should reset.
356        engine.ensure_current_sync_id("new-id")?;
357        // should have cause a reset.
358        assert_reset(&engine)?;
359        Ok(())
360    }
361
362    #[test]
363    fn test_ensure_new_sync_id() -> Result<()> {
364        let strong = new_mem_thread_safe_storage_db();
365        let engine = super::BridgedEngine::new(&strong);
366
367        setup_mock_data(&engine)?;
368
369        {
370            let storage_db = &engine.thread_safe_storage_db()?;
371            let db = storage_db.lock();
372            let conn = db.get_connection()?;
373            put_meta(conn, SYNC_ID_META_KEY, &"old-id".to_string())?;
374        }
375
376        assert_not_reset(&engine)?;
377        assert_eq!(engine.sync_id()?, Some("old-id".to_string()));
378
379        engine.ensure_current_sync_id("new-id")?;
380        // should have cause a reset.
381        assert_reset(&engine)?;
382        // should have the new id.
383        assert_eq!(engine.sync_id()?, Some("new-id".to_string()));
384        Ok(())
385    }
386
387    #[test]
388    fn test_ensure_same_sync_id() -> Result<()> {
389        let strong = new_mem_thread_safe_storage_db();
390        let engine = super::BridgedEngine::new(&strong);
391
392        setup_mock_data(&engine)?;
393        assert_not_reset(&engine)?;
394
395        {
396            let storage_db = &engine.thread_safe_storage_db()?;
397            let db = storage_db.lock();
398            let conn = db.get_connection()?;
399            put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?;
400        }
401
402        engine.ensure_current_sync_id("sync-id")?;
403        // should not have reset.
404        assert_not_reset(&engine)?;
405        Ok(())
406    }
407
408    #[test]
409    fn test_reset_sync_id() -> Result<()> {
410        let strong = new_mem_thread_safe_storage_db();
411        let engine = super::BridgedEngine::new(&strong);
412
413        setup_mock_data(&engine)?;
414
415        {
416            let storage_db = &engine.thread_safe_storage_db()?;
417            let db = storage_db.lock();
418            let conn = db.get_connection()?;
419            put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?;
420        }
421
422        assert_eq!(engine.sync_id()?, Some("sync-id".to_string()));
423        let new_id = engine.reset_sync_id()?;
424        // should have cause a reset.
425        assert_reset(&engine)?;
426        assert_eq!(engine.sync_id()?, Some(new_id));
427        Ok(())
428    }
429}
webext_storage/sync/bridge.rs

webext_storage/sync/
bridge.rs
