refactor(src/queueproc.rs): simplify thread structure — radicle-ci-broker

Change the queue processing to spawn a new thread for each new event to process (run adapters for).

Previously we had N long-lived threads, where N is set in the configuration setting concurrent_adapters. This is awkward if we want to reload configuration, which is an open feature request.

The new structure is also a lot simpler, which hopefully means it’ll be more reliable.

 [package]
 name = "radicle-ci-broker"
 version = "0.21.0"
 edition = "2021"
 edition = "2024"
 rust-version = "1.85"
 authors = ["Lars Wirzenius <liw@liw.fi", "cloudhead <cloudhead@radicle.xyz>"]
 description = "add integration to CI engins or systems to a Radicle node"

 };
 use serde::Serialize;
 use tempfile::{tempdir, TempDir};
 use tempfile::{TempDir, tempdir};
 use url::Url;
 use crate::{

 mod test {
     use std::{fs::write, io::ErrorKind, path::PathBuf, time::Duration};
     use tempfile::{tempdir, NamedTempFile, TempDir};
     use tempfile::{NamedTempFile, TempDir, tempdir};
     use radicle::git::Oid;
     use radicle::prelude::RepoId;

         msg::{MessageError, Response, RunId, RunResult},
         notif::NotificationChannel,
         run::{RunBuilder, Whence},
         test::{mock_adapter, trigger_request, TestResult},
         test::{TestResult, mock_adapter, trigger_request},
     };
     const MAX: Duration = Duration::from_secs(10);

 use clap::Parser;
 use radicle::{git::Oid, prelude::NodeId, Profile};
 use radicle::{Profile, git::Oid, prelude::NodeId};
 use radicle_ci_broker::{
     broker::BrokerError,

     #[error("when adding a branch-update event, the --base option is required")]
     BaseRequired,
     #[error(
         "when adding a patch-create or patch--update event, the --patch-id option is required"
     )]
     #[error("when adding a patch-create or patch--update event, the --patch-id option is required")]
     PatchIdRequired,
     #[error("failed to construct a CI event")]

 use std::{
     convert::TryFrom,
     fs::{read, File},
     fs::{File, read},
     io::{Read, Write},
     path::{Path, PathBuf},
 };

 use std::{
     env::current_exe,
     fs::{read, remove_file, File, OpenOptions},
     fs::{File, OpenOptions, read, remove_file},
     io::{Read, Write},
     net::Shutdown,
     os::unix::net::{UnixListener, UnixStream},

     time::Duration,
 };
 use time::{macros::format_description, OffsetDateTime};
 use tracing::{span, Level};
 use time::{OffsetDateTime, macros::format_description};
 use tracing::{Level, span};
 use radicle::prelude::RepoId;

     storage::RefUpdate,
 };
 use crate::refs::{ref_string, GenericRefName, TagName};
 use crate::refs::{GenericRefName, TagName, ref_string};
 #[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 #[non_exhaustive]

     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 {

 use radicle::{
     git::Oid,
     node::{
         sync::{Announcer, AnnouncerConfig, ReplicationFactor},
         Handle, Node,
         sync::{Announcer, AnnouncerConfig, ReplicationFactor},
     },
     prelude::{Profile, ReadStorage, RepoId},
     storage::git::Repository,

 use serde::{Deserialize, Serialize};
 use sqlite::{Connection, State, Statement};
 use time::{macros::format_description, OffsetDateTime};
 use time::{OffsetDateTime, macros::format_description};
 use uuid::Uuid;
 use crate::{ci_event::CiEvent, msg::RunId, run::Run};

 use std::str::FromStr;
 use radicle::{
     cob::patch::{cache::Patches, Patch, PatchId},
     cob::patch::{Patch, PatchId, cache::Patches},
     git::Oid,
     identity::{Project, RepoId},
     profile::Profile,
     storage::{git::Repository, ReadStorage, RepositoryInfo},
     storage::{ReadStorage, RepositoryInfo, git::Repository},
 };
 /// A Radicle node.

 use radicle::{
     cob::patch::PatchId,
     git::{raw::ObjectType, BranchName, Oid},
     git::{BranchName, Oid, raw::ObjectType},
     node::NodeId,
     prelude::{Profile, RepoId},
     storage::git::Repository,

     use qcheck_macros::quickcheck;
     use radicle::prelude::{Did, RepoId};
     use crate::refs::{branch_from_str, TagName};
     use crate::refs::{TagName, branch_from_str};
     use super::*;

 use clap::ValueEnum;
 use serde_json::Value;
 use tracing::{debug, error, info, trace, warn, Level};
 use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 use tracing::{Level, debug, error, info, trace, warn};
 use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
 use uuid::Uuid;
 use radicle::{git::Oid, identity::RepoId, node::Event, patch::PatchId};

         process::Command,
     };
     use nonempty::{nonempty, NonEmpty};
     use nonempty::{NonEmpty, nonempty};
     use radicle::prelude::{Profile, RepoId};
     use time::{macros::format_description, OffsetDateTime};
     use time::{OffsetDateTime, macros::format_description};
     use super::{MessageError, Oid, Request, Response, RunId, RunResult};

 use std::{fmt, path::PathBuf, time};
 use radicle::{
     node::{Event, Handle},
     Profile,
     node::{Event, Handle},
 };
 use crate::logger;

 //! Notification channel between threads.
 use std::{
     sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender},
     sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel},
     time::Duration,
 };

 use std::{
     collections::HashSet,
     sync::{
         mpsc::{sync_channel, Receiver, RecvTimeoutError, SyncSender},
         Arc, Mutex,
     },
     thread::JoinHandle,
     sync::{Arc, Mutex},
     thread::spawn,
     time::{Duration, Instant},
 };
 use radicle::{identity::RepoId, Profile};
 use radicle::{Profile, identity::RepoId};
 use crate::{
     adapter::{Adapter, Adapters},

     logger,
     msg::{MessageError, RequestBuilder},
     notif::{NotificationReceiver, NotificationSender},
     pull_queue::{PullQueue, PullQueueError},
     worker::{start_thread, Worker},
     worker::Worker,
 };
 #[derive(Default)]

         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 processors = 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(),
                 picked_events.clone(),
                 procssed_tx.clone(),
             );
             processors.push(start_thread(event_procssor));
         }
         Ok(QueueProcessor {
             profile,
             broker,
             filters,
             triggers,
             adapters,
             db: self.db.ok_or(QueueError::Missing("db"))?,
             events_rx: self.events_rx.ok_or(QueueError::Missing("events_rx"))?,
             queue_len_interval: self
                 .queue_len_interval
                 .unwrap_or(DEFAULT_QUEUE_LEN_DURATION),
             prev_queue_len: Instant::now(),
             processors,
             processed_rx: Some(processed_rx),
             picked_events: Some(picked_events),
             concurrent_adapters,
             run_tx,
             current: CurrentlyPicked::default(),
         })
     }

     }
 }
 impl Worker for QueueProcessor {
     const NAME: &str = "queue-processor";
     type Error = QueueError;
     fn work(&mut self) -> Result<(), QueueError> {
         logger::queueproc_start(self.processors.len());
         // Spawn a thread to process results from running adapters.
         #[allow(clippy::unwrap_used)]
         let adapter_results = start_thread(AdapterResults::new(
             self.processed_rx.take().unwrap(),
             self.current.clone(),
         ));
         // 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);
         // Wait for worker threads to terminate. This closes all
         // sender ends for results channel.
         while let Some(proc) = self.processors.pop() {
             let result = proc
                 .join()
                 .map_err(|_| QueueError::JoinEventProcessorThread)?;
             logger::queueproc_processor_thread_result(result.as_ref().map(|_| ()));
         }
         // Wait for results processing thread to terminate.
         adapter_results.join().ok();
         Ok(())
     }
 }
 // The queue processor gets events from the event queue in
 // the database, and processes them concurrently by running
 // the appropriate adapters. To avoid busy looping, the
 // `events_rx` channel is used to receive notification that
 // an event has been added to the database. Processing will
 // end when the channel is closed and the queue is empty,
 // or a "shutdown" event is encountered. In case of shutdown,
 // any currently running adapters will be allowed to finish.
 pub struct QueueProcessor {
     profile: Profile,
     db: Db,
     broker: Broker,
     filters: Vec<EventFilter>,
     triggers: Vec<Trigger>,
     adapters: Adapters,
     concurrent_adapters: usize,
     events_rx: NotificationReceiver,
     queue_len_interval: Duration,
     prev_queue_len: Instant,
     picked_events: Option<PullQueue<Picked>>,
     processors: Vec<JoinHandle<Result<(), QueueError>>>,
     processed_rx: Option<Receiver<RepoId>>,
     run_tx: NotificationSender,
     current: CurrentlyPicked,
 }
 impl QueueProcessor {
     #[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 {
             // 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;
         let mut expecting_new_events = true;
         let mut handles = vec![];
         loop {
             let mut queue_was_emptied = false;
             // If we may spawn another adapter, pick an event from the queue
             // and run the adapters.
             if expecting_new_events && handles.len() < self.concurrent_adapters {
                 match self.pick_event() {
                     Ok(Some(qe)) => {
                         // Remove picked event from queue so we doh't re-process it. If we don't
                         // do this, and we crash when processing the event, we'll re-process it
                         // again and again. It seems better to discard an event, and skip running
                         // CI, rather then getting stuck on a specific event.
                         self.drop_event(&qe)?;
                         match self.matching_adapters(qe.event()) {
                             Ok(Some(adapters)) => {
                                 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));
                                 handles.push((repoid, h));
                             }
                             Ok(None) => (),
                             Err(_) => (),
                         }
                     }
                     Ok(None) => queue_was_emptied = true,
                     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) => {}
                 Err(RecvTimeoutError::Disconnected) => {
                     logger::queueproc_channel_disconnect();
                     done = true;
             // Wait for any threads processing events that have finished. Remove
             // them from the list of currently running threads.
             let mut h2 = vec![];
             for (repoid, h) in handles {
                 if h.is_finished() {
                     if let Some(repoid) = repoid {
                         self.current.remove(repoid);
                     }
                     match h.join() {
                         Err(err) => eprintln!("thread join error {err:?}"),
                         Ok(Err(_)) => (),
                         Ok(Ok(MaybeShutdown::Shutdown)) => {
                             expecting_new_events = false;
                             queue_was_emptied = true;
                         }
                         Ok(Ok(MaybeShutdown::Continue)) => (),
                     }
                 } else {
                     h2.push((repoid, h));
                 }
             }
             handles = h2;
             // If we didn't empty the event queue, but we're still
             // expecting new events, wait for a new event. This prevents
             // a busy loop.
             if expecting_new_events && queue_was_emptied {
                 match self.events_rx.wait_for_notification() {
                     Err(_) => {
                         expecting_new_events = false;
                     }
                     Ok(_) => queue_was_emptied = false,
                 }
             }
             if handles.is_empty() && !expecting_new_events && queue_was_emptied {
                 break;
             }
         }
         Ok(())
     }
     fn processor(&self) -> Result<Processor, QueueError> {
         Ok(Processor {
             profile: self.profile.clone(),
             broker: Broker::new(self.db.filename(), self.broker.max_run_time())
                 .map_err(QueueError::NewBroker)?,
             run_tx: self.run_tx.clone(),
         })
     }
     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.
         // Remove the repositories for which CI is currently running.
         let ids = self.current.list();
         queue = queue
             .iter()

             .collect();
         if let Some(qe) = queue.first() {
             eprintln!("picked event: {qe:?}");
             Ok(Some(qe.clone()))
         } else {
             Ok(None)

             .map_err(QueueError::db)?;
         Ok(())
     }
 }
 #[derive(Default, Clone)]
 struct CurrentlyPicked {
     set: Arc<Mutex<HashSet<RepoId>>>,
 }
 impl CurrentlyPicked {
     fn insert(&mut self, repoid: Option<&RepoId>) {
         if let Some(repoid) = repoid {
             if let Ok(mut set) = self.set.lock() {
                 set.insert(*repoid);
             }
         }
     }
     fn remove(&mut self, repoid: RepoId) {
         if let Ok(mut set) = self.set.lock() {
             set.remove(&repoid);
         }
     }
     fn list(&self) -> Vec<RepoId> {
         if let Ok(set) = self.set.lock() {
             set.iter().copied().collect()
         } else {
             vec![]
         }
     }
     fn is_empty(&self) -> bool {
         if let Ok(set) = self.set.lock() {
             set.is_empty()
         } else {
             false
         }
     }
 }
 struct EventProcessor {
     profile: Profile,
     filters: Vec<EventFilter>,
     triggers: Vec<Trigger>,
     adapters: Adapters,
     broker: Broker,
     run_tx: NotificationSender,
     picked_events: PullQueue<Picked>,
     processed_tx: SyncSender<RepoId>,
 }
 impl EventProcessor {
     #[allow(clippy::too_many_arguments)]
     fn new(
         profile: Profile,
         filters: Vec<EventFilter>,
         triggers: Vec<Trigger>,
         adapters: Adapters,
         broker: Broker,
         run_tx: NotificationSender,
         picked_events: PullQueue<Picked>,
         processed_tx: SyncSender<RepoId>,
     ) -> Self {
         Self {
             profile,
             filters,
             triggers,
             adapters,
             broker,
             run_tx,
             picked_events,
             processed_tx,
         }
     }
     fn process_picked_event(&self, picked: Picked) {
         let res = self.matching_adapters(picked.qe.event());
         logger::queueproc_worker_thread_result(res.as_ref().map(|_| ()));
         if let Ok(Some(adapters)) = res {
             for adapter in adapters {
                 self.run_adapter(&picked.qe, &adapter).ok();
             }
         }
         if let Some(repoid) = picked.qe.event().repository() {
             self.processed_tx.send(*repoid).ok();
         }
     }
     fn matching_adapters(&self, e: &CiEvent) -> Result<Option<Vec<Adapter>>, QueueError> {
         let mut adapters = vec![];

             Ok(Some(adapters))
         }
     }
     fn run_adapter(&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 == MaybeShutdown::Shutdown,
             Err(QueueError::BuildTrigger(_, _)) => done = false,
             Err(err) => Err(err)?,
         }
         self.run_tx.notify()?;
         Ok(done)
     }
     fn process_event(
         &self,
         event: &CiEvent,
         adapter: &Adapter,
     ) -> Result<MaybeShutdown, QueueError> {
         if matches!(event, CiEvent::V1(CiEventV1::Shutdown)) {
             logger::queueproc_action_shutdown();
             Ok(MaybeShutdown::Shutdown)
         } else {
             logger::queueproc_action_run(event);
             let trigger = RequestBuilder::default()
                 .profile(&self.profile)
                 .ci_event(event)
                 .build_trigger_from_ci_event()
                 .map_err(|e| QueueError::build_trigger(event, e));
             logger::queueproc_trigger(&trigger);
             let trigger = trigger?;
             self.broker
                 .execute_ci(adapter, &trigger, &self.run_tx)
                 .map_err(QueueError::execute_ci)?;
             Ok(MaybeShutdown::Continue)
         }
     }
 }
 impl Worker for EventProcessor {
     const NAME: &str = "event-processor";
 impl Worker for QueueProcessor {
     const NAME: &str = "queue-processor";
     type Error = QueueError;
     fn work(&mut self) -> Result<(), QueueError> {
         while let Ok(Some(picked)) = self.picked_events.pop() {
             self.process_picked_event(picked);
         }
         // 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);
         Ok(())
     }
 }
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 pub enum MaybeShutdown {
     Shutdown,
     Continue,
 struct Processor {
     profile: Profile,
     broker: Broker,
     run_tx: NotificationSender,
 }
 struct AdapterResults {
     rx: Receiver<RepoId>,
     current: CurrentlyPicked,
 }
 impl Processor {
     fn pick_and_process_one(
         &self,
         qe: QueuedCiEvent,
         adapters: Vec<Adapter>,
     ) -> Result<MaybeShutdown, QueueError> {
         for adapter in adapters.iter() {
             self.run_tx.notify()?;
             logger::queueproc_picked_event(qe.id(), &qe, adapter);
             match qe.event() {
                 CiEvent::V1(CiEventV1::Shutdown) => {
                     logger::queueproc_action_shutdown();
                     return Ok(MaybeShutdown::Shutdown);
                 }
                 _ => {
                     logger::queueproc_action_run(qe.event());
                     let trigger = RequestBuilder::default()
                         .profile(&self.profile)
                         .ci_event(qe.event())
                         .build_trigger_from_ci_event()
                         .map_err(|e| QueueError::build_trigger(qe.event(), e));
                     logger::queueproc_trigger(&trigger);
                     let trigger = trigger?;
                     self.broker
                         .execute_ci(adapter, &trigger, &self.run_tx)
                         .map_err(QueueError::execute_ci)?;
                 }
             }
         }
 impl AdapterResults {
     fn new(rx: Receiver<RepoId>, current: CurrentlyPicked) -> Self {
         Self { rx, current }
         Ok(MaybeShutdown::Continue)
     }
 }
 impl Worker for AdapterResults {
     const NAME: &str = "adapter-result-processor";
     type Error = QueueError;
     fn work(&mut self) -> Result<(), QueueError> {
         logger::queueproc_start_result_receiver();
         while let Ok(repoid) = self.rx.recv() {
             self.current.remove(repoid);
             logger::queueproc_end_result_receiver();
 #[derive(Default, Clone)]
 struct CurrentlyPicked {
     set: Arc<Mutex<HashSet<RepoId>>>,
 }
 impl CurrentlyPicked {
     fn insert(&mut self, repoid: Option<&RepoId>) {
         if let Some(repoid) = repoid {
             if let Ok(mut set) = self.set.lock() {
                 set.insert(*repoid);
             }
         }
         Ok(())
     }
 }
 #[derive(Debug, Clone)]
 struct Picked {
     qe: QueuedCiEvent,
 }
     fn remove(&mut self, repoid: RepoId) {
         if let Ok(mut set) = self.set.lock() {
             set.remove(&repoid);
         }
     }
 impl Picked {
     fn new(qe: QueuedCiEvent) -> Self {
         Self { qe }
     fn list(&self) -> Vec<RepoId> {
         if let Ok(set) = self.set.lock() {
             set.iter().copied().collect()
         } else {
             vec![]
         }
     }
 }
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 pub enum MaybeShutdown {
     Shutdown,
     Continue,
 }
 #[derive(Debug, thiserror::Error)]
 pub enum QueueError {
     #[error("failed to load node profile")]
     Profile(#[source] radicle::profile::Error),
     #[error("failed to open database")]
     OpenDb(#[source] crate::db::DbError),
     #[error("programming error: QueueProcessorBuilder field {0} was not set")]
     Missing(&'static str),

     #[error("failed to create a new broker instance")]
     NewBroker(#[source] BrokerError),
     #[error("failure when using a pull queue")]
     PullQueue(#[source] PullQueueError),
 }
 impl QueueError {

 use std::fmt;
 use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
 use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
 const PLACEHOLDER: &str = "<REDACTED>";

 use crate::adapter::Adapter;
 use crate::ci_event::{CiEvent, CiEventV1};
 use crate::msg::{Request, RequestBuilder};
 use radicle::Profile;
 use radicle::crypto::ssh::Keystore;
 use radicle::crypto::test::signer::MockSigner;
 use radicle::git::RefString;
 use radicle::profile::{Config, Home};
 use radicle::storage::ReadRepository;
 use radicle::test::setup::Node;
 use radicle::Profile;
 use tempfile::tempdir;

     io::{Read, Seek, Write},
     process::{Child, Command, ExitStatus, Stdio},
     sync::{
         Arc, Condvar, Mutex, MutexGuard,
         mpsc::{
             channel, sync_channel, Receiver, RecvTimeoutError, Sender, SyncSender, TryRecvError,
             Receiver, RecvTimeoutError, Sender, SyncSender, TryRecvError, channel, sync_channel,
         },
         Arc, Condvar, Mutex, MutexGuard,
     },
     thread::{sleep, spawn, JoinHandle},
     thread::{JoinHandle, sleep, spawn},
     time::{Duration, Instant},
 };

     }
     #[test]
     fn sleep_for_too_long_while_reading_with_realtimelines(
     ) -> Result<(), Box<dyn std::error::Error>> {
     fn sleep_for_too_long_while_reading_with_realtimelines()
     -> Result<(), Box<dyn std::error::Error>> {
         let (running, mut stdout) = setup("exec sleep 1000", SHORT_TIMEOUT, None)?;
         while stdout.line().is_some() {}
         let result = running.wait();

 use tempfile::NamedTempFile;
 use time::{
     format_description::{well_known::Rfc2822, FormatItem},
     OffsetDateTime, PrimitiveDateTime,
     format_description::{FormatItem, well_known::Rfc2822},
     macros::format_description,
     parsing::Parsable,
     OffsetDateTime, PrimitiveDateTime,
 };
 use radicle::{
     Profile, Storage,
     cob::ObjectId,
     git::Oid,
     prelude::{NodeId, RepoId},
     storage::ReadStorage,
     Profile, Storage,
 };
 pub fn lookup_repo(profile: &Profile, wanted: &str) -> Result<(RepoId, String), UtilError> {

 //! are implemented and used, as well as logging. A worker thread does
 //! what it does. The abstraction is not meant to constrain that.
 use std::thread::{spawn, JoinHandle};
 use std::thread::{JoinHandle, spawn};
 /// Start a new thread. Caller must catch the thread handle and
 /// join it to wait for thread to end.