manutej/luxor-claude-marketplace/redis-state-management

Claude Code and agent skill by manutej/luxor-claude-marketplace

Trust tier 3
  • community
Source check pending

Checking that the SKILL.md file is still in the repository.

Comprehensive guide for Redis state management including caching strategies, session management, pub/sub patterns, distributed locks, and data structures

Install

npx vskill@latest install manutej/luxor-claude-marketplace/redis-state-management

SKILL.md

Redis State Management

A comprehensive skill for mastering Redis state management patterns in distributed systems. This skill covers caching strategies, session management, pub/sub messaging, distributed locks, data structures, and production-ready patterns using redis-py.

When to Use This Skill

Use this skill when:

  • Implementing high-performance caching layers for web applications
  • Managing user sessions in distributed environments
  • Building real-time messaging and event distribution systems
  • Coordinating distributed processes with locks and synchronization
  • Storing and querying structured data with Redis data structures
  • Optimizing application performance with Redis
  • Scaling applications horizontally with shared state
  • Implementing rate limiting, counters, and analytics
  • Building microservices with Redis as a communication layer
  • Managing temporary data with automatic expiration (TTL)
  • Implementing leaderboards, queues, and real-time features

Core Concepts

Redis Fundamentals

Redis (Remote Dictionary Server) is an in-memory data structure store used as:

  • Database: Persistent key-value storage
  • Cache: High-speed data layer
  • Message Broker: Pub/sub and stream messaging
  • Session Store: Distributed session management

Key Characteristics:

  • In-memory storage (microsecond latency)
  • Optional persistence (RDB snapshots, AOF logs)
  • Rich data structures beyond key-value
  • Atomic operations on complex data types
  • Built-in replication and clustering
  • Pub/sub messaging support
  • Lua scripting for complex operations
  • Pipelining for batch operations
Redis Data Structures

Redis provides multiple data types for different use cases:

  1. Strings: Simple key-value pairs, binary safe
  • Use for: Cache values, counters, flags, JSON objects
  • Max size: 512 MB
  • Commands: SET, GET, INCR, APPEND
  1. Hashes: Field-value maps (objects)
  • Use for: User profiles, configuration objects, small entities
  • Efficient for storing objects with multiple fields
  • Commands: HSET, HGET, HMGET, HINCRBY
  1. Lists: Ordered collections (linked lists)
  • Use for: Queues, activity feeds, recent items
  • Operations at head/tail are O(1)
  • Commands: LPUSH, RPUSH, LPOP, RPOP, LRANGE
  1. Sets: Unordered unique collections
  • Use for: Tags, unique visitors, relationships
  • Set operations: union, intersection, difference
  • Commands: SADD, SMEMBERS, SISMEMBER, SINTER
  1. Sorted Sets: Ordered sets with scores
  • Use for: Leaderboards, time-series, priority queues
  • Range queries by score or rank
  • Commands: ZADD, ZRANGE, ZRANGEBYSCORE, ZRANK
  1. Streams: Append-only logs with consumer groups
  • Use for: Event sourcing, activity logs, message queues
  • Built-in consumer group support
  • Commands: XADD, XREAD, XREADGROUP
Connection Management

Connection Pools: Redis connections are expensive to create. Always use connection pools:

python
import redis
# Connection pool (recommended)
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, max_connections=10)
r = redis.Redis(connection_pool=pool)
# Direct connection (avoid in production)
r = redis.Redis(host='localhost', port=6379, db=0)

Best Practices:

  • Use connection pools for all applications
  • Set appropriate max_connections based on workload
  • Enable decode_responses=True for string data
  • Configure socket_timeout and socket_keepalive
  • Handle connection errors with retries
Data Persistence

Redis offers two persistence mechanisms:

RDB (Redis Database): Point-in-time snapshots

  • Compact binary format
  • Fast restart times
  • Lower disk I/O
  • Potential data loss between snapshots

AOF (Append-Only File): Log of write operations

  • Better durability (fsync policies)
  • Larger files, slower restarts
  • Can be automatically rewritten/compacted
  • Minimal data loss potential

Hybrid Approach: RDB + AOF for best of both worlds

RESP 3 Protocol

Redis Serialization Protocol version 3 offers:

  • Client-side caching support
  • Better data type support
  • Push notifications
  • Performance improvements
python
import redis
from redis.cache import CacheConfig
# Enable RESP3 with client-side caching
r = redis.Redis(host='localhost', port=6379, protocol=3,
                cache_config=CacheConfig())

Caching Strategies

Cache-Aside (Lazy Loading)

Pattern: Application checks cache first, loads from database on miss

python
import redis
import json
from typing import Optional, Dict, Any
r = redis.Redis(decode_responses=True)
def get_user(user_id: int) -> Optional[Dict[str, Any]]:
    """Cache-aside pattern for user data."""
    cache_key = f"user:{user_id}"
    # Try cache first
    cached_data = r.get(cache_key)
    if cached_data:
        return json.loads(cached_data)
    # Cache miss - load from database
    user_data = database.get_user(user_id)  # Your DB query
    if user_data:
        # Store in cache with 1 hour TTL
        r.setex(cache_key, 3600, json.dumps(user_data))
    return user_data

Advantages:

  • Only requested data is cached (efficient memory usage)
  • Cache failures don't break the application
  • Simple to implement

Disadvantages:

  • Cache miss penalty (latency spike)
  • Thundering herd on popular items
  • Stale data until cache expiration
Write-Through Cache

Pattern: Write to cache and database simultaneously

python
def update_user(user_id: int, user_data: Dict[str, Any]) -> bool:
    """Write-through pattern for user updates."""
    cache_key = f"user:{user_id}"
    # Write to database first
    success = database.update_user(user_id, user_data)
    if success:
        # Update cache immediately
        r.setex(cache_key, 3600, json.dumps(user_data))
    return success

Advantages:

  • Cache always consistent with database
  • No read penalty for recently written data

Disadvantages:

  • Write latency increases
  • Unused data may be cached
  • Extra cache write overhead
Write-Behind (Write-Back) Cache

Pattern: Write to cache immediately, sync to database asynchronously

python
import redis
import json
from queue import Queue
from threading import Thread
r = redis.Redis(decode_responses=True)
write_queue = Queue()
def async_writer():
    """Background worker to sync cache to database."""
    while True:
        user_id, user_data = write_queue.get()
        try:
            database.update_user(user_id, user_data)
        except Exception as e:
            # Log error, potentially retry
            print(f"Failed to write user {user_id}: {e}")
        finally:
            write_queue.task_done()
# Start background writer
Thread(target=async_writer, daemon=True).start()
def update_user_fast(user_id: int, user_data: Dict[str, Any]):
    """Write-behind pattern for fast writes."""
    cache_key = f"user:{user_id}"
    # Write to cache immediately (fast)
    r.setex(cache_key, 3600, json.dumps(user_data))
    # Queue database write (async)
    write_queue.put((user_id, user_data))

Advantages:

  • Minimal write latency
  • Can batch database writes
  • Handles write spikes

Disadvantages:

  • Risk of data loss if cache fails
  • Complex error handling
  • Consistency challenges
Cache Invalidation Strategies

Time-based Expiration (TTL):

python
# Set key with expiration
r.setex("session:abc123", 1800, session_data)  # 30 minutes
# Or set TTL on existing key
r.expire("user:profile:123", 3600)  # 1 hour
# Check remaining TTL
ttl = r.ttl("user:profile:123")

Event-based Invalidation:

python
def update_product(product_id: int, product_data: dict):
    """Invalidate cache on update."""
    # Update database
    database.update_product(product_id, product_data)
    # Invalidate related caches
    r.delete(f"product:{product_id}")
    r.delete(f"product_list:category:{product_data['category']}")
    r.delete("products:featured")

Pattern-based Invalidation:

python
# Delete all keys matching pattern
def invalidate_user_cache(user_id: int):
    """Invalidate all cache entries for a user."""
    pattern = f"user:{user_id}:*"
    # Find and delete matching keys
    for key in r.scan_iter(match=pattern, count=100):
        r.delete(key)
Cache Stampede Prevention

Problem: Multiple requests simultaneously miss cache and query database

Solution 1: Probabilistic Early Expiration

python
import time
import random
def get_with_early_expiration(key: str, ttl: int = 3600, beta: float = 1.0):
    """Prevent stampede with probabilistic early recomputation."""
    value = r.get(key)
    if value is None:
        # Cache miss - compute and cache
        value = compute_value(key)
        r.setex(key, ttl, value)
        return value
    # Check if we should recompute early
    current_time = time.time()
    delta = current_time - float(r.get(f"{key}:timestamp") or 0)
    expiry = ttl * random.random() * beta
    if delta > expiry:
        # Recompute in background
        value = compute_value(key)
        r.setex(key, ttl, value)
        r.set(f"{key}:timestamp", current_time)
    return value

Solution 2: Locking

python
from contextlib import contextmanager
@contextmanager
def cache_lock(key: str, timeout: int = 10):
    """Acquire lock for cache computation."""
    lock_key = f"{key}:lock"
    identifier = str(time.time())
    # Try to acquire lock
    if r.set(lock_key, identifier, nx=True, ex=timeout):
        try:
            yield True
        finally:
            # Release lock
            if r.get(lock_key) == identifier:
                r.delete(lock_key)
    else:
        yield False
def get_with_lock(key: str):
    """Use lock to prevent stampede."""
    value = r.get(key)
    if value is None:
        with cache_lock(key) as acquired:
            if acquired:
                # We got the lock - compute value
                value = compute_value(key)
                r.setex(key, 3600, value)
            else:
                # Someone else is computing - wait and retry
                time.sleep(0.1)
                value = r.get(key) or compute_value(key)
    return value

Session Management

Distributed Session Storage

Basic Session Management:

python
import redis
import json
import uuid
from datetime import datetime, timedelta
r = redis.Redis(decode_responses=True)
class SessionManager:
    def __init__(self, ttl: int = 1800):
        """Session manager with Redis backend.
        Args:
            ttl: Session timeout in seconds (default 30 minutes)
        """
        self.ttl = ttl
    def create_session(self, user_id: int, data: dict = None) -> str:
        """Create new session and return session ID."""
        session_id = str(uuid.uuid4())
        session_key = f"session:{session_id}"
        session_data = {
            "user_id": user_id,
            "created_at": datetime.utcnow().isoformat(),
            "data": data or {}
        }
        r.setex(session_key, self.ttl, json.dumps(session_data))
        return session_id
    def get_session(self, session_id: str) -> dict:
        """Retrieve session data and refresh TTL."""
        session_key = f"session:{session_id}"
        session_data = r.get(session_key)
        if session_data:
            # Refresh TTL on access (sliding expiration)
            r.expire(session_key, self.ttl)
            return json.loads(session_data)
        return None
    def update_session(self, session_id: str, data: dict) -> bool:
        """Update session data."""
        session_key = f"session:{session_id}"
        session_data = self.get_session(session_id)
        if session_data:
            session_data["data"].update(data)
            r.setex(session_key, self.ttl, json.dumps(session_data))
            return True
        return False

This is the start of the file. Read the full SKILL.md on GitHub.

Security

  • Security scanPassed, 100 of 100Pattern checks on the skill's files.
  • Model reviewPassed

See the full security report

Quality evaluation

  • Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

See evaluation results

Source

Popularity

Repository stars
54
Repository forks
15
7-day trend score
20

Stars and forks are for the whole GitHub repository, not this skill alone. Refreshed 7 May 2026, 6 months ago.

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

  • Claude Code
  • Cursor
  • GitHub Copilot
  • Windsurf
  • Codex
  • Gemini CLI
  • Cline
  • Amp
  • Roo Code
  • Goose
  • Continue
  • Kilo Code
  • Junie
  • Trae
  • OpenHands
  • Qwen Code
  • AI Agent
  • Tabnine Chat
  • Sourcegraph Cody
  • Amazon Q
  • Firebase Genkit
  • Sweep
  • Devin
  • Phind
  • Mentat
  • Claude Desktop
  • Aider
  • GPT Pilot
  • Plandex
  • bolt.new
  • v0
  • Replit Agent
  • Anse
  • TypingMind
  • BetterChatGPT
  • LibreChat
  • Warp AI
  • Pieces
  • Fig AI
Embed badge
vskill VERIFIED badge
[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/manutej/luxor-claude-marketplace/redis-state-management/badge)](https://verified-skill.com/skills/manutej/luxor-claude-marketplace/redis-state-management)

Something wrong with this skill or its page? Tell us.