From d788eaae2ecce69cd9905e9951d595dfd7e7bc48 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 00:18:57 +0800 Subject: [PATCH] Remove obsolete delivery settings and lazy flush aliases --- src/internal/ports.rs | 3 - src/log_queue/config.rs | 68 +------------ src/log_queue/queue.rs | 17 +--- src/logger.rs | 209 ++-------------------------------------- src/span.rs | 18 ---- src/stream.rs | 2 +- src/test_utils.rs | 9 -- 7 files changed, 10 insertions(+), 316 deletions(-) diff --git a/src/internal/ports.rs b/src/internal/ports.rs index ba9da91..2693c46 100644 --- a/src/internal/ports.rs +++ b/src/internal/ports.rs @@ -48,13 +48,10 @@ impl Drop for SpanOperation { } /// Shared delivery lifecycle, independent of resource HTTP operations. -#[async_trait] pub(crate) trait DeliveryControl: Send + Sync { fn flush(&self) -> crate::Completion; - async fn trigger_flush(&self) -> Result<()>; } -#[async_trait] pub(crate) trait SpanSubmitter: DeliveryControl { fn operation(&self) -> std::result::Result; diff --git a/src/log_queue/config.rs b/src/log_queue/config.rs index b371107..e0924aa 100644 --- a/src/log_queue/config.rs +++ b/src/log_queue/config.rs @@ -66,13 +66,6 @@ fn default_flush_chunk_size() -> usize { DEFAULT_FLUSH_CHUNK_SIZE } -fn default_sync_flush() -> bool { - matches!( - std::env::var("BRAINTRUST_SYNC_FLUSH").ok().as_deref(), - Some("1") | Some("true") | Some("yes") - ) -} - fn default_failed_publish_payloads_dir() -> Option { std::env::var("BRAINTRUST_FAILED_PUBLISH_PAYLOADS_DIR") .ok() @@ -100,7 +93,8 @@ fn default_all_publish_payloads_dir() -> Option { /// ``` #[derive(bon::Builder, Debug, Clone)] pub struct LogQueueConfig { - /// Max queue size (BRAINTRUST_QUEUE_DROP_EXCEEDING_MAXSIZE, default: 15000) + /// Maximum accepted writes, including in-flight work + /// (BRAINTRUST_QUEUE_DROP_EXCEEDING_MAXSIZE, default: 15000). #[builder(default = default_queue_max_size())] queue_max_size: usize, @@ -122,15 +116,6 @@ pub struct LogQueueConfig { #[builder(default = default_num_retries())] num_retries: usize, - /// Legacy compatibility value. Delivery is always automatic; ignored. - #[builder(default = default_sync_flush())] - sync_flush: bool, - - /// Legacy compatibility value. Flush reports covered active failures and - /// delivery remains automatic; ignored. - #[builder(default = false)] - checked_delivery: bool, - /// Chunk size for flush processing (BRAINTRUST_LOG_FLUSH_CHUNK_SIZE, default: 25). /// Rows drained from the queue are merged and sent in sequential chunks of this size. /// Each chunk's batches are sent concurrently (matching TypeScript SDK behavior). @@ -142,11 +127,6 @@ pub struct LogQueueConfig { /// Debug: save all payloads (BRAINTRUST_ALL_PUBLISH_PAYLOADS_DIR) all_publish_payloads_dir: Option, - - /// Legacy compatibility value. Admission is always bounded by queue_max_size, - /// including in-flight work; ignored. - #[builder(default = true)] - enforce_queue_size_limit: bool, } impl Default for LogQueueConfig { @@ -176,14 +156,6 @@ impl LogQueueConfig { self.num_retries } - pub fn sync_flush(&self) -> bool { - self.sync_flush - } - - pub fn checked_delivery(&self) -> bool { - self.checked_delivery - } - pub fn flush_chunk_size(&self) -> usize { self.flush_chunk_size } @@ -199,10 +171,6 @@ impl LogQueueConfig { .clone() .or_else(default_all_publish_payloads_dir) } - - pub fn enforce_queue_size_limit(&self) -> bool { - self.enforce_queue_size_limit - } } #[cfg(test)] @@ -216,16 +184,6 @@ mod tests { assert_eq!(config.batch_max_items(), DEFAULT_BATCH_MAX_ITEMS); assert_eq!(config.batch_max_bytes(), DEFAULT_BATCH_MAX_BYTES); assert_eq!(config.num_retries(), DEFAULT_NUM_RETRIES); - assert!(!config.sync_flush()); - assert!(config.enforce_queue_size_limit()); - } - - #[test] - fn test_enforce_queue_size_limit_can_be_enabled() { - let config = LogQueueConfig::builder() - .enforce_queue_size_limit(true) - .build(); - assert!(config.enforce_queue_size_limit()); } #[test] @@ -234,33 +192,11 @@ mod tests { .queue_max_size(5000) .batch_max_items(50) .num_retries(5) - .sync_flush(true) .build(); assert_eq!(config.queue_max_size(), 5000); assert_eq!(config.batch_max_items(), 50); assert_eq!(config.num_retries(), 5); - assert!(config.sync_flush()); - } - - #[test] - fn test_sync_flush_env_var_accepts_truthy_strings() { - for truthy in &["1", "true", "yes"] { - unsafe { std::env::set_var("BRAINTRUST_SYNC_FLUSH", truthy) }; - assert!( - default_sync_flush(), - "BRAINTRUST_SYNC_FLUSH={truthy} should enable sync flush" - ); - } - for falsy in &["0", "false", "no", ""] { - unsafe { std::env::set_var("BRAINTRUST_SYNC_FLUSH", falsy) }; - assert!( - !default_sync_flush(), - "BRAINTRUST_SYNC_FLUSH={falsy} should not enable sync flush" - ); - } - unsafe { std::env::remove_var("BRAINTRUST_SYNC_FLUSH") }; - assert!(!default_sync_flush(), "unset should not enable sync flush"); } #[test] diff --git a/src/log_queue/queue.rs b/src/log_queue/queue.rs index dd5e1cc..2028949 100644 --- a/src/log_queue/queue.rs +++ b/src/log_queue/queue.rs @@ -638,7 +638,7 @@ impl LogQueueCore { } } -/// Compatibility adapter connecting span preparation to the bounded write service. +/// Span preparation and submission through the bounded write service. #[derive(Clone)] pub struct LogQueue { core: Arc, @@ -654,7 +654,6 @@ impl LogQueue { pub fn new( config: LogQueueConfig, client: reqwest::Client, - _worker_queue_size: usize, execution: ExecutionBinding, ) -> Self { Self { @@ -795,25 +794,13 @@ impl LogQueue { pub(crate) fn is_started(&self) -> bool { matches!(*self.state(), QueueState::Running(_)) } - pub async fn trigger_flush_command(&self) -> Result<()> { - let token = self.flush(); - if let Err(error) = &*token { - return Err(error.clone().into()); - } - Ok(()) - } } -#[async_trait::async_trait] impl DeliveryControl for LogQueue { fn flush(&self) -> Completion { LogQueue::flush(self) } - async fn trigger_flush(&self) -> Result<()> { - self.trigger_flush_command().await - } } -#[async_trait::async_trait] impl SpanSubmitter for LogQueue { fn operation(&self) -> std::result::Result { let lease = { @@ -884,7 +871,6 @@ mod tests { LogQueue::new( LogQueueConfig::builder().num_retries(0).build(), reqwest::Client::new(), - 32, ExecutionBinding::external(tokio::runtime::Handle::current()), ) } @@ -918,7 +904,6 @@ mod tests { .num_retries(0) .build(), reqwest::Client::new(), - 8, ExecutionBinding::external(tokio::runtime::Handle::current()), ); assert!(!queue.is_started()); diff --git a/src/logger.rs b/src/logger.rs index 3b22eb2..b106e44 100644 --- a/src/logger.rs +++ b/src/logger.rs @@ -2,8 +2,6 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use serde::Serialize; -use serde_json::{Map, Value}; use tokio::sync::Notify; use url::Url; @@ -26,7 +24,6 @@ use crate::types::{ParentSpanInfo, SpanAttributes, SpanEventData, SpanObjectType // The TypeScript SDK applies no explicit timeout. const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); const LOGIN_TIMEOUT: Duration = Duration::from_secs(30); -const DEFAULT_QUEUE_SIZE: usize = 256; pub use crate::api::{DEFAULT_API_URL, DEFAULT_APP_URL}; use crate::api::login::LoginResponse; @@ -175,18 +172,14 @@ pub struct BraintrustClientBuilder { org_name: Option, default_project: Option, ca_bundle: Option, - queue_size: usize, blocking_login: bool, skip_login: bool, /// Maximum items per HTTP batch. batch_max_items: Option, /// Maximum bytes per HTTP batch. batch_max_bytes: Option, - /// Maximum queue capacity (None = unlimited). + /// Maximum accepted writes, including in-flight work (None uses environment defaults). queue_max_size: Option, - /// Whether the queue should enforce `queue_max_size`. - enforce_queue_size_limit: bool, - checked_delivery: bool, environment: Option, span_origin: Option, user_agent: Option, @@ -211,14 +204,11 @@ impl BraintrustClientBuilder { org_name: std::env::var("BRAINTRUST_ORG_NAME").ok(), default_project: std::env::var("BRAINTRUST_DEFAULT_PROJECT").ok(), ca_bundle: CaBundleConfig::from_environment(), - queue_size: DEFAULT_QUEUE_SIZE, blocking_login: false, skip_login: false, batch_max_items: None, batch_max_bytes: None, queue_max_size: None, - enforce_queue_size_limit: true, - checked_delivery: false, environment: None, span_origin: None, user_agent: None, @@ -255,29 +245,14 @@ impl BraintrustClientBuilder { self } - /// Set the maximum queue capacity. - /// When full, the newest arriving events are dropped. + /// Set the maximum accepted writes, including in-flight work. + /// When full, new writes return `EnqueueError::Full` at issuance. /// Default: 15,000 (or `BRAINTRUST_QUEUE_DROP_EXCEEDING_MAXSIZE` env var). pub fn with_queue_max_size(mut self, max_size: usize) -> Self { self.queue_max_size = Some(max_size); self } - /// Legacy compatibility setting. Delivery admission is always bounded by - /// `with_queue_max_size`, including in-flight writes. This value is ignored. - pub fn enforce_queue_size_limit(mut self, enforce: bool) -> Self { - self.enforce_queue_size_limit = enforce; - self - } - - /// Legacy compatibility setting. Delivery is always automatic and flush - /// reports failures in covered active work. This value is ignored. Await a - /// write completion to observe that write's result independently of flush. - pub fn checked_delivery(mut self, checked: bool) -> Self { - self.checked_delivery = checked; - self - } - /// Set the app URL (overrides `BRAINTRUST_APP_URL` env var). pub fn app_url(mut self, url: impl Into) -> Self { self.app_url = Some(url.into()); @@ -316,12 +291,6 @@ impl BraintrustClientBuilder { self } - /// Set the internal queue size for buffering log events. - pub fn queue_size(mut self, size: usize) -> Self { - self.queue_size = size; - self - } - /// Wait for login before initialization completes (default: false). /// /// When `false`, the build completion returns a client while login continues @@ -416,17 +385,10 @@ impl BraintrustClientBuilder { .maybe_batch_max_items(self.batch_max_items) .maybe_batch_max_bytes(self.batch_max_bytes) .maybe_queue_max_size(self.queue_max_size) - .enforce_queue_size_limit(self.enforce_queue_size_limit) - .checked_delivery(self.checked_delivery) .build(); // Commands carry their own immutable scope; the worker has no session fallback. - let queue = LogQueue::new( - log_config, - http_client.clone(), - self.queue_size, - execution.clone(), - ); + let queue = LogQueue::new(log_config, http_client.clone(), execution.clone()); let client = BraintrustClient { inner: Arc::new(ClientInner { @@ -524,7 +486,6 @@ impl SpanUpdate { struct ClientInner { lifecycle: Lifecycle, execution: ExecutionBinding, - #[allow(dead_code)] api_url: Url, app_url: Url, queue: LogQueue, @@ -1086,164 +1047,6 @@ impl BraintrustClient { pub fn flush(&self) -> crate::Completion { self.inner.queue.flush() } - - /// Trigger a non-blocking background flush. - /// Does not wait for completion - useful for streaming writes. - pub async fn trigger_flush(&self) -> Result<()> { - self.inner.queue.trigger_flush_command().await - } -} - -#[allow(dead_code)] -#[derive(Debug, Default)] -struct MutableSpanEvent { - input: Option, - output: Option, - expected: Option, - error: Option, - scores: Option>, - metadata: Option>, - metrics: Option>, - tags: Option>, - context: Option, - span_attributes: Option, - extra: HashMap, -} - -#[allow(dead_code)] -#[derive(Serialize)] -struct SerializableSpanEvent { - #[serde(skip_serializing_if = "Option::is_none")] - input: Option, - #[serde(skip_serializing_if = "Option::is_none")] - output: Option, - #[serde(skip_serializing_if = "Option::is_none")] - expected: Option, - #[serde(skip_serializing_if = "Option::is_none")] - error: Option, - #[serde(skip_serializing_if = "Option::is_none")] - scores: Option, - #[serde(skip_serializing_if = "Option::is_none")] - metadata: Option>, - #[serde(skip_serializing_if = "Option::is_none")] - metrics: Option, - #[serde(skip_serializing_if = "Option::is_none")] - tags: Option>, - #[serde(skip_serializing_if = "Option::is_none")] - context: Option, - #[serde(skip_serializing_if = "Option::is_none")] - span_attributes: Option, - #[serde(flatten)] - extra: HashMap, -} - -impl MutableSpanEvent { - fn to_event_map(&self) -> Map { - let serialized = SerializableSpanEvent { - input: self.input.clone(), - output: self.output.clone(), - expected: self.expected.clone(), - error: self.error.clone(), - scores: self.scores.as_ref().map(numeric_map_to_value), - metadata: self.metadata.clone(), - metrics: self.metrics.as_ref().map(numeric_map_to_value), - tags: self.tags.clone(), - context: self.context.clone(), - span_attributes: self - .span_attributes - .clone() - .and_then(|value| serde_json::to_value(value).ok()), - extra: self.extra.clone(), - }; - serde_json::to_value(serialized) - .ok() - .and_then(|value| value.as_object().cloned()) - .unwrap_or_default() - } - - fn apply_merged_event_map(&mut self, mut event: Map) { - self.input = event.remove("input"); - self.output = event.remove("output"); - self.expected = event.remove("expected"); - self.error = event.remove("error"); - self.scores = event.remove("scores").and_then(parse_numeric_map); - self.metadata = event.remove("metadata").and_then(|value| match value { - Value::Object(map) => Some(map), - _ => None, - }); - self.metrics = event.remove("metrics").and_then(parse_numeric_map); - self.tags = event.remove("tags").and_then(parse_string_array); - self.context = event.remove("context"); - self.span_attributes = event - .remove("span_attributes") - .and_then(|value| serde_json::from_value(value).ok()); - self.extra = HashMap::from_iter(event); - } -} - -#[allow(dead_code)] -fn numeric_map_to_value(map: &HashMap) -> Value { - Value::Object(Map::from_iter( - map.iter() - .map(|(key, value)| (key.clone(), Value::from(*value))), - )) -} - -#[allow(dead_code)] -fn apply_propagated_event( - propagated_event: &Map, - event_fields: &mut MutableSpanEvent, -) { - let mut event = event_fields.to_event_map(); - merge_event_maps(&mut event, propagated_event); - event_fields.apply_merged_event_map(event); -} - -#[allow(dead_code)] -fn merge_event_maps(merge_into: &mut Map, merge_from: &Map) { - for (key, merge_from_value) in merge_from { - match (merge_into.get_mut(key), merge_from_value) { - (Some(Value::Object(merge_into_object)), Value::Object(merge_from_object)) => { - merge_event_maps(merge_into_object, merge_from_object); - } - (Some(Value::Array(merge_into_array)), Value::Array(merge_from_array)) - if key == "tags" => - { - for item in merge_from_array { - if !merge_into_array.iter().any(|existing| existing == item) { - merge_into_array.push(item.clone()); - } - } - } - _ => { - merge_into.insert(key.clone(), merge_from_value.clone()); - } - } - } -} - -#[allow(dead_code)] -fn parse_numeric_map(value: Value) -> Option> { - match value { - Value::Object(map) => Some(HashMap::from_iter( - map.into_iter() - .filter_map(|(key, value)| value.as_f64().map(|number| (key, number))), - )), - _ => None, - } -} - -#[allow(dead_code)] -fn parse_string_array(value: Value) -> Option> { - match value { - Value::Array(values) => Some( - values - .into_iter() - .filter_map(|value| value.as_str().map(ToString::to_string)) - .collect(), - ), - _ => None, - } } #[cfg(test)] @@ -1619,13 +1422,13 @@ mod tests { } #[tokio::test] - async fn builder_capacity_is_enforced_even_with_the_legacy_unbounded_setting() { + async fn builder_capacity_is_always_bounded() { let client = BraintrustClient::builder() + .runtime(crate::current_runtime()) .app_url("https://example.com") .api_url("https://example.com") .skip_login(true) .with_queue_max_size(1) - .enforce_queue_size_limit(false) .build() .await .expect("client"); diff --git a/src/span.rs b/src/span.rs index 7d6158f..5ef0b19 100644 --- a/src/span.rs +++ b/src/span.rs @@ -5,8 +5,6 @@ use std::path::PathBuf; use std::sync::{Arc, Mutex}; use std::time::{SystemTime, UNIX_EPOCH}; -#[cfg(test)] -use async_trait::async_trait; use serde_json::{json, Map, Value}; use uuid::Uuid; @@ -678,12 +676,6 @@ impl SpanHandle { .end_time } - /// Trigger a non-blocking background flush. - /// Useful for streaming writes to get partial updates visible immediately. - pub async fn trigger_flush(&self) -> Result<()> { - self.submitter.trigger_flush().await - } - /// Export the span as SpanComponents for passing to other systems or SDKs. /// This creates a serializable representation that can be used as a parent span /// in another context (e.g., across service boundaries via HTTP headers). @@ -1118,7 +1110,6 @@ mod tests { reject: AtomicBool, accepted: Mutex>, } - #[async_trait] impl SpanSubmitter for Submitter { fn operation( &self, @@ -1154,16 +1145,12 @@ mod tests { completion } } - #[async_trait] impl crate::internal::ports::DeliveryControl for Submitter { fn flush(&self) -> crate::Completion { let (sender, completion) = crate::Completion::pair(); let _ = sender.send(Ok(crate::FlushReport::default())); completion } - async fn trigger_flush(&self) -> Result<()> { - Ok(()) - } } let submitter = Arc::new(Submitter { reject: AtomicBool::new(true), @@ -1199,18 +1186,13 @@ mod tests { async fn concurrent_ends_share_one_snapshot_and_one_delivery_result() { use crate::internal::shared_outcome::WriteResponder; struct Submitter(Mutex>); - #[async_trait] impl crate::internal::ports::DeliveryControl for Submitter { fn flush(&self) -> crate::Completion { let (sender, completion) = crate::Completion::pair(); let _ = sender.send(Ok(crate::FlushReport::default())); completion } - async fn trigger_flush(&self) -> Result<()> { - Ok(()) - } } - #[async_trait] impl SpanSubmitter for Submitter { fn operation( &self, diff --git a/src/stream.rs b/src/stream.rs index ea72c71..26a9bf1 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -1091,7 +1091,7 @@ where drop(agg); // Non-blocking flush for real-time updates - let _ = span.trigger_flush().await; + drop(span.flush()); *last_flush.lock().await = Instant::now(); } } diff --git a/src/test_utils.rs b/src/test_utils.rs index a6007c0..96fc203 100644 --- a/src/test_utils.rs +++ b/src/test_utils.rs @@ -1,7 +1,5 @@ use std::sync::{Arc, Mutex}; -use async_trait::async_trait; - use crate::error::Result; use crate::internal::session::CredentialScope; use crate::span::{SpanBuilder, SpanHandle}; @@ -58,7 +56,6 @@ impl MockBraintrustClient { } } -#[async_trait] impl SpanSubmitter for MockBraintrustClient { fn operation( &self, @@ -112,18 +109,12 @@ impl SpanSubmitter for MockBraintrustClient { } } -#[async_trait] impl DeliveryControl for MockBraintrustClient { fn flush(&self) -> crate::Completion { let (sender, completion) = crate::Completion::pair(); let _ = sender.send(Ok(crate::FlushReport::default())); completion } - - async fn trigger_flush(&self) -> Result<()> { - // Mock: no-op for tests - Ok(()) - } } pub(crate) fn mock_span_builder() -> (SpanBuilder, TestSpanCollector) {