refactor: store timeout in receiver, separate event/run senders — radicle-ci-broker

Add new NotificationChannel constructors, for different kinds of channels. Store the receive timeout in the receiver; it depends on kind of channel.

Also add methods to send and receive notifications to senders and receivers. This hides the std::sync::mpsc machinery better, and makes the types easier to use.

Signed-off-by: Lars Wirzenius liw@liw.fi

feat: notify page generation about changes to runs more often

Change src/adapter.rs to send notification via the run notification channel when the adapter is spawned, and when the adapter sends a “triggered” message.

Signed-off-by: Lars Wirzenius liw@liw.fi

     db::{Db, DbError},
     logger,
     msg::{MessageError, Request, Response},
     notif::NotificationSender,
     run::{Run, RunState},
 };

         self.env.iter().map(|(k, v)| (k.as_ref(), v.as_ref()))
     }
     pub fn run(&self, trigger: &Request, run: &mut Run, db: &Db) -> Result<(), AdapterError> {
     pub fn run(
         &self,
         trigger: &Request,
         run: &mut Run,
         db: &Db,
         run_notification: &NotificationSender,
     ) -> Result<(), AdapterError> {
         run.set_state(RunState::Triggered);
         db.update_run(run).map_err(AdapterError::UpdateRun)?;
         let x = self.run_helper(trigger, run, db);
         let x = self.run_helper(trigger, run, db, run_notification);
         run.set_state(RunState::Finished);
         db.update_run(run).map_err(AdapterError::UpdateRun)?;

         x
     }
     fn run_helper(&self, trigger: &Request, run: &mut Run, db: &Db) -> Result<(), AdapterError> {
     fn run_helper(
         &self,
         trigger: &Request,
         run: &mut Run,
         db: &Db,
         run_notification: &NotificationSender,
     ) -> Result<(), AdapterError> {
         assert!(matches!(trigger, Request::Trigger { .. }));
         // Spawn the adapter sub-process.

             .spawn()
             .map_err(|e| AdapterError::SpawnAdapter(self.bin.clone(), e))?;
         run_notification.notify()?;
         // Write the request to trigger a run to the child's stdin.
         // Then close the pipe to prevent the child from trying to
         // read another message that will never be sent.

         if let Some(line) = lines.next() {
             let line = line.map_err(AdapterError::ReadLine)?;
             let resp = Response::from_str(&line).map_err(AdapterError::ParseResponse)?;
             run_notification.notify()?;
             match resp {
                 Response::Triggered { run_id, info_url } => {
                     run.set_state(RunState::Running);

     /// Could no update run in database.
     #[error("failed to update CI run information in database")]
     UpdateRun(#[source] DbError),
     /// Could not send notification of changes to CI runs.
     #[error(transparent)]
     Notif(#[from] crate::notif::NotificationError),
 }
 #[cfg(test)]

     use crate::{
         adapter::AdapterError,
         msg::{MessageError, Response, RunResult},
         notif::NotificationChannel,
         run::Whence,
         test::{mock_adapter, trigger_request, TestResult},
     };

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

         let db = db()?;
         let mut run = run()?;
         let x = Adapter::new(&bin).run(&trigger_request()?, &mut run, &db);
         let mut channel = NotificationChannel::new_run();
         let sender = channel.tx()?;
         let x = Adapter::new(&bin).run(&trigger_request()?, &mut run, &db, &sender);
         eprintln!("{x:#?}");
         assert!(x.is_err());
         assert_eq!(run.result(), Some(&RunResult::Failure));

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

         logger::adapter_config(config);
         broker.set_default_adapter(&adapter);
         let mut event_notifications = NotificationChannel::default();
         let mut event_notifications = NotificationChannel::new_event();
         event_notifications
             .tx()
             .map_err(CibError::notification)?
             .send(())
             .notify()
             .ok();
         let mut run_notifications = NotificationChannel::default();
         let mut run_notifications = NotificationChannel::new_run();
         let db = args.open_db(config)?;
         let processor = QueueProcessorBuilder::default()

         let mut page = StatusPage::default();
         page.set_output_dir(&self.output_dir);
         let mut run_notification = NotificationChannel::default();
         let mut run_notification = NotificationChannel::new_run();
         let thread = page.update_in_thread(
             run_notification.rx().map_err(CibToolError::Notification)?,
             profile,

     }
     #[allow(clippy::result_large_err)]
     pub fn execute_ci(&mut self, trigger: &Request) -> Result<Run, BrokerError> {
     pub fn execute_ci(
         &mut self,
         trigger: &Request,
         run_notification: &NotificationSender,
     ) -> Result<Run, BrokerError> {
         logger::broker_start_run(trigger);
         let run = match trigger {
             Request::Trigger {

                     // We run the adapter, but if that fails, we just
                     // log the error. The `Run` value records the
                     // result of the run.
                     if let Err(e) = adapter.run(trigger, &mut run, &self.db) {
                     if let Err(e) = adapter.run(trigger, &mut run, &self.db, run_notification) {
                         logger::error("failed to run adapter or it failed to run CI", &e);
                     }

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

 //! Notification channel between threads.
 use std::sync::mpsc::{channel, Receiver, Sender};
 use std::{
     sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender},
     time::Duration,
 };
 // Timeout when receiving notification about a new event.
 const EVENT_RECV_TIMEOUT: Duration = Duration::from_secs(1);
 // Maximum interval between update HTML report pages.
 const UPDATE_IMTERVAL: Duration = Duration::from_secs(1);
 /// Channel endpoint for sending notifications.
 pub type NotificationSender = Sender<()>;
 pub struct NotificationSender {
     sender: Sender<()>,
 }
 impl NotificationSender {
     fn new(sender: Sender<()>) -> Self {
         Self { sender }
     }
     pub fn notify(&self) -> Result<(), NotificationError> {
         self.sender.send(()).map_err(NotificationError::Send)
     }
 }
 /// Channel endpoint for receiving notifications.
 pub type NotificationReceiver = Receiver<()>;
 pub struct NotificationReceiver {
     receiver: Receiver<()>,
     max_wait: Duration,
 }
 impl NotificationReceiver {
     fn new(receiver: Receiver<()>, max_wait: Duration) -> Self {
         Self { receiver, max_wait }
     }
     pub fn wait_for_notification(&self) -> Result<(), RecvTimeoutError> {
         self.receiver.recv_timeout(self.max_wait)
     }
 }
 /// Notification channel.
 ///

     rx: Option<NotificationReceiver>,
 }
 impl Default for NotificationChannel {
     fn default() -> Self {
 impl NotificationChannel {
     /// Construct a channel for notifying about new events.
     pub fn new_event() -> Self {
         Self::new(EVENT_RECV_TIMEOUT)
     }
     /// Construct a channel for notifying about new CI runs.
     pub fn new_run() -> Self {
         Self::new(UPDATE_IMTERVAL)
     }
     fn new(max_wait: Duration) -> Self {
         let (tx, rx) = channel();
         Self {
             tx: Some(tx),
             rx: Some(rx),
             tx: Some(NotificationSender::new(tx)),
             rx: Some(NotificationReceiver::new(rx, max_wait)),
         }
     }
 }

     #[error("sender end point already extracted")]
     Sender,
     #[error("error sending to channel")]
     Send(#[source] std::sync::mpsc::SendError<()>),
 }

                     return Ok(());
                 }
                 match run_rx.recv_timeout(UPDATE_INTERVAL) {
                 match run_rx.wait_for_notification() {
                     Ok(_) => (),
                     Err(RecvTimeoutError::Timeout) => (),
                     Err(RecvTimeoutError::Disconnected) => {

     fn push_event(&self, e: CiEvent) -> Result<(), AdderError> {
         self.db.push_queued_ci_event(e)?;
         self.events_tx.send(()).map_err(|_| AdderError::Send)?;
         self.events_tx.notify().map_err(|_| AdderError::Send)?;
         Ok(())
     }
 }

 use std::{
     sync::mpsc::RecvTimeoutError,
     thread::{spawn, JoinHandle},
     time::Duration,
 };
 use radicle::Profile;

     notif::{NotificationReceiver, NotificationSender},
 };
 const RECV_TIMEOUT: Duration = Duration::from_secs(1);
 #[derive(Default)]
 pub struct QueueProcessorBuilder {
     db: Option<Db>,

         let mut done = false;
         while !done {
             while let Some(qe) = self.pick_event()? {
                 self.run_tx.notify()?;
                 logger::queueproc_picked_event(qe.id(), &qe);
                 done = self.process_event(qe.event())?;
                 self.drop_event(qe.id())?;
                 self.run_tx.send(()).map_err(|_| QueueError::NotifyRun)?;
                 self.run_tx.notify()?;
             }
             match self.events_rx.recv_timeout(RECV_TIMEOUT) {
             match self.events_rx.wait_for_notification() {
                 Ok(_) => {}
                 Err(RecvTimeoutError::Timeout) => {}
                 Err(RecvTimeoutError::Disconnected) => {

                     .build_trigger_from_ci_event()
                     .map_err(|e| QueueError::build_trigger(event, e))?;
                 self.broker
                     .execute_ci(&trigger)
                     .execute_ci(&trigger, &self.run_tx)
                     .map_err(QueueError::execute_ci)?;
                 Ok(false)
             }

     #[error("failed to run CI")]
     ExecuteCi(#[source] BrokerError),
     #[error("failed to send notification about new run")]
     NotifyRun,
     #[error(transparent)]
     NotifyRun(#[from] crate::notif::NotificationError),
 }
 impl QueueError {