From f5cdab804a374ab513a3fdce8a5d9c66215d6cf2 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 01:40:44 +0800 Subject: [PATCH] Submit local and recorded evaluation runs eagerly --- examples/eval_sync.rs | 28 +-- src/eval/error.rs | 6 +- src/eval/mod.rs | 14 +- src/eval/recorder.rs | 11 +- src/eval/run.rs | 128 ++++++++++++++ src/eval/runner.rs | 54 ++---- src/logger.rs | 24 ++- tests/eager_eval.rs | 393 ++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 588 insertions(+), 70 deletions(-) create mode 100644 src/eval/run.rs create mode 100644 tests/eager_eval.rs diff --git a/examples/eval_sync.rs b/examples/eval_sync.rs index ac3c245..09e9c50 100644 --- a/examples/eval_sync.rs +++ b/examples/eval_sync.rs @@ -1,15 +1,11 @@ use braintrust_sdk_rust::eval::*; use braintrust_sdk_rust::*; +use std::future::IntoFuture; fn main() -> anyhow::Result<()> { - // Note: No #[tokio::main] needed! SyncEvaluator handles async internally - - // Create Braintrust client - let rt = tokio::runtime::Runtime::new().map_err(|e| anyhow::anyhow!(e))?; - let client = rt.block_on(async { BraintrustClient::builder().build().await })?; - - // Create sync evaluator - let evaluator = SyncEvaluator::new(client)?; + // The client's owned runtime drives work; this thread only waits for tokens. + let client = futures::executor::block_on(BraintrustClient::builder().build().into_future())?; + let evaluator = Evaluator::new(client); // Define test cases let cases = vec![ @@ -40,12 +36,16 @@ fn main() -> anyhow::Result<()> { println!("Starting synchronous evaluation...\n"); let definition = EvalDefinition::new(Box::new(task), scorers)?; - let result = evaluator.run( - EvalOpts::builder() - .name("sync-uppercase-eval".to_string()) - .source(cases.into()) - .definition(definition.clone()) - .build(), + let result = futures::executor::block_on( + evaluator + .run( + EvalOpts::builder() + .name("sync-uppercase-eval".to_string()) + .source(cases.into()) + .definition(definition.clone()) + .build(), + ) + .into_future(), )?; // Print results diff --git a/src/eval/error.rs b/src/eval/error.rs index 6d9dc92..37b11dc 100644 --- a/src/eval/error.rs +++ b/src/eval/error.rs @@ -43,6 +43,8 @@ impl Serialize for CaseError { pub enum EvalConfigError { #[error("evaluation concurrency must be greater than zero")] ZeroConcurrency, + #[error("recorded evaluations inherit their client runtime; omit the run runtime override")] + RecordedRuntime, #[error("scorer at index {index} must have a nonblank name")] EmptyScorerName { index: usize }, #[error("duplicate scorer name: {name}")] @@ -53,6 +55,8 @@ pub enum EvalConfigError { /// Case indices are zero-based; None denotes run-level recording work. #[derive(Debug, thiserror::Error)] pub enum EvalRunFailure { + #[error("failed to initialize evaluation runtime: {0}")] + Runtime(#[source] crate::BraintrustError), #[error(transparent)] Config(#[from] EvalConfigError), #[error("case source failed at index {case_index}: {source}")] @@ -81,7 +85,7 @@ pub struct EvalRunError { pub partial: Option>>, } impl EvalRunError { - fn new(failure: EvalRunFailure) -> Self { + pub(crate) fn new(failure: EvalRunFailure) -> Self { Self { failures: vec![failure], partial: None, diff --git a/src/eval/mod.rs b/src/eval/mod.rs index 8fe357f..1da6849 100644 --- a/src/eval/mod.rs +++ b/src/eval/mod.rs @@ -19,8 +19,6 @@ //! .build() //! .await?; //! -//! let evaluator = Evaluator::new(client); -//! //! let cases = vec![ //! Case { //! input: "hello".to_string(), @@ -36,15 +34,15 @@ //! }))), //! vec![Box::new(ExactMatch::new())], //! )?; -//! let result = evaluator -//! .run( +//! let completion = run(EvalRequest::recorded( //! EvalOpts::builder() //! .name("uppercase-eval".to_string()) //! .source(cases.into()) //! .definition(definition.clone()) //! .build(), -//! ) -//! .await?; +//! client, +//! )); +//! let result = completion.await?; //! //! println!("Results: {:?}", result); //! # Ok(()) @@ -58,6 +56,7 @@ mod error; mod executor; mod recorder; mod reporter; +mod run; mod runner; mod scorer; mod source; @@ -69,7 +68,8 @@ pub use error::{CaseError, EvalConfigError, EvalRunError, EvalRunFailure}; // Re-export public API pub use bt_runner::{summary_to_value, BtEvalRunner}; -pub use runner::{EvalOpts, Evaluator, SyncEvaluator}; +pub use run::{run, EvalRequest}; +pub use runner::{EvalOpts, Evaluator}; pub use scorer::{scorer, ExactMatch, LevenshteinScorer, NumericDiffScorer, Scorer, SyncFnScorer}; pub use source::{ CaseSource, DatasetVersionRef, IterCaseSource, SourceError, StreamCaseSource, VecCaseSource, diff --git a/src/eval/recorder.rs b/src/eval/recorder.rs index 0f8847f..4f8c6e4 100644 --- a/src/eval/recorder.rs +++ b/src/eval/recorder.rs @@ -15,19 +15,21 @@ use std::sync::Arc; type Write = Completion; pub(crate) struct BraintrustRecorder { - client: Arc, + client: BraintrustClient, + submitter: Arc, root: SpanHandle, root_write: tokio::sync::Mutex>, } impl BraintrustRecorder { pub async fn new( - client: Arc, + client: BraintrustClient, + submitter: Arc, name: String, metadata: Option>, tags: Option>, ) -> Result { let root = client - .span_builder() + .span_builder_for_operation(submitter.clone()) .await? .span_type(crate::SpanType::Eval) .build(); @@ -41,6 +43,7 @@ impl BraintrustRecorder { let root_write = tokio::sync::Mutex::new(Some(root.log(log.build().unwrap()))); Ok(Self { client, + submitter, root, root_write, }) @@ -74,7 +77,7 @@ impl async fn start(&self, index: usize, case: &Case) -> Result>> { let span = self .client - .span_builder() + .span_builder_for_operation(self.submitter.clone()) .await? .span_type(crate::SpanType::Eval) .build(); diff --git a/src/eval/run.rs b/src/eval/run.rs new file mode 100644 index 0000000..4625872 --- /dev/null +++ b/src/eval/run.rs @@ -0,0 +1,128 @@ +//! Immediate submission of local and recorded evaluations. +use super::{ + executor::{EvalExecutor, EvalPlan}, + recorder::BraintrustRecorder, + reporter::TerminalReporter, + EvalConfigError, EvalOpts, EvalRunError, EvalRunFailure, EvalSummary, +}; +use crate::{internal::execution::ExecutionBinding, BraintrustClient, Completion}; +use serde::Serialize; + +type RunCompletion = Completion, EvalRunError>; +type SubmitRecorded = fn(BraintrustClient, EvalOpts) -> RunCompletion; + +/// An owned invocation of a reusable definition with its own source and options. +/// +/// Local requests need no client or serialization. Recorded requests require +/// serialization and inherit their client's execution binding and lifecycle. +pub struct EvalRequest { + options: EvalOpts, + target: Target, +} +enum Target { + Local, + Recorded { + client: BraintrustClient, + submit: SubmitRecorded, + }, +} +impl EvalRequest { + /// Run without login, HTTP or recording, on `options.runtime` or an owned runtime. + pub fn local(options: EvalOpts) -> Self { + Self { + options, + target: Target::Local, + } + } +} +impl EvalRequest { + /// Record with this client's runtime. Supplying `options.runtime` is an error. + pub fn recorded(options: EvalOpts, client: BraintrustClient) -> Self { + Self { + options, + target: Target::Recorded { + client, + submit: recorded::, + }, + } + } +} + +/// Submit an evaluation now; awaiting observes the admitted run's result. +/// Dropping the token does not cancel accepted computation or recording. +/// +/// ``` +/// use braintrust_sdk_rust::eval::{self, Case, EvalDefinition, EvalOpts, EvalRequest, SimpleFnTask}; +/// # fn main() -> anyhow::Result<()> { +/// let definition = EvalDefinition::new( +/// Box::new(SimpleFnTask::new(|input: &usize| Ok(input * 2))), vec![], +/// )?; +/// let options = EvalOpts::builder().name("local".into()) +/// .definition(definition).source(vec![Case { input: 3, ..Default::default() }].into()) +/// .quiet(true).build(); +/// let completion = eval::run(EvalRequest::local(options)); +/// assert!(completion.is_ok()); // admission already happened +/// let summary = futures::executor::block_on(async { completion.await })?; +/// assert_eq!(summary.results[0].output, Some(6)); +/// # Ok(()) +/// # } +/// ``` +pub fn run( + request: EvalRequest, +) -> RunCompletion { + if let Err(error) = request.options.validate() { + return Completion::rejected(error.into()); + } + match request.target { + Target::Local => { + let mut options = request.options; + let binding = match ExecutionBinding::new(options.runtime.take()) { + Ok(binding) => binding, + Err(source) => { + return Completion::rejected(EvalRunError::new(EvalRunFailure::Runtime(source))) + } + }; + binding.response(async move { + let mut reporter = TerminalReporter { + quiet: options.quiet, + }; + EvalExecutor::run(EvalPlan::new(options)?, None, &mut reporter).await + }) + } + Target::Recorded { client, submit } => { + if request.options.runtime.is_some() { + return Completion::rejected(EvalConfigError::RecordedRuntime.into()); + } + submit(client, request.options) + } + } +} + +fn recorded( + client: BraintrustClient, + options: EvalOpts, +) -> RunCompletion { + let binding = client.execution_binding(); + binding.response_try(|| { + let operation = client.span_operation()?; + Ok(async move { + let mut reporter = TerminalReporter { + quiet: options.quiet, + }; + let recorder = BraintrustRecorder::new( + client, + operation.submitter.clone(), + options.name.clone(), + options.metadata.clone(), + options.tags.clone(), + ) + .await?; + let result = + EvalExecutor::run(EvalPlan::new(options)?, Some(&recorder), &mut reporter).await; + drop(recorder); + // Revoke write permission before publishing, including escaped spans. + drop(operation); + result + }) + }) +} diff --git a/src/eval/runner.rs b/src/eval/runner.rs index 498b11e..87f9bc3 100644 --- a/src/eval/runner.rs +++ b/src/eval/runner.rs @@ -3,7 +3,7 @@ use serde_json::{Map, Value}; use std::marker::PhantomData; use std::sync::Arc; -use crate::{BraintrustClient, Result}; +use crate::BraintrustClient; use super::{ source::CaseSource, types::EvalSummary, EvalConfigError, EvalDefinition, EvalRunError, @@ -32,6 +32,8 @@ pub struct EvalOpts { /// Suppress progress output (default: false) #[builder(default = false)] pub quiet: bool, + /// Explicit runtime for local runs. Recorded runs inherit their client binding. + pub runtime: Option, } impl EvalOpts { @@ -44,7 +46,7 @@ impl EvalOpts { } } -/// Async evaluator for running evaluations +/// Client-bound convenience facade for recorded evaluations. pub struct Evaluator { client: Arc, _phantom: PhantomData<(I, O)>, @@ -71,50 +73,16 @@ where } } - /// Run the evaluation - pub async fn run( - &self, - opts: EvalOpts, - ) -> std::result::Result, EvalRunError> { - let name = opts.name.clone(); - let metadata = opts.metadata.clone(); - let tags = opts.tags.clone(); - let mut reporter = super::reporter::TerminalReporter { quiet: opts.quiet }; - let plan = super::executor::EvalPlan::new(opts)?; - let recorder = - super::recorder::BraintrustRecorder::new(self.client.clone(), name, metadata, tags) - .await?; - super::executor::EvalExecutor::run(plan, Some(&recorder), &mut reporter).await - } -} - -/// Sync evaluator (wraps async evaluator with Runtime::block_on) -pub struct SyncEvaluator { - evaluator: Evaluator, - runtime: tokio::runtime::Runtime, -} - -impl SyncEvaluator -where - I: Serialize + Send + Sync + 'static, - O: Serialize + Send + Sync + 'static, -{ - /// Create a new sync evaluator - pub fn new(client: BraintrustClient) -> Result { - let runtime = tokio::runtime::Runtime::new() - .map_err(|e| crate::BraintrustError::Background(e.to_string()))?; - Ok(Self { - evaluator: Evaluator::new(client), - runtime, - }) - } - - /// Run the evaluation synchronously + /// Submit a recorded run immediately on the client binding. + /// Dropping the completion does not cancel accepted work. pub fn run( &self, opts: EvalOpts, - ) -> std::result::Result, EvalRunError> { - self.runtime.block_on(self.evaluator.run(opts)) + ) -> crate::Completion, EvalRunError> { + super::run(super::EvalRequest::recorded( + opts, + self.client.as_ref().clone(), + )) } } diff --git a/src/logger.rs b/src/logger.rs index b106e44..191f73f 100644 --- a/src/logger.rs +++ b/src/logger.rs @@ -611,9 +611,31 @@ impl BraintrustClient { .response(async move { client.span_builder_inner().await }) } + pub(crate) fn execution_binding(&self) -> ExecutionBinding { + self.inner.execution.clone() + } + + pub(crate) fn span_operation( + &self, + ) -> std::result::Result { + self.span_submitter().operation() + } + async fn span_builder_inner(&self) -> Result { + self.span_builder_for_operation(self.span_submitter()).await + } + + pub(crate) async fn span_builder_for_operation( + &self, + submitter: Arc, + ) -> Result { + if self.inner.login_skipped { + return Err(BraintrustError::InvalidConfig( + "recording requires login".into(), + )); + } let state = self.wait_for_login_state().await?; - let mut builder = crate::span::SpanBuilder::new(self.span_submitter(), state.scope()?); + let mut builder = crate::span::SpanBuilder::new(submitter, state.scope()?); if let Some(environment) = &self.inner.environment { builder = builder.span_origin_environment(environment.clone()); } diff --git a/tests/eager_eval.rs b/tests/eager_eval.rs new file mode 100644 index 0000000..64853fd --- /dev/null +++ b/tests/eager_eval.rs @@ -0,0 +1,393 @@ +use braintrust_sdk_rust::{ + eval::*, BraintrustClient, EnqueueError, ExecutionError, SpanLog, WriteError, +}; +use std::{ + future::IntoFuture, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, + time::Duration, +}; +use wiremock::{ + matchers::{method, path}, + Mock, MockServer, ResponseTemplate, +}; + +fn options( + definition: EvalDefinition, + cases: Vec>, +) -> EvalOpts { + EvalOpts::builder() + .name("eager".into()) + .definition(definition) + .source(cases.into()) + .quiet(true) + .build() +} +fn numeric() -> EvalOpts { + options( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &usize| Ok(input * 2))), + vec![], + ) + .unwrap(), + vec![Case { + input: 3, + ..Default::default() + }], + ) +} + +#[test] +fn local_runs_from_a_plain_thread_without_serialization_clone_or_debug() { + struct Input(usize); + struct Output(usize); + let definition = EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &Input| Ok(Output(input.0 * 2)))), + vec![], + ) + .unwrap(); + let token = run(EvalRequest::local(options( + definition, + vec![Case { + input: Input(3), + expected: None, + metadata: None, + tags: None, + id: None, + xact_id: None, + created: None, + }], + ))); + assert!(token.is_ok()); + let result = futures::executor::block_on(token.into_future()).unwrap(); + assert_eq!(result.results[0].output.as_ref().unwrap().0, 6); + assert!(result.results[0].span_id.is_none()); +} + +#[test] +fn dropping_the_observer_keeps_the_source_and_callbacks_running() { + let (done, observed) = std::sync::mpsc::channel(); + let definition = EvalDefinition::new( + Box::new(SimpleFnTask::new(move |input: &usize| { + done.send(*input).unwrap(); + Ok(*input) + })), + vec![], + ) + .unwrap(); + let token = run(EvalRequest::local(options( + definition, + vec![Case { + input: 7, + ..Default::default() + }], + ))); + assert!(token.is_ok()); + drop(token); + assert_eq!(observed.recv_timeout(Duration::from_secs(3)).unwrap(), 7); +} + +#[tokio::test] +async fn local_runtime_selection_is_explicit_and_never_uses_ambient_by_default() { + let ambient = tokio::runtime::Handle::current().id(); + let definition = EvalDefinition::new( + Box::new(SimpleFnTask::new(|_: &usize| { + Ok(tokio::runtime::Handle::current().id()) + })), + vec![], + ) + .unwrap(); + for explicit in [false, true] { + let mut opts = options( + definition.clone(), + vec![Case { + input: 0, + ..Default::default() + }], + ); + opts.runtime = explicit.then(tokio::runtime::Handle::current); + let summary = run(EvalRequest::local(opts)).await.unwrap(); + assert_eq!(summary.results[0].output == Some(ambient), explicit); + } +} + +#[test] +fn external_runtime_loss_before_first_poll_reports_worker_loss_without_fallback() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let calls = Arc::new(AtomicUsize::new(0)); + let definition = EvalDefinition::new( + Box::new(SimpleFnTask::new({ + let calls = calls.clone(); + move |input: &usize| { + calls.fetch_add(1, Ordering::SeqCst); + Ok(*input) + } + })), + vec![], + ) + .unwrap(); + let mut opts = options( + definition, + vec![Case { + input: 0, + ..Default::default() + }], + ); + opts.runtime = Some(runtime.handle().clone()); + let token = run(EvalRequest::local(opts)); + assert!(token.is_ok()); + assert_eq!(calls.load(Ordering::SeqCst), 0); + drop(runtime); + let error = futures::executor::block_on(token.into_future()).unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [EvalRunFailure::Execution(ExecutionError::WorkerLost)] + )); + assert_eq!(calls.load(Ordering::SeqCst), 0); +} + +#[test] +fn invalid_options_return_an_immediate_config_error() { + let mut opts = numeric(); + opts.max_concurrency = 0; + let token = run(EvalRequest::local(opts)); + assert!(token.is_err()); + let error = futures::executor::block_on(token.into_future()).unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [EvalRunFailure::Config(EvalConfigError::ZeroConcurrency)] + )); + assert!(error.partial.is_none()); +} + +async fn client(server: &MockServer) -> BraintrustClient { + for (endpoint, body) in [ + ( + "/api/apikey/login", + serde_json::json!({"org_info":[{"id":"org","name":"org"}]}), + ), + ( + "/api/project/register", + serde_json::json!({"project":{"id":"project"}}), + ), + ("/logs3", serde_json::json!({})), + ] { + Mock::given(method("POST")) + .and(path(endpoint)) + .respond_with(ResponseTemplate::new(200).set_body_json(body)) + .mount(server) + .await; + } + Mock::given(method("GET")) + .and(path("/version")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({}))) + .mount(server) + .await; + BraintrustClient::builder() + .runtime(braintrust_sdk_rust::current_runtime()) + .app_url(server.uri()) + .api_url(server.uri()) + .api_key("token") + .default_project("project") + .blocking_login(true) + .build() + .await + .unwrap() +} + +#[tokio::test] +async fn recorded_run_admitted_before_shutdown_finishes_writes_and_revokes_escaped_spans() { + let server = MockServer::start().await; + let client = client(&server).await; + let escaped = Arc::new(Mutex::new(None)); + let definition = EvalDefinition::new( + Box::new(task({ + let escaped = escaped.clone(); + move |input: &usize, hooks| { + let escaped = escaped.clone(); + Box::pin(async move { + let child = hooks + .span + .unwrap() + .child_builder() + .unwrap() + .name("callback-child") + .build(); + child + .log( + SpanLog::builder() + .input(serde_json::json!(input)) + .build() + .unwrap(), + ) + .await + .unwrap(); + *escaped.lock().unwrap() = Some(child); + Ok(*input) + }) + } + })), + vec![], + ) + .unwrap(); + let token = run(EvalRequest::recorded( + options( + definition, + vec![Case { + input: 7, + ..Default::default() + }], + ), + client.clone(), + )); + assert!(token.is_ok()); + let shutdown = client.shutdown(Duration::from_secs(5)); + let rejected = run(EvalRequest::recorded(numeric(), client.clone())); + assert!(rejected.is_err()); + assert!(matches!( + rejected.await.unwrap_err().failures.as_slice(), + [EvalRunFailure::Admission(EnqueueError::Closed)] + )); + let (result, shutdown) = tokio::join!(token, shutdown); + let result = result.unwrap(); + shutdown.unwrap(); + assert_eq!(result.results[0].output, Some(7)); + assert!(result.results[0].span_id.is_some()); + let escaped = escaped.lock().unwrap().take().unwrap(); + assert!(matches!( + escaped.end().await, + Err(WriteError::Admission(EnqueueError::Closed)) + )); +} + +#[tokio::test] +async fn recorded_runtime_override_is_rejected_and_dropped_observer_still_drains() { + let server = MockServer::start().await; + let client = client(&server).await; + let mut opts = numeric(); + opts.runtime = braintrust_sdk_rust::current_runtime(); + let rejected = run(EvalRequest::recorded(opts, client.clone())); + assert!(rejected.is_err()); + assert!(matches!( + rejected.await.unwrap_err().failures.as_slice(), + [EvalRunFailure::Config(EvalConfigError::RecordedRuntime)] + )); + let calls = Arc::new(AtomicUsize::new(0)); + let definition = EvalDefinition::new( + Box::new(SimpleFnTask::new({ + let calls = calls.clone(); + move |input: &usize| { + calls.fetch_add(1, Ordering::SeqCst); + Ok(*input) + } + })), + vec![], + ) + .unwrap(); + drop(run(EvalRequest::recorded( + options( + definition, + vec![Case { + input: 1, + ..Default::default() + }], + ), + client.clone(), + ))); + client.shutdown(Duration::from_secs(5)).await.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn callback_panic_retires_recorded_run_and_revokes_escaped_permission() { + let server = MockServer::start().await; + let client = client(&server).await; + let escaped = Arc::new(Mutex::new(None)); + let definition: EvalDefinition = EvalDefinition::new( + Box::new(task({ + let escaped = escaped.clone(); + move |_: &usize, hooks| { + *escaped.lock().unwrap() = + Some(hooks.span.unwrap().child_builder().unwrap().build()); + Box::pin(async { panic!("application callback panic") }) + } + })), + vec![], + ) + .unwrap(); + let token = run(EvalRequest::recorded( + options( + definition, + vec![Case { + input: 0, + ..Default::default() + }], + ), + client.clone(), + )); + let error = token.await.unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [EvalRunFailure::Execution(ExecutionError::WorkerLost)] + )); + client.shutdown(Duration::from_secs(5)).await.unwrap(); + let escaped = escaped.lock().unwrap().take().unwrap(); + assert!(matches!( + escaped.end().await, + Err(WriteError::Admission(EnqueueError::Closed)) + )); +} + +#[tokio::test] +async fn shutdown_deadline_cancels_pending_run_and_revokes_escaped_permission() { + let server = MockServer::start().await; + let client = client(&server).await; + let escaped = Arc::new(Mutex::new(None)); + let entered = Arc::new(tokio::sync::Notify::new()); + let definition: EvalDefinition = EvalDefinition::new( + Box::new(task({ + let escaped = escaped.clone(); + let entered = entered.clone(); + move |_: &usize, hooks| { + *escaped.lock().unwrap() = + Some(hooks.span.unwrap().child_builder().unwrap().build()); + entered.notify_one(); + Box::pin(std::future::pending()) + } + })), + vec![], + ) + .unwrap(); + let token = run(EvalRequest::recorded( + options( + definition, + vec![Case { + input: 0, + ..Default::default() + }], + ), + client.clone(), + )); + entered.notified().await; + assert!(matches!( + client.shutdown(Duration::from_millis(20)).await, + Err(braintrust_sdk_rust::ShutdownError::Execution( + ExecutionError::TimedOut + )) + )); + let error = token.await.unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [EvalRunFailure::Execution(ExecutionError::TimedOut)] + )); + let escaped = escaped.lock().unwrap().take().unwrap(); + assert!(matches!( + escaped.end().await, + Err(WriteError::Admission(EnqueueError::Closed)) + )); +}