reya/signed
1
0
Fork 0
Code Issues Pull Requests Actions 5 Releases Activity

Detach Background Tasks Directly

Browse Source
This commit is contained in:
reya
2026-09-10 07:26:00 +07:00
parent 5f37f2ff2a
commit 3cfdffdf82
7 changed files with 144 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/signed_state/src/backend.rs
+56 -104
View File
@@ -1,11 +1,9 @@
use std::collections::hash_map::DefaultHasher;
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::future::Future; use std::future::Future;
use std::hash::{Hash, Hasher};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::str::FromStr; use std::str::FromStr;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant}; use std::time::Duration;
use anyhow::{Context as AnyhowContext, Error, anyhow, bail}; use anyhow::{Context as AnyhowContext, Error, anyhow, bail};
use bitcoin_hashes::sha1::Hash as Sha1Hash; use bitcoin_hashes::sha1::Hash as Sha1Hash;
@@ -35,9 +33,6 @@ pub const BOOTSTRAP_RELAYS: [&str; 4] = [
/// Relays used to index the user's NIP-65 relay list. /// Relays used to index the user's NIP-65 relay list.
pub const INDEXER_RELAYS: [&str; 2] = ["wss://indexer.coracle.social", "wss://user.kindpag.es"]; pub const INDEXER_RELAYS: [&str; 2] = ["wss://indexer.coracle.social", "wss://user.kindpag.es"];
/// How long an identical fetch or sync request is suppressed after it started.
const FETCH_DEDUP_WINDOW: Duration = Duration::from_secs(5 * 60);
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub enum BackendEvent { pub enum BackendEvent {
/// User has no signer configured. /// User has no signer configured.
@@ -82,11 +77,8 @@ pub struct Backend {
sync_progress: Option<(u64, u64)>, sync_progress: Option<(u64, u64)>,
/// True when the stored credential is NIP-49 encrypted. /// True when the stored credential is NIP-49 encrypted.
passphrase_required: bool, passphrase_required: bool,
/// Fingerprints of recently started fetches and syncs, a relay plus filter set.
recent_fetches: HashMap<u64, Instant>,
/// Repositories with a push in flight, mirror or checkout based. /// Repositories with a push in flight, mirror or checkout based.
pushing_repos: Arc<Mutex<HashSet<RepoAddr>>>, pushing_repos: Arc<Mutex<HashSet<RepoAddr>>>,
tasks: Vec<Task<Result<(), Error>>>,
} }
struct GlobalBackend(Entity<Backend>); struct GlobalBackend(Entity<Backend>);
@@ -124,7 +116,7 @@ impl Backend {
pub(crate) fn new(client: Client, signer: UniversalSigner, cx: &mut Context<Self>) -> Self { pub(crate) fn new(client: Client, signer: UniversalSigner, cx: &mut Context<Self>) -> Self {
let pump_client = client.clone(); let pump_client = client.clone();
let pump = cx.spawn(async move |this, cx| { let pump: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let mut notifications = pump_client.notifications(); let mut notifications = pump_client.notifications();
while let Some(notification) = notifications.next().await { while let Some(notification) = notifications.next().await {
@@ -145,29 +137,21 @@ impl Backend {
Ok(()) Ok(())
}); });
pump.detach();
let mut this = Self { let mut this = Self {
client, client,
signer, signer,
current_user: None, current_user: None,
sync_progress: None, sync_progress: None,
passphrase_required: false, passphrase_required: false,
recent_fetches: HashMap::new(),
pushing_repos: Arc::new(Mutex::new(HashSet::new())), pushing_repos: Arc::new(Mutex::new(HashSet::new())),
tasks: vec![pump],
}; };
this.bootstrap(cx); this.bootstrap(cx);
this this
} }
/// 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);
}
/// Bootstrap the client. /// Bootstrap the client.
/// ///
/// Restore the saved session, if any. /// Restore the saved session, if any.
@@ -188,7 +172,7 @@ impl Backend {
Ok::<(), Error>(()) Ok::<(), Error>(())
}); });
self.push_task(cx.spawn(async move |this, cx| { let notify_task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
match task.await { match task.await {
Ok(()) => { Ok(()) => {
this.update(cx, |_this, cx| cx.notify())?; this.update(cx, |_this, cx| cx.notify())?;
@@ -198,7 +182,8 @@ impl Backend {
} }
} }
Ok(()) Ok(())
})); });
notify_task.detach();
self.restore_session(cx); self.restore_session(cx);
} }
@@ -216,7 +201,7 @@ impl Backend {
let user = cx.read_credentials(USER_KEYRING); let user = cx.read_credentials(USER_KEYRING);
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let content = match user.await { let content = match user.await {
Ok(Some((_username, secret))) => String::from_utf8(secret)?, Ok(Some((_username, secret))) => String::from_utf8(secret)?,
_ => { _ => {
@@ -264,7 +249,8 @@ impl Backend {
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// Decrypt the NIP-49 keyring credential with the given passphrase. /// Decrypt the NIP-49 keyring credential with the given passphrase.
@@ -982,14 +968,15 @@ impl Backend {
let pubkey = keys.public_key().to_hex(); let pubkey = keys.public_key().to_hex();
let write = cx.write_credentials(USER_KEYRING, &pubkey, nsec.as_bytes()); let write = cx.write_credentials(USER_KEYRING, &pubkey, nsec.as_bytes());
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
if let Err(e) = write.await { if let Err(e) = write.await {
this.update(cx, |_, cx| cx.emit(BackendEvent::error(e.to_string())))?; this.update(cx, |_, cx| cx.emit(BackendEvent::error(e.to_string())))?;
return Ok(()); return Ok(());
} }
this.update(cx, |this, cx| this.set_signer(keys, cx))?; this.update(cx, |this, cx| this.set_signer(keys, cx))?;
Ok(()) Ok(())
})); });
task.detach();
} }
/// Login with a `bunker://...` URI, NIP-46. /// Login with a `bunker://...` URI, NIP-46.
@@ -1008,7 +995,7 @@ impl Backend {
let credential = with_master_key(&uri_string, &keys); let credential = with_master_key(&uri_string, &keys);
let write = cx.write_credentials(USER_KEYRING, "bunker", credential.as_bytes()); let write = cx.write_credentials(USER_KEYRING, "bunker", credential.as_bytes());
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let result = async { let result = async {
let mut signer = NostrConnect::new( let mut signer = NostrConnect::new(
connect_uri, connect_uri,
@@ -1033,14 +1020,15 @@ impl Backend {
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// Remove the saved credential and reset to an anonymous session. /// Remove the saved credential and reset to an anonymous session.
pub fn logout(&mut self, cx: &mut Context<Self>) { pub fn logout(&mut self, cx: &mut Context<Self>) {
let delete = cx.delete_credentials(USER_KEYRING); let delete = cx.delete_credentials(USER_KEYRING);
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
delete.await.ok(); delete.await.ok();
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
@@ -1053,14 +1041,15 @@ impl Backend {
})?; })?;
Ok(()) Ok(())
})); });
task.detach();
} }
/// Fetch the user's grasp list and add the listed grasp servers as relays. /// Fetch the user's grasp list and add the listed grasp servers as relays.
fn bootstrap_user(&mut self, public_key: PublicKey, cx: &mut Context<Self>) { fn bootstrap_user(&mut self, public_key: PublicKey, cx: &mut Context<Self>) {
let client = self.client.clone(); let client = self.client.clone();
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let result = async { let result = async {
let events: Vec<Event> = client let events: Vec<Event> = client
.fetch_events(filters::grasp_list(public_key)) .fetch_events(filters::grasp_list(public_key))
@@ -1082,7 +1071,8 @@ impl Backend {
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// Get the nostr client. /// Get the nostr client.
@@ -1123,7 +1113,7 @@ impl Backend {
<T as AsyncSignEvent>::Error: std::error::Error + Send + Sync + 'static, <T as AsyncSignEvent>::Error: std::error::Error + Send + Sync + 'static,
<T as AsyncNip44>::Error: std::error::Error + Send + Sync + 'static, <T as AsyncNip44>::Error: std::error::Error + Send + Sync + 'static,
{ {
let task = cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
match new_signer.get_public_key_async().await { match new_signer.get_public_key_async().await {
Ok(public_key) => { Ok(public_key) => {
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
@@ -1144,7 +1134,7 @@ impl Backend {
Ok(()) Ok(())
}); });
self.push_task(task); task.detach();
} }
/// Add relays and connect to them. /// Add relays and connect to them.
@@ -1159,7 +1149,7 @@ impl Backend {
Ok::<(), Error>(()) Ok::<(), Error>(())
}); });
self.push_task(cx.spawn(async move |this, cx| { let notify_task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
match task.await { match task.await {
Ok(()) => { Ok(()) => {
this.update(cx, |_this, cx| cx.notify())?; this.update(cx, |_this, cx| cx.notify())?;
@@ -1169,74 +1159,49 @@ impl Backend {
} }
} }
Ok(()) Ok(())
})); });
} notify_task.detach();
/// Whether an identical fetch started within [`FETCH_DEDUP_WINDOW`] is still recent.
///
/// Records the fingerprint when returning `false`, pruning expired entries first.
fn fetch_recently_started(&mut self, fingerprint: u64) -> bool {
self.recent_fetches
.retain(|_, started| started.elapsed() < FETCH_DEDUP_WINDOW);
if self.recent_fetches.contains_key(&fingerprint) {
return true;
}
self.recent_fetches.insert(fingerprint, Instant::now());
false
} }
/// Connect to a repository's announced relays, its NIP-34 `relays` tag. /// Connect to a repository's announced relays, its NIP-34 `relays` tag.
///
/// Callers are responsible for not repeating this for relays they already
/// connected, e.g. `RepoStore::repo_relays`.
pub fn connect_repo_relays( pub fn connect_repo_relays(
&mut self, &mut self,
relays: Vec<RelayUrl>, relays: Vec<RelayUrl>,
filters: Vec<Filter>, filters: Vec<Filter>,
cx: &mut Context<Self>, cx: &mut Context<Self>,
) { ) {
let relay_strs: Vec<&str> = relays.iter().map(|url| url.as_str()).collect();
let fingerprint = fetch_fingerprint(&relay_strs, &filters);
if self.fetch_recently_started(fingerprint) {
log::debug!("skipping duplicate repo relay fetch");
return;
}
let client = self.client.clone(); let client = self.client.clone();
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |_this, _cx| {
if let Err(e) = connect_repo_relays(&client, relays, filters).await { if let Err(e) = connect_repo_relays(&client, relays, filters).await {
log::warn!("repo relay fetch failed: {e}"); log::warn!("repo relay fetch failed: {e}");
// Allow an immediate retry after a failure.
this.update(cx, |this, _cx| {
this.recent_fetches.remove(&fingerprint);
})
.ok();
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// One-shot subscription on the bootstrap relays only. /// One-shot subscription on the bootstrap relays only.
pub fn subscribe_bootstrap(&mut self, filters: Vec<Filter>, cx: &mut Context<Self>) { pub fn subscribe_bootstrap(&mut self, filters: Vec<Filter>, cx: &mut Context<Self>) {
let client = self.client.clone(); let client = self.client.clone();
let task = let fetch =
cx.background_spawn(async move { subscribe_bootstrap_only(&client, filters).await }); cx.background_spawn(async move { subscribe_bootstrap_only(&client, filters).await });
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
if let Err(e) = task.await { if let Err(e) = fetch.await {
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?; this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// Negentropy-sync the given filter against the bootstrap relays. /// Negentropy-sync the given filter against the bootstrap relays.
pub fn sync_bootstrap(&mut self, filter: Filter, cx: &mut Context<Self>) { pub fn sync_bootstrap(&mut self, filter: Filter, cx: &mut Context<Self>) {
let fingerprint = fetch_fingerprint(&BOOTSTRAP_RELAYS, std::slice::from_ref(&filter));
if self.fetch_recently_started(fingerprint) {
log::debug!("skipping duplicate bootstrap sync");
return;
}
let client = self.client.clone(); let client = self.client.clone();
self.sync_progress = Some((0, 0)); self.sync_progress = Some((0, 0));
@@ -1244,7 +1209,7 @@ impl Backend {
let (tx, mut rx) = SyncProgress::channel(); let (tx, mut rx) = SyncProgress::channel();
self.push_task(cx.spawn(async move |this, cx| { let progress_task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let mut last_percent: u64 = 0; let mut last_percent: u64 = 0;
while rx.changed().await.is_ok() { while rx.changed().await.is_ok() {
@@ -1270,15 +1235,16 @@ impl Backend {
} }
Ok(()) Ok(())
})); });
progress_task.detach();
let task = cx.background_spawn(async move { let sync = cx.background_spawn(async move {
let opts = SyncOptions::default().progress(tx); let opts = SyncOptions::default().progress(tx);
sync_bootstrap_only(&client, filter, opts).await sync_bootstrap_only(&client, filter, opts).await
}); });
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
match task.await { match sync.await {
Ok(summary) => { Ok(summary) => {
log::debug!( log::debug!(
"sync done: {} received, {} sent", "sync done: {} received, {} sent",
@@ -1294,14 +1260,13 @@ impl Backend {
Err(e) => { Err(e) => {
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
this.sync_progress = None; this.sync_progress = None;
// Allow an immediate retry after a failure.
this.recent_fetches.remove(&fingerprint);
cx.emit(BackendEvent::error(e.to_string())) cx.emit(BackendEvent::error(e.to_string()))
})?; })?;
} }
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// Sign, broadcast and locally store an event. /// Sign, broadcast and locally store an event.
@@ -1361,17 +1326,18 @@ impl Backend {
/// Sign, broadcast and store an event without awaiting the result. /// Sign, broadcast and store an event without awaiting the result.
fn send_fire_and_forget(&mut self, builder: EventBuilder, cx: &mut Context<Self>) { fn send_fire_and_forget(&mut self, builder: EventBuilder, cx: &mut Context<Self>) {
let task = self.send(builder, cx); let publish = self.send(builder, cx);
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
if let Err(e) = task.await { if let Err(e) = publish.await {
this.update(cx, |_this, cx| { this.update(cx, |_this, cx| {
cx.emit(BackendEvent::error(e.to_string())); cx.emit(BackendEvent::error(e.to_string()));
}) })
.ok(); .ok();
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// Publish NIP-09 deletions for `events`, best-effort. /// Publish NIP-09 deletions for `events`, best-effort.
@@ -1387,14 +1353,15 @@ impl Backend {
tags.push(Tag::parse(["k", &event.kind.to_string()]).expect("valid kind tag")); tags.push(Tag::parse(["k", &event.kind.to_string()]).expect("valid kind tag"));
} }
let task = self.send(EventBuilder::new(Kind::EventDeletion, "").tags(tags), cx); let publish = self.send(EventBuilder::new(Kind::EventDeletion, "").tags(tags), cx);
self.push_task(cx.spawn(async move |_this, _cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |_this, _cx| {
if let Err(e) = task.await { if let Err(e) = publish.await {
log::warn!("failed to retract repository events: {e}"); log::warn!("failed to retract repository events: {e}");
} }
Ok(()) Ok(())
})); });
task.detach();
} }
} }
@@ -1417,21 +1384,6 @@ async fn broadcast_event(client: &Client, event: &Event) -> Result<Event, Error>
Ok(event.clone()) Ok(event.clone())
} }
/// Fingerprint of a relay and filter set, for fetch dedup.
///
/// Relays and filters are sorted first, so the fingerprint is order-independent.
fn fetch_fingerprint(relays: &[&str], filters: &[Filter]) -> u64 {
let mut relays: Vec<&str> = relays.to_vec();
relays.sort_unstable();
let mut filters: Vec<&Filter> = filters.iter().collect();
filters.sort_unstable();
let mut hasher = DefaultHasher::new();
relays.hash(&mut hasher);
filters.hash(&mut hasher);
hasher.finish()
}
/// Add the given relays, connect and fetch the filters. /// Add the given relays, connect and fetch the filters.
async fn connect_repo_relays( async fn connect_repo_relays(
client: &Client, client: &Client,
crates/signed_state/src/checkouts.rs
+13 -23
View File
@@ -3,7 +3,7 @@ use std::path::{Path, PathBuf};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use anyhow::Error; use anyhow::Error;
use gpui::{App, AppContext, Context, Entity, Global, Subscription, Task}; use gpui::{App, AppContext, Context, Entity, Global, Subscription};
use nostr::prelude::*; use nostr::prelude::*;
use settings::{CheckoutRecord, SettingsStore}; use settings::{CheckoutRecord, SettingsStore};
use signed_core::{Announcement, RepoAddr}; use signed_core::{Announcement, RepoAddr};
@@ -105,7 +105,6 @@ pub struct CheckoutsStore {
/// The local pass runs a full pass again once this is older than the /// The local pass runs a full pass again once this is older than the
/// reconciliation cadence, so remote moves still land. /// reconciliation cadence, so remote moves still land.
last_full_sync: Option<Instant>, last_full_sync: Option<Instant>,
tasks: Vec<Task<Result<(), Error>>>,
_subscriptions: Vec<Subscription>, _subscriptions: Vec<Subscription>,
} }
@@ -165,7 +164,6 @@ impl CheckoutsStore {
refresh: RefreshGate::default(), refresh: RefreshGate::default(),
local_pending: false, local_pending: false,
last_full_sync: None, last_full_sync: None,
tasks: Vec::new(),
_subscriptions: subscriptions, _subscriptions: subscriptions,
}; };
@@ -176,14 +174,6 @@ impl CheckoutsStore {
store 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);
}
/// Remember a successful local-checkout use. /// Remember a successful local-checkout use.
pub fn record(&mut self, path: PathBuf, addr: RepoAddr, cx: &mut Context<Self>) { pub fn record(&mut self, path: PathBuf, addr: RepoAddr, cx: &mut Context<Self>) {
if cfg!(target_arch = "wasm32") { if cfg!(target_arch = "wasm32") {
@@ -302,12 +292,11 @@ impl CheckoutsStore {
return; return;
} }
let task = cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
cx.background_executor().timer(REFRESH_DEBOUNCE).await; cx.background_executor().timer(REFRESH_DEBOUNCE).await;
this.update(cx, |this, cx| this.run_refresh(cx)) this.update(cx, |this, cx| this.run_refresh(cx))
}); })
.detach();
self.push_task(task);
} }
/// One full resolve and apply cycle, the debounced entry point. /// One full resolve and apply cycle, the debounced entry point.
@@ -388,7 +377,7 @@ impl CheckoutsStore {
Ok::<_, Error>((associations, statuses, push_statuses)) Ok::<_, Error>((associations, statuses, push_statuses))
}); });
self.push_task(cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
let (associations, statuses, push_statuses) = match work.await { let (associations, statuses, push_statuses) = match work.await {
Ok(results) => results, Ok(results) => results,
Err(_) => { Err(_) => {
@@ -435,7 +424,8 @@ impl CheckoutsStore {
})?; })?;
Ok(()) Ok(())
})); })
.detach();
} }
/// Schedule the fast local status pass, unless one is already pending. /// Schedule the fast local status pass, unless one is already pending.
@@ -449,15 +439,14 @@ impl CheckoutsStore {
} }
self.local_pending = true; self.local_pending = true;
let task = cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
cx.background_executor().timer(LOCAL_POLL).await; cx.background_executor().timer(LOCAL_POLL).await;
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
this.local_pending = false; this.local_pending = false;
this.local_tick(cx); this.local_tick(cx);
}) })
}); })
.detach();
self.push_task(task);
} }
/// The fast local status pass. /// The fast local status pass.
@@ -525,7 +514,7 @@ impl CheckoutsStore {
Ok::<_, Error>((statuses, push_statuses)) Ok::<_, Error>((statuses, push_statuses))
}); });
self.push_task(cx.spawn(async move |this, cx| { let task: gpui::Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let Ok((statuses, push_statuses)) = work.await else { let Ok((statuses, push_statuses)) = work.await else {
// Git reads are best-effort, keep the last results. // Git reads are best-effort, keep the last results.
return Ok(()); return Ok(());
@@ -550,7 +539,8 @@ impl CheckoutsStore {
})?; })?;
Ok(()) Ok(())
})); });
task.detach();
} }
} }
crates/signed_state/src/local_repos.rs
+4 -5
View File
@@ -2,7 +2,7 @@ use std::path::{Path, PathBuf};
use std::sync::Arc; use std::sync::Arc;
use anyhow::Error; use anyhow::Error;
use gpui::{App, AppContext, Context, Entity, Global, Task}; use gpui::{App, AppContext, Context, Entity, Global};
use signed_git::find_git_repos; use signed_git::find_git_repos;
struct GlobalLocalReposStore(Entity<LocalReposStore>); struct GlobalLocalReposStore(Entity<LocalReposStore>);
@@ -19,7 +19,6 @@ pub struct LocalReposStore {
pub scanning: bool, pub scanning: bool,
/// A scan was requested while one was already running. /// A scan was requested while one was already running.
scan_dirty: bool, scan_dirty: bool,
tasks: Vec<Task<Result<(), Error>>>,
} }
impl LocalReposStore { impl LocalReposStore {
@@ -39,7 +38,6 @@ impl LocalReposStore {
repos: Arc::new(Vec::new()), repos: Arc::new(Vec::new()),
scanning: false, scanning: false,
scan_dirty: false, scan_dirty: false,
tasks: Vec::new(),
}; };
store.rescan(cx); store.rescan(cx);
store store
@@ -81,7 +79,7 @@ impl LocalReposStore {
repos repos
}); });
self.tasks.push(cx.spawn(async move |this, cx| { let task: gpui::Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let repos = work.await; let repos = work.await;
let again = this.update(cx, |this, cx| { let again = this.update(cx, |this, cx| {
this.repos = Arc::new(repos); this.repos = Arc::new(repos);
@@ -99,6 +97,7 @@ impl LocalReposStore {
} }
Ok(()) Ok(())
})); });
task.detach();
} }
} }
crates/signed_state/src/profile.rs
+12 -20
View File
@@ -75,7 +75,6 @@ pub struct ProfileStore {
seen: RefCell<HashSet<PublicKey>>, seen: RefCell<HashSet<PublicKey>>,
/// Sender for queuing fetch requests, batched by a background task. /// Sender for queuing fetch requests, batched by a background task.
sender: Sender<PublicKey>, sender: Sender<PublicKey>,
tasks: Vec<Task<Result<(), Error>>>,
_subscription: Subscription, _subscription: Subscription,
} }
@@ -114,17 +113,15 @@ impl ProfileStore {
let (sender, receiver) = flume::unbounded::<PublicKey>(); let (sender, receiver) = flume::unbounded::<PublicKey>();
let entity = cx.entity().downgrade(); let entity = cx.entity().downgrade();
let mut tasks = Vec::new(); cx.spawn(async move |_this, cx| {
tasks.push(cx.spawn(async move |_this, cx| {
Self::handle_requests(entity, &client, &receiver, cx).await Self::handle_requests(entity, &client, &receiver, cx).await
})); })
.detach();
let mut store = Self { let mut store = Self {
profiles: HashMap::new(), profiles: HashMap::new(),
seen: RefCell::new(HashSet::new()), seen: RefCell::new(HashSet::new()),
sender, sender,
tasks,
_subscription: subscription, _subscription: subscription,
}; };
@@ -132,14 +129,6 @@ impl ProfileStore {
store 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. /// Get a profile.
/// ///
/// Returns a placeholder with default metadata. Queues a fetch when the profile is not cached yet. /// Returns a placeholder with default metadata. Queues a fetch when the profile is not cached yet.
@@ -181,7 +170,7 @@ impl ProfileStore {
Ok::<_, Error>(profiles) Ok::<_, Error>(profiles)
}); });
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let profiles = work.await?; let profiles = work.await?;
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
@@ -192,7 +181,8 @@ impl ProfileStore {
})?; })?;
Ok(()) Ok(())
})); });
task.detach();
} }
/// Re-read the latest metadata of an author from the local database. /// Re-read the latest metadata of an author from the local database.
@@ -217,7 +207,7 @@ impl ProfileStore {
Ok::<_, Error>(profile) Ok::<_, Error>(profile)
}); });
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let profile = work.await?; let profile = work.await?;
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
@@ -228,7 +218,8 @@ impl ProfileStore {
})?; })?;
Ok(()) Ok(())
})); });
task.detach();
} }
/// Re-read the latest metadata of every requested author from the local database. /// Re-read the latest metadata of every requested author from the local database.
@@ -273,7 +264,7 @@ impl ProfileStore {
Ok::<_, Error>(profiles) Ok::<_, Error>(profiles)
}); });
self.push_task(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let profiles = work.await?; let profiles = work.await?;
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
@@ -284,7 +275,8 @@ impl ProfileStore {
})?; })?;
Ok(()) Ok(())
})); });
task.detach();
} }
/// Sync metadata for requested authors in batches, debounced to collect requests. /// Sync metadata for requested authors in batches, debounced to collect requests.
crates/signed_state/src/repo.rs
+20 -21
View File
@@ -81,7 +81,6 @@ pub struct RepoStore {
root_fetches: HashSet<EventId>, root_fetches: HashSet<EventId>,
/// Refresh coalescing, see [`RefreshGate`]. /// Refresh coalescing, see [`RefreshGate`].
refresh: RefreshGate, refresh: RefreshGate,
tasks: Vec<Task<Result<(), Error>>>,
_subscription: Subscription, _subscription: Subscription,
} }
@@ -148,7 +147,6 @@ impl RepoStore {
root_fetches: HashSet::new(), root_fetches: HashSet::new(),
refresh: RefreshGate::default(), refresh: RefreshGate::default(),
_subscription: subscription, _subscription: subscription,
tasks: Vec::new(),
}; };
store.subscribe_remote(cx); store.subscribe_remote(cx);
@@ -229,14 +227,12 @@ impl RepoStore {
return; return;
} }
let task = cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
cx.background_executor().timer(REFRESH_DEBOUNCE).await; cx.background_executor().timer(REFRESH_DEBOUNCE).await;
this.update(cx, |this, cx| this.run_refresh(cx)) this.update(cx, |this, cx| this.run_refresh(cx))
}); })
.detach();
self.tasks.retain(|task| !task.is_ready());
self.tasks.push(task);
} }
fn run_refresh(&mut self, cx: &mut Context<Self>) { fn run_refresh(&mut self, cx: &mut Context<Self>) {
@@ -372,9 +368,7 @@ impl RepoStore {
)) ))
}); });
self.tasks.retain(|task| !task.is_ready()); cx.spawn(async move |this, cx| {
self.tasks.push(cx.spawn(async move |this, cx| {
let ( let (
announcement, announcement,
state, state,
@@ -466,7 +460,8 @@ impl RepoStore {
} }
Ok(()) Ok(())
})); })
.detach();
} }
/// Resolve the status of a root event, an issue, patch or PR, per NIP-34. /// Resolve the status of a root event, an issue, patch or PR, per NIP-34.
@@ -638,7 +633,7 @@ impl RepoStore {
.collect() .collect()
}; };
self.tasks.push(cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
// The PR references the root patch event. // The PR references the root patch event.
// Viewers can then find the patch without carrying it inline. // Viewers can then find the patch without carrying it inline.
let root_patch = match publish_patch_series( let root_patch = match publish_patch_series(
@@ -822,7 +817,8 @@ impl RepoStore {
} }
Ok(()) Ok(())
})); })
.detach();
} }
/// Update a pull request. /// Update a pull request.
@@ -894,7 +890,7 @@ impl RepoStore {
.map(|a| a.clone.clone()) .map(|a| a.clone.clone())
.unwrap_or_default(); .unwrap_or_default();
self.tasks.push(cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
if let Err(e) = publish_patch_series( if let Err(e) = publish_patch_series(
&this, &this,
cx, cx,
@@ -943,7 +939,8 @@ impl RepoStore {
} }
Ok(()) Ok(())
})); })
.detach();
} }
/// Set the status of a root event. /// Set the status of a root event.
@@ -1040,7 +1037,7 @@ impl RepoStore {
Ok::<_, Error>(applied) Ok::<_, Error>(applied)
}); });
self.tasks.push(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
match apply.await { match apply.await {
Ok(applied) => { Ok(applied) => {
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
@@ -1062,7 +1059,8 @@ impl RepoStore {
} }
} }
Ok(()) Ok(())
})); });
task.detach();
} }
/// The latest announcement of this repository, /// The latest announcement of this repository,
@@ -1335,17 +1333,18 @@ impl RepoStore {
self.last_error = None; self.last_error = None;
let backend = Backend::global(cx); let backend = Backend::global(cx);
let task = backend.update(cx, |backend, cx| backend.send(builder, cx)); let publish = backend.update(cx, |backend, cx| backend.send(builder, cx));
self.tasks.push(cx.spawn(async move |this, cx| { let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
if let Err(e) = task.await { if let Err(e) = publish.await {
this.update(cx, |this, cx| { this.update(cx, |this, cx| {
this.last_error = Some(e.to_string()); this.last_error = Some(e.to_string());
cx.notify(); cx.notify();
})?; })?;
} }
Ok(()) Ok(())
})); });
task.detach();
} }
} }
crates/signed_state/src/repo_list.rs
+7 -17
View File
@@ -3,7 +3,7 @@ use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use anyhow::Error; use anyhow::Error;
use gpui::{App, AppContext, Context, Entity, Global, Subscription, Task}; use gpui::{App, AppContext, Context, Entity, Global, Subscription};
use nostr_sdk::prelude::*; use nostr_sdk::prelude::*;
use signed_core::{Announcement, Deletions, RepoAddr, filters, repo_addr}; use signed_core::{Announcement, Deletions, RepoAddr, filters, repo_addr};
@@ -55,7 +55,6 @@ pub struct RepoListStore {
pub counts: Arc<HashMap<RepoAddr, RepoActivityCounts>>, pub counts: Arc<HashMap<RepoAddr, RepoActivityCounts>>,
/// Refresh coalescing, see [`RefreshGate`]. /// Refresh coalescing, see [`RefreshGate`].
refresh: RefreshGate, refresh: RefreshGate,
tasks: Vec<Task<Result<(), Error>>>,
_subscription: Subscription, _subscription: Subscription,
} }
@@ -114,7 +113,6 @@ impl RepoListStore {
counts: Arc::new(HashMap::new()), counts: Arc::new(HashMap::new()),
refresh: RefreshGate::default(), refresh: RefreshGate::default(),
_subscription: subscription, _subscription: subscription,
tasks: Vec::new(),
}; };
store.subscribe_remote(cx); store.subscribe_remote(cx);
@@ -133,14 +131,6 @@ impl RepoListStore {
.collect() .collect()
} }
/// 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);
}
/// Negentropy-sync announcements with the bootstrap relays. /// Negentropy-sync announcements with the bootstrap relays.
fn subscribe_remote(&mut self, cx: &mut Context<Self>) { fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
let backend = Backend::global(cx); let backend = Backend::global(cx);
@@ -171,13 +161,12 @@ impl RepoListStore {
return; return;
} }
let task = cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
cx.background_executor().timer(REFRESH_DEBOUNCE).await; cx.background_executor().timer(REFRESH_DEBOUNCE).await;
this.update(cx, |this, cx| this.run_refresh(cx)) this.update(cx, |this, cx| this.run_refresh(cx))
}); })
.detach();
self.push_task(task);
} }
/// One query and apply cycle, the debounced entry point. /// One query and apply cycle, the debounced entry point.
@@ -294,7 +283,7 @@ impl RepoListStore {
Ok::<_, Error>((announcements, last_activity, counts)) Ok::<_, Error>((announcements, last_activity, counts))
}); });
self.push_task(cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
let (announcements, last_activity, counts) = match work.await { let (announcements, last_activity, counts) = match work.await {
Ok(results) => results, Ok(results) => results,
// Database errors are transient, keep the last list. // Database errors are transient, keep the last list.
@@ -321,6 +310,7 @@ impl RepoListStore {
} }
Ok(()) Ok(())
})); })
.detach();
} }
} }
docs/backend-rearchitecture.md
+32 -2
View File
@@ -440,6 +440,18 @@ octal-escaped/non-ASCII quoted paths) before committing to the swap.
## 6. Remove the `tasks: Vec<Task<...>>` + `push_task` boilerplate — use `Task::detach()` ## 6. Remove the `tasks: Vec<Task<...>>` + `push_task` boilerplate — use `Task::detach()`
> **Status: done.** Removed the `tasks` field and `push_task` from all six
> stores (`backend.rs`, `checkouts.rs`, `local_repos.rs`, `profile.rs`,
> `repo.rs`, `repo_list.rs`); every call site now ends in `.detach()`
> instead. As with §14, most `cx.spawn` sites lost their type-inference
> anchor and needed an explicit `let task: Task<Result<(), Error>> = ...`
> (or `gpui::Task<...>` where `Task` wasn't imported) before `.detach()`.
> A few closures that captured a variable also named `task` (the awaited
> inner task) were given a distinct outer name (`notify_task`, `publish`,
> `fetch`, `sync`) to avoid a confusing shadow. `cargo check --workspace`
> and `cargo test -p signed_state` (24 tests) / `cargo test -p workspace`
> (14 tests) all pass.
Verified against the actual pinned GPUI revision Verified against the actual pinned GPUI revision
(`~/.cargo/git/checkouts/zed-a70e2ad075855582/1870e26/crates/scheduler/src/executor.rs:375-573` (`~/.cargo/git/checkouts/zed-a70e2ad075855582/1870e26/crates/scheduler/src/executor.rs:375-573`
and `crates/gpui/src/executor.rs:32-63`). and `crates/gpui/src/executor.rs:32-63`).
@@ -952,6 +964,14 @@ entirely disjoint observers.
## 14. `crates/workspace` has the same task-list pattern as §6 — and there it's an actual bug ## 14. `crates/workspace` has the same task-list pattern as §6 — and there it's an actual bug
> **Status: done.** The `tasks` field and all 17 push sites were removed from
> `RepoDetailView`, `NewPullRequestView`, `CommitDiffView` and
> `PullRequestDetailView`, replaced with `.detach()`. `cargo check -p workspace`
> and `cargo test -p workspace` (14 tests) pass. Removing the field cost each
> `cx.spawn`/`cx.spawn_in` call site its type-inference anchor, so every
> remaining spawn site needed an explicit `let task: gpui::Task<Result<(), ...>> = ...`
> annotation — expect the same when doing §6's `signed_state` half.
§6 covers `signed_state`'s 6 stores, where the unpruned-`Vec<Task>` pattern §6 covers `signed_state`'s 6 stores, where the unpruned-`Vec<Task>` pattern
is a style/complexity concern with no observed failure, because is a style/complexity concern with no observed failure, because
`push_task` always pruned before pushing. `crates/workspace` has the exact `push_task` always pruned before pushing. `crates/workspace` has the exact
@@ -1219,14 +1239,24 @@ method.
## Action plan, in order of risk/reward ## Action plan, in order of risk/reward
1. **Delete the fetch/sync dedup cache** (§3). Pure removal, no behavior 1. ✅ **Delete the fetch/sync dedup cache** (§3). Pure removal, no behavior
change for the intended usage pattern (each call site already has, or change for the intended usage pattern (each call site already has, or
trivially gets, its own guard). Lowest risk, do first. trivially gets, its own guard). Lowest risk, do first.
2. **Remove the `tasks: Vec<Task<...>>` + `push_task` boilerplate**, in
Done: removed `recent_fetches`/`fetch_recently_started`/`fetch_fingerprint`/
`FETCH_DEDUP_WINDOW` and the now-unused `DefaultHasher`/`Hash`/`Hasher`/
`Instant` imports from `signed_state/src/backend.rs`. `connect_repo_relays`
and `sync_bootstrap` no longer fingerprint or gate on a cache; callers keep
their own guards (`RepoStore::repo_relays`, one-shot construction-time call
in `RepoListStore::subscribe_remote`). `cargo check --workspace` and
`cargo test -p signed_state` (24 tests) both pass unchanged.
2. ✅ **Remove the `tasks: Vec<Task<...>>` + `push_task` boilerplate**, in
both `signed_state` (§6) and `crates/workspace` (§14), in favor of both `signed_state` (§6) and `crates/workspace` (§14), in favor of
`.detach()`/`.detach_and_log_err(cx)`. Independent of every other change `.detach()`/`.detach_and_log_err(cx)`. Independent of every other change
here, touches 10 files, all mechanical — and fixes a real unbounded-growth here, touches 10 files, all mechanical — and fixes a real unbounded-growth
bug in `RepoDetailView`/`NewPullRequestView` along the way. bug in `RepoDetailView`/`NewPullRequestView` along the way.
Done: both halves are complete, see §6 and §14 for details.
3. **Fix the relay add/connect calls** (§8): drop `.as_str()`/`ToString` 3. **Fix the relay add/connect calls** (§8): drop `.as_str()`/`ToString`
round trips, replace `add_relay` + blanket `client.connect()`/ round trips, replace `add_relay` + blanket `client.connect()`/
`connect_relay` pairs with `add_relay(url).and_connect()`, and delete `connect_relay` pairs with `add_relay(url).and_connect()`, and delete