refactor

crates/signed_state/src/profile.rs

@@ -3,8 +3,11 @@ use std::collections::{HashMap, HashSet};
 use std::time::{Duration, Instant};
 
 use anyhow::Error;
-use flume::{Receiver, RecvTimeoutError, Sender};
-use gpui::{App, AppContext, Context, Entity, Global, SharedString, Subscription, Task};
+use flume::{Receiver, Sender};
+use gpui::{
+    App, AppContext, AsyncApp, Context, Entity, Global, SharedString, Subscription, Task,
+    WeakEntity,
+};
 use nostr_sdk::prelude::*;
 use utils::shorten_pubkey;
 
@@ -60,14 +63,6 @@ impl Profile {
     }
 }
 
-/// Message from the fetch task to the main thread.
-enum Dispatch {
-    /// A batched sync finished.
-    ///
-    /// Re-read seen profiles from the database.
-    Synced,
-}
-
 /// How long to wait for more requests before firing a batched sync.
 const BATCH_TIMEOUT: Duration = Duration::from_millis(500);
 
@@ -114,23 +109,15 @@ impl ProfileStore {
             _ => {}
         });
 
-        // Fetch requests are queued on a channel.
+        // Fetch requests are queued on a channel, batched into one sync per debounce window.
         let client = backend.read(cx).client();
         let (sender, receiver) = flume::unbounded::<PublicKey>();
-        let (dispatch_tx, dispatch_rx) = flume::unbounded::<Dispatch>();
+        let entity = cx.entity().downgrade();
 
         let mut tasks = Vec::new();
 
-        tasks.push(cx.background_spawn(async move {
-            Self::handle_requests(&client, &dispatch_tx, &receiver).await
-        }));
-
-        // Re-read seen profiles from the database after each batch sync.
-        tasks.push(cx.spawn(async move |this, cx| {
-            while let Ok(Dispatch::Synced) = dispatch_rx.recv_async().await {
-                this.update(cx, |this, cx| this.apply_seen(cx)).ok();
-            }
-            Ok(())
+        tasks.push(cx.spawn(async move |_this, cx| {
+            Self::handle_requests(entity, &client, &receiver, cx).await
         }));
 
         let mut store = Self {
@@ -145,6 +132,14 @@ impl ProfileStore {
         store
     }
 
+    /// Track a spawned task, pruning finished tasks first.
+    ///
+    /// Keeps the store's task list bounded by the number of in-flight tasks.
+    fn push_task(&mut self, task: Task<Result<(), Error>>) {
+        self.tasks.retain(|task| !task.is_ready());
+        self.tasks.push(task);
+    }
+
     /// Get a profile.
     ///
     /// Returns a placeholder with default metadata. Queues a fetch when the profile is not cached yet.
@@ -186,7 +181,7 @@ impl ProfileStore {
             Ok::<_, Error>(profiles)
         });
 
-        self.tasks.push(cx.spawn(async move |this, cx| {
+        self.push_task(cx.spawn(async move |this, cx| {
             let profiles = work.await?;
 
             this.update(cx, |this, cx| {
@@ -222,7 +217,7 @@ impl ProfileStore {
             Ok::<_, Error>(profile)
         });
 
-        self.tasks.push(cx.spawn(async move |this, cx| {
+        self.push_task(cx.spawn(async move |this, cx| {
             let profile = work.await?;
 
             this.update(cx, |this, cx| {
@@ -278,7 +273,7 @@ impl ProfileStore {
             Ok::<_, Error>(profiles)
         });
 
-        self.tasks.push(cx.spawn(async move |this, cx| {
+        self.push_task(cx.spawn(async move |this, cx| {
             let profiles = work.await?;
 
             this.update(cx, |this, cx| {
@@ -294,28 +289,43 @@ impl ProfileStore {
 
     /// Sync metadata for requested authors in batches, debounced to collect requests.
     ///
-    /// Results are dispatched to the main thread, which re-reads the database.
+    /// After each batch, the seen profiles are re-read from the database on the main thread.
     async fn handle_requests(
+        this: WeakEntity<ProfileStore>,
         client: &Client,
-        dispatch: &Sender<Dispatch>,
         receiver: &Receiver<PublicKey>,
+        cx: &mut AsyncApp,
     ) -> Result<(), Error> {
         let mut batch: HashSet<PublicKey> = HashSet::new();
 
         loop {
             // Wait for the first request of a batch.
-            match receiver.recv_timeout(BATCH_TIMEOUT) {
+            match receiver.recv_async().await {
                 Ok(public_key) => {
                     batch.insert(public_key);
                 }
-                Err(RecvTimeoutError::Disconnected) => return Ok(()),
-                Err(RecvTimeoutError::Timeout) => continue,
-            };
+                Err(_) => return Ok(()),
+            }
 
             // Collect everything that arrives within the debounce window.
+            // The channel has no async timeout, race the receive against a timer.
             let deadline = Instant::now() + BATCH_TIMEOUT;
-            while let Ok(public_key) = receiver.recv_deadline(deadline) {
-                batch.insert(public_key);
+            loop {
+                let now = Instant::now();
+                if now >= deadline {
+                    break;
+                }
+                let timer = cx.background_executor().timer(deadline - now);
+                futures::pin_mut!(timer);
+                let recv = receiver.recv_async();
+                futures::pin_mut!(recv);
+                match futures::future::select(recv, timer).await {
+                    futures::future::Either::Left((Ok(public_key), _)) => {
+                        batch.insert(public_key);
+                    }
+                    futures::future::Either::Left((Err(_), _)) => return Ok(()),
+                    futures::future::Either::Right(_) => break,
+                }
             }
 
             let filter = Filter::new()
@@ -327,9 +337,7 @@ impl ProfileStore {
             // Re-apply from the database afterwards.
             match sync_bootstrap_only(client, filter, SyncOptions::default()).await {
                 Ok(_) => {
-                    if dispatch.send(Dispatch::Synced).is_err() {
-                        log::warn!("profile dispatch channel closed, dropping sync result");
-                    }
+                    let _ = this.update(cx, |this, cx| this.apply_seen(cx));
                 }
                 Err(e) => log::warn!("profile sync failed: {e}"),
             }