add rate-limit feature and http-rate crate. (#895)

* add rate-limit feature and http-rate crate.

* fix fmt.

* clippy fix.

* add package meta.

* update change log.

Cargo.toml

@@ -16,6 +16,7 @@ members = [
     "http-encoding",
     "http-file",
     "http-multipart",
+    "http-rate",
     "http-ws",
 ]
 
@@ -34,6 +35,7 @@ xitca-web = { path = "./web" }
 http-encoding = { path = "./http-encoding" }
 http-file = { path = "http-file" }
 http-multipart = { path = "./http-multipart" }
+http-rate = { path = "./http-rate" }
 http-ws = { path = "./http-ws" }
 
 [profile.release]

http-rate/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "http-rate"
+version = "0.1.0"
+edition = "2021"
+authors = ["fakeshadow <[email protected]>"]
+license = "MIT"
+description = "rate limit for http crate types"
+repository = "https://github.com/HFQR/xitca-web"
+keywords = ["http", "rate-limit"]
+readme= "README.md"
+
+[dependencies]
+http = "1"
+
+[dev-dependencies]
+crossbeam = "0.8.0"
+libc = "0.2.70"
+proptest = "1.0.0"
+all_asserts = "2.2.0"

http-rate/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Andreas Fuchs
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

http-rate/README.md

@@ -0,0 +1 @@
+# rate limit for http types

http-rate/src/error.rs

@@ -0,0 +1,76 @@
+use core::fmt;
+
+use std::{error, time::Instant};
+
+use http::{HeaderName, HeaderValue, Response, StatusCode};
+
+use crate::{
+    gcra::NotUntil,
+    timer::{DefaultTimer, Timer},
+};
+
+/// Error happen when client exceeds rate limit.
+#[derive(Debug)]
+pub struct TooManyRequests {
+    after_seconds: u64,
+}
+
+impl fmt::Display for TooManyRequests {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "too many requests. wait for {}", self.after_seconds)
+    }
+}
+
+impl error::Error for TooManyRequests {}
+
+impl From<NotUntil<Instant>> for TooManyRequests {
+    fn from(e: NotUntil<Instant>) -> Self {
+        let after_seconds = e.wait_time_from(DefaultTimer.now()).as_secs();
+
+        Self { after_seconds }
+    }
+}
+
+const X_RT_AFTER: HeaderName = HeaderName::from_static("x-ratelimit-after");
+
+impl TooManyRequests {
+    /// extend response headers with status code and headers
+    /// StatusCode: 429
+    /// Header: `x-ratelimit-after: <num in second>`
+    pub fn extend_response<Ext>(&self, res: &mut Response<Ext>) {
+        *res.status_mut() = StatusCode::TOO_MANY_REQUESTS;
+        res.headers_mut()
+            .insert(X_RT_AFTER, HeaderValue::from(self.after_seconds));
+    }
+}
+
+/// Error indicating that the number of cells tested (the first
+/// argument) is larger than the bucket's capacity.
+///
+/// This means the decision can never have a conforming result. The
+/// argument gives the maximum number of cells that could ever have a
+/// conforming result.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct InsufficientCapacity(pub u32);
+
+impl fmt::Display for InsufficientCapacity {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "required number of cells {} exceeds bucket's capacity", self.0)
+    }
+}
+
+impl std::error::Error for InsufficientCapacity {}
+
+#[cfg(test)]
+mod test {
+    use super::InsufficientCapacity;
+
+    #[test]
+    fn coverage() {
+        let display_output = format!("{}", InsufficientCapacity(3));
+        assert!(display_output.contains('3'));
+        let debug_output = format!("{:?}", InsufficientCapacity(3));
+        assert!(debug_output.contains('3'));
+        assert_eq!(InsufficientCapacity(3), InsufficientCapacity(3));
+    }
+}

http-rate/src/gcra.rs

@@ -0,0 +1,208 @@
+use core::{cmp, fmt, time::Duration};
+
+use crate::{nanos::Nanos, quota::Quota, snapshot::RateSnapshot, state::StateStore, timer};
+
+#[cfg(test)]
+use core::num::NonZeroU32;
+
+#[cfg(test)]
+use crate::error::InsufficientCapacity;
+
+/// A negative rate-limiting outcome.
+///
+/// `NotUntil`'s methods indicate when a caller can expect the next positive
+/// rate-limiting result.
+#[derive(Debug, PartialEq, Eq)]
+pub struct NotUntil<P: timer::Reference> {
+    state: RateSnapshot,
+    start: P,
+}
+
+impl<P: timer::Reference> NotUntil<P> {
+    /// Create a `NotUntil` as a negative rate-limiting result.
+    #[inline]
+    pub(crate) fn new(state: RateSnapshot, start: P) -> Self {
+        Self { state, start }
+    }
+
+    /// Returns the earliest time at which a decision could be
+    /// conforming (excluding conforming decisions made by the Decider
+    /// that are made in the meantime).
+    #[inline]
+    pub fn earliest_possible(&self) -> P {
+        let tat: Nanos = self.state.tat;
+        self.start + tat
+    }
+
+    /// Returns the minimum amount of time from the time that the
+    /// decision was made that must pass before a
+    /// decision can be conforming.
+    ///
+    /// If the time of the next expected positive result is in the past,
+    /// `wait_time_from` returns a zero `Duration`.
+    #[inline]
+    pub fn wait_time_from(&self, from: P) -> Duration {
+        let earliest = self.earliest_possible();
+        earliest.duration_since(earliest.min(from)).into()
+    }
+
+    /// Returns the rate limiting [`Quota`] used to reach the decision.
+    #[inline]
+    pub fn quota(&self) -> Quota {
+        self.state.quota()
+    }
+}
+
+impl<P: timer::Reference> fmt::Display for NotUntil<P> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
+        write!(f, "rate-limited until {:?}", self.start + self.state.tat)
+    }
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) struct Gcra {
+    /// The "weight" of a single packet in units of time.
+    t: Nanos,
+
+    /// The "burst capacity" of the bucket.
+    tau: Nanos,
+}
+
+impl Gcra {
+    pub(crate) fn new(quota: Quota) -> Self {
+        let tau: Nanos = (cmp::max(quota.replenish_1_per, Duration::from_nanos(1)) * quota.max_burst.get()).into();
+        let t: Nanos = quota.replenish_1_per.into();
+        Gcra { t, tau }
+    }
+
+    /// Computes and returns a new ratelimiter state if none exists yet.
+    fn starting_state(&self, t0: Nanos) -> Nanos {
+        t0 + self.t
+    }
+
+    /// Tests a single cell against the rate limiter state and updates it at the given key.
+    pub(crate) fn test_and_update<K, P: timer::Reference, S: StateStore<Key = K>>(
+        &self,
+        start: P,
+        key: &K,
+        state: &S,
+        t0: P,
+    ) -> Result<RateSnapshot, NotUntil<P>> {
+        let t0 = t0.duration_since(start);
+        let tau = self.tau;
+        let t = self.t;
+        state.measure_and_replace(key, |tat| {
+            let tat = tat.unwrap_or_else(|| self.starting_state(t0));
+            let earliest_time = tat.saturating_sub(tau);
+            if t0 < earliest_time {
+                let state = RateSnapshot::new(self.t, self.tau, earliest_time, earliest_time);
+                Err(NotUntil::new(state, start))
+            } else {
+                let next = cmp::max(tat, t0) + t;
+                Ok((RateSnapshot::new(self.t, self.tau, t0, next), next))
+            }
+        })
+    }
+
+    #[cfg(test)]
+    /// Tests whether all `n` cells could be accommodated and updates the rate limiter state, if so.
+    pub(crate) fn test_n_all_and_update<K, P: timer::Reference, S: StateStore<Key = K>>(
+        &self,
+        start: P,
+        key: &K,
+        n: NonZeroU32,
+        state: &S,
+        t0: P,
+    ) -> Result<Result<RateSnapshot, NotUntil<P>>, InsufficientCapacity> {
+        let t0 = t0.duration_since(start);
+        let tau = self.tau;
+        let t = self.t;
+        let additional_weight = t * (n.get() - 1) as u64;
+
+        // check that we can allow enough cells through. Note that `additional_weight` is the
+        // value of the cells *in addition* to the first cell - so add that first cell back.
+        if additional_weight + t > tau {
+            return Err(InsufficientCapacity((tau.as_u64() / t.as_u64()) as u32));
+        }
+        Ok(state.measure_and_replace(key, |tat| {
+            let tat = tat.unwrap_or_else(|| self.starting_state(t0));
+            let earliest_time = (tat + additional_weight).saturating_sub(tau);
+            if t0 < earliest_time {
+                let state = RateSnapshot::new(self.t, self.tau, earliest_time, earliest_time);
+                Err(NotUntil::new(state, start))
+            } else {
+                let next = cmp::max(tat, t0) + t + additional_weight;
+                Ok((RateSnapshot::new(self.t, self.tau, t0, next), next))
+            }
+        }))
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use proptest::prelude::*;
+
+    use crate::quota::Quota;
+
+    use super::*;
+
+    /// Exercise derives and convenience impls on Gcra to make coverage happy
+    #[test]
+    fn gcra_derives() {
+        use all_asserts::assert_gt;
+
+        let g = Gcra::new(Quota::per_second(1));
+        let g2 = Gcra::new(Quota::per_second(2));
+        assert_eq!(g, g);
+        assert_ne!(g, g2);
+        assert_gt!(format!("{:?}", g).len(), 0);
+    }
+
+    /// Exercise derives and convenience impls on NotUntil to make coverage happy
+    #[test]
+    fn notuntil_impls() {
+        use crate::state::RateLimiter;
+        use all_asserts::assert_gt;
+        use timer::FakeRelativeClock;
+
+        let clock = FakeRelativeClock::default();
+        let quota = Quota::per_second(1);
+        let lb = RateLimiter::direct_with_clock(quota, &clock);
+        assert!(lb.check().is_ok());
+        assert!(lb
+            .check()
+            .map_err(|nu| {
+                assert_eq!(nu, nu);
+                assert_gt!(format!("{:?}", nu).len(), 0);
+                assert_eq!(format!("{}", nu), "rate-limited until Nanos(1s)");
+                assert_eq!(nu.quota(), quota);
+            })
+            .is_err());
+    }
+
+    #[derive(Debug)]
+    struct Count(NonZeroU32);
+    impl Arbitrary for Count {
+        type Parameters = ();
+        fn arbitrary_with(_args: ()) -> Self::Strategy {
+            (1..10000u32).prop_map(|x| Count(NonZeroU32::new(x).unwrap())).boxed()
+        }
+
+        type Strategy = BoxedStrategy<Count>;
+    }
+
+    #[test]
+    fn cover_count_derives() {
+        assert_eq!(format!("{:?}", Count(NonZeroU32::new(1).unwrap())), "Count(1)");
+    }
+
+    #[test]
+    fn roundtrips_quota() {
+        proptest!(ProptestConfig::default(), |(per_second: Count, burst: Count)| {
+            let quota = Quota::per_second(per_second.0).allow_burst(burst.0);
+            let gcra = Gcra::new(quota);
+            let back = Quota::from_gcra_parameters(gcra.t, gcra.tau);
+            assert_eq!(quota, back);
+        })
+    }
+}