mod ffi;
use std::ffi::{c_void, CStr};
use std::fmt;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::ptr;
use std::sync::Once;
use std::time::Duration;
pub type Result<T> = std::result::Result<T, Error>;
static INIT: Once = Once::new();
fn init_runtime() {
INIT.call_once(|| unsafe {
ffi::NimMain();
});
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Status {
InvalidArgument,
BufferTooSmall,
Internal,
Unknown(i32),
}
impl Status {
fn from_raw(raw: i32) -> Self {
match raw {
ffi::FB_ERR_INVALID_ARGUMENT => Self::InvalidArgument,
ffi::FB_ERR_BUFFER_TOO_SMALL => Self::BufferTooSmall,
ffi::FB_ERR_INTERNAL => Self::Internal,
other => Self::Unknown(other),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Error {
pub status: Status,
pub message: String,
}
impl Error {
fn from_status(raw: i32) -> Self {
init_runtime();
let message = unsafe {
let ptr = ffi::fb_last_error();
if ptr.is_null() {
String::new()
} else {
CStr::from_ptr(ptr).to_string_lossy().into_owned()
}
};
Self {
status: Status::from_raw(raw),
message,
}
}
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if self.message.is_empty() {
write!(f, "{:?}", self.status)
} else {
write!(f, "{:?}: {}", self.status, self.message)
}
}
}
impl std::error::Error for Error {}
fn check(status: i32) -> Result<()> {
if status == ffi::FB_OK {
Ok(())
} else {
Err(Error::from_status(status))
}
}
fn duration_to_ns(duration: Duration) -> Result<i64> {
i64::try_from(duration.as_nanos()).map_err(|_| Error {
status: Status::InvalidArgument,
message: "duration is too large for FlowBrigade C ABI".to_string(),
})
}
fn ns_to_duration(ns: i64) -> Duration {
if ns <= 0 {
Duration::ZERO
} else {
Duration::from_nanos(ns as u64)
}
}
fn write_to_string<F>(mut write: F) -> Result<String>
where
F: FnMut(*mut i8, usize, *mut usize) -> i32,
{
let mut buf = vec![0_u8; 512];
loop {
let mut out_len = 0_usize;
let status = write(buf.as_mut_ptr().cast(), buf.len(), &mut out_len);
if status == ffi::FB_OK {
buf.truncate(out_len);
return Ok(String::from_utf8_lossy(&buf).into_owned());
}
if status != ffi::FB_ERR_BUFFER_TOO_SMALL {
return Err(Error::from_status(status));
}
let next_len = out_len.max(buf.len() * 2);
buf.resize(next_len, 0);
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RateLimitDecision {
pub allowed: bool,
pub limit: i32,
pub remaining: i32,
pub retry_after: Duration,
pub reset_after: Duration,
}
impl From<ffi::FbRateLimitResult> for RateLimitDecision {
fn from(value: ffi::FbRateLimitResult) -> Self {
Self {
allowed: value.allowed != 0,
limit: value.limit,
remaining: value.remaining,
retry_after: ns_to_duration(value.retry_after_ns),
reset_after: ns_to_duration(value.reset_after_ns),
}
}
}
fn empty_rate_limit_result() -> ffi::FbRateLimitResult {
ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
}
}
impl RateLimitDecision {
fn raw(self) -> ffi::FbRateLimitResult {
ffi::FbRateLimitResult {
allowed: if self.allowed { 1 } else { 0 },
limit: self.limit,
remaining: self.remaining,
retry_after_ns: duration_to_ns(self.retry_after).unwrap_or(i64::MAX),
reset_after_ns: duration_to_ns(self.reset_after).unwrap_or(i64::MAX),
}
}
pub fn to_json(&self) -> Result<String> {
let raw = self.raw();
write_to_string(|buffer, buffer_len, out_len| unsafe {
ffi::fb_rate_limit_result_to_json(&raw, buffer, buffer_len, out_len)
})
}
pub fn to_prometheus(&self) -> Result<String> {
let raw = self.raw();
write_to_string(|buffer, buffer_len, out_len| unsafe {
ffi::fb_rate_limit_result_to_prometheus(&raw, buffer, buffer_len, out_len)
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BulkheadDecision {
pub allowed: bool,
pub capacity: i32,
pub in_use: i32,
pub remaining: i32,
}
impl From<ffi::FbBulkheadResult> for BulkheadDecision {
fn from(value: ffi::FbBulkheadResult) -> Self {
Self {
allowed: value.allowed != 0,
capacity: value.capacity,
in_use: value.in_use,
remaining: value.remaining,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BudgetDecision {
pub allowed: bool,
pub limit: i64,
pub used: i64,
pub remaining: i64,
pub cost: i64,
pub retry_after: Duration,
pub reset_after: Duration,
}
impl From<ffi::FbBudgetResult> for BudgetDecision {
fn from(value: ffi::FbBudgetResult) -> Self {
Self {
allowed: value.allowed != 0,
limit: value.limit,
used: value.used,
remaining: value.remaining,
cost: value.cost,
retry_after: ns_to_duration(value.retry_after_ns),
reset_after: ns_to_duration(value.reset_after_ns),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RetryAllowanceDecision {
pub allowed: bool,
pub limit: i64,
pub originals: i64,
pub retries: i64,
pub remaining: i64,
pub cost: i64,
pub retry_after: Duration,
pub reset_after: Duration,
}
impl From<ffi::FbRetryAllowanceResult> for RetryAllowanceDecision {
fn from(value: ffi::FbRetryAllowanceResult) -> Self {
Self {
allowed: value.allowed != 0,
limit: value.limit,
originals: value.originals,
retries: value.retries,
remaining: value.remaining,
cost: value.cost,
retry_after: ns_to_duration(value.retry_after_ns),
reset_after: ns_to_duration(value.reset_after_ns),
}
}
}
fn empty_retry_allowance_result() -> ffi::FbRetryAllowanceResult {
ffi::FbRetryAllowanceResult {
allowed: 0,
limit: 0,
originals: 0,
retries: 0,
remaining: 0,
cost: 0,
retry_after_ns: 0,
reset_after_ns: 0,
}
}
impl BudgetDecision {
fn raw(self) -> ffi::FbBudgetResult {
ffi::FbBudgetResult {
allowed: if self.allowed { 1 } else { 0 },
limit: self.limit,
used: self.used,
remaining: self.remaining,
cost: self.cost,
retry_after_ns: duration_to_ns(self.retry_after).unwrap_or(i64::MAX),
reset_after_ns: duration_to_ns(self.reset_after).unwrap_or(i64::MAX),
}
}
pub fn to_json(&self, key: &str) -> Result<String> {
let raw = self.raw();
write_to_string(|buffer, buffer_len, out_len| unsafe {
ffi::fb_budget_result_to_json(
&raw,
key.as_ptr().cast(),
key.len(),
buffer,
buffer_len,
out_len,
)
})
}
pub fn to_prometheus(&self, key: &str) -> Result<String> {
let raw = self.raw();
write_to_string(|buffer, buffer_len, out_len| unsafe {
ffi::fb_budget_result_to_prometheus(
&raw,
key.as_ptr().cast(),
key.len(),
buffer,
buffer_len,
out_len,
)
})
}
}
pub fn abi_version() -> i32 {
init_runtime();
unsafe { ffi::fb_abi_version() }
}
pub fn abi_version_string() -> String {
init_runtime();
unsafe {
let ptr = ffi::fb_abi_version_string();
if ptr.is_null() {
String::new()
} else {
CStr::from_ptr(ptr).to_string_lossy().into_owned()
}
}
}
pub fn abi_supports(feature: &str) -> Result<bool> {
init_runtime();
let mut supported = 0;
let status =
unsafe { ffi::fb_abi_supports(feature.as_ptr().cast(), feature.len(), &mut supported) };
check(status)?;
Ok(supported != 0)
}
pub fn parse_duration(input: &str) -> Result<Duration> {
init_runtime();
let mut ns = 0_i64;
let status = unsafe { ffi::fb_duration_parse(input.as_ptr().cast(), input.len(), &mut ns) };
check(status)?;
Ok(ns_to_duration(ns))
}
pub fn format_duration(duration: Duration) -> Result<String> {
init_runtime();
let ns = duration_to_ns(duration)?;
let mut buf = vec![0_u8; 128];
let mut out_len = 0_usize;
let status =
unsafe { ffi::fb_duration_format(ns, buf.as_mut_ptr().cast(), buf.len(), &mut out_len) };
check(status)?;
buf.truncate(out_len);
Ok(String::from_utf8_lossy(&buf).into_owned())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Jitter {
None,
Full,
Equal,
Decorrelated,
}
impl Jitter {
fn raw(self) -> i32 {
match self {
Self::None => ffi::FB_NO_JITTER,
Self::Full => ffi::FB_FULL_JITTER,
Self::Equal => ffi::FB_EQUAL_JITTER,
Self::Decorrelated => ffi::FB_DECORRELATED_JITTER,
}
}
}
pub struct BackoffPolicy {
handle: ffi::FbBackoffPolicy,
}
impl BackoffPolicy {
pub fn fixed(delay: Duration, jitter: Jitter) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
let status = unsafe {
ffi::fb_fixed_backoff_create(duration_to_ns(delay)?, jitter.raw(), &mut handle)
};
check(status)?;
Ok(Self { handle })
}
pub fn linear(
initial: Duration,
increment: Duration,
max_delay: Duration,
jitter: Jitter,
) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
let status = unsafe {
ffi::fb_linear_backoff_create(
duration_to_ns(initial)?,
duration_to_ns(increment)?,
duration_to_ns(max_delay)?,
jitter.raw(),
&mut handle,
)
};
check(status)?;
Ok(Self { handle })
}
pub fn exponential(
initial: Duration,
factor: f64,
max_delay: Duration,
jitter: Jitter,
) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
let status = unsafe {
ffi::fb_exp_backoff_create(
duration_to_ns(initial)?,
factor,
duration_to_ns(max_delay)?,
jitter.raw(),
&mut handle,
)
};
check(status)?;
Ok(Self { handle })
}
pub fn delay_for(&self, attempt: i32) -> Result<Duration> {
init_runtime();
let mut ns = 0_i64;
let status = unsafe { ffi::fb_backoff_delay_for(self.handle, attempt, &mut ns) };
check(status)?;
Ok(ns_to_duration(ns))
}
}
impl Drop for BackoffPolicy {
fn drop(&mut self) {
unsafe {
ffi::fb_backoff_destroy(self.handle);
}
}
}
pub struct TokenBucket {
handle: ffi::FbTokenBucket,
}
impl TokenBucket {
pub fn new(rate: i32, per: Duration, burst: i32) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
let status =
unsafe { ffi::fb_token_bucket_create(rate, duration_to_ns(per)?, burst, &mut handle) };
check(status)?;
Ok(Self { handle })
}
pub fn inspect(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
let status = unsafe { ffi::fb_token_bucket_inspect(self.handle, cost, &mut raw) };
check(status)?;
Ok(raw.into())
}
pub fn consume(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
let status = unsafe { ffi::fb_token_bucket_consume(self.handle, cost, &mut raw) };
check(status)?;
Ok(raw.into())
}
pub fn reset(&mut self) -> Result<()> {
check(unsafe { ffi::fb_token_bucket_reset(self.handle) })
}
pub fn configured_rate(&mut self) -> Result<i32> {
let mut rate = 0;
check(unsafe { ffi::fb_token_bucket_configured_rate(self.handle, &mut rate) })?;
Ok(rate)
}
pub fn configured_period(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_token_bucket_configured_period(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn configured_burst(&mut self) -> Result<i32> {
let mut burst = 0;
check(unsafe { ffi::fb_token_bucket_configured_burst(self.handle, &mut burst) })?;
Ok(burst)
}
pub fn available_tokens(&mut self) -> Result<i32> {
let mut tokens = 0;
check(unsafe { ffi::fb_token_bucket_available_tokens(self.handle, &mut tokens) })?;
Ok(tokens)
}
}
impl Drop for TokenBucket {
fn drop(&mut self) {
unsafe {
ffi::fb_token_bucket_destroy(self.handle);
}
}
}
pub struct GcraLimiter {
handle: ffi::FbGcraLimiter,
}
impl GcraLimiter {
pub fn new(rate: i32, per: Duration, burst: i32) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_gcra_create(rate, duration_to_ns(per)?, burst, &mut handle) })?;
Ok(Self { handle })
}
pub fn inspect(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = empty_rate_limit_result();
check(unsafe { ffi::fb_gcra_inspect(self.handle, cost, &mut raw) })?;
Ok(raw.into())
}
pub fn consume(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = empty_rate_limit_result();
check(unsafe { ffi::fb_gcra_consume(self.handle, cost, &mut raw) })?;
Ok(raw.into())
}
pub fn allow(&mut self, cost: i32) -> Result<bool> {
let mut allowed = 0;
check(unsafe { ffi::fb_gcra_allow(self.handle, cost, &mut allowed) })?;
Ok(allowed != 0)
}
pub fn reset(&mut self) -> Result<()> {
check(unsafe { ffi::fb_gcra_reset(self.handle) })
}
pub fn configured_rate(&mut self) -> Result<i32> {
let mut rate = 0;
check(unsafe { ffi::fb_gcra_configured_rate(self.handle, &mut rate) })?;
Ok(rate)
}
pub fn configured_period(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_gcra_configured_period(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn configured_burst(&mut self) -> Result<i32> {
let mut burst = 0;
check(unsafe { ffi::fb_gcra_configured_burst(self.handle, &mut burst) })?;
Ok(burst)
}
pub fn configured_interval(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_gcra_configured_interval(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn available_capacity(&mut self) -> Result<i32> {
let mut capacity = 0;
check(unsafe { ffi::fb_gcra_available_capacity(self.handle, &mut capacity) })?;
Ok(capacity)
}
}
impl Drop for GcraLimiter {
fn drop(&mut self) {
unsafe {
ffi::fb_gcra_destroy(self.handle);
}
}
}
pub struct FixedWindow {
handle: ffi::FbFixedWindow,
}
impl FixedWindow {
pub fn new(limit: i32, per: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
let status =
unsafe { ffi::fb_fixed_window_create(limit, duration_to_ns(per)?, &mut handle) };
check(status)?;
Ok(Self { handle })
}
pub fn inspect(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe { ffi::fb_fixed_window_inspect(self.handle, cost, &mut raw) })?;
Ok(raw.into())
}
pub fn consume(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe { ffi::fb_fixed_window_consume(self.handle, cost, &mut raw) })?;
Ok(raw.into())
}
pub fn reset(&mut self) -> Result<()> {
check(unsafe { ffi::fb_fixed_window_reset(self.handle) })
}
pub fn configured_limit(&mut self) -> Result<i32> {
let mut limit = 0;
check(unsafe { ffi::fb_fixed_window_configured_limit(self.handle, &mut limit) })?;
Ok(limit)
}
pub fn configured_period(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_fixed_window_configured_period(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn in_use(&mut self) -> Result<i32> {
let mut in_use = 0;
check(unsafe { ffi::fb_fixed_window_in_use(self.handle, &mut in_use) })?;
Ok(in_use)
}
}
impl Drop for FixedWindow {
fn drop(&mut self) {
unsafe {
ffi::fb_fixed_window_destroy(self.handle);
}
}
}
pub struct KeyedFixedWindow {
handle: ffi::FbKeyedFixedWindow,
}
impl KeyedFixedWindow {
pub fn new(limit: i32, per: Duration, max_keys: i32) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
let status = unsafe {
ffi::fb_keyed_fixed_window_create(limit, duration_to_ns(per)?, max_keys, &mut handle)
};
check(status)?;
Ok(Self { handle })
}
pub fn inspect(&mut self, key: &str, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_keyed_fixed_window_inspect(
self.handle,
key.as_ptr().cast(),
key.len(),
cost,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn consume(&mut self, key: &str, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_keyed_fixed_window_consume(
self.handle,
key.as_ptr().cast(),
key.len(),
cost,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn clear(&mut self, key: &str) -> Result<bool> {
let mut cleared = 0;
check(unsafe {
ffi::fb_keyed_fixed_window_clear(
self.handle,
key.as_ptr().cast(),
key.len(),
&mut cleared,
)
})?;
Ok(cleared != 0)
}
pub fn reset(&mut self, key: &str) -> Result<bool> {
let mut reset = 0;
check(unsafe {
ffi::fb_keyed_fixed_window_reset(
self.handle,
key.as_ptr().cast(),
key.len(),
&mut reset,
)
})?;
Ok(reset != 0)
}
pub fn reset_all(&mut self) -> Result<i32> {
let mut removed = 0;
check(unsafe { ffi::fb_keyed_fixed_window_reset_all(self.handle, &mut removed) })?;
Ok(removed)
}
pub fn active_keys(&mut self) -> Result<i32> {
let mut count = 0;
check(unsafe { ffi::fb_keyed_fixed_window_active_keys(self.handle, &mut count) })?;
Ok(count)
}
pub fn configured_limit(&mut self) -> Result<i32> {
let mut limit = 0;
check(unsafe { ffi::fb_keyed_fixed_window_configured_limit(self.handle, &mut limit) })?;
Ok(limit)
}
pub fn configured_period(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_keyed_fixed_window_configured_period(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn key_capacity(&mut self) -> Result<i32> {
let mut capacity = 0;
check(unsafe { ffi::fb_keyed_fixed_window_key_capacity(self.handle, &mut capacity) })?;
Ok(capacity)
}
}
impl Drop for KeyedFixedWindow {
fn drop(&mut self) {
unsafe {
ffi::fb_keyed_fixed_window_destroy(self.handle);
}
}
}
pub struct KeyedGcraLimiter {
handle: ffi::FbKeyedGcraLimiter,
}
impl KeyedGcraLimiter {
pub fn new(rate: i32, per: Duration, burst: i32, max_keys: i32) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe {
ffi::fb_keyed_gcra_create(rate, duration_to_ns(per)?, burst, max_keys, &mut handle)
})?;
Ok(Self { handle })
}
pub fn inspect(&mut self, key: &str, cost: i32) -> Result<RateLimitDecision> {
let mut raw = empty_rate_limit_result();
check(unsafe {
ffi::fb_keyed_gcra_inspect(self.handle, key.as_ptr().cast(), key.len(), cost, &mut raw)
})?;
Ok(raw.into())
}
pub fn consume(&mut self, key: &str, cost: i32) -> Result<RateLimitDecision> {
let mut raw = empty_rate_limit_result();
check(unsafe {
ffi::fb_keyed_gcra_consume(self.handle, key.as_ptr().cast(), key.len(), cost, &mut raw)
})?;
Ok(raw.into())
}
pub fn allow(&mut self, key: &str, cost: i32) -> Result<bool> {
let mut allowed = 0;
check(unsafe {
ffi::fb_keyed_gcra_allow(
self.handle,
key.as_ptr().cast(),
key.len(),
cost,
&mut allowed,
)
})?;
Ok(allowed != 0)
}
pub fn prune_idle(&mut self) -> Result<()> {
check(unsafe { ffi::fb_keyed_gcra_prune_idle(self.handle) })
}
pub fn active_keys(&mut self) -> Result<i32> {
let mut count = 0;
check(unsafe { ffi::fb_keyed_gcra_active_keys(self.handle, &mut count) })?;
Ok(count)
}
pub fn clear(&mut self, key: &str) -> Result<bool> {
let mut cleared = 0;
check(unsafe {
ffi::fb_keyed_gcra_clear(self.handle, key.as_ptr().cast(), key.len(), &mut cleared)
})?;
Ok(cleared != 0)
}
pub fn reset(&mut self, key: &str) -> Result<bool> {
let mut reset = 0;
check(unsafe {
ffi::fb_keyed_gcra_reset(self.handle, key.as_ptr().cast(), key.len(), &mut reset)
})?;
Ok(reset != 0)
}
pub fn reset_all(&mut self) -> Result<i32> {
let mut removed = 0;
check(unsafe { ffi::fb_keyed_gcra_reset_all(self.handle, &mut removed) })?;
Ok(removed)
}
pub fn configured_rate(&mut self) -> Result<i32> {
let mut rate = 0;
check(unsafe { ffi::fb_keyed_gcra_configured_rate(self.handle, &mut rate) })?;
Ok(rate)
}
pub fn configured_period(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_keyed_gcra_configured_period(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn configured_burst(&mut self) -> Result<i32> {
let mut burst = 0;
check(unsafe { ffi::fb_keyed_gcra_configured_burst(self.handle, &mut burst) })?;
Ok(burst)
}
pub fn configured_interval(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_keyed_gcra_configured_interval(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn key_capacity(&mut self) -> Result<i32> {
let mut capacity = 0;
check(unsafe { ffi::fb_keyed_gcra_key_capacity(self.handle, &mut capacity) })?;
Ok(capacity)
}
}
impl Drop for KeyedGcraLimiter {
fn drop(&mut self) {
unsafe {
ffi::fb_keyed_gcra_destroy(self.handle);
}
}
}
pub struct SlidingWindow {
handle: ffi::FbSlidingWindow,
}
impl SlidingWindow {
pub fn new(limit: i32, per: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_sliding_window_create(limit, duration_to_ns(per)?, &mut handle) })?;
Ok(Self { handle })
}
pub fn inspect(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe { ffi::fb_sliding_window_inspect(self.handle, cost, &mut raw) })?;
Ok(raw.into())
}
pub fn consume(&mut self, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe { ffi::fb_sliding_window_consume(self.handle, cost, &mut raw) })?;
Ok(raw.into())
}
pub fn reset(&mut self) -> Result<()> {
check(unsafe { ffi::fb_sliding_window_reset(self.handle) })
}
pub fn configured_limit(&mut self) -> Result<i32> {
let mut limit = 0;
check(unsafe { ffi::fb_sliding_window_configured_limit(self.handle, &mut limit) })?;
Ok(limit)
}
pub fn configured_period(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_sliding_window_configured_period(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn current_use(&mut self) -> Result<i32> {
let mut in_use = 0;
check(unsafe { ffi::fb_sliding_window_current_use(self.handle, &mut in_use) })?;
Ok(in_use)
}
}
impl Drop for SlidingWindow {
fn drop(&mut self) {
unsafe {
ffi::fb_sliding_window_destroy(self.handle);
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CircuitState {
Closed,
Open,
HalfOpen,
Unknown(i32),
}
impl CircuitState {
fn from_raw(raw: i32) -> Self {
match raw {
0 => Self::Closed,
1 => Self::Open,
2 => Self::HalfOpen,
other => Self::Unknown(other),
}
}
}
pub struct CircuitBreaker {
handle: ffi::FbCircuitBreaker,
}
impl CircuitBreaker {
pub fn new(failure_threshold: i32, reset_after: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe {
ffi::fb_circuit_breaker_create(
failure_threshold,
duration_to_ns(reset_after)?,
&mut handle,
)
})?;
Ok(Self { handle })
}
pub fn allow(&mut self) -> Result<bool> {
let mut allowed = 0;
check(unsafe { ffi::fb_circuit_breaker_allow(self.handle, &mut allowed) })?;
Ok(allowed != 0)
}
pub fn record_success(&mut self) -> Result<()> {
check(unsafe { ffi::fb_circuit_breaker_record_success(self.handle) })
}
pub fn record_failure(&mut self) -> Result<()> {
check(unsafe { ffi::fb_circuit_breaker_record_failure(self.handle) })
}
pub fn state(&mut self) -> Result<CircuitState> {
let mut state = 0;
check(unsafe { ffi::fb_circuit_breaker_state(self.handle, &mut state) })?;
Ok(CircuitState::from_raw(state))
}
}
impl Drop for CircuitBreaker {
fn drop(&mut self) {
unsafe {
ffi::fb_circuit_breaker_destroy(self.handle);
}
}
}
pub struct Bulkhead {
handle: ffi::FbBulkhead,
}
impl Bulkhead {
pub fn new(capacity: i32) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_bulkhead_create(capacity, &mut handle) })?;
Ok(Self { handle })
}
pub fn inspect(&mut self) -> Result<BulkheadDecision> {
let mut raw = ffi::FbBulkheadResult {
allowed: 0,
capacity: 0,
in_use: 0,
remaining: 0,
};
check(unsafe { ffi::fb_bulkhead_inspect(self.handle, &mut raw) })?;
Ok(raw.into())
}
pub fn acquire(&mut self) -> Result<BulkheadDecision> {
let mut raw = ffi::FbBulkheadResult {
allowed: 0,
capacity: 0,
in_use: 0,
remaining: 0,
};
check(unsafe { ffi::fb_bulkhead_acquire(self.handle, &mut raw) })?;
Ok(raw.into())
}
pub fn release(&mut self) -> Result<()> {
check(unsafe { ffi::fb_bulkhead_release(self.handle) })
}
}
impl Drop for Bulkhead {
fn drop(&mut self) {
unsafe {
ffi::fb_bulkhead_destroy(self.handle);
}
}
}
pub struct KeyedBulkhead {
handle: ffi::FbKeyedBulkhead,
}
impl KeyedBulkhead {
pub fn new(capacity: i32, max_keys: i32) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_keyed_bulkhead_create(capacity, max_keys, &mut handle) })?;
Ok(Self { handle })
}
pub fn inspect(&mut self, key: &str) -> Result<BulkheadDecision> {
let mut raw = ffi::FbBulkheadResult {
allowed: 0,
capacity: 0,
in_use: 0,
remaining: 0,
};
check(unsafe {
ffi::fb_keyed_bulkhead_inspect(self.handle, key.as_ptr().cast(), key.len(), &mut raw)
})?;
Ok(raw.into())
}
pub fn acquire(&mut self, key: &str) -> Result<BulkheadDecision> {
let mut raw = ffi::FbBulkheadResult {
allowed: 0,
capacity: 0,
in_use: 0,
remaining: 0,
};
check(unsafe {
ffi::fb_keyed_bulkhead_acquire(self.handle, key.as_ptr().cast(), key.len(), &mut raw)
})?;
Ok(raw.into())
}
pub fn release(&mut self, key: &str) -> Result<()> {
check(unsafe {
ffi::fb_keyed_bulkhead_release(self.handle, key.as_ptr().cast(), key.len())
})
}
pub fn clear(&mut self, key: &str) -> Result<bool> {
let mut cleared = 0;
check(unsafe {
ffi::fb_keyed_bulkhead_clear(self.handle, key.as_ptr().cast(), key.len(), &mut cleared)
})?;
Ok(cleared != 0)
}
pub fn active_keys(&mut self) -> Result<i32> {
let mut count = 0;
check(unsafe { ffi::fb_keyed_bulkhead_active_keys(self.handle, &mut count) })?;
Ok(count)
}
}
impl Drop for KeyedBulkhead {
fn drop(&mut self) {
unsafe {
ffi::fb_keyed_bulkhead_destroy(self.handle);
}
}
}
pub struct BudgetLedger {
handle: ffi::FbBudgetLedger,
}
impl BudgetLedger {
pub fn new(limit: i64, per: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_budget_ledger_create(limit, duration_to_ns(per)?, &mut handle) })?;
Ok(Self { handle })
}
pub fn inspect(&mut self, key: &str, cost: i64) -> Result<BudgetDecision> {
let mut raw = ffi::FbBudgetResult {
allowed: 0,
limit: 0,
used: 0,
remaining: 0,
cost: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_budget_inspect(self.handle, key.as_ptr().cast(), key.len(), cost, &mut raw)
})?;
Ok(raw.into())
}
pub fn consume(&mut self, key: &str, cost: i64) -> Result<BudgetDecision> {
let mut raw = ffi::FbBudgetResult {
allowed: 0,
limit: 0,
used: 0,
remaining: 0,
cost: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_budget_consume(self.handle, key.as_ptr().cast(), key.len(), cost, &mut raw)
})?;
Ok(raw.into())
}
pub fn refund(&mut self, key: &str, amount: i64) -> Result<BudgetDecision> {
let mut raw = ffi::FbBudgetResult {
allowed: 0,
limit: 0,
used: 0,
remaining: 0,
cost: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_budget_refund(
self.handle,
key.as_ptr().cast(),
key.len(),
amount,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn reset(&mut self, key: &str) -> Result<BudgetDecision> {
let mut raw = ffi::FbBudgetResult {
allowed: 0,
limit: 0,
used: 0,
remaining: 0,
cost: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_budget_reset(self.handle, key.as_ptr().cast(), key.len(), &mut raw)
})?;
Ok(raw.into())
}
pub fn reset_all(&mut self) -> Result<()> {
check(unsafe { ffi::fb_budget_reset_all(self.handle) })
}
}
impl Drop for BudgetLedger {
fn drop(&mut self) {
unsafe {
ffi::fb_budget_ledger_destroy(self.handle);
}
}
}
pub struct RetryAllowance {
handle: ffi::FbRetryAllowance,
}
impl RetryAllowance {
pub fn new(
retry_ratio: f64,
per: Duration,
minimum_retries: i64,
max_keys: i32,
) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe {
ffi::fb_retry_allowance_create(
retry_ratio,
duration_to_ns(per)?,
minimum_retries,
max_keys,
&mut handle,
)
})?;
Ok(Self { handle })
}
pub fn record_original(&mut self, key: &str, amount: i64) -> Result<RetryAllowanceDecision> {
let mut raw = empty_retry_allowance_result();
check(unsafe {
ffi::fb_retry_allowance_record_original(
self.handle,
key.as_ptr().cast(),
key.len(),
amount,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn inspect_retry(&mut self, key: &str, cost: i64) -> Result<RetryAllowanceDecision> {
let mut raw = empty_retry_allowance_result();
check(unsafe {
ffi::fb_retry_allowance_inspect_retry(
self.handle,
key.as_ptr().cast(),
key.len(),
cost,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn record_retry(&mut self, key: &str, cost: i64) -> Result<RetryAllowanceDecision> {
let mut raw = empty_retry_allowance_result();
check(unsafe {
ffi::fb_retry_allowance_record_retry(
self.handle,
key.as_ptr().cast(),
key.len(),
cost,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn allow_retry(&mut self, key: &str, cost: i64) -> Result<bool> {
let mut allowed = 0;
check(unsafe {
ffi::fb_retry_allowance_allow_retry(
self.handle,
key.as_ptr().cast(),
key.len(),
cost,
&mut allowed,
)
})?;
Ok(allowed != 0)
}
pub fn clear(&mut self, key: &str) -> Result<bool> {
let mut cleared = 0;
check(unsafe {
ffi::fb_retry_allowance_clear(self.handle, key.as_ptr().cast(), key.len(), &mut cleared)
})?;
Ok(cleared != 0)
}
pub fn reset(&mut self, key: &str) -> Result<bool> {
let mut reset = 0;
check(unsafe {
ffi::fb_retry_allowance_reset(self.handle, key.as_ptr().cast(), key.len(), &mut reset)
})?;
Ok(reset != 0)
}
pub fn reset_all(&mut self) -> Result<i32> {
let mut removed = 0;
check(unsafe { ffi::fb_retry_allowance_reset_all(self.handle, &mut removed) })?;
Ok(removed)
}
pub fn prune_expired(&mut self) -> Result<()> {
check(unsafe { ffi::fb_retry_allowance_prune_expired(self.handle) })
}
pub fn active_keys(&mut self) -> Result<i32> {
let mut count = 0;
check(unsafe { ffi::fb_retry_allowance_active_keys(self.handle, &mut count) })?;
Ok(count)
}
pub fn configured_retry_ratio(&mut self) -> Result<f64> {
let mut ratio = 0.0;
check(unsafe { ffi::fb_retry_allowance_configured_retry_ratio(self.handle, &mut ratio) })?;
Ok(ratio)
}
pub fn configured_minimum_retries(&mut self) -> Result<i64> {
let mut minimum_retries = 0;
check(unsafe {
ffi::fb_retry_allowance_configured_minimum_retries(self.handle, &mut minimum_retries)
})?;
Ok(minimum_retries)
}
pub fn configured_period(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_retry_allowance_configured_period(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn key_capacity(&mut self) -> Result<i32> {
let mut capacity = 0;
check(unsafe { ffi::fb_retry_allowance_key_capacity(self.handle, &mut capacity) })?;
Ok(capacity)
}
}
impl Drop for RetryAllowance {
fn drop(&mut self) {
unsafe {
ffi::fb_retry_allowance_destroy(self.handle);
}
}
}
pub struct Timeout {
handle: ffi::FbTimeout,
}
impl Timeout {
pub fn new(after: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_timeout_create(duration_to_ns(after)?, &mut handle) })?;
Ok(Self { handle })
}
pub fn expired(&mut self) -> Result<bool> {
let mut expired = 0;
check(unsafe { ffi::fb_timeout_expired(self.handle, &mut expired) })?;
Ok(expired != 0)
}
pub fn elapsed(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_timeout_elapsed(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn remaining(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_timeout_remaining(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
}
impl Drop for Timeout {
fn drop(&mut self) {
unsafe {
ffi::fb_timeout_destroy(self.handle);
}
}
}
pub struct Deadline {
handle: ffi::FbDeadline,
}
impl Deadline {
pub fn new(after: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_deadline_create(duration_to_ns(after)?, &mut handle) })?;
Ok(Self { handle })
}
pub fn expired(&mut self) -> Result<bool> {
let mut expired = 0;
check(unsafe { ffi::fb_deadline_expired(self.handle, &mut expired) })?;
Ok(expired != 0)
}
pub fn remaining(&mut self) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_deadline_remaining(self.handle, &mut ns) })?;
Ok(ns_to_duration(ns))
}
pub fn clamp(&mut self, requested: Duration) -> Result<Duration> {
let mut ns = 0;
check(unsafe { ffi::fb_deadline_clamp(self.handle, duration_to_ns(requested)?, &mut ns) })?;
Ok(ns_to_duration(ns))
}
}
impl Drop for Deadline {
fn drop(&mut self) {
unsafe {
ffi::fb_deadline_destroy(self.handle);
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LockAcquireDecision {
pub acquired: bool,
pub ttl: Duration,
}
impl From<ffi::FbLockAcquireResult> for LockAcquireDecision {
fn from(value: ffi::FbLockAcquireResult) -> Self {
Self {
acquired: value.acquired != 0,
ttl: ns_to_duration(value.ttl_ns),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LockStatus {
pub held: bool,
pub expired: bool,
pub ttl: Duration,
pub remaining: Duration,
}
impl From<ffi::FbLockStatus> for LockStatus {
fn from(value: ffi::FbLockStatus) -> Self {
Self {
held: value.held != 0,
expired: value.expired != 0,
ttl: ns_to_duration(value.ttl_ns),
remaining: ns_to_duration(value.remaining_ns),
}
}
}
pub struct LockLease {
handle: ffi::FbLockLease,
}
impl Drop for LockLease {
fn drop(&mut self) {
unsafe {
ffi::fb_lock_lease_destroy(self.handle);
}
}
}
pub struct LockStore {
handle: ffi::FbLockStore,
}
impl LockStore {
pub fn new() -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_lock_store_create(&mut handle) })?;
Ok(Self { handle })
}
pub fn acquire(
&mut self,
key: &str,
ttl: Duration,
) -> Result<(LockLease, LockAcquireDecision)> {
let mut lease = ptr::null_mut();
let mut raw = ffi::FbLockAcquireResult {
acquired: 0,
ttl_ns: 0,
};
check(unsafe {
ffi::fb_lock_acquire(
self.handle,
key.as_ptr().cast(),
key.len(),
duration_to_ns(ttl)?,
&mut lease,
&mut raw,
)
})?;
Ok((LockLease { handle: lease }, raw.into()))
}
pub fn release(&mut self, lease: &mut LockLease) -> Result<bool> {
let mut released = 0;
check(unsafe { ffi::fb_lock_release(self.handle, lease.handle, &mut released) })?;
Ok(released != 0)
}
pub fn release_key(&mut self, key: &str) -> Result<bool> {
let mut released = 0;
check(unsafe {
ffi::fb_lock_release_key(self.handle, key.as_ptr().cast(), key.len(), &mut released)
})?;
Ok(released != 0)
}
pub fn refresh(&mut self, lease: &mut LockLease, ttl: Duration) -> Result<LockAcquireDecision> {
let mut raw = ffi::FbLockAcquireResult {
acquired: 0,
ttl_ns: 0,
};
check(unsafe {
ffi::fb_lock_refresh(self.handle, lease.handle, duration_to_ns(ttl)?, &mut raw)
})?;
Ok(raw.into())
}
pub fn inspect(&mut self, lease: &mut LockLease) -> Result<LockStatus> {
let mut raw = ffi::FbLockStatus {
held: 0,
expired: 0,
ttl_ns: 0,
remaining_ns: 0,
};
check(unsafe { ffi::fb_lock_inspect(self.handle, lease.handle, &mut raw) })?;
Ok(raw.into())
}
}
impl Drop for LockStore {
fn drop(&mut self) {
unsafe {
ffi::fb_lock_store_destroy(self.handle);
}
}
}
pub struct Throttle {
handle: ffi::FbThrottle,
}
impl Throttle {
pub fn new(every: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_throttle_create(duration_to_ns(every)?, &mut handle) })?;
Ok(Self { handle })
}
pub fn allow(&mut self) -> Result<bool> {
let mut allowed = 0;
check(unsafe { ffi::fb_throttle_allow(self.handle, &mut allowed) })?;
Ok(allowed != 0)
}
pub fn reset(&mut self) -> Result<()> {
check(unsafe { ffi::fb_throttle_reset(self.handle) })
}
}
impl Drop for Throttle {
fn drop(&mut self) {
unsafe {
ffi::fb_throttle_destroy(self.handle);
}
}
}
pub struct Debouncer {
handle: ffi::FbDebouncer,
}
impl Debouncer {
pub fn new(delay: Duration) -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_debouncer_create(duration_to_ns(delay)?, &mut handle) })?;
Ok(Self { handle })
}
pub fn call(&mut self) -> Result<()> {
check(unsafe { ffi::fb_debouncer_call(self.handle) })
}
pub fn ready(&mut self) -> Result<bool> {
let mut ready = 0;
check(unsafe { ffi::fb_debouncer_ready(self.handle, &mut ready) })?;
Ok(ready != 0)
}
pub fn consume_ready(&mut self) -> Result<bool> {
let mut ready = 0;
check(unsafe { ffi::fb_debouncer_consume_ready(self.handle, &mut ready) })?;
Ok(ready != 0)
}
pub fn cancel(&mut self) -> Result<()> {
check(unsafe { ffi::fb_debouncer_cancel(self.handle) })
}
}
impl Drop for Debouncer {
fn drop(&mut self) {
unsafe {
ffi::fb_debouncer_destroy(self.handle);
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RetryRunResult {
pub succeeded: bool,
pub attempts: i32,
pub last_status: i32,
pub last_delay: Duration,
}
impl From<ffi::FbRetryResult> for RetryRunResult {
fn from(value: ffi::FbRetryResult) -> Self {
Self {
succeeded: value.succeeded != 0,
attempts: value.attempts,
last_status: value.last_status,
last_delay: ns_to_duration(value.last_delay_ns),
}
}
}
struct RetryContext<'a> {
operation: &'a mut dyn FnMut(i32) -> i32,
sleep: Option<&'a mut dyn FnMut(Duration, i32) -> i32>,
panicked: bool,
}
unsafe extern "C" fn retry_operation_trampoline(user_data: *mut c_void, attempt: i32) -> i32 {
let ctx = &mut *(user_data.cast::<RetryContext<'_>>());
match catch_unwind(AssertUnwindSafe(|| (ctx.operation)(attempt))) {
Ok(status) => status,
Err(_) => {
ctx.panicked = true;
ffi::FB_ERR_INTERNAL
}
}
}
unsafe extern "C" fn retry_sleep_trampoline(
user_data: *mut c_void,
delay_ns: i64,
attempt: i32,
) -> i32 {
let ctx = &mut *(user_data.cast::<RetryContext<'_>>());
match ctx.sleep.as_mut() {
Some(sleep) => match catch_unwind(AssertUnwindSafe(|| {
sleep(ns_to_duration(delay_ns), attempt)
})) {
Ok(status) => status,
Err(_) => {
ctx.panicked = true;
ffi::FB_ERR_INTERNAL
}
},
None => ffi::FB_OK,
}
}
pub fn retry_run<'a>(
policy: &BackoffPolicy,
max_attempts: i32,
operation: &'a mut dyn FnMut(i32) -> i32,
sleep: Option<&'a mut dyn FnMut(Duration, i32) -> i32>,
) -> Result<RetryRunResult> {
init_runtime();
let mut ctx = RetryContext {
operation,
sleep,
panicked: false,
};
let mut raw = ffi::FbRetryResult {
succeeded: 0,
attempts: 0,
last_status: 0,
last_delay_ns: 0,
};
let status = unsafe {
ffi::fb_retry_run(
policy.handle,
max_attempts,
Some(retry_operation_trampoline),
Some(retry_sleep_trampoline),
(&mut ctx as *mut RetryContext<'_>).cast(),
&mut raw,
)
};
if ctx.panicked {
return Err(Error {
status: Status::Internal,
message: "Rust callback panicked while running retry".to_string(),
});
}
check(status)?;
Ok(raw.into())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FallbackRunResult {
pub succeeded: bool,
pub attempts: i32,
pub provider_index: i32,
pub failed_count: i32,
pub last_status: i32,
}
impl From<ffi::FbFallbackResult> for FallbackRunResult {
fn from(value: ffi::FbFallbackResult) -> Self {
Self {
succeeded: value.succeeded != 0,
attempts: value.attempts,
provider_index: value.provider_index,
failed_count: value.failed_count,
last_status: value.last_status,
}
}
}
pub struct FallbackProvider {
operation: Box<dyn FnMut(i32) -> i32>,
}
impl FallbackProvider {
pub fn new(operation: impl FnMut(i32) -> i32 + 'static) -> Self {
Self {
operation: Box::new(operation),
}
}
}
struct FallbackContext<'a> {
providers: &'a mut [FallbackProvider],
should_fallback: Option<&'a mut dyn FnMut(i32, i32) -> bool>,
panicked: bool,
}
unsafe extern "C" fn fallback_operation_trampoline(
user_data: *mut c_void,
provider_index: i32,
) -> i32 {
let ctx = &mut *(user_data.cast::<FallbackContext<'_>>());
let Some(provider) = ctx.providers.get_mut(provider_index as usize) else {
return ffi::FB_ERR_INVALID_ARGUMENT;
};
match catch_unwind(AssertUnwindSafe(|| (provider.operation)(provider_index))) {
Ok(status) => status,
Err(_) => {
ctx.panicked = true;
ffi::FB_ERR_INTERNAL
}
}
}
unsafe extern "C" fn fallback_predicate_trampoline(
user_data: *mut c_void,
status: i32,
provider_index: i32,
) -> i32 {
let ctx = &mut *(user_data.cast::<FallbackContext<'_>>());
match ctx.should_fallback.as_mut() {
Some(predicate) => {
match catch_unwind(AssertUnwindSafe(|| predicate(status, provider_index))) {
Ok(true) => 1,
Ok(false) => 0,
Err(_) => {
ctx.panicked = true;
0
}
}
}
None => 1,
}
}
pub fn fallback_run<'a>(
providers: &'a mut [FallbackProvider],
should_fallback: Option<&'a mut dyn FnMut(i32, i32) -> bool>,
) -> Result<FallbackRunResult> {
init_runtime();
let mut ctx = FallbackContext {
providers,
should_fallback,
panicked: false,
};
let raw_provider = ffi::FbFallbackProvider {
operation: Some(fallback_operation_trampoline),
user_data: (&mut ctx as *mut FallbackContext<'_>).cast(),
breaker: ptr::null_mut(),
};
let raw_providers: Vec<_> = ctx.providers.iter_mut().map(|_| raw_provider).collect();
let mut raw = ffi::FbFallbackResult {
succeeded: 0,
attempts: 0,
provider_index: -1,
failed_count: 0,
last_status: 0,
};
let status = unsafe {
ffi::fb_fallback_run(
raw_providers.as_ptr(),
raw_providers.len(),
Some(fallback_predicate_trampoline),
&mut raw,
)
};
if ctx.panicked {
return Err(Error {
status: Status::Internal,
message: "Rust callback panicked while running fallback".to_string(),
});
}
check(status)?;
Ok(raw.into())
}
pub trait FixedWindowStorage {
fn inspect_fixed_window(
&mut self,
key: &str,
limit: i32,
per: Duration,
cost: i32,
current: Duration,
) -> Result<RateLimitDecision>;
fn consume_fixed_window(
&mut self,
key: &str,
limit: i32,
per: Duration,
cost: i32,
current: Duration,
) -> Result<RateLimitDecision>;
fn clear_fixed_window(&mut self, key: &str) -> Result<bool>;
}
struct StorageBridge {
storage: Box<dyn FixedWindowStorage>,
panicked: bool,
}
unsafe extern "C" fn storage_inspect_trampoline(
user_data: *mut c_void,
key: *const i8,
key_len: usize,
limit: i32,
per_ns: i64,
cost: i32,
current_ns: i64,
out_result: *mut ffi::FbRateLimitResult,
) -> i32 {
let bridge = &mut *(user_data.cast::<StorageBridge>());
let key = String::from_utf8_lossy(std::slice::from_raw_parts(key.cast::<u8>(), key_len));
match catch_unwind(AssertUnwindSafe(|| {
bridge.storage.inspect_fixed_window(
&key,
limit,
ns_to_duration(per_ns),
cost,
ns_to_duration(current_ns),
)
})) {
Ok(Ok(decision)) => {
*out_result = decision.raw();
ffi::FB_OK
}
Ok(Err(_)) => ffi::FB_ERR_INTERNAL,
Err(_) => {
bridge.panicked = true;
ffi::FB_ERR_INTERNAL
}
}
}
unsafe extern "C" fn storage_consume_trampoline(
user_data: *mut c_void,
key: *const i8,
key_len: usize,
limit: i32,
per_ns: i64,
cost: i32,
current_ns: i64,
out_result: *mut ffi::FbRateLimitResult,
) -> i32 {
let bridge = &mut *(user_data.cast::<StorageBridge>());
let key = String::from_utf8_lossy(std::slice::from_raw_parts(key.cast::<u8>(), key_len));
match catch_unwind(AssertUnwindSafe(|| {
bridge.storage.consume_fixed_window(
&key,
limit,
ns_to_duration(per_ns),
cost,
ns_to_duration(current_ns),
)
})) {
Ok(Ok(decision)) => {
*out_result = decision.raw();
ffi::FB_OK
}
Ok(Err(_)) => ffi::FB_ERR_INTERNAL,
Err(_) => {
bridge.panicked = true;
ffi::FB_ERR_INTERNAL
}
}
}
unsafe extern "C" fn storage_clear_trampoline(
user_data: *mut c_void,
key: *const i8,
key_len: usize,
out_cleared: *mut i32,
) -> i32 {
let bridge = &mut *(user_data.cast::<StorageBridge>());
let key = String::from_utf8_lossy(std::slice::from_raw_parts(key.cast::<u8>(), key_len));
match catch_unwind(AssertUnwindSafe(|| bridge.storage.clear_fixed_window(&key))) {
Ok(Ok(cleared)) => {
*out_cleared = if cleared { 1 } else { 0 };
ffi::FB_OK
}
Ok(Err(_)) => ffi::FB_ERR_INTERNAL,
Err(_) => {
bridge.panicked = true;
ffi::FB_ERR_INTERNAL
}
}
}
pub struct LimiterRegistry {
handle: ffi::FbLimiterRegistry,
#[allow(clippy::vec_box)]
storages: Vec<Box<StorageBridge>>,
}
impl LimiterRegistry {
pub fn new() -> Result<Self> {
init_runtime();
let mut handle = ptr::null_mut();
check(unsafe { ffi::fb_limiter_registry_create(&mut handle) })?;
Ok(Self {
handle,
storages: Vec::new(),
})
}
pub fn add_fixed_window(&mut self, name: &str, limit: i32, per: Duration) -> Result<()> {
check(unsafe {
ffi::fb_limiter_registry_add_fixed_window(
self.handle,
name.as_ptr().cast(),
name.len(),
limit,
duration_to_ns(per)?,
)
})
}
pub fn add_sliding_window(&mut self, name: &str, limit: i32, per: Duration) -> Result<()> {
check(unsafe {
ffi::fb_limiter_registry_add_sliding_window(
self.handle,
name.as_ptr().cast(),
name.len(),
limit,
duration_to_ns(per)?,
)
})
}
pub fn add_token_bucket(
&mut self,
name: &str,
rate: i32,
per: Duration,
burst: i32,
) -> Result<()> {
check(unsafe {
ffi::fb_limiter_registry_add_token_bucket(
self.handle,
name.as_ptr().cast(),
name.len(),
rate,
duration_to_ns(per)?,
burst,
)
})
}
pub fn add_keyed_fixed_window(
&mut self,
name: &str,
limit: i32,
per: Duration,
max_keys: i32,
) -> Result<()> {
check(unsafe {
ffi::fb_limiter_registry_add_keyed_fixed_window(
self.handle,
name.as_ptr().cast(),
name.len(),
limit,
duration_to_ns(per)?,
max_keys,
)
})
}
pub fn add_stored_fixed_window(
&mut self,
name: &str,
prefix: &str,
limit: i32,
per: Duration,
max_key_length: i32,
storage: impl FixedWindowStorage + 'static,
) -> Result<()> {
let mut bridge = Box::new(StorageBridge {
storage: Box::new(storage),
panicked: false,
});
let raw_storage = ffi::FbRateLimitStorage {
inspect_fixed_window: Some(storage_inspect_trampoline),
consume_fixed_window: Some(storage_consume_trampoline),
clear_fixed_window: Some(storage_clear_trampoline),
user_data: (&mut *bridge as *mut StorageBridge).cast(),
};
check(unsafe {
ffi::fb_limiter_registry_add_stored_fixed_window(
self.handle,
name.as_ptr().cast(),
name.len(),
prefix.as_ptr().cast(),
prefix.len(),
limit,
duration_to_ns(per)?,
&raw_storage,
max_key_length,
)
})?;
self.storages.push(bridge);
Ok(())
}
pub fn add_compound(&mut self, name: &str, child_names: &[&str]) -> Result<()> {
let child_ptrs: Vec<_> = child_names
.iter()
.map(|name| name.as_ptr().cast())
.collect();
let child_lens: Vec<_> = child_names.iter().map(|name| name.len()).collect();
check(unsafe {
ffi::fb_limiter_registry_add_compound(
self.handle,
name.as_ptr().cast(),
name.len(),
child_ptrs.as_ptr(),
child_lens.as_ptr(),
child_names.len(),
)
})
}
pub fn inspect(&mut self, name: &str, key: &str, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_limiter_registry_inspect(
self.handle,
name.as_ptr().cast(),
name.len(),
key.as_ptr().cast(),
key.len(),
cost,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn consume(&mut self, name: &str, key: &str, cost: i32) -> Result<RateLimitDecision> {
let mut raw = ffi::FbRateLimitResult {
allowed: 0,
limit: 0,
remaining: 0,
retry_after_ns: 0,
reset_after_ns: 0,
};
check(unsafe {
ffi::fb_limiter_registry_consume(
self.handle,
name.as_ptr().cast(),
name.len(),
key.as_ptr().cast(),
key.len(),
cost,
&mut raw,
)
})?;
Ok(raw.into())
}
pub fn allow(&mut self, name: &str, key: &str, cost: i32) -> Result<bool> {
let mut allowed = 0;
check(unsafe {
ffi::fb_limiter_registry_allow(
self.handle,
name.as_ptr().cast(),
name.len(),
key.as_ptr().cast(),
key.len(),
cost,
&mut allowed,
)
})?;
Ok(allowed != 0)
}
pub fn clear(&mut self, name: &str, key: &str) -> Result<bool> {
let mut cleared = 0;
check(unsafe {
ffi::fb_limiter_registry_clear(
self.handle,
name.as_ptr().cast(),
name.len(),
key.as_ptr().cast(),
key.len(),
&mut cleared,
)
})?;
Ok(cleared != 0)
}
}
impl Drop for LimiterRegistry {
fn drop(&mut self) {
unsafe {
ffi::fb_limiter_registry_destroy(self.handle);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
#[test]
fn parses_and_formats_duration() {
let parsed = parse_duration("1s500ms").unwrap();
assert_eq!(parsed, Duration::from_millis(1500));
assert_eq!(format_duration(parsed).unwrap(), "1s500ms");
}
#[test]
fn token_bucket_consumes_capacity() {
let mut bucket = TokenBucket::new(2, Duration::from_secs(1), 2).unwrap();
assert!(bucket.consume(1).unwrap().allowed);
assert!(bucket.consume(1).unwrap().allowed);
assert!(!bucket.consume(1).unwrap().allowed);
}
#[test]
fn gcra_consumes_capacity_and_exposes_config() {
let mut limiter = GcraLimiter::new(2, Duration::from_secs(1), 2).unwrap();
assert_eq!(limiter.configured_rate().unwrap(), 2);
assert_eq!(limiter.configured_period().unwrap(), Duration::from_secs(1));
assert_eq!(limiter.configured_burst().unwrap(), 2);
assert_eq!(
limiter.configured_interval().unwrap(),
Duration::from_millis(500)
);
assert_eq!(limiter.available_capacity().unwrap(), 2);
assert!(limiter.consume(1).unwrap().allowed);
assert!(limiter.consume(1).unwrap().allowed);
assert!(!limiter.consume(1).unwrap().allowed);
limiter.reset().unwrap();
assert!(limiter.allow(2).unwrap());
assert!(!limiter.allow(1).unwrap());
}
#[test]
fn fixed_window_denies_after_limit() {
let mut limiter = FixedWindow::new(1, Duration::from_secs(1)).unwrap();
assert!(limiter.consume(1).unwrap().allowed);
assert!(!limiter.consume(1).unwrap().allowed);
}
#[test]
fn keyed_fixed_window_isolates_keys() {
let mut limiter = KeyedFixedWindow::new(1, Duration::from_secs(1), 8).unwrap();
assert!(limiter.consume("a", 1).unwrap().allowed);
assert!(!limiter.consume("a", 1).unwrap().allowed);
assert!(limiter.consume("b", 1).unwrap().allowed);
}
#[test]
fn keyed_gcra_isolates_keys_and_manages_state() {
let mut limiter = KeyedGcraLimiter::new(2, Duration::from_secs(1), 2, 2).unwrap();
assert_eq!(limiter.configured_rate().unwrap(), 2);
assert_eq!(limiter.configured_period().unwrap(), Duration::from_secs(1));
assert_eq!(limiter.configured_burst().unwrap(), 2);
assert_eq!(
limiter.configured_interval().unwrap(),
Duration::from_millis(500)
);
assert_eq!(limiter.key_capacity().unwrap(), 2);
assert!(limiter.inspect("a", 1).unwrap().allowed);
assert_eq!(limiter.active_keys().unwrap(), 0);
assert!(limiter.consume("a", 1).unwrap().allowed);
assert!(limiter.consume("a", 1).unwrap().allowed);
assert!(!limiter.consume("a", 1).unwrap().allowed);
assert_eq!(limiter.active_keys().unwrap(), 1);
assert!(limiter.consume("b", 1).unwrap().allowed);
assert!(limiter.consume("c", 1).is_err());
assert!(limiter.clear("a").unwrap());
assert!(!limiter.clear("a").unwrap());
assert!(limiter.reset("b").unwrap());
assert_eq!(limiter.reset_all().unwrap(), 0);
assert!(limiter.allow("a", 2).unwrap());
assert!(!limiter.allow("a", 1).unwrap());
limiter.prune_idle().unwrap();
}
#[test]
fn bulkhead_tracks_capacity() {
let mut bulkhead = Bulkhead::new(1).unwrap();
assert!(bulkhead.acquire().unwrap().allowed);
assert!(!bulkhead.acquire().unwrap().allowed);
bulkhead.release().unwrap();
assert!(bulkhead.acquire().unwrap().allowed);
}
#[test]
fn budget_ledger_tracks_usage() {
let mut budget = BudgetLedger::new(10, Duration::from_secs(60)).unwrap();
let first = budget.consume("tenant-a", 7).unwrap();
assert!(first.allowed);
let second = budget.consume("tenant-a", 4).unwrap();
assert!(!second.allowed);
}
#[test]
fn retry_allowance_tracks_retry_budget() {
let mut allowance = RetryAllowance::new(0.5, Duration::from_secs(60), 1, 2).unwrap();
assert_eq!(allowance.configured_retry_ratio().unwrap(), 0.5);
assert_eq!(allowance.configured_minimum_retries().unwrap(), 1);
assert_eq!(
allowance.configured_period().unwrap(),
Duration::from_secs(60)
);
assert_eq!(allowance.key_capacity().unwrap(), 2);
let original = allowance.record_original(" tenant-a ", 4).unwrap();
assert!(original.allowed);
assert_eq!(original.originals, 4);
assert_eq!(original.limit, 3);
assert_eq!(original.remaining, 3);
let inspected = allowance.inspect_retry("tenant-a", 2).unwrap();
assert!(inspected.allowed);
assert_eq!(inspected.retries, 2);
let consumed = allowance.record_retry("tenant-a", 2).unwrap();
assert!(consumed.allowed);
assert!(!allowance.record_retry("tenant-a", 2).unwrap().allowed);
assert!(allowance.allow_retry("tenant-a", 1).unwrap());
assert!(!allowance.allow_retry("tenant-a", 1).unwrap());
assert_eq!(allowance.active_keys().unwrap(), 1);
assert!(allowance.clear("tenant-a").unwrap());
assert!(!allowance.clear("tenant-a").unwrap());
allowance.record_original("tenant-b", 1).unwrap();
allowance.record_original("tenant-c", 1).unwrap();
assert!(allowance.record_original("tenant-d", 1).is_err());
assert!(allowance.reset("tenant-b").unwrap());
assert_eq!(allowance.reset_all().unwrap(), 1);
allowance.prune_expired().unwrap();
}
#[test]
fn sliding_window_limits_capacity() {
let mut limiter = SlidingWindow::new(1, Duration::from_secs(1)).unwrap();
assert!(limiter.consume(1).unwrap().allowed);
assert!(!limiter.consume(1).unwrap().allowed);
assert_eq!(limiter.configured_limit().unwrap(), 1);
}
#[test]
fn circuit_breaker_opens_after_failures() {
let mut breaker = CircuitBreaker::new(2, Duration::from_secs(1)).unwrap();
assert!(breaker.allow().unwrap());
breaker.record_failure().unwrap();
breaker.record_failure().unwrap();
assert_eq!(breaker.state().unwrap(), CircuitState::Open);
assert!(!breaker.allow().unwrap());
}
#[test]
fn keyed_bulkhead_isolates_keys() {
let mut bulkhead = KeyedBulkhead::new(1, 8).unwrap();
assert!(bulkhead.acquire("a").unwrap().allowed);
assert!(!bulkhead.acquire("a").unwrap().allowed);
assert!(bulkhead.acquire("b").unwrap().allowed);
bulkhead.release("a").unwrap();
assert!(bulkhead.acquire("a").unwrap().allowed);
}
#[test]
fn timeout_reports_elapsed_and_remaining() {
let mut timeout = Timeout::new(Duration::from_secs(1)).unwrap();
assert!(!timeout.expired().unwrap());
assert!(timeout.remaining().unwrap() <= Duration::from_secs(1));
assert!(timeout.elapsed().unwrap() <= Duration::from_secs(1));
}
#[test]
fn lock_store_acquires_releases_and_inspects() {
let mut store = LockStore::new().unwrap();
let (mut lease, decision) = store.acquire("job", Duration::from_secs(5)).unwrap();
assert!(decision.acquired);
assert!(store.inspect(&mut lease).unwrap().held);
assert!(store.release(&mut lease).unwrap());
assert!(!store.inspect(&mut lease).unwrap().held);
}
#[test]
fn throttle_and_debouncer_expose_state() {
let mut throttle = Throttle::new(Duration::from_secs(1)).unwrap();
assert!(throttle.allow().unwrap());
assert!(!throttle.allow().unwrap());
throttle.reset().unwrap();
assert!(throttle.allow().unwrap());
let mut debouncer = Debouncer::new(Duration::from_millis(10)).unwrap();
assert!(!debouncer.ready().unwrap());
debouncer.call().unwrap();
assert!(!debouncer.consume_ready().unwrap());
debouncer.cancel().unwrap();
}
#[test]
fn retry_runner_uses_callbacks() {
let policy = BackoffPolicy::fixed(Duration::from_millis(1), Jitter::None).unwrap();
let mut attempts = 0;
let mut slept = 0;
let mut operation = |_: i32| {
attempts += 1;
if attempts == 2 {
0
} else {
1
}
};
let mut sleep = |_: Duration, _: i32| {
slept += 1;
0
};
let result = retry_run(&policy, 3, &mut operation, Some(&mut sleep)).unwrap();
assert!(result.succeeded);
assert_eq!(result.attempts, 2);
assert_eq!(slept, 1);
}
#[test]
fn fallback_runner_tries_providers_in_order() {
let mut providers = vec![FallbackProvider::new(|_| 1), FallbackProvider::new(|_| 0)];
let result = fallback_run(&mut providers, None).unwrap();
assert!(result.succeeded);
assert_eq!(result.provider_index, 1);
assert_eq!(result.failed_count, 1);
}
#[test]
fn limiter_registry_uses_named_limiters() {
let mut registry = LimiterRegistry::new().unwrap();
registry
.add_fixed_window("one", 1, Duration::from_secs(1))
.unwrap();
assert!(registry.consume("one", "client", 1).unwrap().allowed);
assert!(!registry.consume("one", "client", 1).unwrap().allowed);
registry
.add_token_bucket("burst", 10, Duration::from_secs(1), 10)
.unwrap();
registry
.add_keyed_fixed_window("keyed", 1, Duration::from_secs(1), 8)
.unwrap();
assert!(registry.consume("keyed", "client", 1).unwrap().allowed);
assert!(!registry.consume("keyed", "client", 1).unwrap().allowed);
assert!(registry.clear("keyed", "client").unwrap());
assert!(registry.consume("keyed", "client", 1).unwrap().allowed);
registry
.add_fixed_window("fresh", 1, Duration::from_secs(1))
.unwrap();
registry.add_compound("both", &["fresh", "burst"]).unwrap();
assert!(registry.inspect("both", "other-client", 1).unwrap().allowed);
}
struct MemoryStorage {
used: HashMap<String, i32>,
}
impl FixedWindowStorage for MemoryStorage {
fn inspect_fixed_window(
&mut self,
key: &str,
limit: i32,
_per: Duration,
cost: i32,
_current: Duration,
) -> Result<RateLimitDecision> {
let used = *self.used.get(key).unwrap_or(&0);
let allowed = used + cost <= limit;
Ok(RateLimitDecision {
allowed,
limit,
remaining: if allowed { limit - used - cost } else { 0 },
retry_after: Duration::ZERO,
reset_after: Duration::from_secs(1),
})
}
fn consume_fixed_window(
&mut self,
key: &str,
limit: i32,
per: Duration,
cost: i32,
current: Duration,
) -> Result<RateLimitDecision> {
let decision = self.inspect_fixed_window(key, limit, per, cost, current)?;
if decision.allowed {
*self.used.entry(key.to_string()).or_insert(0) += cost;
}
Ok(decision)
}
fn clear_fixed_window(&mut self, key: &str) -> Result<bool> {
Ok(self.used.remove(key).is_some())
}
}
#[test]
fn limiter_registry_uses_callback_storage() {
let mut registry = LimiterRegistry::new().unwrap();
registry
.add_stored_fixed_window(
"stored",
"prefix",
1,
Duration::from_secs(1),
128,
MemoryStorage {
used: HashMap::new(),
},
)
.unwrap();
assert!(registry.consume("stored", "client", 1).unwrap().allowed);
assert!(!registry.consume("stored", "client", 1).unwrap().allowed);
assert!(registry.clear("stored", "client").unwrap());
}
#[test]
fn exports_result_text() {
let mut limiter = FixedWindow::new(1, Duration::from_secs(1)).unwrap();
let decision = limiter.consume(1).unwrap();
let rate_json = decision.to_json().unwrap();
assert!(rate_json.contains("\"allowed\""));
assert!(rate_json.contains("true"));
assert!(decision.to_prometheus().unwrap().contains("flowbrigade"));
let mut budget = BudgetLedger::new(10, Duration::from_secs(1)).unwrap();
let budget_decision = budget.consume("tenant", 1).unwrap();
assert!(budget_decision
.to_json("tenant")
.unwrap()
.contains("tenant"));
assert!(budget_decision
.to_prometheus("tenant")
.unwrap()
.contains("flowbrigade"));
}
}