From 2cd1068351bff7f6b410cc6da42d764676babb5d Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Thu, 15 Oct 2026 18:51:10 +0700 Subject: [PATCH] Reshape SyncStatusStore into a thin probe-and-emit store --- crates/signed_state/src/lib.rs | 2 +- crates/signed_state/src/repo.rs | 8 +- crates/signed_state/src/sync_status.rs | 246 ++++++++++++------------- docs/SYNC_STATUS_DIRECTORIES.md | 39 ++-- 4 files changed, 154 insertions(+), 141 deletions(-) diff --git a/crates/signed_state/src/lib.rs b/crates/signed_state/src/lib.rs index 2f9e15e..2c52b8c 100644 --- a/crates/signed_state/src/lib.rs +++ b/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, diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index a1b29d6..c98ac0b 100644 --- a/crates/signed_state/src/repo.rs +++ b/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}")); diff --git a/crates/signed_state/src/sync_status.rs b/crates/signed_state/src/sync_status.rs index 2c26fbb..a534f99 100644 --- a/crates/signed_state/src/sync_status.rs +++ b/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 nostr_sdk::prelude::*; +use flume::Sender; +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::repos::RepoListStore; -use crate::state::{GitRef, SyncState}; +use crate::state::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)>), +} + +/// Probes tracked repositories and broadcasts results. Holds no statuses. pub struct SyncStatusStore { - statuses: HashMap>, - refresh: RefreshGate, + active: RefCell>, + in_flight: bool, + pass_pending: bool, poll_pending: bool, + sender: Sender<()>, + tasks: Vec>>, _subscriptions: Vec, } +impl EventEmitter for SyncStatusStore {} + impl SyncStatusStore { pub fn global(cx: &App) -> Entity { cx.global::().0.clone() @@ -45,150 +56,131 @@ impl SyncStatusStore { } pub fn new(cx: &mut Context) -> 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); - - subscriptions.push(cx.observe(&repos, |this, _repos, cx| { - this.refresh(cx); - })); + let checkouts = CheckoutsStore::global(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 => { - this.statuses.clear(); - cx.notify(); - this.refresh(cx); - } - BackendEvent::RepoUpdates(_) | BackendEvent::Synced => this.refresh(cx), + BackendEvent::SignerChanged => this.clear(cx), + BackendEvent::RepoUpdates(_) | BackendEvent::Synced => this.run_pass(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(), - refresh: RefreshGate::default(), + active: RefCell::new(HashSet::new()), + sender, + in_flight: false, + pass_pending: false, poll_pending: false, + tasks: vec![task], _subscriptions: subscriptions, } } - /// Latest sync status per associated checkout. - pub fn statuses_of(&self, addr: &RepoAddr) -> Vec { - self.statuses.get(addr).cloned().unwrap_or_default() + /// Queue a probe pass; folded into any pass already in flight. + pub fn probe(&self, addrs: impl IntoIterator) { + 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. - pub fn unsynced(&self, addr: &RepoAddr) -> usize { - self.statuses - .get(addr) - .map(|list| { - list.iter() - .map(|checkout| checkout.status.ahead_total) - .sum() - }) - .unwrap_or(0) + // Keeps the in-flight pass so its completion releases the slot exactly once. + fn clear(&mut self, cx: &mut Context) { + let had_addrs = !self.active.borrow().is_empty(); + self.active.borrow_mut().clear(); + self.pass_pending = false; + + if had_addrs { + cx.emit(SyncEvent::Updated(Vec::new())); + } } - pub fn refresh(&mut self, cx: &mut Context) { + fn run_pass(&mut self, cx: &mut Context) { 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) { - 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.refresh.begin(); - let backend = Backend::global(cx); let client = backend.read(cx).client(); - let repos = RepoListStore::global(cx); - let announcements = repos.read(cx).announcements_of(&user); - let checkouts = CheckoutsStore::global(cx); + if backend.read(cx).current_user().is_none() { + self.clear(cx); + return; + } - let targets: Vec<(RepoAddr, Vec)> = announcements - .iter() - .map(|announcement| { - let addr = announcement.addr(); - let paths = checkouts.read(cx).associations_of(&addr); - (addr, paths) - }) - .collect(); + if self.active.borrow().is_empty() { + return; + } + + self.in_flight = true; + self.schedule_poll(cx); + + let targets: Vec<(RepoAddr, Vec)> = { + 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> = HashMap::new(); - for (addr, _) in &targets { + let mut updates: Vec<(RepoAddr, Vec)> = Vec::new(); + + 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; - }; - states.insert(addr.clone(), SyncState::parse(&event).refs); - } - - let mut statuses: HashMap> = HashMap::new(); - for (addr, paths) in &targets { - let Some(nostr_refs) = states.get(addr) else { + updates.push((addr.clone(), Vec::new())); 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() { - statuses.insert(addr.clone(), list); - } + updates.push((addr.clone(), list)); } - Ok::<_, Error>(statuses) + Ok::<_, Error>(updates) }); - cx.spawn(async move |this, cx| { - let statuses = match work.await { - Ok(statuses) => statuses, - Err(error) => { - log::warn!("failed to compute nostr sync statuses: {error}"); - return this.update(cx, |this, _cx| { - this.refresh.abort(); - }); - } - }; + self.tasks.push(cx.spawn(async move |this, cx| { + let updates = work.await?; + let mut again = false; - let again = this.update(cx, |this, cx| { - if this.statuses != statuses { - this.statuses = statuses; - cx.notify(); + this.update(cx, |this, cx| { + this.in_flight = false; + if !updates.is_empty() { + 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>(()) - }) - .detach(); + Ok::<(), anyhow::Error>(()) + })); + } + + fn schedule_poll(&mut self, cx: &mut Context) { + 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>(()) + })); } } diff --git a/docs/SYNC_STATUS_DIRECTORIES.md b/docs/SYNC_STATUS_DIRECTORIES.md index 7d54270..392ddef 100644 --- a/docs/SYNC_STATUS_DIRECTORIES.md +++ b/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 - clear) and observes `CheckoutsStore` (associations changed → probe - active). The `RepoListStore` observation edge disappears: targets now - come from directories, and announcement refreshes re-track through the - sidebar's own data path. + `SignerChanged` → clear active, emit empty so directories clear) and + observes `CheckoutsStore` (associations changed → probe active). The + `RepoListStore` observation edge disappears: targets now come from + directories, and announcement refreshes re-track through the sidebar's + 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)>), @@ -132,8 +134,9 @@ impl EventEmitter for SyncStatusStore {} /// Thin global: probes tracked repos and broadcasts results. Holds no data. pub struct SyncStatusStore { active: RefCell>, - in_flight: RefCell, - pass_pending: RefCell, + sender: Sender<()>, + in_flight: bool, + pass_pending: bool, poll_pending: bool, tasks: Vec>, _subscriptions: Vec, @@ -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 - emits results, and strip the store of `statuses`, `statuses_of`, - `unsynced`, and `RefreshGate` (still used by `RepoListStore` and - `CheckoutsStore`, so `refresh.rs` stays). The `Backend` subscription and - the 180s poll move to the service; the `RepoListStore` observation is - dropped. + `EventEmitter`, move the probe computation into a pass that emits results, + and strip the store of `statuses`, `statuses_of`, `unsynced`, and + `RefreshGate` (still used by `RepoListStore` and `CheckoutsStore`, so + `refresh.rs` stays). The `Backend` subscription and the 180s poll move to + the service; the `RepoListStore` observation is dropped. `RepoStore::run_push` + 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; - a late result for an untracked addr is dropped by `apply`. +- **Signer change mid-pass**: the in-flight pass is left alone so its + 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.