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 cf716c0 [SPARK-59046] Add scg-store-redis and scg-healthcheck crates
cf716c0 is described below
commit cf716c0b8d05d40270997501cbb4e54f0b341cbe
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Thu Aug 27 23:02:29 2026 -0700
[SPARK-59046] Add scg-store-redis and scg-healthcheck 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 more crates that implement/wrap the `scg-routing` traits,
grouped
into one PR:
- `scg-store-redis` — a Redis-backed `AffinityStore` for cross-replica HA
(the
multi-replica counterpart to the in-process `scg-store-memory`).
- `scg-healthcheck` — active gRPC health probing wrapped around a `Pool`,
evicting unhealthy backends from routing.
Both depend on the `scg-routing` crate. They compile and test on their own;
store-redis's Redis integration tests are `#[ignore]` so they compile but
don't
require Docker in CI. Both are appended to the workspace `members` list and
are
consumed later by the `proxy` crate (both are proxy dev-dependencies used
by its
examples/tests). With this PR, all four proxy dev-dependency crates
(store-memory, pool-static, store-redis, healthcheck) are in place, so the
proxy
PR can follow.
### 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
(Redis integration tests are `#[ignore]`, so no Docker is required in CI).
### Was this patch authored or co-authored using generative AI tooling?
Yes, co-authored with Claude Code.
Closes #12 from viirya/import-09-store-redis-healthcheck.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
Cargo.lock | 42 +++
Cargo.toml | 2 +
crates/healthcheck/Cargo.toml | 40 +++
crates/healthcheck/src/lib.rs | 377 ++++++++++++++++++++++++++
crates/healthcheck/tests/probe_eviction.rs | 141 ++++++++++
crates/store-redis/Cargo.toml | 39 +++
crates/store-redis/examples/smoke.rs | 125 +++++++++
crates/store-redis/src/lib.rs | 233 ++++++++++++++++
crates/store-redis/tests/redis_integration.rs | 183 +++++++++++++
9 files changed, 1182 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index c59f45e..856f3d0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2304,6 +2304,20 @@ dependencies = [
"tonic-prost-build",
]
+[[package]]
+name = "scg-healthcheck"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "parking_lot",
+ "scg-routing",
+ "tokio",
+ "tokio-stream",
+ "tonic",
+ "tonic-health",
+ "tracing",
+]
+
[[package]]
name = "scg-observability"
version = "0.1.0"
@@ -2370,6 +2384,20 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "scg-store-redis"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "redis",
+ "scg-routing",
+ "testcontainers-modules",
+ "thiserror 2.0.20",
+ "tokio",
+ "tracing",
+]
+
[[package]]
name = "scg-tenant"
version = "0.1.0"
@@ -2932,6 +2960,7 @@ dependencies = [
"futures-core",
"pin-project-lite",
"tokio",
+ "tokio-util",
]
[[package]]
@@ -2991,6 +3020,19 @@ dependencies = [
"syn 2.0.119",
]
+[[package]]
+name = "tonic-health"
+version = "0.14.6"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "fcfab99db777fba2802f0dfa861d1628d1ae916fb199d29819941f139ae85082"
+dependencies = [
+ "prost",
+ "tokio",
+ "tokio-stream",
+ "tonic",
+ "tonic-prost",
+]
+
[[package]]
name = "tonic-prost"
version = "0.14.6"
diff --git a/Cargo.toml b/Cargo.toml
index c527304..cc94747 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -26,6 +26,8 @@ members = [
"crates/tenant",
"crates/store-memory",
"crates/pool-static",
+ "crates/store-redis",
+ "crates/healthcheck",
]
[workspace.package]
diff --git a/crates/healthcheck/Cargo.toml b/crates/healthcheck/Cargo.toml
new file mode 100644
index 0000000..94e52a8
--- /dev/null
+++ b/crates/healthcheck/Cargo.toml
@@ -0,0 +1,40 @@
+# 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-healthcheck"
+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 = "Active gRPC health-check probing wrapper for the Spark Connect
Gateway's backend pools."
+publish = false
+
+[dependencies]
+scg-routing = { path = "../routing" }
+tokio = { workspace = true }
+tonic = { workspace = true }
+tonic-health = { workspace = true }
+parking_lot = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+tokio-stream = { workspace = true }
+async-trait = { workspace = true }
+tonic-health = { workspace = true }
diff --git a/crates/healthcheck/src/lib.rs b/crates/healthcheck/src/lib.rs
new file mode 100644
index 0000000..e8722b7
--- /dev/null
+++ b/crates/healthcheck/src/lib.rs
@@ -0,0 +1,377 @@
+// 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.
+
+//! Active gRPC health probing wrapped around a [`Pool`].
+//!
+//! The base [`scg_routing::Pool`] trait is purely round-robin: it
+//! hands out backends without knowing whether they actually answer.
+//! A backend pod that has wedged but not crashed (process alive,
+//! gRPC server not responding) will keep getting traffic until K8s
+//! kills the pod or the pool's source-of-truth (e.g. K8s
+//! Endpoints) reflects the failure.
+//!
+//! [`HealthAwarePool`] wraps any [`Pool`] and adds the
+//! [gRPC Health Check Protocol]:
+//!
+//! 1. A background task probes `grpc.health.v1.Health/Check` against
+//! every backend in `inner.all_healthy()` every `interval`.
+//! 2. After `unhealthy_threshold` consecutive failures, the backend
+//! is removed from `pick()` results.
+//! 3. Probes continue against the unhealthy set; after
+//! `healthy_threshold` consecutive successes the backend is
+//! re-admitted.
+//! 4. `pick()` retries up to `inner.all_healthy().len()` times if it
+//! draws an unhealthy backend, so traffic keeps flowing as long
+//! as anything is up.
+//!
+//! [gRPC Health Check Protocol]:
https://github.com/grpc/grpc/blob/master/doc/health-checking.md
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use parking_lot::RwLock;
+use scg_routing::{BackendMember, Pool};
+use tonic_health::pb::health_client::HealthClient;
+use tonic_health::pb::HealthCheckRequest;
+use tonic_health::ServingStatus;
+use tracing::{debug, info, warn};
+
+/// Per-backend health-check tuning. Defaults are chosen for the
+/// "thousands of backends, sub-second tail" world we're not in:
+/// fast enough to evict a wedged pod within ~15s, slow enough not to
+/// pummel the backends.
+#[derive(Debug, Clone)]
+pub struct HealthCheckConfig {
+ pub interval: Duration,
+ pub timeout: Duration,
+ /// Consecutive failures required to mark a backend unhealthy.
+ pub unhealthy_threshold: u32,
+ /// Consecutive successes required to re-admit a backend.
+ pub healthy_threshold: u32,
+}
+
+impl Default for HealthCheckConfig {
+ fn default() -> Self {
+ Self {
+ interval: Duration::from_secs(5),
+ timeout: Duration::from_secs(2),
+ unhealthy_threshold: 3,
+ healthy_threshold: 2,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum HealthState {
+ Healthy,
+ Unhealthy,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct BackendStatus {
+ state: HealthState,
+ /// Counter resets on transition: while Healthy, counts consecutive
+ /// failures; while Unhealthy, counts consecutive successes.
+ streak: u32,
+}
+
+impl BackendStatus {
+ fn fresh() -> Self {
+ // New backends are presumed healthy. We only stop trusting
+ // them once we've seen `unhealthy_threshold` failures —
+ // otherwise a momentary connection blip during the *first*
+ // probe would falsely evict every backend on startup.
+ Self {
+ state: HealthState::Healthy,
+ streak: 0,
+ }
+ }
+}
+
+/// `Pool` adapter that filters its inner pool's `members()` by an
+/// actively-probed health view. Placement over the filtered members
+/// stays with the selection strategy — this wrapper only decides who
+/// is a candidate at all.
+pub struct HealthAwarePool {
+ inner: Arc<dyn Pool>,
+ cfg: HealthCheckConfig,
+ /// Map of `host:port` -> current health status. Backends that
+ /// the inner pool no longer reports are GC'd on each probe round.
+ statuses: RwLock<HashMap<String, BackendStatus>>,
+}
+
+impl HealthAwarePool {
+ /// Wrap `inner` with active health checking using `cfg`. The
+ /// returned pool is immediately usable; spawn the probe task with
+ /// [`HealthAwarePool::spawn_probe`] to start mutating the health
+ /// view.
+ pub fn new(inner: Arc<dyn Pool>, cfg: HealthCheckConfig) -> Arc<Self> {
+ Arc::new(Self {
+ inner,
+ cfg,
+ statuses: RwLock::new(HashMap::new()),
+ })
+ }
+
+ /// Spawn the background probe loop. Returns a `JoinHandle` that
+ /// callers can `abort()` on shutdown; dropping the handle does
+ /// *not* stop the probes (the task holds an `Arc<Self>`).
+ pub fn spawn_probe(self: &Arc<Self>) -> tokio::task::JoinHandle<()> {
+ let me = Arc::clone(self);
+ tokio::spawn(async move {
+ let mut tick = tokio::time::interval(me.cfg.interval);
+ // Burst-tolerant: skip ticks if a probe round took longer
+ // than `interval` rather than queueing them up.
+
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+ loop {
+ tick.tick().await;
+ me.probe_all().await;
+ }
+ })
+ }
+
+ /// One probe round: fetch the inner pool's full backend list,
+ /// dial each in parallel, update each status entry, and GC
+ /// backends the inner pool no longer knows about.
+ async fn probe_all(&self) {
+ let backends: Vec<String> = self.inner.members().into_iter().map(|m|
m.addr).collect();
+ if backends.is_empty() {
+ // Nothing to probe; clear the status map so a restart
+ // doesn't carry over stale entries.
+ self.statuses.write().clear();
+ return;
+ }
+
+ let mut tasks = Vec::with_capacity(backends.len());
+ for addr in &backends {
+ let addr = addr.clone();
+ let timeout = self.cfg.timeout;
+ tasks.push(tokio::spawn(async move {
+ let ok = probe_one(&addr, timeout).await;
+ (addr, ok)
+ }));
+ }
+
+ let mut results = Vec::with_capacity(tasks.len());
+ for t in tasks {
+ if let Ok(r) = t.await {
+ results.push(r);
+ }
+ }
+
+ // Now apply. Hold the write lock for the whole apply so
+ // pick() / all_healthy() see a consistent view.
+ let mut g = self.statuses.write();
+ // GC: drop entries no longer present in `backends`.
+ let live: std::collections::HashSet<&str> =
backends.iter().map(String::as_str).collect();
+ g.retain(|k, _| live.contains(k.as_str()));
+
+ for (addr, ok) in results {
+ let entry =
g.entry(addr.clone()).or_insert_with(BackendStatus::fresh);
+ apply_probe(entry, ok, &self.cfg, &addr);
+ }
+ }
+
+ fn healthy_members(&self) -> Vec<BackendMember> {
+ let g = self.statuses.read();
+ // If a backend is in `inner.members()` but not yet in our
+ // status map, treat it as healthy (default). This avoids a
+ // gap on the first probe round where everything would look
+ // unhealthy.
+ let inner = self.inner.members();
+ let mut out = Vec::with_capacity(inner.len());
+ for m in inner {
+ match g.get(&m.addr) {
+ Some(s) if s.state == HealthState::Unhealthy => continue,
+ _ => out.push(m),
+ }
+ }
+ out
+ }
+}
+
+impl Pool for HealthAwarePool {
+ fn members(&self) -> Vec<BackendMember> {
+ self.healthy_members()
+ }
+
+ fn mark_unhealthy(&self, addr: &str) {
+ // Caller observed a forward error against `addr`. We don't
+ // immediately evict — that would race with the probe loop
+ // and risk oscillation — but we *do* prime the streak so the
+ // next probe failure flips us over the threshold faster.
+ let mut g = self.statuses.write();
+ if let Some(status) = g.get_mut(addr) {
+ if status.state == HealthState::Healthy {
+ status.streak = status.streak.saturating_add(1);
+ debug!(%addr, streak = status.streak, "healthcheck: handler
reported error");
+ }
+ }
+ self.inner.mark_unhealthy(addr);
+ }
+}
+
+/// Open a one-shot connection and call `Health.Check`. We don't
+/// reuse channels here on purpose — a wedged backend may have an
+/// open TCP connection that hangs forever; tearing it down per probe
+/// keeps us honest.
+async fn probe_one(addr: &str, timeout: Duration) -> bool {
+ let url = if addr.starts_with("http://";) || addr.starts_with("https://";) {
+ addr.to_string()
+ } else {
+ format!("http://{}";, addr)
+ };
+ let endpoint = match tonic::transport::Endpoint::from_shared(url) {
+ Ok(e) => e.connect_timeout(timeout).timeout(timeout),
+ Err(e) => {
+ warn!(%addr, error = %e, "healthcheck: invalid endpoint URL");
+ return false;
+ }
+ };
+ let channel = match endpoint.connect().await {
+ Ok(c) => c,
+ Err(e) => {
+ debug!(%addr, error = %e, "healthcheck: connect failed");
+ return false;
+ }
+ };
+ let mut client = HealthClient::new(channel);
+ // Empty service name = "the entire backend." Spark Connect's
+ // tonic server registers Health under the standard convention.
+ let req = HealthCheckRequest {
+ service: String::new(),
+ };
+ match client.check(req).await {
+ Ok(resp) => {
+ let status = resp.into_inner().status;
+ // ServingStatus::Serving == healthy.
+ status == ServingStatus::Serving as i32
+ }
+ Err(e) => {
+ // Backend may be a Spark Connect server without Health
+ // registered. We treat NOT_FOUND / UNIMPLEMENTED as an
+ // ambiguous signal and keep the backend healthy — the
+ // alternative is evicting every backend that doesn't ship
+ // grpc.health.v1, which would be a regression for older
+ // Spark versions.
+ match e.code() {
+ tonic::Code::NotFound | tonic::Code::Unimplemented => {
+ debug!(%addr, "healthcheck: backend has no Health service;
treating as healthy");
+ true
+ }
+ _ => {
+ debug!(%addr, error = %e, "healthcheck: probe failed");
+ false
+ }
+ }
+ }
+ }
+}
+
+/// Update one `BackendStatus` based on the latest probe outcome.
+fn apply_probe(status: &mut BackendStatus, ok: bool, cfg: &HealthCheckConfig,
addr: &str) {
+ match (status.state, ok) {
+ (HealthState::Healthy, true) => {
+ status.streak = 0;
+ }
+ (HealthState::Healthy, false) => {
+ status.streak = status.streak.saturating_add(1);
+ if status.streak >= cfg.unhealthy_threshold {
+ info!(%addr, "healthcheck: backend marked UNHEALTHY");
+ status.state = HealthState::Unhealthy;
+ status.streak = 0;
+ }
+ }
+ (HealthState::Unhealthy, true) => {
+ status.streak = status.streak.saturating_add(1);
+ if status.streak >= cfg.healthy_threshold {
+ info!(%addr, "healthcheck: backend back to HEALTHY");
+ status.state = HealthState::Healthy;
+ status.streak = 0;
+ }
+ }
+ (HealthState::Unhealthy, false) => {
+ status.streak = 0;
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn cfg(unhealthy: u32, healthy: u32) -> HealthCheckConfig {
+ HealthCheckConfig {
+ unhealthy_threshold: unhealthy,
+ healthy_threshold: healthy,
+ ..Default::default()
+ }
+ }
+
+ #[test]
+ fn streak_to_unhealthy_then_back() {
+ let mut s = BackendStatus::fresh();
+ let c = cfg(3, 2);
+
+ // Two failures: still healthy.
+ apply_probe(&mut s, false, &c, "x:1");
+ apply_probe(&mut s, false, &c, "x:1");
+ assert_eq!(s.state, HealthState::Healthy);
+
+ // Third: flip.
+ apply_probe(&mut s, false, &c, "x:1");
+ assert_eq!(s.state, HealthState::Unhealthy);
+
+ // One success: not enough.
+ apply_probe(&mut s, true, &c, "x:1");
+ assert_eq!(s.state, HealthState::Unhealthy);
+
+ // Second success: re-admitted.
+ apply_probe(&mut s, true, &c, "x:1");
+ assert_eq!(s.state, HealthState::Healthy);
+ }
+
+ #[test]
+ fn intermittent_failure_resets_streak() {
+ let mut s = BackendStatus::fresh();
+ let c = cfg(3, 2);
+ apply_probe(&mut s, false, &c, "x:1");
+ apply_probe(&mut s, false, &c, "x:1");
+ // One success in the middle resets the failure streak.
+ apply_probe(&mut s, true, &c, "x:1");
+ apply_probe(&mut s, false, &c, "x:1");
+ assert_eq!(s.state, HealthState::Healthy);
+ }
+
+ #[test]
+ fn intermittent_success_resets_recovery_streak() {
+ let mut s = BackendStatus {
+ state: HealthState::Unhealthy,
+ streak: 0,
+ };
+ let c = cfg(3, 3);
+ apply_probe(&mut s, true, &c, "x:1");
+ apply_probe(&mut s, true, &c, "x:1");
+ // One failure resets the recovery streak.
+ apply_probe(&mut s, false, &c, "x:1");
+ apply_probe(&mut s, true, &c, "x:1");
+ apply_probe(&mut s, true, &c, "x:1");
+ // Still need one more success to re-admit (threshold=3).
+ assert_eq!(s.state, HealthState::Unhealthy);
+ apply_probe(&mut s, true, &c, "x:1");
+ assert_eq!(s.state, HealthState::Healthy);
+ }
+}
diff --git a/crates/healthcheck/tests/probe_eviction.rs
b/crates/healthcheck/tests/probe_eviction.rs
new file mode 100644
index 0000000..d997641
--- /dev/null
+++ b/crates/healthcheck/tests/probe_eviction.rs
@@ -0,0 +1,141 @@
+// 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 test for [`HealthAwarePool`] against real gRPC
+//! servers. Verifies that:
+//!
+//! 1. A backend whose `Health` server is alive stays in
+//! `members()`.
+//! 2. A backend that goes down (server task dropped) is evicted
+//! after `unhealthy_threshold` consecutive failed probes.
+//! 3. `members()` keeps reporting only the surviving backend.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use scg_healthcheck::{HealthAwarePool, HealthCheckConfig};
+use scg_routing::{BackendMember, Pool};
+use tokio::net::TcpListener;
+use tokio::sync::oneshot;
+use tokio_stream::wrappers::TcpListenerStream;
+use tonic::transport::Server;
+use tonic_health::server::health_reporter;
+
+/// Static pool stub: returns whatever vec we feed it. We rebuild it
+/// for each test so we control which backends "exist" from the
+/// inner pool's perspective; the wrapper's job is to filter by
+/// health.
+struct StubPool {
+ backends: parking_lot::RwLock<Vec<String>>,
+}
+
+impl StubPool {
+ fn new(backends: Vec<String>) -> Arc<Self> {
+ Arc::new(Self {
+ backends: parking_lot::RwLock::new(backends),
+ })
+ }
+}
+
+impl Pool for StubPool {
+ fn members(&self) -> Vec<BackendMember> {
+ self.backends
+ .read()
+ .iter()
+ .map(BackendMember::new)
+ .collect()
+ }
+}
+
+/// Spawn a gRPC server with the standard `Health` service registered
+/// and reporting `Serving`. Returns its host:port and a shutdown
+/// sender.
+async fn spawn_health_server() -> (String, oneshot::Sender<()>) {
+ let (reporter, health_svc) = health_reporter();
+ // Empty service name = "the entire backend"; our probe code uses
+ // the same convention.
+ reporter
+ .set_service_status("", tonic_health::ServingStatus::Serving)
+ .await;
+
+ let lis = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = lis.local_addr().unwrap().to_string();
+ let (tx, rx) = oneshot::channel();
+ tokio::spawn(async move {
+ Server::builder()
+ .add_service(health_svc)
+ .serve_with_incoming_shutdown(TcpListenerStream::new(lis), async {
+ let _ = rx.await;
+ })
+ .await
+ .ok();
+ });
+ (addr, tx)
+}
+
+#[tokio::test]
+async fn evicts_dead_backend_then_keeps_alive_one() {
+ let (alive, _alive_kill) = spawn_health_server().await;
+ let (dying, dying_kill) = spawn_health_server().await;
+
+ let stub = StubPool::new(vec![alive.clone(), dying.clone()]);
+ // Aggressive: 200ms interval, 200ms timeout, 2-failure eviction
+ // → eviction of `dying` should land within ~600ms after kill.
+ let cfg = HealthCheckConfig {
+ interval: Duration::from_millis(200),
+ timeout: Duration::from_millis(300),
+ unhealthy_threshold: 2,
+ healthy_threshold: 2,
+ };
+ let wrapper = HealthAwarePool::new(stub.clone() as Arc<dyn Pool>, cfg);
+ let _probe = wrapper.spawn_probe();
+
+ // Let one probe round complete; both should still be healthy.
+ tokio::time::sleep(Duration::from_millis(400)).await;
+ let healthy: Vec<String> = wrapper.members().into_iter().map(|m|
m.addr).collect();
+ assert!(healthy.contains(&alive), "alive backend should be present");
+ assert!(
+ healthy.contains(&dying),
+ "dying backend should still be present pre-kill"
+ );
+
+ // Kill `dying`. Give the OS a moment to actually tear the
+ // listener down before probes start failing.
+ let _ = dying_kill.send(());
+ tokio::time::sleep(Duration::from_millis(200)).await;
+
+ // Wait for the wrapper to evict it. With 200ms interval and
+ // unhealthy_threshold=2, we expect eviction within ~600ms.
+ let mut evicted = false;
+ for _ in 0..30 {
+ let healthy: Vec<String> = wrapper.members().into_iter().map(|m|
m.addr).collect();
+ if !healthy.contains(&dying) && healthy.contains(&alive) {
+ evicted = true;
+ break;
+ }
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
+ assert!(
+ evicted,
+ "dying backend should be evicted within the test window"
+ );
+
+ // members() must keep reporting exactly the surviving backend —
+ // any selection strategy applied on top can only choose it.
+ for _ in 0..5 {
+ let m: Vec<String> = wrapper.members().into_iter().map(|x|
x.addr).collect();
+ assert_eq!(m, vec![alive.clone()]);
+ }
+}
diff --git a/crates/store-redis/Cargo.toml b/crates/store-redis/Cargo.toml
new file mode 100644
index 0000000..0f646e2
--- /dev/null
+++ b/crates/store-redis/Cargo.toml
@@ -0,0 +1,39 @@
+# 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-store-redis"
+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 = "Redis-backed affinity store for cross-replica Spark Connect
Gateway HA."
+publish = false
+
+[dependencies]
+scg-routing = { path = "../routing" }
+async-trait = { workspace = true }
+redis = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+thiserror = { workspace = true }
+anyhow = { workspace = true }
+
+[dev-dependencies]
+testcontainers-modules = { workspace = true }
diff --git a/crates/store-redis/examples/smoke.rs
b/crates/store-redis/examples/smoke.rs
new file mode 100644
index 0000000..bc3ac31
--- /dev/null
+++ b/crates/store-redis/examples/smoke.rs
@@ -0,0 +1,125 @@
+// 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.
+
+//! Smoke test against a real local Redis. Same scenarios as the
+//! testcontainers-backed integration tests but without Docker — when
+//! Docker Hub is unreachable, run a Redis on `:6399` (e.g.
+//! `redis-server --port 6399 --daemonize yes`) and execute:
+//!
+//! ```bash
+//! cargo run -p scg-store-redis --example smoke
+//! ```
+//!
+//! Exits with non-zero status if any invariant fails so CI / scripts
+//! can wrap it.
+
+use std::time::Duration;
+
+use scg_routing::{AffinityStore, SessionKey};
+use scg_store_redis::{RedisStore, RedisStoreConfig};
+
+#[tokio::main]
+async fn main() {
+ let url = std::env::var("REDIS_URL").unwrap_or_else(|_|
"redis://127.0.0.1:6399".into());
+ eprintln!("[smoke] connecting to {}", url);
+
+ // Use a unique key prefix per run so re-runs don't pollute each other.
+ let prefix = format!("scg-smoke-{}", std::process::id());
+ let cfg = RedisStoreConfig {
+ url,
+ key_prefix: prefix,
+ session_ttl: Duration::from_secs(30),
+ op_ttl: Duration::from_secs(30),
+ };
+ let store = RedisStore::connect(cfg.clone())
+ .await
+ .expect("connect to redis");
+
+ // 1. Bind / resolve / forget roundtrip.
+ let k = SessionKey::new("alice", "sess-1");
+ assert!(store.lookup_session(&k).await.is_none(), "fresh key");
+ let bound = store
+ .bind_session_if_absent(k.clone(), "be-a:15002".into())
+ .await;
+ assert_eq!(bound, "be-a:15002");
+ assert_eq!(
+ store.lookup_session(&k).await.as_deref(),
+ Some("be-a:15002")
+ );
+ eprintln!("[smoke] ok: bind+resolve roundtrip");
+
+ // 2. Atomicity / stickiness invariant.
+ let second = store
+ .bind_session_if_absent(k.clone(), "be-b:1".into())
+ .await;
+ assert_eq!(
+ second, "be-a:15002",
+ "stickiness invariant: first write wins"
+ );
+ eprintln!("[smoke] ok: bind_session_if_absent stickiness");
+
+ // 3. Op-id reverse index.
+ assert!(store.lookup_op("op-1").await.is_none());
+ store.bind_op("op-1".into(), "be-a:15002".into()).await;
+ assert_eq!(store.lookup_op("op-1").await.as_deref(), Some("be-a:15002"));
+ store.forget_op("op-1").await;
+ assert!(store.lookup_op("op-1").await.is_none());
+ eprintln!("[smoke] ok: op-id reverse index");
+
+ // 4. Two stores against the same Redis observe each other's writes
+ // — the multi-replica claim.
+ let store_b = RedisStore::connect(cfg.clone()).await.expect("connect b");
+ let k2 = SessionKey::new("alice", "shared");
+ let _ = store
+ .bind_session_if_absent(k2.clone(), "be-x:1".into())
+ .await;
+ assert_eq!(
+ store_b.lookup_session(&k2).await.as_deref(),
+ Some("be-x:1"),
+ "replica B must see replica A's binding"
+ );
+ let race = store_b
+ .bind_session_if_absent(k2.clone(), "be-y:1".into())
+ .await;
+ assert_eq!(race, "be-x:1", "replica A's earlier write wins");
+ eprintln!("[smoke] ok: cross-replica state visibility + atomicity");
+
+ // 5. TTL expiry — short TTL store.
+ let short_cfg = RedisStoreConfig {
+ session_ttl: Duration::from_secs(1),
+ op_ttl: Duration::from_secs(1),
+ key_prefix: format!("scg-smoke-ttl-{}", std::process::id()),
+ ..cfg.clone()
+ };
+ let short = RedisStore::connect(short_cfg).await.expect("connect short");
+ let k3 = SessionKey::new("u", "ttl");
+ short
+ .bind_session_if_absent(k3.clone(), "be-z:1".into())
+ .await;
+ assert!(short.lookup_session(&k3).await.is_some());
+ tokio::time::sleep(Duration::from_millis(1500)).await;
+ assert!(
+ short.lookup_session(&k3).await.is_none(),
+ "TTL-bound key must expire"
+ );
+ eprintln!("[smoke] ok: TTL expiry");
+
+ // Cleanup — forget the persistent keys so the Redis instance stays
+ // tidy for the next smoke run on a shared host.
+ store.forget_session(&k).await;
+ store.forget_session(&k2).await;
+
+ println!("[smoke] all invariants passed");
+}
diff --git a/crates/store-redis/src/lib.rs b/crates/store-redis/src/lib.rs
new file mode 100644
index 0000000..5889834
--- /dev/null
+++ b/crates/store-redis/src/lib.rs
@@ -0,0 +1,233 @@
+// 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 [`AffinityStore`] for cross-replica HA.
+//!
+//! The sibling [`scg-store-memory::MemoryStore`] only works when the
+//! gateway runs as a single replica. When operators want to scale
+//! out (Helm chart with `replicas > 1` and `kind: Deployment`), the
+//! affinity table needs to live somewhere all replicas can see —
+//! otherwise a Spark Connect session pinned to backend `B` by replica
+//! 1 will be re-pinned to a different backend by replica 2 the next
+//! time the client lands on it, breaking the
+//! `(user_id, session_id) -> backend` invariant.
+//!
+//! Schema in Redis:
+//!
+//! * `{prefix}:s:{user_id}|{session_id}` -> backend address. TTL
+//! refreshed on every successful resolve, so idle sessions
+//! eventually fall out and let the gateway re-pick.
+//! * `{prefix}:o:{op_id}` -> backend address. TTL is independent and
+//! typically shorter — operations are bounded by the lifetime of an
+//! `ExecutePlan` invocation.
+//!
+//! Concurrency invariant for sticky routing relies on `SET ... NX
+//! EX`: if two gateway replicas race to bind the same `(user_id,
+//! session_id)`, exactly one `SET` succeeds; the other reads the
+//! winner's value back and returns it from `bind_session_if_absent`.
+//!
+//! Error policy: this layer does not propagate Redis failures up the
+//! `AffinityStore` trait (the trait is intentionally infallible to
+//! keep `Router` simple). On a Redis outage, lookups return `None`
+//! and binds quietly drop, which makes the gateway degrade to
+//! pool-only routing — sessions land on whatever backend the pool
+//! picks each time, the same behaviour you'd see with no affinity
+//! cache at all. We log a `warn!` so the outage shows up in logs.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use redis::aio::ConnectionManager;
+use redis::AsyncCommands;
+use scg_routing::{AffinityStore, SessionKey};
+use tokio::sync::Mutex;
+use tracing::warn;
+
+#[derive(Debug, thiserror::Error)]
+pub enum RedisStoreError {
+ #[error("opening redis client: {0}")]
+ Open(#[from] redis::RedisError),
+}
+
+/// Configuration for [`RedisStore`].
+#[derive(Debug, Clone)]
+pub struct RedisStoreConfig {
+ /// `redis://` URL (e.g. `redis://localhost:6379` or
+ /// `redis://:password@host:6379/2`).
+ pub url: String,
+ /// Key prefix; all keys live under `{prefix}:s:` and `{prefix}:o:`.
+ /// Lets multiple gateway deployments share a Redis without
+ /// colliding.
+ pub key_prefix: String,
+ /// TTL for session bindings. Refreshed on every read. Pick a value
+ /// noticeably longer than typical client idle timeouts so
+ /// reconnecting clients keep their stickiness.
+ pub session_ttl: Duration,
+ /// TTL for op-id bindings. Operations are bounded by their
+ /// `ExecutePlan` lifetime; a few minutes is usually plenty.
+ pub op_ttl: Duration,
+}
+
+impl Default for RedisStoreConfig {
+ fn default() -> Self {
+ Self {
+ url: "redis://127.0.0.1:6379".into(),
+ key_prefix: "scg".into(),
+ session_ttl: Duration::from_secs(60 * 60), // 1h
+ op_ttl: Duration::from_secs(15 * 60), // 15min
+ }
+ }
+}
+
+pub struct RedisStore {
+ conn: Mutex<ConnectionManager>,
+ cfg: RedisStoreConfig,
+}
+
+impl RedisStore {
+ /// Open a connection to Redis. On a successful return the
+ /// `ConnectionManager` is alive but lazy — actual TCP work
+ /// happens on the first command. Failures here are usually
+ /// configuration / DNS errors, not "Redis is down".
+ pub async fn connect(cfg: RedisStoreConfig) -> Result<Self,
RedisStoreError> {
+ let client = redis::Client::open(cfg.url.clone())?;
+ let conn = ConnectionManager::new(client).await?;
+ Ok(Self {
+ conn: Mutex::new(conn),
+ cfg,
+ })
+ }
+
+ fn session_key(&self, k: &SessionKey) -> String {
+ // {prefix}:s:{tenant}|{user_id}|{session_id}
+ format!(
+ "{}:s:{}|{}|{}",
+ self.cfg.key_prefix, k.tenant, k.user_id, k.session_id
+ )
+ }
+
+ fn op_key(&self, op_id: &str) -> String {
+ format!("{}:o:{}", self.cfg.key_prefix, op_id)
+ }
+}
+
+#[async_trait]
+impl AffinityStore for RedisStore {
+ async fn lookup_session(&self, key: &SessionKey) -> Option<String> {
+ let k = self.session_key(key);
+ let mut conn = self.conn.lock().await;
+ // GET + EXPIRE in two round-trips. Could be combined with
+ // GETEX (Redis 6.2+) once we want to drop the round-trip.
+ match conn.get::<_, Option<String>>(&k).await {
+ Ok(Some(v)) => {
+ let _: Result<bool, _> =
+ conn.expire(&k, self.cfg.session_ttl.as_secs() as
i64).await;
+ Some(v)
+ }
+ Ok(None) => None,
+ Err(e) => {
+ warn!(error = %e, key = %k, "redis: lookup_session failed;
treating as miss");
+ None
+ }
+ }
+ }
+
+ async fn bind_session_if_absent(&self, key: SessionKey, backend: String)
-> String {
+ let k = self.session_key(&key);
+ let mut conn = self.conn.lock().await;
+ let ttl_secs = self.cfg.session_ttl.as_secs();
+ // SET NX EX — atomic create-if-absent with TTL.
+ match redis::cmd("SET")
+ .arg(&k)
+ .arg(&backend)
+ .arg("NX")
+ .arg("EX")
+ .arg(ttl_secs)
+ .query_async::<Option<String>>(&mut *conn)
+ .await
+ {
+ Ok(Some(_)) => backend, // SET returned "OK" → we won the race.
+ Ok(None) => {
+ // SET returned nil → key already existed; read it.
+ match conn.get::<_, Option<String>>(&k).await {
+ Ok(Some(existing)) => existing,
+ // The race-tie window where the existing binding
+ // expired between SET NX and GET is rare enough to
+ // accept the rebind; just return our intended value.
+ Ok(None) => backend,
+ Err(e) => {
+ warn!(error = %e, key = %k, "redis: GET after SET-NX
miss failed; rebinding");
+ backend
+ }
+ }
+ }
+ Err(e) => {
+ warn!(error = %e, key = %k, "redis: bind_session_if_absent
failed; degrading to pool pick");
+ // Redis is down. The router will use the backend we
+ // were handed (the pool's pick); subsequent calls
+ // will pick again until Redis recovers.
+ backend
+ }
+ }
+ }
+
+ async fn forget_session(&self, key: &SessionKey) {
+ let k = self.session_key(key);
+ let mut conn = self.conn.lock().await;
+ if let Err(e) = conn.del::<_, i64>(&k).await {
+ warn!(error = %e, key = %k, "redis: forget_session failed");
+ }
+ }
+
+ async fn lookup_op(&self, op_id: &str) -> Option<String> {
+ let k = self.op_key(op_id);
+ let mut conn = self.conn.lock().await;
+ match conn.get::<_, Option<String>>(&k).await {
+ Ok(Some(v)) => {
+ let _: Result<bool, _> = conn.expire(&k,
self.cfg.op_ttl.as_secs() as i64).await;
+ Some(v)
+ }
+ Ok(None) => None,
+ Err(e) => {
+ warn!(error = %e, key = %k, "redis: lookup_op failed; treating
as miss");
+ None
+ }
+ }
+ }
+
+ async fn bind_op(&self, op_id: String, backend: String) {
+ let k = self.op_key(&op_id);
+ let mut conn = self.conn.lock().await;
+ let ttl_secs = self.cfg.op_ttl.as_secs();
+ if let Err(e) = redis::cmd("SET")
+ .arg(&k)
+ .arg(&backend)
+ .arg("EX")
+ .arg(ttl_secs)
+ .query_async::<()>(&mut *conn)
+ .await
+ {
+ warn!(error = %e, key = %k, "redis: bind_op failed");
+ }
+ }
+
+ async fn forget_op(&self, op_id: &str) {
+ let k = self.op_key(op_id);
+ let mut conn = self.conn.lock().await;
+ if let Err(e) = conn.del::<_, i64>(&k).await {
+ warn!(error = %e, key = %k, "redis: forget_op failed");
+ }
+ }
+}
diff --git a/crates/store-redis/tests/redis_integration.rs
b/crates/store-redis/tests/redis_integration.rs
new file mode 100644
index 0000000..4371d22
--- /dev/null
+++ b/crates/store-redis/tests/redis_integration.rs
@@ -0,0 +1,183 @@
+// 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 [`scg_store_redis::RedisStore`] against a
+//! real Redis spun up via testcontainers.
+//!
+//! These tests require a working Docker daemon. CI environments
+//! without Docker can either skip them (cargo test will simply error
+//! on container start, surfaced as an `Err` from `start()`) or run
+//! `cargo test -p scg-store-redis --features ...` once we add an
+//! explicit feature gate. For now we mark them `#[ignore]` so a
+//! plain `cargo test --workspace` doesn't fail in environments where
+//! Docker isn't around. Run them deliberately with:
+//!
+//! ```bash
+//! cargo test -p scg-store-redis -- --ignored
+//! ```
+
+use std::time::Duration;
+
+use scg_routing::{AffinityStore, SessionKey};
+use scg_store_redis::{RedisStore, RedisStoreConfig};
+use testcontainers_modules::{
+ redis::{Redis, REDIS_PORT},
+ testcontainers::{runners::AsyncRunner, ContainerAsync},
+};
+
+/// Spin up a Redis container and connect a fresh `RedisStore` to it.
+/// The returned `_node` keeps the container alive for the test
+/// duration; drop it (end of test) and the container goes away.
+async fn rig() -> (RedisStore, ContainerAsync<Redis>) {
+ let node = Redis::default().start().await.expect("start redis");
+ let host = node.get_host().await.expect("host");
+ let port = node
+ .get_host_port_ipv4(REDIS_PORT)
+ .await
+ .expect("host port");
+ let cfg = RedisStoreConfig {
+ url: format!("redis://{}:{}", host, port),
+ // Keep TTLs comfortable so the test doesn't race the clock.
+ session_ttl: Duration::from_secs(30),
+ op_ttl: Duration::from_secs(30),
+ ..Default::default()
+ };
+ let store = RedisStore::connect(cfg).await.expect("connect redis");
+ (store, node)
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn session_bind_resolve_forget_roundtrip() {
+ let (store, _node) = rig().await;
+ let k = SessionKey::new("alice", "sess-1");
+
+ assert!(store.lookup_session(&k).await.is_none());
+ let bound = store
+ .bind_session_if_absent(k.clone(), "be-a:15002".into())
+ .await;
+ assert_eq!(bound, "be-a:15002");
+ assert_eq!(
+ store.lookup_session(&k).await.as_deref(),
+ Some("be-a:15002")
+ );
+
+ store.forget_session(&k).await;
+ assert!(store.lookup_session(&k).await.is_none());
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn bind_session_if_absent_is_atomic() {
+ let (store, _node) = rig().await;
+ let k = SessionKey::new("alice", "sticky");
+
+ let first = store
+ .bind_session_if_absent(k.clone(), "be-a:1".into())
+ .await;
+ assert_eq!(first, "be-a:1");
+ // A second concurrent caller racing to bind a different value must
+ // observe the first winner — this is the stickiness invariant.
+ let second = store
+ .bind_session_if_absent(k.clone(), "be-b:1".into())
+ .await;
+ assert_eq!(second, "be-a:1");
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn op_reverse_index_roundtrip() {
+ let (store, _node) = rig().await;
+ assert!(store.lookup_op("op-1").await.is_none());
+ store.bind_op("op-1".into(), "be-a:1".into()).await;
+ assert_eq!(store.lookup_op("op-1").await.as_deref(), Some("be-a:1"));
+ store.forget_op("op-1").await;
+ assert!(store.lookup_op("op-1").await.is_none());
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn two_stores_against_same_redis_share_state() {
+ // The whole point of Redis state: replica A binds, replica B
+ // resolves to the same backend. Without this we have no HA.
+ let node = Redis::default().start().await.expect("start redis");
+ let host = node.get_host().await.expect("host");
+ let port = node
+ .get_host_port_ipv4(REDIS_PORT)
+ .await
+ .expect("host port");
+ let url = format!("redis://{}:{}", host, port);
+
+ let store_a = RedisStore::connect(RedisStoreConfig {
+ url: url.clone(),
+ ..Default::default()
+ })
+ .await
+ .expect("connect a");
+ let store_b = RedisStore::connect(RedisStoreConfig {
+ url,
+ ..Default::default()
+ })
+ .await
+ .expect("connect b");
+
+ let k = SessionKey::new("alice", "shared");
+ let a_bound = store_a
+ .bind_session_if_absent(k.clone(), "be-a:1".into())
+ .await;
+ assert_eq!(a_bound, "be-a:1");
+
+ // Replica B must see the same binding without ever calling bind.
+ assert_eq!(store_b.lookup_session(&k).await.as_deref(), Some("be-a:1"));
+
+ // And if replica B races to bind a different backend, replica A's
+ // earlier write wins.
+ let b_bound = store_b
+ .bind_session_if_absent(k.clone(), "be-b:1".into())
+ .await;
+ assert_eq!(b_bound, "be-a:1");
+}
+
+#[tokio::test]
+#[ignore = "requires Docker"]
+async fn session_ttl_expires() {
+ let node = Redis::default().start().await.expect("start redis");
+ let host = node.get_host().await.expect("host");
+ let port = node
+ .get_host_port_ipv4(REDIS_PORT)
+ .await
+ .expect("host port");
+ let store = RedisStore::connect(RedisStoreConfig {
+ url: format!("redis://{}:{}", host, port),
+ // 1-second TTL so this test stays under a few seconds.
+ session_ttl: Duration::from_secs(1),
+ op_ttl: Duration::from_secs(1),
+ ..Default::default()
+ })
+ .await
+ .expect("connect");
+
+ let k = SessionKey::new("u", "ttl");
+ store
+ .bind_session_if_absent(k.clone(), "be-a:1".into())
+ .await;
+ assert!(store.lookup_session(&k).await.is_some());
+
+ // Wait long enough for Redis to expire the key. Each lookup also
+ // refreshes the TTL, so we deliberately don't call lookup during
+ // the wait.
+ tokio::time::sleep(Duration::from_millis(1500)).await;
+ assert!(store.lookup_session(&k).await.is_none());
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
