fix: test concurrent cache-miss

Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>

Author: ulemons

services/libs/data-access-layer/src/members/base.ts

@@ -4,13 +4,15 @@ import {
   DEFAULT_TENANT_ID,
   Error400,
   RawQueryParser,
+  TimeoutError,
   generateUUIDv1,
+  generateUUIDv4,
   getProperDisplayName,
   groupBy,
 } from '@crowd/common'
 import { formatSql, getDbInstance, prepareForModification } from '@crowd/database'
 import { getServiceLogger } from '@crowd/logging'
-import { RedisClient } from '@crowd/redis'
+import { RedisClient, acquireLock, releaseLock, withSingleFlight } from '@crowd/redis'
 import {
   ALL_PLATFORM_TYPES,
   IMemberContribution,
@@ -37,6 +39,20 @@ import { fetchManyMemberIdentities, fetchManyMemberOrgs, fetchManyMemberSegments
 
 const log = getServiceLogger()
 
+// How long a compute lock covers a legitimately slow query. Also used as the wait
+// timeout for requests single-flighting behind it — the two must not drift apart:
+// a shorter wait timeout lets waiters give up and run the query themselves while the
+// original holder is still legitimately within its TTL, reproducing the stampede this
+// lock exists to prevent. See services/libs/redis/src/singleFlight.ts for the mechanics.
+const COMPUTE_LOCK_TTL_SECONDS = 90
+
+const toCountPage = (count: number, limit: number, offset: number): PageData<IDbMemberData> => ({
+  rows: [],
+  count,
+  limit,
+  offset,
+})
+
 interface IQueryMembersAdvancedParams {
   filter?: Record<string, unknown>
   search?: string | null
@@ -255,42 +271,60 @@ export async function queryMembersAdvanced(
     // refreshing it on every hit would fire a COUNT(*) query per request, defeating the cache.
     // The count is kept fresh by: (1) full-result refreshes that also write countCacheKey,
     // (2) natural TTL expiry, (3) explicit cache invalidation on member updates.
-    return {
-      rows: [],
-      count: cachedCount,
-      limit,
-      offset,
-    }
+    return toCountPage(cachedCount, limit, offset)
   }
 
-  log.info(
-    {
-      cacheKey,
-      countCacheKey,
-      segmentId,
-      search: normalizedSearch,
-      limit,
-      offset,
-      orderBy,
-      countOnly,
-    },
-    'Members advanced query cache miss — executing query synchronously',
-  )
+  // Single-flight: only the request that wins the lock hits the DB. Everyone else
+  // piling onto the same cold key waits on the lock instead of firing the same
+  // heavy query in parallel (this is what turns one slow query into a stampede).
+  const computeLockKey = countOnly ? countCacheKey : cacheKey
+
+  const readComputeCache = async (): Promise<PageData<IDbMemberData> | null> => {
+    if (countOnly) {
+      const count = await cache.getCount(countCacheKey)
+      return count !== null ? toCountPage(count, limit, offset) : null
+    }
+    return cache.get(cacheKey)
+  }
 
   try {
-    return await executeQuery(qx, redis, cacheKey, {
-      filter,
-      search: normalizedSearch,
-      limit,
-      offset,
-      orderBy,
-      segmentId,
-      countOnly,
-      fields,
-      include,
-      includeAllAttributes,
-      attributeSettings,
-    })
+    return await withSingleFlight(
+      redis,
+      computeLockKey,
+      {
+        lockTtlSeconds: COMPUTE_LOCK_TTL_SECONDS,
+        waitTimeoutSeconds: COMPUTE_LOCK_TTL_SECONDS,
+      },
+      readComputeCache,
+      () => {
+        log.info(
+          {
+            cacheKey,
+            countCacheKey,
+            segmentId,
+            search: normalizedSearch,
+            limit,
+            offset,
+            orderBy,
+            countOnly,
+          },
+          'Members advanced query cache miss — executing query synchronously',
+        )
+        return executeQuery(qx, redis, cacheKey, {
+          filter,
+          search: normalizedSearch,
+          limit,
+          offset,
+          orderBy,
+          segmentId,
+          countOnly,
+          fields,
+          include,
+          includeAllAttributes,
+          attributeSettings,
+        })
+      },
+    )
   } catch (error) {
     log.warn(
       { cacheKey, countCacheKey, segmentId, search: normalizedSearch, countOnly, err: error },
@@ -417,12 +451,7 @@ export async function executeQuery(
 
     await cache.setCount(countCacheKey, count, 21600)
 
-    return {
-      rows: [],
-      count,
-      limit,
-      offset,
-    }
+    return toCountPage(count, limit, offset)
   }
 
   // Prepare fields for main query
@@ -449,7 +478,12 @@ export async function executeQuery(
   const mainQuery = buildQuery({
     fields: preparedFields,
     withAggregates,
-    includeMemberOrgs: include.memberOrganizations,
+    // Org data for the returned page is always fetched separately via fetchManyMemberOrgs
+    // below — the 'organizations' field is queryable: false, so it's never selected here.
+    // The mo join is only needed when the filter itself references mo.*, which filterHasMo
+    // (computed inside buildQuery) already covers — mirrors the same fix already applied
+    // to buildCountQuery above.
+    includeMemberOrgs: false,
     searchConfig,
     filterString,
     orderBy,
@@ -625,23 +659,36 @@ async function refreshCacheInBackground(
   countOnly = false,
 ): Promise<void> {
   const label = countOnly ? 'count cache' : 'query cache'
-  const cache = new MemberQueryCache(redis)
-  const acquired = await cache.tryAcquireRefreshLock(cacheKey)
-  if (!acquired) {
-    log.debug(
-      { cacheKey },
-      `Members advanced ${label} refresh already in progress — skipping duplicate`,
-    )
+
+  // Same lock (same key, same acquireLock/releaseLock primitive) that the synchronous
+  // cache-miss path takes via withSingleFlight — a single non-blocking attempt (0s wait)
+  // since this is best-effort background work: if the miss path (or another background
+  // refresh) already holds it, just skip rather than duplicate the query. Sharing the
+  // lock is what stops a hit-triggered refresh and a miss-triggered compute from ever
+  // running the same heavy query concurrently for the same cache key.
+  const token = generateUUIDv4()
+  try {
+    await acquireLock(redis, cacheKey, token, COMPUTE_LOCK_TTL_SECONDS, 0)
+  } catch (error) {
+    if (error instanceof TimeoutError) {
+      log.debug(
+        { cacheKey },
+        `Members advanced ${label} refresh already in progress — skipping duplicate`,
+      )
+    } else {
+      log.warn({ cacheKey, err: error }, `Members advanced ${label} refresh lock unavailable`)
+    }
     return
   }
+
   try {
     log.info({ cacheKey }, `Members advanced ${label} background refresh started`)
     await executeQuery(qx, redis, cacheKey, countOnly ? { ...params, countOnly: true } : params)
     log.info({ cacheKey }, `Members advanced ${label} background refresh completed`)
   } catch (error) {
     log.warn({ cacheKey, err: error }, `Members advanced ${label} background refresh failed`)
   } finally {
-    await cache.releaseRefreshLock(cacheKey)
+    await releaseLock(redis, cacheKey, token)
   }
 }
 

services/libs/data-access-layer/src/members/queryBuilder.ts

@@ -31,6 +31,14 @@ const ORDER_FIELD_MAP: Record<string, string> = {
   displayName: 'm."displayName"',
 }
 
+// Hard cap on candidates considered by the activityCount top-N optimization (see
+// buildActivityCountOptimizedQuery). Kept as a shared constant because
+// canUseActivityCountOptimization has to know it too, to refuse the optimized path
+// when a requested page would extend past the cap — otherwise a large limit/offset
+// (e.g. CSV exports requesting effectively unlimited rows) would come back silently
+// truncated instead of falling back to the exhaustive path.
+const ACTIVITY_COUNT_TOP_N_CAP = 50000
+
 export const buildSearchCTE = (
   search: string,
 ): { cte: string; join: string; params: Record<string, string> } => {
@@ -187,6 +195,9 @@ const hasNonIdMemberFieldReferences = (filterString: string): boolean => {
  * @param sortField - The field being sorted by (undefined means default activityCount)
  * @param hasEnrichmentFilters - Whether filter references me.* fields
  * @param hasOrganizationFilters - Whether filter references mo.* fields
+ * @param limit - Requested page size
+ * @param offset - Requested page offset
+ * @param hasNonIdMemberFields - Whether filter uses m.* fields (drives oversampling)
  * @returns true if activityCount optimization can be used
  */
 const canUseActivityCountOptimization = ({
@@ -195,12 +206,18 @@ const canUseActivityCountOptimization = ({
   includeMemberOrgs,
   sortField,
   withAggregates,
+  limit,
+  offset,
+  hasNonIdMemberFields,
 }: {
   filterHasMe: boolean
   filterHasMo: boolean
   sortField: string | undefined
   withAggregates: boolean
   includeMemberOrgs: boolean
+  limit: number
+  offset: number
+  hasNonIdMemberFields: boolean
 }): boolean => {
   // Need aggregates to access activityCount
   if (!withAggregates) return false
@@ -213,6 +230,13 @@ const canUseActivityCountOptimization = ({
 
   if (includeMemberOrgs) return false
 
+  // The top-N CTE is capped at ACTIVITY_COUNT_TOP_N_CAP candidates. If the requested
+  // page reaches past that cap (e.g. CSV exports requesting an effectively unlimited
+  // number of rows), the optimized path would silently return an incomplete page —
+  // fall back to the exhaustive path instead.
+  const oversampleMultiplier = hasNonIdMemberFields ? 10 : 1
+  if ((limit + offset) * oversampleMultiplier > ACTIVITY_COUNT_TOP_N_CAP) return false
+
   return true
 }
 
@@ -316,7 +340,7 @@ const buildActivityCountOptimizedQuery = ({
 
   const baseNeeded = limit + offset
   const oversampleMultiplier = hasNonIdMemberFields ? 10 : 1 // 10x oversampling for m.* filters
-  const totalNeeded = Math.min(baseNeeded * oversampleMultiplier, 50000) // Cap at 50k
+  const totalNeeded = Math.min(baseNeeded * oversampleMultiplier, ACTIVITY_COUNT_TOP_N_CAP)
 
   ctes.push(
     `
@@ -403,6 +427,9 @@ export const buildQuery = ({
     includeMemberOrgs,
     sortField,
     withAggregates,
+    limit,
+    offset,
+    hasNonIdMemberFields: filterHasNonIdMemberFields,
   })
 
   if (useActivityCountOptimized) {

services/libs/data-access-layer/src/members/queryCache.ts

@@ -1,4 +1,4 @@
-import { createHash, randomBytes } from 'crypto'
+import { createHash } from 'crypto'
 
 import { getServiceLogger } from '@crowd/logging'
 import { RedisCache, RedisClient } from '@crowd/redis'
@@ -27,27 +27,6 @@ export class MemberQueryCache {
     this.lockCache = new RedisCache('members-refresh-lock', redis, log)
   }
 
-  // Returns true if lock was acquired (no other refresh in progress for this key).
-  // Uses a cryptographically random token to distinguish "we set it" from "already existed".
-  // TTL ensures the lock auto-expires if the refresh crashes without releasing it.
-  async tryAcquireRefreshLock(cacheKey: string, ttlSeconds = 90): Promise<boolean> {
-    try {
-      const token = randomBytes(16).toString('hex')
-      const stored = await this.lockCache.setIfNotExistsOrGet(cacheKey, token, ttlSeconds)
-      return stored === token
-    } catch {
-      return true // fail open: if Redis is down, let the refresh proceed
-    }
-  }
-
-  async releaseRefreshLock(cacheKey: string): Promise<void> {
-    try {
-      await this.lockCache.delete(cacheKey)
-    } catch {
-      // best effort
-    }
-  }
-
   buildCacheKey(params: {
     fields?: string[]
     filter?: Record<string, unknown>

services/libs/redis/src/index.ts

@@ -6,3 +6,4 @@ export * from './cache'
 export * from './instances'
 export * from './rateLimiter'
 export * from './mutex'
+export * from './singleFlight'

services/libs/redis/src/singleFlight.ts

@@ -0,0 +1,77 @@
+import { TimeoutError, generateUUIDv4, timeout } from '@crowd/common'
+
+import { acquireLock, releaseLock } from './mutex'
+import { RedisClient } from './types'
+
+export interface SingleFlightOptions {
+  lockTtlSeconds: number
+  waitTimeoutSeconds: number
+}
+
+// 100-300ms jittered, matching mutex.ts's own retry spacing.
+const randomPollDelayMs = () => Math.floor(Math.random() * 200) + 100
+
+/**
+ * Ensures only one caller per `lockKey` actually runs `compute()` at a time. A caller
+ * that loses the initial race polls `readCached()` and retries a non-blocking lock
+ * acquisition every ~100-300ms — so it returns as soon as either the holder finishes
+ * and populates the cache, or the lock frees up (holder crashed/released), without
+ * waiting out the full timeout in either case. If Redis itself is unreachable, lock
+ * acquisition fails open (proceeds unlocked) rather than failing the caller.
+ *
+ * `waitTimeoutSeconds` should be >= `lockTtlSeconds`: otherwise a waiter can give up
+ * and run `compute()` unlocked while the original holder is still legitimately working
+ * within its own TTL window, reproducing the exact stampede this helper exists to avoid.
+ */
+export async function withSingleFlight<T>(
+  client: RedisClient,
+  lockKey: string,
+  { lockTtlSeconds, waitTimeoutSeconds }: SingleFlightOptions,
+  readCached: () => Promise<T | null>,
+  compute: () => Promise<T>,
+): Promise<T> {
+  const token = generateUUIDv4()
+  const deadline = Date.now() + waitTimeoutSeconds * 1000
+  let holdsLock = false
+
+  for (;;) {
+    try {
+      await acquireLock(client, lockKey, token, lockTtlSeconds, 0)
+      holdsLock = true
+      break
+    } catch (error) {
+      if (!(error instanceof TimeoutError)) {
+        // Redis unavailable for locking — fail open, compute unprotected.
+        break
+      }
+    }
+
+    const cached = await readCached()
+    if (cached !== null) {
+      return cached
+    }
+
+    if (Date.now() >= deadline) {
+      break
+    }
+
+    await timeout(randomPollDelayMs())
+  }
+
+  try {
+    if (holdsLock) {
+      // Whoever held the lock right before us may have finished and populated the
+      // cache between our last poll and winning it — avoid a redundant compute.
+      const cached = await readCached()
+      if (cached !== null) {
+        return cached
+      }
+    }
+
+    return await compute()
+  } finally {
+    if (holdsLock) {
+      await releaseLock(client, lockKey, token)
+    }
+  }
+}