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

feat: multi-steps push progress (#30)
Rust / build (macos-latest, stable) (push) Canceled after 0s
Details
Rust / build (ubuntu-latest, stable) (push) Canceled after 0s
Details
Rust / build (windows-latest, stable) (push) Canceled after 0s
Details

Browse Source 
Reviewed-on: #30
This commit was merged in pull request #30.
This commit is contained in:
reya
2026-10-14 02:08:20 +00:00
parent b78418d815
commit e72e04c471
12 changed files with 753 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/signed_core/src/lib.rs
-2
View File
@@ -5,7 +5,6 @@ pub mod filters;
pub mod git_event;
pub mod inbox;
pub mod pull_request;
pub mod state;
pub mod status;
pub mod upstream;
@@ -16,6 +15,5 @@ pub use filters::Filters;
pub use git_event::GitEvent;
pub use inbox::{InboxItem, InboxReadState, ThreadResolver};
pub use pull_request::PullRequest;
pub use state::RepoState;
pub use status::RepoStatus;
pub use upstream::Upstream;
crates/signed_core/src/state.rs
-114
View File
@@ -1,114 +0,0 @@
use nostr::prelude::*;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RepoState {
pub refs: Vec<(String, String)>,
pub head: Option<String>,
}
impl RepoState {
pub fn build(id: &str, refs: &[(String, String)], head: Option<&str>) -> EventBuilder {
let mut tags: Vec<Tag> = vec![Tag::identifier(id.to_owned())];
for (name, commit) in refs {
tags.push(Tag::parse([name.as_str(), commit.as_str()]).expect("valid ref tag"));
}
if let Some(head) = head {
tags.push(
Tag::parse(["HEAD", &format!("ref: refs/heads/{head}")]).expect("valid HEAD tag"),
);
}
EventBuilder::new(Kind::RepoState, "").tags(tags)
}
pub fn parse(event: &Event) -> Self {
let mut refs = Vec::new();
let mut head = None;
for tag in event.tags.iter() {
match Nip34Tag::parse(tag.as_slice()) {
Ok(Nip34Tag::Head(branch)) => head = Some(branch),
Ok(Nip34Tag::RefHead { branch, commit }) => {
refs.push((format!("refs/heads/{branch}"), commit.to_string()));
}
Ok(Nip34Tag::RefTag { name, commit }) => {
refs.push((format!("refs/tags/{name}"), commit.to_string()));
}
_ => {}
}
}
Self { refs, head }
}
}
#[cfg(test)]
mod tests {
use super::*;
const COMMIT_A: &str = "aa231c4c6a5777dc89b42207b499891a344add5c";
const COMMIT_B: &str = "59429cfc6cb35b0a1ddace73b5a5c5ed57b8f5ca";
fn keys() -> Keys {
Keys::new(
SecretKey::from_hex("0000000000000000000000000000000000000000000000000000000000000001")
.expect("valid secret key"),
)
}
fn state_event(tags: &[&[&str]]) -> Event {
let tags: Vec<Tag> = tags
.iter()
.map(|t| Tag::parse(t.to_vec()).expect("valid tag"))
.collect();
EventBuilder::new(Kind::RepoState, "")
.tags(tags)
.finalize(&keys())
.expect("signed event")
}
#[test]
fn parses_heads_and_tags() {
let event = state_event(&[
&["HEAD", "ref: refs/heads/main"],
&["refs/heads/main", COMMIT_A],
&["refs/heads/dev", COMMIT_B],
&["refs/tags/v1.0", COMMIT_A],
]);
let state = RepoState::parse(&event);
assert_eq!(state.head.as_deref(), Some("main"));
assert_eq!(
state.refs,
vec![
("refs/heads/main".to_owned(), COMMIT_A.to_owned()),
("refs/heads/dev".to_owned(), COMMIT_B.to_owned()),
("refs/tags/v1.0".to_owned(), COMMIT_A.to_owned()),
]
);
}
#[test]
fn build_round_trips_through_parse() {
let refs = [
("refs/heads/main".to_owned(), COMMIT_A.to_owned()),
("refs/heads/dev".to_owned(), COMMIT_B.to_owned()),
("refs/tags/v1.0".to_owned(), COMMIT_A.to_owned()),
];
let event = RepoState::build("my-repo", &refs, Some("main"))
.finalize(&keys())
.expect("signed event");
assert_eq!(event.kind, Kind::RepoState);
assert_eq!(event.tags.identifier().as_deref(), Some("my-repo"));
let state = RepoState::parse(&event);
assert_eq!(state.refs, refs);
assert_eq!(state.head.as_deref(), Some("main"));
}
}
crates/signed_state/src/backend.rs
+101 -183
View File
@@ -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,8 +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::{GitPushMode, GitRef, PushStepStatus, SyncState};
pub const USER_KEYRING: &str = "Signed Safe Storage";
pub const NOSTR_CONNECT_TIMEOUT: u64 = 60;
@@ -35,6 +36,10 @@ pub enum BackendEvent {
ProfileUpdates(Vec<PublicKey>),
RepoUpdates(Vec<Update>),
Synced,
PushProgress {
addr: RepoAddr,
status: PushStepStatus,
},
Error(String),
}
@@ -523,6 +528,7 @@ impl Backend {
}
let owner = public_key.to_bech32().unwrap();
let addr = RepoAddr::new(public_key, repo_id.clone());
let servers = grasp_servers.clone();
let client = self.client.clone();
let signer = self.signer.clone();
@@ -537,7 +543,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();
@@ -574,20 +580,29 @@ impl Backend {
Some(commit_sha),
);
let event = announce_repository_and_push(
&client,
&signer,
announcement,
&repo_id,
&owner,
&servers,
vec![("refs/heads/main".to_owned(), commit)],
Some("main".to_owned()),
&destination,
cx,
|path, base, owner, repo_id| Repo::open(path)?.push_main(base, owner, repo_id),
)
.await?;
let state = SyncState {
refs: vec![GitRef {
name: "refs/heads/main".to_owned(),
commit: commit_sha,
}],
head: Some("main".to_owned()),
path: Some(destination.clone()),
mode: GitPushMode::Main,
};
let on_step = &mut |status: PushStepStatus, cx: &mut AsyncApp| {
this.update(cx, |_, cx| {
cx.emit(BackendEvent::PushProgress {
addr: addr.clone(),
status,
});
})?;
Ok::<(), Error>(())
};
let event =
announce_and_push_or_retract(&client, &signer, &state, announcement, cx, on_step)
.await?;
let announcement = Announcement::from_event(&event)
.ok_or_else(|| anyhow!("failed to parse announcement"))?;
@@ -635,19 +650,20 @@ impl Backend {
}
let owner = public_key.to_bech32().unwrap();
let addr = RepoAddr::new(public_key, repo_id.clone());
let servers = grasp_servers.clone();
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();
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))
}
})
@@ -660,28 +676,26 @@ 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 event = announce_repository_and_push(
&client,
&signer,
ann,
&repo_id,
&owner,
&servers,
state.refs.clone(),
state.head.clone(),
&path,
cx,
|path, base, owner, repo_id| Repo::open(path)?.push_all(base, owner, repo_id),
)
.await?;
let on_step = &mut |status: PushStepStatus, cx: &mut AsyncApp| {
this.update(cx, |_, cx| {
cx.emit(BackendEvent::PushProgress {
addr: addr.clone(),
status,
});
})?;
Ok::<(), Error>(())
};
let event =
announce_and_push_or_retract(&client, &signer, &state, ann, cx, on_step).await?;
if let Some(base) = servers.first().and_then(grasp_base_url) {
let url = format!("{base}/{owner}/{repo_id}.git");
@@ -754,9 +768,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();
@@ -774,74 +785,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.clone();
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)
})
}
@@ -1187,113 +1174,44 @@ fn build_announcement(
}
}
// Announce the repository, then stage the state event and push.
#[allow(clippy::too_many_arguments)]
async fn announce_repository_and_push<S>(
/// 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<S, P>(
client: &Client,
signer: &S,
state: &SyncState,
announcement: GitRepositoryAnnouncement,
repo_id: &str,
owner: &str,
servers: &[RelayUrl],
refs: Vec<(String, String)>,
head: Option<String>,
path: &Path,
cx: &mut AsyncApp,
push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error> + Send + 'static,
on_step: &mut P,
) -> Result<Event, Error>
where
S: AsyncGetPublicKey + AsyncSignEvent + Clone,
S: AsyncGetPublicKey + AsyncSignEvent + ?Sized,
P: FnMut(PushStepStatus, &mut AsyncApp) -> Result<(), Error>,
{
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::<Vec<String>>()
.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, on_step)
.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 {
crates/signed_state/src/lib.rs
+2
View File
@@ -10,6 +10,7 @@ mod push;
mod refresh;
mod repo;
mod repos;
mod state;
mod sync_status;
use std::path::{Path, PathBuf};
@@ -29,6 +30,7 @@ pub use repo::RepoStore;
pub use repos::RepoListStore;
pub use signed_git::{Nip34Binding, Nip34Kind};
use signed_nostr::NostrBackend;
pub use state::{GitPushMode, GitRef, PushStep, PushStepStatus, SyncState};
pub use sync_status::{CheckoutSyncStatus, SyncStatusStore};
pub fn init(
crates/signed_state/src/push.rs
+3 -96
View File
@@ -1,9 +1,5 @@
use std::path::Path;
use anyhow::Error;
use nostr::prelude::Url;
use nostr_sdk::prelude::*;
use signed_core::RepoState;
/// `ws://` grasp servers, like ngit, use `http(s)://<host>`.
pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option<String> {
@@ -50,16 +46,16 @@ pub struct GraspServer {
}
impl GraspServer {
fn ok(relay: RelayUrl) -> Self {
pub(crate) fn ok(relay: RelayUrl) -> Self {
Self {
relay,
reason: None,
}
}
fn failed(relay: RelayUrl, reason: impl Into<String>) -> Self {
pub(crate) fn failed(relay: &RelayUrl, reason: impl Into<String>) -> Self {
Self {
relay,
relay: relay.to_owned(),
reason: Some(reason.into()),
}
}
@@ -115,93 +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<S>(
client: &Client,
signer: &S,
repo_id: &str,
refs: &[(String, String)],
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.clone(), "no domain"));
continue;
};
let event = match RepoState::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.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()));
}
}
}
outcome
}
crates/signed_state/src/repo.rs
+30 -4
View File
@@ -8,8 +8,7 @@ use nostr::event::IntoEventBuilder;
use nostr_sdk::prelude::*;
use settings::{EventFetchingStrategy, SettingsStore};
use signed_core::{
Announcement, Deletions, Filters, GitEvent, PullRequest, RepoAddr, RepoState, RepoStatus,
filters,
Announcement, Deletions, Filters, GitEvent, PullRequest, RepoAddr, RepoStatus, filters,
};
use signed_git::{Nip34Binding, Repo};
@@ -19,6 +18,7 @@ use crate::checkouts::CheckoutsStore;
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::state::{GitRef, PushStepStatus, SyncState};
use crate::sync_status::SyncStatusStore;
pub struct RepoStore {
@@ -34,7 +34,7 @@ pub struct RepoStore {
pub loaded: bool,
pub head: Option<String>,
// Refs and timestamp of the latest state event, for sync detection.
pub state_refs: Vec<(String, String)>,
pub state_refs: Vec<GitRef>,
pub state_at: Option<Timestamp>,
pub issues: Vec<Event>,
pub patches: Vec<Event>,
@@ -48,6 +48,7 @@ pub struct RepoStore {
// The repository is out of sync on the rejected servers until republished.
pub last_push_warning: Option<String>,
pub pushing: bool,
pub push_steps: Vec<PushStepStatus>,
pub cloning: bool,
// Avoids re-subscribing and re-fetching on every refresh.
repo_relays: HashSet<RelayUrl>,
@@ -101,6 +102,7 @@ impl RepoStore {
last_warning: None,
last_push_warning: None,
pushing: false,
push_steps: Vec::new(),
cloning: false,
repo_relays: HashSet::new(),
root_fetches: HashSet::new(),
@@ -131,6 +133,7 @@ impl RepoStore {
last_warning: None,
last_push_warning: None,
pushing: false,
push_steps: Vec::new(),
cloning: false,
repo_relays: HashSet::new(),
root_fetches: HashSet::new(),
@@ -169,6 +172,14 @@ impl RepoStore {
let backend = Backend::global(cx);
cx.subscribe(&backend, |this, _backend, event, cx| {
if let BackendEvent::PushProgress { addr, status } = event {
if this.addr.as_ref() == Some(addr) {
this.apply_push_step(status.clone(), cx);
}
return;
}
let Some(addr) = this.addr.as_ref() else {
return;
};
@@ -203,6 +214,19 @@ impl RepoStore {
self.addr.as_ref()
}
fn apply_push_step(&mut self, status: PushStepStatus, cx: &mut Context<Self>) {
match self
.push_steps
.iter_mut()
.find(|entry| entry.step() == status.step())
{
Some(entry) => *entry = status,
None => self.push_steps.push(status),
}
cx.notify();
}
pub fn name(&self) -> SharedString {
self.announcement
.as_ref()
@@ -354,7 +378,7 @@ impl RepoStore {
let all_states = states.into_iter().filter(|e| !deletions.is_deleted(e));
let state =
utils::latest(all_states).map(|event| (RepoState::parse(&event), event.created_at));
utils::latest(all_states).map(|event| (SyncState::parse(&event), event.created_at));
let (mut issues, mut patches, mut pull_requests, mut statuses, mut comments) =
(Vec::new(), Vec::new(), Vec::new(), Vec::new(), Vec::new());
@@ -1186,6 +1210,7 @@ impl RepoStore {
cx: &mut Context<Self>,
) -> Task<Result<(), Error>> {
self.pushing = true;
self.push_steps.clear();
self.last_error = None;
self.last_push_warning = None;
cx.notify();
@@ -1195,6 +1220,7 @@ impl RepoStore {
this.update(cx, |this, cx| {
this.pushing = false;
this.push_steps.clear();
match &result {
Ok(outcome) => {
crates/signed_state/src/state.rs
+444
View File
@@ -0,0 +1,444 @@
use std::path::{Path, PathBuf};
use std::str::FromStr;
use anyhow::{Error, anyhow, bail};
use bitcoin_hashes::sha1::Hash as Sha1Hash;
use gpui::{AppContext, AsyncApp};
use nostr::prelude::*;
use nostr_sdk::prelude::*;
use signed_core::Announcement;
use signed_git::Repo;
use crate::push::{GraspServer, PushOutcome, grasp_base_url};
/// A git ref with the commit it points at.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GitRef {
/// Full ref name, like `refs/heads/main` or `refs/tags/v1.0`.
pub name: String,
pub commit: Sha1Hash,
}
impl GitRef {
pub fn new(name: impl Into<String>, commit: &str) -> Result<Self, Error> {
let commit = Sha1Hash::from_str(commit).map_err(|_| anyhow!("invalid commit: {commit}"))?;
Ok(Self {
name: name.into(),
commit,
})
}
}
/// How much to push to the grasp servers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum GitPushMode {
/// Mirror all branches and tags.
#[default]
All,
/// Push `refs/heads/main` only (repo creation).
Main,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncState {
pub refs: Vec<GitRef>,
pub head: Option<String>,
/// Local repository to push from; `None` for states parsed from events.
pub path: Option<PathBuf>,
pub mode: GitPushMode,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PushStep {
AnnounceRepository,
/// Publish the state event to one grasp server.
StageStateEvent(RelayUrl),
/// Git push to one grasp server.
PushObjects(RelayUrl),
/// Fan the state event out to the relays.
BroadcastStateEvent,
}
#[derive(Debug, Clone)]
pub enum PushStepStatus {
Running(PushStep),
Done(PushStep),
Failed(PushStep, String),
}
impl PushStep {
pub fn label(&self) -> String {
match self {
Self::AnnounceRepository => "Announce repository".to_owned(),
Self::StageStateEvent(relay) => format!("Publish state event -> {relay}"),
Self::BroadcastStateEvent => "Publish state event to relays".to_owned(),
Self::PushObjects(relay) => format!(
"Push objects -> {}",
grasp_base_url(relay).unwrap_or_else(|| relay.to_string())
),
}
}
}
impl PushStepStatus {
pub fn step(&self) -> &PushStep {
match self {
Self::Running(step) | Self::Done(step) | Self::Failed(step, _) => step,
}
}
}
impl SyncState {
pub fn build(id: &str, refs: &[GitRef], head: Option<&str>) -> EventBuilder {
let mut tags: Vec<Tag> = vec![Tag::identifier(id.to_owned())];
for git_ref in refs {
let commit = git_ref.commit.to_string();
tags.push(Tag::parse([git_ref.name.as_str(), commit.as_str()]).expect("valid ref tag"));
}
if let Some(head) = head {
tags.push(
Tag::parse(["HEAD", &format!("ref: refs/heads/{head}")]).expect("valid HEAD tag"),
);
}
EventBuilder::new(Kind::RepoState, "").tags(tags)
}
pub fn parse(event: &Event) -> Self {
let mut refs = Vec::new();
let mut head = None;
for tag in event.tags.iter() {
match Nip34Tag::parse(tag.as_slice()) {
Ok(Nip34Tag::Head(branch)) => head = Some(branch),
Ok(Nip34Tag::RefHead { branch, commit }) => {
refs.push(GitRef {
name: format!("refs/heads/{branch}"),
commit,
});
}
Ok(Nip34Tag::RefTag { name, commit }) => {
refs.push(GitRef {
name: format!("refs/tags/{name}"),
commit,
});
}
_ => {}
}
}
Self {
refs,
head,
path: None,
mode: GitPushMode::default(),
}
}
/// Reads the refs and head of a local repository, ready to push.
pub fn from_repo(path: &Path) -> Result<Self, Error> {
let repo = Repo::open(path)?;
let state = repo.ref_state()?;
let refs = state
.refs
.iter()
.map(|(name, commit)| GitRef::new(name, commit))
.collect::<Result<Vec<_>, Error>>()?;
Ok(Self {
refs,
head: state.head,
path: Some(path.to_path_buf()),
mode: GitPushMode::default(),
})
}
pub fn with_mode(mut self, mode: GitPushMode) -> Self {
self.mode = mode;
self
}
/// State event for these refs; the caller signs it.
pub fn state_event(&self, repo_id: &str) -> EventBuilder {
Self::build(repo_id, &self.refs, self.head.as_deref())
}
/// Push the state event and objects to the grasp servers,
/// calling `on_step` for each step.
pub async fn push_to_grasps<P>(
&self,
client: &Client,
event: Event,
announcement: &Announcement,
cx: &mut AsyncApp,
on_step: &mut P,
) -> Result<PushOutcome, Error>
where
P: FnMut(PushStepStatus, &mut AsyncApp) -> Result<(), Error>,
{
let path = self
.path
.as_deref()
.ok_or_else(|| anyhow!("push requires a local repository"))?;
let owner = announcement
.owner
.to_bech32()
.map_err(|error| anyhow!("could not encode the repository owner: {error}"))?;
let mut outcome = PushOutcome::default();
for relay in &announcement.relays {
let step = PushStep::StageStateEvent(relay.clone());
// The relay must accept the state event before the git push is allowed.
on_step(PushStepStatus::Running(step.clone()), cx)?;
// Get base URL for the git push
let base = grasp_base_url(relay).unwrap_or_default();
// Ensure the relay is connected before sending the event.
client.add_relay(relay).and_connect().await?;
let staging = client.send_event(&event).to([relay.clone()]).await;
// Whether the relay accepted the event
let sent_failed = !staging
.as_ref()
.is_ok_and(|output| output.success.contains_key(relay));
if sent_failed {
let reason = match &staging {
Ok(output) => output
.failed
.get(relay)
.cloned()
.unwrap_or_else(|| "relay did not confirm the event".to_owned()),
Err(e) => format!("could not send the state event to {relay}: {e}"),
};
// Run the callback to report the failure
on_step(PushStepStatus::Failed(step, reason.clone()), cx)?;
// Record the failure in the outcome and continue to the next relay
outcome.servers.push(GraspServer::failed(relay, reason));
continue;
}
// Run the callback to report the success
on_step(PushStepStatus::Done(step), cx)?;
// Run the git push in the background, reporting progress through the callback
let step = PushStep::PushObjects(relay.clone());
on_step(PushStepStatus::Running(step.clone()), cx)?;
let pushed = cx.background_spawn({
let path = path.to_path_buf();
let base = base.clone();
let owner = owner.clone();
let repo_id = announcement.id.clone();
let mode = self.mode;
async move {
let repo = Repo::open(&path)?;
match mode {
GitPushMode::Main => repo.push_main(&base, &owner, &repo_id),
GitPushMode::All => repo.push_all(&base, &owner, &repo_id),
}
}
});
match pushed.await {
Ok(()) => {
// Run the callback to report the success
on_step(PushStepStatus::Done(step), cx)?;
outcome.state_event = Some(event.clone());
outcome.servers.push(GraspServer::ok(relay.clone()));
}
Err(e) => {
// Run the callback to report the failure
on_step(PushStepStatus::Failed(step, e.to_string()), cx)?;
outcome
.servers
.push(GraspServer::failed(relay, e.to_string()));
}
}
}
// Fan the state out to the relays once a git server holds the objects.
if outcome.state_event.is_some() {
let step = PushStep::BroadcastStateEvent;
on_step(PushStepStatus::Running(step.clone()), cx)?;
if let Err(e) = client.send_event(&event).broadcast().await {
on_step(PushStepStatus::Failed(step, e.to_string()), cx)?;
log::warn!("failed to broadcast repository state: {e}");
} else {
on_step(PushStepStatus::Done(step), cx)?;
}
}
Ok(outcome)
}
/// Announce the repository and push the state and objects to the grasp servers.
///
/// Both events are signed by the caller, so a failure at the call site can retract the announcement by id.
pub async fn announce_and_push<P>(
&self,
client: &Client,
announcement: Event,
state_event: Event,
cx: &mut AsyncApp,
on_step: &mut P,
) -> Result<Event, Error>
where
P: FnMut(PushStepStatus, &mut AsyncApp) -> Result<(), Error>,
{
let step = PushStep::AnnounceRepository;
on_step(PushStepStatus::Running(step.clone()), cx)?;
let published = Announcement::from_event(&announcement)
.ok_or_else(|| anyhow!("failed to parse announcement"))?;
// Ensure the relays are connected before sending the announcement event.
for url in &published.relays {
client.add_relay(url).and_connect().await.ok();
}
let output = client.send_event(&announcement).broadcast().await?;
// Handle the case where no relay accepted the announcement event.
if output.success.is_empty() {
let reasons = output
.failed
.values()
.cloned()
.collect::<Vec<String>>()
.join(", ");
// Run the callback to report the failure
let step = PushStepStatus::Failed(step, format!("event not accepted: {reasons}"));
on_step(step, cx)?;
bail!("event not accepted: {reasons}");
}
on_step(PushStepStatus::Done(step), cx)?;
let outcome = if self.refs.is_empty() {
PushOutcome::default()
} else {
self.push_to_grasps(client, state_event, &published, cx, on_step)
.await?
};
if outcome.accepted() == 0 {
bail!(
"Repository was announced, but the push to grasp servers failed: {}",
outcome.failure_summary()
);
}
Ok(announcement)
}
/// Retracts a live announcement; call when `announce_and_push` fails.
pub async fn retract_announcement<S>(
client: &Client,
signer: &S,
id: EventId,
) -> Result<(), Error>
where
S: AsyncGetPublicKey + AsyncSignEvent + ?Sized,
{
let builder = EventDeletionRequest::new().id(id);
let deletion = builder.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(())
}
}
#[cfg(test)]
mod tests {
use super::*;
const COMMIT_A: &str = "aa231c4c6a5777dc89b42207b499891a344add5c";
const COMMIT_B: &str = "59429cfc6cb35b0a1ddace73b5a5c5ed57b8f5ca";
fn git_ref(name: &str, commit: &str) -> GitRef {
GitRef::new(name, commit).expect("valid ref")
}
fn keys() -> Keys {
Keys::new(
SecretKey::from_hex("0000000000000000000000000000000000000000000000000000000000000001")
.expect("valid secret key"),
)
}
fn state_event(tags: &[&[&str]]) -> Event {
let tags: Vec<Tag> = tags
.iter()
.map(|t| Tag::parse(t.to_vec()).expect("valid tag"))
.collect();
EventBuilder::new(Kind::RepoState, "")
.tags(tags)
.finalize(&keys())
.expect("signed event")
}
#[test]
fn parses_heads_and_tags() {
let event = state_event(&[
&["HEAD", "ref: refs/heads/main"],
&["refs/heads/main", COMMIT_A],
&["refs/heads/dev", COMMIT_B],
&["refs/tags/v1.0", COMMIT_A],
]);
let state = SyncState::parse(&event);
assert_eq!(state.head.as_deref(), Some("main"));
assert_eq!(
state.refs,
vec![
git_ref("refs/heads/main", COMMIT_A),
git_ref("refs/heads/dev", COMMIT_B),
git_ref("refs/tags/v1.0", COMMIT_A),
]
);
}
#[test]
fn build_round_trips_through_parse() {
let refs = vec![
git_ref("refs/heads/main", COMMIT_A),
git_ref("refs/heads/dev", COMMIT_B),
git_ref("refs/tags/v1.0", COMMIT_A),
];
let event = SyncState::build("my-repo", &refs, Some("main"))
.finalize(&keys())
.expect("signed event");
assert_eq!(event.kind, Kind::RepoState);
assert_eq!(event.tags.identifier().as_deref(), Some("my-repo"));
let state = SyncState::parse(&event);
assert_eq!(state.refs, refs);
assert_eq!(state.head.as_deref(), Some("main"));
}
}
crates/signed_state/src/sync_status.rs
+10 -4
View File
@@ -5,13 +5,14 @@ use std::time::Duration;
use anyhow::Error;
use gpui::{App, AppContext, Context, Entity, Global, Subscription};
use nostr_sdk::prelude::*;
use signed_core::{Deletions, Filters, RepoAddr, RepoState};
use signed_core::{Deletions, Filters, RepoAddr};
use signed_git::{Repo, RepoSyncStatus};
use crate::backend::{Backend, BackendEvent};
use crate::checkouts::CheckoutsStore;
use crate::refresh::{RefreshGate, RefreshRequest};
use crate::repos::RepoListStore;
use crate::state::{GitRef, SyncState};
const SYNC_POLL: Duration = Duration::from_secs(180);
const MAX_SYNC_CHECKOUTS: usize = 8;
@@ -167,7 +168,7 @@ impl SyncStatusStore {
let deletion_events = db.query(Filters::deletions()).await?;
let deletions = Deletions::from_events(deletion_events);
let mut states: HashMap<RepoAddr, Vec<(String, String)>> = HashMap::new();
let mut states: HashMap<RepoAddr, Vec<GitRef>> = HashMap::new();
for (addr, _) in &targets {
let events = db.query(addr.state_filter()).await?;
let latest = utils::latest(
@@ -178,7 +179,7 @@ impl SyncStatusStore {
let Some(event) = latest else {
continue;
};
states.insert(addr.clone(), RepoState::parse(&event).refs);
states.insert(addr.clone(), SyncState::parse(&event).refs);
}
let mut statuses: HashMap<RepoAddr, Vec<CheckoutSyncStatus>> = HashMap::new();
@@ -187,10 +188,15 @@ impl SyncStatusStore {
continue;
};
let remote_refs: Vec<(String, String)> = nostr_refs
.iter()
.map(|git_ref| (git_ref.name.clone(), git_ref.commit.to_string()))
.collect();
let mut list = Vec::new();
for path in paths.iter().take(MAX_SYNC_CHECKOUTS) {
let synced =
Repo::try_open(path).and_then(|repo| repo.sync_status(nostr_refs).ok());
Repo::try_open(path).and_then(|repo| repo.sync_status(&remote_refs).ok());
if let Some(status) = synced {
list.push(CheckoutSyncStatus {
crates/workspace/src/views/dialog_state.rs
+83 -2
View File
@@ -1,12 +1,19 @@
use gpui::prelude::*;
use gpui::{AnyElement, App, SharedString, div};
use gpui_component::ActiveTheme;
use gpui::{AnyElement, App, Context, SharedString, Subscription, div};
use gpui_component::spinner::Spinner;
use gpui_component::{ActiveTheme, Icon, IconName, Sizable, h_flex, v_flex};
use signed_core::RepoAddr;
use signed_state::{Backend, BackendEvent, PushStepStatus};
/// Progress of an async dialog action: a busy flag that disables the form and an error shown below it.
#[derive(Debug, Default)]
pub struct DialogProgress {
pub busy: bool,
pub error: Option<SharedString>,
// The repository whose push checklist is shown; `None` outside a publish.
push_addr: Option<RepoAddr>,
pub push_steps: Vec<PushStepStatus>,
_push_subscription: Option<Subscription>,
}
impl DialogProgress {
@@ -16,6 +23,37 @@ impl DialogProgress {
self.error = None;
}
/// Starts a publish for the repository at `addr` and listens for its push steps.
pub fn begin_push(&mut self, addr: RepoAddr, cx: &mut Context<Self>) {
self.begin();
self.push_addr = Some(addr);
self.push_steps.clear();
if self._push_subscription.is_none() {
let backend = Backend::global(cx);
self._push_subscription = Some(cx.subscribe(&backend, |this, _backend, event, cx| {
let BackendEvent::PushProgress { addr, status } = event else {
return;
};
if this.push_addr.as_ref() != Some(addr) {
return;
}
match this
.push_steps
.iter_mut()
.find(|step| step.step() == status.step())
{
Some(step) => *step = status.clone(),
None => this.push_steps.push(status.clone()),
}
cx.notify();
}));
}
}
/// Marks an action as failed, enabling the form and showing `message`.
pub fn fail(&mut self, message: impl Into<SharedString>) {
self.busy = false;
@@ -33,3 +71,46 @@ pub fn error_row(error: &Option<SharedString>, cx: &App) -> Option<AnyElement> {
.into_any_element()
})
}
/// The push checklist of a running publish, `None` before any step is reported.
pub fn push_steps_row(steps: &[PushStepStatus], cx: &App) -> Option<AnyElement> {
if steps.is_empty() {
return None;
}
let rows = steps
.iter()
.map(|status| {
let label = SharedString::from(status.step().label());
match status {
PushStepStatus::Running(_) => h_flex()
.gap_2()
.text_sm()
.text_color(cx.theme().muted_foreground)
.child(Spinner::new().small())
.child(label)
.into_any_element(),
PushStepStatus::Done(_) => h_flex()
.gap_2()
.text_sm()
.text_color(cx.theme().foreground)
.child(Icon::new(IconName::CircleCheck).small())
.child(label)
.into_any_element(),
PushStepStatus::Failed(_, reason) => h_flex()
.gap_2()
.text_sm()
.text_color(cx.theme().danger)
.child(Icon::new(IconName::Close).small())
.child(SharedString::from(format!(
"{}: {reason}",
status.step().label()
)))
.into_any_element(),
}
})
.collect::<Vec<_>>();
Some(v_flex().gap_1().children(rows).into_any_element())
}
crates/workspace/src/views/repo/init_dialog.rs
+11 -3
View File
@@ -12,12 +12,13 @@ use gpui_component::input::{Input, InputState, Textarea};
use gpui_component::select::{Select, SelectState};
use gpui_component::{ActiveTheme, Disableable, WindowExt};
use settings::SettingsStore;
use signed_core::RepoAddr;
use signed_git::Repo;
use signed_state::Backend;
use signed_ui::SelectOption;
use super::RepoDetailView;
use crate::views::dialog_state::{DialogProgress, error_row};
use crate::views::dialog_state::{DialogProgress, error_row, push_steps_row};
use crate::views::sidebar::grasp_servers::{
GraspServersState, grasp_servers_field, load_user_grasp_servers,
};
@@ -120,6 +121,7 @@ pub fn open(
)
.child(grasp_servers_field(&grasp_state, &relay_input, cx)),
)
.children(push_steps_row(&state.read(cx).push_steps, cx))
.children(error_row(&error, cx))
.child(
DialogFooter::new().justify_end().child(
@@ -189,9 +191,15 @@ fn init_repository(
return;
}
state.update(cx, |state, _| state.begin());
let backend = Backend::global(cx);
let Some(user) = backend.read(cx).current_user() else {
state.update(cx, |state, _| state.fail("Sign in to publish a repository"));
return;
};
let addr = RepoAddr::new(user, RepoAddr::identifier_from_name(&name));
state.update(cx, |state, cx| state.begin_push(addr, cx));
let task = backend.update(cx, |backend, cx| {
backend.publish_local_repo(
local_path.clone(),
crates/workspace/src/views/repo/mod.rs
+58 -1
View File
@@ -26,7 +26,8 @@ use signed_core::{Announcement, RepoAddr, RepoStatus};
use signed_git::{FileCommit, GitCache, RefSync, Repo};
use signed_state::{
Backend, CheckoutStatus, CheckoutSyncStatus, CheckoutsStore, LocalReposStore, Mirrors,
Nip34Binding, Nip34Kind, ProfileStore, RepoListStore, RepoStore, SyncStatusStore,
Nip34Binding, Nip34Kind, ProfileStore, PushStepStatus, RepoListStore, RepoStore,
SyncStatusStore,
};
use signed_ui::{
Avatar, CountBadge, DropdownButton, PixelAvatar, copy_row, menu_copy_row, ref_selector_trigger,
@@ -1747,6 +1748,59 @@ impl RepoDetailView {
)
}
pub(super) fn render_push_progress_banner(&self, cx: &Context<Self>) -> Option<AnyElement> {
let store = self.store.read(cx);
if !store.pushing || store.push_steps.is_empty() {
return None;
}
let steps = store
.push_steps
.iter()
.map(|status| {
let label = SharedString::from(status.step().label());
match status {
PushStepStatus::Running(_) => h_flex()
.gap_2()
.text_sm()
.text_color(cx.theme().muted_foreground)
.child(Spinner::new().small())
.child(label)
.into_any_element(),
PushStepStatus::Done(_) => h_flex()
.gap_2()
.text_sm()
.text_color(cx.theme().foreground)
.child(Icon::new(IconName::CircleCheck).small())
.child(label)
.into_any_element(),
PushStepStatus::Failed(_, reason) => h_flex()
.gap_2()
.text_sm()
.text_color(cx.theme().danger)
.child(Icon::new(IconName::Close).small())
.child(SharedString::from(format!(
"{}: {reason}",
status.step().label()
)))
.into_any_element(),
}
})
.collect::<Vec<_>>();
Some(
v_flex()
.p_4()
.gap_1()
.w_full()
.bg(cx.theme().muted)
.child(div().text_sm().child("Pushing"))
.children(steps)
.into_any_element(),
)
}
pub(super) fn render_push_warning_banner(&self, cx: &Context<Self>) -> Option<AnyElement> {
let store = self.store.read(cx);
let warning = store.last_push_warning.clone()?;
@@ -2189,6 +2243,9 @@ impl Render for RepoDetailView {
.id("repo")
.size_full()
.when_some(banner, |this, banner| this.child(banner))
.when_some(self.render_push_progress_banner(cx), |this, banner| {
this.child(banner)
})
.when_some(self.render_push_warning_banner(cx), |this, banner| {
this.child(banner)
})
crates/workspace/src/views/sidebar/create_repo_dialog.rs
+11 -3
View File
@@ -10,11 +10,12 @@ use gpui_component::form::{field, v_form};
use gpui_component::input::{Input, InputState, Textarea};
use gpui_component::{Disableable, IconName, WindowExt, h_flex};
use settings::SettingsStore;
use signed_core::RepoAddr;
use signed_state::{Backend, CheckoutsStore};
use super::super::open_repo_panel;
use super::grasp_servers::{GraspServersState, grasp_servers_field, load_user_grasp_servers};
use crate::views::dialog_state::{DialogProgress, error_row};
use crate::views::dialog_state::{DialogProgress, error_row, push_steps_row};
/// Progress of the create-repository flow, so async results can be rendered.
pub type CreateRepoState = DialogProgress;
@@ -112,6 +113,7 @@ pub fn open(dock_area: WeakEntity<DockArea>, window: &mut Window, cx: &mut App)
)
.child(grasp_servers_field(&grasp_state, &relay_input, cx)),
)
.children(push_steps_row(&state.read(cx).push_steps, cx))
.children(error_row(&error, cx))
.child(
DialogFooter::new().justify_end().child(
@@ -209,9 +211,15 @@ fn create_repository(
return;
}
state.update(cx, |state, _| state.begin());
let backend = Backend::global(cx);
let Some(user) = backend.read(cx).current_user() else {
state.update(cx, |state, _| state.fail("Sign in to create a repository"));
return;
};
let addr = RepoAddr::new(user, RepoAddr::identifier_from_name(&name));
state.update(cx, |state, cx| state.begin_push(addr, cx));
let task = backend.update(cx, |backend, cx| {
backend.create_repository(&name, &description, folder, servers, cx)
});