Radish alpha
r
radicle-ci-broker
Radicle CI broker
Radicle
Git (anonymous pull)
Log in to clone via SSH
Commits ›
feat: handle `LocalRefsAnnounced` events
✓ CI success Defelo committed 5 months ago
commit 27319c907efe6d43d2c8d836f0c7afc8df827903
parent b6efecdd720e6cf9a3c4769c794bdc544f31cc85
1 passed (1 total) View logs
4 files changed +257 -90
modified src/bin/cibtoolcmd/event.rs
@@ -1,11 +1,13 @@
use std::io::Write;



use clap::ValueEnum;
- 


use radicle::{git::BranchName, patch::PatchId, storage::git::Repository};

use radicle_ci_broker::{
- 
    ergo, filter::EventFilter, node_event_source::NodeEventSource, refs::branch_ref,
- 
    refs::ref_string, util::read_file_as_objectid,
+ 
    ergo,
+ 
    filter::EventFilter,
+ 
    node_event_source::NodeEventSource,
+ 
    refs::{branch_ref, ref_string},
+ 
    util::read_file_as_objectid,

};



use super::*;
@@ -419,6 +421,8 @@ pub struct CiEvents {


impl Leaf for CiEvents {

    fn run(&self, _args: &Args) -> Result<(), CibToolError> {
+ 
        let profile = Profile::load().map_err(CibToolError::Profile)?;
+ 


        let bytes = std::fs::read(&self.input)

            .map_err(|e| CibToolError::ReadEvents(self.input.clone(), e))?;

        let text = String::from_utf8(bytes)
@@ -433,7 +437,7 @@ impl Leaf for CiEvents {


        let mut ci_events: Vec<CiEvent> = vec![];

        for node_event in node_events.iter() {
- 
            if let Ok(mut cevs) = CiEvent::from_node_event(node_event) {
+ 
            if let Ok(mut cevs) = CiEvent::from_node_event(node_event, &profile) {

                ci_events.append(&mut cevs);

            }

        }
modified src/ci_event.rs
@@ -1,16 +1,22 @@
- 
use std::path::{Path, PathBuf};
-
- 
use regex::Regex;
- 
use serde::{Deserialize, Serialize};
+ 
use std::{
+ 
    collections::BTreeMap,
+ 
    path::{Path, PathBuf},
+ 
};



use radicle::{
+ 
    Profile,

    cob::patch::PatchId,

    crypto::PublicKey,
- 
    git::{BranchName, Namespaced, Oid, RefString},
+ 
    git::{BranchName, Namespaced, Oid, Qualified, RefString},

    node::{Event, NodeId},

    prelude::RepoId,
- 
    storage::RefUpdate,
+ 
    storage::{
+ 
        ReadStorage, RefUpdate, RepositoryError,
+ 
        refs::{Refs, RefsAt},
+ 
    },

};
+ 
use regex::Regex;
+ 
use serde::{Deserialize, Serialize};



use crate::{

    msg::RunId,
@@ -276,64 +282,52 @@ impl CiEvent {
    }



    #[allow(clippy::unwrap_used)]
- 
    pub fn from_node_event(event: &Event) -> Result<Vec<Self>, CiEventError> {
- 
        match event {
+ 
    pub fn from_node_event(event: &Event, profile: &Profile) -> Result<Vec<Self>, CiEventError> {
+ 
        let (rid, updates) = match event {

            Event::RefsFetched {

                remote: _,

                rid,

                updated,
+ 
            } => (*rid, updated),
+ 
            Event::LocalRefsAnnounced {
+ 
                rid,
+ 
                refs,
+ 
                timestamp: _,

            } => {
- 
                let mut events = vec![];
- 
                for update in updated {
- 
                    let e = match update {
- 
                        RefUpdate::Created { name, oid } => {
- 
                            let origin = originator(name.to_namespaced().unwrap())?;
- 
                            match ParsedRef::parse_ref(name) {
- 
                                Some(ParsedRef::Branch(branch)) => {
- 
                                    Self::branch_created(origin, *rid, &branch, *oid)?
- 
                                }
- 
                                Some(ParsedRef::Patch(patch_id)) => {
- 
                                    Self::patch_created(origin, *rid, patch_id, *oid)
- 
                                }
- 
                                Some(ParsedRef::Tag(tag_name)) => {
- 
                                    Self::tag_created(origin, *rid, &tag_name, *oid)?
- 
                                }
- 
                                None => continue,
- 
                            }
- 
                        }
- 
                        RefUpdate::Updated { name, old, new } => {
- 
                            let origin = originator(name.to_namespaced().unwrap())?;
- 
                            match ParsedRef::parse_ref(name) {
- 
                                Some(ParsedRef::Branch(branch)) => {
- 
                                    Self::branch_updated(origin, *rid, &branch, *new, *old)?
- 
                                }
- 
                                Some(ParsedRef::Patch(patch_id)) => {
- 
                                    Self::patch_updated(origin, *rid, patch_id, *new)
- 
                                }
- 
                                Some(ParsedRef::Tag(tag_name)) => {
- 
                                    Self::tag_updated(origin, *rid, &tag_name, *new, *old)?
- 
                                }
- 
                                None => continue,
- 
                            }
- 
                        }
- 
                        RefUpdate::Deleted { name, oid } => {
- 
                            let origin = originator(name.to_namespaced().unwrap())?;
- 
                            match ParsedRef::parse_ref(name) {
- 
                                Some(ParsedRef::Branch(branch)) => {
- 
                                    Self::branch_deleted(origin, *rid, &branch, *oid)?
- 
                                }
- 
                                Some(ParsedRef::Patch(_patch_id)) => continue,
- 
                                Some(ParsedRef::Tag(tag_name)) => {
- 
                                    Self::tag_deleted(origin, *rid, &tag_name, *oid)?
- 
                                }
- 
                                None => continue,
- 
                            }
- 
                        }
- 
                        RefUpdate::Skipped { .. } => continue,
- 
                    };
- 
                    events.push(e);
- 
                }
- 
                Ok(events)
+ 
                let repo = profile
+ 
                    .storage
+ 
                    .repository(*rid)
+ 
                    .map_err(CiEventError::Repository)?;
+ 
                let parent = repo
+ 
                    .backend
+ 
                    .find_commit(*refs.at)
+ 
                    .map_err(CiEventError::Git)?
+ 
                    .parent(0)
+ 
                    .map_err(CiEventError::Git)?
+ 
                    .id();
+ 
                let parent_refs = RefsAt {
+ 
                    remote: refs.remote,
+ 
                    at: parent.into(),
+ 
                };
+
+ 
                let signed_refs_new = refs.load(&repo).map_err(CiEventError::StorageRefs)?.sigrefs;
+ 
                let signed_refs_old = parent_refs
+ 
                    .load(&repo)
+ 
                    .map_err(CiEventError::StorageRefs)?
+ 
                    .sigrefs;
+
+ 
                let updates = diff_refs(signed_refs_old.refs, signed_refs_new.refs)
+ 
                    .into_iter()
+ 
                    .filter_map(|mut ref_update| {
+ 
                        let name = ref_update_name_mut(&mut ref_update);
+ 
                        *name = Qualified::from_refstr(&*name)?
+ 
                            .with_namespace(profile.public_key.to_component())
+ 
                            .into_qualified()
+ 
                            .into_refstring();
+ 
                        Some(ref_update)
+ 
                    });
+
+ 
                (*rid, &updates.collect())

            }

            Event::RefsSynced { .. }

            | Event::RefsAnnounced { .. }
@@ -342,10 +336,61 @@ impl CiEvent {
            | Event::SeedDropped { .. }

            | Event::PeerConnected { .. }

            | Event::PeerDisconnected { .. }
- 
            | Event::LocalRefsAnnounced { .. }

            | Event::UploadPack { .. }
- 
            | Event::InventoryAnnounced { .. } => Ok(vec![]),
+ 
            | Event::InventoryAnnounced { .. } => return Ok(vec![]),
+ 
        };
+
+ 
        let mut events = vec![];
+ 
        for update in updates {
+ 
            let e = match update {
+ 
                RefUpdate::Created { name, oid } => {
+ 
                    let origin = originator(name.to_namespaced().unwrap())?;
+ 
                    match ParsedRef::parse_ref(name) {
+ 
                        Some(ParsedRef::Branch(branch)) => {
+ 
                            Self::branch_created(origin, rid, &branch, *oid)?
+ 
                        }
+ 
                        Some(ParsedRef::Patch(patch_id)) => {
+ 
                            Self::patch_created(origin, rid, patch_id, *oid)
+ 
                        }
+ 
                        Some(ParsedRef::Tag(tag_name)) => {
+ 
                            Self::tag_created(origin, rid, &tag_name, *oid)?
+ 
                        }
+ 
                        None => continue,
+ 
                    }
+ 
                }
+ 
                RefUpdate::Updated { name, old, new } => {
+ 
                    let origin = originator(name.to_namespaced().unwrap())?;
+ 
                    match ParsedRef::parse_ref(name) {
+ 
                        Some(ParsedRef::Branch(branch)) => {
+ 
                            Self::branch_updated(origin, rid, &branch, *new, *old)?
+ 
                        }
+ 
                        Some(ParsedRef::Patch(patch_id)) => {
+ 
                            Self::patch_updated(origin, rid, patch_id, *new)
+ 
                        }
+ 
                        Some(ParsedRef::Tag(tag_name)) => {
+ 
                            Self::tag_updated(origin, rid, &tag_name, *new, *old)?
+ 
                        }
+ 
                        None => continue,
+ 
                    }
+ 
                }
+ 
                RefUpdate::Deleted { name, oid } => {
+ 
                    let origin = originator(name.to_namespaced().unwrap())?;
+ 
                    match ParsedRef::parse_ref(name) {
+ 
                        Some(ParsedRef::Branch(branch)) => {
+ 
                            Self::branch_deleted(origin, rid, &branch, *oid)?
+ 
                        }
+ 
                        Some(ParsedRef::Patch(_patch_id)) => continue,
+ 
                        Some(ParsedRef::Tag(tag_name)) => {
+ 
                            Self::tag_deleted(origin, rid, &tag_name, *oid)?
+ 
                        }
+ 
                        None => continue,
+ 
                    }
+ 
                }
+ 
                RefUpdate::Skipped { .. } => continue,
+ 
            };
+ 
            events.push(e);

        }
+ 
        Ok(events)

    }



    pub fn to_pretty_json(&self) -> Result<String, CiEventError> {
@@ -357,6 +402,47 @@ fn originator(name: Namespaced) -> Result<PublicKey, CiEventError> {
    PublicKey::from_namespaced(&name).map_err(|err| CiEventError::key_from_namespaced(&name, err))

}
+ 
fn diff_refs(old_refs: Refs, new_refs: Refs) -> Vec<RefUpdate> {
+ 
    let mut updates = Vec::new();
+ 
    let mut old_refs = BTreeMap::from(old_refs);
+ 
    let mut new_refs = BTreeMap::from(new_refs);
+
+ 
    old_refs.retain(|old_ref, old_oid| match new_refs.remove_entry(old_ref) {
+ 
        Some((new_ref, new_oid)) => {
+ 
            if new_oid != *old_oid {
+ 
                updates.push(RefUpdate::Updated {
+ 
                    name: new_ref,
+ 
                    old: *old_oid,
+ 
                    new: new_oid,
+ 
                })
+ 
            }
+ 
            false
+ 
        }
+ 
        None => true,
+ 
    });
+ 
    updates.extend(
+ 
        old_refs
+ 
            .into_iter()
+ 
            .map(|(old_ref, oid)| RefUpdate::Deleted { name: old_ref, oid }),
+ 
    );
+ 
    updates.extend(
+ 
        new_refs
+ 
            .into_iter()
+ 
            .map(|(new_ref, oid)| RefUpdate::Created { name: new_ref, oid }),
+ 
    );
+
+ 
    updates
+ 
}
+
+ 
fn ref_update_name_mut(ref_update: &mut RefUpdate) -> &mut RefString {
+ 
    match ref_update {
+ 
        RefUpdate::Updated { name, .. } => name,
+ 
        RefUpdate::Created { name, .. } => name,
+ 
        RefUpdate::Deleted { name, .. } => name,
+ 
        RefUpdate::Skipped { name, .. } => name,
+ 
    }
+ 
}
+ 


pub struct CiEvents {

    events: Vec<CiEvent>,

}
@@ -404,6 +490,15 @@ pub enum CiEventError {


    #[error("failed to encode CI event as JSON")]

    ToJson(#[source] Box<dyn std::error::Error + Send + 'static>),
+
+ 
    #[error(transparent)]
+ 
    Repository(RepositoryError),
+
+ 
    #[error(transparent)]
+ 
    StorageRefs(radicle::storage::refs::Error),
+
+ 
    #[error(transparent)]
+ 
    Git(radicle::git::raw::Error),

}



impl CiEventError {
@@ -431,16 +526,22 @@ impl CiEventError {
#[cfg(test)]

#[allow(clippy::unwrap_used)]

mod test {
- 
    use super::*;
- 
    use radicle::{prelude::NodeId, storage::RefUpdate};

    use std::str::FromStr;
- 
    use crate::refs::{branch_from_namespaced, ref_string};
+ 
    use radicle::{prelude::NodeId, storage::RefUpdate};
+
+ 
    use super::*;
+ 
    use crate::{
+ 
        refs::{branch_from_namespaced, ref_string},
+ 
        test::{MockNode, TestResult},
+ 
    };



    const MAIN_BRANCH_REF_NAME: &str =

        "refs/namespaces/z6MkiB8T5cBEQHnrs2MgjMVqvpSVj42X81HjKfFi2XBoMbtr/refs/heads/main";
- 
    const PATCH_REF_NAME: &str = "refs/namespaces/z6MkiB8T5cBEQHnrs2MgjMVqvpSVj42X81HjKfFi2XBoMbtr/refs/heads/patches/f9fa90725474de9002be503ae3cda4670c9a174";
+ 
    const PATCH_REF_NAME: &str = "refs/namespaces/\
+ 
                                  z6MkiB8T5cBEQHnrs2MgjMVqvpSVj42X81HjKfFi2XBoMbtr/refs/heads/\
+ 
                                  patches/f9fa90725474de9002be503ae3cda4670c9a174";

    const PATCH_ID: &str = "f9fa90725474de9002be503ae3cda4670c9a174";



    fn nid() -> NodeId {
@@ -475,19 +576,24 @@ mod test {
    }



    #[test]
- 
    fn nothing_updated() {
+ 
    fn nothing_updated() -> TestResult<()> {
+ 
        let mock_node = MockNode::new()?;
+ 
        let profile = mock_node.profile()?;

        let event = Event::RefsFetched {

            remote: nid(),

            rid: rid(),

            updated: vec![],

        };
- 
        let result = CiEvent::from_node_event(&event);
+ 
        let result = CiEvent::from_node_event(&event, &profile);

        assert!(result.is_ok());

        assert_eq!(result.unwrap(), vec![]);
+ 
        Ok(())

    }



    #[test]
- 
    fn skipped() {
+ 
    fn skipped() -> TestResult<()> {
+ 
        let mock_node = MockNode::new()?;
+ 
        let profile = mock_node.profile()?;

        let event = Event::RefsFetched {

            remote: nid(),

            rid: rid(),
@@ -497,13 +603,16 @@ mod test {
            }],

        };
- 
        let result = CiEvent::from_node_event(&event);
+ 
        let result = CiEvent::from_node_event(&event, &profile);

        assert!(result.is_ok());

        assert_eq!(result.unwrap(), vec![]);
+ 
        Ok(())

    }



    #[test]
- 
    fn branch_created() {
+ 
    fn branch_created() -> TestResult<()> {
+ 
        let mock_node = MockNode::new()?;
+ 
        let profile = mock_node.profile()?;

        let rid = rid();

        let oid = oid();

        let event = Event::RefsFetched {
@@ -514,7 +623,7 @@ mod test {
                oid,

            }],

        };
- 
        let x = CiEvent::from_node_event(&event);
+ 
        let x = CiEvent::from_node_event(&event, &profile);

        eprintln!("result: {x:#?}");

        match x {

            Err(_) => panic!("should succeed"),
@@ -533,10 +642,13 @@ mod test {
            }

            Ok(_) => panic!("empty list of events should not happen"),

        }
+ 
        Ok(())

    }



    #[test]
- 
    fn branch_updated() {
+ 
    fn branch_updated() -> TestResult<()> {
+ 
        let mock_node = MockNode::new()?;
+ 
        let profile = mock_node.profile()?;

        let rid = rid();

        let oid = oid();

        let event = Event::RefsFetched {
@@ -548,7 +660,7 @@ mod test {
                new: oid,

            }],

        };
- 
        let x = CiEvent::from_node_event(&event);
+ 
        let x = CiEvent::from_node_event(&event, &profile);

        eprintln!("result: {x:#?}");

        match x {

            Err(_) => panic!("should succeed"),
@@ -571,10 +683,13 @@ mod test {
            }

            Ok(_) => panic!("empty list of events should not happen"),

        }
+ 
        Ok(())

    }



    #[test]
- 
    fn branch_deleted() {
+ 
    fn branch_deleted() -> TestResult<()> {
+ 
        let mock_node = MockNode::new()?;
+ 
        let profile = mock_node.profile()?;

        let rid = rid();

        let oid = oid();

        let event = Event::RefsFetched {
@@ -585,7 +700,7 @@ mod test {
                oid,

            }],

        };
- 
        let x = CiEvent::from_node_event(&event);
+ 
        let x = CiEvent::from_node_event(&event, &profile);

        eprintln!("result: {x:#?}");

        match x {

            Err(_) => panic!("should succeed"),
@@ -601,10 +716,13 @@ mod test {
            }

            Ok(_) => panic!("empty list of events should not happen"),

        }
+ 
        Ok(())

    }



    #[test]
- 
    fn patch_created() {
+ 
    fn patch_created() -> TestResult<()> {
+ 
        let mock_node = MockNode::new()?;
+ 
        let profile = mock_node.profile()?;

        let rid = rid();

        let patch_id = oid_from(PATCH_ID).into();

        let oid = oid();
@@ -616,7 +734,7 @@ mod test {
                oid,

            }],

        };
- 
        let x = CiEvent::from_node_event(&event);
+ 
        let x = CiEvent::from_node_event(&event, &profile);

        eprintln!("result: {x:#?}");

        match x {

            Err(_) => panic!("should succeed"),
@@ -635,10 +753,13 @@ mod test {
            }

            Ok(_) => panic!("empty list of events should not happen"),

        }
+ 
        Ok(())

    }



    #[test]
- 
    fn patch_updated() {
+ 
    fn patch_updated() -> TestResult<()> {
+ 
        let mock_node = MockNode::new()?;
+ 
        let profile = mock_node.profile()?;

        let rid = rid();

        let patch_id = oid_from(PATCH_ID).into();

        let oid = oid();
@@ -651,7 +772,7 @@ mod test {
                new: oid,

            }],

        };
- 
        let x = CiEvent::from_node_event(&event);
+ 
        let x = CiEvent::from_node_event(&event, &profile);

        eprintln!("result: {x:#?}");

        match x {

            Err(_) => panic!("should succeed"),
@@ -670,6 +791,47 @@ mod test {
            }

            Ok(_) => panic!("empty list of events should not happen"),

        }
+ 
        Ok(())
+ 
    }
+
+ 
    #[test]
+ 
    fn diff_refs() {
+ 
        let oid1 = oid_from("c474d78c37a5c664aba2042db78ec235c0a5d569");
+ 
        let oid2 = oid_from("5b6886f63ff90d61afaeec2f3569a9a1a0984bdb");
+ 
        let foo = RefString::try_from("refs/heads/foo").unwrap();
+ 
        let bar = RefString::try_from("refs/heads/bar").unwrap();
+ 
        let baz = RefString::try_from("refs/heads/baz").unwrap();
+ 
        let xyz = RefString::try_from("refs/heads/xyz").unwrap();
+
+ 
        let old = Refs::from(BTreeMap::from([
+ 
            (foo.clone(), oid1),
+ 
            (bar.clone(), oid1),
+ 
            (xyz.clone(), oid1),
+ 
        ]));
+ 
        let new = Refs::from(BTreeMap::from([
+ 
            (foo.clone(), oid2),
+ 
            (baz.clone(), oid2),
+ 
            (xyz.clone(), oid1),
+ 
        ]));
+ 
        let updates = super::diff_refs(old, new);
+ 
        assert_eq!(
+ 
            updates,
+ 
            [
+ 
                RefUpdate::Updated {
+ 
                    name: foo,
+ 
                    old: oid1,
+ 
                    new: oid2
+ 
                },
+ 
                RefUpdate::Deleted {
+ 
                    name: bar,
+ 
                    oid: oid1,
+ 
                },
+ 
                RefUpdate::Created {
+ 
                    name: baz,
+ 
                    oid: oid2,
+ 
                }
+ 
            ]
+ 
        );

    }

}
@@ -738,9 +900,8 @@ impl ParsedRef {
mod test_parsed_ref {

    use std::str::FromStr;
- 
    use crate::refs::{branch_ref, ref_string};
- 


    use super::*;
+ 
    use crate::refs::{branch_ref, ref_string};



    #[test]

    fn branch() {
modified src/ci_event_source.rs
@@ -10,12 +10,14 @@ use crate::{


pub struct CiEventSource {

    source: NodeEventSource,
+ 
    profile: Profile,

}



impl CiEventSource {
- 
    pub fn new(profile: &Profile) -> Result<Self, CiEventSourceError> {
+ 
    pub fn new(profile: Profile) -> Result<Self, CiEventSourceError> {

        let source = Self {
- 
            source: NodeEventSource::new(profile).map_err(CiEventSourceError::Subscribe)?,
+ 
            source: NodeEventSource::new(&profile).map_err(CiEventSourceError::Subscribe)?,
+ 
            profile,

        };

        logger::ci_event_source_created(&source);

        Ok(source)
@@ -41,8 +43,8 @@ impl CiEventSource {
                Ok(None)

            }

            Ok(Some(event)) => {
- 
                let ci_events =
- 
                    CiEvent::from_node_event(&event).map_err(CiEventSourceError::CiEvent)?;
+ 
                let ci_events = CiEvent::from_node_event(&event, &self.profile)
+ 
                    .map_err(CiEventSourceError::CiEvent)?;

                if !ci_events.is_empty() {

                    logger::ci_event_source_got_events(&ci_events);

                }
modified src/queueadd.rs
@@ -43,7 +43,7 @@ impl QueueAdder {
    fn add_events(&self) -> Result<(), AdderError> {

        let profile = Profile::load().map_err(AdderError::profile)?;
- 
        let mut source = CiEventSource::new(&profile)?;
+ 
        let mut source = CiEventSource::new(profile)?;



        // This loop ends when there's an error, e.g., failure to read an

        // event from the node.