From 5874ae5d9b7a01cde4097ce73f6c0323f91a9f6e Mon Sep 17 00:00:00 2001 From: ld3zian <42660576+ld3z@users.noreply.github.com> Date: Tue, 11 Mar 2025 14:30:10 -0700 Subject: [PATCH] fix flaw --- src/lib/db.ts | 240 ++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 231 insertions(+), 9 deletions(-) diff --git a/src/lib/db.ts b/src/lib/db.ts index 9e2a023..d665da0 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -9,6 +9,8 @@ const BASE_DELAY_MS = 50; const IS_VERCEL = process.env.VERCEL || false; const HEALTH_CHECK_INTERVAL = 60 * 1000; // Check primary health every minute const REPLICA_PROMOTION_THRESHOLD = 5; // After 5 failed primary health checks, promote a replica to primary +const SYNC_BUFFER_SIZE = 100; // Maximum number of operations to buffer before syncing +const SYNC_INTERVAL = 10 * 1000; // Sync replicas every 10 seconds // Redis URLs from environment variables const PRIMARY_REDIS_URL = @@ -31,6 +33,21 @@ let primaryFailedChecks = 0; let replicaClients: Map = new Map(); let activeReplicaUrl: string | null = null; +// Track which replica was used when primary was down +let lastUsedReplica: string | null = null; + +// Operation buffer for sync +interface RedisOperation { + key: string; + value: string; + ttl?: number; + type: 'set' | 'del'; + timestamp: number; +} + +// Buffer to store operations that need to be synced +const operationBuffer: RedisOperation[] = []; + // Redis instance naming for logging const redisInstanceNames = new Map(); @@ -87,7 +104,15 @@ async function createRedisClient(url: string): Promise { // Setup event listeners client - .on("ready", () => console.log(`${redisName} ready`)) + .on("ready", () => { + console.log(`${redisName} ready`); + + // If primary recovers and we had used a replica, initiate sync from replica to primary + if (url === PRIMARY_REDIS_URL && lastUsedReplica) { + syncFromReplicaToPrimary(lastUsedReplica).catch(err => + console.error(`Failed to sync from replica to primary: ${err}`)); + } + }) .on("error", (err) => { console.error(`${redisName} error:`, err); if (url === PRIMARY_REDIS_URL) { @@ -153,6 +178,7 @@ export async function getRedisClient(): Promise { const activeReplica = replicaClients.get(activeReplicaUrl)!; if (activeReplica.status === "ready" && await checkConnection(activeReplica, getRedisName(activeReplicaUrl))) { console.log(`Using active replica: ${getRedisName(activeReplicaUrl)}`); + lastUsedReplica = activeReplicaUrl; // Track that we used this replica return activeReplica; } } @@ -169,6 +195,7 @@ export async function getRedisClient(): Promise { if (await checkConnection(client, getRedisName(url))) { console.log(`Switching to replica: ${getRedisName(url)}`); activeReplicaUrl = url; + lastUsedReplica = url; // Track that we used this replica // If primary has been down for a while, promote this replica to be the new primary if (primaryFailedChecks >= REPLICA_PROMOTION_THRESHOLD) { @@ -191,12 +218,164 @@ export async function getRedisClient(): Promise { return primaryClient; } +/** + * Add an operation to the buffer for replication + */ +function bufferOperation(operation: RedisOperation): void { + operationBuffer.push(operation); + + // If buffer exceeds size limit, trigger a sync + if (operationBuffer.length >= SYNC_BUFFER_SIZE) { + syncReplicas().catch(err => console.error("Failed to sync replicas:", err)); + } +} + +/** + * Synchronize data from a replica to the primary + * Used when primary recovers after being down + */ +async function syncFromReplicaToPrimary(replicaUrl: string): Promise { + if (!primaryClient || primaryClient.status !== "ready" || !replicaClients.has(replicaUrl)) { + console.log("Cannot sync: primary or replica not available"); + return; + } + + const replicaClient = replicaClients.get(replicaUrl)!; + console.log(`Starting sync from ${getRedisName(replicaUrl)} to Primary...`); + + try { + // Get all feed keys using scan to prevent blocking + const feedKeys: string[] = []; + let cursor = '0'; + + do { + const [nextCursor, keys] = await replicaClient.scan( + cursor, + 'MATCH', + '[a-f0-9]*', // Match feed IDs (hex format) + 'COUNT', + '100' + ); + + cursor = nextCursor; + feedKeys.push(...keys); + } while (cursor !== '0'); + + // Get chapter cache keys + const chapterKeys: string[] = []; + cursor = '0'; + + do { + const [nextCursor, keys] = await replicaClient.scan( + cursor, + 'MATCH', + 'chapters:*', + 'COUNT', + '100' + ); + + cursor = nextCursor; + chapterKeys.push(...keys); + } while (cursor !== '0'); + + console.log(`Found ${feedKeys.length} feed keys and ${chapterKeys.length} chapter keys to sync`); + + // Sync feed mappings + if (feedKeys.length > 0) { + const feedPipeline = primaryClient.pipeline(); + + for (const key of feedKeys) { + const value = await replicaClient.get(key); + const ttl = await replicaClient.ttl(key); + + if (value && ttl > 0) { + feedPipeline.set(key, value, 'EX', ttl); + } + } + + await feedPipeline.exec(); + console.log(`Synced ${feedKeys.length} feed mappings from replica to primary`); + } + + // Sync chapter cache + if (chapterKeys.length > 0) { + const chapterPipeline = primaryClient.pipeline(); + + for (const key of chapterKeys) { + const value = await replicaClient.get(key); + const ttl = await replicaClient.ttl(key); + + if (value && ttl > 0) { + chapterPipeline.set(key, value, 'EX', ttl); + } + } + + await chapterPipeline.exec(); + console.log(`Synced ${chapterKeys.length} chapter caches from replica to primary`); + } + + lastUsedReplica = null; // Reset after successful sync + console.log("Replica to primary sync completed successfully"); + } catch (error) { + console.error("Error during replica to primary sync:", error); + } +} + +/** + * Synchronize operations from buffer to all replicas + */ +async function syncReplicas(): Promise { + if (operationBuffer.length === 0) return; + + // Get operations to sync and clear buffer + const operations = [...operationBuffer]; + operationBuffer.length = 0; + + // Skip if no operations or no replicas + if (operations.length === 0 || replicaClients.size === 0) return; + + console.log(`Syncing ${operations.length} operations to ${replicaClients.size} replicas...`); + + // Process each replica + const syncPromises = Array.from(replicaClients.entries()).map(async ([url, client]) => { + try { + if (client.status !== "ready" || !(await checkConnection(client, getRedisName(url)))) { + return; + } + + const pipeline = client.pipeline(); + + // Add all operations to pipeline + for (const op of operations) { + if (op.type === 'set') { + if (op.ttl) { + pipeline.set(op.key, op.value, 'EX', op.ttl); + } else { + pipeline.set(op.key, op.value); + } + } else if (op.type === 'del') { + pipeline.del(op.key); + } + } + + // Execute pipeline + await pipeline.exec(); + console.log(`Successfully synced ${operations.length} operations to ${getRedisName(url)}`); + } catch (error) { + console.error(`Error syncing to ${getRedisName(url)}:`, error); + } + }); + + await Promise.allSettled(syncPromises); +} + /** * Schedule periodic health checks to detect when primary recovers */ function scheduleHealthChecks() { if (typeof setInterval === 'undefined') return; // Skip in environments without setInterval + // Check primary health setInterval(async () => { if (!primaryHealthy && primaryClient) { try { @@ -205,6 +384,12 @@ function scheduleHealthChecks() { console.log("Primary Redis recovered"); primaryHealthy = true; primaryFailedChecks = 0; + + // If we used a replica when primary was down, sync data back + if (lastUsedReplica) { + syncFromReplicaToPrimary(lastUsedReplica).catch(err => + console.error(`Failed to sync from replica to primary: ${err}`)); + } } else { primaryFailedChecks++; console.log(`Primary still unhealthy, failed checks: ${primaryFailedChecks}`); @@ -215,15 +400,20 @@ function scheduleHealthChecks() { } } }, HEALTH_CHECK_INTERVAL); + + // Schedule regular replica sync + setInterval(async () => { + await syncReplicas().catch(err => console.error("Failed to sync replicas:", err)); + }, SYNC_INTERVAL); } -// Start health checks if not in a build context +// Start health checks and sync if not in a build context if (!import.meta.env.SSR) { scheduleHealthChecks(); } /** - * Store a feed mapping in Redis with automatic failover + * Store a feed mapping in Redis with automatic failover and replication */ export async function storeFeedMapping( feedId: string, @@ -236,10 +426,28 @@ export async function storeFeedMapping( created_at: new Date().toISOString() }; + const stringValue = JSON.stringify(mapping); + try { + // First write to primary/active client const redis = await getRedisClient(); // This already handles failover - await redis.set(feedId, JSON.stringify(mapping), "EX", DEFAULT_CACHE_TTL); + await redis.set(feedId, stringValue, "EX", DEFAULT_CACHE_TTL); console.log(`Successfully stored mapping for feed ${feedId}`); + + // Buffer operation for replication to other instances + bufferOperation({ + key: feedId, + value: stringValue, + ttl: DEFAULT_CACHE_TTL, + type: 'set', + timestamp: Date.now() + }); + + // If we're writing to a replica because primary is down, force immediate sync + // to other replicas to ensure consistent data + if (!primaryHealthy && lastUsedReplica) { + await syncReplicas().catch(err => console.error("Failed to sync replicas after write:", err)); + } } catch (error) { console.error(`Error storing feed mapping for ${feedId}:`, error); throw error; @@ -261,11 +469,25 @@ export async function getFeedMapping( // Reset expiration on access await redis.expire(feedId, DEFAULT_CACHE_TTL); - const data = JSON.parse(mapping); - return { - slugs: data.slugs, - lang: data.lang - }; + // Buffer this TTL-reset operation for replication + bufferOperation({ + key: feedId, + value: mapping, + ttl: DEFAULT_CACHE_TTL, + type: 'set', + timestamp: Date.now() + }); + + try { + const data = JSON.parse(mapping); + return { + slugs: data.slugs, + lang: data.lang + }; + } catch (e) { + console.error(`Error parsing feed mapping for ${feedId}:`, e); + return null; + } } catch (error) { console.error(`Error retrieving feed mapping for ${feedId}:`, error); return null;