fix(cache): implement distributed lock for cache stampede prevention

Author: Animesh-86

app/page.tsx

@@ -1,5 +1,6 @@
 'use client';
 import { trackUser } from '@/utils/tracking';
+
 import Link from 'next/link';
 import { useRef, useState, useEffect } from 'react';
 import { AnimatePresence, motion } from 'framer-motion';

lib/cache.ts

@@ -211,6 +211,7 @@ export class DistributedCache<T> {
   private useRedis: boolean;
   private redisUrl: string = '';
   private redisToken: string = '';
+  private localLocks = new Map<string, Promise<T>>();
 
   constructor(maxSize?: number, cleanupIntervalMs?: number) {
     this.localCache = new TTLCache<T>(maxSize, cleanupIntervalMs);
@@ -369,4 +370,121 @@ export class DistributedCache<T> {
   destroy(): void {
     this.localCache.destroy();
   }
+
+  /**
+   * Gets a value from the cache, or executes the load function if missing or stale.
+   * Employs both an in-memory Promise lock (L1) and a Redis Mutex (L2) to prevent Cache Stampedes.
+   *
+   * @param key - Cache key.
+   * @param loadFn - Async function to fetch the data. Receives the stale cached value if one exists.
+   * @param ttlMs - Time to live in milliseconds.
+   * @param shouldFetch - Optional predicate to force fetching even if a cache value exists (e.g. for stale delta sync).
+   */
+  async getOrSet(
+    key: string,
+    loadFn: (cached: T | null) => Promise<T>,
+    ttlMs: number,
+    shouldFetch?: (cached: T) => boolean
+  ): Promise<T> {
+    // 1. L1 & L2 Cache Check
+    const cached = await this.get(key);
+
+    // If we have a cache hit and we don't need to force a refresh, return it early.
+    if (cached !== null && (!shouldFetch || !shouldFetch(cached))) {
+      return cached;
+    }
+
+    // 2. L1 Promise Deduping (Local Lock)
+    const pendingLocal = this.localLocks.get(key);
+    if (pendingLocal) return pendingLocal;
+
+    const executeAndLock = async () => {
+      if (!this.useRedis) {
+        // Fallback: Local execution only
+        const data = await loadFn(cached);
+        await this.set(key, data, ttlMs);
+        return data;
+      }
+
+      const lockKey = `lock:${key}`;
+      const maxPollTime = 8000; // Give up polling after 8 seconds to stay within serverless limits
+      const pollInterval = 400;
+      const start = Date.now();
+
+      while (Date.now() - start < maxPollTime) {
+        try {
+          // Attempt to acquire Redis Mutex
+          const lockRes = await fetch(`${this.redisUrl}/`, {
+            method: 'POST',
+            headers: {
+              Authorization: `Bearer ${this.redisToken}`,
+              'Content-Type': 'application/json',
+            },
+            body: JSON.stringify(['SET', lockKey, '1', 'NX', 'PX', 10000]),
+          });
+
+          if (lockRes.ok) {
+            const lockData = await lockRes.json();
+            if (lockData.result === 'OK') {
+              // Lock acquired! Execute loadFn.
+              try {
+                const freshData = await loadFn(cached);
+                await this.set(key, freshData, ttlMs);
+
+                // Release lock early
+                await fetch(`${this.redisUrl}/`, {
+                  method: 'POST',
+                  headers: {
+                    Authorization: `Bearer ${this.redisToken}`,
+                    'Content-Type': 'application/json',
+                  },
+                  body: JSON.stringify(['DEL', lockKey]),
+                }).catch(() => {});
+
+                return freshData;
+              } catch (err) {
+                // Release lock on error so others can retry
+                await fetch(`${this.redisUrl}/`, {
+                  method: 'POST',
+                  headers: {
+                    Authorization: `Bearer ${this.redisToken}`,
+                    'Content-Type': 'application/json',
+                  },
+                  body: JSON.stringify(['DEL', lockKey]),
+                }).catch(() => {});
+                throw err;
+              }
+            }
+          }
+        } catch (err) {
+          // Redis network error during locking. Fallback to direct execution.
+          console.error(`[DistributedCache] Lock error for "${key}":`, err);
+          const fallbackData = await loadFn(cached);
+          await this.set(key, fallbackData, ttlMs);
+          return fallbackData;
+        }
+
+        // Lock not acquired. Wait and poll L2 cache.
+        await new Promise((resolve) => setTimeout(resolve, pollInterval));
+        const doubleCheck = await this.get(key);
+        // If doubleCheck satisfies the condition, return it
+        if (doubleCheck !== null && (!shouldFetch || !shouldFetch(doubleCheck))) {
+          return doubleCheck;
+        }
+      }
+
+      // Timed out waiting for lock. Fallback to direct execution to avoid hanging the client.
+      const finalFallback = await loadFn(cached);
+      await this.set(key, finalFallback, ttlMs);
+      return finalFallback;
+    };
+
+    // Execute with local Promise lock
+    const promise = executeAndLock().finally(() => {
+      this.localLocks.delete(key);
+    });
+    this.localLocks.set(key, promise);
+
+    return promise;
+  }
 }

lib/github.ts

@@ -251,10 +251,6 @@ const contributionsCache = new DistributedCache<ExtendedContributionData>(1000);
 const profileCache = new DistributedCache<GitHubUserProfile>(1000);
 const reposCache = new DistributedCache<GitHubRepo[]>(500);
 const contributedReposCache = new DistributedCache<Record<string, unknown>[]>(500);
-const pendingContributions = new Map<string, Promise<ExtendedContributionData>>();
-const pendingProfiles = new Map<string, Promise<GitHubUserProfile>>();
-const pendingRepos = new Map<string, Promise<GitHubRepo[]>>();
-const pendingContributedRepos = new Map<string, Promise<Record<string, unknown>[]>>();
 
 interface GitHubUserProfile {
   login: string;
@@ -300,25 +296,6 @@ export function clearGitHubApiCacheForTests(): void {
   profileCache.clear();
   reposCache.clear();
   contributedReposCache.clear();
-  pendingContributions.clear();
-  pendingProfiles.clear();
-  pendingRepos.clear();
-  pendingContributedRepos.clear();
-}
-
-function dedupeRequest<T>(
-  pendingRequests: Map<string, Promise<T>>,
-  key: string,
-  load: () => Promise<T>
-): Promise<T> {
-  const pending = pendingRequests.get(key);
-  if (pending) return pending;
-
-  const request = load().finally(() => {
-    pendingRequests.delete(key);
-  });
-  pendingRequests.set(key, request);
-  return request;
 }
 
 function getGitHubToken(): string {
@@ -397,31 +374,21 @@ export async function fetchGitHubContributions(
   options: FetchOptions = {}
 ): Promise<ExtendedContributionData> {
   const key = cacheKey('contributions', username, options.from, options.to);
+  const LONG_CACHE_TTL = 7 * 24 * 60 * 60 * 1000;
 
-  if (options.bypassCache) {
-    return fetchContributionsUncached(username, key, options, null);
-  }
-
-  // Wrap the entire cache-check + fetch pipeline in dedupeRequest so that
-  // concurrent calls for the same key share a single in-flight promise.
-  // This closes the TOCTOU window where two requests both miss the cache
-  // (during the async DistributedCache.get) and each fires its own API call.
-  const load = async (): Promise<ExtendedContributionData> => {
-    const cached = await contributionsCache.get(key);
-
+  const shouldFetch = (cached: ExtendedContributionData) => {
     const now = Date.now();
-    const isStale = cached?.calendar.lastSyncedAt
+    return cached?.calendar.lastSyncedAt
       ? now - new Date(cached.calendar.lastSyncedAt).getTime() > GITHUB_CACHE_TTL_MS
       : true;
+  };
 
-    if (cached && !isStale) {
-      return cached;
-    }
-
+  const load = async (cached: ExtendedContributionData | null) => {
     return fetchContributionsUncached(username, key, options, cached);
   };
 
-  return dedupeRequest(pendingContributions, key, load);
+  if (options.bypassCache) return load(null);
+  return contributionsCache.getOrSet(key, load, LONG_CACHE_TTL, shouldFetch);
 }
 
 async function fetchContributionsUncached(
@@ -579,17 +546,12 @@ export async function fetchUserProfile(
   const key = cacheKey('profile', username);
   const encodedUsername = encodeURIComponent(username);
 
-  if (options.bypassCache) {
-    return fetchProfileUncached(encodedUsername, key, options);
-  }
-
-  const load = async (): Promise<GitHubUserProfile> => {
-    const cached = await profileCache.get(key);
-    if (cached) return cached;
+  const load = async () => {
     return fetchProfileUncached(encodedUsername, key, options);
   };
 
-  return dedupeRequest(pendingProfiles, key, load);
+  if (options.bypassCache) return load();
+  return profileCache.getOrSet(key, load, GITHUB_CACHE_TTL_MS);
 }
 
 async function fetchProfileUncached(
@@ -627,17 +589,12 @@ export async function fetchUserRepos(
   const key = cacheKey('repos', username);
   const encodedUsername = encodeURIComponent(username);
 
-  if (options.bypassCache) {
-    return fetchReposUncached(encodedUsername, key, options);
-  }
-
-  const load = async (): Promise<GitHubRepo[]> => {
-    const cached = await reposCache.get(key);
-    if (cached) return cached;
+  const load = async () => {
     return fetchReposUncached(encodedUsername, key, options);
   };
 
-  return dedupeRequest(pendingRepos, key, load);
+  if (options.bypassCache) return load();
+  return reposCache.getOrSet(key, load, GITHUB_CACHE_TTL_MS);
 }
 
 async function fetchReposUncached(
@@ -945,10 +902,6 @@ export async function fetchContributedRepos(
   options: FetchOptions = {}
 ): Promise<Record<string, unknown>[]> {
   const key = cacheKey('repos:contributed', username);
-  if (!options.bypassCache) {
-    const cached = await contributedReposCache.get(key);
-    if (cached) return cached;
-  }
 
   const load = async () => {
     const query = `
@@ -983,15 +936,11 @@ export async function fetchContributedRepos(
     if (!res.ok) return [];
     const data = await res.json();
     const result = data?.data?.user?.repositoriesContributedTo?.nodes || [];
-
-    if (!options.bypassCache) {
-      await contributedReposCache.set(key, result, GITHUB_CACHE_TTL_MS);
-    }
     return result;
   };
 
   if (options.bypassCache) return load();
-  return dedupeRequest(pendingContributedRepos, key, load);
+  return contributedReposCache.getOrSet(key, load, GITHUB_CACHE_TTL_MS);
 }
 
 export interface DeveloperScoreInput {