Remove retry logic from grasp push
This commit is contained in:
@@ -812,7 +812,6 @@ impl Backend {
|
|||||||
let relays = relays.clone();
|
let relays = relays.clone();
|
||||||
let refs = refs.clone();
|
let refs = refs.clone();
|
||||||
let head = head.clone();
|
let head = head.clone();
|
||||||
let executor = cx.background_executor().clone();
|
|
||||||
async move {
|
async move {
|
||||||
push_staged_to_grasps(
|
push_staged_to_grasps(
|
||||||
&client,
|
&client,
|
||||||
@@ -823,7 +822,6 @@ impl Backend {
|
|||||||
&path,
|
&path,
|
||||||
&owner,
|
&owner,
|
||||||
&relays,
|
&relays,
|
||||||
&executor,
|
|
||||||
|path, base, owner, repo_id| {
|
|path, base, owner, repo_id| {
|
||||||
Repo::open(path)?.push_all(base, owner, repo_id)
|
Repo::open(path)?.push_all(base, owner, repo_id)
|
||||||
},
|
},
|
||||||
@@ -1234,7 +1232,6 @@ async fn announce_repository_and_push(
|
|||||||
let servers = servers.to_vec();
|
let servers = servers.to_vec();
|
||||||
let head = head.clone();
|
let head = head.clone();
|
||||||
let path = path.to_path_buf();
|
let path = path.to_path_buf();
|
||||||
let executor = cx.background_executor().clone();
|
|
||||||
cx.background_spawn(async move {
|
cx.background_spawn(async move {
|
||||||
push_staged_to_grasps(
|
push_staged_to_grasps(
|
||||||
&client,
|
&client,
|
||||||
@@ -1245,7 +1242,6 @@ async fn announce_repository_and_push(
|
|||||||
&path,
|
&path,
|
||||||
&owner,
|
&owner,
|
||||||
&servers,
|
&servers,
|
||||||
&executor,
|
|
||||||
push,
|
push,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|||||||
+42
-187
@@ -1,17 +1,11 @@
|
|||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::time::Duration;
|
|
||||||
|
|
||||||
use anyhow::Error;
|
use anyhow::Error;
|
||||||
use gpui::BackgroundExecutor;
|
|
||||||
use nostr::prelude::Url;
|
use nostr::prelude::Url;
|
||||||
use nostr_sdk::prelude::*;
|
use nostr_sdk::prelude::*;
|
||||||
use signed_core::RepoState;
|
use signed_core::RepoState;
|
||||||
use signed_git::{Repo, RepoRefState};
|
|
||||||
use signed_nostr::UniversalSigner;
|
use signed_nostr::UniversalSigner;
|
||||||
|
|
||||||
const GRASP_PUSH_ATTEMPTS: usize = 3;
|
|
||||||
const GRASP_RETRY_DELAY: Duration = Duration::from_secs(1);
|
|
||||||
|
|
||||||
/// `ws://` grasp servers, like ngit, use `http(s)://<host>`.
|
/// `ws://` grasp servers, like ngit, use `http(s)://<host>`.
|
||||||
pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option<String> {
|
pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option<String> {
|
||||||
let parsed = Url::parse(relay.as_str()).ok()?;
|
let parsed = Url::parse(relay.as_str()).ok()?;
|
||||||
@@ -135,109 +129,6 @@ impl PushOutcome {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Signs with a `created_at` strictly newer than `last_created_at`,
|
|
||||||
/// since a relay treats a same-id resend as a duplicate.
|
|
||||||
async fn sign_state_event(
|
|
||||||
signer: &UniversalSigner,
|
|
||||||
repo_id: &str,
|
|
||||||
refs: &[(String, String)],
|
|
||||||
head: Option<&str>,
|
|
||||||
last_created_at: u64,
|
|
||||||
) -> Result<(Event, u64), String> {
|
|
||||||
let now = Timestamp::now().as_secs();
|
|
||||||
let created_at = if now > last_created_at {
|
|
||||||
now
|
|
||||||
} else {
|
|
||||||
last_created_at + 1
|
|
||||||
};
|
|
||||||
|
|
||||||
let event = RepoState::build(repo_id, refs, head)
|
|
||||||
.custom_created_at(Timestamp::from_secs(created_at))
|
|
||||||
.finalize_async(signer)
|
|
||||||
.await
|
|
||||||
.map_err(|e| format!("could not sign the state event: {e}"))?;
|
|
||||||
|
|
||||||
Ok((event, created_at))
|
|
||||||
}
|
|
||||||
|
|
||||||
/// On a grasp relay an accepted event is parked in purgatory, which authorizes the paired git push.
|
|
||||||
async fn stage_event_on_relay(
|
|
||||||
client: &Client,
|
|
||||||
relay: &RelayUrl,
|
|
||||||
event: &Event,
|
|
||||||
) -> Result<(), String> {
|
|
||||||
client
|
|
||||||
.add_relay(relay)
|
|
||||||
.and_connect()
|
|
||||||
.await
|
|
||||||
.map_err(|e| format!("could not add relay {relay}: {e}"))?;
|
|
||||||
|
|
||||||
let output = client
|
|
||||||
.send_event(event)
|
|
||||||
.to([relay.clone()])
|
|
||||||
.await
|
|
||||||
.map_err(|e| format!("could not send the state event to {relay}: {e}"))?;
|
|
||||||
|
|
||||||
if output.success.contains_key(relay) {
|
|
||||||
Ok(())
|
|
||||||
} else {
|
|
||||||
let reason = output
|
|
||||||
.failed
|
|
||||||
.get(relay)
|
|
||||||
.cloned()
|
|
||||||
.unwrap_or_else(|| "relay did not confirm the event".to_owned());
|
|
||||||
Err(reason)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Refresh the repo's ref state, keeping the current branch as the head if it is still present.
|
|
||||||
fn refresh_ref_state(path: &Path, head: Option<&str>) -> Option<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();
|
|
||||||
[
|
|
||||||
"no state events in purgatory",
|
|
||||||
"no matching state event",
|
|
||||||
"doesn't match push",
|
|
||||||
"none from authorized publishers",
|
|
||||||
"no repository announcement found",
|
|
||||||
"cannot lock ref",
|
|
||||||
"incorrect old value provided",
|
|
||||||
]
|
|
||||||
.iter()
|
|
||||||
.any(|marker| error.contains(marker))
|
|
||||||
}
|
|
||||||
|
|
||||||
/// The server's own ref alignment moved a ref mid-push; the pushed data is usually already there.
|
|
||||||
fn is_stale_advertisement_race(stderr: &str) -> bool {
|
|
||||||
let error = stderr.to_lowercase();
|
|
||||||
error.contains("cannot lock ref") || error.contains("incorrect old value provided")
|
|
||||||
}
|
|
||||||
|
|
||||||
#[allow(clippy::too_many_arguments)]
|
#[allow(clippy::too_many_arguments)]
|
||||||
pub(crate) async fn push_staged_to_grasps(
|
pub(crate) async fn push_staged_to_grasps(
|
||||||
client: &Client,
|
client: &Client,
|
||||||
@@ -248,12 +139,9 @@ pub(crate) async fn push_staged_to_grasps(
|
|||||||
path: &Path,
|
path: &Path,
|
||||||
owner: &str,
|
owner: &str,
|
||||||
servers: &[RelayUrl],
|
servers: &[RelayUrl],
|
||||||
executor: &BackgroundExecutor,
|
|
||||||
push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>,
|
push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>,
|
||||||
) -> PushOutcome {
|
) -> PushOutcome {
|
||||||
let mut outcome = PushOutcome::default();
|
let mut outcome = PushOutcome::default();
|
||||||
let mut refs = refs.to_vec();
|
|
||||||
let mut head = head.map(str::to_owned);
|
|
||||||
|
|
||||||
for relay in servers {
|
for relay in servers {
|
||||||
let Some(base) = grasp_base_url(relay) else {
|
let Some(base) = grasp_base_url(relay) else {
|
||||||
@@ -263,86 +151,53 @@ pub(crate) async fn push_staged_to_grasps(
|
|||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
|
|
||||||
let git_url = format!("{base}/{owner}/{repo_id}.git");
|
let event = match RepoState::build(repo_id, refs, head)
|
||||||
|
.finalize_async(signer)
|
||||||
let mut reason = None;
|
.await
|
||||||
let mut last_created_at = 0;
|
|
||||||
let mut staged_event = None;
|
|
||||||
|
|
||||||
'server: for attempt in 1..=GRASP_PUSH_ATTEMPTS {
|
|
||||||
if attempt > 1 {
|
|
||||||
executor.timer(GRASP_RETRY_DELAY).await;
|
|
||||||
// The repo's refs may have moved since the last attempt, so refresh the state.
|
|
||||||
if let Some(state) = refresh_ref_state(path, head.as_deref()) {
|
|
||||||
refs = state.refs;
|
|
||||||
head = state.head;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
let (event, created_at) =
|
|
||||||
match sign_state_event(signer, repo_id, &refs, head.as_deref(), last_created_at)
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
Ok(signed) => signed,
|
|
||||||
Err(e) => {
|
|
||||||
reason = Some(e);
|
|
||||||
break 'server;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
last_created_at = created_at;
|
|
||||||
|
|
||||||
// Stage the event on the relay,
|
|
||||||
// so that the push can be accepted if the relay's state is aligned.
|
|
||||||
if let Err(e) = stage_event_on_relay(client, relay, &event).await {
|
|
||||||
let retried =
|
|
||||||
attempt == 1 && stage_event_on_relay(client, relay, &event).await.is_ok();
|
|
||||||
if !retried {
|
|
||||||
reason = Some(e);
|
|
||||||
break 'server;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
staged_event = Some(event.clone());
|
|
||||||
|
|
||||||
match push(path, &base, owner, repo_id) {
|
|
||||||
Ok(()) => {
|
|
||||||
outcome.keep_newest(event);
|
|
||||||
break 'server;
|
|
||||||
}
|
|
||||||
Err(e) => {
|
|
||||||
let text = e.to_string();
|
|
||||||
let retriable =
|
|
||||||
attempt < GRASP_PUSH_ATTEMPTS && is_transient_grasp_denial(&text);
|
|
||||||
reason = Some(text);
|
|
||||||
if !retriable {
|
|
||||||
break 'server;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// If the relay's denial was a stale-ref race,
|
|
||||||
// check if the repo has moved on and retry the push.
|
|
||||||
if let Some(last_reason) = &reason
|
|
||||||
&& is_stale_advertisement_race(last_reason)
|
|
||||||
&& Repo::open(path)
|
|
||||||
.and_then(|repo| repo.remote_has_refs(&git_url, &refs))
|
|
||||||
.unwrap_or(false)
|
|
||||||
{
|
{
|
||||||
if let Some(event) = staged_event {
|
Ok(event) => event,
|
||||||
outcome.keep_newest(event);
|
Err(e) => {
|
||||||
}
|
let reason = format!("could not sign the state event: {e}");
|
||||||
reason = None;
|
|
||||||
}
|
|
||||||
|
|
||||||
match reason {
|
|
||||||
Some(reason) => {
|
|
||||||
log::warn!("grasp push failed: {relay}: {reason}");
|
|
||||||
outcome
|
outcome
|
||||||
.servers
|
.servers
|
||||||
.push(GraspServer::failed(relay.clone(), reason));
|
.push(GraspServer::failed(relay.clone(), reason));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// The relay parks the accepted event in purgatory, authorizing the paired git push.
|
||||||
|
let staging = match client.add_relay(relay).and_connect().await {
|
||||||
|
Ok(_) => match client.send_event(&event).to([relay.clone()]).await {
|
||||||
|
Ok(output) if output.success.contains_key(relay) => Ok(()),
|
||||||
|
Ok(output) => Err(output
|
||||||
|
.failed
|
||||||
|
.get(relay)
|
||||||
|
.cloned()
|
||||||
|
.unwrap_or_else(|| "relay did not confirm the event".to_owned())),
|
||||||
|
Err(e) => Err(format!("could not send the state event to {relay}: {e}")),
|
||||||
|
},
|
||||||
|
Err(e) => Err(format!("could not add relay {relay}: {e}")),
|
||||||
|
};
|
||||||
|
|
||||||
|
if let Err(reason) = staging {
|
||||||
|
log::warn!("grasp push failed: {relay}: {reason}");
|
||||||
|
outcome
|
||||||
|
.servers
|
||||||
|
.push(GraspServer::failed(relay.clone(), reason));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
match push(path, &base, owner, repo_id) {
|
||||||
|
Ok(()) => {
|
||||||
|
outcome.keep_newest(event);
|
||||||
|
outcome.servers.push(GraspServer::ok(relay.clone()));
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
log::warn!("grasp push failed: {relay}: {e}");
|
||||||
|
outcome
|
||||||
|
.servers
|
||||||
|
.push(GraspServer::failed(relay.clone(), e.to_string()));
|
||||||
}
|
}
|
||||||
None => outcome.servers.push(GraspServer::ok(relay.clone())),
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user