fix(sdk): Remove a cyclic reference of Client in ThreadSubscriptionCatchup

crates/matrix-sdk/changelog.d/6594.fixed.md

@@ -0,0 +1,3 @@
+A cyclic reference of `Client` has been detected in `ThreadSubscriptionCatchup`,
+preventing `Client` to drop correctly. This is now fixed, removing a memory leak
+about `Client`.

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

@@ -476,10 +476,9 @@ impl ClientInner {
             EventCache::new(&client, client.base_client.event_cache_store().clone())
         });
 
-        let init_thread_subscription_catchup =
-            client.thread_subscription_catchup.get_or_init(|| async {
-                ThreadSubscriptionCatchup::new(Client { inner: client.clone() })
-            });
+        let init_thread_subscription_catchup = client
+            .thread_subscription_catchup
+            .get_or_init(|| ThreadSubscriptionCatchup::new(Client { inner: client.clone() }));
 
         let _ = join!(init_event_cache, init_thread_subscription_catchup);
 

crates/matrix-sdk/src/client/thread_subscriptions.rs

@@ -131,50 +131,44 @@ pub struct ThreadSubscriptionCatchup {
 }
 
 impl ThreadSubscriptionCatchup {
-    pub fn new(client: Client) -> Arc<Self> {
+    pub async fn new(client: Client) -> Arc<Self> {
         let is_outdated = Arc::new(AtomicBool::new(true));
-
         let weak_client = WeakClient::from_client(&client);
-
         let (ping_sender, ping_receiver) = channel(8);
-
         let uniq_mutex = Arc::new(Mutex::new(()));
 
         let this = Arc::new(Self {
             _task: OnceLock::new(),
             is_outdated,
-            client: weak_client,
+            client: weak_client.clone(),
             ping_sender,
             uniq_mutex,
         });
 
         // Create the task only if the client is configured to handle thread
         // subscriptions.
-        let _ = this._task.get_or_init(|| {
-            let that = this.clone();
-            let client_clone = client.clone();
-
-            client.task_monitor().spawn_infinite_task("client::thread_subscriptions_catchup", async move {
-                match client_clone.enabled_thread_subscriptions().await {
-                    Ok(enabled) => {
-                        if !enabled {
-                            debug!("Thread subscriptions catchup not enabled, not starting the catchup task");
-                            return;
-                        }
-                    }
+        match client.enabled_thread_subscriptions().await {
+            Ok(true) => {
+                let _ = this._task.get_or_init(|| {
+                    let that = this.clone();
+
+                    client
+                        .task_monitor()
+                        .spawn_infinite_task("client::thread_subscriptions_catchup", async move {
+                            Self::thread_subscriptions_catchup_task(that, ping_receiver).await;
+                        })
+                        .abort_on_drop()
+                });
+            }
 
-                    Err(err) => {
-                        warn!("Failed to check if thread subscriptions catchup is enabled: {err}");
-                        return;
-                    }
-                }
+            Ok(false) => {
+                debug!("Thread subscriptions catchup not enabled, not starting the catchup task");
+            }
 
-                Self::thread_subscriptions_catchup_task(
-                    that,
-                    ping_receiver,
-                ).await
-            }).abort_on_drop()
-        });
+            Err(err) => {
+                warn!("Failed to check if thread subscriptions catchup is enabled: {err}");
+            }
+        }
 
         this
     }
@@ -458,3 +452,50 @@ mod tests {
         assert!(tsc.is_outdated().not());
     }
 }
+
+#[cfg(all(test, not(target_family = "wasm")))]
+mod timed_tests {
+    use std::time::Duration;
+
+    use matrix_sdk_base::{ThreadingSupport, sleep::sleep};
+    use matrix_sdk_test::async_test;
+    use tokio::task::yield_now;
+
+    use crate::{client::WeakClient, test_utils::mocks::MatrixMockServer};
+
+    #[async_test]
+    async fn test_issue_6573_client_can_drop_thread_subscriptions_task() {
+        let server = MatrixMockServer::new().await;
+        server.mock_versions().with_thread_subscriptions().ok().mount().await;
+
+        let client = server
+            .client_builder()
+            .no_server_versions()
+            .on_builder(|builder| {
+                builder
+                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
+            })
+            .build()
+            .await;
+
+        let tsc = client.thread_subscription_catchup();
+
+        // Wait for anything to start up.
+        yield_now().await;
+
+        // Ensure the task is running.
+        assert!(tsc._task.get().is_some());
+
+        // Get a weak reference to the client.
+        let weak_client = WeakClient::from_client(&client);
+
+        // Drop the client will drop the task.
+        drop(client);
+
+        // Wait for anything to shutdown.
+        sleep(Duration::from_secs(2)).await;
+
+        // The client has been dropped correctly.
+        assert_eq!(weak_client.strong_count(), 0);
+    }
+}

crates/matrix-sdk/tests/integration/client.rs

@@ -2119,6 +2119,10 @@ async fn test_sync_thread_subscriptions() {
 #[async_test]
 async fn test_sync_thread_subscriptions_with_catchup() {
     let server = MatrixMockServer::new().await;
+
+    // Make sure to advertise support for thread subscriptions.
+    server.mock_versions().with_thread_subscriptions().ok().mount().await;
+
     let client = server
         .client_builder()
         .no_server_versions()
@@ -2135,9 +2139,6 @@ async fn test_sync_thread_subscriptions_with_catchup() {
     let thread2 = owned_event_id!("$thread2:example.com");
     let thread3 = owned_event_id!("$thread3:example.com");
 
-    // Make sure to advertise support for thread subscriptions.
-    server.mock_versions().with_thread_subscriptions().ok().mount().await;
-
     // The provided catchup token will be used to fetch more thread
     // subscriptions via the msc4308 companion endpoint.
     server