diff --git a/crates/headroom-core/src/ccr/backends/in_memory.rs b/crates/headroom-core/src/ccr/backends/in_memory.rs index efb6b5376..d14b04d3d 100644 --- a/crates/headroom-core/src/ccr/backends/in_memory.rs +++ b/crates/headroom-core/src/ccr/backends/in_memory.rs @@ -15,13 +15,16 @@ use std::time::{Duration, Instant}; use dashmap::DashMap; -use crate::ccr::{CcrStore, DEFAULT_CAPACITY, DEFAULT_TTL}; +use crate::ccr::{max_lifetime_for, CcrStore, DEFAULT_CAPACITY, DEFAULT_TTL}; /// In-memory CCR store backed by [`DashMap`] for sharded concurrent /// access. /// -/// - **TTL**: 30 minutes by default. Entries past their TTL are dropped -/// on the next `get` (lazy expiry — no background reaper thread). +/// - **TTL**: 30 minutes by default, treated as an **idle window** — +/// every successful `get` restarts the entry's clock (#2604), bounded +/// by an absolute max lifetime of 8x the idle TTL measured from +/// insertion. Entries past their window are dropped on the next `get` +/// (lazy expiry — no background reaper thread). /// - **Capacity**: 1000 entries by default. When `put` would push us /// past capacity, the oldest entry (per insertion order) is evicted. /// - **Concurrency**: gets and puts on distinct keys do not contend. @@ -36,6 +39,7 @@ pub struct InMemoryCcrStore { /// they actually evict a real entry. order: Mutex>, ttl: Duration, + max_lifetime: Duration, capacity: usize, } @@ -43,19 +47,38 @@ pub struct InMemoryCcrStore { struct Entry { payload: String, inserted: Instant, + last_accessed: Instant, +} + +impl Entry { + /// Expired when idle past `ttl` OR older (since insertion) than + /// `max_lifetime` — the absolute ceiling that keeps constant access + /// from pinning an entry forever. + fn is_expired(&self, ttl: Duration, max_lifetime: Duration) -> bool { + self.last_accessed.elapsed() > ttl || self.inserted.elapsed() > max_lifetime + } } impl InMemoryCcrStore { - /// Default: 1000 entries, 30-minute TTL. + /// Default: 1000 entries, 30-minute idle TTL (8x max lifetime). pub fn new() -> Self { Self::with_capacity_and_ttl(DEFAULT_CAPACITY, DEFAULT_TTL) } + /// `ttl` is the idle window; the absolute max lifetime defaults to + /// 8x that (see [`crate::ccr::DEFAULT_MAX_LIFETIME_MULTIPLIER`]). pub fn with_capacity_and_ttl(capacity: usize, ttl: Duration) -> Self { + Self::with_capacity_and_ttls(capacity, ttl, max_lifetime_for(ttl)) + } + + /// Full-control constructor: idle window and absolute max lifetime + /// specified independently. + pub fn with_capacity_and_ttls(capacity: usize, ttl: Duration, max_lifetime: Duration) -> Self { Self { map: DashMap::with_capacity(capacity), order: Mutex::new(VecDeque::with_capacity(capacity)), ttl, + max_lifetime, capacity, } } @@ -89,8 +112,10 @@ impl CcrStore for InMemoryCcrStore { // in place, leave the order queue alone. Common when the same // tool output flows through multiple times in a session. if let Some(mut existing) = self.map.get_mut(hash) { + let now = Instant::now(); existing.payload = payload.to_string(); - existing.inserted = Instant::now(); + existing.inserted = now; + existing.last_accessed = now; return; } @@ -99,9 +124,11 @@ impl CcrStore for InMemoryCcrStore { if self.map.len() >= self.capacity { self.evict_until_under_capacity(); } + let now = Instant::now(); let entry = Entry { payload: payload.to_string(), - inserted: Instant::now(), + inserted: now, + last_accessed: now, }; let prev = self.map.insert(hash.to_string(), entry); if prev.is_none() { @@ -117,9 +144,12 @@ impl CcrStore for InMemoryCcrStore { } fn get(&self, hash: &str) -> Option { - // Read path: shard read-lock, check TTL, clone payload out. - // No global lock involvement at all — distinct hashes hash to - // distinct shards and never contend. + // Hit path: shard write-lock (get_mut), check the idle window + + // max-lifetime ceiling, refresh `last_accessed`, clone payload + // out. The TTL is a sliding idle window (#2604): every hit + // restarts the clock, so an entry a session keeps touching does + // not expire mid-burst. Distinct hashes hash to distinct shards + // and never contend. // // Lazy expiry uses DashMap's `remove_if` so the check-and-remove // is atomic on the shard. An earlier 2-step (drop read lock, @@ -130,8 +160,9 @@ impl CcrStore for InMemoryCcrStore { // load this manifested as "I just stored it; why is it gone?" // `remove_if` closes the window because the shard write lock // is held across both the predicate evaluation and the removal. - if let Some(entry) = self.map.get(hash) { - if entry.inserted.elapsed() <= self.ttl { + if let Some(mut entry) = self.map.get_mut(hash) { + if !entry.is_expired(self.ttl, self.max_lifetime) { + entry.last_accessed = Instant::now(); return Some(entry.payload.clone()); } } else { @@ -143,7 +174,9 @@ impl CcrStore for InMemoryCcrStore { // and re-fetch its payload. let was_removed = self .map - .remove_if(hash, |_, entry| entry.inserted.elapsed() > self.ttl) + .remove_if(hash, |_, entry| { + entry.is_expired(self.ttl, self.max_lifetime) + }) .is_some(); if was_removed { None diff --git a/crates/headroom-core/src/ccr/backends/redis.rs b/crates/headroom-core/src/ccr/backends/redis.rs index d0070ec39..a05c8b45a 100644 --- a/crates/headroom-core/src/ccr/backends/redis.rs +++ b/crates/headroom-core/src/ccr/backends/redis.rs @@ -10,10 +10,12 @@ //! # Storage model //! //! Each entry maps to a Redis key `ccr:{hash}` containing the original -//! payload bytes, with a `SETEX` TTL applied on every write. Read path -//! is a single `GET`. Redis handles purging via key expiry — no -//! application-side sweep needed (matching the SQLite backend's -//! lazy-purge but at the Redis level). +//! payload bytes, with a `SETEX` TTL applied on every write. The TTL is +//! an **idle window** (#2604): every successful `get` re-arms the key's +//! expiry, bounded by an absolute max lifetime tracked in a companion +//! `ccr:{hash}:born` key whose own expiry marks the ceiling. Redis +//! handles purging via key expiry — no application-side sweep needed +//! (matching the SQLite backend's lazy-purge but at the Redis level). //! //! # Concurrency //! @@ -27,7 +29,7 @@ use redis::Commands; -use crate::ccr::CcrStore; +use crate::ccr::{max_lifetime_for, CcrStore}; /// Key prefix applied to every CCR entry. Configurable per-deployment /// so multiple proxies sharing one Redis don't collide. @@ -38,6 +40,9 @@ pub struct RedisCcrStore { client: redis::Client, key_prefix: String, default_ttl_seconds: u64, + /// Absolute max lifetime (seconds since `put`) that caps the + /// sliding idle window. Defaults to 8x the idle TTL. + max_lifetime_seconds: u64, } impl RedisCcrStore { @@ -59,10 +64,13 @@ impl RedisCcrStore { // signal. let mut conn = client.get_connection()?; let _: String = redis::cmd("PING").query(&mut conn)?; + let max_lifetime_seconds = + max_lifetime_for(std::time::Duration::from_secs(default_ttl_seconds)).as_secs(); Ok(Self { client, key_prefix, default_ttl_seconds, + max_lifetime_seconds, }) } @@ -70,6 +78,12 @@ impl RedisCcrStore { format!("{}:{}", self.key_prefix, hash) } + /// Companion key whose expiry marks the entry's absolute max + /// lifetime; its remaining TTL caps every idle-window re-arm. + fn born_key_for(&self, hash: &str) -> String { + format!("{}:{}:born", self.key_prefix, hash) + } + /// Default TTL (seconds) applied on every `put`. pub fn default_ttl_seconds(&self) -> u64 { self.default_ttl_seconds @@ -102,6 +116,20 @@ impl CcrStore for RedisCcrStore { error = %err, "ccr_redis_put_failed" ); + return; + } + // Companion max-lifetime marker: its remaining TTL caps every + // idle-window re-arm in `get`, so constant access cannot pin an + // entry past `max_lifetime_seconds`. + let born: redis::RedisResult<()> = + conn.set_ex(self.born_key_for(hash), 1_u8, self.max_lifetime_seconds); + if let Err(err) = born { + tracing::warn!( + target = "ccr.redis", + hash = %hash, + error = %err, + "ccr_redis_put_born_failed" + ); } } @@ -120,9 +148,9 @@ impl CcrStore for RedisCcrStore { } }; let bytes: redis::RedisResult>> = conn.get(&key); - match bytes { - Ok(Some(bytes)) => String::from_utf8(bytes).ok(), - Ok(None) => None, + let payload = match bytes { + Ok(Some(bytes)) => String::from_utf8(bytes).ok()?, + Ok(None) => return None, Err(err) => { tracing::warn!( target = "ccr.redis", @@ -130,9 +158,48 @@ impl CcrStore for RedisCcrStore { error = %err, "ccr_redis_get_failed" ); - None + return None; } + }; + + // Sliding idle window (#2604): re-arm the key's expiry on every + // hit, capped by the companion born-key's remaining lifetime. + let born_key = self.born_key_for(hash); + let born_remaining: i64 = conn.ttl(&born_key).unwrap_or(-1); + let remaining = if born_remaining >= 0 { + born_remaining as u64 + } else { + // Legacy entry written by a pre-sliding build (no born key): + // backfill the ceiling from now rather than dropping data. + let backfill: redis::RedisResult<()> = + conn.set_ex(&born_key, 1_u8, self.max_lifetime_seconds); + if let Err(err) = backfill { + tracing::warn!( + target = "ccr.redis", + hash = %hash, + error = %err, + "ccr_redis_born_backfill_failed" + ); + } + self.max_lifetime_seconds + }; + let new_ttl = self.default_ttl_seconds.min(remaining); + if new_ttl == 0 { + // Past the max lifetime: purge rather than serve a pinned + // entry that should have died. + let _: redis::RedisResult<()> = conn.del(&key); + return None; } + let rearm: redis::RedisResult<()> = conn.expire(&key, new_ttl as i64); + if let Err(err) = rearm { + tracing::warn!( + target = "ccr.redis", + hash = %hash, + error = %err, + "ccr_redis_ttl_rearm_failed" + ); + } + Some(payload) } fn len(&self) -> usize { diff --git a/crates/headroom-core/src/ccr/backends/sqlite.rs b/crates/headroom-core/src/ccr/backends/sqlite.rs index 14cc32cc3..906ecfca4 100644 --- a/crates/headroom-core/src/ccr/backends/sqlite.rs +++ b/crates/headroom-core/src/ccr/backends/sqlite.rs @@ -5,16 +5,21 @@ //! //! ```sql //! CREATE TABLE IF NOT EXISTS ccr_entries ( -//! hash TEXT PRIMARY KEY, -//! original BLOB NOT NULL, -//! created_at INTEGER NOT NULL, -- unix-seconds -//! ttl_seconds INTEGER NOT NULL +//! hash TEXT PRIMARY KEY, +//! original BLOB NOT NULL, +//! created_at INTEGER NOT NULL, -- unix-seconds +//! ttl_seconds INTEGER NOT NULL, -- idle window, restarted on get +//! last_accessed INTEGER NOT NULL -- unix-seconds //! ); //! ``` //! -//! On every `get` we lazy-purge stale rows -//! (`WHERE created_at + ttl_seconds <= now`) — no background reaper -//! thread, no cron. +//! The TTL is an **idle window** (#2604): every successful `get` +//! restarts the row's clock via `last_accessed`, bounded by an absolute +//! max lifetime measured from `created_at`. On every `get` we +//! lazy-purge stale rows (`WHERE last_accessed + ttl_seconds <= now OR +//! created_at + max_lifetime <= now`) — no background reaper thread, +//! no cron. DBs created by pre-sliding builds are migrated in place +//! (the `last_accessed` column is added, backfilled from `created_at`). //! //! All hot statements are prepared once on connection setup and reused //! per call (per realignment build constraint #5: performant). Writes @@ -43,14 +48,17 @@ use std::time::{SystemTime, UNIX_EPOCH}; use rusqlite::{params, Connection, OptionalExtension}; -use crate::ccr::CcrStore; +use crate::ccr::{max_lifetime_for, CcrStore}; /// SQLite-backed CCR store. pub struct SqliteCcrStore { conn: Mutex, - /// Default TTL applied on every `put`. Mirrors Python's - /// `compression_store` 30-minute window. + /// Default idle TTL applied on every `put`. Mirrors Python's + /// `compression_store` idle window. default_ttl_seconds: u64, + /// Absolute max lifetime (seconds since `created_at`) that caps the + /// sliding idle window. Defaults to 8x the idle TTL. + max_lifetime_seconds: u64, /// Path the connection was opened against — kept for diagnostics /// and for the proxy-restart simulation test. path: PathBuf, @@ -58,9 +66,24 @@ pub struct SqliteCcrStore { impl SqliteCcrStore { /// Open or create the DB file at `path` and prepare the schema. + /// `default_ttl_seconds` is the idle window; the absolute max + /// lifetime defaults to 8x that (see + /// [`crate::ccr::DEFAULT_MAX_LIFETIME_MULTIPLIER`]). /// Errors surface to the caller (`from_config`); we never silently /// fall back to the in-memory backend (`feedback_no_silent_fallbacks.md`). pub fn open(path: impl AsRef, default_ttl_seconds: u64) -> rusqlite::Result { + let max_lifetime = + max_lifetime_for(std::time::Duration::from_secs(default_ttl_seconds)).as_secs(); + Self::open_with_ttls(path, default_ttl_seconds, max_lifetime) + } + + /// Full-control constructor: idle window and absolute max lifetime + /// specified independently. + pub fn open_with_ttls( + path: impl AsRef, + default_ttl_seconds: u64, + max_lifetime_seconds: u64, + ) -> rusqlite::Result { let path_buf = path.as_ref().to_path_buf(); let conn = Connection::open(&path_buf)?; @@ -73,25 +96,49 @@ impl SqliteCcrStore { conn.execute( "CREATE TABLE IF NOT EXISTS ccr_entries ( - hash TEXT PRIMARY KEY, - original BLOB NOT NULL, - created_at INTEGER NOT NULL, - ttl_seconds INTEGER NOT NULL + hash TEXT PRIMARY KEY, + original BLOB NOT NULL, + created_at INTEGER NOT NULL, + ttl_seconds INTEGER NOT NULL, + last_accessed INTEGER NOT NULL )", [], )?; + Self::migrate_legacy_schema(&conn)?; // No secondary index — the schema is one-row-per-PK and the only // non-PK lookup (the lazy-purge sweep) is a `WHERE` predicate on - // a small table; an index on `created_at + ttl_seconds` would - // cost more than it saves. + // a small table; an index on the expiry expressions would cost + // more than it saves. Ok(Self { conn: Mutex::new(conn), default_ttl_seconds, + max_lifetime_seconds, path: path_buf, }) } + /// DBs created before the sliding-TTL change lack `last_accessed`. + /// Add it in place and backfill from `created_at` so legacy rows + /// keep their original expiry baseline rather than being purged or + /// artificially refreshed. + fn migrate_legacy_schema(conn: &Connection) -> rusqlite::Result<()> { + let has_last_accessed = conn + .prepare("SELECT 1 FROM pragma_table_info('ccr_entries') WHERE name = 'last_accessed'")? + .exists([])?; + if !has_last_accessed { + conn.execute( + "ALTER TABLE ccr_entries ADD COLUMN last_accessed INTEGER NOT NULL DEFAULT 0", + [], + )?; + conn.execute( + "UPDATE ccr_entries SET last_accessed = created_at WHERE last_accessed = 0", + [], + )?; + } + Ok(()) + } + /// Path the connection was opened against. Test helper. pub fn path(&self) -> &Path { &self.path @@ -102,12 +149,15 @@ impl SqliteCcrStore { self.default_ttl_seconds } - /// Drop all expired rows. Lazy — invoked from `get`. Returns the + /// Drop all expired rows: idle past their window, or past the + /// absolute max lifetime. Lazy — invoked from `get`. Returns the /// number of rows purged. - fn purge_expired(conn: &Connection, now: u64) -> rusqlite::Result { + fn purge_expired(&self, conn: &Connection, now: u64) -> rusqlite::Result { let purged = conn.execute( - "DELETE FROM ccr_entries WHERE created_at + ttl_seconds <= ?1", - params![now as i64], + "DELETE FROM ccr_entries + WHERE last_accessed + ttl_seconds <= ?1 + OR created_at + ?2 <= ?1", + params![now as i64, self.max_lifetime_seconds as i64], )?; Ok(purged) } @@ -129,12 +179,13 @@ impl CcrStore for SqliteCcrStore { // Upsert by PK. ON CONFLICT REPLACE matches the in-memory // backend's idempotent re-store semantics. let res = conn.execute( - "INSERT INTO ccr_entries (hash, original, created_at, ttl_seconds) - VALUES (?1, ?2, ?3, ?4) + "INSERT INTO ccr_entries (hash, original, created_at, ttl_seconds, last_accessed) + VALUES (?1, ?2, ?3, ?4, ?3) ON CONFLICT(hash) DO UPDATE SET - original = excluded.original, - created_at = excluded.created_at, - ttl_seconds = excluded.ttl_seconds", + original = excluded.original, + created_at = excluded.created_at, + ttl_seconds = excluded.ttl_seconds, + last_accessed = excluded.last_accessed", params![ hash, payload.as_bytes(), @@ -165,7 +216,7 @@ impl CcrStore for SqliteCcrStore { // Lazy purge sweep, then the real lookup. Both happen under // the same mutex so the row we read is guaranteed not to have // been just-deleted by another caller. - if let Err(err) = Self::purge_expired(&conn, now) { + if let Err(err) = self.purge_expired(&conn, now) { tracing::warn!( target = "ccr.sqlite", error = %err, @@ -176,8 +227,10 @@ impl CcrStore for SqliteCcrStore { let row: Option> = conn .query_row( "SELECT original FROM ccr_entries - WHERE hash = ?1 AND created_at + ttl_seconds > ?2", - params![hash, now as i64], + WHERE hash = ?1 + AND last_accessed + ttl_seconds > ?2 + AND created_at + ?3 > ?2", + params![hash, now as i64, self.max_lifetime_seconds as i64], |r| r.get::<_, Vec>(0), ) .optional() @@ -191,7 +244,22 @@ impl CcrStore for SqliteCcrStore { None }); - row.and_then(|bytes| String::from_utf8(bytes).ok()) + let row = row?; + // Sliding idle window (#2604): a successful hit restarts the + // row's idle clock. Still under the same mutex as the lookup. + if let Err(err) = conn.execute( + "UPDATE ccr_entries SET last_accessed = ?2 WHERE hash = ?1", + params![hash, now as i64], + ) { + tracing::warn!( + target = "ccr.sqlite", + hash = %hash, + error = %err, + "ccr_sqlite_touch_failed" + ); + } + + String::from_utf8(row).ok() } fn len(&self) -> usize { diff --git a/crates/headroom-core/src/ccr/mod.rs b/crates/headroom-core/src/ccr/mod.rs index 2dd3ef28a..d810414ef 100644 --- a/crates/headroom-core/src/ccr/mod.rs +++ b/crates/headroom-core/src/ccr/mod.rs @@ -65,6 +65,20 @@ pub const DEFAULT_CAPACITY: usize = 1000; /// silently converts "lossless with retrieval" into "lossy". pub const DEFAULT_TTL: Duration = Duration::from_secs(1800); +/// The TTL is an **idle window**, not a wall clock: every successful +/// `get` restarts the entry's clock, so an entry a session keeps +/// touching survives a long multi-agent burst (#2604). To keep +/// constant access from pinning an entry forever, an absolute max +/// lifetime of `DEFAULT_MAX_LIFETIME_MULTIPLIER * ttl` (measured from +/// insertion) caps the sliding window. Mirrors the Python +/// `CompressionStore` semantics. +pub const DEFAULT_MAX_LIFETIME_MULTIPLIER: u32 = 8; + +/// Absolute max lifetime for an entry with idle window `idle_ttl`. +pub fn max_lifetime_for(idle_ttl: Duration) -> Duration { + idle_ttl.saturating_mul(DEFAULT_MAX_LIFETIME_MULTIPLIER) +} + /// Compute the canonical CCR key for `payload`. BLAKE3 → first 24 hex /// chars (96 bits — collision-resistant for the bounded LRU population /// the proxy will hold). Centralized here so every call site (live-zone diff --git a/crates/headroom-core/tests/ccr_backends.rs b/crates/headroom-core/tests/ccr_backends.rs index a3ad8f338..5730b0cae 100644 --- a/crates/headroom-core/tests/ccr_backends.rs +++ b/crates/headroom-core/tests/ccr_backends.rs @@ -152,6 +152,157 @@ fn backend_swap_byte_equal_keys() { } } +// ─── Sliding (idle-window) TTL semantics — #2604 ─────────────────────── +// +// The Python `CompressionStore` treats `HEADROOM_CCR_TTL_SECONDS` as an +// idle window that restarts on every successful retrieval, bounded by an +// absolute max lifetime (8x the idle TTL). These tests pin the same +// semantics onto the Rust backends so an entry a session keeps touching +// does not expire mid-burst. + +#[test] +fn in_memory_get_refreshes_idle_ttl() { + let store = InMemoryCcrStore::with_capacity_and_ttl(10, Duration::from_millis(120)); + let hash = compute_key(b"hot entry"); + store.put(&hash, "hot entry"); + // Touch the entry every 60ms for ~4 idle windows' worth of wall + // clock. Wall-clock expiry would kill it at 120ms; a sliding idle + // window keeps it alive because every hit restarts the clock. + for _ in 0..8 { + std::thread::sleep(Duration::from_millis(60)); + assert_eq!( + store.get(&hash).as_deref(), + Some("hot entry"), + "an entry accessed within its idle window must stay alive" + ); + } + // Now go idle past the window: the entry must expire. + std::thread::sleep(Duration::from_millis(200)); + assert_eq!( + store.get(&hash), + None, + "an entry idle past its window must expire" + ); +} + +#[test] +fn in_memory_max_lifetime_caps_sliding_window() { + // Idle TTL 40ms → max lifetime 320ms (8x). Constant access must not + // keep the entry alive forever. + let store = InMemoryCcrStore::with_capacity_and_ttl(10, Duration::from_millis(40)); + let hash = compute_key(b"immortal?"); + store.put(&hash, "immortal?"); + let deadline = std::time::Instant::now() + Duration::from_millis(600); + let mut expired = false; + while std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + if store.get(&hash).is_none() { + expired = true; + break; + } + } + assert!( + expired, + "constant access must not extend an entry past its max lifetime" + ); +} + +#[test] +fn sqlite_get_refreshes_idle_ttl() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("ccr.sqlite"); + // 3-second idle window (unix-second resolution needs whole seconds). + let store = SqliteCcrStore::open(&path, 3).expect("open sqlite store"); + let hash = compute_key(b"sliding sqlite"); + store.put(&hash, "sliding sqlite"); + // t+2s: hit inside the window — restarts the idle clock. + std::thread::sleep(Duration::from_millis(2_000)); + assert_eq!( + store.get(&hash).as_deref(), + Some("sliding sqlite"), + "first access within the idle window must hit" + ); + // t+4s: wall-clock expiry would have purged at t+3s; the refresh at + // t+2s must keep it alive until t+5s. + std::thread::sleep(Duration::from_millis(2_000)); + assert_eq!( + store.get(&hash).as_deref(), + Some("sliding sqlite"), + "an entry accessed within its idle window must stay alive past the wall-clock TTL" + ); + // Go idle past the window. + std::thread::sleep(Duration::from_millis(4_100)); + assert_eq!( + store.get(&hash), + None, + "an entry idle past its window must be purged" + ); +} + +#[test] +fn sqlite_max_lifetime_caps_sliding_window() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("ccr.sqlite"); + // Idle 2s with a 3s ceiling: constant access must not outlive t+3s. + let store = + SqliteCcrStore::open_with_ttls(&path, 2, 3).expect("open sqlite store with ceiling"); + let hash = compute_key(b"capped sqlite"); + store.put(&hash, "capped sqlite"); + std::thread::sleep(Duration::from_millis(1_500)); + assert_eq!( + store.get(&hash).as_deref(), + Some("capped sqlite"), + "entry inside idle window and ceiling must hit" + ); + // Keep touching, but cross the 3s ceiling. + std::thread::sleep(Duration::from_millis(2_600)); + assert_eq!( + store.get(&hash), + None, + "constant access must not extend an entry past its max lifetime" + ); +} + +#[test] +fn sqlite_migrates_legacy_schema_without_last_accessed() { + // A DB created by a pre-sliding-TTL build has no `last_accessed` + // column. Opening it must migrate in place and keep the rows + // retrievable (backfilling last_accessed from created_at). + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("ccr.sqlite"); + let payload = "legacy row"; + let hash = compute_key(payload.as_bytes()); + { + let conn = rusqlite::Connection::open(&path).expect("open raw connection"); + conn.execute( + "CREATE TABLE ccr_entries ( + hash TEXT PRIMARY KEY, + original BLOB NOT NULL, + created_at INTEGER NOT NULL, + ttl_seconds INTEGER NOT NULL + )", + [], + ) + .expect("create legacy schema"); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + conn.execute( + "INSERT INTO ccr_entries (hash, original, created_at, ttl_seconds) + VALUES (?1, ?2, ?3, ?4)", + rusqlite::params![hash, payload.as_bytes(), now, 300_i64], + ) + .expect("insert legacy row"); + } + let store = SqliteCcrStore::open(&path, 300).expect("open must migrate legacy schema"); + assert_eq!( + store.get(&hash).as_deref(), + Some(payload), + "legacy rows must survive the schema migration" + ); +} + // ─── Redis-feature-gated tests ───────────────────────────────────────── #[cfg(feature = "redis")] @@ -196,4 +347,31 @@ mod redis_tests { store.put(&hash, payload); assert_eq!(store.get(&hash).as_deref(), Some(payload)); } + + #[test] + fn redis_get_refreshes_idle_ttl() { + let Some(url) = redis_url() else { + eprintln!("skipping redis_get_refreshes_idle_ttl: HEADROOM_TEST_REDIS_URL not set"); + return; + }; + // 2-second idle window (Redis EXPIRE has 1s resolution). + let store = RedisCcrStore::open_with_prefix(&url, "ccr_test_sliding".to_string(), 2) + .expect("open redis store"); + let payload = "sliding redis"; + let hash = compute_key(payload.as_bytes()); + store.put(&hash, payload); + // Touch at t+1.5s (inside window) — restarts the idle clock. + std::thread::sleep(Duration::from_millis(1_500)); + assert_eq!(store.get(&hash).as_deref(), Some(payload)); + // t+3s: wall-clock expiry would have fired at t+2s. + std::thread::sleep(Duration::from_millis(1_500)); + assert_eq!( + store.get(&hash).as_deref(), + Some(payload), + "an entry accessed within its idle window must stay alive past the wall-clock TTL" + ); + // Go idle past the window. + std::thread::sleep(Duration::from_millis(3_100)); + assert_eq!(store.get(&hash), None); + } }