feat: add "cibtool cob" to add runs to a job COB — radicle-ci-broker

feat: cache job COB ID lookup

feat: make announcing option for “cibool cob”

 *.html
 /result
 /target
 doc/messages.md
 \ No newline at end of file
 doc/messages.md
 perf.*

       # Ambient only fetches dependencies for the current version.
       # Thus, we can't build the old versions. This is an Ambient
       # problem that we can't fix here. So we skip the upgrade tests.
       export RAD_PASSPHRAE=secret
       rad auth --alias ambient
       env
       cargo test --workspace --no-fail-fast -- --skip upgrade --test-threads 2
       subplot docgen ci-broker.subplot -o doc/ci-broker.html

 culpa = "1.0.2"
 qcheck = { version = "1", default-features = false }
 qcheck-macros = { version = "1", default-features = false }
 [profile.release]
 debug = true

     os::unix::process::ExitStatusExt,
     path::{Path, PathBuf},
     process::{Command, ExitStatus},
     sync::mpsc::Sender,
     sync::{Arc, Mutex, mpsc::Sender},
     time::Duration,
 };

 use url::Url;
 use crate::{
     cob::{create_run, failed, succeeded},
     cob::{KnownJobCobs, failed, succeeded},
     config::AdapterSpec,
     db::{Db, DbError},
     logger,

         self.env.iter().map(|(k, v)| (k.as_ref(), v.as_ref()))
     }
     #[allow(clippy::too_many_arguments)]
     pub fn run(
         &self,
         trigger: &Request,

         run_notification: &NotificationSender,
         max_run_time: Duration,
         child_info: Sender<ChildInfo>,
         known_job_cobs: Arc<Mutex<KnownJobCobs>>,
     ) -> Result<(), AdapterError> {
         run.set_state(RunState::Triggered);
         db.update_run(run).map_err(AdapterError::UpdateRun)?;
         let x = self.run_helper(trigger, run, db, run_notification, max_run_time, child_info);
         let x = self.run_helper(
             trigger,
             run,
             db,
             run_notification,
             max_run_time,
             child_info,
             known_job_cobs,
         );
         run.set_state(RunState::Finished);
         db.update_run(run).map_err(AdapterError::UpdateRun)?;

         let mut outcome = MaybeResult::default();
         if let Err(err) = self.read_stdout(run, db, run_notification, stdout) {
         if let Err(err) = self.read_stdout(run, db, run_notification, stdout, known_job_cobs) {
             outcome.set_error(err);
         }

         db: &Db,
         run_notification: &NotificationSender,
         mut stdout: RealtimeLines,
         known_job_cobs: Arc<Mutex<KnownJobCobs>>,
     ) -> Result<(), AdapterError> {
         #[allow(clippy::unwrap_used)]
         let no_url = Url::parse("https://no.url.example.com").unwrap();

                     // there's nothing useful we can do about it, as long as we
                     // let CI run, which want to do.
                     let url = url.unwrap_or(no_url);
                     create_run(
                         run.repo_id(),
                         run.whence().oid(),
                         run.broker_run_id().clone(),
                         &url,
                     );
                     if let Ok(mut known) = known_job_cobs.lock() {
                         known.create_run(
                             run.repo_id(),
                             run.whence().oid(),
                             run.broker_run_id().clone(),
                             &url,
                             false,
                         );
                     }
                 }
                 _ => {
                     return Err(AdapterError::NotTriggered(resp));

     /// Can't write adapter config.
     #[error("failed to write adapter configuration")]
     AdapterConfigWrite(#[source] std::io::Error),
     /// Can't lock mutex.
     #[error("failed to lock mutex")]
     Mutex,
 }
 #[cfg(test)]
 mod test {
     use super::*;
     use std::{fs::write, io::ErrorKind, path::PathBuf, sync::mpsc::channel, time::Duration};
     use tempfile::{NamedTempFile, TempDir, tempdir};

     use super::{Adapter, Db, Run};
     use crate::{
         adapter::AdapterError,
         cob::KnownJobCobs,
         msg::{MessageError, Response, RunId, RunResult},
         notif::NotificationChannel,
         run::{RunBuilder, Whence},

     }
     #[allow(clippy::unwrap_used)]
     fn known() -> Arc<Mutex<KnownJobCobs>> {
         Arc::new(Mutex::new(KnownJobCobs::new().unwrap()))
     }
     #[allow(clippy::unwrap_used)]
     fn adapter(tmp: &TempDir, shell: &'static str) -> PathBuf {
         let bin = tmp.path().join("adapter.sh");
         mock_adapter(&bin, shell).unwrap();

         let (pid_tx, _) = channel();
         let mut channel = NotificationChannel::new_run();
         let sender = channel.tx()?;
         Adapter::new(&bin).run(&trigger_request()?, &mut run, &db, &sender, MAX, pid_tx)?;
         Adapter::new(&bin).run(
             &trigger_request()?,
             &mut run,
             &db,
             &sender,
             MAX,
             pid_tx,
             known(),
         )?;
         assert_eq!(run.result(), Some(&RunResult::Success));
         Ok(())

     #[clap(hide = true)]
     Message(cibtoolcmd::MessageCmd),
     Log(cibtoolcmd::LogCmd),
     #[clap(hide = true)]
     Cob(cibtoolcmd::CobCmd),
 }
 impl Subcommand for Cmd {

             Self::Timeout(x) => x.run(args),
             Self::Message(x) => x.run(args),
             Self::Log(x) => x.run(args),
             Self::Cob(x) => x.run(args),
         }
     }
 }

 #[derive(Debug, thiserror::Error)]
 enum CibToolError {
     #[error("failed to create cache of job COBs")]
     KnownJobCobs(#[source] radicle_ci_broker::cob::JobError),
     #[error("failed to look up node profile")]
     Profile(#[source] radicle::profile::Error),

     #[error(transparent)]
     Run(#[from] cibtoolcmd::RunError),
     #[error(transparent)]
     Cob(#[from] radicle_ci_broker::cob::JobError),
 }

 use std::sync::{Arc, Mutex};
 use url::Url;
 use radicle::{git::Oid, identity::RepoId};
 use radicle_ci_broker::cob::*;
 use super::*;
 #[derive(Parser)]
 pub struct CobCmd {
     /// ID of repository.
     repo_id: RepoId,
     /// Git object ID.
     oid: Oid,
     /// Create this many runs in job COB.
     n: usize,
     /// Announce changes?
     #[clap(long)]
     announce: bool,
 }
 impl Leaf for CobCmd {
     fn run(&self, _args: &Args) -> Result<(), CibToolError> {
         let url = Url::parse("https://liw.fi").unwrap();
         let run_id = RunId::default();
         let known = Arc::new(Mutex::new(
             KnownJobCobs::new().map_err(CibToolError::KnownJobCobs)?,
         ));
         for i in 0..self.n {
             println!("{i}");
             let mut known = known.lock().unwrap();
             known.create_run(self.repo_id, self.oid, run_id.clone(), &url, self.announce);
         }
         Ok(())
     }
 }

 mod log;
 pub use log::*;
 mod cob;
 pub use cob::*;

         // it. We won't do anything about a failure, as there's
         // nothing useful we can do about it, as long as we let CI
         // run, which want to do.
         if let Some(job_id) = create_job(trigger.repo(), oid) {
             run.set_job_id(job_id);
         if let Ok(mut known) = known_job_cobs.lock() {
             if let Some(job_id) = known.create_job(trigger.repo(), oid) {
                 run.set_job_id(job_id);
             }
         }
         // We run the adapter, but if that fails, we just

         let mut channel = NotificationChannel::new_run();
         let sender = channel.tx()?;
         let x = broker.execute_ci(&adapter, &trigger, &sender, pid_tx);
         let x = broker.execute_ci(&adapter, &trigger, &sender, pid_tx, known());
         assert!(x.is_ok());
         let run = x?;
         assert_eq!(run.adapter_run_id(), Some(&RunId::from("xyzzy")));

 //! the creation of a "job COB" for every CI run, and updating it when
 //! the run ends so that the COB captures the result of the run.
 use std::{collections::BTreeSet, str::FromStr, time::Duration};
 use std::{
     collections::{BTreeSet, HashMap},
     str::FromStr,
     time::Duration,
 };
 use url::Url;
 use uuid::Uuid;

 use crate::{logger, msg::RunId};
 /// Create a new job COB, for a given Git object in a given
 /// repository. The new job can be found later by using the same oid,
 /// it is not necessary to keep the job id (indeed, that is not
 /// returned). Separate instances of the CI broker have no way of
 /// communicating the job id.
 pub fn create_job(repo_id: RepoId, oid: Oid) -> Option<JobId> {
     fn fallible_create(repo_id: RepoId, oid: Oid) -> Result<JobId, JobError> {
         let profile = profile()?;
         let repo = repository(&profile, repo_id)?;
         let signer = profile.signer().map_err(JobError::Signer)?;
 /// Lookup cache for job COBs.
 ///
 /// This assumes job COBs are not removed from the node.
 pub struct KnownJobCobs {
     profile: Profile,
     known: HashMap<(RepoId, Oid), JobId>,
 }
 impl KnownJobCobs {
     /// Create a new [`KnownJobCobs`].
     pub fn new() -> Result<Self, JobError> {
         let profile = Profile::load().map_err(JobError::Profile)?;
         Ok(Self {
             profile,
             known: HashMap::new(),
         })
     }
     /// Look up job COB for a specific commit.
     pub fn create_job(&mut self, repo_id: RepoId, oid: Oid) -> Option<JobId> {
         let key = (repo_id, oid);
         if let Some(job_id) = self.known.get(&key) {
             Some(*job_id)
         } else if let Ok(job_id) = self.fallible_create_job(repo_id, oid) {
             self.known.insert(key, job_id);
             Some(job_id)
         } else {
             None
         }
     }
     fn fallible_create_job(&self, repo_id: RepoId, oid: Oid) -> Result<JobId, JobError> {
         let repo = repository(&self.profile, repo_id)?;
         let signer = self.profile.signer().map_err(JobError::Signer)?;
         let mut jobs = jobs(&repo)?;
         match job_for_commit(&jobs, oid) {
             Err(JobError::NoJob(_)) => {
                 let job = jobs.create(oid, &signer).map_err(JobError::CreateJob)?;
                 announce(&profile, repo_id)?;
                 announce(&self.profile, repo_id, false)?;
                 logger::job_create(&repo_id, &oid, job.id());
                 Ok(*job.id())
             }

         }
     }
     fallible_create(repo_id, oid).ok()
 }
     /// Create a new run for an existing job. The run id should be the one
     /// assigned by the CI broker, not the one by the adapter. The log URL
     /// has to be the from the adapter.
     pub fn create_run(
         &mut self,
         repo_id: RepoId,
         oid: Oid,
         run_id: RunId,
         url: &Url,
         announce: bool,
     ) {
         if let Some(job_id) = self.create_job(repo_id, oid) {
             if let Err(err) = self.fallible_create_run(job_id, repo_id, run_id, url, announce) {
                 logger::job_failure(
                     "failed to add a run to a job COB",
                     &repo_id,
                     &oid,
                     Some(&err),
                 );
             }
         }
     }
 /// Create a new run for an existing job. The run id should be the one
 /// assigned by the CI broker, not the one by the adapter. The log URL
 /// has to be the from the adapter.
 pub fn create_run(repo_id: RepoId, oid: Oid, run_id: RunId, url: &Url) {
     fn fallible_crate(repo_id: RepoId, oid: Oid, run_id: RunId, url: &Url) -> Result<(), JobError> {
     fn fallible_create_run(
         &self,
         job_id: JobId,
         repo_id: RepoId,
         run_id: RunId,
         url: &Url,
         a: bool,
     ) -> Result<(), JobError> {
         let uuid = Uuid::from_str(run_id.as_str()).map_err(JobError::Uuid)?;
         let profile = profile()?;
         let repo = repository(&profile, repo_id)?;
         let signer = profile.signer().map_err(JobError::Signer)?;
         let repo = repository(&self.profile, repo_id)?;
         let signer = self.profile.signer().map_err(JobError::Signer)?;
         let mut jobs = jobs(&repo)?;
         let job_id = job_for_commit(&jobs, oid)?;
         let mut job = jobs.get_mut(&job_id).map_err(JobError::GetJobMut)?;
         job.run(uuid, url.clone(), &signer)
             .map_err(JobError::AddRun)?;
         announce(&profile, repo_id)?;
         announce(&self.profile, repo_id, a)?;
         logger::job_run_create(job_id, uuid);
         Ok(())
     }
     if let Err(err) = fallible_crate(repo_id, oid, run_id, url) {
         logger::job_failure(
             "failed to add a run to a job COB",
             &repo_id,
             &oid,
             Some(&err),
         );
     }
 }
 /// Mark a run as having finished successfully.

     let mut job = jobs.get_mut(&job_id).map_err(JobError::GetJobMut)?;
     job.finish(uuid, reason, &signer)
         .map_err(JobError::Finish)?;
     announce(&profile, repo_id)?;
     announce(&profile, repo_id, true)?;
     logger::job_run_finished(job_id, uuid, reason);
     Ok(())

     Err(JobError::NoJob(wanted))
 }
 fn announce(profile: &Profile, repo_id: RepoId) -> Result<(), JobError> {
     const TIMEOUT: Duration = Duration::from_millis(5000);
     let mut node = Node::new(profile.home.socket());
 fn announce(profile: &Profile, repo_id: RepoId, announce: bool) -> Result<(), JobError> {
     if announce {
         const TIMEOUT: Duration = Duration::from_millis(5000);
         let mut node = Node::new(profile.home.socket());
         let (synced, unsynced) = node.seeds(repo_id).map_err(JobError::Seeds)?.iter().fold(
             (BTreeSet::new(), BTreeSet::new()),
             |(mut synced, mut unsynced), seed| {
                 if seed.is_synced() {
                     synced.insert(seed.nid);
                 } else {
                     unsynced.insert(seed.nid);
                 }
                 (synced, unsynced)
             },
         );
     let (synced, unsynced) = node.seeds(repo_id).map_err(JobError::Seeds)?.iter().fold(
         (BTreeSet::new(), BTreeSet::new()),
         |(mut synced, mut unsynced), seed| {
             if seed.is_synced() {
                 synced.insert(seed.nid);
             } else {
                 unsynced.insert(seed.nid);
             }
             (synced, unsynced)
         },
     );
     let announcer = Announcer::new(AnnouncerConfig::public(
         *profile.id(),
         ReplicationFactor::MustReach(1),
         BTreeSet::new(),
         synced,
         unsynced,
     ))
     .map_err(|_| JobError::Announcer)?;
     node.announce(repo_id, TIMEOUT, announcer, |_, _| ())
         .map_err(JobError::Announce)?;
         let announcer = Announcer::new(AnnouncerConfig::public(
             *profile.id(),
             ReplicationFactor::MustReach(1),
             BTreeSet::new(),
             synced,
             unsynced,
         ))
         .map_err(|_| JobError::Announcer)?;
         node.announce(repo_id, TIMEOUT, announcer, |_, _| ())
             .map_err(JobError::Announce)?;
     }
     Ok(())
 }

     adapter::{Adapter, Adapters},
     broker::{Broker, BrokerError},
     ci_event::{CiEvent, CiEventV1},
     cob::KnownJobCobs,
     db::{Db, DbError, QueuedCiEvent},
     filter::{EventFilter, Trigger},
     logger,

             current: CurrentlyPicked::default(),
             child_pid_tx,
             child_pid_rx,
             known_job_cobs: Arc::new(Mutex::new(
                 KnownJobCobs::new().map_err(QueueError::KnownJobCobs)?,
             )),
         })
     }

     current: CurrentlyPicked,
     child_pid_tx: Sender<ChildInfo>,
     child_pid_rx: Receiver<ChildInfo>,
     known_job_cobs: Arc<Mutex<KnownJobCobs>>,
 }
 impl QueueProcessor {

                             let p = self.processor()?;
                             let repoid = qe.event().repository().copied();
                             self.current.insert(qe.event().repository());
                             let h = spawn(move || p.pick_and_process_one(qe, adapters));
                             let known = self.known_job_cobs.clone();
                             let h = spawn(move || p.pick_and_process_one(qe, adapters, known));
                             handles.push((repoid, h));
                         }
                         Ok(None) => {}

         &self,
         qe: QueuedCiEvent,
         adapters: Vec<Adapter>,
         known_job_cobs: Arc<Mutex<KnownJobCobs>>,
     ) -> Result<MaybeShutdown, QueueError> {
         for adapter in adapters.iter() {
             self.run_tx.notify()?;

                     let trigger = trigger?;
                     self.broker
                         .execute_ci(adapter, &trigger, &self.run_tx, self.child_pid_tx.clone())
                         .execute_ci(
                             adapter,
                             &trigger,
                             &self.run_tx,
                             self.child_pid_tx.clone(),
                             known_job_cobs.clone(),
                         )
                         .map_err(QueueError::execute_ci)?;
                 }
             }