From 8699f71945fb0bd1597a00716eb10217bd8dda5d Mon Sep 17 00:00:00 2001 From: olatechy0 <302475381+olatechy0@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:02:33 +0100 Subject: [PATCH] feat(rpc): handle rate limiting and Retry-After Detect HTTP 429 responses in HttpTransport, carry a numeric Retry-After delay on a typed DevkitError::RateLimited, and retry within a configurable cumulative cap (60s by default). Requests that would exceed the cap fail immediately with a clear error instead of sleeping unboundedly. Add mocked and local-server tests plus documentation. Closes #45 --- CHANGELOG.md | 9 ++ crates/devkit-cli/src/main.rs | 1 + crates/devkit-core/src/lib.rs | 34 +++++ crates/devkit-rpc/src/lib.rs | 205 ++++++++++++++++++++++++++++--- docs/ARCHITECTURE.md | 21 +++- docs/CLI.md | 2 +- docs/IMPLEMENTED_CAPABILITIES.md | 3 +- 7 files changed, 257 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bc7e492..bcdc5c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ All notable changes to this project are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- `devkit-rpc`: detect HTTP 429 rate-limit responses, honor numeric + `Retry-After` delays within a configurable cumulative cap (60 seconds by + default), and return a clear `DevkitError::RateLimited` when the cap would be + exceeded. + ## [0.1.0] - 2026-10-28 First tagged release. Prebuilt binaries for Linux x86_64, macOS arm64, and diff --git a/crates/devkit-cli/src/main.rs b/crates/devkit-cli/src/main.rs index fd9e51f..8a893ab 100644 --- a/crates/devkit-cli/src/main.rs +++ b/crates/devkit-cli/src/main.rs @@ -428,6 +428,7 @@ fn runtime(error: devkit_core::DevkitError) -> CliError { DevkitError::InvalidInput(_) => "invalid_input", DevkitError::Config(_) => "configuration", DevkitError::Io(_) => "io", + DevkitError::RateLimited { .. } => "rate_limited", }; CliError::new(EXIT_ERROR, category, error.to_string()) } diff --git a/crates/devkit-core/src/lib.rs b/crates/devkit-core/src/lib.rs index cfe0f3f..61e270d 100644 --- a/crates/devkit-core/src/lib.rs +++ b/crates/devkit-core/src/lib.rs @@ -5,6 +5,8 @@ //! dependency graph. It provides the shared error type, network profiles, and //! output format selection. +use std::time::Duration; + use serde::{Deserialize, Serialize}; /// The CLI binary name. @@ -31,6 +33,14 @@ pub enum DevkitError { /// An I/O operation failed. #[error("I/O error: {0}")] Io(String), + /// The RPC endpoint rate limited the request. + #[error("rate limited: {message}")] + RateLimited { + /// What happened, including any backoff decision. + message: String, + /// The delay the endpoint asked for via `Retry-After`, when provided. + retry_after: Option, + }, } impl DevkitError { @@ -43,6 +53,22 @@ impl DevkitError { pub fn invalid(cause: impl std::fmt::Display) -> Self { DevkitError::InvalidInput(cause.to_string()) } + + /// Builds a rate-limit error with an optional `Retry-After` delay. + pub fn rate_limited(message: impl Into, retry_after: Option) -> Self { + DevkitError::RateLimited { + message: message.into(), + retry_after, + } + } + + /// The `Retry-After` delay carried by a rate-limit error, if any. + pub fn retry_after(&self) -> Option { + match self { + DevkitError::RateLimited { retry_after, .. } => *retry_after, + _ => None, + } + } } /// Output format selection. @@ -170,4 +196,12 @@ mod tests { let err = DevkitError::decode("bad base64"); assert!(err.to_string().contains("bad base64")); } + + #[test] + fn rate_limit_errors_preserve_retry_after() { + let err = DevkitError::rate_limited("HTTP 429", Some(Duration::from_secs(5))); + assert!(err.to_string().contains("rate limited")); + assert_eq!(err.retry_after(), Some(Duration::from_secs(5))); + assert_eq!(DevkitError::decode("x").retry_after(), None); + } } diff --git a/crates/devkit-rpc/src/lib.rs b/crates/devkit-rpc/src/lib.rs index 63ce04b..26c4c9b 100644 --- a/crates/devkit-rpc/src/lib.rs +++ b/crates/devkit-rpc/src/lib.rs @@ -27,6 +27,12 @@ use url::Url; const MAX_RPC_RESPONSE_BYTES: usize = 4 * 1024 * 1024; const MAX_RPC_RESPONSE_DEPTH: usize = 64; +/// Fallback backoff when a rate-limit response has no usable `Retry-After`. +const DEFAULT_RATE_LIMIT_BACKOFF: Duration = Duration::from_secs(1); + +/// Default cap on the total time a single call may spend waiting out limits. +const DEFAULT_MAX_RATE_LIMIT_WAIT: Duration = Duration::from_secs(60); + /// A validated Stellar RPC endpoint. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Endpoint { @@ -128,9 +134,12 @@ impl MockTransport { /// Queues a transport failure. pub fn push_err(&self, message: impl Into) { - self.responses - .borrow_mut() - .push_back(Err(DevkitError::Io(message.into()))); + self.push_error(DevkitError::Io(message.into())); + } + + /// Queues an arbitrary error, including a mocked rate-limit response. + pub fn push_error(&self, error: DevkitError) { + self.responses.borrow_mut().push_back(Err(error)); } /// Returns the recorded `(method, params)` calls. @@ -211,6 +220,17 @@ impl Transport for HttpTransport { .send() .map_err(|error| DevkitError::Io(format!("RPC HTTP request failed: {error}")))?; let status = response.status(); + if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + let retry_after = response + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(parse_retry_after); + return Err(DevkitError::rate_limited( + "RPC endpoint returned HTTP 429 Too Many Requests", + retry_after, + )); + } let mut body = Vec::new(); response .take((MAX_RPC_RESPONSE_BYTES + 1) as u64) @@ -376,6 +396,15 @@ fn response_too_large() -> DevkitError { "RPC response exceeds the {MAX_RPC_RESPONSE_BYTES}-byte limit" )) } + +/// Parses a `Retry-After` header value in seconds. +/// +/// The HTTP-date form is not parsed; callers fall back to a small bounded +/// backoff instead of guessing. +fn parse_retry_after(value: &str) -> Option { + value.trim().parse::().ok().map(Duration::from_secs) +} + /// A typed RPC client over an injected [`Transport`]. pub struct Client { transport: T, @@ -398,8 +427,13 @@ impl Client { } /// Calls a method, retrying retryable transport errors. + /// + /// Rate-limit responses are retried only when the requested `Retry-After` + /// fits inside the policy's total wait cap. The client never sleeps past + /// that cap: it returns a clear rate-limit error instead. fn call(&self, method: &str, params: Value) -> Result { let mut attempt = 0; + let mut waited = Duration::ZERO; loop { match self.transport.call(method, params.clone()) { Ok(value) => { @@ -407,11 +441,27 @@ impl Client { return Ok(value); } Err(error) => { - if self.retry.should_retry(attempt, &error) { - attempt += 1; - continue; + if !self.retry.should_retry(attempt, &error) { + return Err(error); + } + if let Some(delay) = self.retry.rate_limit_delay(&error) { + let total = waited.saturating_add(delay); + if total > self.retry.max_rate_limit_wait { + return Err(DevkitError::rate_limited( + format!( + "RPC rate-limit backoff of {}s would exceed the {}s cap", + delay.as_secs(), + self.retry.max_rate_limit_wait.as_secs() + ), + Some(delay), + )); + } + waited = total; + if !delay.is_zero() { + std::thread::sleep(delay); + } } - return Err(error); + attempt += 1; } } } @@ -457,26 +507,55 @@ impl Client { pub struct RetryPolicy { /// Maximum number of retries (not counting the first attempt). pub max_retries: u32, + /// Maximum total time a single call may spend waiting out rate limits. + pub max_rate_limit_wait: Duration, } impl Default for RetryPolicy { fn default() -> Self { - RetryPolicy { max_retries: 2 } + RetryPolicy { + max_retries: 2, + max_rate_limit_wait: DEFAULT_MAX_RATE_LIMIT_WAIT, + } } } impl RetryPolicy { - /// Creates a policy with `max_retries`. + /// Creates a policy with `max_retries` and the default wait cap. pub fn new(max_retries: u32) -> Self { - RetryPolicy { max_retries } + RetryPolicy { + max_retries, + ..RetryPolicy::default() + } + } + + /// Sets the cap on cumulative rate-limit waiting for a single call. + #[must_use] + pub fn with_max_rate_limit_wait(mut self, max_rate_limit_wait: Duration) -> Self { + self.max_rate_limit_wait = max_rate_limit_wait; + self } /// Returns true when another attempt should be made. /// - /// Only transport-level failures are retryable. Decode and input errors are - /// deterministic and are never retried. + /// Transport failures and rate-limit responses are retryable. Decode and + /// input errors are deterministic and are never retried. pub fn should_retry(&self, attempts_made: u32, error: &DevkitError) -> bool { - attempts_made < self.max_retries && matches!(error, DevkitError::Io(_)) + attempts_made < self.max_retries + && matches!(error, DevkitError::Io(_) | DevkitError::RateLimited { .. }) + } + + /// The delay to apply before retrying a rate-limit response. + /// + /// Uses `Retry-After` when present and a small bounded fallback otherwise. + /// Returns `None` for errors that are not rate limits. + pub fn rate_limit_delay(&self, error: &DevkitError) -> Option { + match error { + DevkitError::RateLimited { retry_after, .. } => { + Some(retry_after.unwrap_or(DEFAULT_RATE_LIMIT_BACKOFF)) + } + _ => None, + } } } @@ -602,7 +681,7 @@ mod tests { } #[test] - fn retry_policy_retries_only_io_errors() { + fn retry_policy_retries_transport_errors_but_not_decode_errors() { let policy = RetryPolicy::new(2); assert!(policy.should_retry(0, &DevkitError::Io("timeout".into()))); assert!(policy.should_retry(1, &DevkitError::Io("timeout".into()))); @@ -618,4 +697,102 @@ mod tests { let client = Client::new(transport).with_retry(RetryPolicy::new(1)); assert_eq!(client.health().unwrap().status, "healthy"); } + + #[test] + fn parses_retry_after_seconds_and_ignores_http_dates() { + assert_eq!(parse_retry_after("0"), Some(Duration::ZERO)); + assert_eq!(parse_retry_after(" 120 "), Some(Duration::from_secs(120))); + assert_eq!(parse_retry_after("Wed, 21 Oct 2015 07:28:00 GMT"), None); + assert_eq!(parse_retry_after("-1"), None); + assert_eq!(parse_retry_after("later"), None); + } + + #[test] + fn http_transport_detects_429_and_carries_retry_after() { + use std::io::Write; + use std::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server"); + let address = listener.local_addr().expect("test server address"); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept request"); + let mut request = [0u8; 1024]; + let _ = stream.read(&mut request); + let body = "{}"; + let response = format!( + "HTTP/1.1 429 Too Many Requests\r\nRetry-After: 7\r\nContent-Length: {}\r\n\r\n{body}", + body.len() + ); + stream + .write_all(response.as_bytes()) + .expect("write response"); + }); + + let endpoint = Endpoint::parse(&format!("http://{address}")).expect("loopback endpoint"); + let transport = HttpTransport::with_default_timeout(endpoint).expect("transport"); + let error = transport.call("getHealth", json!({})).unwrap_err(); + assert!( + matches!( + &error, + DevkitError::RateLimited { + retry_after: Some(retry_after), + .. + } if *retry_after == Duration::from_secs(7) + ), + "unexpected error: {error}" + ); + server.join().expect("test server exits"); + } + + #[test] + fn retry_policy_retries_rate_limits_and_uses_bounded_default_backoff() { + let policy = RetryPolicy::new(2); + let limited = DevkitError::rate_limited("429", None); + assert!(policy.should_retry(0, &limited)); + assert!(policy.should_retry(1, &limited)); + assert!(!policy.should_retry(2, &limited)); + assert_eq!( + policy.rate_limit_delay(&limited), + Some(DEFAULT_RATE_LIMIT_BACKOFF) + ); + assert_eq!(policy.rate_limit_delay(&DevkitError::decode("bad")), None); + } + + #[test] + fn client_retries_a_mocked_rate_limit_response_then_succeeds() { + let transport = MockTransport::default(); + transport.push_error(DevkitError::rate_limited("HTTP 429", Some(Duration::ZERO))); + transport.push_ok(json!({"status": "healthy"})); + let client = Client::new(transport).with_retry(RetryPolicy::new(1)); + assert_eq!(client.health().unwrap().status, "healthy"); + } + + #[test] + fn retry_after_beyond_the_cap_returns_a_clear_error_without_waiting() { + let transport = MockTransport::default(); + transport.push_error(DevkitError::rate_limited( + "HTTP 429", + Some(Duration::from_secs(120)), + )); + let policy = RetryPolicy::new(2).with_max_rate_limit_wait(Duration::from_secs(30)); + let error = Client::new(transport) + .with_retry(policy) + .health() + .unwrap_err(); + assert!(matches!(error, DevkitError::RateLimited { .. })); + assert!(error.to_string().contains("cap")); + assert!(error.to_string().contains("120s")); + } + + #[test] + fn client_returns_the_rate_limit_error_when_retries_are_exhausted() { + let transport = MockTransport::default(); + transport.push_error(DevkitError::rate_limited("HTTP 429", Some(Duration::ZERO))); + let error = Client::new(transport) + .with_retry(RetryPolicy::new(0)) + .health() + .unwrap_err(); + assert!(matches!(error, DevkitError::RateLimited { .. })); + assert_eq!(error.retry_after(), Some(Duration::ZERO)); + } } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index bfef842..a049a0d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -82,8 +82,25 @@ Design rules: timeout and redirects disabled. - Typed models cover `getHealth`, `getNetwork`, and `getLatestLedger`. Other RPC methods return raw JSON until they are modeled. -- `RetryPolicy` retries retryable transport failures. It does not yet implement - exponential backoff or jitter. +- `RetryPolicy` retries transport failures and rate-limit responses. It does not + yet implement exponential backoff or jitter for transport failures. + +### Rate limits + +`HttpTransport` detects HTTP `429 Too Many Requests` and returns a structured +`DevkitError::RateLimited` instead of a generic I/O error. The error carries the +`Retry-After` delay when the endpoint supplies one in seconds. + +`Client` retries rate-limit responses only while the delay fits inside the +policy's cumulative cap (`RetryPolicy::max_rate_limit_wait`, 60 seconds by +default). A missing or non-numeric `Retry-After` falls back to a 1-second +backoff. If the requested wait — or the sum of waits for one call — would exceed +the cap, the client returns a clear rate-limit error immediately and never +sleeps past the cap. `RetryPolicy::with_max_rate_limit_wait` lowers or raises the +cap for a caller; a zero cap turns waiting off. + +The sleep is synchronous and bounded. Callers that cannot block should use a +custom transport and handle `DevkitError::RateLimited` themselves. ## Contract inspection diff --git a/docs/CLI.md b/docs/CLI.md index f907175..1423b54 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -81,7 +81,7 @@ When `--format json` is selected, errors are written to stderr as one JSON objec } ``` -The category is one of `usage`, `decode`, `encode`, `invalid_input`, `configuration`, `io`, or `internal`. Process exit codes remain unchanged (2 for usage errors and 3 for runtime errors). Command-line parse errors use a generic JSON message so supplied argument text is not echoed into logs. Successful output remains on stdout; error JSON is on stderr. +The category is one of `usage`, `decode`, `encode`, `invalid_input`, `configuration`, `io`, `rate_limited`, or `internal`. Process exit codes remain unchanged (2 for usage errors and 3 for runtime errors). Command-line parse errors use a generic JSON message so supplied argument text is not echoed into logs. Successful output remains on stdout; error JSON is on stderr. ## Output stability diff --git a/docs/IMPLEMENTED_CAPABILITIES.md b/docs/IMPLEMENTED_CAPABILITIES.md index 33d03c9..dae4059 100644 --- a/docs/IMPLEMENTED_CAPABILITIES.md +++ b/docs/IMPLEMENTED_CAPABILITIES.md @@ -52,7 +52,8 @@ constructs `HttpTransport`. - Normalized ABI schema for every `contractspecv0` entry type. - Typed models for `getLedgerEntries`, `getTransaction(s)`, `getFeeStats`, and paginated `getEvents`. -- Exponential backoff with jitter, cancellation, and `Retry-After` handling. +- Exponential backoff with jitter and cancellation for transport failures. + (`Retry-After` handling for HTTP 429 is implemented.) - Composite GitHub Action, baseline scanning, and SBOM/attestations. - Tagged release, release binaries, and crates.io publication. - Caching, persistence, and structured logging beyond static diagnostics.