Add a unique, immutable broker run ID to each run, in addition to the — radicle-ci-broker

adapter run ID that may not always be available, and is not guaranteed to be unique.

Use inter-thread channels for threads to notify other threads those other threads have work to do. Event adding thread notifies event processing thread that there are new events, and the event processing thread notifies the report page generation thread that a new run has finished. In addition, the reporting thread reports after a while whether there are any new runs or not, so that the page time stamp gets updated.

 then stdout is exactly ""
 when I run cibtool --db ci-broker.db run list --json
 then stdout contains ""run_id": "xyzzy""
 then stdout contains ""id": "xyzzy""
 ~~~

 when I run cibtool --db x.db run add --id x --repo rad:zwTxygwuz5LDGBq255RA2CbNGrz8 --alias x --url https://x/1 --branch main --commit f1815dde6ae406d8fe3cec0b96c4486766342716 --who x --finished --failure --timestamp 2024-07-09T02:00:00
 when I run cibtool --db x.db run list --json
 then stdout contains "rad:zwTxygwuz5LDGBq255RA2CbNGrz8"
 when I run cibtool --db x.db run list --adapter-run-id abracadabra
 then stdout is exactly ""
 ~~~

 use log::{debug, error};
 use crate::{
     db::{Db, DbError},
     msg::{MessageError, Request, Response},
     run::{Run, RunState},
 };

         self.env.iter().map(|(k, v)| (k.as_ref(), v.as_ref()))
     }
     pub fn run(&self, trigger: &Request, run: &mut Run) -> Result<(), AdapterError> {
     pub fn run(&self, trigger: &Request, run: &mut Run, db: &Db) -> Result<(), AdapterError> {
         debug!("running adapter");
         run.set_state(RunState::Triggered);
         let x = self.run_helper(trigger, run);
         db.update_run(run).map_err(AdapterError::UpdateRun)?;
         let x = self.run_helper(trigger, run, db);
         run.set_state(RunState::Finished);
         db.update_run(run).map_err(AdapterError::UpdateRun)?;
         x
     }
     fn run_helper(&self, trigger: &Request, run: &mut Run) -> Result<(), AdapterError> {
     fn run_helper(&self, trigger: &Request, run: &mut Run, db: &Db) -> Result<(), AdapterError> {
         assert!(matches!(trigger, Request::Trigger { .. }));
         // Spawn the adapter sub-process.

                     if let Some(url) = info_url {
                         run.set_adapter_info_url(&url);
                     }
                     db.update_run(run).map_err(AdapterError::UpdateRun)?;
                 }
                 _ => return Err(AdapterError::NotTriggered(resp)),
             }

             match resp {
                 Response::Finished { result } => {
                     run.set_result(result);
                     db.update_run(run).map_err(AdapterError::UpdateRun)?;
                 }
                 _ => return Err(AdapterError::NotFinished(resp)),
             }

     /// Too many messages from adapter.
     #[error("adapter sent too many messages: first extra is {0:#?}")]
     TooMany(Response),
     /// Could no update run in database.
     #[error("failed to update CI run information in database")]
     UpdateRun(#[source] DbError),
 }
 #[cfg(test)]
 mod test {
     use std::{fs::write, io::ErrorKind};
     use tempfile::tempdir;
     use tempfile::{tempdir, NamedTempFile};
     use radicle::git::Oid;
     use radicle::prelude::RepoId;
     use super::{Adapter, Run};
     use super::{Adapter, Db, Run};
     use crate::{
         adapter::AdapterError,
         msg::{MessageError, Response, RunResult},

         test::{log_in_tests, mock_adapter, trigger_request, TestResult},
     };
     fn db() -> anyhow::Result<Db> {
         let tmp = NamedTempFile::new()?;
         let db = Db::new(tmp.path())?;
         Ok(db)
     }
     fn run() -> anyhow::Result<Run> {
         Ok(Run::new(
             RepoId::from_urn("rad:zwTxygwuz5LDGBq255RA2CbNGrz8")?,

         let bin = tmp.path().join("adapter.sh");
         mock_adapter(&bin, ADAPTER)?;
         let db = db()?;
         let mut run = run()?;
         Adapter::new(&bin).run(&trigger_request()?, &mut run)?;
         Adapter::new(&bin).run(&trigger_request()?, &mut run, &db)?;
         assert_eq!(run.result(), Some(&RunResult::Success));
         Ok(())

         let bin = tmp.path().join("adapter.sh");
         mock_adapter(&bin, ADAPTER)?;
         let db = db()?;
         let mut run = run()?;
         let x = Adapter::new(&bin).run(&trigger_request()?, &mut run);
         let x = Adapter::new(&bin).run(&trigger_request()?, &mut run, &db);
         eprintln!("{x:#?}");
         assert!(x.is_err());
         assert_eq!(run.result(), Some(&RunResult::Failure));

     broker::{Broker, BrokerError},
     config::{Config, ConfigError},
     db::{Db, DbError},
     notif::NotificationChannel,
     pages::{PageBuilder, PageError},
     queueadd::{AdderError, QueueAdderBuilder},
     queueproc::{QueueError, QueueProcessorBuilder},

 impl InsertCmd {
     fn run(&self, args: &Args, config: &Config) -> Result<(), CibError> {
         let mut events_notification = NotificationChannel::default();
         let adder = QueueAdderBuilder::default()
             .events_tx(events_notification.tx())
             .db(args.open_db(config)?)
             .filters(&config.filters)
             .build()

         debug!("default adapter: {adapter:?}");
         broker.set_default_adapter(&adapter);
         let mut event_notifications = NotificationChannel::default();
         event_notifications.tx().send(()).ok();
         let mut run_notifications = NotificationChannel::default();
         let processor = QueueProcessorBuilder::default()
             .events_rx(event_notifications.rx())
             .run_tx(run_notifications.tx())
             .db(db)
             .broker(broker)
             .build()

         if let Some(dirname) = &config.report_dir {
             page.set_output_dir(dirname);
         }
         page.update_in_thread(db, &profile.config.node.alias, false);
         page.update_in_thread(run_notification.rx(), db, &profile.config.node.alias, false);
         let mut broker = Broker::new(config.db()).map_err(CibError::new_broker)?;
         let spec =

     db::{Db, DbError, QueueId},
     event::BrokerEvent,
     msg::{RunId, RunResult},
     notif::NotificationChannel,
     pages::{PageBuilder, PageError},
     run::{Run, RunState, Whence},
 };

 #[derive(Parser)]
 struct AddRun {
     /// Set the run ID.
     /// Set the adapter run ID.
     #[clap(long)]
     id: RunId,
     id: Option<RunId>,
     /// Set alias of node that performed the CI run.
     #[clap(long)]

             who: self.who.clone(),
         };
         let mut run = Run::new(self.repo, &self.alias, whence, self.timestamp.clone());
         run.set_adapter_run_id(RunId::default());
         let id = self.id.clone().unwrap_or_default();
         run.set_adapter_run_id(id);
         if let Some(url) = &self.url {
             run.set_adapter_info_url(url);
         }

             RunResult::Failure
         });
         db.push_run(run)?;
         db.push_run(&run)?;
         Ok(())
     }

 struct ListRuns {
     #[clap(long)]
     json: bool,
     #[clap(long)]
     adapter_run_id: Option<RunId>,
 }
 impl ListRuns {

     fn run(&self, args: &Args) -> Result<(), CibToolError> {
         let db = args.open_db()?;
         let runs = if let Some(wanted) = &self.adapter_run_id {
             db.find_runs(wanted)?
         } else {
             db.get_all_runs()?
         };
         if self.json {
             let runs: Vec<RunInfo> = db.get_all_runs()?.iter().map(RunInfo::from).collect();
             println!(
                 "{}",
                 serde_json::to_string_pretty(&runs).map_err(CibToolError::RunToJson)?
             );
         } else {
             for run in db.get_all_runs()? {
             for run in runs {
                 println!(
                     "{}",
                     "{} {}",
                     run.broker_run_id(),
                     run.adapter_run_id()
                         .map(|id| id.to_string())
                         .unwrap_or("unknown".into())

             .build()?;
         page.set_output_dir(&self.output_dir);
         let thread = page.update_in_thread(db, &profile.config.node.alias, true);
         let mut run_notification = NotificationChannel::default();
         let thread =
             page.update_in_thread(run_notification.rx(), db, &profile.config.node.alias, true);
         thread.join().unwrap()?;
         Ok(())

 use crate::{
     adapter::Adapter,
     db::{Db, DbError},
     msg::{PatchEvent, PushEvent, Request, RunId},
     msg::{PatchEvent, PushEvent, Request},
     run::{Run, Whence},
 };

     pub fn execute_ci(&mut self, trigger: &Request) -> Result<Run, BrokerError> {
         info!("Start CI run");
         debug!("Start CI run on {trigger:#?}");
         let mut run = match trigger {
         let run = match trigger {
             Request::Trigger {
                 common,
                 push,

                     };
                     let mut run = Run::new(*rid, &common.repository.name, whence, now()?);
                     self.db.push_run(&run)?;
                     // We run the adapter, but if that fails, we just
                     // log the error. The `Run` value records the
                     // result of the run.
                     debug!("broker runs adapter");
                     if let Err(e) = adapter.run(trigger, &mut run) {
                     if let Err(e) = adapter.run(trigger, &mut run, &self.db) {
                         error!("failed to run adapter or it failed to run CI: {e}");
                         let mut e = e.source();
                         while let Some(source) = e {

         };
         info!("Finish CI run: {run:?}");
         // If the adapter never gave us a run ID, it has not been set.
         // In that case, we invent one so that it can be used by the
         // database as a key.
         if run.adapter_run_id().is_none() {
             run.set_adapter_run_id(RunId::default());
         }
         self.db.push_run(run.clone())?;
         self.db.update_run(&run)?;
         Ok(run)
     }

         const TABLES: &[&str] = &[
             "CREATE TABLE IF NOT EXISTS counter_test (counter INT)",
             "CREATE TABLE IF NOT EXISTS event_queue (id TEXT PRIMARY KEY, timestamp TEXT, event TEXT)",
             "CREATE TABLE IF NOT EXISTS ci_runs (run_id TEXT PRIMARY KEY, json TEXT)",
             "CREATE TABLE IF NOT EXISTS ci_runs (broker_run_id TEXT PRIMARY KEY, json TEXT)",
         ];
         for table in TABLES.iter() {

     /// Return list of CI runs currently in the database.
     pub fn list_runs(&self) -> Result<Vec<RunId>, DbError> {
         let mut select = self.prepare("SELECT run_id FROM ci_runs")?;
         let mut select = self.prepare("SELECT broker_run_id FROM ci_runs")?;
         let mut run_ids = vec![];

                 Ok(State::Row) => {
                     let run_id: String = select
                         .stmt
                         .read("run_id")
                         .read("broker_run_id")
                         .map_err(|e| DbError::get_run(&select.sql, e))?;
                     let run_id = RunId::from(run_id.as_str());
                     run_ids.push(run_id);

     /// Return a specific CI run, given is id, if one exists.
     pub fn get_run(&self, id: &RunId) -> Result<Option<Run>, DbError> {
         let mut select = self.prepare("SELECT json FROM ci_runs WHERE run_id = :id")?;
         let mut select = self.prepare("SELECT json FROM ci_runs WHERE broker_run_id = :id")?;
         select
             .stmt
             .bind((":id", id.to_string().as_str()))

         Ok(run)
     }
     /// Return a list of broker run IDs that their adapter run ID set
     /// to a given value.
     pub fn find_runs(&self, adapter_runid: &RunId) -> Result<Vec<Run>, DbError> {
         let runs = self
             .get_all_runs()?
             .iter()
             .filter(|run| run.adapter_run_id() == Some(adapter_runid))
             .cloned()
             .collect();
         Ok(runs)
     }
     /// Add a new CI run to the database, returning its id.
     pub fn push_run(&self, run: Run) -> Result<RunId, DbError> {
         let id = run.adapter_run_id().ok_or(DbError::without_id())?.clone();
     pub fn push_run(&self, run: &Run) -> Result<RunId, DbError> {
         let id = run.broker_run_id().clone();
         let json = serde_json::to_string(&run).map_err(DbError::event_to_json)?;
         let mut insert = self.prepare("INSERT INTO ci_runs (run_id, json) VALUES (:id, :json)")?;
         let mut insert =
             self.prepare("INSERT INTO ci_runs (broker_run_id, json) VALUES (:id, :json)")?;
         insert
             .stmt
             .bind((":id", id.to_string().as_str()))

         Ok(id)
     }
     /// Update a CI run in the database.
     pub fn update_run(&self, run: &Run) -> Result<(), DbError> {
         let id = run.broker_run_id().clone();
         let json = serde_json::to_string(&run).map_err(DbError::event_to_json)?;
         let mut update =
             self.prepare("UPDATE ci_runs SET json = :json WHERE broker_run_id = :id")?;
         update
             .stmt
             .bind((":id", id.to_string().as_str()))
             .map_err(|e| DbError::bind(&update.sql, e))?;
         update
             .stmt
             .bind((":json", json.as_str()))
             .map_err(|e| DbError::bind(&update.sql, e))?;
         match update.stmt.next() {
             Ok(_) => (),
             Err(e) => return Err(DbError::update_run(&update.sql, e)),
         }
         Ok(())
     }
     /// Remove a CI run from database, given its id. It's OK if the run is
     /// not in the database, that is just silently ignored.
     pub fn remove_run(&self, id: &RunId) -> Result<(), DbError> {

     #[error("failed to retrieve a CI run from database")]
     GetRun(String, #[source] sqlite::Error),
     #[error("programming error: pushed run does not have its run ID set")]
     WithoutId,
     #[error("failed to insert a CI run into database")]
     PushRun(String, #[source] sqlite::Error),
     #[error("failed to update a CI run in database")]
     UpdateRun(String, #[source] sqlite::Error),
     #[error("failed to remove a CI run from database")]
     RemoveRun(String, #[source] sqlite::Error),
 }

         Self::GetRun(sql.into(), e)
     }
     fn without_id() -> Self {
         Self::WithoutId
     }
     fn push_run(sql: &str, e: sqlite::Error) -> Self {
         Self::PushRun(sql.into(), e)
     }
     fn update_run(sql: &str, e: sqlite::Error) -> Self {
         Self::UpdateRun(sql.into(), e)
     }
     fn remove_run(sql: &str, e: sqlite::Error) -> Self {
         Self::RemoveRun(sql.into(), e)
     }

 pub mod db;
 pub mod event;
 pub mod msg;
 pub mod notif;
 pub mod pages;
 pub mod queueadd;
 pub mod queueproc;

     }
 }
 impl RunId {
     /// Return representation of identifier as a string slice.
     pub fn as_str(&self) -> &str {
         &self.id
     }
 }
 /// The result of a CI run.
 #[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]