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

fix: repository sync (#29)
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: #29
This commit was merged in pull request #29.
This commit is contained in:
reya
2026-10-13 08:58:14 +00:00
parent 7878feb236
commit b78418d815
13 changed files with 499 additions and 997 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+1
View File
@@ -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.
crates/signed_core/src/deletions.rs
-2
View File
@@ -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));
}
}
crates/signed_git/src/repo.rs
+7 -57
View File
@@ -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;
@@ -62,16 +62,13 @@ impl Repo {
pub fn init(path: &Path, name: &str, description: &str) -> Result<String> {
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(),
@@ -1192,11 +1189,13 @@ impl Repo {
],
..Default::default()
};
self.inner
.find_remote("origin")?
.connect(gix::remote::Direction::Fetch)?
.prepare_fetch(Discard, options)?
.receive(Discard, &IS_INTERRUPTED)?;
Ok(())
}
@@ -1263,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<bool> {
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::<Result<Vec<_>, _>>()
.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<String, String> = 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() {
crates/signed_git/src/tests.rs
+47 -25
View File
@@ -56,39 +56,61 @@ fn push_all_mirrors_branches_and_tags() {
}
#[test]
fn remote_has_refs_reports_whether_pushed_refs_landed() {
let server = tempfile::tempdir().unwrap();
bare_server(server.path(), "npub1test", "my-repo");
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();
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"));
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"]);
repo.push_all(
&format!("file://{}", server.path().display()),
"npub1test",
"my-repo",
)
.expect("push");
let listing = |repo: &Repo| {
let mut refs = repo.ref_state().expect("ref state").refs;
refs.sort();
refs
};
assert!(repo.remote_has_refs(&url, &expected).expect("probe"));
// Loose refs.
let loose = listing(&Repo::open(dir).expect("open"));
let stale = vec![("refs/heads/main".to_owned(), "0".repeat(40))];
assert!(!repo.remote_has_refs(&url, &stale).expect("probe"));
// Packed refs.
git_run(dir, &["pack-refs", "--all"]);
let packed = listing(&Repo::open(dir).expect("open"));
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"));
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::<Vec<_>>();
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]
crates/signed_state/src/backend.rs
+116 -76
View File
@@ -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::*;
@@ -21,7 +19,7 @@ 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};
use crate::repos::RepoListStore;
pub const USER_KEYRING: &str = "Signed Safe Storage";
@@ -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,
@@ -452,16 +450,28 @@ 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?;
if output.success.is_empty() {
let reasons = output
.failed
.values()
.cloned()
.collect::<Vec<String>>()
.join(", ");
bail!("event not accepted by any relay: {reasons}");
}
Ok::<(), anyhow::Error>(())
})
.detach();
}
@@ -527,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();
@@ -565,7 +575,6 @@ impl Backend {
);
let event = announce_repository_and_push(
&this,
&client,
&signer,
announcement,
@@ -631,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();
@@ -660,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,
@@ -792,29 +800,29 @@ 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();
let relays = relays.clone();
let refs = refs.clone();
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,
|path, base, owner, repo_id| {
Repo::open(path)?.push_all(base, owner, repo_id)
},
)
.await
}
});
push.await
@@ -1136,17 +1144,22 @@ impl Backend {
// Each target gets its own deletion event.
fn retract_events(&mut self, events: &[Event], cx: &mut Context<Self>) {
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?;
let output = client.send_event(&deletion).broadcast().await?;
if output.success.is_empty() {
bail!("deletion event not accepted by any relay");
}
Ok::<(), Error>(())
})
.detach();
.detach_and_log_err(cx);
}
}
}
@@ -1176,10 +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(
backend: &WeakEntity<Backend>,
async fn announce_repository_and_push<S>(
client: &Client,
signer: &UniversalSigner,
signer: &S,
announcement: GitRepositoryAnnouncement,
repo_id: &str,
owner: &str,
@@ -1189,59 +1201,89 @@ async fn announce_repository_and_push(
path: &Path,
cx: &mut AsyncApp,
push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error> + Send + 'static,
) -> Result<Event, Error> {
) -> Result<Event, Error>
where
S: AsyncGetPublicKey + AsyncSignEvent + Clone,
{
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 = {
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)?
};
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 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 = 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,
&executor,
push,
)
.await
})
.await
})
};
push.await
};
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.
@@ -1292,15 +1334,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())),
}
}
crates/signed_state/src/bootstrap.rs
+2 -8
View File
@@ -78,16 +78,10 @@ pub async fn user_grasp_list_servers(
) -> Result<Vec<RelayUrl>, Error> {
let events: Vec<Event> = 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())
}
crates/signed_state/src/checkouts.rs
+64 -203
View File
@@ -45,6 +45,53 @@ struct Remembered {
last_used: u64,
}
impl Remembered {
fn resolve_associations<'a, A>(
remembered: &[Remembered],
scanned: &[(PathBuf, Option<String>, Option<String>)],
announcements: A,
) -> HashMap<RepoAddr, Vec<PathBuf>>
where
A: IntoIterator<Item = &'a Announcement>,
{
let announcements: Vec<&Announcement> = announcements.into_iter().collect();
let mut out: HashMap<RepoAddr, Vec<PathBuf>> = 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<RepoAddr, Vec<PathBuf>>,
statuses: HashMap<RepoAddr, Vec<CheckoutStatus>>,
@@ -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<RepoAddr, Vec<PathBuf>> = associations
.into_iter()
@@ -435,7 +485,7 @@ impl CheckoutsStore {
Ok::<_, Error>((statuses, push_statuses))
});
let task: gpui::Task<Result<(), Error>> = 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<String>, Option<String>)],
announcements: impl IntoIterator<Item = &'a Announcement>,
) -> HashMap<RepoAddr, Vec<PathBuf>> {
let announcements: Vec<&Announcement> = announcements.into_iter().collect();
let mut out: HashMap<RepoAddr, Vec<PathBuf>> = 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<CheckoutStatus> {
@@ -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<CheckoutStatus> {
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);
}
}
crates/signed_state/src/lib.rs
+1
View File
@@ -4,6 +4,7 @@ mod checkouts;
mod git_store;
mod inbox;
mod local_repos;
mod patch_series;
mod profile;
mod push;
mod refresh;
crates/signed_state/src/patch_series.rs
+50
View File
@@ -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<String>,
}
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<usize> {
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<Sha1Hash> {
self.parts
.last()
.and_then(|part| patch_current_commit(part))
.and_then(|hex| hex.parse::<Sha1Hash>().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 <commit>` 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)
}
crates/signed_state/src/push.rs
+78 -399
View File
@@ -1,25 +1,18 @@
use std::path::Path;
use std::time::Duration;
use anyhow::{Error, bail};
use gpui::BackgroundExecutor;
use nostr::event::IntoEventBuilder;
use anyhow::Error;
use nostr::prelude::Url;
use nostr_sdk::prelude::*;
use signed_core::RepoState;
use signed_git::Repo;
use signed_nostr::UniversalSigner;
pub(crate) const GRASP_PUSH_ATTEMPTS: usize = 3;
const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1);
/// `ws://` grasp servers, like ngit, use `http(s)://<host>`.
pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option<String> {
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 +103,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<String> {
let accepted = self.accepted();
if self.servers.is_empty() || accepted == self.servers.len() {
@@ -122,407 +115,93 @@ impl PushOutcome {
self.failure_summary()
))
}
}
// 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.
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))
}
// 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.
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: Event) {
if state_event
.as_ref()
.is_none_or(|current| event.created_at > current.created_at)
{
*state_event = Some(event);
}
}
#[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<Event, Error> {
if output.success.is_empty() && !output.failed.is_empty() {
let reasons = output
.failed
.values()
.cloned()
.collect::<Vec<String>>()
.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}");
/// 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);
}
}
}
pub(crate) async fn publish_one(&self, builder: EventBuilder) -> Result<Event, Error> {
let event = builder.finalize_async(&self.signer).await?;
self.send_accepted(event).await
}
#[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();
pub(crate) async fn send_accepted(&self, event: Event) -> Result<Event, Error> {
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)
let event = match RepoState::build(repo_id, refs, head)
.finalize_async(signer)
.await
.map_err(|e| format!("could not sign the state event: {e}"))?;
Ok((event, created_at))
}
// `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 {
{
Ok(event) => event,
Err(e) => {
let reason = format!("could not sign the state event: {e}");
outcome
.servers
.push(GraspServer::failed(relay.clone(), "no domain"));
.push(GraspServer::failed(relay.clone(), reason));
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;
}
let (event, created_at) = match self
.sign_state_event(repo_id, refs, head, last_created_at)
.await
{
Ok(signed) => signed,
Err(e) => {
reason = Some(e);
break 'server;
}
};
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;
}
}
}
};
// 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);
}
reason = None;
}
// 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}")),
};
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())),
}
if let Err(reason) = staging {
log::warn!("grasp push failed: {relay}: {reason}");
outcome
.servers
.push(GraspServer::failed(relay.clone(), reason));
continue;
}
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<_>>(),
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)"
));
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
+73 -173
View File
@@ -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,19 +11,16 @@ use signed_core::{
Announcement, Deletions, Filters, GitEvent, PullRequest, RepoAddr, RepoState, RepoStatus,
filters,
};
use signed_git::{Nip34Binding, PatchParser, Repo};
use signed_nostr::UniversalSigner;
use signed_git::{Nip34Binding, Repo};
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::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<RepoAddr>,
// Seeded from the open-time hint, replaced by the database's latest on the
@@ -865,10 +861,22 @@ impl RepoStore {
}
}
let publish_result = {
let pusher = GraspPush::new(client.clone(), signer.clone());
pusher.send_accepted(event).await
};
let publish_result: Result<Event, Error> = async {
let output = client.send_event(&event).broadcast().await?;
if output.success.is_empty() && !output.failed.is_empty() {
let reasons = output
.failed
.values()
.cloned()
.collect::<Vec<String>>()
.join(", ");
bail!("event not accepted by any relay: {reasons}");
}
Ok(event)
}
.await;
let pr_event = match publish_result {
Ok(event) => event,
@@ -1041,10 +1049,23 @@ impl RepoStore {
}
};
let publish_result = {
let pusher = GraspPush::new(client.clone(), signer.clone());
pusher.publish_one(builder).await
};
let publish_result: Result<Event, Error> = async {
let event = builder.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::<Vec<String>>()
.join(", ");
bail!("event not accepted by any relay: {reasons}");
}
Ok(event)
}
.await;
if let Err(e) = publish_result {
return this.update(cx, |this, cx| {
@@ -1297,8 +1318,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>) {
self.last_error = None;
@@ -1309,8 +1329,23 @@ impl RepoStore {
};
let task: Task<Result<(), Error>> = cx.spawn(async move |this, cx| {
let pusher = GraspPush::new(client, signer);
let publish_result = pusher.publish_one(builder).await;
let publish_result: Result<Event, Error> = async {
let event = builder.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::<Vec<String>>()
.join(", ");
bail!("event not accepted by any relay: {reasons}");
}
Ok(event)
}
.await;
if let Err(e) = publish_result {
this.update(cx, |this, cx| {
@@ -1363,65 +1398,21 @@ fn resolve_statuses(
.collect()
}
// The `From <commit>` 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<String>,
}
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<usize> {
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<Sha1Hash> {
self.parts
.last()
.and_then(|part| patch_current_commit(part))
.and_then(|hex| hex.parse::<Sha1Hash>().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(
async fn publish_patch_series<S>(
client: &Client,
signer: &UniversalSigner,
signer: &S,
addr: &RepoAddr,
owner: PublicKey,
euc: Option<&str>,
series: &PatchSeries,
first_marker: &str,
reply_to: Option<EventId>,
) -> Result<Event, Error> {
let pusher = GraspPush::new(client.clone(), signer.clone());
) -> Result<Event, Error>
where
S: AsyncGetPublicKey + AsyncSignEvent + ?Sized,
{
let mut root: Option<Event> = None;
let mut previous = reply_to;
@@ -1468,7 +1459,18 @@ 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?;
if output.success.is_empty() && !output.failed.is_empty() {
let reasons = output
.failed
.values()
.cloned()
.collect::<Vec<String>>()
.join(", ");
bail!("event not accepted by any relay: {reasons}");
}
if root.is_none() {
root = Some(event.clone());
@@ -1503,105 +1505,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 <a@b.c>\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<PublicKey> {
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::<HashSet<_>>(),
expected
);
}
}
crates/signed_state/src/sync_status.rs
+4 -3
View File
@@ -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<SyncStatusStore>);
@@ -229,7 +228,9 @@ impl SyncStatusStore {
})?;
if again {
this.update(cx, |this, cx| this.refresh(cx))?;
this.update(cx, |this, cx| {
this.refresh(cx);
})?;
}
Ok::<_, Error>(())
crates/workspace/src/views/repo/mod.rs
+56 -51
View File
@@ -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<Url> = 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<SharedString> =
snapshot.branches.iter().map(Into::into).collect();
let tags: Vec<SharedString> = 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<SharedString> =
snapshot.branches.iter().map(Into::into).collect();
let tags: Vec<SharedString> =
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(())
});