diff --git a/crates/signed_core/src/filters.rs b/crates/signed_core/src/filters.rs index f275343..3277462 100644 --- a/crates/signed_core/src/filters.rs +++ b/crates/signed_core/src/filters.rs @@ -68,8 +68,10 @@ pub fn announcements_by(public_key: PublicKey) -> Filter { } /// All repository announcements (for global discovery). -pub fn all_announcements(limit: usize) -> Filter { - Filter::new() - .kind(Kind::GitRepoAnnouncement) - .limit(limit) +/// +/// Unbounded: intended for negentropy sync, which reconciles sets +/// efficiently regardless of size. Local database queries with this +/// filter are served by LMDB, so they stay fast as the database grows. +pub fn all_announcements() -> Filter { + Filter::new().kind(Kind::GitRepoAnnouncement) } diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index f47c6e3..8b1f1f9 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -45,6 +45,14 @@ pub enum BackendEvent { /// so stores should re-query (no [`BackendEvent::NostrUpdate`] is fired /// for synced events). Synced, + /// A negentropy sync is in flight. Stores may re-query to render + /// incrementally; UI can show `current`/`total` progress. + SyncProgress { + /// Total events to process. + total: u64, + /// Events processed so far. + current: u64, + }, /// An event built locally was signed, broadcast and stored. Published(Box), /// An error occurred. @@ -67,6 +75,7 @@ pub struct Backend { inner: NostrBackend, current_user: Option, connected: bool, + sync_progress: Option<(u64, u64)>, tasks: Vec>>, } @@ -114,6 +123,7 @@ impl Backend { inner, current_user: None, connected: false, + sync_progress: None, tasks: vec![pump], }; @@ -388,6 +398,11 @@ impl Backend { self.connected } + /// Progress of the in-flight negentropy sync, if any: `(total, current)`. + pub fn sync_progress(&self) -> Option<(u64, u64)> { + self.sync_progress + } + /// Update the signer (any type implementing the async signer traits, /// e.g. `Keys`, `NostrConnect`, a browser extension proxy). pub fn set_signer(&mut self, new_signer: T, cx: &mut Context) @@ -506,12 +521,47 @@ impl Backend { /// Negentropy-sync the given filter against the bootstrap relays: /// reconciles the local database with the relays in both directions. - /// Emits [`BackendEvent::Synced`] on completion. + /// Emits [`BackendEvent::SyncProgress`] while running (throttled to + /// whole-percent changes) and [`BackendEvent::Synced`] on completion. pub fn sync_bootstrap(&mut self, filter: Filter, cx: &mut Context) { let backend = self.inner.clone(); + self.sync_progress = Some((0, 0)); + cx.notify(); + + let (tx, mut rx) = SyncProgress::channel(); + + self.tasks.push(cx.spawn(async move |this, cx| { + let mut last_percent: u64 = 0; + + while rx.changed().await.is_ok() { + let progress = *rx.borrow_and_update(); + let percent = (progress.percentage() * 100.0) as u64; + + if progress.current > 0 && percent != last_percent { + last_percent = percent; + + let alive = this.update(cx, |this, cx| { + this.sync_progress = Some((progress.total, progress.current)); + cx.emit(BackendEvent::SyncProgress { + total: progress.total, + current: progress.current, + }); + cx.notify(); + }); + + if alive.is_err() { + break; + } + } + } + + Ok(()) + })); + let task = cx.background_spawn(async move { - sync_bootstrap_only(&backend.client(), filter).await + let opts = SyncOptions::default().progress(tx); + sync_bootstrap_only(&backend.client(), filter, opts).await }); self.tasks.push(cx.spawn(async move |this, cx| { @@ -522,13 +572,17 @@ impl Backend { summary.received.len(), summary.sent.len() ); - this.update(cx, |_this, cx| { + this.update(cx, |this, cx| { + this.sync_progress = None; cx.emit(BackendEvent::Synced); cx.notify(); })?; } Err(e) => { - this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?; + this.update(cx, |this, cx| { + this.sync_progress = None; + cx.emit(BackendEvent::error(e.to_string())) + })?; } } Ok(()) @@ -596,7 +650,8 @@ pub(crate) async fn subscribe_bootstrap_only(client: &Client, filters: Vec Result { - let output = client.sync(filter).with(BOOTSTRAP_RELAYS).await?; + let output = client.sync(filter).with(BOOTSTRAP_RELAYS).opts(opts).await?; Ok(output.value) } diff --git a/crates/signed_state/src/profile.rs b/crates/signed_state/src/profile.rs index 64742e3..9bd05c0 100644 --- a/crates/signed_state/src/profile.rs +++ b/crates/signed_state/src/profile.rs @@ -257,7 +257,7 @@ impl ProfileStore { // Negentropy-sync with the bootstrap relays. Synced events // are written to the database directly (no NostrUpdate), so // re-apply from the database afterwards. - match sync_bootstrap_only(&client, filter).await { + match sync_bootstrap_only(&client, filter, SyncOptions::default()).await { Ok(_) => { this.update(cx, |this, cx| this.apply_seen(cx))?; } diff --git a/crates/signed_state/src/repo_list.rs b/crates/signed_state/src/repo_list.rs index ed83a7b..b0fdd4b 100644 --- a/crates/signed_state/src/repo_list.rs +++ b/crates/signed_state/src/repo_list.rs @@ -31,7 +31,7 @@ impl RepoListStore { event.kind == Kind::GitRepoAnnouncement && this.author.is_none_or(|a| a == event.pubkey) } - BackendEvent::Synced => true, + BackendEvent::Synced | BackendEvent::SyncProgress { .. } => true, _ => false, }; @@ -69,7 +69,7 @@ impl RepoListStore { backend.update(cx, |backend, cx| { let filter = match author { Some(a) => filters::announcements_by(a), - None => filters::all_announcements(500), + None => filters::all_announcements(), }; backend.sync_bootstrap(filter, cx); }); @@ -93,7 +93,7 @@ impl RepoListStore { loop { let filter = match author { Some(a) => filters::announcements_by(a), - None => filters::all_announcements(500), + None => filters::all_announcements(), }; let events = match client.database().query(filter).await { diff --git a/crates/workspace/src/workspace.rs b/crates/workspace/src/workspace.rs index e0f9d80..e8bbd42 100644 --- a/crates/workspace/src/workspace.rs +++ b/crates/workspace/src/workspace.rs @@ -18,10 +18,15 @@ impl Workspace { let repo_list = cx.new(|cx| RepoListView::new(window, cx)); let connected = backend.read(cx).is_connected(); + let sync_progress = backend.read(cx).sync_progress(); let subscription = cx.subscribe(&backend, |this, _backend, event, cx| { match event { BackendEvent::Connected => this.status = "Connected".into(), + BackendEvent::SyncProgress { total, current } => { + this.status = format!("Syncing repositories... {current}/{total}").into() + } + BackendEvent::Synced => this.status = "Connected".into(), BackendEvent::Error(error) => this.status = error.clone().into(), _ => return, } @@ -30,7 +35,9 @@ impl Workspace { Self { active_screen: repo_list.into(), - status: if connected { + status: if let Some((total, current)) = sync_progress { + format!("Syncing repositories... {current}/{total}").into() + } else if connected { "Connected".into() } else { "Connecting...".into()