diff --git a/services/libs/data-access-layer/src/members/base.ts b/services/libs/data-access-layer/src/members/base.ts index eb5fbd37c3..abebc58303 100644 --- a/services/libs/data-access-layer/src/members/base.ts +++ b/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 => ({ + rows: [], + count, + limit, + offset, +}) + interface IQueryMembersAdvancedParams { filter?: Record 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 | 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,15 +659,28 @@ async function refreshCacheInBackground( countOnly = false, ): Promise { 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) @@ -641,7 +688,7 @@ async function refreshCacheInBackground( } catch (error) { log.warn({ cacheKey, err: error }, `Members advanced ${label} background refresh failed`) } finally { - await cache.releaseRefreshLock(cacheKey) + await releaseLock(redis, cacheKey, token) } } diff --git a/services/libs/data-access-layer/src/members/queryBuilder.ts b/services/libs/data-access-layer/src/members/queryBuilder.ts index 0eb2f044a3..35b88cd4a4 100644 --- a/services/libs/data-access-layer/src/members/queryBuilder.ts +++ b/services/libs/data-access-layer/src/members/queryBuilder.ts @@ -31,6 +31,14 @@ const ORDER_FIELD_MAP: Record = { 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 } => { @@ -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) { diff --git a/services/libs/data-access-layer/src/members/queryCache.ts b/services/libs/data-access-layer/src/members/queryCache.ts index 6eeaff8673..ce511accd4 100644 --- a/services/libs/data-access-layer/src/members/queryCache.ts +++ b/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 { - 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 { - try { - await this.lockCache.delete(cacheKey) - } catch { - // best effort - } - } - buildCacheKey(params: { fields?: string[] filter?: Record diff --git a/services/libs/redis/src/index.ts b/services/libs/redis/src/index.ts index ad58a1c787..ad777b265a 100644 --- a/services/libs/redis/src/index.ts +++ b/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' diff --git a/services/libs/redis/src/singleFlight.ts b/services/libs/redis/src/singleFlight.ts new file mode 100644 index 0000000000..600af94c22 --- /dev/null +++ b/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( + client: RedisClient, + lockKey: string, + { lockTtlSeconds, waitTimeoutSeconds }: SingleFlightOptions, + readCached: () => Promise, + compute: () => Promise, +): Promise { + 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) + } + } +}