ENG-3375: Add TTL to RedisRepository keys

changelog/7884-redis-repository-ttl.yaml

@@ -0,0 +1,3 @@
+type: Fixed
+description: Apply TTL to all RedisRepository keys to prevent unbounded Redis memory growth across PBAC entities.
+pr: 7884

src/fides/service/pbac/redis_repository.py

@@ -10,6 +10,7 @@
 
 from fides.api.common_exceptions import RedisConnectionError
 from fides.api.util.cache import FidesopsRedis
+from fides.config import CONFIG
 
 T = TypeVar("T")
 
@@ -29,8 +30,9 @@ class RedisRepository(ABC, Generic[T]):
 
     PREFIX: str  # e.g. "data_purpose"
 
-    def __init__(self, cache: FidesopsRedis) -> None:
+    def __init__(self, cache: FidesopsRedis, ttl_seconds: Optional[int] = None) -> None:
         self._cache = cache
+        self._ttl_seconds = ttl_seconds if ttl_seconds is not None else CONFIG.redis.default_ttl_seconds
 
     # ── Key construction (private) ────────────────────────────────────
 
@@ -77,11 +79,13 @@ def save(self, entity: T) -> T:
                 for field, value in self._get_index_entries(old_entity):
                     pipe.srem(self._index_key(field, value), pk)
 
-            # Write entity + add new indexes
-            pipe.set(key, self._serialize(entity))
+            # Write entity + add new indexes (all keys get a TTL)
+            pipe.set(key, self._serialize(entity), ex=self._ttl_seconds)
             pipe.sadd(self._all_key(), pk)
+            pipe.expire(self._all_key(), self._ttl_seconds)
             for field, value in self._get_index_entries(entity):
                 pipe.sadd(self._index_key(field, value), pk)
+                pipe.expire(self._index_key(field, value), self._ttl_seconds)
             pipe.execute()
         except RedisConnectionError:
             logger.error("Redis connection error during save of {}:{}", self.PREFIX, pk)

tests/service/pbac/test_redis_repository.py

@@ -0,0 +1,149 @@
+"""Unit tests for the RedisRepository base class."""
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi_pagination import Params
+
+from fides.service.pbac.redis_repository import RedisRepository
+
+TTL = 604800  # 7 days
+
+
+@dataclass
+class SampleEntity:
+    id: str
+    name: str
+    category: str
+    created_at: datetime
+
+
+class SampleRepository(RedisRepository[SampleEntity]):
+    PREFIX = "sample"
+
+    def _serialize(self, entity: SampleEntity) -> str:
+        return json.dumps(
+            {
+                "id": entity.id,
+                "name": entity.name,
+                "category": entity.category,
+                "created_at": entity.created_at.isoformat(),
+            }
+        )
+
+    def _deserialize(self, data: str) -> SampleEntity:
+        d = json.loads(data)
+        d["created_at"] = datetime.fromisoformat(d["created_at"])
+        return SampleEntity(**d)
+
+    def _get_pk(self, entity: SampleEntity) -> str:
+        return entity.id
+
+    def _get_index_entries(self, entity: SampleEntity) -> list[tuple[str, str]]:
+        return [("category", entity.category)]
+
+
+@pytest.fixture()
+def mock_cache():
+    cache = MagicMock()
+    cache.pipeline.return_value = MagicMock()
+    cache.pipeline.return_value.execute.return_value = None
+    return cache
+
+
+@pytest.fixture()
+def repo(mock_cache):
+    return SampleRepository(mock_cache, ttl_seconds=TTL)
+
+
+@pytest.fixture()
+def sample_entity():
+    return SampleEntity(
+        id="abc123",
+        name="Test Entity",
+        category="widgets",
+        created_at=datetime(2024, 1, 1, 12, 0, 0),
+    )
+
+
+class TestSaveTTL:
+    """Verify that save() applies TTL to all Redis keys."""
+
+    def test_entity_key_has_ttl(self, repo, mock_cache, sample_entity):
+        mock_cache.get.return_value = None
+        repo.save(sample_entity)
+
+        pipe = mock_cache.pipeline.return_value
+        pipe.set.assert_called_once_with(
+            "sample:abc123", repo._serialize(sample_entity), ex=TTL
+        )
+
+    def test_all_key_has_ttl(self, repo, mock_cache, sample_entity):
+        mock_cache.get.return_value = None
+        repo.save(sample_entity)
+
+        pipe = mock_cache.pipeline.return_value
+        pipe.sadd.assert_any_call("sample:_all", "abc123")
+        pipe.expire.assert_any_call("sample:_all", TTL)
+
+    def test_index_keys_have_ttl(self, repo, mock_cache, sample_entity):
+        mock_cache.get.return_value = None
+        repo.save(sample_entity)
+
+        pipe = mock_cache.pipeline.return_value
+        pipe.sadd.assert_any_call("sample:idx:category:widgets", "abc123")
+        pipe.expire.assert_any_call("sample:idx:category:widgets", TTL)
+
+    def test_pipeline_is_executed(self, repo, mock_cache, sample_entity):
+        mock_cache.get.return_value = None
+        repo.save(sample_entity)
+
+        pipe = mock_cache.pipeline.return_value
+        pipe.execute.assert_called_once()
+
+    def test_update_cleans_stale_indexes_with_ttl(self, repo, mock_cache, sample_entity):
+        old_entity = SampleEntity(
+            id="abc123",
+            name="Old",
+            category="old_category",
+            created_at=datetime(2024, 1, 1),
+        )
+        mock_cache.get.return_value = repo._serialize(old_entity)
+        repo.save(sample_entity)
+
+        pipe = mock_cache.pipeline.return_value
+        # Old index entry removed
+        pipe.srem.assert_any_call("sample:idx:category:old_category", "abc123")
+        # New entity key still gets TTL
+        pipe.set.assert_called_once_with(
+            "sample:abc123", repo._serialize(sample_entity), ex=TTL
+        )
+
+    @patch("fides.service.pbac.redis_repository.CONFIG")
+    def test_defaults_to_config_ttl(self, mock_config, mock_cache):
+        mock_config.redis.default_ttl_seconds = 86400
+        repo = SampleRepository(mock_cache)
+        assert repo._ttl_seconds == 86400
+
+
+class TestDelete:
+    def test_delete_existing(self, repo, mock_cache, sample_entity):
+        mock_cache.get.return_value = repo._serialize(sample_entity)
+        result = repo.delete("abc123")
+
+        assert result is True
+        pipe = mock_cache.pipeline.return_value
+        pipe.delete.assert_called_once_with("sample:abc123")
+        pipe.srem.assert_any_call("sample:_all", "abc123")
+        pipe.srem.assert_any_call("sample:idx:category:widgets", "abc123")
+        pipe.execute.assert_called()
+
+    def test_delete_missing(self, repo, mock_cache):
+        mock_cache.get.return_value = None
+        result = repo.delete("nonexistent")
+        assert result is False
+
+