signed_state: Simplify refresh coalescing and stop blocking the executor

crates/signed_state/src/backend.rs

@@ -5,7 +5,7 @@ use std::time::{Duration, Instant};
 
 use anyhow::{Error, anyhow, bail};
 use bitcoin_hashes::sha1::Hash as Sha1Hash;
-use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Task};
+use gpui::{App, AppContext, BackgroundExecutor, Context, Entity, EventEmitter, Global, Task};
 use nostr::event::IntoEventBuilder;
 use nostr_connect::prelude::*;
 use nostr_sdk::client::SyncSummary;
@@ -531,6 +531,7 @@ impl Backend {
                 let repo_id = repo_id.clone();
                 let servers = servers.clone();
                 let refs = refs.clone();
+                let executor = cx.background_executor().clone();
                 async move {
                     push_staged_to_grasps(
                         &client,
@@ -541,6 +542,7 @@ impl Backend {
                         &destination,
                         &owner,
                         &servers,
+                        &executor,
                         signed_git::push_main,
                     )
                     .await
@@ -686,6 +688,7 @@ impl Backend {
                     let servers = servers.clone();
                     let refs = refs.clone();
                     let head = head.clone();
+                    let executor = cx.background_executor().clone();
                     async move {
                         push_staged_to_grasps(
                             &client,
@@ -696,6 +699,7 @@ impl Backend {
                             &path,
                             &owner,
                             &servers,
+                            &executor,
                             signed_git::push_all,
                         )
                         .await
@@ -854,6 +858,7 @@ impl Backend {
                     let relays = relays.clone();
                     let refs = refs.clone();
                     let head = head.clone();
+                    let executor = cx.background_executor().clone();
                     async move {
                         push_staged_to_grasps(
                             &client,
@@ -864,6 +869,7 @@ impl Backend {
                             &path,
                             &owner,
                             &relays,
+                            &executor,
                             signed_git::push_all,
                         )
                         .await
@@ -1700,6 +1706,7 @@ async fn push_staged_to_grasps(
     path: &Path,
     owner: &str,
     servers: &[RelayUrl],
+    executor: &BackgroundExecutor,
     push: fn(&Path, &str, &str, &str) -> Result<(), Error>,
 ) -> PushOutcome {
     let mut outcome = PushOutcome::default();
@@ -1722,7 +1729,7 @@ async fn push_staged_to_grasps(
         'server: for attempt in 1..=GRASP_PUSH_ATTEMPTS {
             if attempt > 1 {
                 // Give the server's ingest a moment before re-staging.
-                std::thread::sleep(GRASP_RETRY_DELAY);
+                executor.timer(GRASP_RETRY_DELAY).await;
             }
 
             let (event, created_at) =

crates/signed_state/src/checkouts.rs

@@ -93,6 +93,8 @@ pub struct CheckoutsStore {
     /// A recompute defaults the base the same way.
     requested_head: HashMap<RepoAddr, Option<String>>,
     refresh: RefreshGate,
+    /// True while the timer between a scheduled refresh and its run is pending.
+    debounce_pending: bool,
     local_pending: bool,
     /// When the last full pass (with a remote refresh) completed.
     ///
@@ -163,6 +165,7 @@ impl CheckoutsStore {
             push_statuses: HashMap::new(),
             requested_head: HashMap::new(),
             refresh: RefreshGate::default(),
+            debounce_pending: false,
             local_pending: false,
             last_full_sync: None,
             _subscriptions: subscriptions,
@@ -279,12 +282,15 @@ impl CheckoutsStore {
 
     /// Re-resolve the associations and the requested statuses.
     ///
-    /// Requests arriving while a pass runs fold into a follow-up.
+    /// Requests arriving while a pass runs fold into a follow-up, requests
+    /// arriving while the debounce timer is pending are dropped.
     pub fn refresh(&mut self, cx: &mut Context<Self>) {
-        if self.refresh.request() != RefreshRequest::Schedule {
+        if self.debounce_pending || self.refresh.request() != RefreshRequest::Schedule {
             return;
         }
 
+        self.debounce_pending = true;
+
         cx.spawn(async move |this, cx| {
             cx.background_executor().timer(REFRESH_DEBOUNCE).await;
             this.update(cx, |this, cx| this.run_refresh(cx))
@@ -300,6 +306,7 @@ impl CheckoutsStore {
     /// remote reconciliation cadence ([`Self::local_tick`]); they also restart
     /// the fast local pass.
     fn run_refresh(&mut self, cx: &mut Context<Self>) {
+        self.debounce_pending = false;
         self.refresh.begin();
 
         let records = {
@@ -453,7 +460,7 @@ impl CheckoutsStore {
         }
 
         // A full pass or a fresh request covers this tick, skip it.
-        if self.refresh.running() || self.refresh.debouncing() {
+        if self.refresh.running() || self.debounce_pending {
             self.schedule_local_pass(cx);
             return;
         }
@@ -515,7 +522,7 @@ impl CheckoutsStore {
             this.update(cx, |this, cx| {
                 // A full pass or a fresh request will apply fresher data
                 // (the tracking refs move only when a full pass fetches).
-                if this.refresh.running() || this.refresh.debouncing() {
+                if this.refresh.running() || this.debounce_pending {
                     return;
                 }
 

crates/signed_state/src/refresh.rs

@@ -3,14 +3,13 @@
 pub struct RefreshGate {
     running: bool,
     dirty: bool,
-    debouncing: bool,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum RefreshRequest {
-    /// No run or timer covers the request, start the debounce timer.
+    /// No run covers the request, start one now.
     Schedule,
-    /// A run or pending timer already covers the request.
+    /// A run is in flight and covers the request, fold it into a follow-up.
     Fold,
 }
 
@@ -19,28 +18,20 @@ impl RefreshGate {
         self.running
     }
 
-    pub fn debouncing(&self) -> bool {
-        self.debouncing
-    }
-
     /// A new refresh request arrived.
     ///
-    /// Folded into a follow-up run while one is in flight, dropped while the
-    /// debounce timer is pending, otherwise starts the timer.
+    /// Folded into a follow-up run while one is in flight, otherwise the
+    /// caller starts the run itself.
     pub fn request(&mut self) -> RefreshRequest {
         if self.running {
             self.dirty = true;
             RefreshRequest::Fold
-        } else if self.debouncing {
-            RefreshRequest::Fold
         } else {
-            self.debouncing = true;
             RefreshRequest::Schedule
         }
     }
 
     pub fn begin(&mut self) {
-        self.debouncing = false;
         self.running = true;
     }
 
@@ -55,3 +46,48 @@ impl RefreshGate {
         self.running = false;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_request_while_running_folds_into_a_follow_up() {
+        let mut gate = RefreshGate::default();
+        gate.begin();
+
+        assert_eq!(gate.request(), RefreshRequest::Fold);
+        assert!(gate.finish());
+    }
+
+    #[test]
+    fn a_request_without_a_run_schedules() {
+        let mut gate = RefreshGate::default();
+
+        assert_eq!(gate.request(), RefreshRequest::Schedule);
+        assert!(!gate.running());
+    }
+
+    #[test]
+    fn a_request_after_a_run_schedules_again() {
+        let mut gate = RefreshGate::default();
+        gate.begin();
+        assert_eq!(gate.request(), RefreshRequest::Fold);
+        assert!(gate.finish());
+
+        assert_eq!(gate.request(), RefreshRequest::Schedule);
+    }
+
+    #[test]
+    fn abort_keeps_the_pending_request() {
+        let mut gate = RefreshGate::default();
+        gate.begin();
+        assert_eq!(gate.request(), RefreshRequest::Fold);
+
+        gate.abort();
+        assert!(!gate.running());
+
+        gate.begin();
+        assert!(gate.finish());
+    }
+}

crates/workspace/src/views/inbox.rs

@@ -177,7 +177,6 @@ impl InboxView {
     }
 
     fn refresh_initial(&mut self, cx: &mut Context<Self>) {
-        debug_assert!(!self.refresh.debouncing());
         if self.refresh.running() {
             self.refresh.request();
             return;