From d5acef65a6e77b97ae57c8656c80a34c8cea4b84 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Wed, 14 Oct 2026 08:11:14 +0700 Subject: [PATCH] Refactor push logic into SyncState methods --- crates/signed_state/src/backend.rs | 269 ++++++++--------------------- crates/signed_state/src/push.rs | 88 ---------- 2 files changed, 76 insertions(+), 281 deletions(-) diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index ef7ae38..3acef17 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -1,5 +1,5 @@ use std::collections::HashSet; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::str::FromStr; use std::time::{Duration, Instant}; @@ -19,9 +19,9 @@ 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}; +use crate::push::{PushOutcome, grasp_base_url, grasp_clone_url}; use crate::repos::RepoListStore; -use crate::state::GitRef; +use crate::state::{GitPushMode, GitRef, PushStepStatus, SyncState}; pub const USER_KEYRING: &str = "Signed Safe Storage"; pub const NOSTR_CONNECT_TIMEOUT: u64 = 60; @@ -36,6 +36,10 @@ pub enum BackendEvent { ProfileUpdates(Vec), RepoUpdates(Vec), Synced, + PushProgress { + addr: RepoAddr, + status: PushStepStatus, + }, Error(String), } @@ -575,23 +579,18 @@ impl Backend { Some(commit_sha), ); - let event = announce_repository_and_push( - &client, - &signer, - announcement, - &repo_id, - &owner, - &servers, - vec![GitRef { + let state = SyncState { + refs: vec![GitRef { name: "refs/heads/main".to_owned(), commit: commit_sha, }], - Some("main".to_owned()), - &destination, - cx, - |path, base, owner, repo_id| Repo::open(path)?.push_main(base, owner, repo_id), - ) - .await?; + head: Some("main".to_owned()), + path: Some(destination.clone()), + mode: GitPushMode::Main, + }; + + let event = + announce_and_push_or_retract(&client, &signer, &state, announcement, cx).await?; let announcement = Announcement::from_event(&event) .ok_or_else(|| anyhow!("failed to parse announcement"))?; @@ -650,8 +649,8 @@ impl Backend { let path = path.clone(); async move { let repo = Repo::open(&path)?; - let state = repo.ref_state()?; let euc = repo.root_commit()?; + let state = SyncState::from_repo(&path)?; Ok::<_, Error>((state, euc)) } }) @@ -664,34 +663,14 @@ impl Backend { if let Some(branch) = chosen { let wanted = format!("refs/heads/{branch}"); - if state.refs.iter().any(|(name, _)| name == &wanted) { + if state.refs.iter().any(|git_ref| git_ref.name == wanted) { state.head = Some(branch.to_owned()); } } - let euc = euc.and_then(|commit| Sha1Hash::from_str(&commit).ok()); + let euc = euc.and_then(|commit: String| Sha1Hash::from_str(&commit).ok()); let ann = build_announcement(&repo_id, &name, &description, &owner, &servers, euc); - - let refs = state - .refs - .iter() - .map(|(name, commit)| GitRef::new(name, commit)) - .collect::, Error>>()?; - - let event = announce_repository_and_push( - &client, - &signer, - ann, - &repo_id, - &owner, - &servers, - refs, - state.head.clone(), - &path, - cx, - |path, base, owner, repo_id| Repo::open(path)?.push_all(base, owner, repo_id), - ) - .await?; + let event = announce_and_push_or_retract(&client, &signer, &state, ann, cx).await?; if let Some(base) = servers.first().and_then(grasp_base_url) { let url = format!("{base}/{owner}/{repo_id}.git"); @@ -764,9 +743,6 @@ impl Backend { cx.notify(); }); - let owner = announcement.owner.to_bech32().unwrap(); - let repo_id = announcement.id.clone(); - let relays = announcement.relays.clone(); let client = self.client.clone(); let signer = self.signer.clone(); @@ -784,78 +760,50 @@ impl Backend { let mut state = { let work = cx.background_spawn({ let path = path.clone(); - async move { Repo::open(&path).and_then(|repo| repo.ref_state()) } + async move { SyncState::from_repo(&path) } }); work.await? }; // Keep the announced default branch in `HEAD` when it is among the pushed refs. - let heads: Vec<&str> = state - .refs - .iter() - .filter_map(|(name, _)| name.strip_prefix("refs/heads/")) - .collect(); - if let Some(head) = announced_head - && heads.iter().any(|branch| *branch == head) + && state + .refs + .iter() + .any(|git_ref| git_ref.name.strip_prefix("refs/heads/") == Some(head.as_str())) { state.head = Some(head); } - // Grasp servers authorize a push by the state event they hold in purgatory. - let refs = state - .refs - .iter() - .map(|(name, commit)| GitRef::new(name, commit)) - .collect::, Error>>()?; - let head = state.head.clone(); - - let outcome = if refs.is_empty() { + let outcome = if state.refs.is_empty() { PushOutcome::default() } else { - let push = cx.background_spawn({ - let client = client.clone(); - let signer = signer.clone(); - let path = path.clone(); - let owner = owner.clone(); - let repo_id = repo_id.clone(); - let relays = relays.clone(); - let refs = refs.clone(); - let head = head.clone(); - async move { - push_staged_to_grasps( - &client, - &signer, - &repo_id, - &refs, - head.as_deref(), - &path, - &owner, - &relays, - |path, base, owner, repo_id| { - Repo::open(path)?.push_all(base, owner, repo_id) - }, - ) - .await - } - }); - push.await + // Grasp servers authorize a push by the state event they hold in purgatory. + let builder = state.state_event(&announcement.id); + let event = builder.finalize_async(&signer).await?; + + let on_step = &mut |status: PushStepStatus, cx: &mut AsyncApp| { + this.update(cx, |_, cx| { + cx.emit(BackendEvent::PushProgress { + addr: addr.clone(), + status, + }); + })?; + Ok::<(), Error>(()) + }; + + state + .push_to_grasps(&client, event, &announcement, cx, on_step) + .await? }; - if !refs.is_empty() && outcome.accepted() == 0 { + if !state.refs.is_empty() && outcome.accepted() == 0 { bail!( "could not push the repository to any grasp server: {}", outcome.failure_summary() ); } - // Fan the state out to the relays once a git server holds the objects. - if let Some(state_event) = &outcome.state_event - && let Err(e) = client.send_event(state_event).broadcast().await - { - log::warn!("failed to broadcast repository state: {e}"); - } - Ok(outcome) }) } @@ -1201,113 +1149,48 @@ fn build_announcement( } } -// Announce the repository, then stage the state event and push. -#[allow(clippy::too_many_arguments)] -async fn announce_repository_and_push( +/// Announces a repository and pushes its state to grasp servers, +/// or retracts the announcement if the push fails. +async fn announce_and_push_or_retract( client: &Client, signer: &S, + state: &SyncState, announcement: GitRepositoryAnnouncement, - repo_id: &str, - owner: &str, - servers: &[RelayUrl], - refs: Vec, - head: Option, - path: &Path, cx: &mut AsyncApp, - push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error> + Send + 'static, ) -> Result where - S: AsyncGetPublicKey + AsyncSignEvent + Clone, + S: AsyncGetPublicKey + AsyncSignEvent + ?Sized, { - for url in servers { - if let Err(e) = client.add_relay(url).and_connect().await { - log::warn!("failed to connect grasp relay {url}: {e}"); - } - } + let state_event = state + .state_event(&announcement.id) + .finalize_async(signer) + .await?; + let announcement_event = announcement.finalize_async(signer).await?; - let event = announcement.finalize_async(signer).await?; - let output = client.send_event(&event).broadcast().await?; - - if output.success.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() { - PushOutcome::default() - } else { - let push = { - let client = client.clone(); - let signer = S::clone(signer); - 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(); - cx.background_spawn(async move { - push_staged_to_grasps( - &client, - &signer, - &repo_id, - &refs, - head.as_deref(), - &path, - &owner, - &servers, - push, - ) - .await - }) - }; - push.await - }; - - if outcome.accepted() == 0 { - // Retract the announcement so the repository is not left announced without content. - let retracted = async { - let deletion = EventDeletionRequest::new() - .id(event.id) - .finalize_async(signer) - .await?; - let output = client.send_event(&deletion).broadcast().await?; - - // 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. - if let Some(state_event) = &outcome.state_event - && let Err(e) = client.send_event(state_event).broadcast().await + match state + .announce_and_push( + client, + announcement_event.clone(), + state_event, + cx, + &mut |_, _| Ok(()), + ) + .await { - log::warn!("failed to broadcast repository state: {e}"); - } + Ok(event) => Ok(event), + Err(e) => { + let retracted = + SyncState::retract_announcement(client, signer, announcement_event.id).await; + let retraction = match retracted { + Ok(()) => "The announcement has been retracted".to_owned(), + Err(error) => format!( + "Retracting the announcement failed: {error}; delete the repository to clean it up" + ), + }; - Ok(event) + bail!("{e}. {retraction}"); + } + } } enum UpdateEvent { diff --git a/crates/signed_state/src/push.rs b/crates/signed_state/src/push.rs index ca95968..006ff6a 100644 --- a/crates/signed_state/src/push.rs +++ b/crates/signed_state/src/push.rs @@ -1,11 +1,6 @@ -use std::path::Path; - -use anyhow::Error; use nostr::prelude::Url; use nostr_sdk::prelude::*; -use crate::state::{GitRef, SyncState}; - /// `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()?; @@ -116,87 +111,4 @@ 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); - } - } -} - -#[allow(clippy::too_many_arguments)] -pub(crate) async fn push_staged_to_grasps( - client: &Client, - signer: &S, - repo_id: &str, - refs: &[GitRef], - head: Option<&str>, - path: &Path, - owner: &str, - servers: &[RelayUrl], - push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>, -) -> PushOutcome -where - S: AsyncGetPublicKey + AsyncSignEvent + ?Sized, -{ - let mut outcome = PushOutcome::default(); - - for relay in servers { - let Some(base) = grasp_base_url(relay) else { - outcome - .servers - .push(GraspServer::failed(relay, "no domain")); - continue; - }; - - let event = match SyncState::build(repo_id, refs, head) - .finalize_async(signer) - .await - { - Ok(event) => event, - Err(e) => { - let reason = format!("could not sign the state event: {e}"); - outcome.servers.push(GraspServer::failed(relay, 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 { - outcome.servers.push(GraspServer::failed(relay, reason)); - continue; - } - - match push(path, &base, owner, repo_id) { - Ok(()) => { - outcome.keep_newest(event); - outcome.servers.push(GraspServer::ok(relay.clone())); - } - Err(e) => { - outcome - .servers - .push(GraspServer::failed(relay, e.to_string())); - } - } - } - - outcome }