f834f3e fix: handle adapter process exiting non-zero — radicle-ci-broker

modified src/adapter.rs

@@ -10,18 +10,21 @@

 use std::{
     collections::HashMap,
     ffi::OsStr,
     io::{BufRead, BufReader},
     os::unix::process::ExitStatusExt,
     io::{BufRead, BufReader, Read},
     path::{Path, PathBuf},
     process::{Command, Stdio},
 };
 use log::{debug, error};
 use crate::{
     msg::{MessageError, Request, Response},
     pages::StatusPage,
     run::{Run, RunState},
 };
 const NOT_EXITED: i32 = 999;
 /// An external executable that runs CI on request.
 #[derive(Debug, Default, Clone, Eq, PartialEq)]
 pub struct Adapter {

@@ -54,6 +57,7 @@ impl Adapter {

         run: &mut Run,
         status: &mut StatusPage,
     ) -> Result<(), AdapterError> {
         debug!("running adapter");
         run.set_state(RunState::Triggered);
         status.push_run(run.clone());
         let x = self.run_helper(trigger, run, status);

@@ -71,9 +75,11 @@ impl Adapter {

         assert!(matches!(trigger, Request::Trigger { .. }));
         // Spawn the adapter sub-process.
         debug!("spawn adapter sub-process");
         let mut child = Command::new(&self.bin)
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
             .envs(self.envs())
             .spawn()
             .map_err(|e| AdapterError::SpawnAdapter(self.bin.clone(), e))?;

@@ -81,6 +87,7 @@ impl Adapter {

         // 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.
         debug!("send request to sub-process stdin");
         {
             let stdin = child.stdin.take().ok_or(AdapterError::StdinHandle)?;
             trigger.to_writer(stdin)?;

@@ -95,6 +102,7 @@ impl Adapter {

         if let Some(line) = lines.next() {
             let line = line.map_err(AdapterError::ReadLine)?;
             let resp = Response::from_str(&line)?;
             debug!("first response {resp:#?}");
             match resp {
                 Response::Triggered { run_id } => {
                     run.set_state(RunState::Running);

@@ -103,11 +111,14 @@ impl Adapter {

                 }
                 _ => return Err(AdapterError::NotTriggered(resp)),
             }
         } else {
             debug!("no first response message");
         }
         if let Some(line) = lines.next() {
             let line = line.map_err(AdapterError::ReadLine)?;
             let resp = Response::from_str(&line)?;
             debug!("second response {resp:#?}");
             match resp {
                 Response::Finished { result } => {
                     run.set_result(result);

@@ -115,18 +126,36 @@ impl Adapter {

                 }
                 _ => return Err(AdapterError::NotTriggered(resp)),
             }
         } else {
             debug!("no second response message");
         }
         if let Some(line) = lines.next() {
             let line = line.map_err(AdapterError::ReadLine)?;
             let resp = Response::from_str(&line)?;
             debug!("third response is too many {resp:#?}");
             return Err(AdapterError::TooMany(resp));
         } else {
             debug!("no further response messages");
         }
         let wait = child.wait().map_err(AdapterError::Wait)?;
         let wait = wait.into_raw();
         if wait != 0 {
             return Err(AdapterError::Failed(wait));
         debug!("adapter wait result {wait:#?}");
         if let Some(exit) = wait.code() {
             debug!("adapter exit was {exit}");
             if exit != 0 {
                 let mut stderr = child.stderr.take().ok_or(AdapterError::StderrHandle)?;
                 let mut buf = vec![];
                 stderr
                     .read_to_end(&mut buf)
                     .map_err(AdapterError::ReadStderr)?;
                 let stderr = String::from_utf8_lossy(&buf);
                 debug!("adapter stderr: {stderr:?}");
                 return Err(AdapterError::Failed(exit));
             }
         } else {
             error!("adapter sub-process did not exit voluntarily");
             return Err(AdapterError::Failed(NOT_EXITED));
         }
         Ok(())

@@ -151,6 +180,14 @@ pub enum AdapterError {

     #[error("failed to get handle for adapter's stdout")]
     StdoutHandle,
     /// Error getting the file handle for the adapter's stderr.
     #[error("failed to get handle for adapter's stderr")]
     StderrHandle,
     /// Error reading adapter's stderr.
     #[error("failed to read the adapter's stderr")]
     ReadStderr(#[source] std::io::Error),
     #[error("failed to read from adapter stdout")]
     ReadLine(#[source] std::io::Error),

@@ -186,7 +223,7 @@ mod test {

         msg::{MessageError, Response, RunResult},
         pages::PageBuilder,
         run::Whence,
         test::{mock_adapter, trigger_request, TestResult},
         test::{log_in_tests, mock_adapter, trigger_request, TestResult},
     };
     fn run() -> Run {

@@ -210,6 +247,8 @@ mod test {

     #[test]
     fn adapter_reports_success() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"success"}'

@@ -229,6 +268,8 @@ echo '{"response":"finished","result":"success"}'

     #[test]
     fn adapter_reports_failure() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"failure"}'

@@ -247,7 +288,33 @@ echo '{"response":"finished","result":"failure"}'

     }
     #[test]
     fn adapter_exits_nonzero() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"failure"}'
 echo woe be me 1>&2
 exit 1
 "#;
         let tmp = tempdir()?;
         let bin = tmp.path().join("adapter.sh");
         mock_adapter(&bin, ADAPTER)?;
         let mut run = run();
         let mut status = status_page();
         let x = Adapter::new(&bin).run(&trigger_request()?, &mut run, &mut status);
         assert!(x.is_err());
         assert_eq!(run.result(), Some(&RunResult::Failure));
         Ok(())
     }
     #[test]
     fn adapter_is_killed_before_any_messages() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 kill -9 $BASHPID
 "#;

@@ -266,6 +333,8 @@ kill -9 $BASHPID

     #[test]
     fn adapter_is_killed_after_first_message() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 kill -9 $BASHPID

@@ -285,6 +354,8 @@ kill -9 $BASHPID

     #[test]
     fn adapter_is_killed_after_second_message() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"success"}'

@@ -305,6 +376,8 @@ kill -9 $BASHPID

     #[test]
     fn adapter_produces_as_bad_message() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"success","bad":"field"}'

@@ -327,6 +400,8 @@ echo '{"response":"finished","result":"success","bad":"field"}'

     #[test]
     fn adapter_first_message_isnt_triggered() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"finished","result":"success"}'
 "#;

@@ -350,6 +425,8 @@ echo '{"response":"finished","result":"success"}'

     #[test]
     fn adapter_outputs_too_many_messages() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"success"}'

@@ -375,6 +452,8 @@ echo '{"response":"finished","result":"success"}'

     #[test]
     fn adapter_does_not_exist() -> TestResult<()> {
         log_in_tests();
         let tmp = tempdir()?;
         let bin = tmp.path().join("adapter.sh");

@@ -394,6 +473,8 @@ echo '{"response":"finished","result":"success"}'

     #[test]
     fn adapter_is_not_executable() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"success"}'

@@ -419,6 +500,8 @@ echo '{"response":"finished","result":"success"}'

     #[test]
     fn adapter_has_bad_interpreter() -> TestResult<()> {
         log_in_tests();
         // We test this with a shebang. However, the same kind of code
         // paths and errors should happen when a binary can't be
         // loaded due to missing dynamic linker or library or such.

modified src/bin/ci-broker.rs

@@ -5,7 +5,7 @@ use std::{

     time::Duration,
 };
 use log::{debug, info};
 use log::{debug, error, info};
 use radicle::prelude::Profile;
 use radicle_ci_broker::{

@@ -111,7 +111,9 @@ fn fallible_main() -> Result<(), BrokerError> {

                 .profile(&profile)
                 .broker_event(&e)
                 .build_trigger()?;
             broker.execute_ci(&req, &mut page)?;
             if let Err(e) = broker.execute_ci(&req, &mut page) {
                 error!("failed to run adapter, or adapter failed to run CI: {e}");
             }
         }
     }
 }

modified src/broker.rs

@@ -5,7 +5,7 @@

 use std::{collections::HashMap, path::Path};
 use log::{debug, info};
 use log::{debug, error, info};
 use time::{macros::format_description, OffsetDateTime};
 use radicle::prelude::RepoId;

@@ -101,7 +101,16 @@ impl Broker {

                     };
                     let mut run = Run::new(*rid, &common.repository.name, whence, now());
                     adapter.run(trigger, &mut run, status)?;
                     // 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, status) {
                         error!("failed to run adapter or it failed to run CI: {e}");
                     }
                     debug!("broker run {run:#?}");
                     run
                 } else {
                     return Err(BrokerError::NoAdapter(*rid));

@@ -109,7 +118,7 @@ impl Broker {

             }
         };
         self.db.push_run(&run)?;
         info!("Finish CI run: {run:#?}");
         debug!("Finish CI run: {run:#?}");
         Ok(run)
     }

@@ -130,7 +139,7 @@ mod test {

         msg::{RunId, RunResult},
         pages::{PageBuilder, StatusPage},
         run::RunState,
         test::{mock_adapter, trigger_request, TestResult},
         test::{log_in_tests, mock_adapter, trigger_request, TestResult},
     };
     fn broker(filename: &Path) -> Broker {

@@ -255,4 +264,37 @@ echo '{"response":"finished","result":"success"}'

         Ok(())
     }
     #[test]
     fn adapter_fails() -> TestResult<()> {
         log_in_tests();
         const ADAPTER: &str = r#"#!/bin/bash
 echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'
 echo '{"response":"finished","result":"success"}'
 echo woe be me 1>&2
 exit 1
 "#;
         let tmp = tempdir()?;
         let bin = tmp.path().join("adapter.sh");
         let adapter = mock_adapter(&bin, ADAPTER)?;
         let tmp = tempdir().unwrap();
         let db = tmp.path().join("db.db");
         let mut broker = broker(&db);
         broker.set_default_adapter(&adapter);
         let trigger = trigger_request()?;
         let mut status = status_page();
         let x = broker.execute_ci(&trigger, &mut status);
         assert!(x.is_ok());
         let run = x.unwrap();
         assert_eq!(run.adapter_run_id(), Some(&RunId::from("xyzzy")));
         assert_eq!(run.state(), RunState::Finished);
         assert_eq!(run.result(), Some(&RunResult::Success));
         Ok(())
     }
 }

modified src/test.rs

@@ -94,3 +94,11 @@ pub fn mock_adapter(filename: &Path, script: &str) -> TestResult<Adapter> {

     Ok(Adapter::new(filename))
 }
 pub fn log_in_tests() {
     pretty_env_logger::formatted_builder()
         .filter_level(log::LevelFilter::Debug)
         .is_test(true)
         .try_init()
         .ok();
 }