add concurrent adapters — radicle-ci-broker

One adapter at a time per repository.

Not all commits in this patch build or pass tests separately.

 | field | summary |
 |:------|:--------|
 | `adapters` | list of CI adapters |
 | `concurrent_adapters` | max number of adapters to run at the same time |
 | `db` | database |
 | `default_adapter` |  this will become deprecated, use `triggers` instead |
 | `filters` | this will become deprecated, use `triggers` instead |

 use std::{
     collections::HashMap,
     ffi::OsStr,
     os::unix::process::ExitStatusExt,
     path::{Path, PathBuf},
     process::Command,
     process::{Command, ExitStatus},
     time::Duration,
 };

     notif::NotificationSender,
     run::{Run, RunState},
     sensitive::Sensitive,
     timeoutcmd::{LineReceiver, TimeoutCommand, TimeoutError},
     timeoutcmd::{RealtimeLines, TimeoutCommand, TimeoutError},
 };
 const NOT_EXITED: i32 = 999;
 /// The set of all configured adapters.
 #[derive(Clone)]
 pub struct Adapters {

         let mut cmd = Command::new(&self.bin);
         cmd.envs(self.envs());
         let mut child = TimeoutCommand::new(max_run_time);
         let stdout = child.stdout();
         child.feed_stdin(trigger.to_string().as_bytes());
         let child = child.spawn(cmd).map_err(|err| match err {
             TimeoutError::Spawn(_, err) => AdapterError::SpawnAdapter(self.bin.clone(), err),
             _ => AdapterError::TimeoutCommand(err),
         })?;
         let child = match child.spawn(cmd) {
             Ok(child) => child,
             Err(TimeoutError::Spawn(_, err)) => {
                 Err(AdapterError::SpawnAdapter(self.bin.clone(), err))?
             }
             Err(err) => Err(AdapterError::TimeoutCommand(err))?,
         };
         run_notification.notify()?;
         let stdout = child.stdout();
         let stderr = child.stderr();
         let mut outcome = MaybeResult::default();
         if let Err(err) = self.read_stdout(run, db, run_notification, stdout) {
             outcome.set_error(err);
         }
         self.read_stderr(stderr);
         if outcome.has_error() {
             child.kill().ok();
         }
         let wait_result = child.wait().expect("FIXME");
         if wait_result.timed_out() {
             logger::adapter_did_not_exit_voluntarily();
             outcome.set_error(AdapterError::Failed(NOT_EXITED));
         } else if let Some(exit) = wait_result.status().code() {
             logger::adapter_result(exit);
             if exit != 0 {
                 outcome.set_error(AdapterError::Failed(exit));
             }
         let result = if outcome.has_error() {
             child.kill()
         } else {
             logger::adapter_did_not_exit();
             outcome.set_error(AdapterError::Signal);
             child.wait()
         };
         match result {
             Ok(finished) => {
                 let stderr = finished.stderr();
                 self.log_stderr(stderr);
                 let exit = finished.exit_code();
                 logger::adapter_result(exit);
                 if !exit.success() {
                     if let Some(signal) = exit.signal() {
                         outcome.set_error(AdapterError::Signal(signal));
                     } else {
                         outcome.set_error(AdapterError::Failed(exit));
                     }
                 }
             }
             Err(TimeoutError::TimedOut) => {
                 logger::adapter_did_not_exit_voluntarily();
                 outcome.set_error(AdapterError::FailedNotExited);
             }
             Err(err) => {
                 logger::adapter_did_not_exit(err);
                 outcome.set_error(AdapterError::Signal(9));
             }
         }
         if let Some(err) = outcome.error() {

         run: &mut Run,
         db: &Db,
         run_notification: &NotificationSender,
         stdout: &LineReceiver,
         mut stdout: RealtimeLines,
     ) -> Result<(), AdapterError> {
         if let Some(line) = stdout.line() {
             let resp = Response::from_str(&line).map_err(AdapterError::ParseResponse)?;

         Ok(())
     }
     fn read_stderr(&self, stderr: &LineReceiver) {
         while let Some(line) = stderr.line() {
             logger::adapter_stderr_line(&line);
     fn log_stderr(&self, stderr: &[u8]) {
         for line in String::from_utf8_lossy(stderr).lines() {
             logger::adapter_stderr_line(line);
         }
     }
 }

     Wait(#[source] std::io::Error),
     /// Child process failed.
     #[error("child process failed with wait status {0}")]
     Failed(i32),
     #[error("child process failed with wait status {0:?}")]
     Failed(ExitStatus),
     /// Child process failed: didn't exit.
     #[error("child process failed without exiting")]
     FailedNotExited,
     /// Child process was killed.
     #[error("child process terminated by signal")]
     Signal,
     #[error("child process terminated by signal {0}")]
     Signal(i32),
     /// First message is not `Response::Triggered`
     #[error("adapter's first message is not 'triggered', but {0:?}")]

         let mut channel = NotificationChannel::new_run();
         let sender = channel.tx()?;
         let x = Adapter::new(&bin).run(&trigger_request()?, &mut run, &db, &sender, MAX);
         eprintln!("{x:#?}");
         assert!(matches!(x, Err(AdapterError::Signal)));
         eprintln!("Adapter::run result: {x:#?}");
         if let Err(AdapterError::Failed(x)) = x {
             use std::os::unix::process::ExitStatusExt;
             eprintln!("Adapter::run result: signal={:?}", x.signal());
         }
         assert!(matches!(x, Err(AdapterError::Signal(9))));
         Ok(())
     }

             .filters(config.filters())
             .triggers(&config.to_triggers())
             .adapters(&adapters)
             .concurrent_adapters(config.concurrent_adapters())
             .build()
             .map_err(CibError::process_queue)?;
         let thread = processor.process_in_thread();

 use radicle::patch::PatchId;
 use radicle_ci_broker::refs::branch_ref;
 #[allow(unused_imports)] // FIXME
 use radicle_ci_broker::{
     filter::EventFilter, node_event_source::NodeEventSource, refs::ref_string,
     util::read_file_as_objectid,

         cmd.arg("-c").arg(&self.script);
         let mut to = TimeoutCommand::new(Duration::from_secs(self.timeout));
         let mut stdout = to.stdout();
         if let Some(bytes) = self.generate {
             let mut stdin: Vec<u8> = vec![];

         println!("spawn child");
         let running = to.spawn(cmd)?;
         if let Some(secs) = self.kill_after {
             sleep(Duration::from_secs(secs));
             running.kill().unwrap();
         }
         let mut stdout_bytes = 0;
         if !self.fill_buffers {
             let stdout = running.stdout();
             while let Some(line) = stdout.line() {
                 stdout_bytes += line.as_bytes().len();
                 if self.verbose {
                     println!("stdout: {line:?}");
         let tor = if let Some(secs) = self.kill_after {
             sleep(Duration::from_secs(secs));
             running.kill()?
         } else {
             if !self.fill_buffers {
                 while let Some(line) = stdout.line() {
                     stdout_bytes += line.as_bytes().len();
                     if self.verbose {
                         println!("stdout: {line:?}");
                     }
                 }
                 println!("finished reading stdout");
             }
             println!("finished reading stdout");
             let stderr = running.stderr();
             while let Some(line) = stderr.line() {
                 if self.verbose {
                     println!("stderr: {line:?}");
                 }
             running.wait()?
         };
         let stderr = tor.stderr();
         for line in stderr {
             if self.verbose {
                 println!("stderr: {line:?}");
             }
             println!("finished reading stderr");
         }
         println!("finished reading stderr");
         let tor = running.wait()?;
         let elapsed = started.elapsed();
         let speed = (stdout_bytes as f64) / elapsed.as_secs_f64();
         println!("stdout bytes: {stdout_bytes}");
         println!("duration: {} ms", elapsed.as_millis());
         println!("speed: {:.0} B/s", speed);
         println!("exit: {}", tor.status());
         println!("timed out? {}", tor.timed_out());
         println!("exit: {}", tor.exit_code());
         Ok(())
     }

         })
     }
     pub fn max_run_time(&self) -> Duration {
         self.max_run_time
     }
     pub fn db(&self) -> &Db {
         &self.db
     }
     #[allow(clippy::result_large_err)]
     pub fn execute_ci(
         &mut self,
         &self,
         adapter: &Adapter,
         trigger: &Request,
         run_notification: &NotificationSender,

         let tmp = tempdir()?;
         let db = tmp.path().join("db.db");
         let mut broker = broker(&db)?;
         let broker = broker(&db)?;
         let trigger = trigger_request()?;

 const DEFAULT_QUEUE_LEN_INTERVAL: Duration = Duration::from_secs(3600);
 const DEFAULT_STATUS_PAGE_UPDATE_INTERVAL: u64 = 10;
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
 struct NonzeroCount {
     count: Option<usize>,
 }
 impl NonzeroCount {
     fn count(&self) -> usize {
         match self.count {
             None | Some(1) => 1,
             Some(n) => n,
         }
     }
 }
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct Config {

     #[serde(deserialize_with = "deserialize_duration")]
     #[serde(default = "default_queue_len_interval")]
     queue_len_interval: Duration,
     #[serde(default)]
     concurrent_adapters: NonzeroCount,
 }
 fn default_max_run_time() -> Duration {

         self.report_dir.as_deref()
     }
     pub fn concurrent_adapters(&self) -> usize {
         self.concurrent_adapters.count()
     }
     pub fn to_adapters(&self) -> Result<Adapters, AdapterError> {
         Adapters::new(&self.adapters, self.default_adapter.as_deref())
     }

 pub mod node_event_source;
 pub mod notif;
 pub mod pages;
 pub mod pull_queue;
 pub mod queueadd;
 pub mod queueproc;
 pub mod refs;

 use std::{path::Path, process::ExitStatus, time::Duration};
 use clap::ValueEnum;
 use radicle::{git::Oid, identity::RepoId, node::Event, patch::PatchId};
 use radicle::{identity::RepoId, node::Event, patch::PatchId};
 use serde_json::Value;
 use tracing::{debug, error, info, trace, warn, Level};
 use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};

     queueadd::AdderError,
     queueproc::QueueError,
     run::Run,
     timeoutcmd::TimeoutError,
 };
 #[derive(Debug, thiserror::Error)]

     );
 }
 pub fn queueproc_start() {
 pub fn queueproc_start(concurrent_adapters: usize) {
     info!(
         msg_id = ?Id::QueueProcStart,
         kind = %Kind::Startup,
         concurrent_adapters,
         "start thread to process events until a shutdown event"
     );
 }

     }
 }
 pub fn queueproc_start_result_receiver() {
     info!(
         msg_id = ?Id::QueueProcStart,
         kind = %Kind::Startup,
         "start thread to process results of CI runs"
     );
 }
 pub fn queueproc_end_result_receiver() {
     info!(
         msg_id = ?Id::QueueProcStart,
         kind = %Kind::Startup,
         "end thread to process results of CI runs"
     );
 }
 pub fn queueproc_channel_disconnect() {
     info!(
         msg_id = ?Id::QueueProcDisconnected,

         ?id,
         ?event,
         ?adapter,
         "picked event from queue"
         "running adapter on event picked from queue"
     );
 }

     );
 }
 pub fn queueproc_remove_event(id: &QueueId) {
 pub fn queueproc_remove_event(event: &QueuedCiEvent) {
     info!(
         msg_id = ?Id::QueueProcRemoveEvent,
         kind = %Kind::Debug,
         ?id,
         ?event,
         "remove event from queue"
     );
 }
 pub fn queueproc_action_run(rid: &RepoId, oid: &Oid, msg: &str) {
 pub fn queueproc_action_run(event: &CiEvent) {
     info!(
         msg_id = ?Id::QueueProcActionRun,
         kind = %Kind::Debug,
         ?rid,
         ?oid,
         msg = msg,
         ?event,
         "Action: run"
     );
 }

 }
 pub fn broker_db(filename: &Path) {
     info!(
     debug!(
         msg_id = ?Id::BrokerDatabase,
         kind = %Kind::Startup,
         filename = %filename.display(),

     );
 }
 pub fn adapter_result(exit: i32) {
 pub fn adapter_result(exit: ExitStatus) {
     debug!(
         msg_id = ?Id::AdapterExitCode,
         kind = %Kind::Debug,
         exit_code = exit,
         ?exit,
         "adapter exit code"
     );
 }

     );
 }
 pub fn adapter_did_not_exit() {
 pub fn adapter_did_not_exit(error: TimeoutError) {
     warn!(
         msg_id = ?Id::AdapterNoExit,
         kind = %Kind::Debug,
         ?error,
         "adapter did not exit: probably killed by signal"
     );
 }

 const UPDATE_IMTERVAL: Duration = Duration::from_secs(1);
 /// Channel endpoint for sending notifications.
 #[derive(Clone)]
 pub struct NotificationSender {
     sender: Sender<()>,
 }

 use std::sync::{Arc, Condvar, Mutex};
 /// A pull queue.
 ///
 /// A producer produces an item when a consumer is ready for it, not
 /// before. There can be any number of consumers.
 ///
 /// This type has interior mutability. Each thread should own
 /// their clone of this, and all clones refer to the same queue.
 ///
 /// # Example
 /// ```
 /// # use std::thread::spawn;
 /// # use radicle_ci_broker::pull_queue::PullQueue;
 /// let mut queue = PullQueue::new();
 /// let mut clone = queue.clone();
 /// let producer = spawn(move || {
 ///     for i in 0..10 {
 ///         queue.push(i);
 ///     }
 /// });
 /// let consumer = spawn(move ||
 ///     while let Ok(Some(i)) = clone.pop() {
 ///         println!("{i}");
 ///     }
 /// );
 /// producer.join().unwrap();
 /// consumer.join().unwrap();
 /// ```
 pub struct PullQueue<T> {
     q: Arc<(Mutex<Unlocked<T>>, Condvar)>,
 }
 impl<T> Clone for PullQueue<T> {
     fn clone(&self) -> Self {
         Self { q: self.q.clone() }
     }
 }
 impl<T: Clone> Default for PullQueue<T> {
     fn default() -> Self {
         Self::new()
     }
 }
 impl<T> Drop for PullQueue<T> {
     fn drop(&mut self) {
         let (mutex, var) = &*self.q;
         let mut locked = mutex
             .lock()
             .map_err(|_| PullQueueError::Mutex)
             .expect("FATAL: mutex was poisoned");
         locked.end();
         var.notify_all();
     }
 }
 impl<T: Clone> PullQueue<T> {
     /// Create a new [`PullQueue`].
     pub fn new() -> Self {
         Self {
             q: Arc::new((Mutex::new(Unlocked::new()), Condvar::new())),
         }
     }
     /// Push a new item to the queue. This blocks until there is a
     /// consumer who wants the item.
     pub fn push(&mut self, i: T) -> Result<(), PullQueueError> {
         let (mutex, var) = &*self.q;
         let mut locked = mutex.lock().map_err(|_| PullQueueError::Mutex)?;
         loop {
             if locked.try_set(i.clone()) {
                 var.notify_all();
                 break;
             } else {
                 // Wait for the previous item to be consumed.
                 locked = var.wait(locked).map_err(|_| PullQueueError::Mutex)?;
             }
         }
         Ok(())
     }
     /// Consume an item, if one is available. This will block until an
     /// item is available or not more items will ever be produced.
     /// Return `None` when no more items will ever be produced.
     pub fn pop(&mut self) -> Result<Option<T>, PullQueueError> {
         let (mutex, var) = &*self.q;
         let mut locked = mutex.lock().map_err(|_| PullQueueError::Mutex)?;
         loop {
             if locked.is_ended() {
                 // Queue is empty and won't get anything more. We're
                 // done.
                 var.notify_all();
                 return Ok(None);
             } else if let Some(i) = locked.get() {
                 // There was something in the queue, return it.
                 var.notify_all();
                 return Ok(Some(i));
             } else {
                 locked = var.wait(locked).map_err(|_| PullQueueError::Mutex)?;
             }
         }
     }
 }
 struct Unlocked<T> {
     ended: bool,
     item: Option<T>,
 }
 impl<T> Unlocked<T> {
     fn new() -> Self {
         Self {
             ended: false,
             item: None,
         }
     }
     fn end(&mut self) {
         self.ended = true;
     }
     fn is_ended(&self) -> bool {
         self.ended && self.item.is_none()
     }
     fn try_set(&mut self, item: T) -> bool {
         if self.item.is_none() {
             self.item = Some(item);
             true
         } else {
             false
         }
     }
     fn get(&mut self) -> Option<T> {
         self.item.take()
     }
 }
 /// All possible errors from [`PullQueue`].
 #[derive(Debug, thiserror::Error)]
 pub enum PullQueueError {
     /// Mutex locking failed.
     #[error("failed to lock mutex for pull queue")]
     Mutex,
 }

 #![allow(clippy::result_large_err)]
 use std::{
     sync::mpsc::RecvTimeoutError,
     collections::HashSet,
     sync::{
         mpsc::{sync_channel, Receiver, RecvTimeoutError, SyncSender},
         Arc, Mutex,
     },
     thread::{spawn, JoinHandle},
     time::{Duration, Instant},
 };
 use radicle::Profile;
 use radicle::{identity::RepoId, Profile};
 use crate::{
     adapter::{Adapter, Adapters},
     broker::{Broker, BrokerError},
     ci_event::{CiEvent, CiEventV1},
     db::{Db, DbError, QueueId, QueuedCiEvent},
     db::{Db, DbError, QueuedCiEvent},
     filter::{EventFilter, Trigger},
     logger,
     msg::{MessageError, RequestBuilder},
     notif::{NotificationReceiver, NotificationSender},
     pull_queue::{PullQueue, PullQueueError},
 };
 #[derive(Default)]

     events_rx: Option<NotificationReceiver>,
     run_tx: Option<NotificationSender>,
     queue_len_interval: Option<Duration>,
     concurrent_adapters: Option<usize>,
 }
 const DEFAULT_QUEUE_LEN_DURATION: Duration = Duration::from_secs(10);
 impl QueueProcessorBuilder {
     pub fn build(self) -> Result<QueueProcessor, QueueError> {
         let profile = Profile::load().map_err(QueueError::Profile)?;
         let broker = self.broker.ok_or(QueueError::Missing("broker"))?;
         let filters = self.filters.ok_or(QueueError::Missing("filters"))?;
         let triggers = self.triggers.ok_or(QueueError::Missing("triggers"))?;
         let adapters = self.adapters.ok_or(QueueError::Missing("adapters"))?;
         let run_tx = self.run_tx.ok_or(QueueError::Missing("run_tx"))?;
         let concurrent_adapters = self
             .concurrent_adapters
             .ok_or(QueueError::Missing("concurrent_adapters"))?;
         let (procssed_tx, processed_rx) = sync_channel(1);
         let picked_events = PullQueue::new();
         let mut descs = vec![];
         for _ in 0..concurrent_adapters {
             let broker = Broker::new(broker.db().filename(), broker.max_run_time())
                 .map_err(QueueError::NewBroker)?;
             let event_procssor = EventProcessor::new(
                 profile.clone(),
                 filters.clone(),
                 triggers.clone(),
                 adapters.clone(),
                 broker,
                 run_tx.clone(),
                 procssed_tx.clone(),
             );
             descs.push(ProcessorDescription::new(
                 event_procssor,
                 picked_events.clone(),
             ));
         }
         Ok(QueueProcessor {
             db: self.db.ok_or(QueueError::Missing("db"))?,
             profile: Profile::load().map_err(QueueError::Profile)?,
             broker: self.broker.ok_or(QueueError::Missing("broker"))?,
             filters: self.filters.ok_or(QueueError::Missing("filters"))?,
             triggers: self.triggers.ok_or(QueueError::Missing("triggers"))?,
             adapters: self.adapters.ok_or(QueueError::Missing("adapters"))?,
             events_rx: self.events_rx.ok_or(QueueError::Missing("events_rx"))?,
             run_tx: self.run_tx.ok_or(QueueError::Missing("run_tx"))?,
             queue_len_interval: self
                 .queue_len_interval
                 .unwrap_or(DEFAULT_QUEUE_LEN_DURATION),
             prev_queue_len: Instant::now(),
             processors: descs,
             processed_rx: Some(processed_rx),
             picked_events: Some(picked_events),
             current: CurrentlyPicked::default(),
         })
     }

         self
     }
     pub fn concurrent_adapters(mut self, n: usize) -> Self {
         self.concurrent_adapters = Some(n);
         self
     }
     pub fn broker(mut self, broker: Broker) -> Self {
         self.broker = Some(broker);
         self

 pub struct QueueProcessor {
     db: Db,
     profile: Profile,
     filters: Vec<EventFilter>,
     triggers: Vec<Trigger>,
     broker: Broker,
     adapters: Adapters,
     events_rx: NotificationReceiver,
     run_tx: NotificationSender,
     queue_len_interval: Duration,
     prev_queue_len: Instant,
     picked_events: Option<PullQueue<Picked>>,
     processors: Vec<ProcessorDescription>,
     processed_rx: Option<Receiver<RepoId>>,
     current: CurrentlyPicked,
 }
 impl QueueProcessor {
     pub fn process_in_thread(mut self) -> JoinHandle<Result<(), QueueError>> {
         spawn(move || {
             logger::queueproc_start();
             logger::queueproc_start(self.processors.len());
             // Spawn a thread to process results from running adapters.
             #[allow(clippy::unwrap_used)]
             let rx = self.processed_rx.take().unwrap();
             let mut current = self.current.clone();
             let results = spawn(move || {
                 logger::queueproc_start_result_receiver();
                 while let Ok(repoid) = rx.recv() {
                     current.remove(repoid);
                     logger::queueproc_end_result_receiver();
                 }
             });
             // Pick events from queue, send to worker threads that run
             // adapters. Results are processed by thread above.
             let result = self.process_until_shutdown();
             logger::queueproc_end(&result);
             result
             // Wait for worker threads to terminate. This closes all
             // sender ends for results channel.
             for proc in self.processors {
                 proc.join().ok();
             }
             // Wait for results processing thread to terminate.
             results.join().ok();
             Ok(())
         })
     }
     #[allow(clippy::unwrap_used)]
     fn process_until_shutdown(&mut self) -> Result<(), QueueError> {
         let mut done = false;
         let mut picked_events = self.picked_events.take().unwrap();
         while !done {
             let mut first_error = None;
             while let Some(qe) = self.pick_event()? {
                 // We always drop the picked event at once so that if
                 // anything goes wrong, we don't try to process the
                 // same event again. It's better for the failure to be
                 // logged and let the humans deal with whatever
                 // complicated failure situation is happening.
                 self.drop_event(qe.id())?;
                 match self.pick_adapters(qe.event()) {
                     Err(err) => {
                         if first_error.is_none() {
                             first_error = Some(err);
                         }
                     }
                     Ok(None) => (), // We already removed event from queue.
                     Ok(Some(adapters)) => {
                         for adapter in adapters {
                             match self.run_adapter(&qe, &adapter) {
                                 Err(err) => {
                                     if first_error.is_none() {
                                         first_error = Some(err)
                                     }
                                 }
                                 Ok(finished) => done = finished,
                             }
                         }
                     }
             // Process all events currently in the event queue, if a
             // filter allows it.
             loop {
                 if let Some(qe) = self.pick_event()? {
                     let picked = Picked::new(qe.clone());
                     self.current.insert(picked.qe.event().repository());
                     picked_events.push(picked).map_err(QueueError::PullQueue)?;
                     // We always drop the picked event as soon as it has
                     // been pushed to a processing thread, so that if
                     // anything goes wrong, we're less likely try to
                     // process the same event again. It's better for the
                     // failure to be logged and let the humans deal with
                     // whatever complicated failure situation is
                     // happening.
                     self.drop_event(&qe)?;
                 } else if self.current.is_empty() {
                     break;
                 }
             }
             if let Some(err) = first_error {
                 return Err(err);
             }
             // Wait for a notification of new events in the queue.
             // This prevents the loop from being a busy loop.
             match self.events_rx.wait_for_notification() {
                 Ok(_) => {}
                 Err(RecvTimeoutError::Timeout) => {}

         Ok(())
     }
     fn run_adapter(&mut self, qe: &QueuedCiEvent, adapter: &Adapter) -> Result<bool, QueueError> {
         let mut done = false;
         self.run_tx.notify()?;
         logger::queueproc_picked_event(qe.id(), qe, adapter);
         let res = self.process_event(qe.event(), adapter);
         logger::queueproc_processed_event(&res);
         match res {
             Ok(shut_down) => done = shut_down,
             Err(QueueError::BuildTrigger(_, _)) => done = false,
             Err(err) => Err(err)?,
         }
         self.run_tx.notify()?;
         Ok(done)
     }
     fn pick_event(&mut self) -> Result<Option<QueuedCiEvent>, QueueError> {
         let ids = self.db.queued_ci_events().map_err(QueueError::db)?;

         }
         queue.sort_by_cached_key(|qe| qe.timestamp().to_string());
         // If we can access the set of repositories for which CI is
         // currently running, remove those repositories from the list.
         let ids = self.current.list();
         queue = queue
             .iter()
             .filter(|qe| {
                 if let Some(repoid) = qe.event().repository() {
                     !ids.contains(repoid)
                 } else {
                     true
                 }
             })
             .cloned()
             .collect();
         if let Some(qe) = queue.first() {
             Ok(Some(qe.clone()))
         } else {