chore: and reconcile pushes with server only refs
Rust / build (macos-latest, stable) (push) Canceled after 0s
Rust / build (ubuntu-latest, stable) (push) Canceled after 0s
Rust / build (windows-latest, stable) (push) Canceled after 0s

This commit is contained in:
2026-10-16 20:07:13 +07:00
parent a2e7907193
commit 4b4b8b5cf2
8 changed files with 546 additions and 82 deletions
+26 -3
View File
@@ -21,7 +21,7 @@ use crate::git_store::Mirrors;
use crate::inbox::Inbox;
use crate::push::{PushOutcome, grasp_base_url, grasp_clone_url};
use crate::repos::RepoListStore;
use crate::state::{GitPushMode, GitRef, PushStepStatus, SyncState};
use crate::state::{GitPushMode, GitPushPolicy, GitRef, PushStepStatus, SyncState};
pub const USER_KEYRING: &str = "Signed Safe Storage";
pub const NOSTR_CONNECT_TIMEOUT: u64 = 60;
@@ -588,6 +588,7 @@ impl Backend {
head: Some("main".to_owned()),
path: Some(destination.clone()),
mode: GitPushMode::Main,
policy: GitPushPolicy::default(),
};
let on_step = &mut |status: PushStepStatus, cx: &mut AsyncApp| {
@@ -735,7 +736,17 @@ impl Backend {
cx: &mut Context<Self>,
) -> Task<Result<PushOutcome, Error>> {
let path = Mirrors::path(&announcement.addr());
self.push_repo_from(announcement, path, None, cx)
self.push_repo_from(announcement, path, None, GitPushPolicy::Reconcile, cx)
}
/// Republishes the mirror and deletes grasp refs the mirror does not have.
pub fn reset_repository(
&mut self,
announcement: Announcement,
cx: &mut Context<Self>,
) -> Task<Result<PushOutcome, Error>> {
let path = Mirrors::path(&announcement.addr());
self.push_repo_from(announcement, path, None, GitPushPolicy::ResetRemote, cx)
}
pub fn push_checkout(
@@ -745,7 +756,13 @@ impl Backend {
announced_head: Option<String>,
cx: &mut Context<Self>,
) -> Task<Result<PushOutcome, Error>> {
self.push_repo_from(announcement, checkout, announced_head, cx)
self.push_repo_from(
announcement,
checkout,
announced_head,
GitPushPolicy::Reconcile,
cx,
)
}
fn push_repo_from(
@@ -753,6 +770,7 @@ impl Backend {
announcement: Announcement,
path: PathBuf,
announced_head: Option<String>,
policy: GitPushPolicy,
cx: &mut Context<Self>,
) -> Task<Result<PushOutcome, Error>> {
let addr = announcement.addr();
@@ -789,6 +807,11 @@ impl Backend {
});
work.await?
};
state.policy = policy;
// Fetch grasp refs the local repository lacks before the state event is signed,
// so the servers' full ref map is declared.
state.reconcile(&announcement, cx).await?;
// Keep the announced default branch in `HEAD` when it is among the pushed refs.
if let Some(head) = announced_head
+1 -1
View File
@@ -28,7 +28,7 @@ pub use repo::RepoStore;
pub use repos::RepoListStore;
pub use signed_git::{Nip34Binding, Nip34Kind};
use signed_nostr::NostrBackend;
pub use state::{GitPushMode, GitRef, PushStep, PushStepStatus, SyncState};
pub use state::{GitPushMode, GitPushPolicy, GitRef, PushStep, PushStepStatus, SyncState};
pub use sync_status::{CheckoutSyncStatus, SyncEvent, SyncStatusDirectory, SyncStatusStore};
pub fn init(
+18
View File
@@ -1172,6 +1172,24 @@ impl RepoStore {
self.run_push(push, None, cx)
}
/// Republishes the mirror and deletes grasp refs the mirror does not have.
pub fn reset_repository(&mut self, cx: &mut Context<Self>) -> Task<Result<(), Error>> {
if self.pushing {
return Task::ready(Err(anyhow::anyhow!(
"A push to this repository is already in progress"
)));
}
let Some(announcement) = self.action_announcement(cx) else {
return self.action_error("Repository announcement is not loaded yet", cx);
};
let backend = Backend::global(cx);
let push = backend.update(cx, |backend, cx| backend.reset_repository(announcement, cx));
self.run_push(push, None, cx)
}
pub fn push_checkout(
&mut self,
path: PathBuf,
+164 -74
View File
@@ -40,6 +40,17 @@ pub enum GitPushMode {
Main,
}
/// How to handle server-only refs when pushing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum GitPushPolicy {
/// Reconcile the local refs with the grasp servers,
/// fetching missing refs before pushing.
#[default]
Reconcile,
/// Declare the local refs only and delete server-only refs while pushing.
ResetRemote,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncState {
pub refs: Vec<GitRef>,
@@ -47,6 +58,7 @@ pub struct SyncState {
/// Local repository to push from; `None` for states parsed from events.
pub path: Option<PathBuf>,
pub mode: GitPushMode,
pub policy: GitPushPolicy,
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -135,6 +147,7 @@ impl SyncState {
head,
path: None,
mode: GitPushMode::default(),
policy: GitPushPolicy::default(),
}
}
@@ -154,6 +167,7 @@ impl SyncState {
head: state.head,
path: Some(path.to_path_buf()),
mode: GitPushMode::default(),
policy: GitPushPolicy::default(),
})
}
@@ -162,13 +176,127 @@ impl SyncState {
self
}
pub fn with_policy(mut self, policy: GitPushPolicy) -> Self {
self.policy = policy;
self
}
/// Reconciles the local repository with the grasp servers, fetching missing refs.
pub async fn reconcile(
&mut self,
announcement: &Announcement,
cx: &mut AsyncApp,
) -> Result<(), Error> {
if self.policy == GitPushPolicy::ResetRemote || self.refs.is_empty() {
return Ok(());
}
let Some(path) = self.path.clone() else {
return Ok(());
};
let local: Vec<(String, String)> = self
.refs
.iter()
.map(|git_ref| (git_ref.name.clone(), git_ref.commit.to_string()))
.collect();
let urls: Vec<String> = announcement
.clone
.iter()
.map(|url| url.to_string())
.collect();
let snapshot_path = path.clone();
let missing = cx
.background_spawn(async move {
let repo = Repo::open(&path)?;
// Refs the servers advertise that the local repository lacks
let mut missing: Vec<(String, Vec<String>)> = Vec::new();
// Diverged server tips that exist only on the servers
let mut backups: Vec<(String, String, Vec<String>)> = Vec::new();
for url in &urls {
let Ok(remote_refs) = repo.ls_remote(url) else {
continue;
};
for (name, oid) in remote_refs {
let Some((_, local_oid)) =
local.iter().find(|(local_name, _)| local_name == &name)
else {
match missing.iter_mut().find(|(known, _)| known == &name) {
Some((_, sources)) => sources.push(url.clone()),
None => missing.push((name, vec![url.clone()])),
}
continue;
};
if local_oid == &oid {
continue;
}
if repo.merge_base(local_oid, &oid).is_err() {
match backups.iter_mut().find(|(known, _, _)| known == &name) {
Some((_, _, sources)) => sources.push(url.clone()),
None => {
backups.push((name.clone(), oid.clone(), vec![url.clone()]))
}
}
}
}
}
let mut failures = Vec::new();
for (name, sources) in &missing {
let refspec = format!("+{name}:{name}");
if let Err(error) = repo.fetch_refs(sources, &refspec) {
log::warn!("failed to fetch {name} from grasp servers: {error}");
failures.push(name.clone());
}
}
for (name, oid, sources) in &backups {
let refspec = format!("+{name}:refs/nostr/backup/{oid}");
if let Err(error) = repo.fetch_refs(sources, &refspec) {
log::warn!("failed to back up {name} from grasp servers: {error}");
}
}
match failures.is_empty() {
true => Ok(missing
.into_iter()
.map(|(name, _)| name)
.collect::<Vec<_>>()),
false => bail!(
"could not fetch {} from any grasp server",
failures.join(", ")
),
}
})
.await?;
if !missing.is_empty() {
let (mode, policy) = (self.mode, self.policy);
*self = Self::from_repo(&snapshot_path)?;
self.mode = mode;
self.policy = policy;
}
Ok(())
}
/// State event for these refs; the caller signs it.
pub fn state_event(&self, repo_id: &str) -> EventBuilder {
Self::build(repo_id, &self.refs, self.head.as_deref())
}
/// Push the state event and objects to the grasp servers,
/// calling `on_step` for each step.
/// Push the state event and objects to the grasp servers.
///
/// Calling `on_step` for each step.
pub async fn push_to_grasps<P>(
&self,
client: &Client,
@@ -191,6 +319,12 @@ impl SyncState {
.map_err(|error| anyhow!("could not encode the repository owner: {error}"))?;
let mut outcome = PushOutcome::default();
let policy = self.policy;
let declared: Vec<String> = self
.refs
.iter()
.map(|git_ref| git_ref.name.clone())
.collect();
for relay in &announcement.relays {
let step = PushStep::StageStateEvent(relay.clone());
@@ -241,12 +375,24 @@ impl SyncState {
let owner = owner.clone();
let repo_id = announcement.id.clone();
let mode = self.mode;
let declared = declared.clone();
async move {
let repo = Repo::open(&path)?;
match mode {
GitPushMode::Main => repo.push_main(&base, &owner, &repo_id),
GitPushMode::All => repo.push_all(&base, &owner, &repo_id),
GitPushMode::All => {
let mut refspecs = force_refspecs();
if policy == GitPushPolicy::ResetRemote {
let url = format!("{base}/{owner}/{repo_id}.git");
let server_refs = repo.ls_remote(&url)?;
refspecs.extend(deletion_refspecs(&declared, &server_refs));
}
let refspecs = refspecs.iter().map(String::as_str).collect::<Vec<_>>();
repo.push_refspecs(&base, &owner, &repo_id, &refspecs)
}
}
}
});
@@ -370,75 +516,19 @@ impl SyncState {
}
}
#[cfg(test)]
mod tests {
use super::*;
const COMMIT_A: &str = "aa231c4c6a5777dc89b42207b499891a344add5c";
const COMMIT_B: &str = "59429cfc6cb35b0a1ddace73b5a5c5ed57b8f5ca";
fn git_ref(name: &str, commit: &str) -> GitRef {
GitRef::new(name, commit).expect("valid ref")
}
fn keys() -> Keys {
Keys::new(
SecretKey::from_hex("0000000000000000000000000000000000000000000000000000000000000001")
.expect("valid secret key"),
)
}
fn state_event(tags: &[&[&str]]) -> Event {
let tags: Vec<Tag> = tags
.iter()
.map(|t| Tag::parse(t.to_vec()).expect("valid tag"))
.collect();
EventBuilder::new(Kind::RepoState, "")
.tags(tags)
.finalize(&keys())
.expect("signed event")
}
#[test]
fn parses_heads_and_tags() {
let event = state_event(&[
&["HEAD", "ref: refs/heads/main"],
&["refs/heads/main", COMMIT_A],
&["refs/heads/dev", COMMIT_B],
&["refs/tags/v1.0", COMMIT_A],
]);
let state = SyncState::parse(&event);
assert_eq!(state.head.as_deref(), Some("main"));
assert_eq!(
state.refs,
vec![
git_ref("refs/heads/main", COMMIT_A),
git_ref("refs/heads/dev", COMMIT_B),
git_ref("refs/tags/v1.0", COMMIT_A),
]
);
}
#[test]
fn build_round_trips_through_parse() {
let refs = vec![
git_ref("refs/heads/main", COMMIT_A),
git_ref("refs/heads/dev", COMMIT_B),
git_ref("refs/tags/v1.0", COMMIT_A),
];
let event = SyncState::build("my-repo", &refs, Some("main"))
.finalize(&keys())
.expect("signed event");
assert_eq!(event.kind, Kind::RepoState);
assert_eq!(event.tags.identifier().as_deref(), Some("my-repo"));
let state = SyncState::parse(&event);
assert_eq!(state.refs, refs);
assert_eq!(state.head.as_deref(), Some("main"));
}
/// Refspecs to force push all heads and tags to the grasp servers.
fn force_refspecs() -> Vec<String> {
vec![
"+refs/heads/*:refs/heads/*".to_owned(),
"+refs/tags/*:refs/tags/*".to_owned(),
]
}
/// Refspecs to delete server-only refs from the grasp servers.
fn deletion_refspecs(declared: &[String], server_refs: &[(String, String)]) -> Vec<String> {
server_refs
.iter()
.filter(|(name, _)| !declared.contains(name))
.map(|(name, _)| format!(":{name}"))
.collect()
}