Make span finalization observable and idempotent

.github/workflows/ci.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
 
 env:
   CARGO_TERM_COLOR: always

Cargo.toml

@@ -14,13 +14,11 @@ categories = ["api-bindings", "asynchronous", "development-tools"]
 anyhow = "1"
 base64 = "0.22"
 bon = "3"
-arc-swap = "1"
 async-trait = "0.1"
 backoff = { version = "0.4", features = ["tokio"] }
 bytes = "1"
 futures = "0.3"
 chrono = { version = "0.4", features = ["serde"] }
-crossbeam = "0.8"
 indexmap = "2"
 reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "stream", "rustls-tls-native-roots"] }
 serde = { version = "1", features = ["derive"] }

src/api/btql.rs

@@ -0,0 +1,70 @@
+//! BTQL dataset readback, independent of SDK ingestion.
+use super::{Client, Error, Plane};
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value};
+
+/// A record within a dataset.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[non_exhaustive]
+pub struct DatasetRecord {
+    /// Unique record ID.
+    pub id: String,
+    /// Input data for the operation.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub input: Option<Value>,
+    /// Expected output for evaluation.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub expected: Option<Value>,
+    /// Key-value metadata.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub metadata: Option<Map<String, Value>>,
+    /// String tags for filtering.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub tags: Option<Vec<String>>,
+    /// Transaction ID (for versioning).
+    #[serde(rename = "_xact_id", skip_serializing_if = "Option::is_none")]
+    pub xact_id: Option<String>,
+    /// ISO 8601 timestamp.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub created: Option<String>,
+}
+
+/// BTQL query for fetching dataset records.
+#[derive(Debug, Clone, Serialize)]
+pub struct BTQLQuery {
+    pub query: BTQLQueryInner,
+}
+
+/// Inner BTQL query structure.
+#[derive(Debug, Clone, Serialize)]
+pub struct BTQLQueryInner {
+    /// Dataset reference: dataset('id')
+    pub from: String,
+    /// Maximum records per page.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub limit: Option<usize>,
+    /// Pagination cursor token.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cursor: Option<String>,
+}
+
+/// Response from BTQL query.
+#[derive(Debug, Clone, Deserialize)]
+pub struct BTQLResponse {
+    /// Records in JSONL format.
+    #[serde(default)]
+    pub data: Vec<DatasetRecord>,
+    /// Cursor for next page (None if last page).
+    pub cursor: Option<String>,
+}
+
+#[derive(Clone)]
+pub struct Btql {
+    pub(crate) client: Client,
+}
+impl Btql {
+    /// Execute one dataset page request; no visibility polling or pagination loop.
+    pub async fn query_dataset_records(&self, query: BTQLQuery) -> Result<BTQLResponse, Error> {
+        self.client.post_json(Plane::Data, "btql", &query).await
+    }
+}

src/api/client.rs

@@ -0,0 +1,267 @@
+use std::sync::Arc;
+
+use reqwest::{
+    header::{HeaderMap, HeaderValue, AUTHORIZATION},
+    RequestBuilder, Response,
+};
+use serde::{de::DeserializeOwned, Serialize};
+use url::Url;
+
+use super::{Config, Error};
+use crate::http::{build_http_client_builder, CaBundleConfig};
+
+#[derive(Clone, Copy)]
+pub(crate) enum Plane {
+    Control,
+    Data,
+}
+
+/// Immutable HTTP client shared by lightweight endpoint views.
+#[derive(Clone)]
+pub struct Client {
+    inner: Arc<Inner>,
+}
+struct Inner {
+    http: reqwest::Client,
+    api_url: Url,
+    app_url: Url,
+    headers: HeaderMap,
+}
+impl Client {
+    pub fn new(config: Config) -> Result<Self, Error> {
+        let api_url = base_url(&config.api_url)?;
+        let app_url = base_url(&config.app_url)?;
+        if config.api_key.trim().is_empty() {
+            return Err(Error::Config("api_key must not be empty".into()));
+        }
+        let headers = scoped_headers(&config.api_key, config.org_name.as_deref())?;
+        let ca = match (config.custom_ca_bundle_path, config.custom_ca_bundle_pem) {
+            (Some(_), Some(_)) => {
+                return Err(Error::Config(
+                    "choose a CA file or PEM data, not both".into(),
+                ))
+            }
+            (Some(path), None) => Some(CaBundleConfig::File(path)),
+            (None, Some(pem)) => Some(CaBundleConfig::from_pem(pem, "API configuration")),
+            (None, None) => CaBundleConfig::from_environment(),
+        };
+        let http =
+            build_http_client_builder(config.timeout, ca.as_ref(), config.user_agent.as_deref())
+                .map_err(|err| Error::Config(err.to_string()))?
+                // Redirects would be additional HTTP requests and can change routing.
+                .redirect(reqwest::redirect::Policy::none())
+                .build()
+                .map_err(|_| Error::Config("failed to build HTTP client".into()))?;
+        Ok(Self {
+            inner: Arc::new(Inner {
+                http,
+                api_url,
+                app_url,
+                headers,
+            }),
+        })
+    }
+    /// Reuse the SDK's configured transport without copying session state into
+    /// the API layer. Credentials and endpoint routes stay immutable.
+    pub(crate) fn for_scope(
+        http: reqwest::Client,
+        api_url: &Url,
+        app_url: &Url,
+        token: &str,
+        org_name: Option<&str>,
+    ) -> Result<Self, Error> {
+        let api_url = base_url(api_url.as_str())?;
+        let app_url = base_url(app_url.as_str())?;
+        let headers = scoped_headers(token, org_name)?;
+        Ok(Self {
+            inner: Arc::new(Inner {
+                http,
+                app_url,
+                api_url,
+                headers,
+            }),
+        })
+    }
+
+    pub fn logs3(&self) -> super::logs3::Logs3 {
+        super::logs3::Logs3 {
+            client: self.clone(),
+        }
+    }
+    pub fn btql(&self) -> super::btql::Btql {
+        super::btql::Btql {
+            client: self.clone(),
+        }
+    }
+    pub fn datasets(&self) -> super::datasets::Datasets {
+        super::datasets::Datasets {
+            client: self.clone(),
+        }
+    }
+    pub fn experiments(&self) -> super::experiments::Experiments {
+        super::experiments::Experiments {
+            client: self.clone(),
+        }
+    }
+    pub fn projects(&self) -> super::projects::Projects {
+        super::projects::Projects {
+            client: self.clone(),
+        }
+    }
+    /// One login request. Organization selection and client reconfiguration are explicit.
+    pub async fn login(&self) -> Result<super::login::LoginResponse, Error> {
+        let response = self
+            .send(
+                self.request(Plane::Control, "api/apikey/login", true)?
+                    .header("content-type", "application/json"),
+                "api/apikey/login",
+            )
+            .await?;
+        decode(response, "api/apikey/login").await
+    }
+    fn request(
+        &self,
+        plane: Plane,
+        path: &'static str,
+        post: bool,
+    ) -> Result<RequestBuilder, Error> {
+        let base = if matches!(plane, Plane::Control) {
+            &self.inner.app_url
+        } else {
+            &self.inner.api_url
+        };
+        let url = base
+            .join(path)
+            .map_err(|_| Error::Config("invalid endpoint path".into()))?;
+        let request = if post {
+            self.inner.http.post(url)
+        } else {
+            self.inner.http.get(url)
+        };
+        Ok(request.headers(self.inner.headers.clone()))
+    }
+    async fn send(
+        &self,
+        request: RequestBuilder,
+        operation: &'static str,
+    ) -> Result<Response, Error> {
+        let response = request
+            .send()
+            .await
+            .map_err(|source| transport(operation, source))?;
+        if response.status().is_success() {
+            return Ok(response);
+        }
+        let status = response.status().as_u16();
+        let body = response
+            .text()
+            .await
+            .map_err(|source| transport(operation, source))?;
+        Err(Error::Http {
+            operation,
+            status,
+            body,
+        })
+    }
+    pub(crate) async fn get_json<T: DeserializeOwned>(
+        &self,
+        plane: Plane,
+        path: &'static str,
+        query: &[(&str, &str)],
+    ) -> Result<T, Error> {
+        let response = self
+            .send(self.request(plane, path, false)?.query(query), path)
+            .await?;
+        decode(response, path).await
+    }
+    pub(crate) async fn post_json<Q: Serialize + ?Sized, T: DeserializeOwned>(
+        &self,
+        plane: Plane,
+        path: &'static str,
+        body: &Q,
+    ) -> Result<T, Error> {
+        let response = self
+            .send(self.request(plane, path, true)?.json(body), path)
+            .await?;
+        decode(response, path).await
+    }
+    pub(crate) async fn post_empty<Q: Serialize + ?Sized>(
+        &self,
+        plane: Plane,
+        path: &'static str,
+        body: &Q,
+    ) -> Result<(), Error> {
+        self.send(self.request(plane, path, true)?.json(body), path)
+            .await?;
+        Ok(())
+    }
+    pub(crate) async fn post_bytes(
+        &self,
+        path: &'static str,
+        payload: Vec<u8>,
+    ) -> Result<(), Error> {
+        self.send(
+            self.request(Plane::Data, path, true)?
+                .header("content-type", "application/json")
+                .body(payload),
+            path,
+        )
+        .await?;
+        Ok(())
+    }
+}
+fn scoped_headers(token: &str, org_name: Option<&str>) -> Result<HeaderMap, Error> {
+    let mut headers = HeaderMap::new();
+    let mut token = HeaderValue::from_str(&format!("Bearer {token}"))
+        .map_err(|_| Error::Config("invalid api_key header".into()))?;
+    token.set_sensitive(true);
+    headers.insert(AUTHORIZATION, token);
+    if let Some(org) = org_name {
+        headers.insert(
+            "x-bt-org-name",
+            HeaderValue::from_str(org)
+                .map_err(|_| Error::Config("invalid organization header".into()))?,
+        );
+    }
+    Ok(headers)
+}
+
+fn base_url(value: &str) -> Result<Url, Error> {
+    let mut url = Url::parse(value).map_err(|_| Error::Config("invalid endpoint URL".into()))?;
+    if !matches!(url.scheme(), "http" | "https")
+        || url.host_str().is_none()
+        || !url.username().is_empty()
+        || url.password().is_some()
+        || url.query().is_some()
+        || url.fragment().is_some()
+    {
+        return Err(Error::Config(
+            "endpoint must be an HTTP(S) base URL without credentials, query or fragment".into(),
+        ));
+    }
+    if !url.path().ends_with('/') {
+        url.set_path(&format!("{}/", url.path()));
+    }
+    Ok(url)
+}
+fn transport(operation: &'static str, source: reqwest::Error) -> Error {
+    Error::Transport {
+        operation,
+        source: source.without_url(),
+    }
+}
+async fn decode<T: DeserializeOwned>(
+    response: Response,
+    operation: &'static str,
+) -> Result<T, Error> {
+    let status = response.status().as_u16();
+    let body = response
+        .bytes()
+        .await
+        .map_err(|source| transport(operation, source))?;
+    serde_json::from_slice(&body).map_err(|source| Error::Decode {
+        operation,
+        status,
+        source,
+    })
+}