Replace legacy bulk uploads with eager ingestion completions

README.md

@@ -180,6 +180,9 @@ and [synchronous example](https://github.com/braintrustdata/braintrust-sdk-rust/
 - `eval` separates definitions, case sources, execution, recording and reporting.
 - `api::Client` exposes direct endpoint operations with one HTTP attempt per
   call. Use it when your application owns delivery and retry policy.
+- `ingest::BatchUploader` uploads caller-owned bulk import batches with SDK
+  splitting, retries and overflow handling. It accepts `api::Config`, returns
+  eager completions from `upload_rows(rows, max_items)`, and supports `shutdown`.
 - Usage extractors support OpenAI and Anthropic responses; streaming helpers
   support provider response assembly.
 

src/api/client.rs

@@ -39,6 +39,15 @@ impl Client {
         T: Send + 'static,
         F: FnOnce(Transport) -> W,
         W: Future<Output = Result<T, Error>> + Send + 'static,
+    {
+        self.submit_operation(make)
+    }
+    pub(crate) fn submit_operation<T, E, F, W>(&self, make: F) -> Completion<T, E>
+    where
+        T: Send + 'static,
+        E: From<crate::ExecutionError> + From<crate::EnqueueError> + Send + 'static,
+        F: FnOnce(Transport) -> W,
+        W: Future<Output = Result<T, E>> + Send + 'static,
     {
         self.execution
             .response_try(|| Ok(make(self.transport.clone())))

src/api/transport.rs

@@ -28,6 +28,9 @@ struct Inner {
     headers: HeaderMap,
 }
 impl Transport {
+    pub(crate) fn http_client(&self) -> reqwest::Client {
+        self.inner.http.clone()
+    }
     pub fn new(config: Config) -> Result<Self, Error> {
         let api_url = base_url(&config.api_url)?;
         let app_url = base_url(&config.app_url)?;

src/ingest.rs

@@ -0,0 +1,137 @@
+//! Reliable ingestion of caller-owned batches, separate from resource logging.
+//!
+//! Use this adapter for bulk imports whose batch membership belongs to the
+//! application. It shares SDK encoding, splitting, retry and overflow handling.
+//! `api::Client::logs3()` remains the one-attempt endpoint surface.
+//!
+//! ```no_run
+//! use braintrust_sdk_rust::{api::Config, ingest::BatchUploader};
+//! use serde_json::{Map, Value};
+//! use std::time::Duration;
+//! # async fn example() -> anyhow::Result<()> {
+//! let uploader = BatchUploader::new(Config::new("your-api-key"))?;
+//! let row = Map::from_iter([
+//!     ("id".into(), Value::String("row-id".into())),
+//!     ("dataset_id".into(), Value::String("dataset-id".into())),
+//!     ("input".into(), Value::String("hello".into())),
+//! ]);
+//! let delivery = uploader.upload_rows(vec![row], 100);
+//! assert!(delivery.is_ok()); // admission has already happened
+//! let result = delivery.await?;
+//! assert_eq!(result.rows_uploaded, 1);
+//! uploader.shutdown(Duration::from_secs(10)).await?;
+//! # Ok(())
+//! # }
+//! ```
+use crate::internal::{
+    ingest::{EncodedBatch, EncodedRow},
+    upload::{Capabilities, UploadCoordinator, UploadPolicy},
+};
+use crate::{api, BraintrustError, Completion, ShutdownError};
+use serde_json::{Map, Value};
+use std::{sync::Arc, time::Duration};
+use tokio::sync::Mutex;
+
+#[derive(Debug, Clone, Default)]
+pub struct UploadResult {
+    pub rows_uploaded: usize,
+    pub bytes_processed: usize,
+    pub requests_sent: usize,
+}
+
+/// Concrete bulk-ingestion client. Clones share a runtime, admission capacity,
+/// immutable credentials, capability discovery and shutdown.
+#[derive(Clone)]
+pub struct BatchUploader {
+    api: api::Client,
+    limits: Arc<Mutex<Option<Capabilities>>>,
+}
+impl BatchUploader {
+    /// Validate explicit configuration without login; runtime selection matches
+    /// `api::Client`: None owns a runtime, Some uses the caller's handle.
+    pub fn new(config: api::Config) -> Result<Self, api::Error> {
+        Ok(Self {
+            api: api::Client::new(config)?,
+            limits: Arc::new(Mutex::new(None)),
+        })
+    }
+    /// Submit this owned batch immediately. Await for full delivery; dropping
+    /// the observer leaves accepted work running. A terminal error belongs only
+    /// to this operation and does not prevent the next batch from proceeding.
+    pub fn upload_rows(
+        &self,
+        rows: Vec<Map<String, Value>>,
+        batch_max_num_items: usize,
+    ) -> Completion<UploadResult, BraintrustError> {
+        let cache = self.limits.clone();
+        self.api.submit_operation(move |transport| async move {
+            if rows.is_empty() {
+                return Ok(UploadResult::default());
+            }
+            let coordinator = UploadCoordinator::for_transport(transport, UploadPolicy::Standalone);
+            let limits = {
+                let mut cached = cache.lock().await;
+                match *cached {
+                    Some(limits) => limits,
+                    None => {
+                        let limits = coordinator.capabilities(6 * 1024 * 1024, None).await;
+                        *cached = Some(limits);
+                        limits
+                    }
+                }
+            };
+            let row_payloads = rows
+                .iter()
+                .map(EncodedRow::from_map)
+                .collect::<std::result::Result<Vec<_>, _>>()
+                .map_err(|error| {
+                    BraintrustError::InvalidConfig(format!(
+                        "failed to serialize logs3 row: {error}"
+                    ))
+                })?;
+            let batches = EncodedBatch::partition(
+                row_payloads,
+                batch_max_num_items.max(1),
+                (limits.max_request_size / 2).max(1),
+            );
+
+            let mut result = UploadResult::default();
+            for batch in batches {
+                let batch_result = coordinator.upload(batch, limits).await?;
+                result.bytes_processed += batch_result.bytes_processed;
+                result.requests_sent += batch_result.requests_sent;
+                result.rows_uploaded += batch_result.rows_uploaded;
+            }
+            Ok(result)
+        })
+    }
+    /// Close admission and drain accepted batch operations. Individual upload
+    /// failures stay on their tokens; this reports execution drain/timeout.
+    pub fn shutdown(&self, timeout: Duration) -> Completion<(), ShutdownError> {
+        self.api.shutdown(timeout)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[tokio::test]
+    async fn uploader_rejects_malformed_environment_ca_bundle() {
+        let _lock = crate::http::tests::ENV_LOCK.lock().await;
+        let original = std::env::var_os(crate::http::CUSTOM_CA_BUNDLE_ENV);
+        std::env::set_var(
+            crate::http::CUSTOM_CA_BUNDLE_ENV,
+            "-----BEGIN CERTIFICATE-----\nnot-base64\n-----END CERTIFICATE-----\n",
+        );
+        let result = BatchUploader::new(api::Config::new("test-key"));
+        match original {
+            Some(value) => std::env::set_var(crate::http::CUSTOM_CA_BUNDLE_ENV, value),
+            None => std::env::remove_var(crate::http::CUSTOM_CA_BUNDLE_ENV),
+        }
+        assert!(result
+            .err()
+            .expect("malformed CA bundle should fail")
+            .to_string()
+            .contains(crate::http::CUSTOM_CA_BUNDLE_ENV));
+    }
+}

src/internal/upload.rs

@@ -102,6 +102,16 @@ impl UploadCoordinator {
         })
     }
 
+    pub(crate) fn for_transport(logs: api::Transport, policy: UploadPolicy) -> Self {
+        Self {
+            objects: ObjectStoreTransport {
+                http: logs.http_client(),
+            },
+            logs,
+            policy,
+        }
+    }
+
     pub async fn capabilities(
         &self,
         fallback: usize,

src/lib.rs

@@ -10,11 +10,11 @@ pub mod eval;
 mod experiments;
 mod extractors;
 mod http;
+pub mod ingest;
 mod internal;
 mod json_merge;
 mod log_queue;
 mod logger;
-mod logs3;
 mod runtime;
 mod span;
 mod span_components;
@@ -45,7 +45,6 @@ pub use logger::{
     BraintrustClient, BraintrustClientBuilder, LoginState, OrgInfo, DEFAULT_API_URL,
     DEFAULT_APP_URL,
 };
-pub use logs3::{Logs3BatchUploader, Logs3UploadResult};
 pub use runtime::current_runtime;
 pub use span::{
     SpanBuilder, SpanHandle, SpanLog, SpanLogBuilder, SpanLogBuilderError, SpanOrigin,