diff --git a/Cargo.lock b/Cargo.lock index eeaea5a..466e065 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -792,15 +792,14 @@ checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" [[package]] name = "openssl" -version = "0.10.78" +version = "0.10.79" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f38c4372413cdaaf3cc79dd92d29d7d9f5ab09b51b10dded508fb90bb70b9222" +checksum = "bf0b434746ee2832f4f0baf10137e1cabb18cbe6912c69e2e33263c45250f542" dependencies = [ "bitflags", "cfg-if", "foreign-types", "libc", - "once_cell", "openssl-macros", "openssl-sys", ] @@ -824,9 +823,9 @@ checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" [[package]] name = "openssl-sys" -version = "0.9.114" +version = "0.9.115" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13ce1245cd07fcc4cfdb438f7507b0c7e4f3849a69fd84d52374c66d83741bb6" +checksum = "158fe5b292746440aa6e7a7e690e55aeb72d41505e2804c23c6973ad0e9c9781" dependencies = [ "cc", "libc", diff --git a/lago-client/examples/credit_note.rs b/lago-client/examples/credit_note.rs index 23a9c8c..59172f9 100644 --- a/lago-client/examples/credit_note.rs +++ b/lago-client/examples/credit_note.rs @@ -154,18 +154,18 @@ async fn main() -> Result<(), Box> { } // Step 6: Filter credit notes by customer - if let Some(customer) = &invoice.customer { - if let Some(ref external_id) = customer.external_id { - println!("\n=== Filtering credit notes by customer ==="); - let filter = CreditNoteFilter::new().with_external_customer_id(external_id.clone()); - let request = ListCreditNotesRequest::new().with_filters(filter); - let filtered = client.list_credit_notes(Some(request)).await?; - println!( - "Found {} credit notes for customer '{}'", - filtered.credit_notes.len(), - external_id - ); - } + if let Some(customer) = &invoice.customer + && let Some(ref external_id) = customer.external_id + { + println!("\n=== Filtering credit notes by customer ==="); + let filter = CreditNoteFilter::new().with_external_customer_id(external_id.clone()); + let request = ListCreditNotesRequest::new().with_filters(filter); + let filtered = client.list_credit_notes(Some(request)).await?; + println!( + "Found {} credit notes for customer '{}'", + filtered.credit_notes.len(), + external_id + ); } // Step 7: Filter by reason diff --git a/lago-client/examples/invoice.rs b/lago-client/examples/invoice.rs index c014149..43b809d 100644 --- a/lago-client/examples/invoice.rs +++ b/lago-client/examples/invoice.rs @@ -41,17 +41,16 @@ async fn main() -> Result<(), Box> { ); // Example 3: List invoices for the same customer - if let Some(customer) = &first_invoice.customer { - if let Some(customer_id) = &customer.external_id { - let list_customer_request = ListCustomerInvoicesRequest::new(customer_id.clone()); - let customer_invoices = - client.list_customer_invoices(list_customer_request).await?; - println!( - "\nCustomer {} has {} invoices", - customer_id, - customer_invoices.invoices.len() - ); - } + if let Some(customer) = &first_invoice.customer + && let Some(customer_id) = &customer.external_id + { + let list_customer_request = ListCustomerInvoicesRequest::new(customer_id.clone()); + let customer_invoices = client.list_customer_invoices(list_customer_request).await?; + println!( + "\nCustomer {} has {} invoices", + customer_id, + customer_invoices.invoices.len() + ); } // Example 4: Update an invoice's metadata diff --git a/lago-client/src/client.rs b/lago-client/src/client.rs index cfc1e42..4e3c6e6 100644 --- a/lago-client/src/client.rs +++ b/lago-client/src/client.rs @@ -1,5 +1,7 @@ use reqwest::{Client as HttpClient, Response}; use serde::de::DeserializeOwned; +use std::panic::AssertUnwindSafe; +use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::time::sleep; @@ -7,8 +9,12 @@ use lago_types::error::{LagoError, Result}; use crate::{Config, RetryMode}; -/// Information about rate limit headers from the API response -#[derive(Debug, Clone)] +/// Information about rate limit headers from the API response. +/// +/// Delivered to the [`RateLimitInfoCallback`] after every successful request so +/// callers can build observability around the rate limit (warn at thresholds, +/// emit metrics, etc.). +#[derive(Debug, Clone, Default)] pub struct RateLimitInfo { /// Maximum number of requests allowed in the rate limit window pub limit: Option, @@ -16,6 +22,24 @@ pub struct RateLimitInfo { pub remaining: Option, /// Number of seconds until the rate limit window resets pub reset: Option, + /// HTTP method of the call (GET, POST, ...). + pub method: String, + /// Request URL. + pub url: String, +} + +impl RateLimitInfo { + /// Returns the fraction of the rate limit currently used in `[0.0, 1.0]`, + /// or `None` when the headers aren't usable (missing limit, zero limit, + /// missing remaining). + pub fn usage_pct(&self) -> Option { + match (self.limit, self.remaining) { + (Some(limit), Some(remaining)) if limit > 0 => { + Some(1.0 - f64::from(remaining) / f64::from(limit)) + } + _ => None, + } + } } impl std::fmt::Display for RateLimitInfo { @@ -34,6 +58,14 @@ impl std::fmt::Display for RateLimitInfo { } } +/// Callback invoked after every successful response with parsed rate limit +/// headers. +/// +/// Use this to build observability around the rate limit (warn at thresholds, +/// emit metrics, etc.). Panics raised from the callback are caught and logged +/// so they cannot break the underlying request flow. +pub type RateLimitInfoCallback = Arc; + /// The main client for interacting with the Lago API /// /// This client handles HTTP requests, authentication, retries, and error handling @@ -140,15 +172,18 @@ impl LagoClient { } }; - // Parse rate limit headers before consuming the response - let rate_limit_info = if response.status().as_u16() == 429 { - Some(self.parse_rate_limit_headers(&response)) - } else { - None - }; + // Parse rate limit headers before consuming the response. Used + // both to time 429 retries and to feed the on_rate_limit_info + // callback after a successful response. + let rate_limit_info = self.parse_rate_limit_info(&response, method, url); match self.handle_response(response).await { - Ok(result) => return Ok(result), + Ok(result) => { + if let Some(info) = &rate_limit_info { + self.emit_rate_limit_info(info); + } + return Ok(result); + } Err(e) => { if !self.should_retry(&e, attempt) { return Err(e); @@ -163,6 +198,20 @@ impl LagoClient { } } + /// Invokes the configured `on_rate_limit_info` callback (if any) with + /// parsed rate limit info, catching panics so a buggy observer cannot + /// break the request flow. + fn emit_rate_limit_info(&self, info: &RateLimitInfo) { + let Some(callback) = self.config.on_rate_limit_info() else { + return; + }; + + let result = std::panic::catch_unwind(AssertUnwindSafe(|| callback(info))); + if result.is_err() { + eprintln!("lago: on_rate_limit_info callback panicked; suppressing"); + } + } + /// Determines the appropriate delay before the next retry attempt /// /// For rate limit errors (429), uses the `x-ratelimit-reset` header value @@ -189,11 +238,17 @@ impl LagoClient { delay.min(Self::MAX_RETRY_DELAY) } - /// Extracts rate limit information from response headers + /// Extracts rate limit information from response headers. /// - /// Parses the x-ratelimit-* headers that are present on every response - /// from the Lago API to provide rate limit context. - fn parse_rate_limit_headers(&self, response: &Response) -> RateLimitInfo { + /// Returns `None` when no `x-ratelimit-*` headers are present (for + /// example, on a self-hosted Lago instance that doesn't enforce limits) + /// so callers can skip emission entirely when there's nothing to report. + fn parse_rate_limit_info( + &self, + response: &Response, + method: &str, + url: &str, + ) -> Option { let limit = response .headers() .get("x-ratelimit-limit") @@ -212,11 +267,17 @@ impl LagoClient { .and_then(|h| h.to_str().ok()) .and_then(|s| s.parse::().ok()); - RateLimitInfo { + if limit.is_none() && remaining.is_none() && reset.is_none() { + return None; + } + + Some(RateLimitInfo { limit, remaining, reset, - } + method: method.to_string(), + url: url.to_string(), + }) } /// Processes the HTTP response and converts it to the expected type @@ -780,6 +841,7 @@ mod tests { limit: Some(100), remaining: Some(0), reset: Some(10), + ..Default::default() }; let delay = client.get_retry_delay(Some(&info), &LagoError::RateLimit, 1); @@ -798,6 +860,7 @@ mod tests { limit: Some(100), remaining: Some(0), reset: Some(120), + ..Default::default() }; let delay = client.get_retry_delay(Some(&info), &LagoError::RateLimit, 1); @@ -816,6 +879,7 @@ mod tests { limit: Some(100), remaining: Some(0), reset: None, + ..Default::default() }; let delay = client.get_retry_delay(Some(&info), &LagoError::RateLimit, 1); @@ -844,4 +908,163 @@ mod tests { "Should use exponential backoff for non-rate-limit errors" ); } + + // ------------------------------------------------------------------ + // RateLimitInfo + on_rate_limit_info observability + // ------------------------------------------------------------------ + + use std::sync::Mutex; + + #[test] + fn test_rate_limit_info_usage_pct() { + let info = RateLimitInfo { + limit: Some(100), + remaining: Some(20), + ..Default::default() + }; + assert_eq!(info.usage_pct(), Some(0.80)); + + let saturated = RateLimitInfo { + limit: Some(100), + remaining: Some(0), + ..Default::default() + }; + assert_eq!(saturated.usage_pct(), Some(1.0)); + } + + #[test] + fn test_rate_limit_info_usage_pct_unusable() { + // Missing limit + assert_eq!( + RateLimitInfo { + limit: None, + remaining: Some(20), + ..Default::default() + } + .usage_pct(), + None, + ); + // Missing remaining + assert_eq!( + RateLimitInfo { + limit: Some(100), + remaining: None, + ..Default::default() + } + .usage_pct(), + None, + ); + // Zero limit + assert_eq!( + RateLimitInfo { + limit: Some(0), + remaining: Some(0), + ..Default::default() + } + .usage_pct(), + None, + ); + } + + fn create_observed_client(base_url: &str, callback: RateLimitInfoCallback) -> LagoClient { + let config = Config::builder() + .credentials(Credentials::new("test-api-key".to_string())) + .region(Region::Custom(base_url.to_string())) + .timeout(Duration::from_secs(10)) + .on_rate_limit_info(callback) + .build(); + + LagoClient::new(config) + } + + #[tokio::test] + async fn test_on_rate_limit_info_fires_on_success() { + let mut server = Server::new_async().await; + let mock = server + .mock("GET", "/test") + .with_status(200) + .with_header("x-ratelimit-limit", "100") + .with_header("x-ratelimit-remaining", "20") + .with_header("x-ratelimit-reset", "5") + .with_body(json!({"id": "1", "name": "ok"}).to_string()) + .create_async() + .await; + + let captured: Arc>> = Arc::new(Mutex::new(Vec::new())); + let captured_clone = captured.clone(); + let callback: RateLimitInfoCallback = Arc::new(move |info: &RateLimitInfo| { + captured_clone.lock().unwrap().push(info.clone()); + }); + + let client = create_observed_client(&server.url(), callback); + let url = format!("{}/test", server.url()); + + let _: Result = client.make_request("GET", &url, None::<&()>).await; + + // Snapshot under the lock then drop the guard before awaiting again + // (clippy::await_holding_lock). + let snapshot: Vec = { + let guard = captured.lock().unwrap(); + guard.clone() + }; + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot[0].limit, Some(100)); + assert_eq!(snapshot[0].remaining, Some(20)); + assert_eq!(snapshot[0].reset, Some(5)); + assert_eq!(snapshot[0].method, "GET"); + assert_eq!(snapshot[0].usage_pct(), Some(0.80)); + + mock.assert_async().await; + } + + #[tokio::test] + async fn test_on_rate_limit_info_not_called_when_headers_absent() { + let mut server = Server::new_async().await; + let mock = server + .mock("GET", "/test") + .with_status(200) + .with_body(json!({"id": "1", "name": "ok"}).to_string()) + .create_async() + .await; + + let counter: Arc> = Arc::new(Mutex::new(0)); + let counter_clone = counter.clone(); + let callback: RateLimitInfoCallback = Arc::new(move |_: &RateLimitInfo| { + *counter_clone.lock().unwrap() += 1; + }); + + let client = create_observed_client(&server.url(), callback); + let url = format!("{}/test", server.url()); + + let _: Result = client.make_request("GET", &url, None::<&()>).await; + + assert_eq!(*counter.lock().unwrap(), 0); + mock.assert_async().await; + } + + #[tokio::test] + async fn test_on_rate_limit_info_panic_is_caught() { + let mut server = Server::new_async().await; + let mock = server + .mock("GET", "/test") + .with_status(200) + .with_header("x-ratelimit-limit", "100") + .with_header("x-ratelimit-remaining", "1") + .with_header("x-ratelimit-reset", "5") + .with_body(json!({"id": "1", "name": "ok"}).to_string()) + .create_async() + .await; + + let callback: RateLimitInfoCallback = Arc::new(|_: &RateLimitInfo| { + panic!("intentional"); + }); + + let client = create_observed_client(&server.url(), callback); + let url = format!("{}/test", server.url()); + + let result: Result = client.make_request("GET", &url, None::<&()>).await; + assert!(result.is_ok(), "callback panic must not break the request"); + + mock.assert_async().await; + } } diff --git a/lago-client/src/config.rs b/lago-client/src/config.rs index 4f91cf6..35da739 100644 --- a/lago-client/src/config.rs +++ b/lago-client/src/config.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use std::time::Duration; use super::{ + client::RateLimitInfoCallback, credentials::{ Credentials, CredentialsProvider, EnvironmentCredentialsProvider, StaticCredentialsProvider, }, @@ -20,6 +21,7 @@ pub struct Config { pub(crate) timeout: Duration, pub(crate) retry_config: RetryConfig, pub(crate) user_agent: String, + pub(crate) on_rate_limit_info: Option, } impl Config { @@ -70,6 +72,14 @@ impl Config { pub fn user_agent(&self) -> &str { &self.user_agent } + + /// Gets the configured `on_rate_limit_info` callback, if any. + /// + /// The callback is invoked after every successful response with parsed + /// rate limit headers. See [`crate::RateLimitInfoCallback`]. + pub fn on_rate_limit_info(&self) -> Option<&RateLimitInfoCallback> { + self.on_rate_limit_info.as_ref() + } } impl Default for Config { @@ -84,6 +94,7 @@ impl Default for Config { timeout: Duration::from_secs(30), retry_config: RetryConfig::default(), user_agent: format!("lago-rust-client/{}", env!("CARGO_PKG_VERSION")), + on_rate_limit_info: None, } } } @@ -98,6 +109,7 @@ pub struct ConfigBuilder { timeout: Option, retry_config: Option, user_agent: Option, + on_rate_limit_info: Option, } impl ConfigBuilder { @@ -112,6 +124,7 @@ impl ConfigBuilder { timeout: None, retry_config: None, user_agent: None, + on_rate_limit_info: None, } } @@ -199,6 +212,22 @@ impl ConfigBuilder { self } + /// Sets a callback to receive rate limit info after every successful + /// response. + /// + /// Use this to build observability around the rate limit (warn at + /// thresholds, emit metrics, etc.). See [`crate::RateLimitInfoCallback`]. + /// + /// # Arguments + /// * `callback` - The callback to invoke with parsed rate limit info + /// + /// # Returns + /// The builder instance for method chaining + pub fn on_rate_limit_info(mut self, callback: RateLimitInfoCallback) -> Self { + self.on_rate_limit_info = Some(callback); + self + } + /// Builds the final configuration instance /// /// Any unset values will use the defaults from `Config::default()`. @@ -218,6 +247,7 @@ impl ConfigBuilder { timeout: self.timeout.unwrap_or(default_config.timeout), retry_config: self.retry_config.unwrap_or(default_config.retry_config), user_agent: self.user_agent.unwrap_or(default_config.user_agent), + on_rate_limit_info: self.on_rate_limit_info, } } } diff --git a/lago-client/src/lib.rs b/lago-client/src/lib.rs index 84d3290..7b2ea1b 100644 --- a/lago-client/src/lib.rs +++ b/lago-client/src/lib.rs @@ -1,6 +1,7 @@ pub mod client; pub mod config; pub mod credentials; +pub mod observability; pub mod queries; pub mod region; pub mod retry; diff --git a/lago-client/src/observability.rs b/lago-client/src/observability.rs new file mode 100644 index 0000000..b80628e --- /dev/null +++ b/lago-client/src/observability.rs @@ -0,0 +1,89 @@ +//! Optional observability helpers for Lago rate limits. +//! +//! Provides a ready-to-use [`RateLimitInfoCallback`](crate::RateLimitInfoCallback) +//! so callers can opt into rate limit observability without writing their +//! own callback. + +use std::sync::Arc; + +use crate::client::{RateLimitInfo, RateLimitInfoCallback}; + +/// Default usage thresholds (80%, 90%, 95%) at which the logging observer +/// emits a warning. +pub const DEFAULT_RATE_LIMIT_THRESHOLDS: &[f64] = &[0.80, 0.90, 0.95]; + +/// Returns a [`RateLimitInfoCallback`] that logs a warning to stderr each time +/// rate limit usage crosses one of the configured thresholds. +/// +/// Pass `None` for `thresholds` to use [`DEFAULT_RATE_LIMIT_THRESHOLDS`]. +/// +/// # Example +/// +/// ```no_run +/// use lago_client::{Config, Credentials, LagoClient, Region, observability::logging_rate_limit_observer}; +/// +/// let config = Config::builder() +/// .credentials(Credentials::new("api-key".to_string())) +/// .region(Region::Us) +/// .on_rate_limit_info(logging_rate_limit_observer(None)) +/// .build(); +/// +/// let _client = LagoClient::new(config); +/// ``` +pub fn logging_rate_limit_observer(thresholds: Option<&[f64]>) -> RateLimitInfoCallback { + let mut sorted: Vec = thresholds.unwrap_or(DEFAULT_RATE_LIMIT_THRESHOLDS).to_vec(); + // Descending so we report the highest matching threshold first. + sorted.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal)); + + Arc::new(move |info: &RateLimitInfo| { + let Some(pct) = info.usage_pct() else { + return; + }; + + if sorted.iter().any(|threshold| pct >= *threshold) { + eprintln!( + "lago: rate limit at {:.0}% (limit={:?}, remaining={:?}, reset={:?}s, {} {})", + pct * 100.0, + info.limit, + info.remaining, + info.reset, + info.method, + info.url, + ); + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn info(limit: Option, remaining: Option) -> RateLimitInfo { + RateLimitInfo { + limit, + remaining, + reset: Some(10), + method: "GET".to_string(), + url: "https://x".to_string(), + } + } + + #[test] + fn observer_does_not_panic_for_valid_info() { + let observer = logging_rate_limit_observer(Some(&[0.80])); + observer(&info(Some(100), Some(4))); // 96% + } + + #[test] + fn observer_no_op_when_usage_pct_unavailable() { + let observer = logging_rate_limit_observer(None); + observer(&info(None, None)); + } + + #[test] + fn default_thresholds_used_when_none_passed() { + // Smoke test: should not panic with the default thresholds. + let observer = logging_rate_limit_observer(None); + observer(&info(Some(100), Some(50))); + } +}