You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2026/08/25 02:10:30 UTC
(spark-connect-gateway) branch main updated: [SPARK-58961] Add scg-audit and scg-ratelimit crates
This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-connect-gateway.git
The following commit(s) were added to refs/heads/main by this push:
new 8415202 [SPARK-58961] Add scg-audit and scg-ratelimit crates
8415202 is described below
commit 8415202509f8dfef30d7f83c8ab29aebb44175a6
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Mon Aug 24 19:10:22 2026 -0700
[SPARK-58961] Add scg-audit and scg-ratelimit crates
### What changes were proposed in this pull request?
Part of importing the Spark Connect Gateway reference implementation into
`apache/spark-connect-gateway` as a series of small, bottom-up PRs (tracked in
the do-not-merge reference PR #1).
This adds two small leaf crates, grouped into one PR:
- `scg-audit` — structured audit logging for the gateway.
- `scg-ratelimit` — per-tenant (and optional per-user) token-bucket rate
limiting, with a Redis-backed distributed token bucket for multi-replica
deployments.
Both are leaf crates (no internal dependencies), so they compile and test on
their own (audit: 6 unit tests; ratelimit: 9 tests, plus Redis integration tests
that are `#[ignore]` so they compile but don't require Docker in CI). They are
appended to the workspace `members` list, and are consumed later by the `proxy`
crate.
### Why are the changes needed?
Implements the reference gateway from the approved SPIP (SPARK-58455), imported
one reviewable piece at a time.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`cargo fmt --all --check`, `cargo clippy --workspace --all-targets -- -D warnings`,
`cargo build --workspace`, and `cargo test --workspace` all pass locally
(audit 6 tests, ratelimit 9 tests; Redis integration tests are `#[ignore]`).
### Was this patch authored or co-authored using generative AI tooling?
Yes, co-authored with Claude Code.
Closes #9 from viirya/import-06-audit-ratelimit.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <li...@databricks.com>
---
Cargo.lock | 979 +++++++++++++++++++++++++++-
Cargo.toml | 2 +
crates/audit/Cargo.toml | 34 +
crates/audit/src/lib.rs | 431 ++++++++++++
crates/ratelimit/Cargo.toml | 38 ++
crates/ratelimit/src/lib.rs | 517 +++++++++++++++
crates/ratelimit/src/redis.rs | 350 ++++++++++
crates/ratelimit/tests/redis_integration.rs | 281 ++++++++
8 files changed, 2617 insertions(+), 15 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 49fc8fb..c436a36 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11,12 +11,36 @@ dependencies = [
"memchr",
]
+[[package]]
+name = "android_system_properties"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc"
+dependencies = [
+ "libc",
+]
+
[[package]]
name = "anyhow"
version = "1.0.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
+[[package]]
+name = "arc-swap"
+version = "1.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b"
+dependencies = [
+ "rustversion",
+]
+
+[[package]]
+name = "arcstr"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d"
+
[[package]]
name = "assert-json-diff"
version = "2.0.2"
@@ -27,6 +51,55 @@ dependencies = [
"serde_json",
]
+[[package]]
+name = "astral-tokio-tar"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b18457efd137254e016bbde5e1d88df61c4e1a5ae2223746e56123bac6af2463"
+dependencies = [
+ "futures-core",
+ "libc",
+ "portable-atomic",
+ "rustc-hash",
+ "rustix",
+ "tokio",
+ "tokio-stream",
+ "xattr",
+]
+
+[[package]]
+name = "async-lock"
+version = "3.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
+dependencies = [
+ "event-listener",
+ "event-listener-strategy",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-stream"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
[[package]]
name = "async-trait"
version = "0.1.91"
@@ -93,24 +166,128 @@ dependencies = [
"tower-service",
]
+[[package]]
+name = "backon"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef"
+dependencies = [
+ "fastrand",
+]
+
[[package]]
name = "base64"
version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+[[package]]
+name = "base64"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
+
[[package]]
name = "base64ct"
version = "1.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06"
+[[package]]
+name = "bitflags"
+version = "1.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+
[[package]]
name = "bitflags"
version = "2.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da"
+[[package]]
+name = "bollard"
+version = "0.20.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee04c4c84f1f811b017f2fbb7dd8815c976e7ca98593de9c1e2afad0f636bff4"
+dependencies = [
+ "async-stream",
+ "base64 0.22.1",
+ "bitflags 2.13.1",
+ "bollard-buildkit-proto",
+ "bollard-stubs",
+ "bytes",
+ "futures-core",
+ "futures-util",
+ "hex",
+ "home",
+ "http",
+ "http-body-util",
+ "hyper",
+ "hyper-named-pipe",
+ "hyper-rustls",
+ "hyper-util",
+ "hyperlocal",
+ "log",
+ "num",
+ "pin-project-lite",
+ "rand 0.9.5",
+ "rustls",
+ "rustls-native-certs",
+ "rustls-pki-types",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "serde_urlencoded",
+ "thiserror 2.0.20",
+ "time",
+ "tokio",
+ "tokio-stream",
+ "tokio-util",
+ "tonic",
+ "tower-service",
+ "url",
+ "winapi",
+]
+
+[[package]]
+name = "bollard-buildkit-proto"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85a885520bf6249ab931a764ffdb87b0ceef48e6e7d807cfdb21b751e086e1ad"
+dependencies = [
+ "prost",
+ "prost-types",
+ "tonic",
+ "tonic-prost",
+ "ureq",
+]
+
+[[package]]
+name = "bollard-stubs"
+version = "1.52.1-rc.29.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f0a8ca8799131c1837d1282c3f81f31e76ceb0ce426e04a7fe1ccee3287c066"
+dependencies = [
+ "base64 0.22.1",
+ "bollard-buildkit-proto",
+ "bytes",
+ "prost",
+ "serde",
+ "serde_json",
+ "serde_repr",
+ "time",
+]
+
+[[package]]
+name = "bs58"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf88ba1141d185c399bee5288d850d63b8369520c1eafc32a0430b5b6c287bf4"
+dependencies = [
+ "tinyvec",
+]
+
[[package]]
name = "bumpalo"
version = "3.20.3"
@@ -162,7 +339,24 @@ version = "0.4.45"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
dependencies = [
+ "iana-time-zone",
"num-traits",
+ "serde",
+ "windows-link",
+]
+
+[[package]]
+name = "combine"
+version = "4.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "memchr",
+ "pin-project-lite",
+ "tokio",
+ "tokio-util",
]
[[package]]
@@ -171,6 +365,22 @@ version = "0.9.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
+[[package]]
+name = "core-foundation"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
+[[package]]
+name = "core-foundation-sys"
+version = "0.8.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
+
[[package]]
name = "cpufeatures"
version = "0.3.0"
@@ -190,6 +400,40 @@ dependencies = [
"typenum",
]
+[[package]]
+name = "darling"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d"
+dependencies = [
+ "darling_core",
+ "darling_macro",
+]
+
+[[package]]
+name = "darling_core"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0"
+dependencies = [
+ "ident_case",
+ "proc-macro2",
+ "quote",
+ "strsim",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "darling_macro"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d"
+dependencies = [
+ "darling_core",
+ "quote",
+ "syn 2.0.119",
+]
+
[[package]]
name = "deadpool"
version = "0.12.3"
@@ -208,6 +452,37 @@ version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b"
+[[package]]
+name = "defmt"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1"
+dependencies = [
+ "bitflags 1.3.2",
+ "defmt-macros",
+]
+
+[[package]]
+name = "defmt-macros"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8"
+dependencies = [
+ "defmt-parser",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "defmt-parser"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e"
+dependencies = [
+ "thiserror 2.0.20",
+]
+
[[package]]
name = "der"
version = "0.7.10"
@@ -224,6 +499,9 @@ name = "deranged"
version = "0.5.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c"
+dependencies = [
+ "serde_core",
+]
[[package]]
name = "digest"
@@ -246,6 +524,23 @@ dependencies = [
"syn 3.0.3",
]
+[[package]]
+name = "docker_credential"
+version = "1.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "29547a1dc60885a552306986316bc9701ba120c1a8db6769fa68691529ad373d"
+dependencies = [
+ "base64 0.22.1",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "dyn-clone"
+version = "1.0.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
+
[[package]]
name = "either"
version = "1.17.0"
@@ -268,12 +563,53 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "etcetera"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de48cc4d1c1d97a20fd819def54b890cadde72ed3ad0c614822a0a433361be96"
+dependencies = [
+ "cfg-if",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "event-listener"
+version = "5.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2"
+dependencies = [
+ "parking",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "event-listener-strategy"
+version = "0.5.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
+dependencies = [
+ "event-listener",
+ "pin-project-lite",
+]
+
[[package]]
name = "fastrand"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223"
+[[package]]
+name = "ferroid"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee93edf3c501f0035bbeffeccfed0b79e14c311f12195ec0e661e114a0f60da4"
+dependencies = [
+ "portable-atomic",
+ "rand 0.10.2",
+ "web-time",
+]
+
[[package]]
name = "find-msvc-tools"
version = "0.1.10"
@@ -456,13 +792,19 @@ dependencies = [
"futures-core",
"futures-sink",
"http",
- "indexmap",
+ "indexmap 2.14.0",
"slab",
"tokio",
"tokio-util",
"tracing",
]
+[[package]]
+name = "hashbrown"
+version = "0.12.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
+
[[package]]
name = "hashbrown"
version = "0.15.5"
@@ -490,6 +832,21 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
+[[package]]
+name = "hex"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
+
+[[package]]
+name = "home"
+version = "0.5.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
[[package]]
name = "http"
version = "1.5.0"
@@ -557,6 +914,20 @@ dependencies = [
"want",
]
+[[package]]
+name = "hyper-named-pipe"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fab3637d6b04a8037af8a266fdf6cf92ea957e8c53981a2bf6136572531025bf"
+dependencies = [
+ "hex",
+ "hyper",
+ "hyper-util",
+ "pin-project-lite",
+ "tokio",
+ "tower-service",
+]
+
[[package]]
name = "hyper-rustls"
version = "0.27.9"
@@ -592,7 +963,7 @@ version = "0.1.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"futures-channel",
"futures-util",
@@ -609,6 +980,45 @@ dependencies = [
"tracing",
]
+[[package]]
+name = "hyperlocal"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "986c5ce3b994526b3cd75578e62554abd09f0899d6206de48b3e96ab34ccc8c7"
+dependencies = [
+ "hex",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
+ "pin-project-lite",
+ "tokio",
+ "tower-service",
+]
+
+[[package]]
+name = "iana-time-zone"
+version = "0.1.65"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470"
+dependencies = [
+ "android_system_properties",
+ "core-foundation-sys",
+ "iana-time-zone-haiku",
+ "js-sys",
+ "log",
+ "wasm-bindgen",
+ "windows-core",
+]
+
+[[package]]
+name = "iana-time-zone-haiku"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f"
+dependencies = [
+ "cc",
+]
+
[[package]]
name = "icu_collections"
version = "2.2.0"
@@ -691,6 +1101,12 @@ dependencies = [
"zerovec",
]
+[[package]]
+name = "ident_case"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
+
[[package]]
name = "idna"
version = "1.1.0"
@@ -712,6 +1128,17 @@ dependencies = [
"icu_properties",
]
+[[package]]
+name = "indexmap"
+version = "1.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99"
+dependencies = [
+ "autocfg",
+ "hashbrown 0.12.3",
+ "serde",
+]
+
[[package]]
name = "indexmap"
version = "2.14.0"
@@ -720,6 +1147,8 @@ checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
dependencies = [
"equivalent",
"hashbrown 0.17.1",
+ "serde",
+ "serde_core",
]
[[package]]
@@ -743,6 +1172,59 @@ version = "1.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
+[[package]]
+name = "jiff"
+version = "0.2.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "668b7183bd07af9a4885f5c35b0cc5c83c4607a913c16b7e17291832910d2dcc"
+dependencies = [
+ "defmt",
+ "jiff-core",
+ "jiff-static",
+ "jiff-tzdb-platform",
+ "log",
+ "portable-atomic",
+ "portable-atomic-util",
+ "serde_core",
+ "windows-link",
+]
+
+[[package]]
+name = "jiff-core"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7feca88439efe53da3754500c1851dedf3cb36c524dd5cf8225cc0794de95d09"
+dependencies = [
+ "defmt",
+]
+
+[[package]]
+name = "jiff-static"
+version = "0.2.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a69dcb3a21cfb32ce1cd056169337ca284af0766dd766e7878819b251a49204"
+dependencies = [
+ "jiff-core",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "jiff-tzdb"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "142bd39932ad231f10513df9ab62661fead8719872150b7ad02a2df79f4e141e"
+
+[[package]]
+name = "jiff-tzdb-platform"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8"
+dependencies = [
+ "jiff-tzdb",
+]
+
[[package]]
name = "js-sys"
version = "0.3.104"
@@ -760,7 +1242,7 @@ version = "9.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a87cc7a48537badeae96744432de36f4be2b4a34a05a5ef32e9dd8a1c169dde"
dependencies = [
- "base64",
+ "base64 0.22.1",
"js-sys",
"pem",
"ring",
@@ -876,6 +1358,20 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "num"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23"
+dependencies = [
+ "num-bigint",
+ "num-complex",
+ "num-integer",
+ "num-iter",
+ "num-rational",
+ "num-traits",
+]
+
[[package]]
name = "num-bigint"
version = "0.4.8"
@@ -902,6 +1398,15 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "num-complex"
+version = "0.4.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495"
+dependencies = [
+ "num-traits",
+]
+
[[package]]
name = "num-conv"
version = "0.2.2"
@@ -927,6 +1432,17 @@ dependencies = [
"num-traits",
]
+[[package]]
+name = "num-rational"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824"
+dependencies = [
+ "num-bigint",
+ "num-integer",
+ "num-traits",
+]
+
[[package]]
name = "num-traits"
version = "0.2.19"
@@ -953,6 +1469,12 @@ version = "1.21.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
+[[package]]
+name = "openssl-probe"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
+
[[package]]
name = "opentelemetry"
version = "0.31.0"
@@ -1030,6 +1552,12 @@ dependencies = [
"tokio-stream",
]
+[[package]]
+name = "parking"
+version = "2.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
+
[[package]]
name = "parking_lot"
version = "0.12.5"
@@ -1053,13 +1581,38 @@ dependencies = [
"windows-link",
]
+[[package]]
+name = "parse-display"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "914a1c2265c98e2446911282c6ac86d8524f495792c38c5bd884f80499c7538a"
+dependencies = [
+ "parse-display-derive",
+ "regex",
+ "regex-syntax",
+]
+
+[[package]]
+name = "parse-display-derive"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2ae7800a4c974efd12df917266338e79a7a74415173caf7e70aa0a0707345281"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "regex",
+ "regex-syntax",
+ "structmeta",
+ "syn 2.0.119",
+]
+
[[package]]
name = "pem"
version = "3.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be"
dependencies = [
- "base64",
+ "base64 0.22.1",
"serde_core",
]
@@ -1086,7 +1639,7 @@ checksum = "8701b58ea97060d5e5b155d383a69952a60943f0e6dfe30b04c287beb0b27455"
dependencies = [
"fixedbitset",
"hashbrown 0.15.5",
- "indexmap",
+ "indexmap 2.14.0",
]
[[package]]
@@ -1136,6 +1689,21 @@ dependencies = [
"spki",
]
+[[package]]
+name = "portable-atomic"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85"
+
+[[package]]
+name = "portable-atomic-util"
+version = "0.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618"
+dependencies = [
+ "portable-atomic",
+]
+
[[package]]
name = "potential_utf"
version = "0.1.5"
@@ -1273,7 +1841,7 @@ version = "0.13.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e9f068eba8e7071c5f9511831b44f32c740d5adf574e990f946ddb53db2f314e"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"memchr",
"unicase",
]
@@ -1449,13 +2017,60 @@ dependencies = [
"rand_core 0.10.1",
]
+[[package]]
+name = "redis"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e37a4ca5c6ca42aa3e6df2fd32b987a65d32a4c2159a6f3fe0fd1df306a2658f"
+dependencies = [
+ "arc-swap",
+ "arcstr",
+ "async-lock",
+ "backon",
+ "bytes",
+ "cfg-if",
+ "combine",
+ "futures-channel",
+ "futures-util",
+ "itoa",
+ "percent-encoding",
+ "pin-project-lite",
+ "ryu",
+ "sha1_smol",
+ "socket2",
+ "tokio",
+ "tokio-util",
+ "url",
+ "xxhash-rust",
+]
+
[[package]]
name = "redox_syscall"
version = "0.5.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
+]
+
+[[package]]
+name = "ref-cast"
+version = "1.0.26"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "216e8f773d7923bcba9ceb86a86c93cabb3903a11872fc3f138c49630e50b96d"
+dependencies = [
+ "ref-cast-impl",
+]
+
+[[package]]
+name = "ref-cast-impl"
+version = "1.0.26"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c9283685feec7d69af75fb0e858d5e7378f33fe4fc699383b2916ab9273e03c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
]
[[package]]
@@ -1493,7 +2108,7 @@ version = "0.12.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"futures-core",
"http",
@@ -1571,7 +2186,7 @@ version = "1.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"errno",
"libc",
"linux-raw-sys",
@@ -1593,6 +2208,18 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "rustls-native-certs"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
+dependencies = [
+ "openssl-probe",
+ "rustls-pki-types",
+ "schannel",
+ "security-framework",
+]
+
[[package]]
name = "rustls-pki-types"
version = "1.15.1"
@@ -1626,12 +2253,21 @@ version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
+[[package]]
+name = "scg-audit"
+version = "0.1.0"
+dependencies = [
+ "parking_lot",
+ "tracing",
+ "tracing-subscriber",
+]
+
[[package]]
name = "scg-auth"
version = "0.1.0"
dependencies = [
"async-trait",
- "base64",
+ "base64 0.22.1",
"jsonwebtoken",
"parking_lot",
"rand 0.8.7",
@@ -1693,6 +2329,19 @@ dependencies = [
"uuid",
]
+[[package]]
+name = "scg-ratelimit"
+version = "0.1.0"
+dependencies = [
+ "parking_lot",
+ "redis",
+ "testcontainers-modules",
+ "thiserror 2.0.20",
+ "tokio",
+ "tonic",
+ "tracing",
+]
+
[[package]]
name = "scg-routing"
version = "0.1.0"
@@ -1703,12 +2352,68 @@ dependencies = [
"tonic",
]
+[[package]]
+name = "schannel"
+version = "0.1.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "schemars"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4cd191f9397d57d581cddd31014772520aa448f65ef991055d7f61582c65165f"
+dependencies = [
+ "dyn-clone",
+ "ref-cast",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "schemars"
+version = "1.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a"
+dependencies = [
+ "dyn-clone",
+ "ref-cast",
+ "serde",
+ "serde_json",
+]
+
[[package]]
name = "scopeguard"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
+[[package]]
+name = "security-framework"
+version = "3.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d"
+dependencies = [
+ "bitflags 2.13.1",
+ "core-foundation",
+ "core-foundation-sys",
+ "libc",
+ "security-framework-sys",
+]
+
+[[package]]
+name = "security-framework-sys"
+version = "2.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
[[package]]
name = "serde"
version = "1.0.229"
@@ -1752,6 +2457,17 @@ dependencies = [
"zmij",
]
+[[package]]
+name = "serde_repr"
+version = "0.1.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8d3b1629de253c70a0508c3899572da79ca359fdab27c7920ff00406df418906"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
[[package]]
name = "serde_urlencoded"
version = "0.7.1"
@@ -1764,19 +2480,58 @@ dependencies = [
"serde",
]
+[[package]]
+name = "serde_with"
+version = "3.22.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee78f1fbe43ac4a0e47aadb3dbd357b69eb0d3793e948624cd03dd2750ab1c0a"
+dependencies = [
+ "base64 0.22.1",
+ "bs58",
+ "chrono",
+ "hex",
+ "indexmap 1.9.3",
+ "indexmap 2.14.0",
+ "jiff",
+ "schemars 0.9.0",
+ "schemars 1.2.2",
+ "serde_core",
+ "serde_json",
+ "serde_with_macros",
+ "time",
+]
+
+[[package]]
+name = "serde_with_macros"
+version = "3.22.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8705578779c2b6bd90d84d66eb2e206b708b1a4d7b9f17641b293545bf1c7e46"
+dependencies = [
+ "darling",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
[[package]]
name = "serde_yaml"
version = "0.9.34+deprecated"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
dependencies = [
- "indexmap",
+ "indexmap 2.14.0",
"itoa",
"ryu",
"serde",
"unsafe-libyaml",
]
+[[package]]
+name = "sha1_smol"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d"
+
[[package]]
name = "sharded-slab"
version = "0.1.7"
@@ -1868,6 +2623,35 @@ version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
+[[package]]
+name = "strsim"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
+
+[[package]]
+name = "structmeta"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e1575d8d40908d70f6fd05537266b90ae71b15dbbe7a8b7dffa2b759306d329"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "structmeta-derive",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "structmeta-derive"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "152a0b65a590ff6c3da95cabe2353ee04e6167c896b28e3b14478c2636c922fc"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
[[package]]
name = "subtle"
version = "2.6.1"
@@ -1929,6 +2713,46 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "testcontainers"
+version = "0.27.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfd5785b5483672915ed5fe3cddf9f546802779fc1eceff0a6fb7321fac81c1e"
+dependencies = [
+ "astral-tokio-tar",
+ "async-trait",
+ "bollard",
+ "bytes",
+ "docker_credential",
+ "either",
+ "etcetera",
+ "ferroid",
+ "futures",
+ "http",
+ "itertools",
+ "log",
+ "memchr",
+ "parse-display",
+ "pin-project-lite",
+ "serde",
+ "serde_json",
+ "serde_with",
+ "thiserror 2.0.20",
+ "tokio",
+ "tokio-stream",
+ "tokio-util",
+ "url",
+]
+
+[[package]]
+name = "testcontainers-modules"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5985fde5befe4ffa77a052e035e16c2da86e8bae301baa9f9904ad3c494d357"
+dependencies = [
+ "testcontainers",
+]
+
[[package]]
name = "thiserror"
version = "1.0.69"
@@ -2104,7 +2928,7 @@ checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
dependencies = [
"async-trait",
"axum",
- "base64",
+ "base64 0.22.1",
"bytes",
"h2",
"http",
@@ -2174,7 +2998,7 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
dependencies = [
"futures-core",
"futures-util",
- "indexmap",
+ "indexmap 2.14.0",
"pin-project-lite",
"slab",
"sync_wrapper",
@@ -2191,7 +3015,7 @@ version = "0.6.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"bytes",
"futures-util",
"http",
@@ -2341,6 +3165,33 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
+[[package]]
+name = "ureq"
+version = "3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "972d7902c8735f2695410b8aed7df6ed12a47394aa1c8d7af49f0497b731a94d"
+dependencies = [
+ "base64 0.23.1",
+ "log",
+ "percent-encoding",
+ "rustls",
+ "rustls-pki-types",
+ "ureq-proto",
+ "utf8-zero",
+]
+
+[[package]]
+name = "ureq-proto"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da5f78b09e6941e1a0f2e30e695e4b120377b54d5e0aec11b594bb57b3971613"
+dependencies = [
+ "base64 0.23.1",
+ "http",
+ "httparse",
+ "log",
+]
+
[[package]]
name = "url"
version = "2.5.8"
@@ -2351,8 +3202,15 @@ dependencies = [
"idna",
"percent-encoding",
"serde",
+ "serde_derive",
]
+[[package]]
+name = "utf8-zero"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e"
+
[[package]]
name = "utf8_iter"
version = "1.0.4"
@@ -2490,12 +3348,87 @@ dependencies = [
"rustls-pki-types",
]
+[[package]]
+name = "winapi"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
+dependencies = [
+ "winapi-i686-pc-windows-gnu",
+ "winapi-x86_64-pc-windows-gnu",
+]
+
+[[package]]
+name = "winapi-i686-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
+
+[[package]]
+name = "winapi-x86_64-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+
+[[package]]
+name = "windows-core"
+version = "0.62.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
+dependencies = [
+ "windows-implement",
+ "windows-interface",
+ "windows-link",
+ "windows-result",
+ "windows-strings",
+]
+
+[[package]]
+name = "windows-implement"
+version = "0.60.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "windows-interface"
+version = "0.59.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
[[package]]
name = "windows-link"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
+[[package]]
+name = "windows-result"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
+dependencies = [
+ "windows-link",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
+dependencies = [
+ "windows-link",
+]
+
[[package]]
name = "windows-sys"
version = "0.52.0"
@@ -2585,7 +3518,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031"
dependencies = [
"assert-json-diff",
- "base64",
+ "base64 0.22.1",
"deadpool",
"futures",
"http",
@@ -2613,6 +3546,22 @@ version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
+[[package]]
+name = "xattr"
+version = "1.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156"
+dependencies = [
+ "libc",
+ "rustix",
+]
+
+[[package]]
+name = "xxhash-rust"
+version = "0.8.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aee1b19627c7c60102ab80d3a9cbe18de90bfe03bfa6c3715447681f0e8c8af6"
+
[[package]]
name = "yoke"
version = "0.8.3"
diff --git a/Cargo.toml b/Cargo.toml
index 754343f..066345a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -21,6 +21,8 @@ members = [
"crates/config",
"crates/auth",
"crates/observability",
+ "crates/audit",
+ "crates/ratelimit",
]
[workspace.package]
diff --git a/crates/audit/Cargo.toml b/crates/audit/Cargo.toml
new file mode 100644
index 0000000..594a84a
--- /dev/null
+++ b/crates/audit/Cargo.toml
@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[package]
+name = "scg-audit"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+repository.workspace = true
+readme.workspace = true
+keywords.workspace = true
+categories.workspace = true
+description = "Structured audit event emitter for the Spark Connect Gateway, on top of tracing."
+publish = false
+
+[dependencies]
+tracing = { workspace = true }
+
+[dev-dependencies]
+tracing-subscriber = { workspace = true }
+parking_lot = { workspace = true }
diff --git a/crates/audit/src/lib.rs b/crates/audit/src/lib.rs
new file mode 100644
index 0000000..4a2a05d
--- /dev/null
+++ b/crates/audit/src/lib.rs
@@ -0,0 +1,431 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Structured audit logging for the Spark Connect Gateway.
+//!
+//! Records security- and compliance-relevant events as
+//! `tracing::info!` events with `target = "scg::audit"`. The
+//! existing JSON formatter installed by `scg-observability` picks
+//! them up automatically; operators filter by target in their log
+//! aggregator (Loki, Splunk, etc.) to get an auditable stream
+//! distinct from operational logs.
+//!
+//! ## Event taxonomy
+//!
+//! Four event types, deliberately narrow:
+//!
+//! | Event | When |
+//! |-------|------|
+//! | `session.create` | First time a `(tenant, user, session_id)` is bound to a backend |
+//! | `session.release` | Client called `ReleaseSession` and the gateway forgot the binding |
+//! | `auth.failure` | Authentication rejected an RPC (any of the failure reasons from `scg-auth`) |
+//! | `rpc.error` | Handler returned a non-OK `Status` — surfaced separately so compliance can see failures without mining every log line |
+//!
+//! Successful RPCs are deliberately *not* logged by default —
+//! they're already counted in `scg_rpcs_total{code="OK"}` and
+//! filling the audit stream with every Config call defeats the
+//! purpose. The optional `log_successful_rpcs` config switch turns
+//! them on for strict-monitoring environments.
+//!
+//! ## Why not a sink trait
+//!
+//! We pipe through `tracing` rather than introducing a separate
+//! "AuditSink" abstraction because:
+//! * The structured-log infrastructure (JSON formatter, levels,
+//! field types) already exists.
+//! * Operators already have one log pipeline; one fewer thing to
+//! route.
+//! * Filtering by `target` is a one-line query in Loki/Splunk.
+//!
+//! If we ever need a dedicated sink (file, Kafka, S3) we can add
+//! a `tracing_subscriber::Layer` that intercepts `target =
+//! "scg::audit"` events without changing this API.
+
+use tracing::info;
+
+/// Audit-logging configuration. Constructed from `scg-config`'s
+/// `AuditSettings`.
+#[derive(Debug, Clone)]
+pub struct AuditConfig {
+ /// Master switch. Defaults `true` — compliance value is high
+ /// and the per-event cost (one `tracing::info!`) is negligible.
+ /// Operators can flip to `false` in dev / local environments.
+ pub enabled: bool,
+ /// When `true`, every successful RPC also emits an `rpc.ok`
+ /// audit event. Defaults `false` to keep the audit stream
+ /// signal-rich; only switch on under strict monitoring.
+ pub log_successful_rpcs: bool,
+}
+
+impl Default for AuditConfig {
+ fn default() -> Self {
+ Self {
+ enabled: true,
+ log_successful_rpcs: false,
+ }
+ }
+}
+
+/// Audit logger. Cheap to clone (config is small + `Copy`-like).
+#[derive(Debug, Clone)]
+pub struct AuditLogger {
+ cfg: AuditConfig,
+}
+
+impl AuditLogger {
+ pub fn new(cfg: AuditConfig) -> Self {
+ Self { cfg }
+ }
+
+ /// Build a no-op logger (`enabled = false`). Used by tests
+ /// that don't care about audit output, and as the proxy's
+ /// default when the operator hasn't configured audit.
+ pub fn disabled() -> Self {
+ Self::new(AuditConfig {
+ enabled: false,
+ log_successful_rpcs: false,
+ })
+ }
+
+ /// `session.create` — a `(tenant, user, session_id)` was bound
+ /// to `backend` for the first time. Fires on the binding-path
+ /// of the affinity store (not on every lookup).
+ ///
+ /// `groups` is the verified identity's group membership (from
+ /// the JWT/OIDC `groups` claim or the static-token config).
+ /// Recorded as a comma-joined string so the audit stream stays
+ /// flat JSON; downstream tooling can split if it needs an array.
+ /// Per-RPC events deliberately *don't* carry groups — they fire
+ /// frequently enough that doubling the field count matters.
+ pub fn session_create(
+ &self,
+ rid: &str,
+ tenant: &str,
+ user_id: &str,
+ groups: &[String],
+ session_id: &str,
+ backend: &str,
+ ) {
+ if !self.cfg.enabled {
+ return;
+ }
+ info!(
+ target: "scg::audit",
+ event = "session.create",
+ rid = %rid,
+ tenant = %tenant,
+ user_id = %user_id,
+ groups = %groups.join(","),
+ session_id = %session_id,
+ backend = %backend,
+ "session created",
+ );
+ }
+
+ /// `session.release` — the client called `ReleaseSession`. The
+ /// gateway has dropped its affinity record; the backend has
+ /// dropped the SparkSession.
+ pub fn session_release(
+ &self,
+ rid: &str,
+ tenant: &str,
+ user_id: &str,
+ groups: &[String],
+ session_id: &str,
+ backend: &str,
+ ) {
+ if !self.cfg.enabled {
+ return;
+ }
+ info!(
+ target: "scg::audit",
+ event = "session.release",
+ rid = %rid,
+ tenant = %tenant,
+ user_id = %user_id,
+ groups = %groups.join(","),
+ session_id = %session_id,
+ backend = %backend,
+ "session released",
+ );
+ }
+
+ /// `auth.failure` — the auth interceptor rejected an RPC.
+ /// `reason` is the same fixed-cardinality string used by the
+ /// metric (`missing_token`, `invalid_token`, `expired`,
+ /// `unknown_kid`, `unknown`).
+ pub fn auth_failure(&self, rid: &str, rpc: &str, reason: &str) {
+ if !self.cfg.enabled {
+ return;
+ }
+ info!(
+ target: "scg::audit",
+ event = "auth.failure",
+ rid = %rid,
+ rpc = %rpc,
+ reason = %reason,
+ "authentication failed",
+ );
+ }
+
+ /// `config.redacted` — a client asked for a config key the
+ /// gateway refuses to relay (currently the backend's own
+ /// pre-shared token) and the value was stripped from the
+ /// response. Security-relevant on its own: a legitimate client
+ /// has no reason to read the gateway↔backend credential, so
+ /// these events are worth alerting on.
+ pub fn config_redacted(&self, rid: &str, tenant: &str, user_id: &str, key: &str) {
+ if !self.cfg.enabled {
+ return;
+ }
+ info!(
+ target: "scg::audit",
+ event = "config.redacted",
+ rid = %rid,
+ tenant = %tenant,
+ user_id = %user_id,
+ key = %key,
+ "config key withheld from client",
+ );
+ }
+
+ /// `rpc.error` — a handler returned a non-OK Status. `code` is
+ /// the canonical gRPC code name (e.g. `Unauthenticated`,
+ /// `PermissionDenied`, `ResourceExhausted`). Cancelled and
+ /// OK results are filtered by the caller.
+ pub fn rpc_error(
+ &self,
+ rid: &str,
+ rpc: &str,
+ tenant: &str,
+ user_id: &str,
+ code: &str,
+ message: &str,
+ ) {
+ if !self.cfg.enabled {
+ return;
+ }
+ info!(
+ target: "scg::audit",
+ event = "rpc.error",
+ rid = %rid,
+ rpc = %rpc,
+ tenant = %tenant,
+ user_id = %user_id,
+ code = %code,
+ message = %message,
+ "rpc returned non-OK status",
+ );
+ }
+
+ /// `rpc.ok` — optional. Only emitted when
+ /// `log_successful_rpcs = true`. Off by default to keep the
+ /// audit stream signal-rich.
+ pub fn rpc_ok(&self, rid: &str, rpc: &str, tenant: &str, user_id: &str) {
+ if !self.cfg.enabled || !self.cfg.log_successful_rpcs {
+ return;
+ }
+ info!(
+ target: "scg::audit",
+ event = "rpc.ok",
+ rid = %rid,
+ rpc = %rpc,
+ tenant = %tenant,
+ user_id = %user_id,
+ "rpc completed ok",
+ );
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use parking_lot::Mutex;
+ use std::sync::Arc;
+ use tracing::subscriber::with_default;
+ use tracing_subscriber::layer::SubscriberExt;
+ use tracing_subscriber::Layer;
+
+ /// Layer that captures all events with `target = "scg::audit"`
+ /// into a Vec. Used to verify the audit API actually emits
+ /// what it promises without needing to install the global JSON
+ /// formatter.
+ #[derive(Clone, Default)]
+ struct CaptureLayer {
+ events: Arc<Mutex<Vec<CapturedEvent>>>,
+ }
+
+ #[derive(Debug)]
+ struct CapturedEvent {
+ target: String,
+ fields: std::collections::HashMap<String, String>,
+ }
+
+ impl<S: tracing::Subscriber> Layer<S> for CaptureLayer {
+ fn on_event(
+ &self,
+ event: &tracing::Event<'_>,
+ _ctx: tracing_subscriber::layer::Context<'_, S>,
+ ) {
+ let target = event.metadata().target().to_string();
+ let mut fields = std::collections::HashMap::new();
+ struct Vis<'a>(&'a mut std::collections::HashMap<String, String>);
+ impl<'a> tracing::field::Visit for Vis<'a> {
+ fn record_debug(
+ &mut self,
+ field: &tracing::field::Field,
+ value: &dyn std::fmt::Debug,
+ ) {
+ self.0.insert(field.name().into(), format!("{:?}", value));
+ }
+ fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+ self.0.insert(field.name().into(), value.into());
+ }
+ }
+ event.record(&mut Vis(&mut fields));
+ self.events.lock().push(CapturedEvent { target, fields });
+ }
+ }
+
+ fn with_capture<F: FnOnce(&CaptureLayer)>(f: F) {
+ let cap = CaptureLayer::default();
+ let sub = tracing_subscriber::registry().with(cap.clone());
+ with_default(sub, || f(&cap));
+ }
+
+ #[test]
+ fn disabled_logger_emits_nothing() {
+ with_capture(|cap| {
+ let a = AuditLogger::disabled();
+ a.session_create("rid-1", "t", "u", &[], "s", "b:1");
+ a.auth_failure("rid-2", "Config", "missing_token");
+ a.rpc_error("rid-3", "Config", "t", "u", "Internal", "boom");
+ assert!(cap.events.lock().is_empty());
+ });
+ }
+
+ #[test]
+ fn session_create_emits_expected_fields() {
+ with_capture(|cap| {
+ let a = AuditLogger::new(AuditConfig::default());
+ a.session_create(
+ "rid-1",
+ "team-a",
+ "alice",
+ &["devs".into(), "admins".into()],
+ "sess-1",
+ "be:15002",
+ );
+ let events = cap.events.lock();
+ assert_eq!(events.len(), 1);
+ let e = &events[0];
+ assert_eq!(e.target, "scg::audit");
+ assert_eq!(
+ e.fields.get("event").map(|s| s.as_str()),
+ Some("session.create")
+ );
+ assert_eq!(e.fields.get("tenant").map(|s| s.as_str()), Some("team-a"));
+ assert_eq!(e.fields.get("user_id").map(|s| s.as_str()), Some("alice"));
+ assert_eq!(
+ e.fields.get("session_id").map(|s| s.as_str()),
+ Some("sess-1")
+ );
+ assert_eq!(
+ e.fields.get("backend").map(|s| s.as_str()),
+ Some("be:15002")
+ );
+ assert_eq!(
+ e.fields.get("groups").map(|s| s.as_str()),
+ Some("devs,admins"),
+ );
+ });
+ }
+
+ #[test]
+ fn session_create_empty_groups_renders_empty_string() {
+ with_capture(|cap| {
+ let a = AuditLogger::new(AuditConfig::default());
+ a.session_create("rid-1", "team-a", "alice", &[], "sess-1", "be:15002");
+ let events = cap.events.lock();
+ assert_eq!(events.len(), 1);
+ assert_eq!(events[0].fields.get("groups").map(|s| s.as_str()), Some(""));
+ });
+ }
+
+ #[test]
+ fn auth_failure_emits_reason() {
+ with_capture(|cap| {
+ let a = AuditLogger::new(AuditConfig::default());
+ a.auth_failure("rid-1", "Config", "invalid_token");
+ let events = cap.events.lock();
+ assert_eq!(events.len(), 1);
+ let e = &events[0];
+ assert_eq!(
+ e.fields.get("event").map(|s| s.as_str()),
+ Some("auth.failure")
+ );
+ assert_eq!(
+ e.fields.get("reason").map(|s| s.as_str()),
+ Some("invalid_token")
+ );
+ });
+ }
+
+ #[test]
+ fn rpc_error_emits_code_and_message() {
+ with_capture(|cap| {
+ let a = AuditLogger::new(AuditConfig::default());
+ a.rpc_error(
+ "rid-1",
+ "ExecutePlan",
+ "team-b",
+ "bob",
+ "ResourceExhausted",
+ "rate limit exceeded",
+ );
+ let events = cap.events.lock();
+ assert_eq!(events.len(), 1);
+ let e = &events[0];
+ assert_eq!(e.fields.get("event").map(|s| s.as_str()), Some("rpc.error"));
+ assert_eq!(
+ e.fields.get("code").map(|s| s.as_str()),
+ Some("ResourceExhausted")
+ );
+ });
+ }
+
+ #[test]
+ fn rpc_ok_only_fires_when_enabled() {
+ with_capture(|cap| {
+ // Default config: log_successful_rpcs is false.
+ let a = AuditLogger::new(AuditConfig::default());
+ a.rpc_ok("rid-1", "Config", "t", "u");
+ assert!(cap.events.lock().is_empty());
+
+ // Opt in.
+ let a = AuditLogger::new(AuditConfig {
+ enabled: true,
+ log_successful_rpcs: true,
+ });
+ a.rpc_ok("rid-2", "Config", "t", "u");
+ let events = cap.events.lock();
+ assert_eq!(events.len(), 1);
+ assert_eq!(
+ events[0].fields.get("event").map(|s| s.as_str()),
+ Some("rpc.ok")
+ );
+ });
+ }
+}
diff --git a/crates/ratelimit/Cargo.toml b/crates/ratelimit/Cargo.toml
new file mode 100644
index 0000000..5c5375b
--- /dev/null
+++ b/crates/ratelimit/Cargo.toml
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[package]
+name = "scg-ratelimit"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+repository.workspace = true
+readme.workspace = true
+keywords.workspace = true
+categories.workspace = true
+description = "Per-tenant token-bucket rate limiting (in-memory or Redis-shared) for the Spark Connect Gateway."
+publish = false
+
+[dependencies]
+parking_lot = { workspace = true }
+redis = { workspace = true, features = ["script"] }
+tokio = { workspace = true }
+tonic = { workspace = true }
+tracing = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+testcontainers-modules = { workspace = true }
diff --git a/crates/ratelimit/src/lib.rs b/crates/ratelimit/src/lib.rs
new file mode 100644
index 0000000..326f654
--- /dev/null
+++ b/crates/ratelimit/src/lib.rs
@@ -0,0 +1,517 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Per-tenant (+ optional per-user) token-bucket rate limiting.
+//!
+//! Two backends:
+//!
+//! * **In-memory** — fine for single-replica deployments or
+//! multi-replica setups behind a sticky LB. Each gateway replica
+//! enforces its own bucket; effective quota is
+//! `N × configured_rate` for an N-replica deployment.
+//! * **Redis** — atomic token bucket via a Lua script, shared
+//! across all gateway replicas. The quota is enforced
+//! cluster-wide. See [`redis`] for the wire format and the Lua
+//! contract.
+//!
+//! ## Why token bucket
+//!
+//! Spark Connect traffic is bursty by nature: a client opens a
+//! session, fires a quick `Config` + `AnalyzePlan` pair, then waits
+//! while a long `ExecutePlan` stream runs. A leaky bucket would
+//! penalize that warm-up burst; a fixed-window counter would let
+//! you double-burst across the window boundary. Token bucket gives
+//! you the average-rate guarantee you want for fair sharing while
+//! still tolerating the natural burstiness.
+//!
+//! ## Bucket scopes
+//!
+//! Two buckets per (tenant, user) pair:
+//!
+//! * **Tenant bucket** — every RPC for the tenant consumes one
+//! token. This is the primary defence against a single tenant
+//! overwhelming the shared backends.
+//! * **User bucket** (optional, off by default) — every RPC for
+//! the specific user inside the tenant also consumes one token.
+//! Lets a tenant limit any one user inside it.
+//!
+//! Both buckets must have a token available for the RPC to
+//! proceed. The user bucket is only consulted when its rate is
+//! configured > 0; otherwise it's a no-op.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+use parking_lot::Mutex;
+use tonic::Status;
+use tracing::debug;
+
+/// Configuration for one bucket (tenant or user). Rates are in
+/// RPCs/second. `burst` is the bucket capacity — the maximum number
+/// of consecutive RPCs before the limiter kicks in.
+///
+/// A rate of `0` disables the bucket entirely (every RPC is
+/// admitted without consulting it). This is how operators opt out
+/// of the per-user dimension without touching the per-tenant
+/// dimension.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct BucketRate {
+ pub rpcs_per_second: f64,
+ pub burst: u64,
+}
+
+impl BucketRate {
+ pub fn disabled() -> Self {
+ Self {
+ rpcs_per_second: 0.0,
+ burst: 0,
+ }
+ }
+ pub(crate) fn is_enabled(&self) -> bool {
+ self.rpcs_per_second > 0.0 && self.burst > 0
+ }
+}
+
+/// Per-tenant rate-limit config — one for the per-tenant bucket,
+/// one for the per-user bucket. Per-user defaults disabled so
+/// operators can turn on tenant-level limits without immediately
+/// having to decide a sensible per-user value.
+#[derive(Debug, Clone, Copy)]
+pub struct TenantLimits {
+ pub tenant: BucketRate,
+ pub per_user: BucketRate,
+}
+
+impl Default for TenantLimits {
+ fn default() -> Self {
+ Self {
+ tenant: BucketRate::disabled(),
+ per_user: BucketRate::disabled(),
+ }
+ }
+}
+
+/// Reasons a [`RateLimiter`] might reject an RPC. Used by callers
+/// that want to label metrics or log details — the `Status`
+/// returned by [`RateLimiter::check`] is always
+/// `RESOURCE_EXHAUSTED`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RejectScope {
+ Tenant,
+ User,
+}
+
+impl RejectScope {
+ pub fn as_str(&self) -> &'static str {
+ match self {
+ RejectScope::Tenant => "tenant",
+ RejectScope::User => "user",
+ }
+ }
+}
+
+/// Observer trait so the proxy can wire metrics in without
+/// `scg-ratelimit` knowing about `scg-observability`. The metrics
+/// crate registers a counter; this trait's `on_reject` callback
+/// bumps it.
+pub trait LimiterObserver: Send + Sync + 'static {
+ fn on_reject(&self, tenant: &str, scope: RejectScope);
+}
+
+/// No-op observer used by the in-crate unit tests. Production
+/// callers (gateway main) always wire a metrics-backed observer.
+#[cfg(test)]
+struct NoopObserver;
+#[cfg(test)]
+impl LimiterObserver for NoopObserver {
+ fn on_reject(&self, _tenant: &str, _scope: RejectScope) {}
+}
+
+/// One token-bucket instance. Tokens refill continuously at
+/// `rate_per_second` up to `capacity`. Calls to [`Self::try_take`]
+/// account for elapsed wall-clock time since the last call.
+///
+/// The bucket holds *fractional* tokens (`f64`) so a low refill rate
+/// (say 0.5 RPS) still lets through one RPC every 2 seconds rather
+/// than rounding to zero forever.
+struct Bucket {
+ rate_per_second: f64,
+ capacity: f64,
+ tokens: f64,
+ last_refill: Instant,
+}
+
+impl Bucket {
+ fn new(cfg: BucketRate) -> Self {
+ Self {
+ rate_per_second: cfg.rpcs_per_second,
+ capacity: cfg.burst as f64,
+ // Start full so a fresh tenant gets its full burst.
+ tokens: cfg.burst as f64,
+ last_refill: Instant::now(),
+ }
+ }
+
+ fn try_take(&mut self) -> bool {
+ let now = Instant::now();
+ let elapsed = now.duration_since(self.last_refill).as_secs_f64();
+ self.tokens = (self.tokens + elapsed * self.rate_per_second).min(self.capacity);
+ self.last_refill = now;
+
+ if self.tokens >= 1.0 {
+ self.tokens -= 1.0;
+ true
+ } else {
+ false
+ }
+ }
+}
+
+/// Cluster-wide observer hook for Redis-related errors. Fail-open
+/// deployments use this to bump `scg_rate_limit_redis_errors_total`
+/// — fail-closed deployments still bump it before returning
+/// `ResourceExhausted`.
+pub trait RedisErrorObserver: Send + Sync + 'static {
+ fn on_redis_error(&self, tenant: &str, reason: &'static str);
+}
+
+/// No-op variant used when redis errors aren't being counted.
+pub struct NoopRedisErrorObserver;
+impl RedisErrorObserver for NoopRedisErrorObserver {
+ fn on_redis_error(&self, _: &str, _: &'static str) {}
+}
+
+/// Behaviour when the Redis backend is unreachable.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FailMode {
+ /// Admit the RPC. Recommended default — availability over
+ /// strict-quota enforcement, mirrors the Redis affinity-store's
+ /// fail-soft behaviour. The error metric still fires so
+ /// operators can see the outage in real time.
+ Open,
+ /// Reject the RPC with `ResourceExhausted`. Pick this for
+ /// strict-SaaS isolation policies where a Redis outage must not
+ /// become a quota-bypass attack vector. Note: this makes Redis
+ /// a hard dependency of the request path.
+ Closed,
+}
+
+/// Public rate-limiter handle — an enum so the same proxy call site
+/// works for both backends. Cheap to clone (each variant is behind
+/// `Arc`).
+#[derive(Clone)]
+pub enum RateLimiter {
+ Memory(MemoryLimiter),
+ Redis(redis::RedisLimiter),
+}
+
+impl RateLimiter {
+ /// Convenience: build the in-memory limiter.
+ pub fn new(
+ default: TenantLimits,
+ overrides: HashMap<String, TenantLimits>,
+ observer: Arc<dyn LimiterObserver>,
+ ) -> Self {
+ Self::Memory(MemoryLimiter::new(default, overrides, observer))
+ }
+
+ /// True when *some* bucket is enabled — caller can skip the
+ /// check-and-acquire dance when it's not.
+ pub fn is_active(&self) -> bool {
+ match self {
+ Self::Memory(m) => m.is_active(),
+ Self::Redis(r) => r.is_active(),
+ }
+ }
+
+ /// Take one token from the (tenant, user) bucket pair. Returns
+ /// `Ok(())` when both buckets had tokens available, or
+ /// `Err(Status::ResourceExhausted)` when either was empty.
+ ///
+ /// On rejection the observer is notified with the *first* scope
+ /// that failed — typically the more-restrictive of the two —
+ /// so metrics show which dimension is the bottleneck.
+ pub async fn check(&self, tenant: &str, user: &str) -> Result<(), Status> {
+ match self {
+ Self::Memory(m) => m.check(tenant, user),
+ Self::Redis(r) => r.check(tenant, user).await,
+ }
+ }
+}
+
+/// In-memory rate limiter. Cheap to clone (everything's behind an
+/// `Arc`). Operator-supplied config is fixed for the lifetime of
+/// the process.
+#[derive(Clone)]
+pub struct MemoryLimiter {
+ inner: Arc<MemoryLimiterInner>,
+}
+
+struct MemoryLimiterInner {
+ /// Per-tenant config: explicit overrides keyed by tenant name,
+ /// plus a `default` fallback applied to any tenant not listed.
+ overrides: HashMap<String, TenantLimits>,
+ default: TenantLimits,
+ /// Live state: one tenant bucket + per-user bucket map per tenant.
+ state: Mutex<HashMap<String, TenantState>>,
+ observer: Arc<dyn LimiterObserver>,
+}
+
+struct TenantState {
+ tenant_bucket: Bucket,
+ user_buckets: HashMap<String, Bucket>,
+}
+
+impl MemoryLimiter {
+ /// Build an in-memory limiter with the given default + overrides.
+ /// Both buckets being disabled (the default) is a no-op
+ /// limiter; callers can keep the limiter wired into the proxy
+ /// without overhead.
+ pub fn new(
+ default: TenantLimits,
+ overrides: HashMap<String, TenantLimits>,
+ observer: Arc<dyn LimiterObserver>,
+ ) -> Self {
+ Self {
+ inner: Arc::new(MemoryLimiterInner {
+ overrides,
+ default,
+ state: Mutex::new(HashMap::new()),
+ observer,
+ }),
+ }
+ }
+
+ pub fn is_active(&self) -> bool {
+ self.inner.default.tenant.is_enabled()
+ || self.inner.default.per_user.is_enabled()
+ || self
+ .inner
+ .overrides
+ .values()
+ .any(|l| l.tenant.is_enabled() || l.per_user.is_enabled())
+ }
+
+ pub fn check(&self, tenant: &str, user: &str) -> Result<(), Status> {
+ let limits = self
+ .inner
+ .overrides
+ .get(tenant)
+ .copied()
+ .unwrap_or(self.inner.default);
+
+ // Fast path: both buckets disabled for this tenant.
+ if !limits.tenant.is_enabled() && !limits.per_user.is_enabled() {
+ return Ok(());
+ }
+
+ let mut state = self.inner.state.lock();
+ let entry = state
+ .entry(tenant.to_string())
+ .or_insert_with(|| TenantState {
+ tenant_bucket: Bucket::new(limits.tenant),
+ user_buckets: HashMap::new(),
+ });
+
+ // Check the tenant bucket first — a tenant-level violation
+ // is the more interesting metric for an operator (it means
+ // the entire tenant is hot, not just one user inside it).
+ if limits.tenant.is_enabled() && !entry.tenant_bucket.try_take() {
+ drop(state);
+ debug!(%tenant, %user, "rate_limit: rejected at tenant scope");
+ self.inner.observer.on_reject(tenant, RejectScope::Tenant);
+ return Err(Status::resource_exhausted(format!(
+ "tenant {:?} rate limit exceeded",
+ tenant
+ )));
+ }
+
+ if limits.per_user.is_enabled() {
+ let user_bucket = entry
+ .user_buckets
+ .entry(user.to_string())
+ .or_insert_with(|| Bucket::new(limits.per_user));
+ if !user_bucket.try_take() {
+ drop(state);
+ debug!(%tenant, %user, "rate_limit: rejected at user scope");
+ self.inner.observer.on_reject(tenant, RejectScope::User);
+ return Err(Status::resource_exhausted(format!(
+ "user {:?} rate limit exceeded (tenant {:?})",
+ user, tenant
+ )));
+ }
+ }
+
+ Ok(())
+ }
+}
+
+pub mod redis;
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::time::Duration;
+
+ fn limits(rps: f64, burst: u64) -> TenantLimits {
+ TenantLimits {
+ tenant: BucketRate {
+ rpcs_per_second: rps,
+ burst,
+ },
+ per_user: BucketRate::disabled(),
+ }
+ }
+
+ fn limits_with_user(t_rps: f64, t_burst: u64, u_rps: f64, u_burst: u64) -> TenantLimits {
+ TenantLimits {
+ tenant: BucketRate {
+ rpcs_per_second: t_rps,
+ burst: t_burst,
+ },
+ per_user: BucketRate {
+ rpcs_per_second: u_rps,
+ burst: u_burst,
+ },
+ }
+ }
+
+ #[test]
+ fn disabled_limiter_is_inactive_and_admits_everything() {
+ let r = MemoryLimiter::new(
+ TenantLimits::default(),
+ HashMap::new(),
+ Arc::new(NoopObserver),
+ );
+ assert!(!r.is_active());
+ for _ in 0..1000 {
+ r.check("any-tenant", "any-user").unwrap();
+ }
+ }
+
+ #[test]
+ fn burst_allows_consecutive_then_rejects() {
+ let r = MemoryLimiter::new(limits(1.0, 5), HashMap::new(), Arc::new(NoopObserver));
+ assert!(r.is_active());
+ // 5 in a row should succeed (full burst).
+ for _ in 0..5 {
+ r.check("t", "u").unwrap();
+ }
+ // The 6th immediately fails — refill hasn't had time.
+ let err = r.check("t", "u").unwrap_err();
+ assert_eq!(err.code(), tonic::Code::ResourceExhausted);
+ }
+
+ #[test]
+ fn refill_lets_more_rpcs_through_after_a_wait() {
+ // 10 RPS, burst 2. After ~250ms we should have ~2.5 tokens.
+ let r = MemoryLimiter::new(limits(10.0, 2), HashMap::new(), Arc::new(NoopObserver));
+ r.check("t", "u").unwrap();
+ r.check("t", "u").unwrap();
+ // Bucket empty.
+ assert!(r.check("t", "u").is_err());
+ std::thread::sleep(Duration::from_millis(300));
+ // ~3 tokens refilled, capped at burst=2. At least one
+ // succeeds; the second probably does too (clock noise can
+ // make it just under 2.0).
+ assert!(r.check("t", "u").is_ok());
+ }
+
+ #[test]
+ fn override_takes_precedence_over_default() {
+ // Default very restrictive; override generous.
+ let mut overrides = HashMap::new();
+ overrides.insert("team-a".to_string(), limits(100.0, 100));
+ let r = MemoryLimiter::new(limits(1.0, 1), overrides, Arc::new(NoopObserver));
+
+ // Default tenant: only 1 RPC before reject.
+ r.check("anyone-else", "u").unwrap();
+ assert!(r.check("anyone-else", "u").is_err());
+
+ // team-a: 50 RPCs no problem.
+ for _ in 0..50 {
+ r.check("team-a", "u").unwrap();
+ }
+ }
+
+ #[test]
+ fn tenant_buckets_are_independent_across_tenants() {
+ let r = MemoryLimiter::new(limits(1.0, 2), HashMap::new(), Arc::new(NoopObserver));
+ // Exhaust team-a's bucket.
+ r.check("team-a", "u").unwrap();
+ r.check("team-a", "u").unwrap();
+ assert!(r.check("team-a", "u").is_err());
+ // team-b has its own, fresh bucket.
+ r.check("team-b", "u").unwrap();
+ r.check("team-b", "u").unwrap();
+ }
+
+ #[test]
+ fn per_user_bucket_protects_one_user_without_blocking_others() {
+ // Tenant bucket generous, per-user bucket tight: 2 burst.
+ let r = MemoryLimiter::new(
+ limits_with_user(100.0, 100, 1.0, 2),
+ HashMap::new(),
+ Arc::new(NoopObserver),
+ );
+ r.check("t", "alice").unwrap();
+ r.check("t", "alice").unwrap();
+ let err = r.check("t", "alice").unwrap_err();
+ assert_eq!(err.code(), tonic::Code::ResourceExhausted);
+ // bob is unaffected.
+ r.check("t", "bob").unwrap();
+ r.check("t", "bob").unwrap();
+ }
+
+ #[test]
+ fn observer_is_notified_with_scope() {
+ use std::sync::atomic::{AtomicU64, Ordering};
+ struct Counter {
+ tenant_rejects: AtomicU64,
+ user_rejects: AtomicU64,
+ }
+ impl LimiterObserver for Counter {
+ fn on_reject(&self, _tenant: &str, scope: RejectScope) {
+ match scope {
+ RejectScope::Tenant => self.tenant_rejects.fetch_add(1, Ordering::Relaxed),
+ RejectScope::User => self.user_rejects.fetch_add(1, Ordering::Relaxed),
+ };
+ }
+ }
+ let counter = Arc::new(Counter {
+ tenant_rejects: AtomicU64::new(0),
+ user_rejects: AtomicU64::new(0),
+ });
+ // Tenant burst 1, user burst 1: first RPC consumes both;
+ // second is rejected by *tenant* (checked first), third —
+ // after we let tenant refill — should hit user.
+ let r = MemoryLimiter::new(
+ limits_with_user(1000.0, 1, 1000.0, 1),
+ HashMap::new(),
+ counter.clone(),
+ );
+ r.check("t", "u").unwrap();
+ let _ = r.check("t", "u");
+ // Both burst=1 with 1000 RPS refill ≈ 1ms — bucket is full
+ // again on the next iteration. Counts will be at least 1 of
+ // some scope; we only assert one happened.
+ assert!(
+ counter.tenant_rejects.load(Ordering::Relaxed)
+ + counter.user_rejects.load(Ordering::Relaxed)
+ >= 1
+ );
+ }
+}
diff --git a/crates/ratelimit/src/redis.rs b/crates/ratelimit/src/redis.rs
new file mode 100644
index 0000000..8ef193f
--- /dev/null
+++ b/crates/ratelimit/src/redis.rs
@@ -0,0 +1,350 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Redis-backed distributed token bucket.
+//!
+//! The in-memory limiter in the parent module enforces quotas per
+//! *gateway replica*. A 3-replica deployment with `rpcsPerSecond: 100`
+//! actually admits up to 300 RPS cluster-wide before any throttling
+//! fires — fine for back-pressure, wrong for a strict-SaaS quota.
+//! This module shares the bucket state in Redis so all replicas
+//! enforce the same numbers.
+//!
+//! ## Algorithm: token bucket via Lua
+//!
+//! Each `(tenant, user)` pair is one Redis hash with two fields:
+//!
+//! * `tokens` — floating-point token balance.
+//! * `ts` — last-refill timestamp in milliseconds since the Unix epoch.
+//!
+//! Every RPC runs [`TOKEN_BUCKET_SCRIPT`] via `EVAL` (or `EVALSHA`
+//! once cached). The Lua does the entire refill-and-take atomically
+//! under Redis's single-threaded execution, so no two replicas can
+//! ever observe the same `tokens` value and both decrement it. The
+//! script returns `1` on admit, `0` on reject, which the Rust caller
+//! turns into an `Ok(()) / ResourceExhausted` pair.
+//!
+//! ## Why not `redis-cell`?
+//!
+//! `CL.THROTTLE` from the redis-cell module is a one-line GCRA
+//! implementation that we'd happily use — except it requires
+//! `loadmodule redis-cell.so`, which most managed Redis offerings
+//! (ElastiCache, MemoryStore, Upstash) don't permit. Sticking to
+//! plain `EVAL` keeps the chart usable everywhere.
+//!
+//! ## Fail mode
+//!
+//! When Redis is unreachable, [`RedisLimiter::check`] consults
+//! [`FailMode`]:
+//!
+//! * `Open` (default) — admit the RPC. Availability over strict
+//! quota; matches the Redis affinity-store's fail-soft
+//! behaviour. The `redis_error_observer` is notified so operators
+//! can alert on a sustained nonzero rate.
+//! * `Closed` — reject the RPC. Use when a Redis outage must not
+//! become a quota-bypass attack vector. Makes Redis a hard
+//! dependency of every RPC.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use ::redis::aio::ConnectionManager;
+use ::redis::{RedisError, Script};
+use tokio::sync::Mutex;
+use tonic::Status;
+use tracing::{debug, warn};
+
+use crate::{BucketRate, FailMode, LimiterObserver, RedisErrorObserver, RejectScope, TenantLimits};
+
+/// Token-bucket Lua script. Inputs and outputs:
+///
+/// * `KEYS[1]` — the bucket hash key (e.g. `scg-rl:t:team-a`)
+/// * `ARGV[1]` — refill rate (tokens/second, float)
+/// * `ARGV[2]` — bucket capacity (max tokens, float)
+/// * `ARGV[3]` — current time in milliseconds (we send it from Rust
+/// instead of using `TIME` so the test can inject deterministic
+/// clocks; production uses `now()` in `RedisLimiter::check`)
+/// * `ARGV[4]` — TTL for the key in seconds (so abandoned buckets
+/// GC themselves; default to a generous multiple of the refill
+/// period)
+///
+/// Returns `1` when a token was consumed, `0` otherwise.
+///
+/// The script is idempotent across retries only in the sense that
+/// each `EVAL` is a single atomic operation — there is no "undo"
+/// for an admitted call. Callers that retry on transient network
+/// errors will *not* double-count rejections.
+pub(crate) const TOKEN_BUCKET_SCRIPT: &str = r#"
+local key = KEYS[1]
+local rate = tonumber(ARGV[1])
+local capacity = tonumber(ARGV[2])
+local now_ms = tonumber(ARGV[3])
+local ttl = tonumber(ARGV[4])
+
+local data = redis.call('HMGET', key, 'tokens', 'ts')
+local tokens = tonumber(data[1])
+local last_ms = tonumber(data[2])
+
+if tokens == nil then
+ -- Fresh bucket starts full so a new tenant gets its burst.
+ tokens = capacity
+ last_ms = now_ms
+end
+
+-- Refill based on wall-clock elapsed; cap at capacity.
+local elapsed_s = math.max(0, (now_ms - last_ms) / 1000.0)
+tokens = math.min(capacity, tokens + elapsed_s * rate)
+
+local admitted = 0
+if tokens >= 1.0 then
+ tokens = tokens - 1.0
+ admitted = 1
+end
+
+redis.call('HMSET', key, 'tokens', tokens, 'ts', now_ms)
+redis.call('EXPIRE', key, ttl)
+return admitted
+"#;
+
+/// Configuration for the Redis-backed limiter. Keep it small —
+/// extension points (key prefix, TTLs) are tuneable but rarely
+/// need to be.
+#[derive(Debug, Clone)]
+pub struct RedisLimiterConfig {
+ /// Redis URL (e.g. `redis://host:6379`). Reuses the same client
+ /// flavour as `scg-store-redis`.
+ pub url: String,
+ /// All keys live under `{key_prefix}:t:*` (tenant bucket) and
+ /// `{key_prefix}:u:*` (user bucket). Defaults to `"scg-rl"`.
+ /// Use a different prefix from the affinity store so a `FLUSH`
+ /// of one doesn't take out the other.
+ pub key_prefix: String,
+ /// Bucket-key TTL. Abandoned `(tenant, user)` pairs are
+ /// GC'd by Redis after this. Defaults to one hour — long enough
+ /// that an idle bucket survives a coffee break, short enough
+ /// that a one-off CI test doesn't leak forever.
+ pub key_ttl: Duration,
+ /// What to do when Redis is unreachable. See [`FailMode`].
+ pub fail_mode: FailMode,
+}
+
+impl Default for RedisLimiterConfig {
+ fn default() -> Self {
+ Self {
+ url: "redis://localhost:6379".into(),
+ key_prefix: "scg-rl".into(),
+ key_ttl: Duration::from_secs(3600),
+ fail_mode: FailMode::Open,
+ }
+ }
+}
+
+/// Errors that surface during connection setup. Per-request errors
+/// (timeouts, ConnectionRefused at runtime) don't bubble up — they
+/// route through [`FailMode`] inside `check`.
+#[derive(Debug, thiserror::Error)]
+pub enum RedisLimiterError {
+ #[error("redis connect failed: {0}")]
+ Connect(#[from] RedisError),
+}
+
+/// Distributed rate limiter. Cheap to clone — connection + script
+/// handle live behind `Arc`.
+#[derive(Clone)]
+pub struct RedisLimiter {
+ inner: Arc<Inner>,
+}
+
+struct Inner {
+ cfg: RedisLimiterConfig,
+ /// Per-tenant config; everything not in `overrides` falls back
+ /// to `default`. Same shape as the in-memory limiter.
+ overrides: HashMap<String, TenantLimits>,
+ default: TenantLimits,
+ /// Shared async connection. `ConnectionManager` reconnects
+ /// transparently after Redis hiccups; the inner `Mutex` is so
+ /// `&self` can hand out a `&mut Connection` to issue commands.
+ conn: Mutex<ConnectionManager>,
+ /// Reject observer (same one the memory limiter uses) so metric
+ /// labels stay consistent across the two backends.
+ observer: Arc<dyn LimiterObserver>,
+ /// Redis-error observer. Bumps a separate metric (errors are
+ /// distinct from rejections — a fail-open RPC fires this
+ /// without firing the reject observer).
+ redis_error_observer: Arc<dyn RedisErrorObserver>,
+ /// Pre-compiled Lua script. Redis caches it by SHA on first use;
+ /// subsequent calls go through EVALSHA implicitly.
+ script: Script,
+}
+
+impl RedisLimiter {
+ /// Connect to Redis. Fails fast on misconfigured URL or
+ /// unreachable host at *boot* time (operators should hear about
+ /// these immediately). Per-request errors take the [`FailMode`]
+ /// path instead.
+ pub async fn connect(
+ cfg: RedisLimiterConfig,
+ default: TenantLimits,
+ overrides: HashMap<String, TenantLimits>,
+ observer: Arc<dyn LimiterObserver>,
+ redis_error_observer: Arc<dyn RedisErrorObserver>,
+ ) -> Result<Self, RedisLimiterError> {
+ let client = ::redis::Client::open(cfg.url.clone())?;
+ let conn = ConnectionManager::new(client).await?;
+ let script = Script::new(TOKEN_BUCKET_SCRIPT);
+ Ok(Self {
+ inner: Arc::new(Inner {
+ cfg,
+ overrides,
+ default,
+ conn: Mutex::new(conn),
+ observer,
+ redis_error_observer,
+ script,
+ }),
+ })
+ }
+
+ pub fn is_active(&self) -> bool {
+ self.inner.default.tenant.is_enabled()
+ || self.inner.default.per_user.is_enabled()
+ || self
+ .inner
+ .overrides
+ .values()
+ .any(|l| l.tenant.is_enabled() || l.per_user.is_enabled())
+ }
+
+ fn tenant_key(&self, tenant: &str) -> String {
+ format!("{}:t:{}", self.inner.cfg.key_prefix, tenant)
+ }
+
+ fn user_key(&self, tenant: &str, user: &str) -> String {
+ format!("{}:u:{}|{}", self.inner.cfg.key_prefix, tenant, user)
+ }
+
+ /// Run the token-bucket script for a single bucket and return
+ /// `Ok(true)` (admitted), `Ok(false)` (rejected), or
+ /// `Err(redis_error)`. Callers translate that triple into the
+ /// fail-mode + observer dance.
+ async fn take_one(&self, key: String, rate: &BucketRate) -> Result<bool, RedisError> {
+ let now_ms = current_time_ms();
+ let ttl = self.inner.cfg.key_ttl.as_secs() as i64;
+ let mut conn = self.inner.conn.lock().await;
+ let admitted: i64 = self
+ .inner
+ .script
+ .key(&key)
+ .arg(rate.rpcs_per_second)
+ .arg(rate.burst as f64)
+ .arg(now_ms)
+ .arg(ttl)
+ .invoke_async(&mut *conn)
+ .await?;
+ Ok(admitted == 1)
+ }
+
+ /// Tenant-then-user bucket check. Returns Ok(()) on admit, or
+ /// `ResourceExhausted` on reject. Redis errors route through
+ /// `FailMode`.
+ pub async fn check(&self, tenant: &str, user: &str) -> Result<(), Status> {
+ let limits = self
+ .inner
+ .overrides
+ .get(tenant)
+ .copied()
+ .unwrap_or(self.inner.default);
+
+ if !limits.tenant.is_enabled() && !limits.per_user.is_enabled() {
+ return Ok(());
+ }
+
+ // Tenant bucket first: a tenant-wide reject is the more
+ // diagnostically useful signal for an operator.
+ if limits.tenant.is_enabled() {
+ match self.take_one(self.tenant_key(tenant), &limits.tenant).await {
+ Ok(true) => {}
+ Ok(false) => {
+ debug!(%tenant, %user, "rate_limit(redis): rejected at tenant scope");
+ self.inner.observer.on_reject(tenant, RejectScope::Tenant);
+ return Err(Status::resource_exhausted(format!(
+ "tenant {:?} rate limit exceeded",
+ tenant
+ )));
+ }
+ Err(e) => {
+ if let Some(s) = self.handle_redis_error(tenant, "tenant_bucket", e) {
+ return Err(s);
+ }
+ // fail-open: fall through to the user check.
+ }
+ }
+ }
+
+ if limits.per_user.is_enabled() {
+ match self
+ .take_one(self.user_key(tenant, user), &limits.per_user)
+ .await
+ {
+ Ok(true) => {}
+ Ok(false) => {
+ debug!(%tenant, %user, "rate_limit(redis): rejected at user scope");
+ self.inner.observer.on_reject(tenant, RejectScope::User);
+ return Err(Status::resource_exhausted(format!(
+ "user {:?} rate limit exceeded (tenant {:?})",
+ user, tenant
+ )));
+ }
+ Err(e) => {
+ if let Some(s) = self.handle_redis_error(tenant, "user_bucket", e) {
+ return Err(s);
+ }
+ }
+ }
+ }
+
+ Ok(())
+ }
+
+ /// Centralised fail-mode routing. Returns `Some(Status)` when
+ /// the caller should reject, `None` when it should keep going
+ /// (fail-open). Bumps the error observer either way.
+ fn handle_redis_error(
+ &self,
+ tenant: &str,
+ reason: &'static str,
+ err: RedisError,
+ ) -> Option<Status> {
+ warn!(%tenant, reason, error = %err, "rate_limit(redis): backend error");
+ self.inner
+ .redis_error_observer
+ .on_redis_error(tenant, reason);
+ match self.inner.cfg.fail_mode {
+ FailMode::Open => None,
+ FailMode::Closed => Some(Status::resource_exhausted(
+ "rate limiter unavailable (fail-closed)",
+ )),
+ }
+ }
+}
+
+fn current_time_ms() -> i64 {
+ use std::time::{SystemTime, UNIX_EPOCH};
+ SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .map(|d| d.as_millis() as i64)
+ .unwrap_or(0)
+}
diff --git a/crates/ratelimit/tests/redis_integration.rs b/crates/ratelimit/tests/redis_integration.rs
new file mode 100644
index 0000000..c5c2fc3
--- /dev/null
+++ b/crates/ratelimit/tests/redis_integration.rs
@@ -0,0 +1,281 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Integration tests for the Redis-backed [`RedisLimiter`].
+//!
+//! Spins up a real Redis via testcontainers and drives the limiter
+//! against it. Requires a working Docker daemon; tests are marked
+//! `#[ignore]` so plain `cargo test --workspace` doesn't fail in
+//! Docker-less environments. Run deliberately with:
+//!
+//! ```bash
+//! cargo test -p scg-ratelimit -- --ignored
+//! ```
+//!
+//! Coverage:
+//!
+//! * Burst-then-reject across the bucket capacity.
+//! * Two tenants are independent (one exhausting doesn't affect
+//! the other).
+//! * Two-replica simulation: two `RedisLimiter` instances pointing
+//! at the same Redis enforce a shared bucket — what an in-memory
+//! limiter explicitly does *not* do.
+//! * Fail-mode semantics when Redis is unreachable: `Open` admits,
+//! `Closed` rejects.
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use scg_ratelimit::redis::{RedisLimiter, RedisLimiterConfig};
+use scg_ratelimit::{
+ BucketRate, FailMode, LimiterObserver, NoopRedisErrorObserver, RedisErrorObserver, RejectScope,
+ TenantLimits,
+};
+use testcontainers_modules::{
+ redis::{Redis, REDIS_PORT},
+ testcontainers::{runners::AsyncRunner, ContainerAsync},
+};
+
+#[derive(Default)]
+struct Counter {
+ tenant: AtomicU64,
+ user: AtomicU64,
+}
+
+impl LimiterObserver for Counter {
+ fn on_reject(&self, _tenant: &str, scope: RejectScope) {
+ match scope {
+ RejectScope::Tenant => self.tenant.fetch_add(1, Ordering::Relaxed),
+ RejectScope::User => self.user.fetch_add(1, Ordering::Relaxed),
+ };
+ }
+}
+
+#[derive(Default)]
+struct RedisErrCounter {
+ errors: AtomicU64,
+}
+
+impl RedisErrorObserver for RedisErrCounter {
+ fn on_redis_error(&self, _: &str, _: &'static str) {
+ self.errors.fetch_add(1, Ordering::Relaxed);
+ }
+}
+
+/// Start a real Redis container, return its URL plus the container
+/// handle (drop to stop). Container start can fail if Docker isn't
+/// running — that surfaces as the `Err` from `start`.
+async fn spawn_redis() -> (String, ContainerAsync<Redis>) {
+ let node = Redis::default().start().await.expect("start redis");
+ let host = node.get_host().await.expect("redis host");
+ let port = node
+ .get_host_port_ipv4(REDIS_PORT)
+ .await
+ .expect("redis port");
+ let url = format!("redis://{}:{}", host, port);
+ (url, node)
+}
+
+fn limits(rps: f64, burst: u64) -> TenantLimits {
+ TenantLimits {
+ tenant: BucketRate {
+ rpcs_per_second: rps,
+ burst,
+ },
+ per_user: BucketRate::disabled(),
+ }
+}
+
+/// Build a limiter bound to the given URL with the given per-tenant
+/// limits as the default.
+async fn limiter_for(url: &str, default: TenantLimits, fail_mode: FailMode) -> RedisLimiter {
+ let cfg = RedisLimiterConfig {
+ url: url.into(),
+ key_prefix: format!("scg-rl-test-{}", rand_suffix()),
+ key_ttl: Duration::from_secs(30),
+ fail_mode,
+ };
+ let observer = Arc::new(Counter::default());
+ let err_obs: Arc<dyn RedisErrorObserver> = Arc::new(NoopRedisErrorObserver);
+ RedisLimiter::connect(cfg, default, HashMap::new(), observer, err_obs)
+ .await
+ .expect("connect redis limiter")
+}
+
+/// Tests share a Redis instance; each test gets its own key prefix
+/// so they don't collide.
+fn rand_suffix() -> String {
+ use std::time::{SystemTime, UNIX_EPOCH};
+ SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .map(|d| format!("{}-{}", d.as_nanos(), std::process::id()))
+ .unwrap_or_else(|_| "0".into())
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn burst_admits_then_rejects() {
+ let (url, _node) = spawn_redis().await;
+ // 1 RPS, burst 5: 5 admits, 6th rejects (refill < 1 token).
+ let l = limiter_for(&url, limits(1.0, 5), FailMode::Open).await;
+ for _ in 0..5 {
+ l.check("t", "u").await.expect("within burst");
+ }
+ let err = l.check("t", "u").await.unwrap_err();
+ assert_eq!(err.code(), tonic::Code::ResourceExhausted);
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn tenants_are_independent() {
+ let (url, _node) = spawn_redis().await;
+ let l = limiter_for(&url, limits(1.0, 2), FailMode::Open).await;
+
+ // Exhaust team-a.
+ l.check("team-a", "u").await.unwrap();
+ l.check("team-a", "u").await.unwrap();
+ assert!(l.check("team-a", "u").await.is_err());
+
+ // team-b's bucket is independent.
+ l.check("team-b", "u").await.unwrap();
+ l.check("team-b", "u").await.unwrap();
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn two_replicas_share_the_bucket() {
+ // The whole point of the Redis backend: an in-memory limiter
+ // would double the effective quota when you double the
+ // replicas. The Redis one must not.
+ let (url, _node) = spawn_redis().await;
+
+ // Share a *fixed* key prefix between both limiters so they
+ // actually contend on the same Redis keys.
+ let cfg = |fm| RedisLimiterConfig {
+ url: url.clone(),
+ key_prefix: "scg-rl-test-shared".into(),
+ key_ttl: Duration::from_secs(30),
+ fail_mode: fm,
+ };
+ let obs = Arc::new(Counter::default());
+ let err_obs: Arc<dyn RedisErrorObserver> = Arc::new(NoopRedisErrorObserver);
+ let l1 = RedisLimiter::connect(
+ cfg(FailMode::Open),
+ limits(1.0, 3),
+ HashMap::new(),
+ obs.clone(),
+ err_obs.clone(),
+ )
+ .await
+ .unwrap();
+ let l2 = RedisLimiter::connect(
+ cfg(FailMode::Open),
+ limits(1.0, 3),
+ HashMap::new(),
+ obs.clone(),
+ err_obs,
+ )
+ .await
+ .unwrap();
+
+ // Burst=3 shared. Two RPCs through replica 1, two through
+ // replica 2: the 4th overall must reject regardless of which
+ // replica it hits.
+ l1.check("shared-t", "u").await.unwrap();
+ l1.check("shared-t", "u").await.unwrap();
+ l2.check("shared-t", "u").await.unwrap();
+ let err = l2.check("shared-t", "u").await.unwrap_err();
+ assert_eq!(err.code(), tonic::Code::ResourceExhausted);
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn refill_admits_after_wait() {
+ let (url, _node) = spawn_redis().await;
+ // 10 RPS, burst 2 — ~200ms gives us 2 fresh tokens.
+ let l = limiter_for(&url, limits(10.0, 2), FailMode::Open).await;
+ l.check("t", "u").await.unwrap();
+ l.check("t", "u").await.unwrap();
+ assert!(l.check("t", "u").await.is_err());
+ tokio::time::sleep(Duration::from_millis(300)).await;
+ assert!(l.check("t", "u").await.is_ok());
+}
+
+#[tokio::test]
+async fn fail_open_admits_when_redis_unreachable() {
+ // Connect against a port we know is closed. The connect itself
+ // may succeed in the lazy `ConnectionManager` path — what
+ // matters is that `check` does not return an error.
+ let cfg = RedisLimiterConfig {
+ // Reserved port that should be closed.
+ url: "redis://127.0.0.1:1".into(),
+ key_prefix: "scg-rl-test-failopen".into(),
+ key_ttl: Duration::from_secs(30),
+ fail_mode: FailMode::Open,
+ };
+ let obs = Arc::new(Counter::default());
+ let err_obs = Arc::new(RedisErrCounter::default());
+ // ConnectionManager may still succeed here (it's lazy); skip
+ // the test if connect itself fails because that would be the
+ // boot-time error path, not the per-RPC fail-mode path.
+ let limiter = match RedisLimiter::connect(
+ cfg,
+ limits(1.0, 1),
+ HashMap::new(),
+ obs.clone(),
+ err_obs.clone(),
+ )
+ .await
+ {
+ Ok(l) => l,
+ Err(_) => return,
+ };
+
+ // The check must not return an error — fail-open admits.
+ let res = limiter.check("t", "u").await;
+ assert!(res.is_ok(), "fail-open should admit, got {:?}", res);
+ // …and the error metric should fire.
+ assert!(err_obs.errors.load(Ordering::Relaxed) >= 1);
+}
+
+#[tokio::test]
+async fn fail_closed_rejects_when_redis_unreachable() {
+ let cfg = RedisLimiterConfig {
+ url: "redis://127.0.0.1:1".into(),
+ key_prefix: "scg-rl-test-failclosed".into(),
+ key_ttl: Duration::from_secs(30),
+ fail_mode: FailMode::Closed,
+ };
+ let obs = Arc::new(Counter::default());
+ let err_obs = Arc::new(RedisErrCounter::default());
+ let limiter = match RedisLimiter::connect(
+ cfg,
+ limits(1.0, 1),
+ HashMap::new(),
+ obs.clone(),
+ err_obs.clone(),
+ )
+ .await
+ {
+ Ok(l) => l,
+ Err(_) => return,
+ };
+
+ let err = limiter.check("t", "u").await.unwrap_err();
+ assert_eq!(err.code(), tonic::Code::ResourceExhausted);
+ assert!(err_obs.errors.load(Ordering::Relaxed) >= 1);
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org