Bug 1641005 - Implement `mozISyncedExtensionStorageArea.fetchPendingSyncChanges()`. r=markh

This new method fetches pending synced changes from the extension
storage store, and passes them to `storage.onChanged` listeners.
This allows extensions that listen for these events to know when a
sync happened, which Kinto supported as well.

To guard against misuse, this method is implemented on a separate
`mozISyncedExtensionStorageArea` interface. To avoid multiple
inheritance (if `mozI{Synced, Configurable}ExtensionStorageArea` both
inherited from `mozIExtensionStorageArea`, which base method is called
when?), both of these internal-ish interfaces now inherit from
`nsISupports` instead.

Finally, because `fetchPendingSyncChanges` can return changes for
multiple extensions, `mozIExtensionStorageListener.onChanged` now takes
the affected extension ID as its first argument.

Differential Revision: https://phabricator.services.mozilla.com/D76976

services/sync/modules/engines/extension-storage.js

@@ -15,7 +15,7 @@ const { XPCOMUtils } = ChromeUtils.import(
 
 XPCOMUtils.defineLazyModuleGetters(this, {
   BridgedEngine: "resource://services-sync/bridged_engine.js",
-  extensionStorageSync: "resource://gre/modules/ExtensionStorageSyncKinto.jsm",
+  extensionStorageSync: "resource://gre/modules/ExtensionStorageSync.jsm",
   Svc: "resource://services-sync/util.js",
   SyncEngine: "resource://services-sync/engines.js",
   Tracker: "resource://services-sync/engines.js",
@@ -23,6 +23,13 @@ XPCOMUtils.defineLazyModuleGetters(this, {
   MULTI_DEVICE_THRESHOLD: "resource://services-sync/constants.js",
 });
 
+XPCOMUtils.defineLazyModuleGetter(
+  this,
+  "extensionStorageSyncKinto",
+  "resource://gre/modules/ExtensionStorageSyncKinto.jsm",
+  "extensionStorageSync"
+);
+
 XPCOMUtils.defineLazyServiceGetter(
   this,
   "StorageSyncService",
@@ -73,6 +80,48 @@ ExtensionStorageEngineBridge.prototype = {
   // we don't support repair at all!
   _skipPercentageChance: 100,
 
+  _notifyPendingChanges() {
+    return new Promise(resolve => {
+      this._bridge
+        .QueryInterface(Ci.mozISyncedExtensionStorageArea)
+        .fetchPendingSyncChanges({
+          QueryInterface: ChromeUtils.generateQI([
+            Ci.mozIExtensionStorageListener,
+            Ci.mozIExtensionStorageCallback,
+          ]),
+          onChanged: (extId, json) => {
+            try {
+              extensionStorageSync.notifyListeners(extId, JSON.parse(json));
+            } catch (ex) {
+              this._log.warn(
+                `Error notifying change listeners for ${extId}`,
+                ex
+              );
+            }
+          },
+          handleSuccess: resolve,
+          handleError: (code, message) => {
+            this._log.warn(
+              "Error fetching pending synced changes",
+              message,
+              code
+            );
+          },
+        });
+    });
+  },
+
+  async _processIncoming() {
+    await super._processIncoming();
+    try {
+      await this._notifyPendingChanges();
+    } catch (ex) {
+      // Failing to notify `storage.onChanged` observers is bad, but shouldn't
+      // interrupt syncing.
+      this._log.warn("Error notifying about synced changes", ex);
+    }
+  },
+
   get enabled() {
     return getEngineEnabled();
   },
@@ -117,7 +166,7 @@ ExtensionStorageEngineKinto.prototype = {
   allowSkippedRecord: false,
 
   async _sync() {
-    return extensionStorageSync.syncAll();
+    return extensionStorageSyncKinto.syncAll();
   },
 
   get enabled() {
@@ -131,7 +180,7 @@ ExtensionStorageEngineKinto.prototype = {
   },
 
   _wipeClient() {
-    return extensionStorageSync.clearAll();
+    return extensionStorageSyncKinto.clearAll();
   },
 
   shouldSkipSync(syncReason) {

toolkit/components/extensions/ExtensionStorageSync.jsm

@@ -38,10 +38,9 @@ XPCOMUtils.defineLazyGetter(this, "storageSvc", () =>
 
 // The interfaces which define the callbacks used by the bridge. There's a
 // callback for success, failure, and to record data changes.
-function ExtensionStorageApiCallback(resolve, reject, extId, changeCallback) {
+function ExtensionStorageApiCallback(resolve, reject, changeCallback) {
   this.resolve = resolve;
   this.reject = reject;
-  this.extId = extId;
   this.changeCallback = changeCallback;
 }
 
@@ -70,10 +69,10 @@ ExtensionStorageApiCallback.prototype = {
     this.reject(new ExtensionUtils.ExtensionError(sanitized));
   },
 
-  onChanged(json) {
+  onChanged(extId, json) {
     if (this.changeCallback && json) {
       try {
-        this.changeCallback(this.extId, JSON.parse(json));
+        this.changeCallback(extId, JSON.parse(json));
       } catch (ex) {
         Cu.reportError(ex);
       }
@@ -91,8 +90,7 @@ class ExtensionStorageSync {
   // * Ensures the API is allowed to be used.
   // * Works out what "extension id" to use.
   // * Turns the callback API into a promise API.
-  async _promisify(fn, extension, ...args) {
-    let extId = extension.id;
+  async _promisify(fn, ...args) {
     if (prefPermitsStorageSync !== true) {
       throw new ExtensionUtils.ExtensionError(
         `Please set ${STORAGE_SYNC_ENABLED_PREF} to true in about:config`
@@ -102,27 +100,30 @@ class ExtensionStorageSync {
       let callback = new ExtensionStorageApiCallback(
         resolve,
         reject,
-        extId,
         (extId, changes) => this.notifyListeners(extId, changes)
       );
-      fn(extId, ...args, callback);
+      fn(...args, callback);
     });
   }
 
   set(extension, items, context) {
-    return this._promisify(storageSvc.set, extension, JSON.stringify(items));
+    let extId = extension.id;
+    return this._promisify(storageSvc.set, extId, JSON.stringify(items));
   }
 
   remove(extension, keys, context) {
-    return this._promisify(storageSvc.remove, extension, JSON.stringify(keys));
+    let extId = extension.id;
+    return this._promisify(storageSvc.remove, extId, JSON.stringify(keys));
   }
 
   clear(extension, context) {
-    return this._promisify(storageSvc.clear, extension);
+    let extId = extension.id;
+    return this._promisify(storageSvc.clear, extId);
   }
 
   get(extension, spec, context) {
-    return this._promisify(storageSvc.get, extension, JSON.stringify(spec));
+    let extId = extension.id;
+    return this._promisify(storageSvc.get, extId, JSON.stringify(spec));
   }
 
   getBytesInUse(extension, keys, context) {

toolkit/components/extensions/storage/ExtensionStorageComponents.h

@@ -11,8 +11,7 @@
 extern "C" {
 
 // Implemented in Rust, in the `webext_storage_bridge` crate.
-nsresult NS_NewExtensionStorageSyncArea(
-    mozIConfigurableExtensionStorageArea** aResult);
+nsresult NS_NewExtensionStorageSyncArea(mozIExtensionStorageArea** aResult);
 
 }  // extern "C"
 
@@ -25,8 +24,8 @@ namespace storage {
 // `already_AddRefed<T>`, but Rust doesn't have such a type. So we call the
 // Rust constructor using a `nsCOMPtr` (which is compatible with Rust's
 // `xpcom::RefPtr`) out param, and return that.
-already_AddRefed<mozIConfigurableExtensionStorageArea> NewSyncArea() {
-  nsCOMPtr<mozIConfigurableExtensionStorageArea> storage;
+already_AddRefed<mozIExtensionStorageArea> NewSyncArea() {
+  nsCOMPtr<mozIExtensionStorageArea> storage;
   nsresult rv = NS_NewExtensionStorageSyncArea(getter_AddRefs(storage));
   if (NS_WARN_IF(NS_FAILED(rv))) {
     return nullptr;

toolkit/components/extensions/storage/components.conf

@@ -8,7 +8,7 @@ Classes = [
     {
         'cid': '{f1e424f2-67fe-4f69-a8f8-3993a71f44fa}',
         'contract_ids': ['@mozilla.org/extensions/storage/internal/sync-area;1'],
-        'type': 'mozIConfigurableExtensionStorageArea',
+        'type': 'mozIExtensionStorageArea',
         'headers': ['mozilla/extensions/storage/ExtensionStorageComponents.h'],
         'constructor': 'mozilla::extensions::storage::NewSyncArea',
     },

toolkit/components/extensions/storage/mozIExtensionStorageArea.idl

@@ -61,10 +61,12 @@ interface mozIExtensionStorageArea : nsISupports {
                        in mozIExtensionStorageCallback callback);
 };
 
-// A configurable storage area has additional methods for setting up and tearing
-// down its underlying database connection.
+// Implements additional methods for setting up and tearing down the underlying
+// database connection for a storage area. This is a separate interface because
+// these methods are not part of the `StorageArea` API, and have restrictions on
+// when they can be called.
 [scriptable, uuid(2b008295-1bcc-4610-84f1-ad4cab2fa9ee)]
-interface mozIConfigurableExtensionStorageArea : mozIExtensionStorageArea {
+interface mozIConfigurableExtensionStorageArea : nsISupports {
     // Sets up the storage area. An area can only be configured once; calling
     // `configure` multiple times will throw. `configure` must also be called
     // before any of the `mozIExtensionStorageArea` methods, or they'll fail
@@ -78,12 +80,28 @@ interface mozIConfigurableExtensionStorageArea : mozIExtensionStorageArea {
     void teardown(in mozIExtensionStorageCallback callback);
 };
 
+// Implements additional methods for syncing a storage area. This is a separate
+// interface because these methods are not part of the `StorageArea` API, and
+// have restrictions on when they can be called.
+[scriptable, uuid(6dac82c9-1d8a-4893-8c0f-6e626aef802c)]
+interface mozISyncedExtensionStorageArea : nsISupports {
+    // If a sync is in progress, this method fetches pending change
+    // notifications for all extensions whose storage areas were updated.
+    // `callback` should implement `mozIExtensionStorageListener` to forward
+    // the records to `storage.onChanged` listeners. This method should only
+    // be called by Sync, after `mozIBridgedSyncEngine.apply` and before
+    // `syncFinished`. It fetches nothing if called at any other time.
+    void fetchPendingSyncChanges(in mozIExtensionStorageCallback callback);
+};
+
 // A listener for storage area notifications.
 [scriptable, uuid(8cb3c7e4-d0ca-4353-bccd-2673b4e11510)]
 interface mozIExtensionStorageListener : nsISupports {
     // Notifies that an operation has data to pass to `storage.onChanged`
-    // listeners. `json` is a JSON array of listener infos.
-    void onChanged(in AUTF8String json);
+    // listeners for the given `extensionId`. `json` is a JSON array of listener
+    // infos. If an operation affects multiple extensions, this method will be
+    // called multiple times, once per extension.
+    void onChanged(in AUTF8String extensionId, in AUTF8String json);
 };
 
 // A generic callback for a storage operation. Either `handleSuccess` or

toolkit/components/extensions/storage/webext_storage_bridge/src/area.rs

@@ -35,7 +35,9 @@ use crate::store::{LazyStore, LazyStoreConfig};
 /// threads. In Rust terms, it's `Send`, but not `Sync`.
 #[derive(xpcom)]
 #[xpimplements(
+    mozIExtensionStorageArea,
     mozIConfigurableExtensionStorageArea,
+    mozISyncedExtensionStorageArea,
     mozIInterruptible,
     mozIBridgedSyncEngine
 )]
@@ -268,6 +270,16 @@ fn teardown(
     Ok(())
 }
 
+/// `mozISyncedExtensionStorageArea` implementation.
+impl StorageSyncArea {
+    xpcom_method!(
+        fetch_pending_sync_changes => FetchPendingSyncChanges(callback: *const mozIExtensionStorageCallback)
+    );
+    fn fetch_pending_sync_changes(&self, callback: &mozIExtensionStorageCallback) -> Result<()> {
+        self.dispatch(Punt::FetchPendingSyncChanges, callback)
+    }
+}
+
 /// `mozIInterruptible` implementation.
 impl StorageSyncArea {
     xpcom_method!(

toolkit/components/extensions/storage/webext_storage_bridge/src/lib.rs

@@ -40,7 +40,7 @@ mod punt;
 mod store;
 
 use nserror::{nsresult, NS_OK};
-use xpcom::{interfaces::mozIConfigurableExtensionStorageArea, RefPtr};
+use xpcom::{interfaces::mozIExtensionStorageArea, RefPtr};
 
 use crate::area::StorageSyncArea;
 
@@ -53,12 +53,11 @@ use crate::area::StorageSyncArea;
 /// This function is unsafe because it dereferences `result`.
 #[no_mangle]
 pub unsafe extern "C" fn NS_NewExtensionStorageSyncArea(
-    result: *mut *const mozIConfigurableExtensionStorageArea,
+    result: *mut *const mozIExtensionStorageArea,
 ) -> nsresult {
     match StorageSyncArea::new() {
         Ok(bridge) => {
-            RefPtr::new(bridge.coerce::<mozIConfigurableExtensionStorageArea>())
-                .forget(&mut *result);
+            RefPtr::new(bridge.coerce::<mozIExtensionStorageArea>()).forget(&mut *result);
             NS_OK
         }
         Err(err) => err.into(),

toolkit/components/extensions/storage/webext_storage_bridge/src/punt.rs

@@ -37,6 +37,9 @@ pub enum Punt {
     Clear { ext_id: String },
     /// Returns the bytes in use for the specified, or all, keys.
     GetBytesInUse { ext_id: String, keys: JsonValue },
+    /// Fetches all pending Sync change notifications to pass to
+    /// `storage.onChanged` listeners.
+    FetchPendingSyncChanges,
 }
 
 impl Punt {
@@ -49,6 +52,7 @@ impl Punt {
             Punt::Remove { .. } => "webext_storage::remove",
             Punt::Clear { .. } => "webext_storage::clear",
             Punt::GetBytesInUse { .. } => "webext_storage::get_bytes_in_use",
+            Punt::FetchPendingSyncChanges => "webext_storage::fetch_pending_sync_changes",
         }
     }
 }
@@ -56,27 +60,45 @@ impl Punt {
 /// A storage operation result, punted from the background queue back to the
 /// main thread.
 #[derive(Default)]
-pub struct PuntResult {
-    pub changes: Option<String>,
-    pub value: Option<String>,
+struct PuntResult {
+    changes: Vec<Change>,
+    value: Option<String>,
+}
+
+/// A change record for an extension.
+struct Change {
+    ext_id: String,
+    json: String,
 }
 
 impl PuntResult {
-    /// Creates a result with changes to pass to `onChanged`, and no return
-    /// value for `handleSuccess`. The `Borrow` bound lets this method take
-    /// either a borrowed reference or an owned value.
-    pub fn with_changes<T: Borrow<S>, S: Serialize>(changes: T) -> Result<Self> {
+    /// Creates a result with a single change to pass to `onChanged`, and no
+    /// return value for `handleSuccess`. The `Borrow` bound lets this method
+    /// take either a borrowed reference or an owned value.
+    fn with_change<T: Borrow<S>, S: Serialize>(ext_id: &str, changes: T) -> Result<Self> {
         Ok(PuntResult {
-            changes: Some(serde_json::to_string(changes.borrow())?),
+            changes: vec![Change {
+                ext_id: ext_id.into(),
+                json: serde_json::to_string(changes.borrow())?,
+            }],
             value: None,
         })
     }
 
+    /// Creates a result with changes for multiple extensions to pass to
+    /// `onChanged`, and no return value for `handleSuccess`.
+    fn with_changes(changes: Vec<Change>) -> Self {
+        PuntResult {
+            changes,
+            value: None,
+        }
+    }
+
     /// Creates a result with no changes to pass to `onChanged`, and a return
     /// value for `handleSuccess`.
-    pub fn with_value<T: Borrow<S>, S: Serialize>(value: T) -> Result<Self> {
+    fn with_value<T: Borrow<S>, S: Serialize>(value: T) -> Result<Self> {
         Ok(PuntResult {
-            changes: None,
+            changes: Vec::new(),
             value: Some(serde_json::to_string(value.borrow())?),
         })
     }
@@ -137,21 +159,32 @@ impl PuntTask {
     fn inner_run(&self, punt: Punt) -> Result<PuntResult> {
         Ok(match punt {
             Punt::Set { ext_id, value } => {
-                PuntResult::with_changes(self.store()?.get()?.set(&ext_id, value)?)
+                PuntResult::with_change(&ext_id, self.store()?.get()?.set(&ext_id, value)?)?
             }
             Punt::Get { ext_id, keys } => {
-                PuntResult::with_value(self.store()?.get()?.get(&ext_id, keys)?)
+                PuntResult::with_value(self.store()?.get()?.get(&ext_id, keys)?)?
             }
             Punt::Remove { ext_id, keys } => {
-                PuntResult::with_changes(self.store()?.get()?.remove(&ext_id, keys)?)
+                PuntResult::with_change(&ext_id, self.store()?.get()?.remove(&ext_id, keys)?)?
             }
             Punt::Clear { ext_id } => {
-                PuntResult::with_changes(self.store()?.get()?.clear(&ext_id)?)
+                PuntResult::with_change(&ext_id, self.store()?.get()?.clear(&ext_id)?)?
             }
             Punt::GetBytesInUse { ext_id, keys } => {
-                PuntResult::with_value(self.store()?.get()?.get_bytes_in_use(&ext_id, keys)?)
+                PuntResult::with_value(self.store()?.get()?.get_bytes_in_use(&ext_id, keys)?)?
             }
-        }?)
+            Punt::FetchPendingSyncChanges => PuntResult::with_changes(
+                self.store()?
+                    .get()?
+                    .get_synced_changes()?
+                    .into_iter()
+                    .map(|info| Change {
+                        ext_id: info.ext_id,
+                        json: info.changes,
+                    })
+                    .collect(),
+            ),
+        })
     }
 }
 
@@ -176,12 +209,13 @@ impl Task for PuntTask {
             Ok(PuntResult { changes, value }) => {
                 // If we have change data, and the callback implements the
                 // listener interface, notify about it first.
-                if let (Some(listener), Some(json)) = (
-                    callback.query_interface::<mozIExtensionStorageListener>(),
-                    changes,
-                ) {
-                    // Ignore errors.
-                    let _ = unsafe { listener.OnChanged(&*nsCString::from(json)) };
+                if let Some(listener) = callback.query_interface::<mozIExtensionStorageListener>() {
+                    for Change { ext_id, json } in changes {
+                        // Ignore errors.
+                        let _ = unsafe {
+                            listener.OnChanged(&*nsCString::from(ext_id), &*nsCString::from(json))
+                        };
+                    }
                 }
                 let result = value.map(nsCString::from).into_variant();
                 unsafe { callback.HandleSuccess(result.coerce()) }

toolkit/components/extensions/test/xpcshell/test_StorageSyncService.js

@@ -22,8 +22,8 @@ function promisify(func, ...params) {
         Ci.mozIBridgedSyncEngineCallback,
         Ci.mozIBridgedSyncEngineApplyCallback,
       ]),
-      onChanged(json) {
-        changes.push(JSON.parse(json));
+      onChanged(extId, json) {
+        changes.push({ extId, changes: JSON.parse(json) });
       },
       handleSuccess(value) {
         resolve({
@@ -58,11 +58,14 @@ add_task(async function test_storage_sync_service() {
       changes,
       [
         {
-          hi: {
-            newValue: "hello! 💖",
-          },
-          bye: {
-            newValue: "adiós",
+          extId: "ext-1",
+          changes: {
+            hi: {
+              newValue: "hello! 💖",
+            },
+            bye: {
+              newValue: "adiós",
+            },
           },
         },
       ],
@@ -158,12 +161,17 @@ add_task(async function test_storage_sync_bridged_engine() {
   info("Merge");
   // Three levels of JSON wrapping: each outgoing envelope, the cleartext in
   // each envelope, and the extension storage data in each cleartext.
-  // TODO: Should we reduce to 2? Extension storage data could be a map...
   let { value: outgoingEnvelopesAsJSON } = await promisify(area.apply);
   let outgoingEnvelopes = outgoingEnvelopesAsJSON.map(json => JSON.parse(json));
   let parsedCleartexts = outgoingEnvelopes.map(e => JSON.parse(e.cleartext));
   let parsedData = parsedCleartexts.map(c => JSON.parse(c.data));
 
+  let { changes } = await promisify(
+    area.QueryInterface(Ci.mozISyncedExtensionStorageArea)
+      .fetchPendingSyncChanges
+  );
+  deepEqual(changes, [], "Should return pending synced changes for observers");
+
   // ext-1 doesn't exist remotely yet, so the Rust sync layer will generate
   // a GUID for it. We don't know what it is, so we find it by the extension
   // ID.