From 4a57a01f4a9e49d05b9a733e49271123680e1df9 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Tue, 13 Oct 2026 09:05:51 +0700 Subject: [PATCH 1/6] fix sync --- AGENTS.md | 1 + crates/signed_git/src/repo.rs | 2 + crates/signed_git/src/tests.rs | 58 +++ crates/signed_state/src/backend.rs | 92 ++-- crates/signed_state/src/push.rs | 563 +++++++++---------------- crates/signed_state/src/repo.rs | 36 +- crates/workspace/src/views/repo/mod.rs | 107 ++--- 7 files changed, 394 insertions(+), 465 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 08edbba..bddfca0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,6 +5,7 @@ * Keep doc comments and comments simple, one line, straightforward, and unbloated. * Prefer implementing functionality in existing files unless it is a new logical component. Avoid creating many small files. * Prefer `impl` blocks over standalone functions: place functionality as methods or associated functions on the relevant type, extending the type's existing `impl` block instead of creating a duplicate one in another file. +* Do not wrap clients or SDK dependencies in project structs or pass-through helper functions, call their methods directly where needed. Extract a helper only when it holds real logic that is used in more than one place. * Bind a global store handle to a local before using it (`let state = GlobalState::global(cx);` then `state.update(...)`) instead of chaining on `CheckoutsStore::global(cx)` inline. * Avoid using functions that panic like `unwrap()`, instead use mechanisms like `?` to propagate errors. * Be careful with operations like indexing which may panic if the indexes are out of bounds. diff --git a/crates/signed_git/src/repo.rs b/crates/signed_git/src/repo.rs index 780ac3b..2240653 100644 --- a/crates/signed_git/src/repo.rs +++ b/crates/signed_git/src/repo.rs @@ -1192,11 +1192,13 @@ impl Repo { ], ..Default::default() }; + self.inner .find_remote("origin")? .connect(gix::remote::Direction::Fetch)? .prepare_fetch(Discard, options)? .receive(Discard, &IS_INTERRUPTED)?; + Ok(()) } diff --git a/crates/signed_git/src/tests.rs b/crates/signed_git/src/tests.rs index 62a31cf..59dca08 100644 --- a/crates/signed_git/src/tests.rs +++ b/crates/signed_git/src/tests.rs @@ -55,6 +55,64 @@ fn push_all_mirrors_branches_and_tags() { assert!(refs.contains("refs/tags/v1.0")); } +#[test] +fn ref_state_matches_the_git_cli_ref_listing() { + let (dir, repo) = fixture(&[("a.txt", b"one")]); + commit_all(&repo, "initial"); + let dir = dir.path(); + + git_run(dir, &["branch", "browse-group"]); + git_run(dir, &["checkout", "-q", "-b", "new-search"]); + std::fs::write(dir.join("b.txt"), b"two").expect("write"); + commit_all(&repo, "search work"); + git_run(dir, &["checkout", "-q", "-"]); + git_run(dir, &["tag", "v1"]); + + let listing = |repo: &Repo| { + let mut refs = repo.ref_state().expect("ref state").refs; + refs.sort(); + refs + }; + + // Loose refs. + let loose = listing(&Repo::open(dir).expect("open")); + + // Packed refs. + git_run(dir, &["pack-refs", "--all"]); + let packed = listing(&Repo::open(dir).expect("open")); + + let output = git_run( + dir, + &[ + "for-each-ref", + "--format=%(refname) %(objectname)", + "refs/heads", + "refs/tags", + ], + ); + let mut cli = String::from_utf8_lossy(&output.stdout) + .lines() + .map(|line| { + let (name, oid) = line.split_once(' ').expect("for-each-ref format"); + (name.to_owned(), oid.to_owned()) + }) + .collect::>(); + cli.sort(); + + assert_eq!(loose, cli); + assert_eq!(packed, cli); + assert!( + loose + .iter() + .any(|(name, _)| name == "refs/heads/new-search") + ); + assert!( + loose + .iter() + .any(|(name, _)| name == "refs/heads/browse-group") + ); +} + #[test] fn remote_has_refs_reports_whether_pushed_refs_landed() { let server = tempfile::tempdir().unwrap(); diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index e2aca0a..a4ea705 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -21,7 +21,9 @@ use crate::bootstrap::{ }; use crate::git_store::Mirrors; use crate::inbox::Inbox; -use crate::push::{GraspPush, PushOutcome, grasp_base_url, grasp_clone_url}; +use crate::push::{ + PushOutcome, grasp_base_url, grasp_clone_url, push_staged_to_grasps, require_relay_accepted, +}; use crate::repos::RepoListStore; pub const USER_KEYRING: &str = "Signed Safe Storage"; @@ -452,16 +454,19 @@ impl Backend { .map(|url| RelayUrl::parse(url).expect("valid relay URL")) .collect(); - let pusher = GraspPush::new(this.client.clone(), this.signer.clone()); - for builder in [ RelayList::new(relays).into_event_builder(), metadata, GitUserGraspList { grasp_servers }.into_event_builder(), ] { - let pusher = pusher.clone(); + let client = this.client.clone(); + let signer = this.signer.clone(); cx.background_spawn(async move { - pusher.publish_best_effort(builder).await; + let event = builder.finalize_async(&signer).await?; + let output = client.send_event(&event).broadcast().await?; + require_relay_accepted(output, event)?; + + Ok::<(), anyhow::Error>(()) }) .detach(); } @@ -792,7 +797,8 @@ impl Backend { PushOutcome::default() } else { let push = cx.background_spawn({ - let pusher = GraspPush::new(client.clone(), signer.clone()); + let client = client.clone(); + let signer = signer.clone(); let path = path.clone(); let owner = owner.clone(); let repo_id = repo_id.clone(); @@ -801,20 +807,21 @@ impl Backend { let head = head.clone(); let executor = cx.background_executor().clone(); async move { - pusher - .push_staged_to_grasps( - &repo_id, - &refs, - head.as_deref(), - &path, - &owner, - &relays, - &executor, - |path, base, owner, repo_id| { - Repo::open(path)?.push_all(base, owner, repo_id) - }, - ) - .await + push_staged_to_grasps( + &client, + &signer, + &repo_id, + &refs, + head.as_deref(), + &path, + &owner, + &relays, + &executor, + |path, base, owner, repo_id| { + Repo::open(path)?.push_all(base, owner, repo_id) + }, + ) + .await } }); push.await @@ -1136,15 +1143,16 @@ impl Backend { // Each target gets its own deletion event. fn retract_events(&mut self, events: &[Event], cx: &mut Context) { - let pusher = GraspPush::new(self.client.clone(), self.signer.clone()); - for event in events.iter().cloned() { - let pusher = pusher.clone(); + let client = self.client.clone(); + let signer = self.signer.clone(); - cx.spawn(async move |_this, _cx| { - if let Err(e) = pusher.retract_event(&event).await { - log::warn!("failed to retract event {}: {e}", event.id); - } + cx.background_spawn(async move { + let builder = EventDeletionRequest::new().id(event.id); + let deletion = builder.finalize_async(&signer).await?; + client.send_event(&deletion).broadcast().await?; + + Ok::<(), anyhow::Error>(()) }) .detach(); } @@ -1198,23 +1206,26 @@ async fn announce_repository_and_push( let builder = announcement.into_event_builder(); let event = builder.finalize_async(signer).await?; let output = client.send_event(&event).broadcast().await?; - GraspPush::require_relay_accepted(output, event)? + require_relay_accepted(output, event)? }; // The state event is the push authorization. let outcome = if refs.is_empty() { PushOutcome::default() } else { - let pusher = GraspPush::new(client.clone(), signer.clone()); - let repo_id = repo_id.to_owned(); - let owner = owner.to_owned(); - let servers = servers.to_vec(); - let head = head.clone(); - let path = path.to_path_buf(); - let executor = cx.background_executor().clone(); - cx.background_spawn(async move { - pusher - .push_staged_to_grasps( + let push = { + let client = client.clone(); + let signer = signer.clone(); + let repo_id = repo_id.to_owned(); + let owner = owner.to_owned(); + let servers = servers.to_vec(); + let head = head.clone(); + let path = path.to_path_buf(); + let executor = cx.background_executor().clone(); + cx.background_spawn(async move { + push_staged_to_grasps( + &client, + &signer, &repo_id, &refs, head.as_deref(), @@ -1225,8 +1236,9 @@ async fn announce_repository_and_push( push, ) .await - }) - .await + }) + }; + push.await }; if outcome.accepted() == 0 { diff --git a/crates/signed_state/src/push.rs b/crates/signed_state/src/push.rs index bc18945..4773e7a 100644 --- a/crates/signed_state/src/push.rs +++ b/crates/signed_state/src/push.rs @@ -3,14 +3,13 @@ use std::time::Duration; use anyhow::{Error, bail}; use gpui::BackgroundExecutor; -use nostr::event::IntoEventBuilder; use nostr::prelude::Url; use nostr_sdk::prelude::*; use signed_core::RepoState; -use signed_git::Repo; +use signed_git::{Repo, RepoRefState}; use signed_nostr::UniversalSigner; -pub(crate) const GRASP_PUSH_ATTEMPTS: usize = 3; +const GRASP_PUSH_ATTEMPTS: usize = 3; const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1); @@ -18,8 +17,10 @@ const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1); pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option { let parsed = Url::parse(relay.as_str()).ok()?; let host = parsed.host_str()?; - let port = parsed.port().map(|p| format!(":{p}")).unwrap_or_default(); - // `ws://` grasp servers, e.g. local dev relays, speak plain HTTP. + let port = parsed + .port() + .map(|port| format!(":{port}")) + .unwrap_or_default(); let scheme = if relay.scheme().is_secure() { "https" } else { @@ -110,7 +111,7 @@ impl PushOutcome { .join("; ") } - // `None` when every server accepted the push or nothing was pushed. + /// `None` when every server accepted the push or nothing was pushed. pub fn partial_warning(&self) -> Option { let accepted = self.accepted(); if self.servers.is_empty() || accepted == self.servers.len() { @@ -122,25 +123,119 @@ impl PushOutcome { self.failure_summary() )) } + + /// All staged events carry the same refs; the newest timestamp wins on the relays. + fn keep_newest(&mut self, event: Event) { + if self + .state_event + .as_ref() + .is_none_or(|current| event.created_at > current.created_at) + { + self.state_event = Some(event); + } + } } -// Reasons a push attempt should be retried with a freshly staged state event -// and a fresh git advertisement. Two families are retried: -// -// - **Purgatory denials**: the grasp server sends these when the state event -// for the push has not reached its purgatory yet. Re-staging a fresh event -// resolves them. -// - **Stale advertisement races**: `git receive-pack` compares each ref -// update against the value it advertised when the push started. The grasp -// server's own background sync can move a ref in between - typically by -// aligning the repository to a parked state event once the objects of an -// earlier attempt land - so the compare-and-swap fails with `cannot lock -// ref` / `incorrect old value provided`. A retry against the fresh -// advertisement converges, and when the race is lost the pushed data is -// usually already on the server (see `is_stale_advertisement_race` and -// the convergence probe in `push_staged_to_grasps`). -// -// Other rejections are not retried. +pub(crate) fn require_relay_accepted( + output: SendEventOutput, + event: Event, +) -> Result { + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + + bail!("event not accepted by any relay: {reasons}"); + } + + Ok(event) +} + +/// Signs with a `created_at` strictly newer than `last_created_at`, +/// since a relay treats a same-id resend as a duplicate. +async fn sign_state_event( + signer: &UniversalSigner, + repo_id: &str, + refs: &[(String, String)], + head: Option<&str>, + last_created_at: u64, +) -> Result<(Event, u64), String> { + let now = Timestamp::now().as_secs(); + let created_at = if now > last_created_at { + now + } else { + last_created_at + 1 + }; + + let event = RepoState::build(repo_id, refs, head) + .custom_created_at(Timestamp::from_secs(created_at)) + .finalize_async(signer) + .await + .map_err(|e| format!("could not sign the state event: {e}"))?; + + Ok((event, created_at)) +} + +/// On a grasp relay an accepted event is parked in purgatory, which authorizes the paired git push. +async fn stage_event_on_relay( + client: &Client, + relay: &RelayUrl, + event: &Event, +) -> Result<(), String> { + client + .add_relay(relay) + .and_connect() + .await + .map_err(|e| format!("could not add relay {relay}: {e}"))?; + + let output = client + .send_event(event) + .to([relay.clone()]) + .await + .map_err(|e| format!("could not send the state event to {relay}: {e}"))?; + + if output.success.contains_key(relay) { + Ok(()) + } else { + let reason = output + .failed + .get(relay) + .cloned() + .unwrap_or_else(|| "relay did not confirm the event".to_owned()); + Err(reason) + } +} + +/// Refresh the repo's ref state, keeping the current branch as the head if it is still present. +fn refresh_ref_state(path: &Path, head: Option<&str>) -> Option { + let state = Repo::open(path).and_then(|repo| repo.ref_state()).ok()?; + + if state.refs.is_empty() { + return None; + } + + let head = match head { + Some(current) => { + let name = format!("refs/heads/{current}"); + if state.refs.iter().any(|(known, _)| *known == name) { + Some(current.to_owned()) + } else { + state.head + } + } + None => state.head, + }; + + Some(RepoRefState { + refs: state.refs, + head, + }) +} + +/// Denials that staging a fresh state event and pushing against the refreshed advertisement resolve. fn is_transient_grasp_denial(stderr: &str) -> bool { let error = stderr.to_lowercase(); [ @@ -156,180 +251,55 @@ fn is_transient_grasp_denial(stderr: &str) -> bool { .any(|marker| error.contains(marker)) } -// A push rejected because `git receive-pack`'s compare-and-swap lost to the -// grasp server's own background ref alignment: the ref moved between this -// push's advertisement and its ref transaction (`cannot lock ref ... is at -// ... but expected ...` / `incorrect old value provided`). The pushed data -// is usually already on the server by then. +/// The server's own ref alignment moved a ref mid-push; the pushed data is usually already there. fn is_stale_advertisement_race(stderr: &str) -> bool { let error = stderr.to_lowercase(); error.contains("cannot lock ref") || error.contains("incorrect old value provided") } -// All staged events carry the same refs; the newest timestamp wins on the relays. -fn keep_newest(state_event: &mut Option, event: Event) { - if state_event - .as_ref() - .is_none_or(|current| event.created_at > current.created_at) - { - *state_event = Some(event); - } -} +#[allow(clippy::too_many_arguments)] +pub(crate) async fn push_staged_to_grasps( + client: &Client, + signer: &UniversalSigner, + repo_id: &str, + refs: &[(String, String)], + head: Option<&str>, + path: &Path, + owner: &str, + servers: &[RelayUrl], + executor: &BackgroundExecutor, + push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>, +) -> PushOutcome { + let mut outcome = PushOutcome::default(); + let mut refs = refs.to_vec(); + let mut head = head.map(str::to_owned); -#[derive(Clone)] -pub(crate) struct GraspPush { - client: Client, - signer: UniversalSigner, -} - -impl GraspPush { - pub(crate) fn new(client: Client, signer: UniversalSigner) -> Self { - Self { client, signer } - } - - pub(crate) fn require_relay_accepted( - output: SendEventOutput, - event: Event, - ) -> Result { - if output.success.is_empty() && !output.failed.is_empty() { - let reasons = output - .failed - .values() - .cloned() - .collect::>() - .join(", "); - bail!("event not accepted by any relay: {reasons}"); - } - - Ok(event) - } - - // A relay hiccup should not block sign-up, so failures are only logged. - pub(crate) async fn publish_best_effort(&self, builder: EventBuilder) { - let result: Result<(), Error> = async { - let event = builder.finalize_async(&self.signer).await?; - let output = self.client.send_event(&event).broadcast().await?; - Self::require_relay_accepted(output, event)?; - Ok(()) - } - .await; - - if let Err(e) = result { - log::warn!("failed to publish identity bootstrap event: {e}"); - } - } - - pub(crate) async fn publish_one(&self, builder: EventBuilder) -> Result { - let event = builder.finalize_async(&self.signer).await?; - self.send_accepted(event).await - } - - pub(crate) async fn send_accepted(&self, event: Event) -> Result { - let output = self.client.send_event(&event).broadcast().await?; - Self::require_relay_accepted(output, event) - } - - pub(crate) async fn retract_event(&self, event: &Event) -> Result<(), Error> { - let builder = EventDeletionRequest::new() - .id(event.id) - .into_event_builder(); - - let deletion = builder.finalize_async(&self.signer).await?; - self.client.send_event(&deletion).broadcast().await?; - - Ok(()) - } - - // Retries within the same second get the next second: a grasp relay - // treats a same-id resend as a duplicate and does not re-run its ingest, - // so an identical resend cannot re-park a state event lost from its purgatory. - async fn sign_state_event( - &self, - repo_id: &str, - refs: &[(String, String)], - head: Option<&str>, - last_created_at: u64, - ) -> Result<(Event, u64), String> { - let now = Timestamp::now().as_secs(); - let created_at = if now > last_created_at { - now - } else { - last_created_at + 1 + for relay in servers { + let Some(base) = grasp_base_url(relay) else { + outcome + .servers + .push(GraspServer::failed(relay.clone(), "no domain")); + continue; }; - let event = RepoState::build(repo_id, refs, head) - .custom_created_at(Timestamp::from_secs(created_at)) - .finalize_async(&self.signer) - .await - .map_err(|e| format!("could not sign the state event: {e}"))?; + let git_url = format!("{base}/{owner}/{repo_id}.git"); - Ok((event, created_at)) - } + let mut reason = None; + let mut last_created_at = 0; + let mut staged_event = None; - // `Ok` only when the relay confirmed the event. On a grasp relay the - // accept parks the event in purgatory, which authorizes the paired git push. - async fn stage_event_on_relay(&self, relay: &RelayUrl, event: &Event) -> Result<(), String> { - self.client - .add_relay(relay) - .and_connect() - .await - .map_err(|e| format!("could not add relay {relay}: {e}"))?; - - let output = self - .client - .send_event(event) - .to([relay.clone()]) - .await - .map_err(|e| format!("could not send the state event to {relay}: {e}"))?; - - if output.success.contains_key(relay) { - Ok(()) - } else { - let reason = output - .failed - .get(relay) - .cloned() - .unwrap_or_else(|| "relay did not confirm the event".to_owned()); - Err(reason) - } - } - - #[allow(clippy::too_many_arguments)] - pub(crate) async fn push_staged_to_grasps( - &self, - repo_id: &str, - refs: &[(String, String)], - head: Option<&str>, - path: &Path, - owner: &str, - servers: &[RelayUrl], - executor: &BackgroundExecutor, - push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>, - ) -> PushOutcome { - let mut outcome = PushOutcome::default(); - - for relay in servers { - let Some(base) = grasp_base_url(relay) else { - outcome - .servers - .push(GraspServer::failed(relay.clone(), "no domain")); - continue; - }; - let git_url = format!("{base}/{owner}/{repo_id}.git"); - - let mut reason = None; - let mut last_created_at = 0; - // For the convergence probe when every push attempt lost the - // stale-ref race. - let mut staged_event = None; - - 'server: for attempt in 1..=GRASP_PUSH_ATTEMPTS { - if attempt > 1 { - executor.timer(GRASP_RETRY_DELAY).await; + 'server: for attempt in 1..=GRASP_PUSH_ATTEMPTS { + if attempt > 1 { + executor.timer(GRASP_RETRY_DELAY).await; + // The repo's refs may have moved since the last attempt, so refresh the state. + if let Some(state) = refresh_ref_state(path, head.as_deref()) { + refs = state.refs; + head = state.head; } + } - let (event, created_at) = match self - .sign_state_event(repo_id, refs, head, last_created_at) + let (event, created_at) = + match sign_state_event(signer, repo_id, &refs, head.as_deref(), last_created_at) .await { Ok(signed) => signed, @@ -339,190 +309,61 @@ impl GraspPush { } }; - last_created_at = created_at; + last_created_at = created_at; - // A failed stage means the grasp never parked the state, so - // the git push would be denied anyway: skip it. - if let Err(e) = self.stage_event_on_relay(relay, &event).await { - // One retry absorbs a relay connect blip, on the first - // attempt only. - if attempt == 1 && self.stage_event_on_relay(relay, &event).await.is_ok() { - } else { - reason = Some(e); - break 'server; - } - } - staged_event = Some(event.clone()); - - match push(path, &base, owner, repo_id) { - Ok(()) => { - keep_newest(&mut outcome.state_event, event); - break 'server; - } - Err(e) => { - let text = e.to_string(); - if attempt < GRASP_PUSH_ATTEMPTS && is_transient_grasp_denial(&text) { - reason = Some(text); - continue 'server; - } - reason = Some(text); - break 'server; - } + // Stage the event on the relay, + // so that the push can be accepted if the relay's state is aligned. + if let Err(e) = stage_event_on_relay(client, relay, &event).await { + let retried = + attempt == 1 && stage_event_on_relay(client, relay, &event).await.is_ok(); + if !retried { + reason = Some(e); + break 'server; } } + staged_event = Some(event.clone()); - // The grasp's own background sync aligns refs to staged state - // events as soon as the objects land, which can beat every push - // attempt's compare-and-swap. When the last denial was that race - // the sync has usually finished by now: verify the advertised refs - // and accept the server when the pushed data is already there. - if let Some(last_reason) = &reason - && is_stale_advertisement_race(last_reason) - && Repo::open(path) - .and_then(|repo| repo.remote_has_refs(&git_url, refs)) - .unwrap_or(false) - { - if let Some(event) = staged_event { - keep_newest(&mut outcome.state_event, event); + match push(path, &base, owner, repo_id) { + Ok(()) => { + outcome.keep_newest(event); + break 'server; } - reason = None; - } - - match reason { - Some(reason) => { - log::warn!("grasp push failed: {relay}: {reason}"); - outcome - .servers - .push(GraspServer::failed(relay.clone(), reason)); + Err(e) => { + let text = e.to_string(); + let retriable = + attempt < GRASP_PUSH_ATTEMPTS && is_transient_grasp_denial(&text); + reason = Some(text); + if !retriable { + break 'server; + } } - None => outcome.servers.push(GraspServer::ok(relay.clone())), } } - outcome - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn grasp_base_url_maps_schemes_like_ngit() { - let wss = RelayUrl::parse("wss://relay.ngit.dev").expect("url"); - assert_eq!( - grasp_base_url(&wss).as_deref(), - Some("https://relay.ngit.dev") - ); - - let ws = RelayUrl::parse("ws://localhost:8080").expect("url"); - assert_eq!( - grasp_base_url(&ws).as_deref(), - Some("http://localhost:8080") - ); - } - - #[test] - fn grasp_clone_url_matches_ngit_format() { - let relay = RelayUrl::parse("wss://gitnostr.com").expect("url"); - let url = grasp_clone_url(&relay, "npub1test", "my-repo").expect("url"); - assert_eq!( - url.to_string(), - "https://gitnostr.com/npub1test/my-repo.git" - ); - } - - #[test] - fn grasp06_prs_url_matches_ngit_format() { - assert_eq!( - grasp06_prs_url("https://relay.ngit.dev", "npub1author", "my-repo"), - "https://relay.ngit.dev/prs/npub1author/my-repo.git" - ); - assert_eq!( - grasp06_prs_url("http://localhost:8080", "npub1author", "my-repo"), - "http://localhost:8080/prs/npub1author/my-repo.git" - ); - } - - #[test] - fn pr_clone_urls_orders_author_first_and_deduplicates() { - let prs = vec![ - Url::parse("https://a.example/prs/npub1me/repo.git").expect("url"), - Url::parse("https://a.example/prs/npub1me/repo.git").expect("url"), - ]; - let base = vec![ - Url::parse("https://a.example/npub1owner/repo.git").expect("url"), - Url::parse("https://b.example/npub1owner/repo.git").expect("url"), - Url::parse("https://b.example/npub1owner/repo.git").expect("url"), - ]; - - let urls = pr_clone_urls(prs, base); - assert_eq!( - urls.iter().map(ToString::to_string).collect::>(), - vec![ - "https://a.example/prs/npub1me/repo.git", - "https://a.example/npub1owner/repo.git", - "https://b.example/npub1owner/repo.git", - ] - ); - } - - #[test] - fn transient_grasp_denials_are_classified() { - let reported = "remote: ERR authorisation failed: No state events in purgatory\n\ - fatal: the remote end hung up unexpectedly\n\ - error: failed to push some refs to 'https://relay.ngit.dev/...git'"; - assert!(is_transient_grasp_denial(reported)); - - assert!(is_transient_grasp_denial( - "remote: ERR authorisation failed: No matching state event found in purgatory" - )); - assert!(is_transient_grasp_denial( - "remote: ERR authorisation failed: 1 state event in purgatory from authorized \ - publisher but doesn't match push" - )); - assert!(is_transient_grasp_denial( - "remote: ERR authorisation failed: 2 state events in purgatory but none from \ - authorized publishers" - )); - assert!(is_transient_grasp_denial( - "remote: ERR authorisation failed: No repository announcement found" - )); - - assert!(!is_transient_grasp_denial( - "remote: ERR authorisation failed: not a maintainer of this repository" - )); - assert!(!is_transient_grasp_denial( - "fatal: unable to access 'https://relay.ngit.dev/...': The requested URL returned \ - error: 403" - )); - assert!(!is_transient_grasp_denial( - "fatal: unable to access 'https://relay.ngit.dev/...': Could not resolve host" - )); - } - - #[test] - fn stale_ref_races_are_retried() { - let reported = "remote: error: cannot lock ref 'refs/heads/main': is at \ - cac2ac91b6f5fb8dfcb6962785babc6e65350cb3 but expected \ - bc5e892aa84dc6240a5fbcd59367a4857d26f49b\n\ - To https://relay.ngit.dev/npub1owner/signed-test.git\n\ - ! [remote rejected] main -> main (incorrect old value provided)\n\ - error: failed to push some refs to 'https://relay.ngit.dev/npub1owner/signed-test.git'"; - assert!(is_transient_grasp_denial(reported)); - assert!(is_stale_advertisement_race(reported)); - - assert!(is_stale_advertisement_race( - "cannot lock ref 'refs/heads/main'" - )); - assert!(is_stale_advertisement_race( - "! [remote rejected] main -> main (incorrect old value provided)" - )); - - assert!(!is_stale_advertisement_race("No state events in purgatory")); - - assert!(!is_transient_grasp_denial( - " ! [rejected] main -> main (non-fast-forward)" - )); + // If the relay's denial was a stale-ref race, + // check if the repo has moved on and retry the push. + if let Some(last_reason) = &reason + && is_stale_advertisement_race(last_reason) + && Repo::open(path) + .and_then(|repo| repo.remote_has_refs(&git_url, &refs)) + .unwrap_or(false) + { + if let Some(event) = staged_event { + outcome.keep_newest(event); + } + reason = None; + } + + match reason { + Some(reason) => { + log::warn!("grasp push failed: {relay}: {reason}"); + outcome + .servers + .push(GraspServer::failed(relay.clone(), reason)); + } + None => outcome.servers.push(GraspServer::ok(relay.clone())), + } } + + outcome } diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index 1ea0659..2059eb5 100644 --- a/crates/signed_state/src/repo.rs +++ b/crates/signed_state/src/repo.rs @@ -18,7 +18,9 @@ use signed_nostr::UniversalSigner; use crate::backend::{Backend, BackendEvent}; use crate::bootstrap::user_grasp_list_servers; use crate::checkouts::CheckoutsStore; -use crate::push::{GraspPush, PushOutcome, grasp_base_url, grasp06_prs_url, pr_clone_urls}; +use crate::push::{ + PushOutcome, grasp_base_url, grasp06_prs_url, pr_clone_urls, require_relay_accepted, +}; use crate::repos::RepoListStore; use crate::sync_status::SyncStatusStore; @@ -865,10 +867,11 @@ impl RepoStore { } } - let publish_result = { - let pusher = GraspPush::new(client.clone(), signer.clone()); - pusher.send_accepted(event).await - }; + let publish_result: Result = async { + let output = client.send_event(&event).broadcast().await?; + require_relay_accepted(output, event) + } + .await; let pr_event = match publish_result { Ok(event) => event, @@ -1041,10 +1044,12 @@ impl RepoStore { } }; - let publish_result = { - let pusher = GraspPush::new(client.clone(), signer.clone()); - pusher.publish_one(builder).await - }; + let publish_result: Result = async { + let event = builder.finalize_async(&signer).await?; + let output = client.send_event(&event).broadcast().await?; + require_relay_accepted(output, event) + } + .await; if let Err(e) = publish_result { return this.update(cx, |this, cx| { @@ -1309,8 +1314,12 @@ impl RepoStore { }; let task: Task> = cx.spawn(async move |this, cx| { - let pusher = GraspPush::new(client, signer); - let publish_result = pusher.publish_one(builder).await; + let publish_result: Result = async { + let event = builder.finalize_async(&signer).await?; + let output = client.send_event(&event).broadcast().await?; + require_relay_accepted(output, event) + } + .await; if let Err(e) = publish_result { this.update(cx, |this, cx| { @@ -1421,7 +1430,6 @@ async fn publish_patch_series( first_marker: &str, reply_to: Option, ) -> Result { - let pusher = GraspPush::new(client.clone(), signer.clone()); let mut root: Option = None; let mut previous = reply_to; @@ -1468,7 +1476,9 @@ async fn publish_patch_series( } let builder = EventBuilder::new(Kind::GitPatch, part.clone()).tags(tags); - let event = pusher.publish_one(builder).await?; + let event = builder.finalize_async(signer).await?; + let output = client.send_event(&event).broadcast().await?; + let event = require_relay_accepted(output, event)?; if root.is_none() { root = Some(event.clone()); diff --git a/crates/workspace/src/views/repo/mod.rs b/crates/workspace/src/views/repo/mod.rs index 14c39c4..c750ed5 100644 --- a/crates/workspace/src/views/repo/mod.rs +++ b/crates/workspace/src/views/repo/mod.rs @@ -321,9 +321,6 @@ impl RepoDetailView { (store.announcement.clone(), store.path.clone()) }; - // A repository with a local worktree shows it directly. An announced one - // still loads its announcement and activity from the store, which is - // subscribed to the relays independently. if let Some(local_path) = local_path { self.repo_started = true; @@ -358,6 +355,7 @@ impl RepoDetailView { self.repo_started = true; + let store = self.store.downgrade(); let addr = announcement.addr(); let clone_urls: Vec = announcement.clone.clone(); @@ -399,64 +397,71 @@ impl RepoDetailView { })?; // Refresh the clone from the network in the background. - // When it completes, update the refs and commit list. if !had_clone { return Ok(()); } - let refresh = { - let addr = addr.clone(); + // Whether a push is in progress + let pushing = store.read_with(cx, |this, _| this.pushing)?; - cx.background_spawn(async move { - let Some(repo) = Mirrors::open(&addr)? else { - return Ok::<_, Error>(None); - }; + if !pushing { + let refresh = { + let addr = addr.clone(); - // Best-effort, a fetch failure, e.g. offline, keeps the cached state. - repo.fetch().ok(); + cx.background_spawn(async move { + let Some(repo) = Mirrors::open(&addr)? else { + return Ok::<_, Error>(None); + }; - let moved = repo.fast_forward_branches().unwrap_or(false); + // Best-effort, a fetch failure, e.g. offline, keeps the cached state. + repo.fetch().ok(); - Ok::<_, Error>(Some((moved, repo.snapshot()?))) - }) - } - .await; + let moved = repo.fast_forward_branches().unwrap_or(false); - this.update_in(cx, |this, window, cx| { - if let Ok(Some((moved, snapshot))) = refresh { - let branches: Vec = - snapshot.branches.iter().map(Into::into).collect(); - let tags: Vec = snapshot.tags.iter().map(Into::into).collect(); - let current_branch = snapshot.current_branch.clone(); - let head_commit = snapshot.head_commit.clone(); - - let branches_changed = this.refs.set_branches( - branches, - current_branch.map(Into::into), - window, - cx, - ); - - let tags_changed = this.refs.set_tags(tags, window, cx); - let new_head_commit = head_commit.as_ref().map(|c| &c.id); - let current_head_commit = this.head_commit.as_ref().map(|c| &c.id); - - let head_changed = new_head_commit != current_head_commit; - this.head_commit = head_commit; - - if head_changed { - this.history.update(cx, |history, cx| history.reload(cx)); - } - - if moved { - this.catch_up_worktree(cx); - } - - if branches_changed || tags_changed || head_changed { - cx.notify(); - } + Ok::<_, Error>(Some((moved, repo.snapshot()?))) + }) } - })?; + .await; + + this.update_in(cx, |this, window, cx| { + if let Ok(Some((moved, snapshot))) = refresh { + let branches: Vec = + snapshot.branches.iter().map(Into::into).collect(); + + let tags: Vec = + snapshot.tags.iter().map(Into::into).collect(); + + let current_branch = snapshot.current_branch.clone(); + let head_commit = snapshot.head_commit.clone(); + + let branches_changed = this.refs.set_branches( + branches, + current_branch.map(Into::into), + window, + cx, + ); + + let tags_changed = this.refs.set_tags(tags, window, cx); + let new_head_commit = head_commit.as_ref().map(|c| &c.id); + let current_head_commit = this.head_commit.as_ref().map(|c| &c.id); + + let head_changed = new_head_commit != current_head_commit; + this.head_commit = head_commit; + + if head_changed { + this.history.update(cx, |history, cx| history.reload(cx)); + } + + if moved { + this.catch_up_worktree(cx); + } + + if branches_changed || tags_changed || head_changed { + cx.notify(); + } + } + })?; + } Ok(()) }); -- 2.54.0 From b4c35654843eb2ac95fbed1e1be0d99e4ba5aff3 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Tue, 13 Oct 2026 09:20:08 +0700 Subject: [PATCH 2/6] Extract PatchSeries module and remove require_relay_accepted helper --- crates/signed_core/src/deletions.rs | 2 - crates/signed_state/src/backend.rs | 33 ++-- crates/signed_state/src/lib.rs | 1 + crates/signed_state/src/patch_series.rs | 50 ++++++ crates/signed_state/src/push.rs | 21 +-- crates/signed_state/src/repo.rs | 214 ++++++------------------ crates/signed_state/src/sync_status.rs | 4 +- 7 files changed, 129 insertions(+), 196 deletions(-) create mode 100644 crates/signed_state/src/patch_series.rs diff --git a/crates/signed_core/src/deletions.rs b/crates/signed_core/src/deletions.rs index 354077a..cc6b407 100644 --- a/crates/signed_core/src/deletions.rs +++ b/crates/signed_core/src/deletions.rs @@ -28,8 +28,6 @@ impl Deletions { .map(|c| (c, event.pubkey, event.created_at)), ); } else if event.kind == Kind::RequestToVanish { - // We can't verify which relay the request targeted, so any - // vanish request is honored for the author's events. vanished.push((event.pubkey, event.created_at)); } } diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index a4ea705..32e2aed 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -21,9 +21,7 @@ use crate::bootstrap::{ }; use crate::git_store::Mirrors; use crate::inbox::Inbox; -use crate::push::{ - PushOutcome, grasp_base_url, grasp_clone_url, push_staged_to_grasps, require_relay_accepted, -}; +use crate::push::{PushOutcome, grasp_base_url, grasp_clone_url, push_staged_to_grasps}; use crate::repos::RepoListStore; pub const USER_KEYRING: &str = "Signed Safe Storage"; @@ -464,7 +462,16 @@ impl Backend { cx.background_spawn(async move { let event = builder.finalize_async(&signer).await?; let output = client.send_event(&event).broadcast().await?; - require_relay_accepted(output, event)?; + + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + bail!("event not accepted by any relay: {reasons}"); + } Ok::<(), anyhow::Error>(()) }) @@ -1202,12 +1209,18 @@ async fn announce_repository_and_push( client.add_relay(url).and_connect().await.ok(); } - let event = { - let builder = announcement.into_event_builder(); - let event = builder.finalize_async(signer).await?; - let output = client.send_event(&event).broadcast().await?; - require_relay_accepted(output, event)? - }; + let event = announcement.finalize_async(signer).await?; + let output = client.send_event(&event).broadcast().await?; + + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + bail!("event not accepted by any relay: {reasons}"); + } // The state event is the push authorization. let outcome = if refs.is_empty() { diff --git a/crates/signed_state/src/lib.rs b/crates/signed_state/src/lib.rs index 5ca064c..bcbd7e1 100644 --- a/crates/signed_state/src/lib.rs +++ b/crates/signed_state/src/lib.rs @@ -4,6 +4,7 @@ mod checkouts; mod git_store; mod inbox; mod local_repos; +mod patch_series; mod profile; mod push; mod refresh; diff --git a/crates/signed_state/src/patch_series.rs b/crates/signed_state/src/patch_series.rs new file mode 100644 index 0000000..b32c3d4 --- /dev/null +++ b/crates/signed_state/src/patch_series.rs @@ -0,0 +1,50 @@ +use bitcoin_hashes::sha1::Hash as Sha1Hash; +use signed_git::PatchParser; + +/// NIP-34 suggests patches when each event is under 60kb. +pub(crate) const MAX_PATCH_EVENT_BYTES: usize = 60 * 1024; + +pub(crate) struct PatchSeries { + pub(crate) parts: Vec, +} + +impl PatchSeries { + pub(crate) fn parse(patch: &str) -> Self { + Self { + parts: PatchParser::split_patch_series(patch) + .into_iter() + .map(str::to_owned) + .collect(), + } + } + + /// Returns the length of the first part that exceeds the NIP-34 recommended size. + pub(crate) fn oversized_length(&self) -> Option { + self.parts + .iter() + .map(String::len) + .find(|length| *length > MAX_PATCH_EVENT_BYTES) + } + + /// The last commit of the series, which is the tip of the branch. + pub(crate) fn tip_commit(&self) -> Option { + self.parts + .last() + .and_then(|part| patch_current_commit(part)) + .and_then(|hex| hex.parse::().ok()) + } + + pub(crate) fn commit_of(&self, index: usize) -> Option<&str> { + self.parts + .get(index) + .and_then(|part| patch_current_commit(part)) + .filter(|hex| hex.len() == 40) + } +} + +/// The `From ` header on the first line. +fn patch_current_commit(patch: &str) -> Option<&str> { + let line = patch.lines().next()?; + let hex = line.strip_prefix("From ")?; + hex.split_whitespace().next().filter(|hex| hex.len() == 40) +} diff --git a/crates/signed_state/src/push.rs b/crates/signed_state/src/push.rs index 4773e7a..a5471a4 100644 --- a/crates/signed_state/src/push.rs +++ b/crates/signed_state/src/push.rs @@ -1,7 +1,7 @@ use std::path::Path; use std::time::Duration; -use anyhow::{Error, bail}; +use anyhow::Error; use gpui::BackgroundExecutor; use nostr::prelude::Url; use nostr_sdk::prelude::*; @@ -10,7 +10,6 @@ use signed_git::{Repo, RepoRefState}; use signed_nostr::UniversalSigner; const GRASP_PUSH_ATTEMPTS: usize = 3; - const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1); /// `ws://` grasp servers, like ngit, use `http(s)://`. @@ -136,24 +135,6 @@ impl PushOutcome { } } -pub(crate) fn require_relay_accepted( - output: SendEventOutput, - event: Event, -) -> Result { - if output.success.is_empty() && !output.failed.is_empty() { - let reasons = output - .failed - .values() - .cloned() - .collect::>() - .join(", "); - - bail!("event not accepted by any relay: {reasons}"); - } - - Ok(event) -} - /// Signs with a `created_at` strictly newer than `last_created_at`, /// since a relay treats a same-id resend as a duplicate. async fn sign_state_event( diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index 2059eb5..b482fb7 100644 --- a/crates/signed_state/src/repo.rs +++ b/crates/signed_state/src/repo.rs @@ -3,7 +3,6 @@ use std::collections::{HashMap, HashSet}; use std::path::PathBuf; use anyhow::{Error, bail}; -use bitcoin_hashes::sha1::Hash as Sha1Hash; use gpui::{App, AppContext, Context, SharedString, Subscription, Task}; use nostr::event::IntoEventBuilder; use nostr_sdk::prelude::*; @@ -12,21 +11,17 @@ use signed_core::{ Announcement, Deletions, Filters, GitEvent, PullRequest, RepoAddr, RepoState, RepoStatus, filters, }; -use signed_git::{Nip34Binding, PatchParser, Repo}; +use signed_git::{Nip34Binding, Repo}; use signed_nostr::UniversalSigner; use crate::backend::{Backend, BackendEvent}; use crate::bootstrap::user_grasp_list_servers; use crate::checkouts::CheckoutsStore; -use crate::push::{ - PushOutcome, grasp_base_url, grasp06_prs_url, pr_clone_urls, require_relay_accepted, -}; +use crate::patch_series::{MAX_PATCH_EVENT_BYTES, PatchSeries}; +use crate::push::{PushOutcome, grasp_base_url, grasp06_prs_url, pr_clone_urls}; use crate::repos::RepoListStore; use crate::sync_status::SyncStatusStore; -// NIP-34 suggests patches when each event is under 60kb. -const MAX_PATCH_EVENT_BYTES: usize = 60 * 1024; - pub struct RepoStore { addr: Option, // Seeded from the open-time hint, replaced by the database's latest on the @@ -869,7 +864,18 @@ impl RepoStore { let publish_result: Result = async { let output = client.send_event(&event).broadcast().await?; - require_relay_accepted(output, event) + + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + bail!("event not accepted by any relay: {reasons}"); + } + + Ok(event) } .await; @@ -1047,7 +1053,18 @@ impl RepoStore { let publish_result: Result = async { let event = builder.finalize_async(&signer).await?; let output = client.send_event(&event).broadcast().await?; - require_relay_accepted(output, event) + + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + bail!("event not accepted by any relay: {reasons}"); + } + + Ok(event) } .await; @@ -1302,8 +1319,7 @@ impl RepoStore { Task::ready(Err(anyhow::anyhow!("{message}"))) } - // Every one-shot repository event (issue, comment, status) goes through - // this; multi-step flows call the SDK directly instead. + /// Publish an event to the repository's relays, and update the store with any error. fn publish(&mut self, builder: EventBuilder, cx: &mut Context) { self.last_error = None; @@ -1317,7 +1333,18 @@ impl RepoStore { let publish_result: Result = async { let event = builder.finalize_async(&signer).await?; let output = client.send_event(&event).broadcast().await?; - require_relay_accepted(output, event) + + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + bail!("event not accepted by any relay: {reasons}"); + } + + Ok(event) } .await; @@ -1372,53 +1399,7 @@ fn resolve_statuses( .collect() } -// The `From ` header on the first line. -fn patch_current_commit(patch: &str) -> Option<&str> { - let line = patch.lines().next()?; - let hex = line.strip_prefix("From ")?; - hex.split_whitespace().next().filter(|hex| hex.len() == 40) -} - -struct PatchSeries { - parts: Vec, -} - -impl PatchSeries { - fn parse(patch: &str) -> Self { - Self { - parts: PatchParser::split_patch_series(patch) - .into_iter() - .map(str::to_owned) - .collect(), - } - } - - // The byte length of the first part over the NIP-34 size suggestion. - fn oversized_length(&self) -> Option { - self.parts - .iter() - .map(String::len) - .find(|length| *length > MAX_PATCH_EVENT_BYTES) - } - - // The tip of the series is its last commit; `git format-patch` orders - // patches oldest first. - fn tip_commit(&self) -> Option { - self.parts - .last() - .and_then(|part| patch_current_commit(part)) - .and_then(|hex| hex.parse::().ok()) - } - - fn commit_of(&self, index: usize) -> Option<&str> { - self.parts - .get(index) - .and_then(|part| patch_current_commit(part)) - .filter(|hex| hex.len() == 40) - } -} - -// Returns the root event, the one a PR references. +/// Publish a patch series as a sequence of `git/patch` events. #[allow(clippy::too_many_arguments)] async fn publish_patch_series( client: &Client, @@ -1478,7 +1459,16 @@ async fn publish_patch_series( let builder = EventBuilder::new(Kind::GitPatch, part.clone()).tags(tags); let event = builder.finalize_async(signer).await?; let output = client.send_event(&event).broadcast().await?; - let event = require_relay_accepted(output, event)?; + + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + bail!("event not accepted by any relay: {reasons}"); + } if root.is_none() { root = Some(event.clone()); @@ -1513,105 +1503,3 @@ fn comment_builder( .into_event_builder() .tags([Tag::coordinate(addr.clone().into(), None)]) } - -#[cfg(test)] -mod tests { - use std::collections::HashSet; - - use nostr_sdk::prelude::*; - use signed_core::GitEvent; - - use super::{RepoStore, comment_builder, patch_current_commit}; - - #[test] - fn parses_format_patch_header() { - let patch = "From 1f6c0c5f3f1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a Mon Sep 17 00:00:00 2001\nFrom: A \nSubject: [PATCH] fix\n\n---\n"; - assert_eq!( - patch_current_commit(patch), - Some("1f6c0c5f3f1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a") - ); - } - - #[test] - fn comment_builder_follows_nip22() { - let keys = Keys::generate(); - let root = EventBuilder::new(Kind::GitIssue, "issue body") - .finalize(&keys) - .expect("signed event"); - let addr = signed_core::RepoAddr::new(root.pubkey, "my-repo"); - let relay = RelayUrl::parse("wss://relay.example.com").expect("valid relay URL"); - - let event = comment_builder(&root, None, Some(&relay), &addr, "hi".into()) - .finalize(&keys) - .expect("signed event"); - - assert_eq!(event.kind, Kind::Comment); - - let kinds: Vec<&str> = event.tags.iter().map(Tag::kind).collect(); - for expected in ["E", "K", "P", "e", "k", "p", "a"] { - assert!(kinds.contains(&expected), "missing {expected} tag"); - } - - let e = event.tags.iter().find(|t| t.kind() == "E").expect("E tag"); - let slice = e.as_slice(); - assert_eq!(slice[1], root.id.to_hex()); - assert_eq!(slice[2], relay.as_str()); - assert_eq!(slice[3], root.pubkey.to_hex()); - - let e = event.tags.iter().find(|t| t.kind() == "e").expect("e tag"); - assert_eq!(e.as_slice()[1], root.id.to_hex()); - - assert!(event.references_root(&root.id)); - } - - #[test] - fn maintainer_filters_name_owner_and_maintainers() { - let owner = Keys::generate().public_key(); - let maintainer = Keys::generate().public_key(); - let addr = signed_core::RepoAddr::new(owner, "my-repo"); - - let filters = RepoStore::maintainer_filters(&addr, &[maintainer]); - assert_eq!(filters.len(), 4); - - let expected = HashSet::from([owner, maintainer]); - - let named = |filter: &Filter| -> HashSet { - let authors = filter.authors.iter().flatten().copied(); - let p_tag = filter - .generic_tags - .get(&SingleLetterTag::LOWERCASE_P) - .into_iter() - .flatten() - .filter_map(|value| PublicKey::from_hex(value).ok()); - authors.chain(p_tag).collect() - }; - - let announcement = &filters[0]; - assert_eq!(named(announcement), expected); - assert!( - announcement - .generic_tags - .contains_key(&SingleLetterTag::LOWERCASE_D) - ); - - for filter in &filters[1..3] { - assert_eq!(named(filter), expected); - assert!( - filter - .generic_tags - .contains_key(&SingleLetterTag::LOWERCASE_A) - ); - } - - let deletions = &filters[3]; - assert_eq!( - deletions - .authors - .clone() - .unwrap_or_default() - .into_iter() - .collect::>(), - expected - ); - } -} diff --git a/crates/signed_state/src/sync_status.rs b/crates/signed_state/src/sync_status.rs index 43d19f9..c6aba00 100644 --- a/crates/signed_state/src/sync_status.rs +++ b/crates/signed_state/src/sync_status.rs @@ -229,7 +229,9 @@ impl SyncStatusStore { })?; if again { - this.update(cx, |this, cx| this.refresh(cx))?; + this.update(cx, |this, cx| { + this.refresh(cx); + })?; } Ok::<_, Error>(()) -- 2.54.0 From 0e3a5ed865aa41047f456a5c7fc456066617ead6 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Tue, 13 Oct 2026 09:40:58 +0700 Subject: [PATCH 3/6] clean up --- crates/signed_git/src/repo.rs | 11 +- crates/signed_state/src/bootstrap.rs | 10 +- crates/signed_state/src/checkouts.rs | 267 ++++++------------------- crates/signed_state/src/sync_status.rs | 3 +- 4 files changed, 71 insertions(+), 220 deletions(-) diff --git a/crates/signed_git/src/repo.rs b/crates/signed_git/src/repo.rs index 2240653..de8a895 100644 --- a/crates/signed_git/src/repo.rs +++ b/crates/signed_git/src/repo.rs @@ -62,16 +62,13 @@ impl Repo { pub fn init(path: &Path, name: &str, description: &str) -> Result { use gix::refs::transaction::{Change, LogChange, PreviousValue, RefEdit, RefLog}; - std::fs::create_dir_all(path) - .with_context(|| format!("failed to create {}", path.display()))?; - + std::fs::create_dir_all(path)?; let repo = gix::init(path)?; let (signature, mut time_buf) = Self::repository_signature(); - let signature = signature.to_ref(&mut time_buf); - let head = gix::refs::FullName::try_from("HEAD") - .map_err(|e| anyhow::anyhow!("invalid ref name: {e}"))?; + let signature = signature.to_ref(&mut time_buf); + let head = gix::refs::FullName::try_from("HEAD")?; repo.edit_references_as( [RefEdit { @@ -99,7 +96,7 @@ impl Repo { format!("# {name}\n\n{description}\n") }; - std::fs::write(path.join("README.md"), &readme).context("failed to write README.md")?; + std::fs::write(path.join("README.md"), &readme)?; let blob = repo.write_object(gix::objs::Blob { data: readme.into_bytes(), diff --git a/crates/signed_state/src/bootstrap.rs b/crates/signed_state/src/bootstrap.rs index dd0056f..0fe0fa7 100644 --- a/crates/signed_state/src/bootstrap.rs +++ b/crates/signed_state/src/bootstrap.rs @@ -78,16 +78,10 @@ pub async fn user_grasp_list_servers( ) -> Result, Error> { let events: Vec = client .database() - .query(Filters::grasp_list(user)) + .query(Filters::grasp_list(user).limit(1)) .await? .into_iter() .collect(); - let latest = events - .into_iter() - .max_by_key(|event| event.created_at) - .map(|event| grasp_list_servers(&event)) - .unwrap_or_default(); - - Ok(latest) + Ok(events.first().map(grasp_list_servers).unwrap_or_default()) } diff --git a/crates/signed_state/src/checkouts.rs b/crates/signed_state/src/checkouts.rs index 91db84e..b72892e 100644 --- a/crates/signed_state/src/checkouts.rs +++ b/crates/signed_state/src/checkouts.rs @@ -45,6 +45,53 @@ struct Remembered { last_used: u64, } +impl Remembered { + fn resolve_associations<'a, A>( + remembered: &[Remembered], + scanned: &[(PathBuf, Option, Option)], + announcements: A, + ) -> HashMap> + where + A: IntoIterator, + { + let announcements: Vec<&Announcement> = announcements.into_iter().collect(); + let mut out: HashMap> = HashMap::new(); + + let mut sorted: Vec<&Remembered> = remembered.iter().collect(); + sorted.sort_by_key(|record| std::cmp::Reverse(record.last_used)); + for record in sorted { + let paths = out.entry(record.addr.clone()).or_default(); + if !paths.contains(&record.path) { + paths.push(record.path.clone()); + } + } + + for (path, origin, root) in scanned { + for announcement in &announcements { + let url_match = origin.as_deref().is_some_and(|origin| { + announcement + .clone + .iter() + .any(|url| same_repo_url(origin, url.as_str())) + }); + + let euc_match = root + .as_deref() + .is_some_and(|root| announcement.euc.as_deref() == Some(root)); + + if url_match || euc_match { + let paths = out.entry(announcement.addr()).or_default(); + if !paths.contains(path) { + paths.push(path.clone()); + } + } + } + } + + out + } +} + pub struct CheckoutsStore { by_repo: HashMap>, statuses: HashMap>, @@ -117,9 +164,10 @@ impl CheckoutsStore { if !cfg!(target_arch = "wasm32") { let weak = cx.entity().downgrade(); cx.defer(move |cx| { - if let Err(error) = weak.update(cx, |this, cx| this.refresh(cx)) { - log::warn!("checkouts store dropped before initial refresh could run: {error}"); - } + weak.update(cx, |this, cx| { + this.refresh(cx); + }) + .ok(); }); } @@ -143,13 +191,15 @@ impl CheckoutsStore { if cfg!(target_arch = "wasm32") { return; } + + let settings = SettingsStore::global(cx); + let addr_str = addr.to_string(); + let last_used = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_secs()) .unwrap_or(0); - let addr_str = addr.to_string(); - let settings = SettingsStore::global(cx); settings.update(cx, |settings, cx| { settings.edit( |s| { @@ -306,7 +356,7 @@ impl CheckoutsStore { } let associations = - CheckoutsStore::resolve_associations(&remembered, &facts, announcements.iter()); + Remembered::resolve_associations(&remembered, &facts, announcements.iter()); let associations: HashMap> = associations .into_iter() @@ -435,7 +485,7 @@ impl CheckoutsStore { Ok::<_, Error>((statuses, push_statuses)) }); - let task: gpui::Task> = cx.spawn(async move |this, cx| { + cx.spawn(async move |this, cx| { let Ok((statuses, push_statuses)) = work.await else { return Ok(()); }; @@ -456,53 +506,9 @@ impl CheckoutsStore { } })?; - Ok(()) - }); - task.detach(); - } -} - -impl CheckoutsStore { - fn resolve_associations<'a>( - remembered: &[Remembered], - scanned: &[(PathBuf, Option, Option)], - announcements: impl IntoIterator, - ) -> HashMap> { - let announcements: Vec<&Announcement> = announcements.into_iter().collect(); - let mut out: HashMap> = HashMap::new(); - - let mut sorted: Vec<&Remembered> = remembered.iter().collect(); - sorted.sort_by_key(|record| std::cmp::Reverse(record.last_used)); - for record in sorted { - let paths = out.entry(record.addr.clone()).or_default(); - if !paths.contains(&record.path) { - paths.push(record.path.clone()); - } - } - - for (path, origin, root) in scanned { - for announcement in &announcements { - let url_match = origin.as_deref().is_some_and(|origin| { - announcement - .clone - .iter() - .any(|url| same_repo_url(origin, url.as_str())) - }); - - let euc_match = root - .as_deref() - .is_some_and(|root| announcement.euc.as_deref() == Some(root)); - - if url_match || euc_match { - let paths = out.entry(announcement.addr()).or_default(); - if !paths.contains(path) { - paths.push(path.clone()); - } - } - } - } - - out + Ok::<(), anyhow::Error>(()) + }) + .detach(); } fn checkout_status(path: &Path, announced_head: Option<&str>) -> Option { @@ -535,7 +541,7 @@ impl CheckoutsStore { }) } - // Never fetches the checked-out refs; reads the tracking refs as-is. + /// Returns checkouts that are ahead of their remote tracking branch. fn checkout_push_status(path: &Path, fetch: bool) -> Option { let repo = Repo::try_open(path)?; if repo.is_dirty() { @@ -546,15 +552,13 @@ impl CheckoutsStore { let head = repo.head()?; let origin = repo.origin_url().ok().flatten()?; - if fetch { - repo.fetch_refs(&[origin], "+refs/heads/*:refs/remotes/origin/*") - .ok(); + if fetch && let Err(e) = repo.fetch_refs(&[origin], "+refs/heads/*:refs/remotes/origin/*") { + log::warn!("failed to fetch refs for {path:?}: {e}"); } let remote = format!("refs/remotes/origin/{branch}"); - // A branch never fetched or pushed yet compares against the remote - // HEAD, the fork point in practice. + // If the branch is new and has no remote tracking branch, fall back to origin/HEAD. let base = if repo.ref_exists(&remote) { remote } else if repo.ref_exists("refs/remotes/origin/HEAD") { @@ -648,146 +652,3 @@ impl CheckoutsStore { branch_matches || tip_matches } } - -#[cfg(test)] -mod tests { - use std::process::Command; - - use super::*; - - #[test] - fn same_repo_url_ignores_the_transport_scheme() { - assert!(same_repo_url( - "grasp://relay.ngit.dev/npub1test/repo", - "https://relay.ngit.dev/npub1test/repo.git" - )); - assert!(same_repo_url( - "ws://localhost:8080/npub1test/repo", - "http://localhost:8080/npub1test/repo" - )); - assert!(!same_repo_url( - "wss://localhost:8081/npub1test/repo", - "wss://localhost:8080/npub1test/repo" - )); - assert!(!same_repo_url( - "wss://host/npub1test/repo", - "wss://host/npub1other/repo" - )); - assert!(same_repo_url("/local/path", "/local/path")); - assert!(!same_repo_url("/local/path", "/local/other")); - } - - #[test] - fn checkout_status_reports_ahead_branches_only() { - let dir = tempfile::tempdir().expect("tempdir"); - let path = dir.path().join("repo"); - let _initial = Repo::init(&path, "My Repo", "").expect("init"); - let run = |args: &[&str]| { - let status = Command::new("git") - .current_dir(&path) - .env("GIT_AUTHOR_NAME", "Test Author") - .env("GIT_AUTHOR_EMAIL", "test@example.com") - .env("GIT_COMMITTER_NAME", "Test Author") - .env("GIT_COMMITTER_EMAIL", "test@example.com") - .env("GIT_EDITOR", "true") - .args(args) - .status() - .expect("git"); - assert!(status.success(), "git {args:?} failed"); - }; - let commit = |message: &str| { - run(&["add", "-A"]); - run(&["commit", "-m", message]); - }; - - run(&["checkout", "-b", "feature"]); - std::fs::write(path.join("feature.txt"), "x\n").expect("write"); - commit("feature work"); - let status = CheckoutsStore::checkout_status(&path, Some("main")).expect("status"); - assert_eq!(status.branch, "feature"); - assert_eq!(status.base, "main"); - assert_eq!(status.ahead, 1); - assert_eq!(status.head.len(), 40); - - std::fs::write(path.join("uncommitted.txt"), "y\n").expect("write"); - assert!(CheckoutsStore::checkout_status(&path, Some("main")).is_none()); - run(&["checkout", "--", "."]); - - run(&["checkout", "main"]); - assert_eq!(CheckoutsStore::checkout_status(&path, Some("main")), None); - } - - #[test] - fn checkout_push_status_counts_unpushed_commits_only() { - let dir = tempfile::tempdir().expect("tempdir"); - let remote = dir.path().join("remote"); - Repo::init(&remote, "My Repo", "").expect("init"); - let config = Command::new("git") - .args(["config", "receive.denyCurrentBranch", "ignore"]) - .current_dir(&remote) - .status() - .expect("git config"); - assert!(config.success(), "git config failed"); - - let checkout = dir.path().join("checkout"); - let status = Command::new("git") - .args([ - "clone", - "-q", - remote.to_str().unwrap(), - checkout.to_str().unwrap(), - ]) - .status() - .expect("git clone"); - assert!(status.success(), "git clone failed"); - - let run = |args: &[&str]| { - let status = Command::new("git") - .current_dir(&checkout) - .env("GIT_AUTHOR_NAME", "Test Author") - .env("GIT_AUTHOR_EMAIL", "test@example.com") - .env("GIT_COMMITTER_NAME", "Test Author") - .env("GIT_COMMITTER_EMAIL", "test@example.com") - .env("GIT_EDITOR", "true") - .args(args) - .status() - .expect("git"); - assert!(status.success(), "git {args:?} failed"); - }; - - // A fresh clone has nothing to push. - assert_eq!(CheckoutsStore::checkout_push_status(&checkout, true), None); - - std::fs::write(checkout.join("work.txt"), "x\n").expect("write"); - run(&["add", "-A"]); - run(&["commit", "-m", "local work"]); - let status = CheckoutsStore::checkout_push_status(&checkout, true).expect("status"); - assert_eq!(status.branch, "main"); - assert_eq!(status.base, "refs/remotes/origin/main"); - assert_eq!(status.ahead, 1); - assert_eq!(status.head.len(), 40); - - let local = CheckoutsStore::checkout_push_status(&checkout, false).expect("local status"); - assert_eq!(local.ahead, 1); - - run(&["push", "origin", "main"]); - assert_eq!(CheckoutsStore::checkout_push_status(&checkout, true), None); - - let remote_run = |args: &[&str]| { - let status = Command::new("git") - .current_dir(&remote) - .env("GIT_AUTHOR_NAME", "Other Author") - .env("GIT_AUTHOR_EMAIL", "other@example.com") - .env("GIT_COMMITTER_NAME", "Other Author") - .env("GIT_COMMITTER_EMAIL", "other@example.com") - .args(args) - .status() - .expect("git"); - assert!(status.success(), "git {args:?} failed"); - }; - std::fs::write(remote.join("other.txt"), "y\n").expect("write"); - remote_run(&["add", "-A"]); - remote_run(&["commit", "-m", "remote work"]); - assert_eq!(CheckoutsStore::checkout_push_status(&checkout, true), None); - } -} diff --git a/crates/signed_state/src/sync_status.rs b/crates/signed_state/src/sync_status.rs index c6aba00..cfbe4a9 100644 --- a/crates/signed_state/src/sync_status.rs +++ b/crates/signed_state/src/sync_status.rs @@ -13,8 +13,7 @@ use crate::checkouts::CheckoutsStore; use crate::refresh::{RefreshGate, RefreshRequest}; use crate::repos::RepoListStore; -// Nostr state changes rarely, so drift detection runs on a slow poll. -const SYNC_POLL: Duration = Duration::from_secs(60); +const SYNC_POLL: Duration = Duration::from_secs(180); const MAX_SYNC_CHECKOUTS: usize = 8; struct GlobalSyncStatusStore(Entity); -- 2.54.0 From f02f699aceab53c5a3c08754420a8ac48d9b5285 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Tue, 13 Oct 2026 10:08:02 +0700 Subject: [PATCH 4/6] Remove retry logic from grasp push --- crates/signed_state/src/backend.rs | 4 - crates/signed_state/src/push.rs | 229 ++++++----------------------- 2 files changed, 42 insertions(+), 191 deletions(-) diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index 32e2aed..5c6a164 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -812,7 +812,6 @@ 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, @@ -823,7 +822,6 @@ impl Backend { &path, &owner, &relays, - &executor, |path, base, owner, repo_id| { Repo::open(path)?.push_all(base, owner, repo_id) }, @@ -1234,7 +1232,6 @@ async fn announce_repository_and_push( let servers = servers.to_vec(); let head = head.clone(); let path = path.to_path_buf(); - let executor = cx.background_executor().clone(); cx.background_spawn(async move { push_staged_to_grasps( &client, @@ -1245,7 +1242,6 @@ async fn announce_repository_and_push( &path, &owner, &servers, - &executor, push, ) .await diff --git a/crates/signed_state/src/push.rs b/crates/signed_state/src/push.rs index a5471a4..920bcd6 100644 --- a/crates/signed_state/src/push.rs +++ b/crates/signed_state/src/push.rs @@ -1,17 +1,11 @@ use std::path::Path; -use std::time::Duration; use anyhow::Error; -use gpui::BackgroundExecutor; use nostr::prelude::Url; use nostr_sdk::prelude::*; use signed_core::RepoState; -use signed_git::{Repo, RepoRefState}; use signed_nostr::UniversalSigner; -const GRASP_PUSH_ATTEMPTS: usize = 3; -const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1); - /// `ws://` grasp servers, like ngit, use `http(s)://`. pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option { let parsed = Url::parse(relay.as_str()).ok()?; @@ -135,109 +129,6 @@ impl PushOutcome { } } -/// Signs with a `created_at` strictly newer than `last_created_at`, -/// since a relay treats a same-id resend as a duplicate. -async fn sign_state_event( - signer: &UniversalSigner, - repo_id: &str, - refs: &[(String, String)], - head: Option<&str>, - last_created_at: u64, -) -> Result<(Event, u64), String> { - let now = Timestamp::now().as_secs(); - let created_at = if now > last_created_at { - now - } else { - last_created_at + 1 - }; - - let event = RepoState::build(repo_id, refs, head) - .custom_created_at(Timestamp::from_secs(created_at)) - .finalize_async(signer) - .await - .map_err(|e| format!("could not sign the state event: {e}"))?; - - Ok((event, created_at)) -} - -/// On a grasp relay an accepted event is parked in purgatory, which authorizes the paired git push. -async fn stage_event_on_relay( - client: &Client, - relay: &RelayUrl, - event: &Event, -) -> Result<(), String> { - client - .add_relay(relay) - .and_connect() - .await - .map_err(|e| format!("could not add relay {relay}: {e}"))?; - - let output = client - .send_event(event) - .to([relay.clone()]) - .await - .map_err(|e| format!("could not send the state event to {relay}: {e}"))?; - - if output.success.contains_key(relay) { - Ok(()) - } else { - let reason = output - .failed - .get(relay) - .cloned() - .unwrap_or_else(|| "relay did not confirm the event".to_owned()); - Err(reason) - } -} - -/// Refresh the repo's ref state, keeping the current branch as the head if it is still present. -fn refresh_ref_state(path: &Path, head: Option<&str>) -> Option { - let state = Repo::open(path).and_then(|repo| repo.ref_state()).ok()?; - - if state.refs.is_empty() { - return None; - } - - let head = match head { - Some(current) => { - let name = format!("refs/heads/{current}"); - if state.refs.iter().any(|(known, _)| *known == name) { - Some(current.to_owned()) - } else { - state.head - } - } - None => state.head, - }; - - Some(RepoRefState { - refs: state.refs, - head, - }) -} - -/// Denials that staging a fresh state event and pushing against the refreshed advertisement resolve. -fn is_transient_grasp_denial(stderr: &str) -> bool { - let error = stderr.to_lowercase(); - [ - "no state events in purgatory", - "no matching state event", - "doesn't match push", - "none from authorized publishers", - "no repository announcement found", - "cannot lock ref", - "incorrect old value provided", - ] - .iter() - .any(|marker| error.contains(marker)) -} - -/// The server's own ref alignment moved a ref mid-push; the pushed data is usually already there. -fn is_stale_advertisement_race(stderr: &str) -> bool { - let error = stderr.to_lowercase(); - error.contains("cannot lock ref") || error.contains("incorrect old value provided") -} - #[allow(clippy::too_many_arguments)] pub(crate) async fn push_staged_to_grasps( client: &Client, @@ -248,12 +139,9 @@ pub(crate) async fn push_staged_to_grasps( path: &Path, owner: &str, servers: &[RelayUrl], - executor: &BackgroundExecutor, push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>, ) -> PushOutcome { let mut outcome = PushOutcome::default(); - let mut refs = refs.to_vec(); - let mut head = head.map(str::to_owned); for relay in servers { let Some(base) = grasp_base_url(relay) else { @@ -263,86 +151,53 @@ pub(crate) async fn push_staged_to_grasps( continue; }; - let git_url = format!("{base}/{owner}/{repo_id}.git"); - - let mut reason = None; - let mut last_created_at = 0; - let mut staged_event = None; - - 'server: for attempt in 1..=GRASP_PUSH_ATTEMPTS { - if attempt > 1 { - executor.timer(GRASP_RETRY_DELAY).await; - // The repo's refs may have moved since the last attempt, so refresh the state. - if let Some(state) = refresh_ref_state(path, head.as_deref()) { - refs = state.refs; - head = state.head; - } - } - - let (event, created_at) = - match sign_state_event(signer, repo_id, &refs, head.as_deref(), last_created_at) - .await - { - Ok(signed) => signed, - Err(e) => { - reason = Some(e); - break 'server; - } - }; - - last_created_at = created_at; - - // Stage the event on the relay, - // so that the push can be accepted if the relay's state is aligned. - if let Err(e) = stage_event_on_relay(client, relay, &event).await { - let retried = - attempt == 1 && stage_event_on_relay(client, relay, &event).await.is_ok(); - if !retried { - reason = Some(e); - break 'server; - } - } - staged_event = Some(event.clone()); - - match push(path, &base, owner, repo_id) { - Ok(()) => { - outcome.keep_newest(event); - break 'server; - } - Err(e) => { - let text = e.to_string(); - let retriable = - attempt < GRASP_PUSH_ATTEMPTS && is_transient_grasp_denial(&text); - reason = Some(text); - if !retriable { - break 'server; - } - } - } - } - - // If the relay's denial was a stale-ref race, - // check if the repo has moved on and retry the push. - if let Some(last_reason) = &reason - && is_stale_advertisement_race(last_reason) - && Repo::open(path) - .and_then(|repo| repo.remote_has_refs(&git_url, &refs)) - .unwrap_or(false) + let event = match RepoState::build(repo_id, refs, head) + .finalize_async(signer) + .await { - if let Some(event) = staged_event { - outcome.keep_newest(event); - } - reason = None; - } - - match reason { - Some(reason) => { - log::warn!("grasp push failed: {relay}: {reason}"); + Ok(event) => event, + Err(e) => { + let reason = format!("could not sign the state event: {e}"); outcome .servers .push(GraspServer::failed(relay.clone(), reason)); + continue; + } + }; + + // The relay parks the accepted event in purgatory, authorizing the paired git push. + let staging = match client.add_relay(relay).and_connect().await { + Ok(_) => match client.send_event(&event).to([relay.clone()]).await { + Ok(output) if output.success.contains_key(relay) => Ok(()), + Ok(output) => Err(output + .failed + .get(relay) + .cloned() + .unwrap_or_else(|| "relay did not confirm the event".to_owned())), + Err(e) => Err(format!("could not send the state event to {relay}: {e}")), + }, + Err(e) => Err(format!("could not add relay {relay}: {e}")), + }; + + if let Err(reason) = staging { + log::warn!("grasp push failed: {relay}: {reason}"); + outcome + .servers + .push(GraspServer::failed(relay.clone(), reason)); + continue; + } + + match push(path, &base, owner, repo_id) { + Ok(()) => { + outcome.keep_newest(event); + outcome.servers.push(GraspServer::ok(relay.clone())); + } + Err(e) => { + log::warn!("grasp push failed: {relay}: {e}"); + outcome + .servers + .push(GraspServer::failed(relay.clone(), e.to_string())); } - None => outcome.servers.push(GraspServer::ok(relay.clone())), } } -- 2.54.0 From 3c307691aa950d82e74213da5240fc160e7a166f Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Tue, 13 Oct 2026 10:27:27 +0700 Subject: [PATCH 5/6] clean up --- crates/signed_git/src/repo.rs | 51 +------------------ crates/signed_git/src/tests.rs | 36 -------------- crates/signed_state/src/backend.rs | 78 ++++++++++++++++++------------ 3 files changed, 48 insertions(+), 117 deletions(-) diff --git a/crates/signed_git/src/repo.rs b/crates/signed_git/src/repo.rs index de8a895..811f840 100644 --- a/crates/signed_git/src/repo.rs +++ b/crates/signed_git/src/repo.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeSet, HashMap, HashSet}; +use std::collections::{BTreeSet, HashSet}; use std::io::Write; use std::path::{Path, PathBuf}; use std::process::Stdio; @@ -1262,55 +1262,6 @@ impl Repo { Ok(()) } - /// Checks that `url` advertises exactly the expected ref oids. - pub fn remote_has_refs(&self, url: &str, expected: &[(String, String)]) -> Result { - if expected.is_empty() { - return Ok(true); - } - - let url = Self::transport_url(url); - - let refspecs = expected - .iter() - .map(|(name, _)| { - gix::refspec::parse( - gix::bstr::BStr::new(format!("+{name}:{name}").as_bytes()), - gix::refspec::parse::Operation::Fetch, - ) - .map(|spec| spec.to_owned()) - }) - .collect::, _>>() - .into_anyhow() - .context("invalid refspec")?; - - let options = gix::remote::ref_map::Options { - extra_refspecs: refspecs, - ..Default::default() - }; - - let (refs, _) = self - .inner - .remote_at(url.as_str()) - .with_context(|| format!("cannot use remote {url}"))? - .connect(gix::remote::Direction::Fetch) - .with_context(|| format!("cannot connect to {url}"))? - .ref_map(Discard, options) - .with_context(|| format!("listing refs of {url} failed"))?; - - let advertised: HashMap = refs - .remote_refs - .iter() - .filter_map(|reference| { - let (name, object, _peeled) = reference.unpack(); - object.map(|oid| (String::from_utf8_lossy(name).into_owned(), oid.to_string())) - }) - .collect(); - - Ok(expected - .iter() - .all(|(name, oid)| advertised.get(name.as_str()) == Some(oid))) - } - /// Sets the `origin` URL when the repository has no remote yet. pub fn ensure_origin(&self, url: &str) -> Result<()> { if self.inner.find_remote("origin").is_ok() { diff --git a/crates/signed_git/src/tests.rs b/crates/signed_git/src/tests.rs index 59dca08..8e4a8d0 100644 --- a/crates/signed_git/src/tests.rs +++ b/crates/signed_git/src/tests.rs @@ -113,42 +113,6 @@ fn ref_state_matches_the_git_cli_ref_listing() { ); } -#[test] -fn remote_has_refs_reports_whether_pushed_refs_landed() { - let server = tempfile::tempdir().unwrap(); - bare_server(server.path(), "npub1test", "my-repo"); - let (dir, repo) = fixture(&[("a.txt", b"one")]); - commit_all(&repo, "initial"); - let dir = dir.path(); - let main = git_in(dir, &["rev-parse", "refs/heads/main"]).expect("main oid"); - let url = format!("file://{}/npub1test/my-repo.git", server.path().display()); - let expected = vec![("refs/heads/main".to_owned(), main.clone())]; - let repo = Repo::open(dir).expect("open"); - - assert!(!repo.remote_has_refs(&url, &expected).expect("probe")); - - repo.push_all( - &format!("file://{}", server.path().display()), - "npub1test", - "my-repo", - ) - .expect("push"); - - assert!(repo.remote_has_refs(&url, &expected).expect("probe")); - - let stale = vec![("refs/heads/main".to_owned(), "0".repeat(40))]; - assert!(!repo.remote_has_refs(&url, &stale).expect("probe")); - - git_run(dir, &["tag", "v1.0"]); - repo.push_all( - &format!("file://{}", server.path().display()), - "npub1test", - "my-repo", - ) - .expect("push"); - assert!(repo.remote_has_refs(&url, &expected).expect("probe")); -} - #[test] fn repo_ref_state_lists_branches_tags_and_head() { let (_dir, repo) = fixture(&[("a.txt", b"hello")]); diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index 5c6a164..6d847db 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -5,9 +5,7 @@ use std::time::{Duration, Instant}; use anyhow::{Error, anyhow, bail}; use bitcoin_hashes::sha1::Hash as Sha1Hash; -use gpui::{ - App, AppContext, AsyncApp, Context, Entity, EventEmitter, Global, Task, TaskExt, WeakEntity, -}; +use gpui::{App, AppContext, AsyncApp, Context, Entity, EventEmitter, Global, Task, TaskExt}; use nostr::event::IntoEventBuilder; use nostr::nips::nip19::Nip19Coordinate; use nostr_connect::prelude::*; @@ -194,7 +192,7 @@ impl Backend { this.set_signer(keys, cx); })?; } else if content.starts_with("bunker://") { - let (base, keys) = extract_master_key(&content); + let (base, keys) = extract_master_key(&content)?; let uri = NostrConnectUri::parse(base)?; let mut signer = NostrConnect::new( uri, @@ -463,7 +461,7 @@ impl Backend { let event = builder.finalize_async(&signer).await?; let output = client.send_event(&event).broadcast().await?; - if output.success.is_empty() && !output.failed.is_empty() { + if output.success.is_empty() { let reasons = output .failed .values() @@ -539,7 +537,7 @@ impl Backend { folder.join(dir_name) }; - cx.spawn(async move |this, cx| { + cx.spawn(async move |_this, cx| { let work = cx.background_spawn({ let destination = destination.clone(); let name = name.clone(); @@ -577,7 +575,6 @@ impl Backend { ); let event = announce_repository_and_push( - &this, &client, &signer, announcement, @@ -643,7 +640,7 @@ impl Backend { let client = self.client.clone(); let signer = self.signer.clone(); - cx.spawn(async move |this, cx| { + cx.spawn(async move |_this, cx| { let (mut state, euc) = cx .background_spawn({ let path = path.clone(); @@ -672,7 +669,6 @@ impl Backend { let ann = build_announcement(&repo_id, &name, &description, &owner, &servers, euc); let event = announce_repository_and_push( - &this, &client, &signer, ann, @@ -1155,11 +1151,15 @@ impl Backend { cx.background_spawn(async move { let builder = EventDeletionRequest::new().id(event.id); let deletion = builder.finalize_async(&signer).await?; - client.send_event(&deletion).broadcast().await?; + let output = client.send_event(&deletion).broadcast().await?; - Ok::<(), anyhow::Error>(()) + if output.success.is_empty() { + bail!("deletion event not accepted by any relay"); + } + + Ok::<(), Error>(()) }) - .detach(); + .detach_and_log_err(cx); } } } @@ -1190,7 +1190,6 @@ fn build_announcement( // Announce the repository, then stage the state event and push. #[allow(clippy::too_many_arguments)] async fn announce_repository_and_push( - backend: &WeakEntity, client: &Client, signer: &UniversalSigner, announcement: GitRepositoryAnnouncement, @@ -1204,13 +1203,15 @@ async fn announce_repository_and_push( push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error> + Send + 'static, ) -> Result { for url in servers { - client.add_relay(url).and_connect().await.ok(); + if let Err(e) = client.add_relay(url).and_connect().await { + log::warn!("failed to connect grasp relay {url}: {e}"); + } } let event = announcement.finalize_async(signer).await?; let output = client.send_event(&event).broadcast().await?; - if output.success.is_empty() && !output.failed.is_empty() { + if output.success.is_empty() { let reasons = output .failed .values() @@ -1252,17 +1253,34 @@ async fn announce_repository_and_push( if outcome.accepted() == 0 { // Retract the announcement so the repository is not left announced without content. - backend - .update(cx, |backend, cx| { - backend.retract_events(std::slice::from_ref(&event), cx); - }) - .ok(); + let retracted = async { + let deletion = EventDeletionRequest::new() + .id(event.id) + .finalize_async(signer) + .await?; + let output = client.send_event(&deletion).broadcast().await?; - return Err(anyhow!( - "The repository was announced, but the push to every grasp server failed: {}. \ - The announcement has been retracted", - outcome.failure_summary() - )); + // Without relay acceptance the deletion request stays live. + if output.success.is_empty() { + bail!("deletion event not accepted by any relay"); + } + + Ok::<(), Error>(()) + } + .await; + + return Err(match retracted { + Ok(()) => anyhow!( + "The repository was announced, but the push to every grasp server failed: {}. \ + The announcement has been retracted", + outcome.failure_summary() + ), + Err(e) => anyhow!( + "The repository was announced, but the push to every grasp server failed: {}. \ + Retracting the announcement failed: {e}; delete the repository to clean it up", + outcome.failure_summary() + ), + }); } // Fan the state out to the relays once a git server holds the objects. @@ -1313,15 +1331,13 @@ impl UpdateEvent { } } -fn extract_master_key(credential: &str) -> (&str, Keys) { +fn extract_master_key(credential: &str) -> Result<(&str, Keys), Error> { match credential.split_once("master=") { Some((base, nsec)) => { - let keys = SecretKey::parse(nsec) - .map(Keys::new) - .unwrap_or_else(|_| Keys::generate()); - (base.trim_end_matches(['?', '&']), keys) + let keys = Keys::new(SecretKey::parse(nsec)?); + Ok((base.trim_end_matches(['?', '&']), keys)) } - None => (credential, Keys::generate()), + None => Ok((credential, Keys::generate())), } } -- 2.54.0 From 56d3924e46bf0450383ff0f6af4e37dba1c499a3 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Tue, 13 Oct 2026 10:41:44 +0700 Subject: [PATCH 6/6] Make signing functions generic over the signer trait --- crates/signed_state/src/backend.rs | 11 +++++++---- crates/signed_state/src/push.rs | 10 ++++++---- crates/signed_state/src/repo.rs | 10 ++++++---- 3 files changed, 19 insertions(+), 12 deletions(-) diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index 6d847db..de1a045 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -1189,9 +1189,9 @@ fn build_announcement( // Announce the repository, then stage the state event and push. #[allow(clippy::too_many_arguments)] -async fn announce_repository_and_push( +async fn announce_repository_and_push( client: &Client, - signer: &UniversalSigner, + signer: &S, announcement: GitRepositoryAnnouncement, repo_id: &str, owner: &str, @@ -1201,7 +1201,10 @@ async fn announce_repository_and_push( path: &Path, cx: &mut AsyncApp, push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error> + Send + 'static, -) -> Result { +) -> Result +where + S: AsyncGetPublicKey + AsyncSignEvent + Clone, +{ for url in servers { if let Err(e) = client.add_relay(url).and_connect().await { log::warn!("failed to connect grasp relay {url}: {e}"); @@ -1227,7 +1230,7 @@ async fn announce_repository_and_push( } else { let push = { let client = client.clone(); - let signer = signer.clone(); + let signer = S::clone(signer); let repo_id = repo_id.to_owned(); let owner = owner.to_owned(); let servers = servers.to_vec(); diff --git a/crates/signed_state/src/push.rs b/crates/signed_state/src/push.rs index 920bcd6..56f7171 100644 --- a/crates/signed_state/src/push.rs +++ b/crates/signed_state/src/push.rs @@ -4,7 +4,6 @@ use anyhow::Error; use nostr::prelude::Url; use nostr_sdk::prelude::*; use signed_core::RepoState; -use signed_nostr::UniversalSigner; /// `ws://` grasp servers, like ngit, use `http(s)://`. pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option { @@ -130,9 +129,9 @@ impl PushOutcome { } #[allow(clippy::too_many_arguments)] -pub(crate) async fn push_staged_to_grasps( +pub(crate) async fn push_staged_to_grasps( client: &Client, - signer: &UniversalSigner, + signer: &S, repo_id: &str, refs: &[(String, String)], head: Option<&str>, @@ -140,7 +139,10 @@ pub(crate) async fn push_staged_to_grasps( owner: &str, servers: &[RelayUrl], push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>, -) -> PushOutcome { +) -> PushOutcome +where + S: AsyncGetPublicKey + AsyncSignEvent + ?Sized, +{ let mut outcome = PushOutcome::default(); for relay in servers { diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index b482fb7..fba0a34 100644 --- a/crates/signed_state/src/repo.rs +++ b/crates/signed_state/src/repo.rs @@ -12,7 +12,6 @@ use signed_core::{ filters, }; use signed_git::{Nip34Binding, Repo}; -use signed_nostr::UniversalSigner; use crate::backend::{Backend, BackendEvent}; use crate::bootstrap::user_grasp_list_servers; @@ -1401,16 +1400,19 @@ fn resolve_statuses( /// Publish a patch series as a sequence of `git/patch` events. #[allow(clippy::too_many_arguments)] -async fn publish_patch_series( +async fn publish_patch_series( client: &Client, - signer: &UniversalSigner, + signer: &S, addr: &RepoAddr, owner: PublicKey, euc: Option<&str>, series: &PatchSeries, first_marker: &str, reply_to: Option, -) -> Result { +) -> Result +where + S: AsyncGetPublicKey + AsyncSignEvent + ?Sized, +{ let mut root: Option = None; let mut previous = reply_to; -- 2.54.0