cpex_session_valkey/
store.rs

1// Location: ./builtins/session/valkey/src/store.rs
2// Copyright 2026
3// SPDX-License-Identifier: Apache-2.0
4// Authors: Fred Araujo
5//
6// `ValkeySessionStore` — the Valkey-backed `SessionStore`. Labels live in
7// a Redis SET per session so `append_labels` is a single atomic
8// server-side union (`SADD`), never a client-side read-modify-write that
9// would lose labels under concurrent cross-node appends.
10//
11// # Fail-closed mapping
12//
13//   - `SMEMBERS` on a missing key returns an empty set → `Ok(empty)`
14//     (unknown session). It is NOT an error.
15//   - connection/timeout/protocol/decode failures → `Err(Backend)` so the
16//     caller fails the request closed.
17//
18// # Sliding TTL
19//
20// `append_labels` issues `SADD` + `EXPIRE` in one atomic pipeline.
21// `load_labels` refreshes the TTL fail-open: the read already succeeded,
22// so a refresh failure is alarmed but the labels are still returned.
23
24use std::fmt::Write as _;
25use std::time::Duration;
26
27use apl_cpex::{SessionStore, SessionStoreError};
28use async_trait::async_trait;
29use deadpool_redis::{Connection, Pool};
30use redis::AsyncCommands;
31use sha2::{Digest, Sha256};
32
33use crate::config::ValkeyConfig;
34use crate::connection::build_pool;
35use crate::error::BuildError;
36
37/// Valkey-backed session label store.
38pub struct ValkeySessionStore {
39    pool: Pool,
40    key_prefix: String,
41    ttl_seconds: Option<u64>,
42    connect_timeout: Duration,
43    command_timeout: Duration,
44}
45
46impl ValkeySessionStore {
47    /// Build from validated config. The pool is created lazily, so this
48    /// does not dial Valkey — connection failures surface on first use
49    /// and correctly fail the request closed.
50    pub fn from_config(cfg: &ValkeyConfig) -> Result<Self, BuildError> {
51        Ok(Self {
52            pool: build_pool(cfg)?,
53            key_prefix: cfg.key_prefix.clone(),
54            ttl_seconds: cfg.ttl_seconds,
55            connect_timeout: Duration::from_millis(cfg.connect_timeout_ms),
56            command_timeout: Duration::from_millis(cfg.command_timeout_ms),
57        })
58    }
59
60    /// Key schema: `<prefix>:<hex(sha256(session_id))>`. The full-width
61    /// digest keeps the Valkey keyspace collision-free and removes raw
62    /// session ids from it.
63    fn key(&self, session_id: &str) -> String {
64        let mut hasher = Sha256::new();
65        hasher.update(session_id.as_bytes());
66        let digest = hasher.finalize();
67        let mut hex = String::with_capacity(digest.len() * 2);
68        for byte in digest {
69            let _ = write!(hex, "{byte:02x}");
70        }
71        format!("{}:{}", self.key_prefix, hex)
72    }
73
74    /// Acquire a pooled connection, bounded by the connect timeout (the
75    /// fail-fast knob for a dead/slow endpoint, distinct from the
76    /// per-command timeout applied to SMEMBERS/SADD below).
77    async fn conn(&self) -> Result<Connection, SessionStoreError> {
78        match tokio::time::timeout(self.connect_timeout, self.pool.get()).await {
79            Ok(Ok(conn)) => Ok(conn),
80            Ok(Err(e)) => Err(backend(e)),
81            Err(_) => Err(SessionStoreError::Backend(
82                "valkey connection acquire timed out".to_string(),
83            )),
84        }
85    }
86}
87
88/// Map any backend failure to the fail-closed `SessionStoreError`.
89fn backend(e: impl std::fmt::Display) -> SessionStoreError {
90    SessionStoreError::Backend(e.to_string())
91}
92
93#[async_trait]
94impl SessionStore for ValkeySessionStore {
95    async fn load_labels(&self, session_id: &str) -> Result<Vec<String>, SessionStoreError> {
96        let key = self.key(session_id);
97        let mut conn = self.conn().await?;
98
99        // SMEMBERS on a missing key returns an empty set (Ok), so an
100        // unknown session naturally maps to Ok(empty). Only a real
101        // backend failure becomes Err.
102        let labels: Vec<String> =
103            match tokio::time::timeout(self.command_timeout, conn.smembers(&key)).await {
104                Ok(res) => res.map_err(backend)?,
105                Err(_) => {
106                    return Err(SessionStoreError::Backend(
107                        "valkey SMEMBERS timed out".to_string(),
108                    ))
109                },
110            };
111
112        // Sliding-TTL refresh is fail-open for the read: the labels were
113        // read successfully, so a refresh failure is alarmed, not failed
114        // closed. A persistently-failing refresh risks silent key
115        // expiry across requests — see the operator runbook.
116        if let Some(ttl) = self.ttl_seconds {
117            let refresh: Result<bool, _> =
118                match tokio::time::timeout(self.command_timeout, conn.expire(&key, ttl as i64))
119                    .await
120                {
121                    Ok(res) => res,
122                    Err(_) => Ok(false), // treat timeout as a failed refresh
123                };
124            if let Err(e) = refresh {
125                tracing::warn!(
126                    alarm = "session_store_ttl_refresh_failed",
127                    error = %e,
128                    "valkey TTL refresh on load failed; returning read labels (fail-open)"
129                );
130            }
131        }
132
133        Ok(labels)
134    }
135
136    async fn append_labels(
137        &self,
138        session_id: &str,
139        labels: &[String],
140    ) -> Result<(), SessionStoreError> {
141        if labels.is_empty() {
142            return Ok(());
143        }
144        let key = self.key(session_id);
145        let mut conn = self.conn().await?;
146
147        // Atomic server-side union + optional TTL refresh in one round
148        // trip (MULTI/EXEC). SADD is a commutative merge, so concurrent
149        // cross-node appends never lose labels.
150        let mut pipe = redis::pipe();
151        pipe.atomic();
152        pipe.sadd(&key, labels).ignore();
153        if let Some(ttl) = self.ttl_seconds {
154            pipe.expire(&key, ttl as i64).ignore();
155        }
156
157        match tokio::time::timeout(self.command_timeout, pipe.query_async::<()>(&mut conn)).await {
158            Ok(res) => res.map_err(backend)?,
159            Err(_) => {
160                return Err(SessionStoreError::Backend(
161                    "valkey append (SADD+EXPIRE) timed out".to_string(),
162                ))
163            },
164        }
165        Ok(())
166    }
167}
cpex_session_valkey/store.rs

cpex_session_valkey/
store.rs
