recovery: Fix the recovery state not being updated after backups are enabled via secret send

crates/matrix-sdk/src/encryption/backups/mod.rs

@@ -423,8 +423,12 @@ impl Backups {
     }
 
     /// Set the state of the backup.
-    fn set_state(&self, state: BackupState) {
-        self.client.inner.e2ee.backup_state.global_state.set(state);
+    fn set_state(&self, new_state: BackupState) {
+        let old_state = self.client.inner.e2ee.backup_state.global_state.set(new_state);
+
+        if old_state != new_state {
+            info!("Backup state changed from {old_state:?} to {new_state:?}");
+        }
     }
 
     /// Set the backup state to the `Enabled` variant and insert the backup key
@@ -860,6 +864,7 @@ impl Backups {
 
     /// Listen for `m.secret.send` to-device messages and check the secret inbox
     /// if we do receive one.
+    #[instrument(skip_all)]
     pub(crate) async fn secret_send_event_handler(_: ToDeviceSecretSendEvent, client: Client) {
         let olm_machine = client.olm_machine().await;
 
@@ -921,7 +926,6 @@ impl Backups {
     /// removed on the homeserver.
     async fn handle_deleted_backup_version(&self, olm_machine: &OlmMachine) -> Result<(), Error> {
         olm_machine.backup_machine().disable_backup().await?;
-        self.client.encryption().recovery().update_state_after_backup_disabling().await;
         self.set_state(BackupState::Unknown);
 
         Ok(())

crates/matrix-sdk/src/encryption/mod.rs

@@ -135,6 +135,20 @@ impl EncryptionData {
             tasks.download_room_keys = Some(BackupDownloadTask::new(weak_client));
         }
     }
+
+    /// Initialize the background task which listens for changes in the
+    /// [`backups::BackupState`] and updataes the [`recovery::RecoveryState`].
+    ///
+    /// This should happen after the usual tasks have been set up and after the
+    /// E2EE initialization tasks have been set up.
+    pub fn initialize_recovery_state_update_task(&self, client: &Client) {
+        let mut guard = self.tasks.lock().unwrap();
+
+        let future = Recovery::update_state_after_backup_state_change(client);
+        let join_handle = spawn(future);
+
+        guard.update_recovery_state_after_backup = Some(join_handle);
+    }
 }
 
 /// Settings for end-to-end encryption features.

crates/matrix-sdk/src/encryption/recovery/mod.rs

@@ -90,7 +90,8 @@
 //!
 //! [`Recovery key`]: https://spec.matrix.org/v1.8/client-server-api/#recovery-key
 
-use futures_core::Stream;
+use futures_core::{Future, Stream};
+use futures_util::StreamExt as _;
 use ruma::{
     api::client::keys::get_keys,
     events::{
@@ -105,7 +106,7 @@ use crate::encryption::{
     backups::Backups,
     secret_storage::{SecretStorage, SecretStore},
 };
-use crate::Client;
+use crate::{client::WeakClient, encryption::backups::BackupState, Client};
 
 pub mod futures;
 mod types;
@@ -437,6 +438,7 @@ impl Recovery {
 
         self.client.add_event_handler(Self::default_key_event_handler);
         self.client.add_event_handler(Self::secret_send_event_handler);
+        self.client.inner.e2ee.initialize_recovery_state_update_task(&self.client);
 
         self.update_recovery_state().await?;
 
@@ -488,7 +490,11 @@ impl Recovery {
 
     async fn update_recovery_state(&self) -> Result<()> {
         let new_state = self.check_recovery_state().await?;
-        self.client.inner.e2ee.recovery_state.set(new_state);
+        let old_state = self.client.inner.e2ee.recovery_state.set(new_state);
+
+        if new_state != old_state {
+            info!("Recovery state changed from {old_state:?} to {new_state:?}");
+        }
 
         Ok(())
     }
@@ -509,12 +515,44 @@ impl Recovery {
         client.encryption().recovery().update_recovery_state_no_fail().await;
     }
 
-    #[instrument]
-    pub(crate) async fn update_state_after_backup_disabling(&self) {
-        // TODO: This is quite ugly, this method is called by the backups subsystem.
-        // Backups shouldn't depend on recovery, recovery should listen to the
-        // backup state change.
-        self.update_recovery_state_no_fail().await;
+    /// Listen for changes in the [`BackupState`] and, if necessary, update the
+    /// [`RecoveryState`] accordingly.
+    ///
+    /// This should not be called directly, this method is put into a background
+    /// task which is always listening for updates in the [`BackupState`].
+    pub(crate) fn update_state_after_backup_state_change(
+        client: &Client,
+    ) -> impl Future<Output = ()> {
+        let mut stream = client.encryption().backups().state_stream();
+        let weak = WeakClient::from_client(client);
+
+        async move {
+            while let Some(update) = stream.next().await {
+                if let Some(client) = weak.get() {
+                    match update {
+                        Ok(update) => {
+                            // The recovery state only cares about these two states, the
+                            // intermediate states that tell us that
+                            // we're creating a backup are not interesting.
+                            if matches!(update, BackupState::Unknown | BackupState::Enabled) {
+                                client
+                                    .encryption()
+                                    .recovery()
+                                    .update_recovery_state_no_fail()
+                                    .await;
+                            }
+                        }
+                        Err(_) => {
+                            // We missed some updates, let's update our state in case something
+                            // changed.
+                            client.encryption().recovery().update_recovery_state_no_fail().await;
+                        }
+                    }
+                } else {
+                    break;
+                }
+            }
+        }
     }
 
     #[instrument]

crates/matrix-sdk/src/encryption/tasks.rs

@@ -43,6 +43,8 @@ pub(crate) struct ClientTasks {
     pub(crate) upload_room_keys: Option<BackupUploadingTask>,
     #[cfg(feature = "e2e-encryption")]
     pub(crate) download_room_keys: Option<BackupDownloadTask>,
+    #[cfg(feature = "e2e-encryption")]
+    pub(crate) update_recovery_state_after_backup: Option<JoinHandle<()>>,
     pub(crate) setup_e2ee: Option<JoinHandle<()>>,
 }
 

crates/matrix-sdk/src/utils.rs

@@ -76,10 +76,16 @@ impl<T: 'static + Send + Clone> ChannelObservable<T> {
     }
 
     /// Set the underlying data to the new value.
-    pub(crate) fn set(&self, new_value: T) {
-        *self.value.write().unwrap() = new_value.to_owned();
+    pub(crate) fn set(&self, new_value: T) -> T {
+        let old_value = {
+            let mut guard = self.value.write().unwrap();
+            std::mem::replace(&mut (*guard), new_value.clone())
+        };
+
         // We're ignoring the error case where no receivers exist.
         let _ = self.channel.send(new_value);
+
+        old_value
     }
 
     /// Get the current value of the underlying data.
@@ -183,3 +189,17 @@ impl IntoRawStateEventContent for &Box<RawJsonValue> {
         self.clone().into_raw_state_event_content()
     }
 }
+
+#[cfg(test)]
+mod test {
+    #[cfg(feature = "e2e-encryption")]
+    #[test]
+    fn test_channel_observable_get_set() {
+        let observable = super::ChannelObservable::new(0);
+
+        assert_eq!(observable.get(), 0);
+        assert_eq!(observable.set(1), 0);
+        assert_eq!(observable.set(10), 1);
+        assert_eq!(observable.get(), 10);
+    }
+}

crates/matrix-sdk/tests/integration/encryption/recovery.rs

@@ -66,7 +66,7 @@ async fn test_client(user_id: &UserId) -> (Client, wiremock::MockServer) {
             "errcode": "M_NOT_FOUND",
             "error": "Account data not found"
         })))
-        .expect(1)
+        .expect(2)
         .named("m.secret_storage.default_key account data GET")
         .mount_as_scoped(&server)
         .await;
@@ -319,10 +319,13 @@ async fn enable(
         )))
         .and(header("authorization", "Bearer 1234"))
         .respond_with(move |_: &wiremock::Request| {
-            let content = default_key_content.lock().unwrap().take().unwrap();
-            ResponseTemplate::new(200).set_body_json(content)
+            if let Some(content) = default_key_content.lock().unwrap().as_ref() {
+                ResponseTemplate::new(200).set_body_json(content)
+            } else {
+                ResponseTemplate::new(404)
+            }
         })
-        .expect(1)
+        .expect(2)
         .mount_as_scoped(server)
         .await;
 
@@ -633,10 +636,13 @@ async fn recovery_disabling() {
         )))
         .and(header("authorization", "Bearer 1234"))
         .respond_with(move |_: &wiremock::Request| {
-            let content = default_key_content.lock().unwrap().take().unwrap();
-            ResponseTemplate::new(200).set_body_json(content)
+            if let Some(content) = default_key_content.lock().unwrap().as_ref() {
+                ResponseTemplate::new(200).set_body_json(content)
+            } else {
+                ResponseTemplate::new(404)
+            }
         })
-        .expect(1)
+        .expect(2)
         .named("m.secret_storage.default_key account data GET")
         .mount(&server)
         .await;