Reshape SyncStatusStore into a thin probe-and-emit store

crates/signed_state/src/lib.rs

@@ -29,7 +29,7 @@ pub use repos::RepoListStore;
 pub use signed_git::{Nip34Binding, Nip34Kind};
 use signed_nostr::NostrBackend;
 pub use state::{GitPushMode, GitRef, PushStep, PushStepStatus, SyncState};
-pub use sync_status::{CheckoutSyncStatus, SyncStatusStore};
+pub use sync_status::{CheckoutSyncStatus, SyncEvent, SyncStatusStore};
 
 pub fn init(
     db_path: impl AsRef<Path>,

crates/signed_state/src/repo.rs

@@ -1226,6 +1226,7 @@ impl RepoStore {
                     Ok(outcome) => {
                         this.last_error = None;
                         this.last_push_warning = outcome.partial_warning();
+
                         if let Some((addr, path)) = &pushed_checkout {
                             let checkouts = CheckoutsStore::global(cx);
                             checkouts.update(cx, |store, cx| {
@@ -1233,8 +1234,13 @@ impl RepoStore {
                             });
                         }
 
+                        // Re-probe so sync badges reflect the pushed state.
                         let sync_status = SyncStatusStore::global(cx);
-                        sync_status.update(cx, |store, cx| store.refresh(cx));
+                        let addr = pushed_checkout
+                            .as_ref()
+                            .map(|(addr, _)| addr.clone())
+                            .or_else(|| this.addr.clone());
+                        sync_status.read(cx).probe(addr);
                     }
                     Err(e) => {
                         this.last_error = Some(format!("Push failed: {e}"));

crates/signed_state/src/sync_status.rs

@@ -1,18 +1,17 @@
-use std::collections::HashMap;
+use std::cell::RefCell;
+use std::collections::HashSet;
 use std::path::PathBuf;
 use std::time::Duration;
 
 use anyhow::Error;
-use gpui::{App, AppContext, Context, Entity, Global, Subscription};
+use flume::Sender;
-use nostr_sdk::prelude::*;
+use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
 use signed_core::{Deletions, Filters, RepoAddr};
 use signed_git::{Repo, RepoSyncStatus};
 
 use crate::backend::{Backend, BackendEvent};
 use crate::checkouts::CheckoutsStore;
-use crate::refresh::{RefreshGate, RefreshRequest};
+use crate::state::SyncState;
-use crate::repos::RepoListStore;
-use crate::state::{GitRef, SyncState};
 
 const SYNC_POLL: Duration = Duration::from_secs(180);
 const MAX_SYNC_CHECKOUTS: usize = 8;
@@ -28,13 +27,25 @@ pub struct CheckoutSyncStatus {
     pub status: RepoSyncStatus,
 }
 
+/// Statuses for these repositories may have changed; empty on signer change.
+#[derive(Debug, Clone)]
+pub enum SyncEvent {
+    Updated(Vec<(RepoAddr, Vec<CheckoutSyncStatus>)>),
+}
+
+/// Probes tracked repositories and broadcasts results. Holds no statuses.
 pub struct SyncStatusStore {
-    statuses: HashMap<RepoAddr, Vec<CheckoutSyncStatus>>,
+    active: RefCell<HashSet<RepoAddr>>,
-    refresh: RefreshGate,
+    in_flight: bool,
+    pass_pending: bool,
     poll_pending: bool,
+    sender: Sender<()>,
+    tasks: Vec<Task<Result<(), Error>>>,
     _subscriptions: Vec<Subscription>,
 }
 
+impl EventEmitter<SyncEvent> for SyncStatusStore {}
+
 impl SyncStatusStore {
     pub fn global(cx: &App) -> Entity<Self> {
         cx.global::<GlobalSyncStatusStore>().0.clone()
@@ -45,150 +56,131 @@ impl SyncStatusStore {
     }
 
     pub fn new(cx: &mut Context<Self>) -> Self {
+        let (sender, receiver) = flume::unbounded::<()>();
         let mut subscriptions = Vec::new();
 
         if !cfg!(target_arch = "wasm32") {
-            let repos = RepoListStore::global(cx);
-            let checkouts = CheckoutsStore::global(cx);
             let backend = Backend::global(cx);
-
+            let checkouts = CheckoutsStore::global(cx);
-            subscriptions.push(cx.observe(&repos, |this, _repos, cx| {
-                this.refresh(cx);
-            }));
 
             // Checkout associations decide which paths get compared.
             subscriptions.push(cx.observe(&checkouts, |this, _checkouts, cx| {
-                this.refresh(cx);
+                this.run_pass(cx);
             }));
 
             subscriptions.push(
                 cx.subscribe(&backend, |this, _backend, event, cx| match event {
-                    BackendEvent::SignerChanged => {
+                    BackendEvent::SignerChanged => this.clear(cx),
-                        this.statuses.clear();
+                    BackendEvent::RepoUpdates(_) | BackendEvent::Synced => this.run_pass(cx),
-                        cx.notify();
-                        this.refresh(cx);
-                    }
-                    BackendEvent::RepoUpdates(_) | BackendEvent::Synced => this.refresh(cx),
                     _ => {}
                 }),
             );
-
-            let weak = cx.entity().downgrade();
-            cx.defer(move |cx| {
-                if let Err(error) = weak.update(cx, |this, cx| this.refresh(cx)) {
-                    log::warn!("sync status store dropped before the initial refresh: {error}");
-                }
-            });
         }
 
+        let task = cx.spawn(async move |this, cx| {
+            loop {
+                receiver.recv_async().await?;
+
+                // One pass covers every signal queued since the wake.
+                while receiver.try_recv().is_ok() {}
+
+                this.update(cx, |this, cx| {
+                    this.run_pass(cx);
+                })?;
+            }
+        });
+
         Self {
-            statuses: HashMap::new(),
+            active: RefCell::new(HashSet::new()),
-            refresh: RefreshGate::default(),
+            sender,
+            in_flight: false,
+            pass_pending: false,
             poll_pending: false,
+            tasks: vec![task],
             _subscriptions: subscriptions,
         }
     }
 
-    /// Latest sync status per associated checkout.
+    /// Queue a probe pass; folded into any pass already in flight.
-    pub fn statuses_of(&self, addr: &RepoAddr) -> Vec<CheckoutSyncStatus> {
+    pub fn probe(&self, addrs: impl IntoIterator<Item = RepoAddr>) {
-        self.statuses.get(addr).cloned().unwrap_or_default()
+        self.active.borrow_mut().extend(addrs);
+
+        if let Err(e) = self.sender.send(()) {
+            log::warn!("failed to queue sync status probe: {e}");
+        }
     }
 
-    /// Commits available locally but missing on Nostr.
+    // Keeps the in-flight pass so its completion releases the slot exactly once.
-    pub fn unsynced(&self, addr: &RepoAddr) -> usize {
+    fn clear(&mut self, cx: &mut Context<Self>) {
-        self.statuses
+        let had_addrs = !self.active.borrow().is_empty();
-            .get(addr)
+        self.active.borrow_mut().clear();
-            .map(|list| {
+        self.pass_pending = false;
-                list.iter()
+
-                    .map(|checkout| checkout.status.ahead_total)
+        if had_addrs {
-                    .sum()
+            cx.emit(SyncEvent::Updated(Vec::new()));
-            })
+        }
-            .unwrap_or(0)
     }
 
-    pub fn refresh(&mut self, cx: &mut Context<Self>) {
+    fn run_pass(&mut self, cx: &mut Context<Self>) {
         if cfg!(target_arch = "wasm32") {
             return;
         }
-        if self.refresh.request() != RefreshRequest::Schedule {
+
+        if self.in_flight {
+            self.pass_pending = true;
             return;
         }
 
-        let backend = Backend::global(cx);
-        let Some(user) = backend.read(cx).current_user() else {
-            let changed = !self.statuses.is_empty();
-            self.statuses.clear();
-            if changed {
-                cx.notify();
-            }
-            return;
-        };
-
-        self.schedule_poll(cx);
-        self.run_refresh(user, cx);
-    }
-
-    fn schedule_poll(&mut self, cx: &mut Context<Self>) {
-        if self.poll_pending {
-            return;
-        }
-        self.poll_pending = true;
-
-        cx.spawn(async move |this, cx| {
-            cx.background_executor().timer(SYNC_POLL).await;
-            this.update(cx, |this, cx| {
-                this.poll_pending = false;
-                this.refresh(cx);
-            })
-        })
-        .detach();
-    }
-
-    fn run_refresh(&mut self, user: PublicKey, cx: &mut Context<Self>) {
-        self.refresh.begin();
-
         let backend = Backend::global(cx);
         let client = backend.read(cx).client();
 
-        let repos = RepoListStore::global(cx);
+        if backend.read(cx).current_user().is_none() {
-        let announcements = repos.read(cx).announcements_of(&user);
+            self.clear(cx);
-        let checkouts = CheckoutsStore::global(cx);
+            return;
+        }
 
-        let targets: Vec<(RepoAddr, Vec<PathBuf>)> = announcements
+        if self.active.borrow().is_empty() {
-            .iter()
+            return;
-            .map(|announcement| {
+        }
-                let addr = announcement.addr();
+
-                let paths = checkouts.read(cx).associations_of(&addr);
+        self.in_flight = true;
-                (addr, paths)
+        self.schedule_poll(cx);
-            })
+
-            .collect();
+        let targets: Vec<(RepoAddr, Vec<PathBuf>)> = {
+            let active = self.active.borrow();
+            let checkouts = CheckoutsStore::global(cx);
+
+            active
+                .iter()
+                .map(|addr| {
+                    let paths = checkouts.read(cx).associations_of(addr);
+                    (addr.clone(), paths)
+                })
+                .collect()
+        };
 
         let work = cx.background_spawn(async move {
             let db = client.database();
             let deletion_events = db.query(Filters::deletions()).await?;
             let deletions = Deletions::from_events(deletion_events);
 
-            let mut states: HashMap<RepoAddr, Vec<GitRef>> = HashMap::new();
+            let mut updates: Vec<(RepoAddr, Vec<CheckoutSyncStatus>)> = Vec::new();
-            for (addr, _) in &targets {
+
+            for (addr, paths) in &targets {
                 let events = db.query(addr.state_filter()).await?;
                 let latest = utils::latest(
                     events
                         .into_iter()
                         .filter(|event| !deletions.is_deleted(event)),
                 );
+
                 let Some(event) = latest else {
-                    continue;
+                    updates.push((addr.clone(), Vec::new()));
-                };
-                states.insert(addr.clone(), SyncState::parse(&event).refs);
-            }
-
-            let mut statuses: HashMap<RepoAddr, Vec<CheckoutSyncStatus>> = HashMap::new();
-            for (addr, paths) in &targets {
-                let Some(nostr_refs) = states.get(addr) else {
                     continue;
                 };
 
-                let remote_refs: Vec<(String, String)> = nostr_refs
+                let remote_refs: Vec<(String, String)> = SyncState::parse(&event)
+                    .refs
                     .iter()
                     .map(|git_ref| (git_ref.name.clone(), git_ref.commit.to_string()))
                     .collect();
@@ -206,41 +198,49 @@ impl SyncStatusStore {
                     }
                 }
 
-                if !list.is_empty() {
+                updates.push((addr.clone(), list));
-                    statuses.insert(addr.clone(), list);
-                }
             }
 
-            Ok::<_, Error>(statuses)
+            Ok::<_, Error>(updates)
         });
 
-        cx.spawn(async move |this, cx| {
+        self.tasks.push(cx.spawn(async move |this, cx| {
-            let statuses = match work.await {
+            let updates = work.await?;
-                Ok(statuses) => statuses,
+            let mut again = false;
-                Err(error) => {
-                    log::warn!("failed to compute nostr sync statuses: {error}");
-                    return this.update(cx, |this, _cx| {
-                        this.refresh.abort();
-                    });
-                }
-            };
 
-            let again = this.update(cx, |this, cx| {
+            this.update(cx, |this, cx| {
-                if this.statuses != statuses {
+                this.in_flight = false;
-                    this.statuses = statuses;
+                if !updates.is_empty() {
-                    cx.notify();
+                    cx.emit(SyncEvent::Updated(updates));
                 }
-                this.refresh.finish()
+                again = std::mem::take(&mut this.pass_pending);
             })?;
 
             if again {
                 this.update(cx, |this, cx| {
-                    this.refresh(cx);
+                    this.run_pass(cx);
                 })?;
             }
 
-            Ok::<_, Error>(())
+            Ok::<(), anyhow::Error>(())
-        })
+        }));
-        .detach();
+    }
+
+    fn schedule_poll(&mut self, cx: &mut Context<Self>) {
+        if self.poll_pending {
+            return;
+        }
+        self.poll_pending = true;
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            cx.background_executor().timer(SYNC_POLL).await;
+
+            this.update(cx, |this, cx| {
+                this.poll_pending = false;
+                this.run_pass(cx);
+            })?;
+
+            Ok::<(), anyhow::Error>(())
+        }));
     }
 }

docs/SYNC_STATUS_DIRECTORIES.md

@@ -79,11 +79,13 @@ Two pieces replace the fat global:
      directories cannot cheaply recompute git statuses, so re-querying per
      directory would multiply the expensive part.
    - Subscribes to `Backend` (`RepoUpdates`/`Synced` → probe active;
-     `SignerChanged` → clear active + in-flight, emit empty so directories
+     `SignerChanged` → clear active, emit empty so directories clear) and
-     clear) and observes `CheckoutsStore` (associations changed → probe
+     observes `CheckoutsStore` (associations changed → probe active). The
-     active). The `RepoListStore` observation edge disappears: targets now
+     `RepoListStore` observation edge disappears: targets now come from
-     come from directories, and announcement refreshes re-track through the
+     directories, and announcement refreshes re-track through the sidebar's
-     sidebar's own data path.
+     own data path.
+   - Probe requests reach the loop over the same flume-channel wake the
+     profile fetch service uses, so `probe` needs only `&self`.
    - Polls `active` every 180s while it is non-empty; no user signed in →
      clear and idle, same as today.
 2. `SyncStatusDirectory` (per surface entity): the statuses one surface
@@ -121,7 +123,7 @@ mechanism the profile directories rely on, and it composes with
 ```rust
 // signed_state/src/sync_status.rs
 
-/// Fresh statuses for these repositories; empty on signer change.
+/// Statuses for these repositories may have changed; empty on signer change.
 #[derive(Debug, Clone)]
 pub enum SyncEvent {
     Updated(Vec<(RepoAddr, Vec<CheckoutSyncStatus>)>),
@@ -132,8 +134,9 @@ impl EventEmitter<SyncEvent> for SyncStatusStore {}
 /// Thin global: probes tracked repos and broadcasts results. Holds no data.
 pub struct SyncStatusStore {
     active: RefCell<HashSet<RepoAddr>>,
-    in_flight: RefCell<bool>,
+    sender: Sender<()>,
-    pass_pending: RefCell<bool>,
+    in_flight: bool,
+    pass_pending: bool,
     poll_pending: bool,
     tasks: Vec<Task<()>>,
     _subscriptions: Vec<Subscription>,
@@ -198,12 +201,13 @@ these two views.
 ## Migration steps
 
 1. `signed_state`: reshape `sync_status.rs` — add `SyncEvent` +
-   `EventEmitter`, move the probe computation into a service method that
+   `EventEmitter`, move the probe computation into a pass that emits results,
-   emits results, and strip the store of `statuses`, `statuses_of`,
+   and strip the store of `statuses`, `statuses_of`, `unsynced`, and
-   `unsynced`, and `RefreshGate` (still used by `RepoListStore` and
+   `RefreshGate` (still used by `RepoListStore` and `CheckoutsStore`, so
-   `CheckoutsStore`, so `refresh.rs` stays). The `Backend` subscription and
+   `refresh.rs` stays). The `Backend` subscription and the 180s poll move to
-   the 180s poll move to the service; the `RepoListStore` observation is
+   the service; the `RepoListStore` observation is dropped. `RepoStore::run_push`
-   dropped.
+   replaces its post-push `refresh(cx)` with `probe(addr)` so badges update
+   without waiting for the relay pump.
 2. `signed_state`: add `SyncStatusDirectory` in the same file and export it
    from `lib.rs`. `init` keeps creating the global store.
 3. `workspace`: rewire `SidebarPanel` (directory + `track` in `refresh`,
@@ -221,8 +225,11 @@ these two views.
 - **Checkout association added mid-pass**: `CheckoutsStore` observation
   schedules a follow-up pass, so the new path is picked up promptly, not on
   the next 180s tick.
-- **Signer change mid-pass**: directories clear on the empty `Updated` event;
+- **Signer change mid-pass**: the in-flight pass is left alone so its
-  a late result for an untracked addr is dropped by `apply`.
+  completion releases the slot exactly once; its stale results are dropped by
+  directories because the addrs are no longer tracked. The empty `Updated`
+  event clears directories; a later probe re-runs over the new signer's
+  addrs.
 - **No published state for a repo**: the pass skips it (`states.get` miss),
   statuses stay empty, and the pending set is released like today's
   missing-entry behavior.