Distributed Caching

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

import Redis from 'ioredis';
 
const redis = new Redis({
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: 3,
  // Connection pooling via ioredis is automatic
});
 
// Basic caching
async function cacheExample() {
  // SET with expiration
  await redis.setex('user:123', 300, JSON.stringify({ name: 'Alice', email: 'alice@example.com' }));
  
  // GET with type safety
  const userData = await redis.get('user:123');
  const user = userData ? JSON.parse(userData) : null;
  
  // SET only if not exists (useful for locks, deduplication)
  const wasSet = await redis.setnx('lock:resource', 'holder-id');
  
  // SET with options: NX (not exists), EX (expiration)
  await redis.set('key', 'value', 'EX', 300, 'NX');
}
 
// Hash operations (structured data without JSON parsing)
async function hashExample() {
  // Store user as hash
  await redis.hset('user:123', {
    name: 'Alice',
    email: 'alice@example.com',
    loginCount: '42',
  });
  
  // Get single field
  const name = await redis.hget('user:123', 'name');
  
  // Get all fields
  const user = await redis.hgetall('user:123');
  
  // Increment field atomically
  await redis.hincrby('user:123', 'loginCount', 1);
}
 
// Sorted set for leaderboards/rankings
async function leaderboardExample() {
  // Add scores
  await redis.zadd('leaderboard:daily', 1500, 'player:123');
  await redis.zadd('leaderboard:daily', 2300, 'player:456');
  await redis.zadd('leaderboard:daily', 1800, 'player:789');
  
  // Get top 10 with scores
  const topPlayers = await redis.zrevrange('leaderboard:daily', 0, 9, 'WITHSCORES');
  // ['player:456', '2300', 'player:789', '1800', 'player:123', '1500']
  
  // Get rank of specific player
  const rank = await redis.zrevrank('leaderboard:daily', 'player:123');
}
 
// Atomic operations with Lua scripting
async function atomicOperation() {
  // Rate limiter: increment and check in one atomic operation
  const rateLimitScript = `
    local current = redis.call('INCR', KEYS[1])
    if current == 1 then
      redis.call('EXPIRE', KEYS[1], ARGV[1])
    end
    return current
  `;
  
  const requestCount = await redis.eval(
    rateLimitScript,
    1,                     // Number of keys
    'rate:user:123',       // KEYS[1]
    '60'                   // ARGV[1]: 60 second window
  );
  
  if (requestCount > 100) {
    throw new Error('Rate limit exceeded');
  }
}
 
// Pipeline for batch operations
async function pipelineExample() {
  const pipeline = redis.pipeline();
  
  // Queue multiple operations
  pipeline.get('user:123');
  pipeline.get('user:456');
  pipeline.hgetall('session:abc');
  pipeline.zrange('recent:items', 0, 9);
  
  // Execute all at once (single round trip)
  const results = await pipeline.exec();
  // [[null, 'user123data'], [null, 'user456data'], [null, {...}], [null, [...]]]
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

import { Cluster } from 'ioredis';
 
// Connect to Redis Cluster
const cluster = new Cluster([
  { host: 'node1.redis.example.com', port: 6379 },
  { host: 'node2.redis.example.com', port: 6379 },
  { host: 'node3.redis.example.com', port: 6379 },
], {
  // Cluster-specific options
  scaleReads: 'slave',  // Read from replicas for read scaling
  slotsRefreshTimeout: 2000,
  redisOptions: {
    password: process.env.REDIS_PASSWORD,
    tls: {}, // Enable TLS in production
  },
});
 
// Normal operations work transparently
await cluster.set('user:123', 'data');
const data = await cluster.get('user:123');
 
// Hash tags for co-locating related keys
// Keys with same {tag} go to same slot
await cluster.set('{user:123}:profile', JSON.stringify(profile));
await cluster.set('{user:123}:settings', JSON.stringify(settings));
await cluster.set('{user:123}:sessions', JSON.stringify(sessions));
 
// Now multi-key operations work (same slot)
const results = await cluster.mget(
  '{user:123}:profile',
  '{user:123}:settings',
  '{user:123}:sessions'
);
 
// Warning: This WILL fail in cluster mode (different slots)
// await cluster.mget('user:123', 'user:456');  // CrossSlotError
 
// Pipeline with cluster (automatically groups by slot)
const pipeline = cluster.pipeline();
pipeline.set('{order:1}:status', 'pending');
pipeline.set('{order:1}:items', JSON.stringify(items));
await pipeline.exec();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

import Memcached from 'memcached';
 
// Connect to Memcached cluster (client-side consistent hashing)
const memcached = new Memcached([
  'cache1.example.com:11211',
  'cache2.example.com:11211',
  'cache3.example.com:11211',
], {
  poolSize: 10,
  retries: 2,
  timeout: 5000,
  idle: 30000,
});
 
// Promisify for async/await
const memcachedAsync = {
  get: (key: string): Promise<any> =>
    new Promise((resolve, reject) => {
      memcached.get(key, (err, data) => {
        if (err) reject(err);
        else resolve(data);
      });
    }),
  
  set: (key: string, value: any, ttl: number): Promise<boolean> =>
    new Promise((resolve, reject) => {
      memcached.set(key, value, ttl, (err) => {
        if (err) reject(err);
        else resolve(true);
      });
    }),
  
  delete: (key: string): Promise<boolean> =>
    new Promise((resolve, reject) => {
      memcached.del(key, (err) => {
        if (err) reject(err);
        else resolve(true);
      });
    }),
  
  // Atomic increment
  incr: (key: string, amount: number): Promise<number> =>
    new Promise((resolve, reject) => {
      memcached.incr(key, amount, (err, result) => {
        if (err) reject(err);
        else resolve(result);
      });
    }),
};
 
// Usage
async function cacheUser(userId: string, user: User): Promise<void> {
  // Must serialize to string (no native data structures)
  await memcachedAsync.set(
    `user:${userId}`,
    JSON.stringify(user),
    300  // TTL in seconds
  );
}
 
async function getUser(userId: string): Promise<User | null> {
  const data = await memcachedAsync.get(`user:${userId}`);
  return data ? JSON.parse(data) : null;
}
 
// Multi-get for batch operations
function multiGet(keys: string[]): Promise<Record<string, any>> {
  return new Promise((resolve, reject) => {
    memcached.getMulti(keys, (err, data) => {
      if (err) reject(err);
      else resolve(data);
    });
  });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

// Pattern 1: Cache-Aside with Invalidation
async function updateUserCacheAside(
  userId: string, 
  updates: Partial<User>
): Promise<User> {
  // Write to database first
  const user = await db.users.update({
    where: { id: userId },
    data: updates,
  });
  
  // Invalidate cache (next read will repopulate)
  await redis.del(`user:${userId}`);
  
  return user;
}
 
// Pattern 2: Write-Through with Cache Update
async function updateUserWriteThrough(
  userId: string,
  updates: Partial<User>
): Promise<User> {
  // Write to database
  const user = await db.users.update({
    where: { id: userId },
    data: updates,
  });
  
  // Update cache with new value (not invalidate)
  await redis.setex(
    `user:${userId}`,
    300,
    JSON.stringify(user)
  );
  
  return user;
}
 
// Pattern 3: Handling the Delete-Then-Read Race
// Problem: Thread A deletes cache, Thread B reads stale DB, caches it,
//          Thread A writes to DB. Cache is now stale.
 
// Solution: Double-delete pattern
async function updateUserDoubleDelete(
  userId: string,
  updates: Partial<User>
): Promise<User> {
  const cacheKey = `user:${userId}`;
  
  // First delete
  await redis.del(cacheKey);
  
  // Write to database
  const user = await db.users.update({
    where: { id: userId },
    data: updates,
  });
  
  // Delay to allow in-flight reads to complete
  setTimeout(async () => {
    // Second delete (cleans up any stale repopulation)
    await redis.del(cacheKey);
  }, 500);
  
  return user;
}
 
// Pattern 4: Lock-based consistency
async function updateUserWithLock(
  userId: string,
  updates: Partial<User>
): Promise<User> {
  const cacheKey = `user:${userId}`;
  const lockKey = `lock:${cacheKey}`;
  
  // Acquire lock
  const lockAcquired = await redis.set(lockKey, '1', 'EX', 5, 'NX');
  if (!lockAcquired) {
    throw new Error('Could not acquire lock');
  }
  
  try {
    // Write to database
    const user = await db.users.update({
      where: { id: userId },
      data: updates,
    });
    
    // Update cache
    await redis.setex(cacheKey, 300, JSON.stringify(user));
    
    return user;
  } finally {
    // Release lock
    await redis.del(lockKey);
  }
}
 
// Pattern 5: Version-based invalidation
interface VersionedCache {
  version: number;
  data: any;
}
 
async function getUserVersioned(userId: string): Promise<User | null> {
  const currentVersion = await redis.get(`user:${userId}:version`);
  const cached = await redis.get(`user:${userId}`);
  
  if (cached) {
    const parsed: VersionedCache = JSON.parse(cached);
    if (parsed.version.toString() === currentVersion) {
      return parsed.data;
    }
    // Version mismatch - cache is stale
  }
  
  // Fetch fresh data
  const user = await db.users.findUnique({ where: { id: userId } });
  
  if (user) {
    const version = parseInt(currentVersion || '0') + 1;
    await redis.setex(`user:${userId}`, 300, JSON.stringify({
      version,
      data: user,
    }));
    await redis.set(`user:${userId}:version`, version.toString());
  }
  
  return user;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

import Redis from 'ioredis';
import CircuitBreaker from 'opossum';
 
// Configure Redis with resilience
const redis = new Redis({
  host: 'cache.example.com',
  port: 6379,
  
  // Connection resilience
  retryStrategy: (times) => {
    if (times > 3) return null;  // Stop retrying
    return Math.min(times * 200, 2000);  // Exponential backoff
  },
  
  // Command timeout
  commandTimeout: 500,  // Fast fail
  
  // Connection timeout
  connectTimeout: 3000,
  
  // Auto-reconnect
  maxRetriesPerRequest: 1,
  enableOfflineQueue: false,  // Don't queue when disconnected
});
 
// Circuit breaker for cache operations
const cacheBreaker = new CircuitBreaker(
  async (operation: () => Promise<any>) => operation(),
  {
    timeout: 1000,           // Consider call failed after 1s
    errorThresholdPercentage: 50,  // Open after 50% failures
    resetTimeout: 30000,     // Try again after 30s
    volumeThreshold: 10,     // Minimum calls before tripping
  }
);
 
cacheBreaker.on('open', () => {
  console.warn('Cache circuit breaker OPEN - falling back to database');
});
 
cacheBreaker.on('halfOpen', () => {
  console.info('Cache circuit breaker testing recovery...');
});
 
cacheBreaker.on('close', () => {
  console.info('Cache circuit breaker CLOSED - cache operational');
});
 
// Resilient cache wrapper
class ResilientCache {
  async get<T>(key: string): Promise<T | null> {
    try {
      return await cacheBreaker.fire(async () => {
        const data = await redis.get(key);
        return data ? JSON.parse(data) : null;
      });
    } catch (error) {
      // Circuit open or cache error - return null (cache miss)
      console.warn(`Cache get failed for ${key}:`, error.message);
      return null;
    }
  }
 
  async set(key: string, value: any, ttlSeconds: number): Promise<void> {
    try {
      await cacheBreaker.fire(async () => {
        await redis.setex(key, ttlSeconds, JSON.stringify(value));
      });
    } catch (error) {
      // Cache write failure is non-fatal
      console.warn(`Cache set failed for ${key}:`, error.message);
      // Don't throw - application continues without caching
    }
  }
 
  async delete(key: string): Promise<void> {
    try {
      await cacheBreaker.fire(async () => {
        await redis.del(key);
      });
    } catch (error) {
      // Invalidation failure is concerning but not fatal
      console.error(`Cache delete failed for ${key}:`, error.message);
      // Consider: queue for retry, or accept stale data with TTL
    }
  }
}
 
// Usage with graceful degradation
const cache = new ResilientCache();
 
async function getUser(userId: string): Promise<User> {
  // Try cache (gracefully handles failure)
  const cached = await cache.get<User>(`user:${userId}`);
  if (cached) {
    return cached;
  }
  
  // Fallback to database
  const user = await db.users.findUnique({ where: { id: userId } });
  
  // Attempt to cache (fire-and-forget, non-blocking)
  if (user) {
    cache.set(`user:${userId}`, user, 300).catch(() => {
      // Already logged in ResilientCache
    });
  }
  
  return user;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# Redis production configuration
 
# Memory limits
maxmemory 4gb
maxmemory-policy allkeys-lru  # Evict any key using LRU when memory full
# Alternatives:
# - volatile-lru: Only evict keys with TTL set
# - allkeys-random: Random eviction
# - volatile-ttl: Evict keys closest to expiration
# - noeviction: Return errors when memory full
 
# Persistence (choose based on durability needs)
# Option 1: No persistence (pure cache)
save ""
appendonly no
 
# Option 2: RDB snapshots
save 900 1       # Snapshot if 1 key changed in 15 min
save 300 10      # Snapshot if 10 keys changed in 5 min
save 60 10000    # Snapshot if 10000 keys changed in 1 min
 
# Option 3: AOF (append-only file) for durability
appendonly yes
appendfsync everysec  # Sync every second (balance durability/performance)
 
# Network security
bind 10.0.0.0/8      # Only listen on private network
requirepass <strong-password>
tls-port 6379
tls-cert-file /path/to/cert.pem
tls-key-file /path/to/key.pem
 
# Connection limits
maxclients 10000
timeout 300           # Close idle connections after 5 min
 
# Slow query logging
slowlog-log-slower-than 10000  # Log commands > 10ms
slowlog-max-len 128
 
# Disable dangerous commands in production
rename-command FLUSHALL ""
rename-command FLUSHDB ""
rename-command DEBUG ""
rename-command KEYS ""  # KEYS is O(n), use SCAN instead

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# Redis monitoring commands
 
# Real-time stats
redis-cli INFO stats
redis-cli INFO memory
redis-cli INFO replication
 
# Slow query log
redis-cli SLOWLOG GET 10
 
# Memory analysis
redis-cli MEMORY DOCTOR
 
# Big keys scan (run during low traffic)
redis-cli --bigkeys
 
# Sample key analysis
redis-cli DEBUG OBJECT key-name
 
# Live monitoring
redis-cli MONITOR  # WARNING: Production impact, use briefly
 
# Latency check
redis-cli --latency
 
# Key expiration analysis
redis-cli DEBUG SLEEP 0  # Check key TTL distribution