fix: capture and log all stderr from adapter — radicle-ci-broker

Previously, we failed to capture all of stderr from the adapter if the adapter sent messages to its stdout could not be parsed, or otherwise meant the CI broker deduced the CI run had failed. Now we read all of stderr (module time limits) before returning an error result.

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

 _Who:_ `adapter-devs`, `node-ops`
 ~~~scenario
 given a Radicle node, with CI configured with broker.yaml and adapter dummy.sh
 given a Radicle node, with CI configured with broker.yaml and adapter broken-adapter.sh
 given a Git repository xyzzy in the Radicle node
 given the Radicle node emits a refsUpdated event for xyzzy
 when I run ./env.sh synthetic-events synt.sock event.json --log log.txt

 given a directory reports
 when I run ./env.sh cib --config broker.yaml queued
 then stderr contains "Rivendell"
 then stderr contains "Mordor"
 ~~~
 This adapter outputs a broken response message, and after that
 something to its stderr. The CI broker is meant to read and log both.
 ~~~{#broken-adapter.sh .file .sh}
 #!/bin/bash
 set -euo pipefail
 cat > /dev/null
 echo '{"response":"Rivendell"}}'
 echo "This is an adapter error: Mordor" 1>&2
 ~~~
 ## Allows setting minimum log level
 _What:_ The node admin should be able to set the minimum log level for

     time::Duration,
 };
 //use tracing::Instrument;
 use crate::{
     db::{Db, DbError},
     logger,

     notif::NotificationSender,
     run::{Run, RunState},
     sensitive::Sensitive,
     timeoutcmd::{TimeoutCommand, TimeoutError},
     timeoutcmd::{LineReceiver, TimeoutCommand, TimeoutError},
 };
 const NOT_EXITED: i32 = 999;

         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");
         logger::debug2(format!(
             "wait result? {wait_result:?} status.code: {:?}",
             wait_result.status().code()
         ));
         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));
             }
         } else {
             logger::adapter_did_not_exit();
             outcome.set_error(AdapterError::Signal);
         }
         if let Some(err) = outcome.error() {
             Err(err)
         } else {
             Ok(())
         }
     }
     fn read_stdout(
         &self,
         run: &mut Run,
         db: &Db,
         run_notification: &NotificationSender,
         stdout: &LineReceiver,
     ) -> Result<(), AdapterError> {
         if let Some(line) = stdout.line() {
             let resp = Response::from_str(&line).map_err(AdapterError::ParseResponse)?;
             run_notification.notify()?;

                     db.update_run(run).map_err(AdapterError::UpdateRun)?;
                 }
                 _ => {
                     child.kill().ok();
                     return Err(AdapterError::NotTriggered(resp));
                 }
             }
         } else {
             logger::adapter_no_first_response();
             child.kill().ok();
             return Err(AdapterError::NoFirstMessage);
         }

                     db.update_run(run).map_err(AdapterError::UpdateRun)?;
                 }
                 _ => {
                     child.kill().ok();
                     return Err(AdapterError::NotFinished(resp));
                 }
             }
         } else {
             logger::adapter_no_second_response();
             child.kill().ok();
             return Err(AdapterError::NoSecondMessage);
         }
         if let Some(line) = stdout.line() {
             let resp = Response::from_str(&line).map_err(AdapterError::ParseResponse)?;
             logger::adapter_too_many_responses();
             child.kill().ok();
             return Err(AdapterError::TooMany(resp));
         }
         let stderr = child.stderr();
         Ok(())
     }
     fn read_stderr(&self, stderr: &LineReceiver) {
         while let Some(line) = stderr.line() {
             logger::adapter_stderr_line(&line);
         }
     }
 }
         let result = child.wait().expect("FIXME");
         logger::debug2(format!(
             "wait result? {result:?} status.code: {:?}",
             result.status().code()
         ));
 #[derive(Default)]
 struct MaybeResult {
     error: Option<AdapterError>,
 }
         if result.timed_out() {
             logger::adapter_did_not_exit_voluntarily();
             return Err(AdapterError::Failed(NOT_EXITED));
         } else if let Some(exit) = result.status().code() {
             logger::adapter_result(exit);
             if exit != 0 {
                 return Err(AdapterError::Failed(exit));
             }
         } else {
             logger::adapter_did_not_exit();
             return Err(AdapterError::Signal);
 impl MaybeResult {
     fn set_error(&mut self, err: AdapterError) {
         if self.error.is_none() {
             self.error = Some(err);
         }
     }
         Ok(())
     fn has_error(&self) -> bool {
         self.error.is_some()
     }
     fn error(self) -> Option<AdapterError> {
         self.error
     }
 }

             spawn(move || NonBlockingReader::new("stdout", stdout, stdout_lines_tx).read_to_end());
         let stdout_lines = LineReceiver::new("stdout", stdout_lines_rx, stdout_termination_rx);
         // Set up thread to capture child stdout.
         // Set up thread to capture child stderr.
         let stderr = child.stderr.take().ok_or(TimeoutError::TakeStderr)?;
         let (stderr_termination_tx, stderr_termination_rx) = sync_channel(1);
         let (stderr_lines_tx, stderr_lines_rx) = sync_channel(MAX_OUTPUT_BYTES);