recovery: Ensure that we don't miss updates to the backup state

This patch switches the way we update the recovery state upon changes in
the backup state. Previously two places updated the recovery state after
the backup state changed:

    1. A method living in the recovery subsystem that the backup
       subsystem itself calls.
    2. An event handler which is called when we receive a m.secret.send
       event.

The first method is a hack because it introduces a circular dependency
between the recovery and backup subsystems.

More importantly, the second method can miss updates, because the backup
subsystem has a similar event handler which then processes the secret we
received and if the secret was a backup recovery key, enables backups.

Depending on the order these event handlers are called, the recovery
subsystem might update the recovery state before the secret has been
handled.

The backup subsystem provides an async stream which broadcasts updates
to the backup state, letting the recovery subsystem listen to this
stream and update its state if we notice such updates fixes both
problems we listed above.

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

@@ -926,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?;
 
@@ -513,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/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;