fix: when adapter times out, return correct error variant — radicle-ci-broker

We need to handle “run failed for an intrinsic reason” differently from “run didn’t finish in time”.

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

fix: expected result in test case

wrong result used to be returned by code, test case was always wrong

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

refactor(src/adapter.rs): reduce code duplication in tests

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

feat: say run timed out on report page if that happened

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

         run.set_state(RunState::Finished);
         db.update_run(run).map_err(AdapterError::UpdateRun)?;
         if matches!(x, Err(AdapterError::FailedTimeout)) {
             run.set_timed_out();
         }
         x
     }

             outcome.set_error(err);
         }
         let result = if outcome.has_error() {
             child.kill()
         } else {
             child.wait()
         };
         let result = child.wait();
         match result {
             Ok(finished) => {

                 }
             }
             Err(TimeoutError::TimedOut) => {
                 logger::adapter_did_not_exit_voluntarily();
                 outcome.set_error(AdapterError::FailedNotExited);
                 logger::adapter_timed_out();
                 outcome.set_error(AdapterError::FailedTimeout);
             }
             Err(err) => {
                 logger::adapter_did_not_exit(err);
                 outcome.set_error(AdapterError::Signal(9));
                 logger::adapter_did_not_exit(&err);
                 outcome.set_error(AdapterError::TimeoutWait(err));
             }
         }
         if let Some(err) = outcome.error() {
             run.set_result(RunResult::Failure);
             Err(err)
         } else {
             Ok(())

 impl MaybeResult {
     fn set_error(&mut self, err: AdapterError) {
         if self.error.is_none() {
             self.error = Some(err);
         }
     }
     fn has_error(&self) -> bool {
         self.error.is_some()
         self.error = Some(err);
     }
     fn error(self) -> Option<AdapterError> {

     #[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 terminated due to taking too long.
     #[error("child process was terminated due to taking too long")]
     FailedTimeout,
     /// Child process was killed.
     #[error("child process terminated by signal {0}")]
     Signal(i32),
     /// Child process did not terminate.
     #[error("child process did not terminate")]
     TimeoutWait(#[source] crate::timeoutcmd::TimeoutError),
     /// First message is not `Response::Triggered`
     #[error("adapter's first message is not 'triggered', but {0:?}")]
     NotTriggered(Response),

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

             .build())
     }
     #[allow(clippy::unwrap_used)]
     fn adapter(tmp: &TempDir, shell: &'static str) -> PathBuf {
         let bin = tmp.path().join("adapter.sh");
         mock_adapter(&bin, shell).unwrap();
         bin
     }
     #[test]
     fn adapter_reports_success() -> TestResult<()> {
         const ADAPTER: &str = r#"#!/bin/sh
         let tmp = tempdir()?;
         let bin = adapter(
             &tmp,
             r#"#!/bin/sh
 read
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"success"}'
 "#;
         let tmp = tempdir()?;
         let bin = tmp.path().join("adapter.sh");
         mock_adapter(&bin, ADAPTER)?;
 "#,
         );
         let db = db()?;
         let mut run = run()?;

         let run = x?;
         assert_eq!(run.adapter_run_id(), Some(&RunId::from("xyzzy")));
         assert_eq!(run.state(), RunState::Finished);
         assert_eq!(run.result(), Some(&RunResult::Success));
         assert_eq!(run.result(), Some(&RunResult::Failure));
         Ok(())
     }

     AdapterStderrLine,
     AdapterStdoutLine,
     AdapterTempConfig,
     AdapterTimedOut,
     AdapterTooManyMessages,
     BrokerDatabase,

     );
 }
 pub fn adapter_did_not_exit(error: TimeoutError) {
 pub fn adapter_did_not_exit(error: &TimeoutError) {
     warn!(
         msg_id = ?Id::AdapterNoExit,
         kind = %Kind::Debug,

     );
 }
 pub fn adapter_timed_out() {
     info!(
         msg_id = ?Id::AdapterTimedOut,
         kind = %Kind::FinishRun,
         "adapter was terminated due to taking too long"
     );
 }
 pub fn job_failure(
     msg: &'static str,
     repo_id: &RepoId,

         let mut runs = self.runs(rid);
         runs.sort_by_cached_key(|run| run.timestamp());
         runs.reverse();
         for run in runs {
             let result = if let Some(result) = run.result() {
                 result.to_string()
             } else {
                 "unknown".into()
             };
             let mut status = Element::new(Tag::Span)
                 .with_class(&result)
                 .with_text(&result);
             if run.timed_out() {
                 status.push_text(" (timed out)");
             }
             let current = match run.state() {
                 RunState::Triggered => Element::new(Tag::Span)
                     .with_attribute("state", "triggered")

             whence,
             state: RunState::Triggered,
             result: None,
             timed_out: None,
         }
     }
 }

     broker_run_id: RunId,
     adapter_run_id: Option<RunId>,
     adapter_info_url: Option<String>,
     job_id: Option<JobId>,
     repo_id: RepoId,
     #[serde(alias = "repo_alias")]
     repo_name: String,

     whence: Whence,
     state: RunState,
     result: Option<RunResult>,
     // These fields need to be options we can de-serialize from old
     // runs in the database.
     job_id: Option<JobId>,
     timed_out: Option<bool>,
 }
 impl Run {

     pub fn result(&self) -> Option<&RunResult> {
         self.result.as_ref()
     }
     /// Mark run as having timed out.
     pub fn set_timed_out(&mut self) {
         self.timed_out = Some(true);
     }
     /// Was run marked as having timed out?
     pub fn timed_out(&self) -> bool {
         self.timed_out == Some(true)
     }
 }
 /// State of CI run.

     }
     pub fn kill(mut self) -> Result<FinishedProcess, TimeoutError> {
         self.kill_helper();
         self.wait()
     }
     fn kill_helper(&mut self) {
         // Kill the whole process groups. This includes any child
         // processes of the adapter. We terminate with extreme
         // prejudice, as the adapter had all the time the node

             libc::killpg(self.child.id() as i32, libc::SIGKILL);
         }
         self.stdout.finish();
         self.wait()
     }
     pub fn wait(mut self) -> Result<FinishedProcess, TimeoutError> {
         let result = self.child.try_wait();
         match result {
             Ok(Some(status)) => {
                 let stderr = self
                     .stderr_thread
                     .join()
                     .map_err(|_| TimeoutError::Thread)??;
                 return Ok(FinishedProcess {
                     exit: status,
                     stderr,
                 });
             }
             Ok(None) => (),
             Err(err) => return Err(TimeoutError::Wait(err)),
         }
         self.stdout.finish();
         let max_wait = self.deadline - Instant::now();

         match self.stdout_rx.recv_timeout(max_wait) {
             Ok(_) | Err(RecvTimeoutError::Disconnected) => {}
             Err(RecvTimeoutError::Timeout) => {
                 self.kill_helper();
                 return Err(TimeoutError::TimedOut);
             }
         }