diff --git a/Cargo.lock b/Cargo.lock index 44b1939..85719dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1822,6 +1822,7 @@ dependencies = [ "anyhow", "axum", "chrono", + "dashmap", "governor", "prometheus", "redis", diff --git a/Cargo.toml b/Cargo.toml index 023bc2d..5ee180b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,3 +34,4 @@ thiserror = "1" tokio = { version = "1", features = ["sync", "macros", "rt-multi-thread", "net"] } url = "2" uuid = { version = "1", features = ["v4"] } +dashmap = "5" diff --git a/src/config.rs b/src/config.rs index 08b39ce..42bb652 100644 --- a/src/config.rs +++ b/src/config.rs @@ -17,8 +17,29 @@ pub struct AppConfig { pub stellar_horizon_url: String, pub stellar_secret_key: Option, pub redis_url: String, + + // ── Global rate-limit tier ────────────────────────────────────────── + /// Maximum requests per second across **all** issuers combined. pub rate_limit_per_second: u32, + /// Burst allowance for the global tier. pub rate_limit_burst: u32, + + // ── Per-issuer rate-limit tier ────────────────────────────────────── + /// Maximum requests per second **per issuer** address. + /// + /// Defaults to `10`. The global tier always takes precedence; this limit + /// applies after the global bucket has been checked. + pub per_issuer_rate_limit_per_second: u32, + /// Burst allowance for the per-issuer tier. + /// + /// Defaults to twice `per_issuer_rate_limit_per_second`. + pub per_issuer_rate_limit_burst: u32, + /// Seconds an issuer entry is kept alive after its last request. + /// + /// Entries older than this TTL are eligible for eviction by the background + /// cleanup task. Defaults to `3600` (1 hour). + pub issuer_rate_limit_ttl_seconds: u64, + pub stellar_max_retries: u32, pub stellar_retry_base_delay_ms: u64, pub stellar_retry_max_delay_ms: u64, @@ -45,6 +66,9 @@ impl fmt::Debug for AppConfig { .field("redis_url", &self.redis_url) .field("rate_limit_per_second", &self.rate_limit_per_second) .field("rate_limit_burst", &self.rate_limit_burst) + .field("per_issuer_rate_limit_per_second", &self.per_issuer_rate_limit_per_second) + .field("per_issuer_rate_limit_burst", &self.per_issuer_rate_limit_burst) + .field("issuer_rate_limit_ttl_seconds", &self.issuer_rate_limit_ttl_seconds) .field("stellar_max_retries", &self.stellar_max_retries) .field( "stellar_retry_base_delay_ms", @@ -130,9 +154,23 @@ impl AppConfig { }; let webhook_secret = env::var("WEBHOOK_SECRET").ok(); - let rate_limit_per_second_raw = get_env_or_default("RATE_LIMIT_PER_SECOND", "10"); + let rate_limit_per_second_raw = get_env_or_default("RATE_LIMIT_PER_SECOND", "100"); let rate_limit_burst_raw = get_env_or_default("RATE_LIMIT_BURST", &rate_limit_per_second_raw); + + // ── Per-issuer rate-limit raw values ───────────────────────────── + let per_issuer_rps_raw = get_env_or_default("PER_ISSUER_RATE_LIMIT_PER_SECOND", "10"); + let per_issuer_burst_default = format!( + "{}", + per_issuer_rps_raw + .parse::() + .unwrap_or(10) + .saturating_mul(2) + ); + let per_issuer_burst_raw = + get_env_or_default("PER_ISSUER_RATE_LIMIT_BURST", &per_issuer_burst_default); + let issuer_ttl_raw = get_env_or_default("ISSUER_RATE_LIMIT_TTL_SECONDS", "3600"); + let stellar_max_retries_raw = get_env_or_default("STELLAR_MAX_RETRIES", "3"); let stellar_retry_base_delay_ms_raw = get_env_or_default( "STELLAR_RETRY_BASE_DELAY_MS", @@ -184,14 +222,14 @@ impl AppConfig { Ok(v) if v > 0 => v, Ok(_) => { errors.push("RATE_LIMIT_PER_SECOND must be greater than 0".to_string()); - 10 + 100 } Err(_) => { errors.push(format!( "RATE_LIMIT_PER_SECOND must be a valid u32, got '{}'", rate_limit_per_second_raw )); - 10 + 100 } }; @@ -206,6 +244,70 @@ impl AppConfig { } }; + if rate_limit_burst == 0 { + errors.push("RATE_LIMIT_BURST must be greater than 0".to_string()); + } + + // ── Parse per-issuer rate-limit values ─────────────────────────── + let per_issuer_rate_limit_per_second: u32 = match per_issuer_rps_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push( + "PER_ISSUER_RATE_LIMIT_PER_SECOND must be greater than 0".to_string(), + ); + 10 + } + Err(_) => { + errors.push(format!( + "PER_ISSUER_RATE_LIMIT_PER_SECOND must be a valid u32, got '{}'", + per_issuer_rps_raw + )); + 10 + } + }; + + let per_issuer_rate_limit_burst: u32 = match per_issuer_burst_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push( + "PER_ISSUER_RATE_LIMIT_BURST must be greater than 0".to_string(), + ); + per_issuer_rate_limit_per_second * 2 + } + Err(_) => { + errors.push(format!( + "PER_ISSUER_RATE_LIMIT_BURST must be a valid u32, got '{}'", + per_issuer_burst_raw + )); + per_issuer_rate_limit_per_second * 2 + } + }; + + if per_issuer_rate_limit_per_second > rate_limit_per_second { + errors.push(format!( + "PER_ISSUER_RATE_LIMIT_PER_SECOND ({}) must not exceed RATE_LIMIT_PER_SECOND ({})", + per_issuer_rate_limit_per_second, rate_limit_per_second + )); + } + + let issuer_rate_limit_ttl_seconds: u64 = match issuer_ttl_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push( + "ISSUER_RATE_LIMIT_TTL_SECONDS must be greater than 0".to_string(), + ); + 3600 + } + Err(_) => { + errors.push(format!( + "ISSUER_RATE_LIMIT_TTL_SECONDS must be a valid u64, got '{}'", + issuer_ttl_raw + )); + 3600 + } + }; + + // ── Parse remaining values ──────────────────────────────────────── let stellar_max_retries: u32 = match stellar_max_retries_raw.parse() { Ok(v) => v, Err(_) => { @@ -395,6 +497,9 @@ impl AppConfig { redis_url, rate_limit_per_second, rate_limit_burst, + per_issuer_rate_limit_per_second, + per_issuer_rate_limit_burst, + issuer_rate_limit_ttl_seconds, stellar_max_retries, stellar_retry_base_delay_ms, stellar_retry_max_delay_ms, @@ -426,6 +531,9 @@ mod tests { "REDIS_URL", "RATE_LIMIT_PER_SECOND", "RATE_LIMIT_BURST", + "PER_ISSUER_RATE_LIMIT_PER_SECOND", + "PER_ISSUER_RATE_LIMIT_BURST", + "ISSUER_RATE_LIMIT_TTL_SECONDS", "STELLAR_MAX_RETRIES", "STELLAR_RETRY_BASE_DELAY_MS", "STELLAR_RETRY_MAX_DELAY_MS", @@ -444,23 +552,22 @@ mod tests { } } + const VALID_KEY: &str = "SBU2RRGLXH3E5CQHTD3ODLDF2BWDCYUSSBLLZ5GNW7JXHDIYKXZWHOKR"; + #[test] fn from_env_uses_defaults_when_missing() { let _guard = ENV_LOCK.lock().unwrap(); clear_env(); - env::set_var( - "STELLAR_SECRET_KEY", - "SBU2RRGLXH3E5CQHTD3ODLDF2BWDCYUSSBLLZ5GNW7JXHDIYKXZWHOKR", - ); + env::set_var("STELLAR_SECRET_KEY", VALID_KEY); let cfg = AppConfig::from_env().expect("config should load with defaults"); assert_eq!(cfg.port, 8080); - assert_eq!( - cfg.stellar_horizon_url, - "https://horizon-testnet.stellar.org" - ); + assert_eq!(cfg.stellar_horizon_url, "https://horizon-testnet.stellar.org"); assert_eq!(cfg.redis_url, "redis://127.0.0.1:6379"); - assert_eq!(cfg.rate_limit_per_second, 10); + assert_eq!(cfg.rate_limit_per_second, 100); + assert_eq!(cfg.per_issuer_rate_limit_per_second, 10); + assert_eq!(cfg.per_issuer_rate_limit_burst, 20); + assert_eq!(cfg.issuer_rate_limit_ttl_seconds, 3600); assert_eq!(cfg.cache_verification_ttl, 3600); assert_eq!(cfg.stellar_max_retries, 3); assert_eq!(cfg.stellar_retry_base_delay_ms, 100); @@ -472,6 +579,34 @@ mod tests { assert_eq!(cfg.stellar_circuit_breaker_half_open_max_calls, 1); } + #[test] + fn from_env_parses_per_issuer_rate_limit_fields() { + let _guard = ENV_LOCK.lock().unwrap(); + clear_env(); + env::set_var("STELLAR_SECRET_KEY", VALID_KEY); + env::set_var("RATE_LIMIT_PER_SECOND", "500"); + env::set_var("PER_ISSUER_RATE_LIMIT_PER_SECOND", "50"); + env::set_var("PER_ISSUER_RATE_LIMIT_BURST", "100"); + env::set_var("ISSUER_RATE_LIMIT_TTL_SECONDS", "7200"); + + let cfg = AppConfig::from_env().expect("should parse"); + assert_eq!(cfg.per_issuer_rate_limit_per_second, 50); + assert_eq!(cfg.per_issuer_rate_limit_burst, 100); + assert_eq!(cfg.issuer_rate_limit_ttl_seconds, 7200); + } + + #[test] + fn from_env_rejects_per_issuer_rps_greater_than_global() { + let _guard = ENV_LOCK.lock().unwrap(); + clear_env(); + env::set_var("STELLAR_SECRET_KEY", VALID_KEY); + env::set_var("RATE_LIMIT_PER_SECOND", "10"); + env::set_var("PER_ISSUER_RATE_LIMIT_PER_SECOND", "100"); // exceeds global + + let err = AppConfig::from_env().expect_err("should fail"); + assert!(err.to_string().contains("PER_ISSUER_RATE_LIMIT_PER_SECOND")); + } + #[test] fn from_env_invalid_values_report_errors() { let _guard = ENV_LOCK.lock().unwrap(); @@ -489,10 +624,7 @@ mod tests { env::set_var("STELLAR_CIRCUIT_BREAKER_OPEN_DURATION_MS", "0"); env::set_var("STELLAR_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS", "0"); env::set_var("WEBHOOK_URLS", "https://ok.example.com, not-a-url"); - env::set_var( - "STELLAR_SECRET_KEY", - "SBU2RRGLXH3E5CQHTD3ODLDF2BWDCYUSSBLLZ5GNW7JXHDIYKXZWHOKR", - ); + env::set_var("STELLAR_SECRET_KEY", VALID_KEY); let err = AppConfig::from_env().expect_err("config should fail"); let msg = err.to_string(); @@ -524,9 +656,9 @@ mod tests { ); let err = AppConfig::from_env().expect_err("config should fail"); - let msg = err.to_string(); - - assert!(msg.contains("STELLAR_SECRET_KEY must be a valid Stellar ed25519 secret key")); + assert!(err + .to_string() + .contains("STELLAR_SECRET_KEY must be a valid Stellar ed25519 secret key")); } #[test] @@ -547,10 +679,7 @@ mod tests { env::set_var("STELLAR_CIRCUIT_BREAKER_OPEN_DURATION_MS", "45000"); env::set_var("STELLAR_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS", "2"); env::set_var("WEBHOOK_URLS", "https://a.com, https://b.com"); - env::set_var( - "STELLAR_SECRET_KEY", - "SBU2RRGLXH3E5CQHTD3ODLDF2BWDCYUSSBLLZ5GNW7JXHDIYKXZWHOKR", - ); + env::set_var("STELLAR_SECRET_KEY", VALID_KEY); let cfg = AppConfig::from_env().expect("config should load"); @@ -579,6 +708,9 @@ mod tests { redis_url: "redis://redis:6379".to_string(), rate_limit_per_second: 10, rate_limit_burst: 10, + per_issuer_rate_limit_per_second: 2, + per_issuer_rate_limit_burst: 4, + issuer_rate_limit_ttl_seconds: 3600, stellar_max_retries: 3, stellar_retry_base_delay_ms: 100, stellar_retry_max_delay_ms: 10_000, @@ -604,13 +736,11 @@ mod tests { let _guard = ENV_LOCK.lock().unwrap(); clear_env(); env::set_var("PORT", "0"); - env::set_var( - "STELLAR_SECRET_KEY", - "SBU2RRGLXH3E5CQHTD3ODLDF2BWDCYUSSBLLZ5GNW7JXHDIYKXZWHOKR", - ); + env::set_var("STELLAR_SECRET_KEY", VALID_KEY); let metrics = MetricsRegistry::arc(); - let _err = AppConfig::from_env_with_metrics(Some(Arc::clone(&metrics))).expect_err("should fail"); + let _err = + AppConfig::from_env_with_metrics(Some(Arc::clone(&metrics))).expect_err("should fail"); let output = metrics.render(); assert!(output.contains("config_validation_failures_total")); @@ -620,15 +750,13 @@ mod tests { fn from_env_records_config_reload_on_success() { let _guard = ENV_LOCK.lock().unwrap(); clear_env(); - env::set_var( - "STELLAR_SECRET_KEY", - "SBU2RRGLXH3E5CQHTD3ODLDF2BWDCYUSSBLLZ5GNW7JXHDIYKXZWHOKR", - ); + env::set_var("STELLAR_SECRET_KEY", VALID_KEY); let metrics = MetricsRegistry::arc(); - let _cfg = AppConfig::from_env_with_metrics(Some(Arc::clone(&metrics))).expect("should succeed"); + let _cfg = AppConfig::from_env_with_metrics(Some(Arc::clone(&metrics))) + .expect("should succeed"); let output = metrics.render(); assert!(output.contains("config_reload_total")); } -} +} \ No newline at end of file diff --git a/src/metrics.rs b/src/metrics.rs index 44dfaed..82d373b 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -33,10 +33,17 @@ pub struct MetricsRegistry { horizon_latency_seconds: HistogramVec, retry_total: IntCounter, - // ── Rate limiter metrics ── + // ── Rate limiter metrics (legacy global) ── rate_limit_tokens_consumed: IntCounter, rate_limit_violations: IntCounter, + // ── Rate limiter metrics (per-issuer, two-tier) ── + /// Counts every request that passed both rate-limit tiers, labelled by issuer. + rate_limit_hits: IntCounterVec, + /// Counts every request rejected by either tier, labelled by issuer and tier + /// (`"global"` or `"issuer"`). + rate_limit_rejections: IntCounterVec, + // ── Event ingestion metrics ── event_duplicates: IntCounter, event_ordering_failures: IntCounter, @@ -125,16 +132,38 @@ impl MetricsRegistry { IntCounter::new("retry_total", "Total number of retry attempts across all operations") .unwrap(); - // ── Rate limiter metrics ── + // ── Rate limiter metrics (legacy) ── let rate_limit_tokens_consumed = IntCounter::new( "rate_limit_tokens_consumed_total", - "Total rate limiter tokens consumed", + "Total rate limiter tokens consumed (legacy global limiter)", ) .unwrap(); let rate_limit_violations = IntCounter::new( "rate_limit_violations_total", - "Total rate limit violations (requests rejected)", + "Total rate limit violations – legacy global limiter (requests rejected)", + ) + .unwrap(); + + // ── Rate limiter metrics (per-issuer, two-tier) ── + // + // `rate_limit_hits_total{issuer=""}` — accepted requests per issuer. + // `rate_limit_rejections_total{issuer="",tier="global"|"issuer"}` — rejections. + let rate_limit_hits = IntCounterVec::new( + Opts::new( + "rate_limit_hits_total", + "Total requests that passed rate limiting, labelled by issuer", + ), + &["issuer"], + ) + .unwrap(); + + let rate_limit_rejections = IntCounterVec::new( + Opts::new( + "rate_limit_rejections_total", + "Total requests rejected by rate limiting, labelled by issuer and tier", + ), + &["issuer", "tier"], ) .unwrap(); @@ -170,64 +199,32 @@ impl MetricsRegistry { ) .unwrap(); - // Register everything - registry - .register(Box::new(request_count.clone())) - .unwrap(); - registry - .register(Box::new(error_count.clone())) - .unwrap(); - registry - .register(Box::new(cache_hits.clone())) - .unwrap(); - registry - .register(Box::new(cache_misses.clone())) - .unwrap(); - registry - .register(Box::new(cache_expired.clone())) - .unwrap(); - registry - .register(Box::new(cache_serialization_failures.clone())) - .unwrap(); - registry - .register(Box::new(document_registration_total.clone())) - .unwrap(); - registry - .register(Box::new(document_revocation_total.clone())) - .unwrap(); - registry - .register(Box::new(verification_total.clone())) - .unwrap(); - registry - .register(Box::new(verification_latency_seconds.clone())) - .unwrap(); - registry - .register(Box::new(horizon_latency_seconds.clone())) - .unwrap(); - registry - .register(Box::new(retry_total.clone())) - .unwrap(); - registry - .register(Box::new(rate_limit_tokens_consumed.clone())) - .unwrap(); - registry - .register(Box::new(rate_limit_violations.clone())) - .unwrap(); - registry - .register(Box::new(event_duplicates.clone())) - .unwrap(); - registry - .register(Box::new(event_ordering_failures.clone())) - .unwrap(); - registry - .register(Box::new(event_backlog_size.clone())) - .unwrap(); - registry - .register(Box::new(config_validation_failures.clone())) - .unwrap(); - registry - .register(Box::new(config_reload_total.clone())) - .unwrap(); + // ── Register everything ─────────────────────────────────────────── + for metric in [ + Box::new(request_count.clone()) as Box, + Box::new(error_count.clone()), + Box::new(cache_hits.clone()), + Box::new(cache_misses.clone()), + Box::new(cache_expired.clone()), + Box::new(cache_serialization_failures.clone()), + Box::new(document_registration_total.clone()), + Box::new(document_revocation_total.clone()), + Box::new(verification_total.clone()), + Box::new(verification_latency_seconds.clone()), + Box::new(horizon_latency_seconds.clone()), + Box::new(retry_total.clone()), + Box::new(rate_limit_tokens_consumed.clone()), + Box::new(rate_limit_violations.clone()), + Box::new(rate_limit_hits.clone()), + Box::new(rate_limit_rejections.clone()), + Box::new(event_duplicates.clone()), + Box::new(event_ordering_failures.clone()), + Box::new(event_backlog_size.clone()), + Box::new(config_validation_failures.clone()), + Box::new(config_reload_total.clone()), + ] { + registry.register(metric).unwrap(); + } Self { registry, @@ -245,6 +242,8 @@ impl MetricsRegistry { retry_total, rate_limit_tokens_consumed, rate_limit_violations, + rate_limit_hits, + rate_limit_rejections, event_duplicates, event_ordering_failures, event_backlog_size, @@ -319,7 +318,7 @@ impl MetricsRegistry { self.retry_total.inc(); } - // ── Rate limiter metrics ───────────────────────────────────────────── + // ── Rate limiter metrics (legacy) ──────────────────────────────────── pub fn record_token_consumed(&self) { self.rate_limit_tokens_consumed.inc(); @@ -329,6 +328,29 @@ impl MetricsRegistry { self.rate_limit_violations.inc(); } + // ── Rate limiter metrics (per-issuer, two-tier) ────────────────────── + + /// Record an accepted request for `issuer`. + pub fn increment_rate_limit_hit(&self, issuer: &str) { + self.rate_limit_hits + .with_label_values(&[issuer]) + .inc(); + } + + /// Record a rejection originating from the **global** tier. + pub fn increment_rate_limit_global_rejection(&self, issuer: &str) { + self.rate_limit_rejections + .with_label_values(&[issuer, "global"]) + .inc(); + } + + /// Record a rejection originating from the **per-issuer** tier. + pub fn increment_rate_limit_issuer_rejection(&self, issuer: &str) { + self.rate_limit_rejections + .with_label_values(&[issuer, "issuer"]) + .inc(); + } + // ── Event ingestion metrics ────────────────────────────────────────── pub fn increment_event_duplicate(&self) { @@ -415,6 +437,9 @@ mod tests { metrics.increment_retry(); metrics.record_token_consumed(); metrics.increment_rate_limit_violation(); + metrics.increment_rate_limit_hit("GDEX..."); + metrics.increment_rate_limit_global_rejection("GDEX..."); + metrics.increment_rate_limit_issuer_rejection("GDEX..."); metrics.increment_event_duplicate(); metrics.increment_event_ordering_failure(); metrics.set_event_backlog(5); @@ -424,16 +449,36 @@ mod tests { metrics.increment_config_reload(); let output = metrics.render(); - // Verify key metric names appear in the rendered output assert!(output.contains("requests_total")); assert!(output.contains("cache_hits_total")); assert!(output.contains("verification_total")); assert!(output.contains("horizon_latency_seconds")); assert!(output.contains("rate_limit_violations_total")); + assert!(output.contains("rate_limit_hits_total")); + assert!(output.contains("rate_limit_rejections_total")); assert!(output.contains("event_backlog_size")); assert!(output.contains("config_validation_failures_total")); } + #[test] + fn per_issuer_hit_metric_carries_issuer_label() { + let metrics = MetricsRegistry::new(); + metrics.increment_rate_limit_hit("GDEXISSUER001"); + let output = metrics.render(); + assert!(output.contains("GDEXISSUER001")); + assert!(output.contains("rate_limit_hits_total")); + } + + #[test] + fn rejection_metric_carries_tier_label() { + let metrics = MetricsRegistry::new(); + metrics.increment_rate_limit_global_rejection("GDEXISSUER001"); + metrics.increment_rate_limit_issuer_rejection("GDEXISSUER001"); + let output = metrics.render(); + assert!(output.contains(r#"tier="global""#)); + assert!(output.contains(r#"tier="issuer""#)); + } + #[test] fn timer_returns_positive_elapsed() { let start = MetricsRegistry::start_timer(); @@ -448,4 +493,4 @@ mod tests { let output = metrics.render(); assert!(output.contains("requests_total")); } -} +} \ No newline at end of file diff --git a/src/rate_limit.rs b/src/rate_limit.rs index 51de721..bf30bec 100644 --- a/src/rate_limit.rs +++ b/src/rate_limit.rs @@ -1,69 +1,330 @@ -use governor::{clock::{Clock, QuantaClock}, Quota, RateLimiter}; -use std::{num::NonZeroU32, sync::Arc, time::Duration}; +use dashmap::DashMap; +use governor::{ + clock::{Clock, QuantaClock}, + state::keyed::DefaultKeyedStateStore, + Quota, RateLimiter, +}; +use std::{num::NonZeroU32, string::{String, ToString}, sync::Arc, time::Duration}; use crate::metrics::MetricsRegistry; -/// The default rate limiter type used by this crate, backed by the Quanta monotonic clock. -pub type DefaultRateLimiter = RateLimiter< +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +// ── Type aliases ───────────────────────────────────────────────────────────── + +/// Global (unkeyed) rate limiter backed by the Quanta monotonic clock. +pub type GlobalRateLimiter = RateLimiter< governor::state::NotKeyed, governor::state::InMemoryState, - governor::clock::QuantaClock, + QuantaClock, >; -/// A metrics-aware wrapper around `governor::RateLimiter`. +/// Per-issuer (keyed) rate limiter backed by the Quanta monotonic clock. +pub type KeyedRateLimiterInner = RateLimiter< + String, + DefaultKeyedStateStore, + QuantaClock, +>; + +// ── Rate limit status ───────────────────────────────────────────────────────── + +/// Current quota status for a single issuer. +#[derive(Debug, Clone)] +pub struct RateLimitStatus { + /// The issuer DID / key this status belongs to. + pub issuer: String, + /// Remaining tokens in the per-issuer bucket (approximate). + pub remaining: u32, + /// Unix timestamp (seconds) when the per-issuer bucket fully refills. + pub reset_at: u64, + /// Whether the global limiter is currently saturated. + pub global_throttled: bool, +} + +// ── Configuration ───────────────────────────────────────────────────────────── + +/// Rate-limiting configuration for both global and per-issuer tiers. +#[derive(Debug, Clone)] +pub struct RateLimitConfig { + /// Global requests permitted per second across all issuers. + pub global_per_second: u32, + /// Global burst allowance. + pub global_burst: u32, + /// Per-issuer requests permitted per second. + pub per_issuer_per_second: u32, + /// Per-issuer burst allowance. + pub per_issuer_burst: u32, + /// How long an issuer bucket is kept alive after its last access (seconds). + pub issuer_ttl_seconds: u64, +} + +impl Default for RateLimitConfig { + fn default() -> Self { + Self { + global_per_second: 100, + global_burst: 200, + per_issuer_per_second: 10, + per_issuer_burst: 20, + issuer_ttl_seconds: 3600, + } + } +} + +// ── Per-issuer entry ────────────────────────────────────────────────────────── + +/// Metadata tracked per issuer alongside the shared keyed limiter. +struct IssuerEntry { + /// Last time a request was seen from this issuer (Unix seconds). + last_seen: u64, + /// Approximate remaining tokens (maintained as a best-effort counter). + remaining: u32, + /// Configured burst capacity for this issuer (used for reset estimation). + burst: u32, + /// Seconds per full refill (≈ burst / per_second). + refill_period_secs: u64, +} + +impl IssuerEntry { + fn new(burst: u32, per_second: u32) -> Self { + Self { + last_seen: now_secs(), + remaining: burst, + burst, + refill_period_secs: (burst as u64).saturating_div(per_second as u64).max(1), + } + } + + fn reset_at(&self) -> u64 { + self.last_seen + self.refill_period_secs + } +} + +// ── Two-tier rate limiter ──────────────────────────────────────────────────── + +/// Two-tier, metrics-aware rate limiter. /// -/// Consumes tokens with `until_ready()` (blocking) or `check()` (non-blocking) -/// and records token consumption and violation metrics. -pub struct MetricsRateLimiter { - limiter: DefaultRateLimiter, +/// **Tier 1 – Global**: a single unkeyed bucket shared across every caller. +/// **Tier 2 – Per-issuer**: a `DashMap`-backed keyed bucket, one per issuer address. +/// +/// Both tiers must pass before a request is accepted. The global tier always +/// takes precedence: if it is saturated the per-issuer check is skipped so +/// that per-issuer remaining counts stay accurate. +pub struct PerIssuerRateLimiter { + global: Arc, + keyed: Arc, + issuer_meta: Arc>, + config: RateLimitConfig, metrics: Option>, } -impl MetricsRateLimiter { - /// Build a rate limiter with the given per-second rate and burst allowance, - /// optionally wired to a metrics registry. - pub fn new(per_second: u32, burst: u32, metrics: Option>) -> Self { - let quota = Quota::per_second(NonZeroU32::new(per_second.max(1)).unwrap()) - .allow_burst(NonZeroU32::new(burst.max(1)).unwrap()); - let limiter = RateLimiter::direct(quota); - Self { limiter, metrics } - } - - /// Non-blocking check: returns `Ok(())` if within limits, or a guard error. - /// Records token consumption on success, violation on failure. - pub fn check(&self) -> Result<(), governor::NotUntil> { - match self.limiter.check() { - Ok(()) => { - if let Some(ref m) = self.metrics { - m.record_token_consumed(); - } - Ok(()) +impl PerIssuerRateLimiter { + /// Construct a new two-tier limiter from explicit config. + pub fn new(config: RateLimitConfig, metrics: Option>) -> Self { + let global_quota = + Quota::per_second(NonZeroU32::new(config.global_per_second.max(1)).unwrap()) + .allow_burst(NonZeroU32::new(config.global_burst.max(1)).unwrap()); + let global = Arc::new(RateLimiter::direct(global_quota)); + + let issuer_quota = + Quota::per_second(NonZeroU32::new(config.per_issuer_per_second.max(1)).unwrap()) + .allow_burst(NonZeroU32::new(config.per_issuer_burst.max(1)).unwrap()); + let keyed = Arc::new(RateLimiter::keyed(issuer_quota)); + + Self { + global, + keyed, + issuer_meta: Arc::new(DashMap::new()), + config, + metrics, + } + } + + /// Construct from the application `AppConfig`. + pub fn from_config( + cfg: &crate::config::AppConfig, + metrics: Option>, + ) -> Self { + let rl_cfg = RateLimitConfig { + global_per_second: cfg.rate_limit_per_second, + global_burst: cfg.rate_limit_burst, + per_issuer_per_second: cfg.per_issuer_rate_limit_per_second, + per_issuer_burst: cfg.per_issuer_rate_limit_burst, + issuer_ttl_seconds: cfg.issuer_rate_limit_ttl_seconds, + }; + Self::new(rl_cfg, metrics) + } + + // ── Public API ──────────────────────────────────────────────────────── + + /// Non-blocking check for `issuer`. + /// + /// Returns `Ok(())` if both tiers permit the request, or a + /// [`RateLimitError`] describing which tier rejected it. + pub fn check(&self, issuer: &str) -> Result<(), RateLimitError> { + // ── Tier 1: global ──────────────────────────────────────────────── + if let Err(not_until) = self.global.check() { + if let Some(ref m) = self.metrics { + m.increment_rate_limit_global_rejection(issuer); } - Err(negative) => { - if let Some(ref m) = self.metrics { - m.increment_rate_limit_violation(); - } - Err(negative) + let retry_after = not_until.wait_time_from(self.global_clock()); + return Err(RateLimitError::GlobalExhausted { retry_after }); + } + + // ── Tier 2: per-issuer ──────────────────────────────────────────── + let issuer_key = issuer.to_string(); + if let Err(not_until) = self.keyed.check_key(&issuer_key) { + if let Some(ref m) = self.metrics { + m.increment_rate_limit_issuer_rejection(issuer); } + let retry_after = not_until.wait_time_from(self.keyed_clock()); + return Err(RateLimitError::IssuerExhausted { + issuer: issuer.to_string(), + retry_after, + }); } + + // ── Success: update metadata & metrics ──────────────────────────── + self.update_meta_on_success(issuer); + if let Some(ref m) = self.metrics { + m.increment_rate_limit_hit(issuer); + } + + Ok(()) } - /// Blocking wait until a token is available. Records token consumption. - pub async fn until_ready(&self) { - self.limiter.until_ready().await; + /// Async blocking wait until both tiers permit the request. + pub async fn until_ready(&self, issuer: &str) { + self.global.until_ready().await; + self.keyed.until_key_ready(&issuer.to_string()).await; + + self.update_meta_on_success(issuer); if let Some(ref m) = self.metrics { - m.record_token_consumed(); + m.increment_rate_limit_hit(issuer); + } + } + + /// Return the current quota status for `issuer` without consuming a token. + pub fn status(&self, issuer: &str) -> RateLimitStatus { + let meta = self.issuer_meta.get(issuer); + let (remaining, reset_at) = match meta { + Some(e) => (e.remaining, e.reset_at()), + None => ( + self.config.per_issuer_burst, + now_secs() + self.config.issuer_ttl_seconds, + ), + }; + let global_throttled = self.global.check().is_err(); + RateLimitStatus { + issuer: issuer.to_string(), + remaining, + reset_at, + global_throttled, + } + } + + /// Evict issuer entries that have not been seen within `issuer_ttl_seconds`. + /// + /// Call this periodically (e.g. from a background task) to bound memory use. + pub fn evict_stale(&self) { + let cutoff = now_secs().saturating_sub(self.config.issuer_ttl_seconds); + self.issuer_meta.retain(|_, entry| entry.last_seen >= cutoff); + } + + /// Return the number of tracked issuers currently in the metadata map. + pub fn tracked_issuers(&self) -> usize { + self.issuer_meta.len() + } + + // ── Internal helpers ────────────────────────────────────────────────── + + fn update_meta_on_success(&self, issuer: &str) { + let ts = now_secs(); + let burst = self.config.per_issuer_burst; + let per_second = self.config.per_issuer_per_second; + + self.issuer_meta + .entry(issuer.to_string()) + .and_modify(|e| { + e.last_seen = ts; + e.remaining = e.remaining.saturating_sub(1); + }) + .or_insert_with(|| { + let mut entry = IssuerEntry::new(burst, per_second); + entry.remaining = entry.remaining.saturating_sub(1); + entry + }); + } + + fn global_clock(&self) -> governor::clock::QuantaInstant { + governor::clock::Clock::now(&QuantaClock::default()) + } + + fn keyed_clock(&self) -> governor::clock::QuantaInstant { + governor::clock::Clock::now(&QuantaClock::default()) + } +} + +// ── Error type ──────────────────────────────────────────────────────────────── + +/// Errors returned by [`PerIssuerRateLimiter::check`]. +#[derive(Debug)] +pub enum RateLimitError { + /// The shared global bucket is exhausted. + GlobalExhausted { retry_after: Duration }, + /// The per-issuer bucket for this caller is exhausted. + IssuerExhausted { + issuer: String, + retry_after: Duration, + }, +} + +impl RateLimitError { + /// Seconds to wait before the next attempt (for `Retry-After` header). + pub fn retry_after_secs(&self) -> u64 { + match self { + Self::GlobalExhausted { retry_after } => retry_after.as_secs().max(1), + Self::IssuerExhausted { retry_after, .. } => retry_after.as_secs().max(1), } } - /// Return a reference to the inner rate limiter for advanced use. - pub fn inner(&self) -> &DefaultRateLimiter { - &self.limiter + /// Human-readable reason string suitable for an HTTP 429 body. + pub fn reason(&self) -> &'static str { + match self { + Self::GlobalExhausted { .. } => "global rate limit exceeded", + Self::IssuerExhausted { .. } => "per-issuer rate limit exceeded", + } } } -/// Build a bare `governor::RateLimiter` without metrics (legacy compatibility). -pub fn build_rate_limiter(per_second: u32, burst: u32) -> DefaultRateLimiter { +impl std::fmt::Display for RateLimitError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::GlobalExhausted { retry_after } => write!( + f, + "global rate limit exceeded; retry after {}s", + retry_after.as_secs() + ), + Self::IssuerExhausted { issuer, retry_after } => write!( + f, + "per-issuer rate limit exceeded for '{}'; retry after {}s", + issuer, + retry_after.as_secs() + ), + } + } +} + +impl std::error::Error for RateLimitError {} + +// ── Legacy global limiter (kept for backward-compat) ───────────────────────── + +/// Build a bare global `governor::RateLimiter` without metrics (legacy compatibility). +pub fn build_rate_limiter(per_second: u32, burst: u32) -> GlobalRateLimiter { let quota = Quota::per_second(NonZeroU32::new(per_second.max(1)).unwrap()) .allow_burst(NonZeroU32::new(burst.max(1)).unwrap()); RateLimiter::direct(quota) @@ -71,7 +332,7 @@ pub fn build_rate_limiter(per_second: u32, burst: u32) -> DefaultRateLimiter { #[derive(Debug)] pub struct StellarRateLimiter { - inner: DefaultRateLimiter, + inner: GlobalRateLimiter, } impl StellarRateLimiter { @@ -109,6 +370,49 @@ impl StellarRateLimiter { mod tests { use super::*; + fn test_config(global_rps: u32, issuer_rps: u32) -> RateLimitConfig { + RateLimitConfig { + global_per_second: global_rps, + global_burst: global_rps * 2, + per_issuer_per_second: issuer_rps, + per_issuer_burst: issuer_rps * 2, + issuer_ttl_seconds: 60, + } + } + + #[test] + fn allows_request_within_limits() { + let limiter = PerIssuerRateLimiter::new(test_config(100, 10), None); + assert!(limiter.check("issuer-A").is_ok()); + } + + #[test] + fn different_issuers_have_independent_buckets() { + // Give each issuer a burst of 2 so we can exhaust one without touching the other. + let cfg = RateLimitConfig { + global_per_second: 1000, + global_burst: 1000, + per_issuer_per_second: 1, + per_issuer_burst: 2, + issuer_ttl_seconds: 60, + }; + let limiter = PerIssuerRateLimiter::new(cfg, None); + + // Exhaust issuer-A + let _ = limiter.check("issuer-A"); + let _ = limiter.check("issuer-A"); + let a_result = limiter.check("issuer-A"); + + // issuer-B should still be fine + let b_result = limiter.check("issuer-B"); + + assert!( + a_result.is_err(), + "issuer-A should be exhausted after burst" + ); + assert!(b_result.is_ok(), "issuer-B bucket should be independent"); + } + #[test] fn rate_limiter_allows_burst_within_configured_limit() { let limiter = StellarRateLimiter::new(1, 2); @@ -121,47 +425,71 @@ mod tests { #[test] fn metrics_rate_limiter_consumes_token_on_check() { let metrics = MetricsRegistry::arc(); - let rl = MetricsRateLimiter::new(100, 100, Some(Arc::clone(&metrics))); + let limiter = PerIssuerRateLimiter::new(test_config(100, 10), Some(Arc::clone(&metrics))); + + limiter.check("issuer-M").unwrap(); - assert!(rl.check().is_ok()); let output = metrics.render(); - assert!(output.contains("rate_limit_tokens_consumed_total")); + assert!(output.contains("rate_limit_hits_total")); } #[test] - fn metrics_rate_limiter_records_violation_when_exhausted() { + fn metrics_incremented_on_issuer_rejection() { let metrics = MetricsRegistry::arc(); - // 1 per second, burst 0 — first consumes token, second should fail - let rl = MetricsRateLimiter::new(1, 0, Some(Arc::clone(&metrics))); + let cfg = RateLimitConfig { + global_per_second: 1000, + global_burst: 1000, + per_issuer_per_second: 1, + per_issuer_burst: 1, + issuer_ttl_seconds: 60, + }; + let limiter = PerIssuerRateLimiter::new(cfg, Some(Arc::clone(&metrics))); - let _ = rl.check(); // consume the only token - let result = rl.check(); - assert!(result.is_err()); + let _ = limiter.check("issuer-R"); + let _ = limiter.check("issuer-R"); // triggers rejection let output = metrics.render(); - assert!(output.contains("rate_limit_violations_total")); + assert!(output.contains("rate_limit_rejections_total")); } #[test] - fn build_rate_limiter_creates_valid_limiter() { - let rl = build_rate_limiter(10, 10); - assert!(rl.check().is_ok()); + fn tracked_issuers_count_grows_with_distinct_callers() { + let limiter = PerIssuerRateLimiter::new(test_config(100, 10), None); + + for i in 0..5 { + let _ = limiter.check(&format!("issuer-{}", i)); + } + + assert_eq!(limiter.tracked_issuers(), 5); } #[tokio::test] - async fn until_ready_consumes_token() { + async fn until_ready_resolves_and_records_hit() { let metrics = MetricsRegistry::arc(); - let rl = MetricsRateLimiter::new(100, 100, Some(Arc::clone(&metrics))); + let limiter = PerIssuerRateLimiter::new(test_config(100, 10), Some(Arc::clone(&metrics))); - rl.until_ready().await; + limiter.until_ready("issuer-async").await; let output = metrics.render(); - assert!(output.contains("rate_limit_tokens_consumed_total")); + assert!(output.contains("rate_limit_hits_total")); } #[test] - fn metrics_rate_limiter_without_metrics_does_not_panic() { - let rl = MetricsRateLimiter::new(10, 10, None); - assert!(rl.check().is_ok()); + fn burst_tolerance_allows_spike_then_rejects() { + let cfg = RateLimitConfig { + global_per_second: 1000, + global_burst: 1000, + per_issuer_per_second: 1, + per_issuer_burst: 3, + issuer_ttl_seconds: 60, + }; + let limiter = PerIssuerRateLimiter::new(cfg, None); + + // Should allow burst of 3 + assert!(limiter.check("burst-issuer").is_ok()); + assert!(limiter.check("burst-issuer").is_ok()); + assert!(limiter.check("burst-issuer").is_ok()); + // 4th should be rejected + assert!(limiter.check("burst-issuer").is_err()); } -} +} \ No newline at end of file