From a8255a2e074aa07df5fa4b4fee5a3703b8d41d94 Mon Sep 17 00:00:00 2001 From: Fintan Halpenny Date: Tue, 19 Aug 2025 15:35:18 +0100 Subject: [PATCH] Introduce a node event for canonical reference updates Whenever the node fetches new updates, it checks if canonical references can be updated. The node has learned how to return these results and emit them as node events. This is a breaking change since it adds a new variant the `Event` type, which is not forwards-compatible. --- crates/radicle-node/src/test/simulator.rs | 1 + crates/radicle-node/src/tests.rs | 1 + crates/radicle-node/src/worker/fetch.rs | 25 ++++++++++---- crates/radicle-protocol/src/service.rs | 11 +++++++ crates/radicle-protocol/src/worker/fetch.rs | 36 ++++++++++++++++++++- crates/radicle/CHANGELOG.md | 8 +++++ crates/radicle/src/node/events.rs | 26 ++++++++++++++- 7 files changed, 99 insertions(+), 9 deletions(-) diff --git a/crates/radicle-node/src/test/simulator.rs b/crates/radicle-node/src/test/simulator.rs index 38ef3400..849a006b 100644 --- a/crates/radicle-node/src/test/simulator.rs +++ b/crates/radicle-node/src/test/simulator.rs @@ -663,6 +663,7 @@ where remote, Rc::new(Ok(fetch::FetchResult { updated: vec![], + canonical: fetch::UpdatedCanonicalRefs::default(), namespaces: HashSet::new(), clone: true, doc: arbitrary::gen(1), diff --git a/crates/radicle-node/src/tests.rs b/crates/radicle-node/src/tests.rs index 75d1c4f3..bb232db7 100644 --- a/crates/radicle-node/src/tests.rs +++ b/crates/radicle-node/src/tests.rs @@ -1589,6 +1589,7 @@ fn test_queued_fetch_from_ann_same_rid() { name: refname.clone(), oid, }], + canonical: fetch::UpdatedCanonicalRefs::default(), namespaces: [carol.id()].into_iter().collect(), clone: false, doc: arbitrary::gen(1), diff --git a/crates/radicle-node/src/worker/fetch.rs b/crates/radicle-node/src/worker/fetch.rs index 53428355..1b495b3b 100644 --- a/crates/radicle-node/src/worker/fetch.rs +++ b/crates/radicle-node/src/worker/fetch.rs @@ -18,7 +18,7 @@ use radicle::storage::{ use radicle::{cob, git, node, Storage}; use radicle_fetch::git::refs::Applied; use radicle_fetch::{Allowed, BlockList, FetchLimit}; -pub use radicle_protocol::worker::fetch::FetchResult; +pub use radicle_protocol::worker::fetch::{FetchResult, UpdatedCanonicalRefs}; use super::channels::ChannelsFlush; @@ -128,9 +128,13 @@ impl Handle { Err(e) => return Err(e.into()), } - if let Err(e) = set_canonical_refs(&repo, &applied) { - log::warn!(target: "worker", "Failed to set canonical references: {e}"); - } + let canonical = match set_canonical_refs(&repo, &applied) { + Ok(updates) => updates.unwrap_or_default(), + Err(e) => { + log::warn!(target: "worker", "Failed to set canonical references: {e}"); + UpdatedCanonicalRefs::default() + } + }; // Notifications are only posted for pulls, not clones. if let Some(mut store) = notifs { @@ -147,6 +151,7 @@ impl Handle { Ok(FetchResult { updated: applied.updated, + canonical, namespaces: remotes.into_iter().collect(), doc: repo.identity_doc()?, clone, @@ -356,17 +361,21 @@ where Ok(()) } -fn set_canonical_refs(repo: &Repository, applied: &Applied) -> Result<(), error::Canonical> { +fn set_canonical_refs( + repo: &Repository, + applied: &Applied, +) -> Result, error::Canonical> { let identity = repo.identity()?; let rules = match identity .canonical_refs()? .map(|crefs| crefs.rules().clone()) .filter(|rules| !rules.is_empty()) { - None => return Ok(()), + None => return Ok(None), Some(rules) => rules, }; + let mut updated_refs = UpdatedCanonicalRefs::default(); for update in applied.updated.iter() { let name = match update { RefUpdate::Updated { name, .. } | RefUpdate::Created { name, .. } => name, @@ -421,9 +430,11 @@ fn set_canonical_refs(repo: &Repository, applied: &Applied) -> Result<(), error: target: "worker", "Failed to set canonical reference {refname}->{oid}: {e}" ); + } else { + updated_refs.updated(refname, oid); } } } } - Ok(()) + Ok(Some(updated_refs)) } diff --git a/crates/radicle-protocol/src/service.rs b/crates/radicle-protocol/src/service.rs index 3c373dd9..ae8052d5 100644 --- a/crates/radicle-protocol/src/service.rs +++ b/crates/radicle-protocol/src/service.rs @@ -1177,6 +1177,7 @@ where match result { Ok(crate::worker::fetch::FetchResult { updated, + canonical, namespaces, clone, doc, @@ -1198,6 +1199,16 @@ where rid, updated: updated.clone(), }); + self.emitter.emit_all( + canonical + .into_iter() + .map(|(refname, target)| Event::CanonicalRefUpdated { + rid, + refname, + target, + }) + .collect(), + ); // Announce our new inventory if this fetch was a full clone. // Only update and announce inventory for public repositories. diff --git a/crates/radicle-protocol/src/worker/fetch.rs b/crates/radicle-protocol/src/worker/fetch.rs index 1ff587aa..1ca13f4b 100644 --- a/crates/radicle-protocol/src/worker/fetch.rs +++ b/crates/radicle-protocol/src/worker/fetch.rs @@ -1,14 +1,17 @@ pub mod error; -use std::collections::HashSet; +use std::collections::{BTreeMap, HashSet}; use radicle::crypto::PublicKey; +use radicle::git; use radicle::{identity::DocAt, storage::RefUpdate}; #[derive(Debug, Clone)] pub struct FetchResult { /// The set of updated references. pub updated: Vec, + /// The canonical references that were updated as part of the fetch process. + pub canonical: UpdatedCanonicalRefs, /// The set of remote namespaces that were updated. pub namespaces: HashSet, /// The fetch was a full clone. @@ -21,6 +24,7 @@ impl FetchResult { pub fn new(doc: DocAt) -> Self { Self { updated: vec![], + canonical: UpdatedCanonicalRefs::default(), namespaces: HashSet::new(), clone: false, doc, @@ -28,11 +32,41 @@ impl FetchResult { } } +/// The set of canonical references, updated after a fetch, and their +/// corresponding targets. +#[derive(Clone, Default, Debug)] +pub struct UpdatedCanonicalRefs { + inner: BTreeMap, git::Oid>, +} + +impl IntoIterator for UpdatedCanonicalRefs { + type Item = (git::Qualified<'static>, git::Oid); + type IntoIter = std::collections::btree_map::IntoIter, git::Oid>; + + fn into_iter(self) -> Self::IntoIter { + self.inner.into_iter() + } +} + +impl UpdatedCanonicalRefs { + /// Insert a new updated entry for the canonical reference identified by + /// `refname` and its new `target.` + pub fn updated(&mut self, refname: git::Qualified<'static>, target: git::Oid) { + self.inner.insert(refname, target); + } + + /// Return an iterator of all the updates. + pub fn iter(&self) -> impl Iterator, &git::Oid)> { + self.inner.iter() + } +} + #[cfg(any(test, feature = "test"))] impl qcheck::Arbitrary for FetchResult { fn arbitrary(g: &mut qcheck::Gen) -> Self { FetchResult { updated: vec![], + canonical: UpdatedCanonicalRefs::default(), namespaces: HashSet::arbitrary(g), clone: bool::arbitrary(g), doc: DocAt::arbitrary(g), diff --git a/crates/radicle/CHANGELOG.md b/crates/radicle/CHANGELOG.md index 4fbd1d7d..faabf97a 100644 --- a/crates/radicle/CHANGELOG.md +++ b/crates/radicle/CHANGELOG.md @@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Introduce a node event for canonical reference updates, `Event::CanonicalRefUpdated`. + Whenever the node fetches new updates, it checks if canonical references can + be updated. The node has learned how to return these results and emit them as + node events. This is a breaking change since it adds a new variant the `Event` + type. +- Add `#[non_exhaustive]` to `Event` to prevent any further breaking changes + when adding new variants. + ### Changed - `radicle::profile::Home::socket` defaults to the path `\\.\pipe\radicle-node` diff --git a/crates/radicle/src/node/events.rs b/crates/radicle/src/node/events.rs index 20618b5e..de63491e 100644 --- a/crates/radicle/src/node/events.rs +++ b/crates/radicle/src/node/events.rs @@ -9,7 +9,7 @@ use std::time; use crossbeam_channel as chan; -use crate::git::Oid; +use crate::git::{Oid, Qualified}; use crate::node; use crate::prelude::*; use crate::storage::{refs, RefUpdate}; @@ -21,6 +21,7 @@ pub const MAX_PENDING_EVENTS: usize = 8192; /// /// The node emits events of this type to its control socket for other /// programs to consume. +#[non_exhaustive] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase", tag = "type")] pub enum Event { @@ -122,6 +123,15 @@ pub enum Event { }, /// The node has uploaded a Git pack file to another node. UploadPack(upload_pack::UploadPack), + /// A canonical reference was updated after a fetch. + CanonicalRefUpdated { + /// The repository the canonical reference was updated for. + rid: RepoId, + /// The reference name of the canonical reference update. + refname: Qualified<'static>, + /// The new target of the reference, after the update. + target: Oid, + }, } impl From for Event { @@ -214,6 +224,20 @@ impl Emitter { .retain(|s| s.try_send(event.clone()).is_ok()); } + /// Emit a batch of events to subscribers and drop those who can't receive + /// them. + /// N.b. subscribers are also dropped if their channel is full. + pub fn emit_all(&self, events: Vec) { + // SAFETY: We deliberately propagate panics from other threads holding the lock. + #[allow(clippy::unwrap_used)] + self.subscribers.lock().unwrap().retain(|s| { + events + .clone() + .into_iter() + .all(|event| s.try_send(event).is_ok()) + }); + } + /// Subscribe to events stream. pub fn subscribe(&self) -> chan::Receiver { let (sender, receiver) = chan::bounded(MAX_PENDING_EVENTS);