constructorfabric · cyberantonz · Aug 3, 2026 · Aug 3, 2026 · coderabbitai · Aug 3, 2026
@@ -208,6 +208,7 @@ jobs:
             -e MARIADB_USER=insight -e MARIADB_PASSWORD=insight \
             -e MARIADB_ROOT_PASSWORD=root \
             -p 3306:3306 mariadb:11.4
+          docker run -d --name insight-redis -p 6379:6379 redis:7-alpine
           for i in $(seq 1 60); do
             if docker exec insight-maria healthcheck.sh --connect --innodb_initialized 2>/dev/null; then
               echo "MariaDB ready"; break
@@ -238,6 +239,13 @@ jobs:
               cargo llvm-cov run --no-report --package "${{ matrix.entry.package }}" $feats -- \
                 -c "services/${{ matrix.entry.name }}/config/insight.yaml" migrate
             cargo llvm-cov --no-report --package "${{ matrix.entry.package }}" $feats -- --include-ignored
+            # Redis-backed cache live tests: opt-in via the env var and run
+            # serially — flush_all wipes the shared keyspace, so parallel
+            # cache tests race each other (the other live tests never see
+            # the env var and keep their parallel run above).
+            INTEGRATION_TESTS_REDIS_URL="redis://127.0.0.1:6379" \
+              cargo llvm-cov --no-report --package "${{ matrix.entry.package }}" $feats -- \
+                --include-ignored --test-threads=1 infra::cache::live_tests
           else
             cargo llvm-cov --no-report --package "${{ matrix.entry.package }}" $feats
           fi

@@ -314,7 +314,7 @@ The system **MUST** expose `GET /internal/authz` on the main listener as the gat
 
 The response carries no correlation id -- it is cacheable, so per-request correlation ids are generated at the edge (see [Gateway DESIGN](../gateway/DESIGN.md)).
 
-**Rationale**: This endpoint replaces the deleted Router's in-process session check + JWT injection; the Cache-Control contract keeps the authenticator in control of gateway-side staleness (revocation takes effect at the gateway within `authz_cache_max_age`, default 30 s).
+**Rationale**: This endpoint is the gateway's session check + JWT injection; the Cache-Control contract keeps the authenticator in control of gateway-side staleness (revocation takes effect at the gateway within `authz_cache_max_age`, default 30 s).
 
 **Actors**: `cpt-insightspec-actor-nginx-gateway`
 
@@ -487,7 +487,7 @@ Viewer identity remains exclusively gateway-authored: no client-supplied header
 
 - [ ] `p2` - **ID**: `cpt-insightspec-nfr-auth-exchange-p95`
 
-The `/internal/authz` exchange (token mapping + session + JWT reads) **MUST** complete within 5 ms p95 under normal load, keeping total gateway overhead comfortably inside the 15 ms p95 budget the deleted gateway spec carried.
+The `/internal/authz` exchange (token mapping + session + JWT reads) **MUST** complete within 5 ms p95 under normal load, keeping total gateway overhead comfortably inside a 15 ms p95 budget.
 
 **Threshold**: 5 ms p95 for the exchange; two Redis reads on the hot path.
 

@@ -219,7 +219,7 @@ mod tests {
 
         async fn invalidate(&self, tenant_id: Uuid, mode: InvalidateMode) -> anyhow::Result<()> {
             // Tenant-prefix purge: remove every entry whose tenant matches.
-            // Matches the production `SCAN cat:v1:{tenant}:* + UNLINK`.
+            // Matches the production `UNLINK cat:v1:{tenant}` hash drop.
             let mut g = self
                 .store
                 .lock()

@@ -31,9 +31,9 @@
 //!   below `hash-max-listpack-entries` (default 128). A tenant with ≤ 128
 //!   distinct `(role, team)` combinations pays one allocation, not one per
 //!   entry.
-//! - `flush_all()` still uses `SCAN cat:v1:* + UNLINK`, but now enumerates
-//!   tenant hashes (a small number) rather than per-entry keys (potentially
-//!   thousands).
+//! - `flush_all()` walks the `cat:v1:tenants` registry set and `UNLINK`s
+//!   per key — no `SCAN` cursor and no multi-key commands, both invalid on
+//!   Redis Cluster.
 //!
 //! DESIGN §3.2's swap-ability OQ (§4 γ) explicitly allows changing the cache
 //! mechanism behind the trait — the public surface
@@ -63,6 +63,7 @@ use std::sync::Mutex;
 use std::time::{Duration, Instant};
 
 use async_trait::async_trait;
+use futures::StreamExt as _;
 use redis::AsyncCommands;
 use uuid::Uuid;
 
@@ -74,6 +75,14 @@ use crate::domain::catalog::response::CatalogResponse;
 /// admin-write invalidations both walk.
 pub const CACHE_KEY_PREFIX: &str = "cat:v1:";
 
+/// Registry of live tenant hash keys — `flush_all` walks this instead of a
+/// keyspace `SCAN`. Stale members cost a no-op `UNLINK`; size is bounded by
+/// the number of tenants ever cached.
+const TENANT_REGISTRY_KEY: &str = "cat:v1:tenants";
+
+/// Concurrent `UNLINK`s in flight during `flush_all`.
+const FLUSH_UNLINK_CONCURRENCY: usize = 16;
+
 /// Default per-entry TTL — internal to this module. PRD §5.3
 /// `cpt-metric-cat-fr-cache` mandates 5 minutes; admin writes invalidate
 /// ahead of TTL so users don't observe "I changed the threshold, nothing
@@ -316,39 +325,6 @@ impl RedisCatalogCache {
             skip_until: SkipUntilMap::default(),
         })
     }
-
-    /// `SCAN MATCH pattern + UNLINK` — used only by `flush_all`, which has
-    /// to enumerate tenant hashes. NEVER `KEYS`, NEVER `FLUSHDB`. `UNLINK`
-    /// is preferred over `DEL` because it is asynchronous on the server
-    /// side and won't block large purges.
-    ///
-    /// Per-tenant invalidation does NOT use this — it's a single
-    /// `UNLINK cat:v1:{tenant}` against the hash key (see `invalidate`).
-    async fn scan_and_unlink(&self, pattern: &str) -> anyhow::Result<()> {
-        let mut conn = self.conn.clone();
-        let mut cursor: u64 = 0;
-        loop {
-            let (next, batch): (u64, Vec<String>) = redis::cmd("SCAN")
-                .arg(cursor)
-                .arg("MATCH")
-                .arg(pattern)
-                .arg("COUNT")
-                .arg(100)
-                .query_async(&mut conn)
-                .await?;
-            if !batch.is_empty() {
-                let _: i64 = redis::cmd("UNLINK")
-                    .arg(&batch)
-                    .query_async(&mut conn)
-                    .await?;
-            }
-            cursor = next;
-            if cursor == 0 {
-                break;
-            }
-        }
-        Ok(())
-    }
 }
 
 #[async_trait]
@@ -415,6 +391,10 @@ impl CatalogCache for RedisCatalogCache {
         let field = cache_field(role_slug, team_id);
         let mut conn = self.conn.clone();
         let bytes = serde_json::to_vec(payload)?;
+        // INVARIANT: register before writing, so the registry is always a
+        // superset of live keys. Separate command — different slot on a
+        // cluster, so it can't join the pipeline below.
+        let _: i64 = conn.sadd(TENANT_REGISTRY_KEY, &hash_key).await?;
         // `HSET` has no TTL of its own. Refresh the per-hash TTL with
         // `EXPIRE` after every write so an actively-read tenant keeps its
         // cache warm; a quiet tenant lets its hash expire cleanly. Both
@@ -445,9 +425,29 @@ impl CatalogCache for RedisCatalogCache {
     }
 
     async fn flush_all(&self) -> anyhow::Result<()> {
-        // `cat:v1:*` — NEVER `FLUSHDB`. The Redis instance is shared with
-        // sibling namespaces and a global flush would clobber them.
-        self.scan_and_unlink(&format!("{CACHE_KEY_PREFIX}*")).await
+        // INVARIANT: NEVER `KEYS`, NEVER `FLUSHDB` — the instance is shared
+        // with sibling namespaces.
+        let mut conn = self.conn.clone();
+        let keys: Vec<String> = conn.smembers(TENANT_REGISTRY_KEY).await?;
+        if keys.is_empty() {
+            return Ok(());
+        }
+
+        // One key per UNLINK (a batch could span cluster slots), issued
+        // concurrently on the multiplexed connection, bounded in flight.
+        let mut unlinks = futures::stream::iter(keys.iter().cloned().map(|key| {
+            let mut conn = self.conn.clone();
+            async move { conn.unlink::<_, i64>(&key).await }
+        }))
+        .buffer_unordered(FLUSH_UNLINK_CONCURRENCY);
+        while let Some(res) = unlinks.next().await {
+            res?;
+        }
+
+        // INVARIANT: SREM only what we enumerated — a concurrent `put` may
+        // have just registered a key we must not drop.
+        let _: i64 = conn.srem(TENANT_REGISTRY_KEY, &keys).await?;
+        Ok(())
     }
 
     fn should_skip(&self, tenant_id: Uuid) -> bool {

@@ -1,11 +1,12 @@
 //! Index janitor (PRD 5.5.8, DESIGN §4.3).
 //!
-//! Per-key Redis TTLs remove session records and token mappings, but ZSET
-//! index members (`asm:user_sessions:*`) and refresh-schedule orphans linger
-//! until trimmed. One leader (Redis lock, DD-BFF-09 — same election as the
-//! refresher) runs a pass every `janitor_interval_seconds` (default 30 s) and
-//! emits removed/backlog metrics; a rising backlog means no pod is running
-//! passes.
+//! Per-key Redis TTLs remove session records and token mappings; the
+//! login-state live index and refresh-schedule orphans linger until trimmed.
+//! Per-user session indexes are trimmed inline by writers and TTL-bounded,
+//! so the pass issues no keyspace SCAN. One leader (Redis lock, DD-BFF-09 —
+//! same election as the refresher) runs a pass every
+//! `janitor_interval_seconds` (default 30 s) and emits removed/backlog
+//! metrics; a rising backlog means no pod is running passes.
 
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};

@@ -214,10 +214,31 @@ impl SessionManager {
     pub async fn connect(redis_url: &str) -> anyhow::Result<Self> {
         anyhow::ensure!(!redis_url.is_empty(), "redis_url is required (fail closed)");
         let client = redis::Client::open(redis_url).context("open Redis client")?;
-        let conn = client
+        let mut conn = client
             .get_connection_manager()
             .await
             .context("establish Redis connection manager")?;
+
+        // EXPIREAT NX|GT needs Redis >= 7.0; fail at boot, not per login.
+        let info: String = redis::cmd("INFO")
+            .arg("server")
+            .query_async(&mut conn)
+            .await
+            .context("read Redis server info")?;
+        let version = info
+            .lines()
+            .find_map(|l| l.strip_prefix("redis_version:"))
+            .map_or("", str::trim);
+        let major: u64 = version
+            .split('.')
+            .next()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(0);
+        anyhow::ensure!(
+            major >= 7,
+            "Redis >= 7.0 required (EXPIREAT NX/GT), server reports {version:?}"
+        );
+
         Ok(Self { conn })
     }
 
@@ -377,12 +398,27 @@ impl SessionManager {
                 .with_expiration(redis::SetExpiry::EX(s.jwt_reissue_after_seconds)),
         )
         .ignore();
-        // User-session index (score = expiry).
-        pipe.zadd(user_sessions_key(&r.person_id), &s.session_id, expires_at)
+        // User-session index (score = expiry): inline trim + guarded TTL
+        // stand in for the removed janitor SCAN.
+        let ukey = user_sessions_key(&r.person_id);
+        let created = i64::try_from(r.created_at).unwrap_or(0);
+        let absolute = i64::try_from(r.absolute_expires_at).unwrap_or(i64::MAX);
+        pipe.zrembyscore(&ukey, 0, created).ignore();
+        pipe.zadd(&ukey, &s.session_id, expires_at).ignore();
+        // INVARIANT: index TTL >= every member's absolute expiry. NX seeds
+        // a TTL (GT treats no-TTL as infinite); GT only ever extends.
+        pipe.cmd("EXPIREAT")
+            .arg(&ukey)
+            .arg(absolute)
+            .arg("NX")
+            .ignore();
+        pipe.cmd("EXPIREAT")
+            .arg(&ukey)
+            .arg(absolute)
+            .arg("GT")
             .ignore();
         // Back-channel logout indexes: by OIDC `sid` (when the IdP supplies
         // one) and by `(iss, sub)` — the sub-only fallback path.
-        let absolute = i64::try_from(r.absolute_expires_at).unwrap_or(i64::MAX);
         // A view-as session is ALSO indexed under the real principal, so
         // revoke-by-person against the impersonator (admin deprovisioning,
         // self "log out everywhere") reaches it. Scored at the absolute cap:
@@ -391,22 +427,49 @@ impl SessionManager {
         // Reads stay correct — a dead session's record is gone, so index
         // readers skip it.
         if !r.impersonator_person_id.is_empty() {
-            pipe.zadd(
-                user_sessions_key(&r.impersonator_person_id),
-                &s.session_id,
-                absolute,
-            )
-            .ignore();
+            let ikey = user_sessions_key(&r.impersonator_person_id);
+            pipe.zrembyscore(&ikey, 0, created).ignore();
+            pipe.zadd(&ikey, &s.session_id, absolute).ignore();
+            pipe.cmd("EXPIREAT")
+                .arg(&ikey)
+                .arg(absolute)
+                .arg("NX")
+                .ignore();
+            pipe.cmd("EXPIREAT")
+                .arg(&ikey)
+                .arg(absolute)
+                .arg("GT")
+                .ignore();
         }
         if let Some(sid) = &r.idp_sid {
             let idx = sid_index_key(&r.idp_iss, sid);
             pipe.sadd(&idx, &s.session_id).ignore();
-            pipe.expire_at(&idx, absolute).ignore();
+            // INVARIANT: NX/GT, never plain EXPIREAT — the set is shared and
+            // a shorter-lived session must not cut the TTL under a live one.
+            pipe.cmd("EXPIREAT")
+                .arg(&idx)
+                .arg(absolute)
+                .arg("NX")
+                .ignore();
+            pipe.cmd("EXPIREAT")
+                .arg(&idx)
+                .arg(absolute)
+                .arg("GT")
+                .ignore();
         }
         if !r.idp_sub.is_empty() {
             let idx = sub_index_key(&r.idp_iss, &r.idp_sub);
             pipe.sadd(&idx, &s.session_id).ignore();
-            pipe.expire_at(&idx, absolute).ignore();
+            pipe.cmd("EXPIREAT")
+                .arg(&idx)
+                .arg(absolute)
+                .arg("NX")
+                .ignore();
+            pipe.cmd("EXPIREAT")
+                .arg(&idx)
+                .arg(absolute)
+                .arg("GT")
+                .ignore();
         }
         // IdP refresh schedule (consumer lands in step 10).
         if let Some(due) = s.refresh_due_at {
@@ -759,11 +822,11 @@ impl SessionManager {
         Ok(())
     }
 
-    /// One janitor pass (DESIGN §4.3): trim expired members from every
-    /// `asm:user_sessions:*` ZSET (`ZREMRANGEBYSCORE 0 now` — per-key TTLs
-    /// removed the records, the index members linger) and drop long-overdue
-    /// orphans from the refresh schedule (live sessions are re-scheduled by
-    /// the refresher; an entry still due after `orphan_grace` has no owner).
+    /// One janitor pass (DESIGN §4.3): trim expired members from the
+    /// login-state index and drop long-overdue orphans from the refresh
+    /// schedule (live sessions are re-scheduled by the refresher; an entry
+    /// still due after `orphan_grace` has no owner). Per-user session
+    /// indexes are trimmed inline by writers and TTL-bounded — no SCAN.
     /// Returns (removed members, overdue-backlog size before trimming).
     ///
     /// # Errors
@@ -774,38 +837,6 @@ impl SessionManager {
         let mut removed = 0u64;
         let mut backlog = 0u64;
 
-        // SCAN, never KEYS — bounded batches on a shared Redis.
-        let mut cursor: u64 = 0;
-        loop {
-            let (next, keys): (u64, Vec<String>) = redis::cmd("SCAN")
-                .arg(cursor)
-                .arg("MATCH")
-                .arg("asm:user_sessions:*")
-                .arg("COUNT")
-                .arg(100)
-                .query_async(&mut conn)
-                .await
-                .context("scan user-session indexes")?;
-            for key in keys {
-                let expired: u64 = conn
-                    .zcount(&key, 0, now_i)
-                    .await
-                    .context("count expired index members")?;
-                if expired > 0 {
-                    backlog += expired;
-                    let n: u64 = conn
-                        .zrembyscore(&key, 0, now_i)
-                        .await
-                        .context("trim expired index members")?;
-                    removed += n;
-                }
-            }
-            cursor = next;
-            if cursor == 0 {
-                break;
-            }
-        }
-
         // Expired login-state index members (the HASH keys expired via TTL).
         let stale_states: u64 = conn
             .zrembyscore(LOGIN_STATE_LIVE_KEY, 0, now_i)
@@ -927,7 +958,7 @@ impl SessionManager {
 
     /// List a person's live sessions from the per-user index (score > `now`),
     /// loading each record. Index members whose record has already expired are
-    /// skipped (the janitor trims them).
+    /// skipped (the next session-create on this index trims them).
     ///
     /// # Errors
     /// Fails on a Redis error.