Caching Strategy#
Redis caching patterns, TTL strategies, and cache invalidation across the ops-db-api.
Cache Types#
The API uses Redis for multiple caching purposes:
Transaction Buffer - Pending operations queue
Write-Through Cache - Generated IDs immediately available
Buffered Data Cache - Records pending replication
Read Buffer - Mutable updates to buffered records
Query Result Cache - Expensive query results
Endpoint Cache - Full endpoint responses
Endpoint Caching#
The @cached_endpoint decorator:
reset_transaction_builder()
# If the result is already a response object, update it with transaction info
if hasattr(result, "transaction_id"):
result.transaction_id = transaction_id
return result
else:
# Return the original result (let FastAPI handle the response)
return result
else:
# Main site: execute the function (which builds transaction steps),
# then execute those steps directly against the database. A
# write that can't actually be executed must fail loud here —
# this is the only write path on the main site, so silently
# returning the endpoint's success result would report a
# write that never happened.
result = await func(*args, **kwargs)
# Find the transaction builder populated by the function
transaction_builder = None
for key, value in kwargs.items():
if isinstance(value, SQLAlchemyTransactionBuilder):
transaction_builder = value
break
if transaction_builder is None and "_transaction_builder" in kwargs:
transaction_builder = kwargs["_transaction_builder"]
# Also check for the global reference, mirroring the
# secondary-site lookup above.
if transaction_builder is None:
from ..dependencies import get_transaction_builder
transaction_builder = get_transaction_builder()
if transaction_builder is None:
logger.error(
"No transaction builder found for direct execution on main site"
)
raise HTTPException(
status_code=500,
detail="Transaction builder not available for direct execution",
)
if transaction_builder.steps:
transaction = transaction_builder.build()
transaction_manager = get_transaction_manager()
if transaction_manager.executor:
try:
await transaction_manager.executor.execute_transaction(
transaction
)
except Exception:
# The endpoint already wrote a read-buffer overlay
# for this transaction, making the record visible to
# smart-query reads. On the main site LSN-based
# cleanup never runs, so a failed write would leave
# that overlay readable until its TTL. Drop it, then
# fail loud with the original error.
await _invalidate_failed_write_read_buffer(
transaction.transaction_id
)
raise
# The direct write committed. On the main site the write
# is not buffered, so a response still claiming
# status="buffered" (with the null transaction_id the
# secondary branch would otherwise fill in) misreports
# what happened. Correct it to reflect the direct write.
if getattr(result, "status", None) == "buffered":
result.status = "completed"
else:
Usage:
from ccat_ops_db_api.transaction_buffering import cached_endpoint
@router.get("/expensive-calculation/{id}")
@cached_endpoint(ttl=600) # Cache for 10 minutes
async def expensive_calculation(id: int):
# Complex computation
result = await perform_expensive_calculation(id)
return result
Cache key format:
site:{site_name}:cache:{function_name}:{args_hash}
TTL Strategies#
Different caches have different TTL strategies:
Cache Type |
TTL |
Strategy |
|---|---|---|
Transaction buffer |
None |
Durable until processed |
Write-through cache |
Dynamic |
Extended until LSN confirms |
Buffered data cache |
Dynamic |
Extended until LSN confirms |
Read buffer |
Dynamic |
Extended until LSN confirms |
Query results |
Fixed |
5-60 minutes typical |
Endpoint responses |
Fixed |
1-10 minutes typical |
Cache Invalidation#
LSN-Based Invalidation:
When LSN tracker confirms replication:
async def cleanup_caches(transaction_id):
transaction = await get_transaction(transaction_id)
for step in transaction.steps:
# Remove write-through cache
await redis.delete(f"site:{site}:cache:ids:{step.model}:{step.id}")
# Remove buffered data cache
await redis.delete(f"site:{site}:buffered:{step.model}:{step.id}")
# Remove read buffer
await redis.delete(f"site:{site}:read_buffer:{step.model}:{step.id}")
Time-Based Expiration:
Most caches use TTL for automatic cleanup:
await redis.setex(cache_key, ttl=300, value=data) # 5 minutes
Manual Invalidation:
For critical updates:
# Invalidate specific cache
await redis.delete(cache_key)
# Invalidate pattern
keys = await redis.keys(f"site:{site}:cache:visibility:*")
if keys:
await redis.delete(*keys)
Cache Monitoring#
Cache Hit Rate:
cache_hits = await redis.get("metrics:cache:hits")
cache_misses = await redis.get("metrics:cache:misses")
hit_rate = cache_hits / (cache_hits + cache_misses)
Cache Size:
redis-cli
> INFO memory
> DBSIZE
Monitor Cache Operations:
redis-cli MONITOR
Best Practices#
DO:
Use appropriate TTLs (shorter for frequently changing data)
Namespace keys by site
Monitor cache hit rates
Invalidate on updates
Use write-through for generated IDs
DON’T:
Cache user-specific data without user ID in key
Use very long TTLs for volatile data
Forget to handle cache misses
Cache large objects (> 1MB) without compression
Example: Visibility Caching#
Visibility calculations are expensive, so we cache aggressively:
@router.get("/visibility/{source_id}")
@cached_endpoint(ttl=3600) # 1 hour
async def get_visibility(
source_id: int,
date_start: datetime,
date_end: datetime
):
# Expensive calculation
visibility = await calculate_visibility(
source_id,
date_start,
date_end
)
return visibility
Key includes: source_id, date_start, date_end
TTL: 1 hour (visibility doesn’t change rapidly)
Invalidation: Admin can trigger precalculation
Summary#
Caching in ops-db-api:
Multiple types: Transaction, write-through, buffered, query, endpoint
Dynamic TTLs: LSN-based for transaction caches
Namespaced keys: Site-aware cache isolation
Smart invalidation: LSN confirms when safe to cleanup
Monitoring: Track hit rates and cache size
Next Steps#
Transaction Manager - Write-through cache implementation
LSN Tracking - LSN-based invalidation
Redis Inspection - Redis debugging