diff --git a/src/api/btql.rs b/src/api/btql.rs index f4075ab..3fe4d45 100644 --- a/src/api/btql.rs +++ b/src/api/btql.rs @@ -33,6 +33,9 @@ pub struct DatasetRecord { #[derive(Debug, Clone, Serialize)] pub struct BTQLQuery { pub query: BTQLQueryInner, + /// Optional immutable dataset snapshot version. + #[serde(skip_serializing_if = "Option::is_none")] + pub version: Option, } /// Inner BTQL query structure. diff --git a/src/dataset.rs b/src/dataset.rs index dc4a6a4..29ece2f 100644 --- a/src/dataset.rs +++ b/src/dataset.rs @@ -595,6 +595,7 @@ struct DatasetCursor { scope: CredentialScope, dataset_id: String, batch_size: usize, + version: Option, cursor: Option, buffer: VecDeque, done: bool, @@ -634,6 +635,7 @@ impl DatasetIterator { scope, dataset_id, batch_size, + version: None, cursor: None, buffer: VecDeque::new(), done: false, @@ -690,6 +692,8 @@ impl DatasetIterator { } } +pub(crate) mod read_session; + impl DatasetCursor { /// Get the next record, fetching a new batch if needed. async fn next(&mut self) -> Result> { @@ -707,6 +711,7 @@ impl DatasetCursor { limit: Some(self.batch_size), cursor: previous_cursor.clone(), }, + version: self.version.clone(), }; // A failed request leaves the cursor and EOF state untouched. let response = self diff --git a/src/dataset/read_session.rs b/src/dataset/read_session.rs new file mode 100644 index 0000000..b6571d0 --- /dev/null +++ b/src/dataset/read_session.rs @@ -0,0 +1,143 @@ +//! One admitted, demand-driven dataset cursor on its resource's fixed binding. +use super::*; +use std::future::IntoFuture; +use tokio::sync::{mpsc, oneshot}; + +#[derive(Debug, Clone)] +pub(crate) struct ReadError(Arc); +impl std::fmt::Display for ReadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} +impl std::error::Error for ReadError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.0.as_ref()) + } +} +impl From for ReadError { + fn from(error: BraintrustError) -> Self { + Self(Arc::new(error)) + } +} +impl From for ReadError { + fn from(error: crate::ExecutionError) -> Self { + BraintrustError::from(error).into() + } +} +impl From for ReadError { + fn from(error: crate::EnqueueError) -> Self { + BraintrustError::from(error).into() + } +} + +type Reply = oneshot::Sender, ReadError>>; +type Driver = futures::future::Shared>>; + +pub(crate) struct ReadSession { + requests: Option>, + ready: Option>, + metadata: Option, + driver: Driver, + done: bool, +} +impl ReadSession { + pub fn start( + dataset: Dataset, + batch_size: Option, + version: Option, + ) -> std::result::Result { + let (requests, mut receiver) = mpsc::channel::(1); + let (ready, metadata) = oneshot::channel(); + let execution = dataset.execution.clone(); + let completion = execution.response(async move { + let metadata = dataset.ensure_registered().await?.clone(); + let mut cursor = DatasetCursor { + fetcher: dataset.api.clone(), + scope: dataset.scope.clone(), + dataset_id: metadata.dataset_id.clone(), + batch_size: batch_size.unwrap_or(1000).max(1), + version, + cursor: None, + buffer: VecDeque::new(), + done: false, + }; + if ready.send(metadata).is_err() { + return Ok(()); + } + while let Some(reply) = receiver.recv().await { + let result = cursor.next().await.map_err(ReadError::from); + let terminal = !matches!(result, Ok(Some(_))); + let failure = result.as_ref().err().cloned(); + if reply.send(result).is_err() { + break; + } + if let Some(error) = failure { + return Err(error); + } + if terminal { + break; + } + } + Ok(()) + }); + if let Err(error) = &*completion { + return Err(error.clone()); + } + Ok(Self { + requests: Some(requests), + ready: Some(metadata), + metadata: None, + driver: completion.into_future().boxed().shared(), + done: false, + }) + } + + pub async fn prepare(&mut self) -> std::result::Result<&DatasetMetadata, ReadError> { + if let Some(ready) = self.ready.take() { + self.metadata = Some(self.receive(ready).await?); + } + self.metadata.as_ref().ok_or_else(stopped) + } + + pub async fn next(&mut self) -> std::result::Result, ReadError> { + self.prepare().await?; + if self.done { + return Ok(None); + } + let (reply, received) = oneshot::channel(); + if self + .requests + .as_ref() + .ok_or_else(stopped)? + .send(reply) + .await + .is_err() + { + return Err(self.driver.clone().await.err().unwrap_or_else(stopped)); + } + let result = self.receive(received).await?; + let result = result?; + if result.is_none() { + self.done = true; + self.requests = None; + } + Ok(result) + } + + async fn receive( + &self, + receiver: oneshot::Receiver, + ) -> std::result::Result { + use futures::future::{select, Either}; + // Prefer a reply already produced before a terminal driver result. + match select(receiver, self.driver.clone()).await { + Either::Left((Ok(value), _)) => Ok(value), + Either::Left((Err(_), driver)) => Err(driver.await.err().unwrap_or_else(stopped)), + Either::Right((result, _)) => Err(result.err().unwrap_or_else(stopped)), + } + } +} +fn stopped() -> ReadError { + BraintrustError::Background("dataset read session stopped before replying".into()).into() +} diff --git a/src/eval/executor.rs b/src/eval/executor.rs index 9161636..e0dab25 100644 --- a/src/eval/executor.rs +++ b/src/eval/executor.rs @@ -87,12 +87,19 @@ impl EvalExecutor { binding: Option<&crate::internal::execution::ExecutionBinding>, ) -> std::result::Result, EvalRunError> { reporter.started(&plan.name); + let prepared = plan.source.prepare().await; let dataset = plan.source.provenance(); let mut results = Vec::new(); let mut failures = Vec::new(); + if let Err(source) = prepared { + failures.push(EvalRunFailure::Source { + case_index: 0, + source, + }); + } let mut case_count = 0; let mut pending = FuturesUnordered::new(); - loop { + while failures.is_empty() { // Keep one source read alive while cases run. Sources may await // case progress and need not support cancellation/restarting next(). let next_case = { diff --git a/src/eval/mod.rs b/src/eval/mod.rs index 121f193..8a556b3 100644 --- a/src/eval/mod.rs +++ b/src/eval/mod.rs @@ -73,7 +73,8 @@ 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, + CaseSource, DatasetCaseSource, DatasetVersionRef, IterCaseSource, SourceError, + StreamCaseSource, VecCaseSource, }; pub use task::{task, SimpleFnTask, SyncFnTask, Task}; pub use types::{ diff --git a/src/eval/source.rs b/src/eval/source.rs index 025f2ee..2456b2e 100644 --- a/src/eval/source.rs +++ b/src/eval/source.rs @@ -34,6 +34,11 @@ pub struct DatasetVersionRef { /// without requiring Sync, cloning user values, or serializing local cases. #[async_trait] pub trait CaseSource: Send { + /// Resolve source identity before provenance is captured or cases are read. + async fn prepare(&mut self) -> Result<(), SourceError> { + Ok(()) + } + async fn next(&mut self) -> Result>, SourceError>; fn provenance(&self) -> Option { None @@ -103,3 +108,94 @@ impl From>> for Box anyhow::Result<()> { +/// let mut cases = DatasetCaseSource::::new( +/// dataset, Some("snapshot-version".into()), Some(100), +/// )?; +/// cases.prepare().await?; +/// while let Some(case) = cases.next().await? { println!("{}", case.input); } +/// # Ok(()) +/// # } +/// ``` +pub struct DatasetCaseSource { + session: crate::dataset::read_session::ReadSession, + provenance: Option, + version: Option, + marker: std::marker::PhantomData (I, O)>, +} +impl DatasetCaseSource { + pub fn new( + dataset: crate::Dataset, + version: Option, + batch_size: Option, + ) -> Result { + Ok(Self { + session: crate::dataset::read_session::ReadSession::start( + dataset, + batch_size, + version.clone(), + )?, + provenance: None, + version, + marker: std::marker::PhantomData, + }) + } +} +#[async_trait] +impl CaseSource + for DatasetCaseSource +{ + async fn prepare(&mut self) -> Result<(), SourceError> { + let metadata = self.session.prepare().await?; + if self.provenance.is_none() { + self.provenance = Some(DatasetVersionRef { + dataset_id: metadata.dataset_id.clone(), + version: self.version.clone(), + }); + } + Ok(()) + } + fn provenance(&self) -> Option { + self.provenance.clone() + } + async fn next(&mut self) -> Result>, SourceError> { + self.prepare().await?; + let Some(record) = self.session.next().await? else { + return Ok(None); + }; + let input = record.input.unwrap_or(serde_json::Value::Null); + let input = serde_json::from_value(input).map_err(|error| { + anyhow::Error::new(error).context(format!("dataset record {} input", record.id)) + })?; + let expected = record + .expected + .map(serde_json::from_value) + .transpose() + .map_err(|error| { + anyhow::Error::new(error).context(format!("dataset record {} expected", record.id)) + })?; + Ok(Some(Case { + input, + expected, + metadata: record.metadata, + tags: record.tags, + id: Some(record.id), + xact_id: record.xact_id, + created: record.created, + })) + } +} diff --git a/src/internal/upload.rs b/src/internal/upload.rs index 57a6258..d03b63c 100644 --- a/src/internal/upload.rs +++ b/src/internal/upload.rs @@ -313,7 +313,11 @@ mod tests { .any(|key| key.as_str() == "x-bt-org-name")); assert_eq!(request.body, data); } - let grant: Value = serde_json::from_slice(&requests[0].body).unwrap(); + let grant_request = requests + .iter() + .find(|request| request.url.path() == "/logs3/overflow") + .expect("overflow grant request"); + let grant: Value = serde_json::from_slice(&grant_request.body).unwrap(); assert_eq!(grant["size_bytes"], data.len()); assert_eq!(grant["rows"][0]["object_ids"]["project_id"], "project"); } diff --git a/tests/api_client.rs b/tests/api_client.rs index 450d79a..3a908e2 100644 --- a/tests/api_client.rs +++ b/tests/api_client.rs @@ -147,6 +147,7 @@ async fn btql_readback_preserves_cursor_and_request_errors() { limit: Some(5), cursor: Some("next".into()), }, + version: None, }) .await .unwrap(); @@ -169,7 +170,8 @@ async fn btql_readback_preserves_cursor_and_request_errors() { from: "dataset('id')".into(), limit: None, cursor: None, - } + }, + version: None, }) .await, Err(api::Error::Http { status: 500, .. }) diff --git a/tests/dataset_case_source.rs b/tests/dataset_case_source.rs new file mode 100644 index 0000000..bd45263 --- /dev/null +++ b/tests/dataset_case_source.rs @@ -0,0 +1,417 @@ +use braintrust_sdk_rust::{eval::*, BraintrustClient, BraintrustError, Dataset, EnqueueError}; +use serde::Deserialize; +use serde_json::json; +use std::{ + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, + time::Duration, +}; +use wiremock::{ + matchers::{header, method, path}, + Mock, MockServer, ResponseTemplate, +}; + +async fn setup(recording: bool) -> (BraintrustClient, Dataset, MockServer, MockServer) { + let control = MockServer::start().await; + let data = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/api/apikey/login")) + .respond_with(ResponseTemplate::new(200).set_body_json( + json!({"org_info":[{"id":"record-org","name":"record-org","api_url":data.uri()}]}), + )) + .mount(&control) + .await; + Mock::given(method("POST")).and(path("/api/dataset/register")) + .and(header("authorization", "Bearer dataset-token")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"project":{"id":"project","name":"project"},"dataset":{"id":"dataset-id","name":"cases"}}))) + .mount(&control).await; + Mock::given(method("POST")).and(path("/api/experiment/register")) + .and(header("authorization", "Bearer record-token")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"project":{"id":"record-project","name":"record-project"},"experiment":{"id":"experiment","name":"experiment"}}))) + .mount(&control).await; + Mock::given(method("GET")) + .and(path("/version")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&data) + .await; + Mock::given(method("POST")) + .and(path("/logs3")) + .respond_with(ResponseTemplate::new(204)) + .mount(&data) + .await; + let client = BraintrustClient::builder() + .runtime(braintrust_sdk_rust::current_runtime()) + .app_url(control.uri()) + .api_url(data.uri()) + .api_key("record-token") + .skip_login(!recording) + .blocking_login(recording) + .default_project("record-project") + .build() + .await + .unwrap(); + let dataset = client + .dataset_builder_with_credentials("dataset-token", "dataset-org") + .project_name("project") + .dataset_name("cases") + .build() + .unwrap(); + (client, dataset, control, data) +} +fn page(response: ResponseTemplate) -> Mock { + Mock::given(method("POST")) + .and(path("/btql")) + .and(header("authorization", "Bearer dataset-token")) + .respond_with(response) +} +#[derive(Deserialize)] +struct Input { + value: usize, +} +#[derive(Deserialize)] +struct Output { + value: usize, +} +fn options(source: DatasetCaseSource) -> EvalOpts { + EvalOpts::builder() + .name("dataset".into()) + .source(Box::new(source)) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &Input| { + Ok(Output { + value: input.value * 2, + }) + })), + vec![], + ) + .unwrap(), + ) + .quiet(true) + .build() +} + +#[tokio::test] +async fn typed_pages_pin_version_and_preserve_case_fields_without_clone_or_serialize() { + let (client, dataset, control, data) = setup(false).await; + page(ResponseTemplate::new(200).set_body_json(json!({"data":[{"id":"one","input":{"value":1},"expected":{"value":2},"metadata":{"source":true},"tags":["source"],"_xact_id":"row-version","created":"created"}],"cursor":"a"}))) + .with_priority(1).up_to_n_times(1).expect(1).mount(&data).await; + page(ResponseTemplate::new(200).set_body_json(json!({"data":[],"cursor":"b"}))) + .with_priority(2) + .up_to_n_times(1) + .expect(1) + .mount(&data) + .await; + page( + ResponseTemplate::new(200) + .set_body_json(json!({"data":[{"id":"two","input":{"value":2}}],"cursor":null})), + ) + .with_priority(3) + .expect(1) + .mount(&data) + .await; + let source = DatasetCaseSource::new(dataset, Some("snapshot".into()), Some(1)).unwrap(); + let summary = run(EvalRequest::local(options(source))).await.unwrap(); + assert_eq!( + summary.dataset.unwrap(), + DatasetVersionRef { + dataset_id: "dataset-id".into(), + version: Some("snapshot".into()) + } + ); + assert!(summary.recording.is_none()); + assert_eq!( + summary + .results + .iter() + .map(|case| case.output.as_ref().unwrap().value) + .collect::>(), + [2, 4] + ); + let case = &summary.results[0]; + assert_eq!(case.expected.as_ref().unwrap().value, 2); + assert_eq!(case.case_id.as_deref(), Some("one")); + assert_eq!(case.xact_id.as_deref(), Some("row-version")); + assert_eq!(case.created.as_deref(), Some("created")); + assert_eq!(case.metadata.as_ref().unwrap()["source"], true); + assert_eq!(case.tags.as_ref().unwrap(), &["source"]); + let queries: Vec = data + .received_requests() + .await + .unwrap() + .iter() + .filter(|request| request.url.path() == "/btql") + .map(|request| serde_json::from_slice(&request.body).unwrap()) + .collect(); + assert_eq!(queries.len(), 3); + for query in &queries { + assert_eq!(query["version"], "snapshot"); + assert_eq!(query["query"]["from"], "dataset('dataset-id')"); + assert_eq!(query["query"]["limit"], 1); + } + assert!(queries[0]["query"].get("cursor").is_none()); + assert_eq!(queries[1]["query"]["cursor"], "a"); + assert_eq!(queries[2]["query"]["cursor"], "b"); + assert!(!control + .received_requests() + .await + .unwrap() + .iter() + .any(|request| request.url.path() == "/btql")); + client.shutdown(Duration::from_secs(5)).await.unwrap(); +} + +#[tokio::test] +async fn recorded_source_admitted_before_shutdown_can_fetch_all_pages_and_finish_recording() { + let (client, dataset, _, data) = setup(true).await; + page( + ResponseTemplate::new(200) + .set_body_json(json!({"data":[{"id":"one","input":1}],"cursor":"next"})), + ) + .with_priority(1) + .up_to_n_times(1) + .mount(&data) + .await; + page( + ResponseTemplate::new(200) + .set_body_json(json!({"data":[{"id":"two","input":2}],"cursor":null})), + ) + .with_priority(2) + .mount(&data) + .await; + let source = DatasetCaseSource::::new(dataset, None, Some(1)).unwrap(); + let opts = EvalOpts::builder() + .name("recorded-dataset".into()) + .source(Box::new(source)) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &usize| Ok(*input))), + vec![], + ) + .unwrap(), + ) + .quiet(true) + .build(); + let token = run(EvalRequest::recorded(opts, client.clone())); + assert!(token.is_ok()); + let shutdown = client.shutdown(Duration::from_secs(5)); + let (result, shutdown) = tokio::join!(token, shutdown); + let summary = result.unwrap(); + shutdown.unwrap(); + assert_eq!(summary.total_cases, 2); + assert_eq!(summary.dataset.unwrap().version, None); + assert_eq!(summary.recording.unwrap().experiment_id, "experiment"); + assert!(summary.results.iter().all(|case| case.span_id.is_some())); +} + +#[tokio::test] +async fn registration_failure_is_a_typed_source_error_before_cases_are_consumed() { + let (client, dataset, control, data) = setup(false).await; + Mock::given(method("POST")) + .and(path("/api/dataset/register")) + .respond_with(ResponseTemplate::new(403).set_body_string("denied")) + .with_priority(1) + .mount(&control) + .await; + let calls = Arc::new(AtomicUsize::new(0)); + let source = DatasetCaseSource::::new(dataset, None, None).unwrap(); + let opts = EvalOpts::builder() + .name("failure".into()) + .source(Box::new(source)) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new({ + let calls = calls.clone(); + move |input: &usize| { + calls.fetch_add(1, Ordering::SeqCst); + Ok(*input) + } + })), + vec![], + ) + .unwrap(), + ) + .quiet(true) + .build(); + let error = run(EvalRequest::local(opts)).await.unwrap_err(); + let [EvalRunFailure::Source { + case_index: 0, + source, + }] = error.failures.as_slice() + else { + panic!("expected source failure") + }; + assert!(source.chain().any(|error| matches!( + error.downcast_ref::(), + Some(BraintrustError::Api { status: 403, .. }) + ))); + assert_eq!(calls.load(Ordering::SeqCst), 0); + let partial = error.partial.unwrap(); + assert_eq!(partial.total_cases, 0); + assert!(partial.dataset.is_none()); + assert!(data.received_requests().await.unwrap().is_empty()); + client.shutdown(Duration::from_secs(5)).await.unwrap(); +} + +#[tokio::test] +async fn later_page_failure_retains_completed_cases_and_snapshot_identity() { + let (client, dataset, _, data) = setup(false).await; + page( + ResponseTemplate::new(200) + .set_body_json(json!({"data":[{"id":"one","input":{"value":4}}],"cursor":"next"})), + ) + .with_priority(1) + .up_to_n_times(1) + .mount(&data) + .await; + page(ResponseTemplate::new(502).set_body_string("page failed")) + .with_priority(2) + .mount(&data) + .await; + let error = run(EvalRequest::local(options( + DatasetCaseSource::new(dataset, Some("pinned".into()), Some(1)).unwrap(), + ))) + .await + .err() + .expect("page should fail"); + assert!(matches!( + &error.failures[..], + [EvalRunFailure::Source { case_index: 1, .. }] + )); + let summary = error.partial.unwrap(); + assert_eq!(summary.dataset.unwrap().version.as_deref(), Some("pinned")); + assert_eq!(summary.results[0].output.as_ref().unwrap().value, 8); + client.shutdown(Duration::from_secs(5)).await.unwrap(); +} + +#[tokio::test] +async fn malformed_row_keeps_decode_cause_and_preceding_results() { + let (client, dataset, _, data) = setup(false).await; + page(ResponseTemplate::new(200).set_body_json(json!({"data":[{"id":"good","input":{"value":2}},{"id":"bad","input":"invalid input"}],"cursor":null}))).mount(&data).await; + let error = run(EvalRequest::local(options( + DatasetCaseSource::new(dataset, None, None).unwrap(), + ))) + .await + .err() + .expect("decode should fail"); + let [EvalRunFailure::Source { + case_index: 1, + source, + }] = error.failures.as_slice() + else { + panic!("expected row failure") + }; + assert!(source + .chain() + .any(|error| error.downcast_ref::().is_some())); + assert!(source.to_string().contains("bad input")); + assert_eq!( + error.partial.unwrap().results[0] + .output + .as_ref() + .unwrap() + .value, + 4 + ); + client.shutdown(Duration::from_secs(5)).await.unwrap(); +} + +#[tokio::test] +async fn dropping_an_unread_source_releases_accepted_work_and_closed_admission_is_immediate() { + let (client, dataset, _, data) = setup(false).await; + let source = DatasetCaseSource::::new(dataset.clone(), None, None).unwrap(); + drop(source); + client.shutdown(Duration::from_secs(5)).await.unwrap(); + assert!(data.received_requests().await.unwrap().is_empty()); + let error = DatasetCaseSource::::new(dataset, None, None) + .err() + .expect("source admission must be closed"); + assert!(error.chain().any(|error| matches!( + error.downcast_ref::(), + Some(BraintrustError::Admission(EnqueueError::Closed)) + ))); +} + +#[tokio::test] +async fn empty_live_source_reports_identity_without_claiming_a_snapshot() { + let (client, dataset, _, data) = setup(false).await; + page(ResponseTemplate::new(200).set_body_json(json!({"data":[],"cursor":null}))) + .expect(1) + .mount(&data) + .await; + let summary = run(EvalRequest::local(options( + DatasetCaseSource::new(dataset, None, None).unwrap(), + ))) + .await + .unwrap(); + assert_eq!(summary.total_cases, 0); + assert_eq!( + summary.dataset.unwrap(), + DatasetVersionRef { + dataset_id: "dataset-id".into(), + version: None + } + ); + let requests = data.received_requests().await.unwrap(); + let query: serde_json::Value = serde_json::from_slice(&requests[0].body).unwrap(); + assert!(query.get("version").is_none()); + client.shutdown(Duration::from_secs(5)).await.unwrap(); +} + +#[tokio::test] +async fn source_runtime_loss_reports_original_worker_loss_without_running_reads_elsewhere() { + let control = MockServer::start().await; + let data = MockServer::start().await; + let app_url = control.uri(); + let api_url = data.uri(); + let source = std::thread::spawn(move || { + use std::future::IntoFuture; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let client = runtime + .block_on( + BraintrustClient::builder() + .runtime(Some(runtime.handle().clone())) + .app_url(app_url) + .api_url(api_url) + .skip_login(true) + .build() + .into_future(), + ) + .unwrap(); + let dataset = client + .dataset_builder_with_credentials("token", "org") + .project_name("project") + .dataset_name("dataset") + .build() + .unwrap(); + let source = DatasetCaseSource::new(dataset, None, None).unwrap(); + drop(runtime); + source + }) + .join() + .unwrap(); + let error = run(EvalRequest::local(options(source))) + .await + .err() + .expect("source runtime was lost"); + let [EvalRunFailure::Source { + case_index: 0, + source, + }] = error.failures.as_slice() + else { + panic!("expected source failure") + }; + assert!(source.chain().any(|error| matches!( + error.downcast_ref::(), + Some(BraintrustError::Execution( + braintrust_sdk_rust::ExecutionError::WorkerLost + )) + ))); + assert!(control.received_requests().await.unwrap().is_empty()); + assert!(data.received_requests().await.unwrap().is_empty()); +} diff --git a/tests/eager_api.rs b/tests/eager_api.rs index 2e15533..b36f52c 100644 --- a/tests/eager_api.rs +++ b/tests/eager_api.rs @@ -107,6 +107,7 @@ async fn all_endpoint_families_run_unobserved_from_a_plain_thread_and_drain_afte limit: None, cursor: None, }, + version: None, })); abandon(client.experiments().register(ExperimentRegisterRequest { project_name: "p".into(),