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

fix sync

Browse Source
This commit is contained in:
reya
2026-10-13 09:05:51 +07:00
parent 7878feb236
commit 4a57a01f4a
7 changed files with 394 additions and 465 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_git/src/repo.rs
+2
View File
@@ -1192,11 +1192,13 @@ impl Repo {
],
..Default::default()
};
self.inner
.find_remote("origin")?
.connect(gix::remote::Direction::Fetch)?
.prepare_fetch(Discard, options)?
.receive(Discard, &IS_INTERRUPTED)?;
Ok(())
}
crates/signed_git/src/tests.rs
+58
View File
@@ -55,6 +55,64 @@ fn push_all_mirrors_branches_and_tags() {
assert!(refs.contains("refs/tags/v1.0"));
}
#[test]
fn ref_state_matches_the_git_cli_ref_listing() {
let (dir, repo) = fixture(&[("a.txt", b"one")]);
commit_all(&repo, "initial");
let dir = dir.path();
git_run(dir, &["branch", "browse-group"]);
git_run(dir, &["checkout", "-q", "-b", "new-search"]);
std::fs::write(dir.join("b.txt"), b"two").expect("write");
commit_all(&repo, "search work");
git_run(dir, &["checkout", "-q", "-"]);
git_run(dir, &["tag", "v1"]);
let listing = |repo: &Repo| {
let mut refs = repo.ref_state().expect("ref state").refs;
refs.sort();
refs
};
// Loose refs.
let loose = listing(&Repo::open(dir).expect("open"));
// Packed refs.
git_run(dir, &["pack-refs", "--all"]);
let packed = listing(&Repo::open(dir).expect("open"));
let output = git_run(
dir,
&[
"for-each-ref",
"--format=%(refname) %(objectname)",
"refs/heads",
"refs/tags",
],
);
let mut cli = String::from_utf8_lossy(&output.stdout)
.lines()
.map(|line| {
let (name, oid) = line.split_once(' ').expect("for-each-ref format");
(name.to_owned(), oid.to_owned())
})
.collect::<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]
fn remote_has_refs_reports_whether_pushed_refs_landed() {
let server = tempfile::tempdir().unwrap();
crates/signed_state/src/backend.rs
+52 -40
View File
@@ -21,7 +21,9 @@ use crate::bootstrap::{
};
use crate::git_store::Mirrors;
use crate::inbox::Inbox;
use crate::push::{GraspPush, PushOutcome, grasp_base_url, grasp_clone_url};
use crate::push::{
PushOutcome, grasp_base_url, grasp_clone_url, push_staged_to_grasps, require_relay_accepted,
};
use crate::repos::RepoListStore;
pub const USER_KEYRING: &str = "Signed Safe Storage";
@@ -452,16 +454,19 @@ impl Backend {
.map(|url| RelayUrl::parse(url).expect("valid relay URL"))
.collect();
let pusher = GraspPush::new(this.client.clone(), this.signer.clone());
for builder in [
RelayList::new(relays).into_event_builder(),
metadata,
GitUserGraspList { grasp_servers }.into_event_builder(),
] {
let pusher = pusher.clone();
let client = this.client.clone();
let signer = this.signer.clone();
cx.background_spawn(async move {
pusher.publish_best_effort(builder).await;
let event = builder.finalize_async(&signer).await?;
let output = client.send_event(&event).broadcast().await?;
require_relay_accepted(output, event)?;
Ok::<(), anyhow::Error>(())
})
.detach();
}
@@ -792,7 +797,8 @@ impl Backend {
PushOutcome::default()
} else {
let push = cx.background_spawn({
let pusher = GraspPush::new(client.clone(), signer.clone());
let client = client.clone();
let signer = signer.clone();
let path = path.clone();
let owner = owner.clone();
let repo_id = repo_id.clone();
@@ -801,20 +807,21 @@ impl Backend {
let head = head.clone();
let executor = cx.background_executor().clone();
async move {
pusher
.push_staged_to_grasps(
&repo_id,
&refs,
head.as_deref(),
&path,
&owner,
&relays,
&executor,
|path, base, owner, repo_id| {
Repo::open(path)?.push_all(base, owner, repo_id)
},
)
.await
push_staged_to_grasps(
&client,
&signer,
&repo_id,
&refs,
head.as_deref(),
&path,
&owner,
&relays,
&executor,
|path, base, owner, repo_id| {
Repo::open(path)?.push_all(base, owner, repo_id)
},
)
.await
}
});
push.await
@@ -1136,15 +1143,16 @@ impl Backend {
// Each target gets its own deletion event.
fn retract_events(&mut self, events: &[Event], cx: &mut Context<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?;
client.send_event(&deletion).broadcast().await?;
Ok::<(), anyhow::Error>(())
})
.detach();
}
@@ -1198,23 +1206,26 @@ async fn announce_repository_and_push(
let builder = announcement.into_event_builder();
let event = builder.finalize_async(signer).await?;
let output = client.send_event(&event).broadcast().await?;
GraspPush::require_relay_accepted(output, event)?
require_relay_accepted(output, event)?
};
// The state event is the push authorization.
let outcome = if refs.is_empty() {
PushOutcome::default()
} else {
let pusher = GraspPush::new(client.clone(), signer.clone());
let repo_id = repo_id.to_owned();
let owner = owner.to_owned();
let servers = servers.to_vec();
let head = head.clone();
let path = path.to_path_buf();
let executor = cx.background_executor().clone();
cx.background_spawn(async move {
pusher
.push_staged_to_grasps(
let push = {
let client = client.clone();
let signer = signer.clone();
let repo_id = repo_id.to_owned();
let owner = owner.to_owned();
let servers = servers.to_vec();
let head = head.clone();
let path = path.to_path_buf();
let executor = cx.background_executor().clone();
cx.background_spawn(async move {
push_staged_to_grasps(
&client,
&signer,
&repo_id,
&refs,
head.as_deref(),
@@ -1225,8 +1236,9 @@ async fn announce_repository_and_push(
push,
)
.await
})
.await
})
};
push.await
};
if outcome.accepted() == 0 {
crates/signed_state/src/push.rs
+202 -361
View File
@@ -3,14 +3,13 @@ use std::time::Duration;
use anyhow::{Error, bail};
use gpui::BackgroundExecutor;
use nostr::event::IntoEventBuilder;
use nostr::prelude::Url;
use nostr_sdk::prelude::*;
use signed_core::RepoState;
use signed_git::Repo;
use signed_git::{Repo, RepoRefState};
use signed_nostr::UniversalSigner;
pub(crate) const GRASP_PUSH_ATTEMPTS: usize = 3;
const GRASP_PUSH_ATTEMPTS: usize = 3;
const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1);
@@ -18,8 +17,10 @@ const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1);
pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option<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 +111,7 @@ impl PushOutcome {
.join("; ")
}
// `None` when every server accepted the push or nothing was pushed.
/// `None` when every server accepted the push or nothing was pushed.
pub fn partial_warning(&self) -> Option<String> {
let accepted = self.accepted();
if self.servers.is_empty() || accepted == self.servers.len() {
@@ -122,25 +123,119 @@ impl PushOutcome {
self.failure_summary()
))
}
/// All staged events carry the same refs; the newest timestamp wins on the relays.
fn keep_newest(&mut self, event: Event) {
if self
.state_event
.as_ref()
.is_none_or(|current| event.created_at > current.created_at)
{
self.state_event = Some(event);
}
}
}
// Reasons a push attempt should be retried with a freshly staged state event
// and a fresh git advertisement. Two families are retried:
//
// - **Purgatory denials**: the grasp server sends these when the state event
// for the push has not reached its purgatory yet. Re-staging a fresh event
// resolves them.
// - **Stale advertisement races**: `git receive-pack` compares each ref
// update against the value it advertised when the push started. The grasp
// server's own background sync can move a ref in between - typically by
// aligning the repository to a parked state event once the objects of an
// earlier attempt land - so the compare-and-swap fails with `cannot lock
// ref` / `incorrect old value provided`. A retry against the fresh
// advertisement converges, and when the race is lost the pushed data is
// usually already on the server (see `is_stale_advertisement_race` and
// the convergence probe in `push_staged_to_grasps`).
//
// Other rejections are not retried.
pub(crate) fn require_relay_accepted(
output: SendEventOutput,
event: Event,
) -> Result<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)
}
/// Signs with a `created_at` strictly newer than `last_created_at`,
/// since a relay treats a same-id resend as a duplicate.
async fn sign_state_event(
signer: &UniversalSigner,
repo_id: &str,
refs: &[(String, String)],
head: Option<&str>,
last_created_at: u64,
) -> Result<(Event, u64), String> {
let now = Timestamp::now().as_secs();
let created_at = if now > last_created_at {
now
} else {
last_created_at + 1
};
let event = RepoState::build(repo_id, refs, head)
.custom_created_at(Timestamp::from_secs(created_at))
.finalize_async(signer)
.await
.map_err(|e| format!("could not sign the state event: {e}"))?;
Ok((event, created_at))
}
/// On a grasp relay an accepted event is parked in purgatory, which authorizes the paired git push.
async fn stage_event_on_relay(
client: &Client,
relay: &RelayUrl,
event: &Event,
) -> Result<(), String> {
client
.add_relay(relay)
.and_connect()
.await
.map_err(|e| format!("could not add relay {relay}: {e}"))?;
let output = client
.send_event(event)
.to([relay.clone()])
.await
.map_err(|e| format!("could not send the state event to {relay}: {e}"))?;
if output.success.contains_key(relay) {
Ok(())
} else {
let reason = output
.failed
.get(relay)
.cloned()
.unwrap_or_else(|| "relay did not confirm the event".to_owned());
Err(reason)
}
}
/// Refresh the repo's ref state, keeping the current branch as the head if it is still present.
fn refresh_ref_state(path: &Path, head: Option<&str>) -> Option<RepoRefState> {
let state = Repo::open(path).and_then(|repo| repo.ref_state()).ok()?;
if state.refs.is_empty() {
return None;
}
let head = match head {
Some(current) => {
let name = format!("refs/heads/{current}");
if state.refs.iter().any(|(known, _)| *known == name) {
Some(current.to_owned())
} else {
state.head
}
}
None => state.head,
};
Some(RepoRefState {
refs: state.refs,
head,
})
}
/// Denials that staging a fresh state event and pushing against the refreshed advertisement resolve.
fn is_transient_grasp_denial(stderr: &str) -> bool {
let error = stderr.to_lowercase();
[
@@ -156,180 +251,55 @@ fn is_transient_grasp_denial(stderr: &str) -> bool {
.any(|marker| error.contains(marker))
}
// A push rejected because `git receive-pack`'s compare-and-swap lost to the
// grasp server's own background ref alignment: the ref moved between this
// push's advertisement and its ref transaction (`cannot lock ref ... is at
// ... but expected ...` / `incorrect old value provided`). The pushed data
// is usually already on the server by then.
/// The server's own ref alignment moved a ref mid-push; the pushed data is usually already there.
fn is_stale_advertisement_race(stderr: &str) -> bool {
let error = stderr.to_lowercase();
error.contains("cannot lock ref") || error.contains("incorrect old value provided")
}
// All staged events carry the same refs; the newest timestamp wins on the relays.
fn keep_newest(state_event: &mut Option<Event>, event: Event) {
if state_event
.as_ref()
.is_none_or(|current| event.created_at > current.created_at)
{
*state_event = Some(event);
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn push_staged_to_grasps(
client: &Client,
signer: &UniversalSigner,
repo_id: &str,
refs: &[(String, String)],
head: Option<&str>,
path: &Path,
owner: &str,
servers: &[RelayUrl],
executor: &BackgroundExecutor,
push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>,
) -> PushOutcome {
let mut outcome = PushOutcome::default();
let mut refs = refs.to_vec();
let mut head = head.map(str::to_owned);
#[derive(Clone)]
pub(crate) struct GraspPush {
client: Client,
signer: UniversalSigner,
}
impl GraspPush {
pub(crate) fn new(client: Client, signer: UniversalSigner) -> Self {
Self { client, signer }
}
pub(crate) fn require_relay_accepted(
output: SendEventOutput,
event: Event,
) -> Result<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}");
}
}
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
}
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)
.await
.map_err(|e| format!("could not sign the state event: {e}"))?;
let git_url = format!("{base}/{owner}/{repo_id}.git");
Ok((event, created_at))
}
let mut reason = None;
let mut last_created_at = 0;
let mut staged_event = None;
// `Ok` only when the relay confirmed the event. On a grasp relay the
// accept parks the event in purgatory, which authorizes the paired git push.
async fn stage_event_on_relay(&self, relay: &RelayUrl, event: &Event) -> Result<(), String> {
self.client
.add_relay(relay)
.and_connect()
.await
.map_err(|e| format!("could not add relay {relay}: {e}"))?;
let output = self
.client
.send_event(event)
.to([relay.clone()])
.await
.map_err(|e| format!("could not send the state event to {relay}: {e}"))?;
if output.success.contains_key(relay) {
Ok(())
} else {
let reason = output
.failed
.get(relay)
.cloned()
.unwrap_or_else(|| "relay did not confirm the event".to_owned());
Err(reason)
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn push_staged_to_grasps(
&self,
repo_id: &str,
refs: &[(String, String)],
head: Option<&str>,
path: &Path,
owner: &str,
servers: &[RelayUrl],
executor: &BackgroundExecutor,
push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>,
) -> PushOutcome {
let mut outcome = PushOutcome::default();
for relay in servers {
let Some(base) = grasp_base_url(relay) else {
outcome
.servers
.push(GraspServer::failed(relay.clone(), "no domain"));
continue;
};
let git_url = format!("{base}/{owner}/{repo_id}.git");
let mut reason = None;
let mut last_created_at = 0;
// For the convergence probe when every push attempt lost the
// stale-ref race.
let mut staged_event = None;
'server: for attempt in 1..=GRASP_PUSH_ATTEMPTS {
if attempt > 1 {
executor.timer(GRASP_RETRY_DELAY).await;
'server: for attempt in 1..=GRASP_PUSH_ATTEMPTS {
if attempt > 1 {
executor.timer(GRASP_RETRY_DELAY).await;
// The repo's refs may have moved since the last attempt, so refresh the state.
if let Some(state) = refresh_ref_state(path, head.as_deref()) {
refs = state.refs;
head = state.head;
}
}
let (event, created_at) = match self
.sign_state_event(repo_id, refs, head, last_created_at)
let (event, created_at) =
match sign_state_event(signer, repo_id, &refs, head.as_deref(), last_created_at)
.await
{
Ok(signed) => signed,
@@ -339,190 +309,61 @@ impl GraspPush {
}
};
last_created_at = created_at;
last_created_at = created_at;
// A failed stage means the grasp never parked the state, so
// the git push would be denied anyway: skip it.
if let Err(e) = self.stage_event_on_relay(relay, &event).await {
// One retry absorbs a relay connect blip, on the first
// attempt only.
if attempt == 1 && self.stage_event_on_relay(relay, &event).await.is_ok() {
} else {
reason = Some(e);
break 'server;
}
}
staged_event = Some(event.clone());
match push(path, &base, owner, repo_id) {
Ok(()) => {
keep_newest(&mut outcome.state_event, event);
break 'server;
}
Err(e) => {
let text = e.to_string();
if attempt < GRASP_PUSH_ATTEMPTS && is_transient_grasp_denial(&text) {
reason = Some(text);
continue 'server;
}
reason = Some(text);
break 'server;
}
// Stage the event on the relay,
// so that the push can be accepted if the relay's state is aligned.
if let Err(e) = stage_event_on_relay(client, relay, &event).await {
let retried =
attempt == 1 && stage_event_on_relay(client, relay, &event).await.is_ok();
if !retried {
reason = Some(e);
break 'server;
}
}
staged_event = Some(event.clone());
// The grasp's own background sync aligns refs to staged state
// events as soon as the objects land, which can beat every push
// attempt's compare-and-swap. When the last denial was that race
// the sync has usually finished by now: verify the advertised refs
// and accept the server when the pushed data is already there.
if let Some(last_reason) = &reason
&& is_stale_advertisement_race(last_reason)
&& Repo::open(path)
.and_then(|repo| repo.remote_has_refs(&git_url, refs))
.unwrap_or(false)
{
if let Some(event) = staged_event {
keep_newest(&mut outcome.state_event, event);
match push(path, &base, owner, repo_id) {
Ok(()) => {
outcome.keep_newest(event);
break 'server;
}
reason = None;
}
match reason {
Some(reason) => {
log::warn!("grasp push failed: {relay}: {reason}");
outcome
.servers
.push(GraspServer::failed(relay.clone(), reason));
Err(e) => {
let text = e.to_string();
let retriable =
attempt < GRASP_PUSH_ATTEMPTS && is_transient_grasp_denial(&text);
reason = Some(text);
if !retriable {
break 'server;
}
}
None => outcome.servers.push(GraspServer::ok(relay.clone())),
}
}
outcome
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn grasp_base_url_maps_schemes_like_ngit() {
let wss = RelayUrl::parse("wss://relay.ngit.dev").expect("url");
assert_eq!(
grasp_base_url(&wss).as_deref(),
Some("https://relay.ngit.dev")
);
let ws = RelayUrl::parse("ws://localhost:8080").expect("url");
assert_eq!(
grasp_base_url(&ws).as_deref(),
Some("http://localhost:8080")
);
}
#[test]
fn grasp_clone_url_matches_ngit_format() {
let relay = RelayUrl::parse("wss://gitnostr.com").expect("url");
let url = grasp_clone_url(&relay, "npub1test", "my-repo").expect("url");
assert_eq!(
url.to_string(),
"https://gitnostr.com/npub1test/my-repo.git"
);
}
#[test]
fn grasp06_prs_url_matches_ngit_format() {
assert_eq!(
grasp06_prs_url("https://relay.ngit.dev", "npub1author", "my-repo"),
"https://relay.ngit.dev/prs/npub1author/my-repo.git"
);
assert_eq!(
grasp06_prs_url("http://localhost:8080", "npub1author", "my-repo"),
"http://localhost:8080/prs/npub1author/my-repo.git"
);
}
#[test]
fn pr_clone_urls_orders_author_first_and_deduplicates() {
let prs = vec![
Url::parse("https://a.example/prs/npub1me/repo.git").expect("url"),
Url::parse("https://a.example/prs/npub1me/repo.git").expect("url"),
];
let base = vec![
Url::parse("https://a.example/npub1owner/repo.git").expect("url"),
Url::parse("https://b.example/npub1owner/repo.git").expect("url"),
Url::parse("https://b.example/npub1owner/repo.git").expect("url"),
];
let urls = pr_clone_urls(prs, base);
assert_eq!(
urls.iter().map(ToString::to_string).collect::<Vec<_>>(),
vec![
"https://a.example/prs/npub1me/repo.git",
"https://a.example/npub1owner/repo.git",
"https://b.example/npub1owner/repo.git",
]
);
}
#[test]
fn transient_grasp_denials_are_classified() {
let reported = "remote: ERR authorisation failed: No state events in purgatory\n\
fatal: the remote end hung up unexpectedly\n\
error: failed to push some refs to 'https://relay.ngit.dev/...git'";
assert!(is_transient_grasp_denial(reported));
assert!(is_transient_grasp_denial(
"remote: ERR authorisation failed: No matching state event found in purgatory"
));
assert!(is_transient_grasp_denial(
"remote: ERR authorisation failed: 1 state event in purgatory from authorized \
publisher but doesn't match push"
));
assert!(is_transient_grasp_denial(
"remote: ERR authorisation failed: 2 state events in purgatory but none from \
authorized publishers"
));
assert!(is_transient_grasp_denial(
"remote: ERR authorisation failed: No repository announcement found"
));
assert!(!is_transient_grasp_denial(
"remote: ERR authorisation failed: not a maintainer of this repository"
));
assert!(!is_transient_grasp_denial(
"fatal: unable to access 'https://relay.ngit.dev/...': The requested URL returned \
error: 403"
));
assert!(!is_transient_grasp_denial(
"fatal: unable to access 'https://relay.ngit.dev/...': Could not resolve host"
));
}
#[test]
fn stale_ref_races_are_retried() {
let reported = "remote: error: cannot lock ref 'refs/heads/main': is at \
cac2ac91b6f5fb8dfcb6962785babc6e65350cb3 but expected \
bc5e892aa84dc6240a5fbcd59367a4857d26f49b\n\
To https://relay.ngit.dev/npub1owner/signed-test.git\n\
! [remote rejected] main -> main (incorrect old value provided)\n\
error: failed to push some refs to 'https://relay.ngit.dev/npub1owner/signed-test.git'";
assert!(is_transient_grasp_denial(reported));
assert!(is_stale_advertisement_race(reported));
assert!(is_stale_advertisement_race(
"cannot lock ref 'refs/heads/main'"
));
assert!(is_stale_advertisement_race(
"! [remote rejected] main -> main (incorrect old value provided)"
));
assert!(!is_stale_advertisement_race("No state events in purgatory"));
assert!(!is_transient_grasp_denial(
" ! [rejected] main -> main (non-fast-forward)"
));
// If the relay's denial was a stale-ref race,
// check if the repo has moved on and retry the push.
if let Some(last_reason) = &reason
&& is_stale_advertisement_race(last_reason)
&& Repo::open(path)
.and_then(|repo| repo.remote_has_refs(&git_url, &refs))
.unwrap_or(false)
{
if let Some(event) = staged_event {
outcome.keep_newest(event);
}
reason = None;
}
match reason {
Some(reason) => {
log::warn!("grasp push failed: {relay}: {reason}");
outcome
.servers
.push(GraspServer::failed(relay.clone(), reason));
}
None => outcome.servers.push(GraspServer::ok(relay.clone())),
}
}
outcome
}
crates/signed_state/src/repo.rs
+23 -13
View File
@@ -18,7 +18,9 @@ use signed_nostr::UniversalSigner;
use crate::backend::{Backend, BackendEvent};
use crate::bootstrap::user_grasp_list_servers;
use crate::checkouts::CheckoutsStore;
use crate::push::{GraspPush, PushOutcome, grasp_base_url, grasp06_prs_url, pr_clone_urls};
use crate::push::{
PushOutcome, grasp_base_url, grasp06_prs_url, pr_clone_urls, require_relay_accepted,
};
use crate::repos::RepoListStore;
use crate::sync_status::SyncStatusStore;
@@ -865,10 +867,11 @@ impl RepoStore {
}
}
let publish_result = {
let pusher = GraspPush::new(client.clone(), signer.clone());
pusher.send_accepted(event).await
};
let publish_result: Result<Event, Error> = async {
let output = client.send_event(&event).broadcast().await?;
require_relay_accepted(output, event)
}
.await;
let pr_event = match publish_result {
Ok(event) => event,
@@ -1041,10 +1044,12 @@ impl RepoStore {
}
};
let publish_result = {
let pusher = GraspPush::new(client.clone(), signer.clone());
pusher.publish_one(builder).await
};
let publish_result: Result<Event, Error> = async {
let event = builder.finalize_async(&signer).await?;
let output = client.send_event(&event).broadcast().await?;
require_relay_accepted(output, event)
}
.await;
if let Err(e) = publish_result {
return this.update(cx, |this, cx| {
@@ -1309,8 +1314,12 @@ impl RepoStore {
};
let task: Task<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?;
require_relay_accepted(output, event)
}
.await;
if let Err(e) = publish_result {
this.update(cx, |this, cx| {
@@ -1421,7 +1430,6 @@ async fn publish_patch_series(
first_marker: &str,
reply_to: Option<EventId>,
) -> Result<Event, Error> {
let pusher = GraspPush::new(client.clone(), signer.clone());
let mut root: Option<Event> = None;
let mut previous = reply_to;
@@ -1468,7 +1476,9 @@ async fn publish_patch_series(
}
let builder = EventBuilder::new(Kind::GitPatch, part.clone()).tags(tags);
let event = pusher.publish_one(builder).await?;
let event = builder.finalize_async(signer).await?;
let output = client.send_event(&event).broadcast().await?;
let event = require_relay_accepted(output, event)?;
if root.is_none() {
root = Some(event.clone());
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(())
});