From 446b73bb6bad3e79ba333026c625fb257632698a Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 5 Jul 2026 11:31:06 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=AF=20fix:=20Harden=20Principals=20Cac?= =?UTF-8?q?he=20Invalidation=20and=20Lock-Wait=20Paths=20(#14120)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 🧯 fix: Harden Principals Cache Invalidation and Lock-Wait Paths Follow-up hardening for the group-membership principals cache: - Coalesce deferred transactional invalidations into one session `ended` listener so many membership writes in a transaction cannot trip the emitter's max-listeners warning - Re-attempt the build lock inside the lock-wait poll so readers take over when the holder skips its write (invalidated away, failed, or crashed) instead of stalling for the full wait budget - Read cache builds from the primary so secondaryPreferred deployments cannot pin a lagging secondary's pre-mutation memberships for the TTL - Decouple the delayed stale-rewrite eviction pass from build locking via a crossProcess store marker, so USER_PRINCIPALS_LOCK_TTL_MS=0 disables locking only - Treat aggregation-pipeline bulk updates touching memberIds as indeterminate (namespace clear) instead of silently skipping invalidation * ⏲️ fix: Floor the Stale-Eviction Delay for Lockless Cache Builds The second invalidation pass ran at only 500ms when build locking is disabled (USER_PRINCIPALS_LOCK_TTL_MS=0), so a cross-container build slower than that could still re-cache revoked memberships until the TTL. The store now supplies an explicit staleEvictionDelayMs (lock wait plus one build round-trip, floored at 3s), keeping the default locked budget unchanged. --- .env.example | 3 +- packages/api/src/cache/principals.ts | 12 +- .../src/methods/userGroup.spec.ts | 142 ++++++++++++++++++ .../data-schemas/src/methods/userGroup.ts | 89 ++++++++--- packages/data-schemas/src/types/cache.ts | 4 + 5 files changed, 226 insertions(+), 24 deletions(-) diff --git a/.env.example b/.env.example index d29b6d0c85..c8049f1d94 100644 --- a/.env.example +++ b/.env.example @@ -997,7 +997,8 @@ HELP_AND_FAQ_URL=https://librechat.ai # TTL in milliseconds for cached group memberships used in ACL permission checks (default: 300000 / 5 minutes; 0 disables) # Membership changes invalidate affected entries immediately; the TTL bounds staleness from cross-process races. # USER_PRINCIPALS_CACHE_TTL_MS=300000 -# Redis lock TTL in milliseconds for cross-container cache builds (default: 5000; 0 disables locking) +# Redis lock TTL in milliseconds for cross-container cache builds (default: 5000) +# 0 disables build locking only; the delayed stale-rewrite eviction pass still runs on Redis-backed stores. # Only used when the USER_PRINCIPALS namespace is Redis-backed; non-Redis deployments use in-process deduplication. # USER_PRINCIPALS_LOCK_TTL_MS=5000 # Maximum time in milliseconds to wait for another container holding the lock to fill the cache diff --git a/packages/api/src/cache/principals.ts b/packages/api/src/cache/principals.ts index ad0009d284..5730a3c7ed 100644 --- a/packages/api/src/cache/principals.ts +++ b/packages/api/src/cache/principals.ts @@ -11,7 +11,9 @@ const lockTtl = math(process.env.USER_PRINCIPALS_LOCK_TTL_MS, 5000); const lockWait = math(process.env.USER_PRINCIPALS_LOCK_WAIT_MS, lockTtl); export type UserPrincipalsCache = Keyv & { + crossProcess?: boolean; lockWaitMs?: number; + staleEvictionDelayMs?: number; acquireLock?: (key: string) => Promise; releaseLock?: (key: string, token: string) => Promise; }; @@ -44,8 +46,16 @@ export function userPrincipalsCache(): UserPrincipalsCache | undefined { const isRedisBacked = keyvRedisClient != null && !cacheConfig.FORCED_IN_MEMORY_CACHE_NAMESPACES?.includes(CacheKeys.USER_PRINCIPALS); - if (isRedisBacked && redisClient != null && lockTtl > 0) { + if (isRedisBacked) { + /** Marks the store shared across containers; the delayed stale-rewrite eviction + * pass depends on this even when build locking is disabled (lock TTL of 0). */ + cache.crossProcess = true; cache.lockWaitMs = Math.max(lockWait, 0); + /** Lock wait plus one build round-trip, floored so lockless configurations + * (lock TTL of 0) still cover multi-second builds in other containers. */ + cache.staleEvictionDelayMs = Math.max(cache.lockWaitMs + 500, 3000); + } + if (isRedisBacked && redisClient != null && lockTtl > 0) { cache.acquireLock = async (key) => { const token = randomUUID(); const result = await redisClient.set(key, token, 'PX', lockTtl, 'NX'); diff --git a/packages/data-schemas/src/methods/userGroup.spec.ts b/packages/data-schemas/src/methods/userGroup.spec.ts index 0fcc849a44..a825952823 100644 --- a/packages/data-schemas/src/methods/userGroup.spec.ts +++ b/packages/data-schemas/src/methods/userGroup.spec.ts @@ -1062,9 +1062,11 @@ describe('userGroup methods', () => { }); const cache = createFakeCache(); const lockedCache = Object.assign(cache, { + crossProcess: true, acquireLock: jest.fn(async () => 'lock-token'), releaseLock: jest.fn(async () => undefined), lockWaitMs: 0, + staleEvictionDelayMs: 200, }); const cachedMethods = createUserGroupMethods(mongoose, { getCache: jest.fn(() => lockedCache), @@ -1088,6 +1090,146 @@ describe('userGroup methods', () => { ]); }); + it('runs the delayed second invalidation on cross-process stores without build locks', async () => { + const user = await createTestUser({ idOnTheSource: 'nolock-ext-1' }); + await Group.create({ + name: 'No Lock Team', + source: 'entra', + idOnTheSource: 'nolock-grp-1', + memberIds: ['nolock-ext-1'], + }); + const cache = createFakeCache(); + const crossProcessCache = Object.assign(cache, { + crossProcess: true, + lockWaitMs: 0, + staleEvictionDelayMs: 200, + }); + const cachedMethods = createUserGroupMethods(mongoose, { + getCache: jest.fn(() => crossProcessCache), + }); + + await cachedMethods.removeUserFromAllGroups(user._id.toString()); + expect(cache.delete).toHaveBeenCalledTimes(1); + + cache.store.set(user._id.toString(), []); + await new Promise((resolve) => setTimeout(resolve, 700)); + + expect(cache.store.has(user._id.toString())).toBe(false); + }); + + it('takes over the build when the lock frees without a cache fill', async () => { + const user = await createTestUser({ idOnTheSource: 'takeover-ext-1' }); + const group = await Group.create({ + name: 'Takeover Team', + source: 'entra', + idOnTheSource: 'takeover-grp-1', + memberIds: ['takeover-ext-1'], + }); + const cache = createFakeCache(); + const lockedCache = Object.assign(cache, { + crossProcess: true, + acquireLock: jest + .fn(async (): Promise => 'lock-token') + .mockResolvedValueOnce(null), + releaseLock: jest.fn(async () => undefined), + lockWaitMs: 5000, + }); + const cachedMethods = createUserGroupMethods(mongoose, { + getCache: jest.fn(() => lockedCache), + }); + + const startedAt = Date.now(); + const principals = await cachedMethods.getUserPrincipals({ + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'takeover-ext-1', + }); + + /** The poller re-attempts the lock instead of sleeping out the 5s wait budget */ + expect(Date.now() - startedAt).toBeLessThan(1500); + expect(groupPrincipalIds(principals)).toEqual([group._id.toString()]); + expect(lockedCache.acquireLock).toHaveBeenCalledTimes(2); + expect(lockedCache.releaseLock).toHaveBeenCalledWith( + 'USER_PRINCIPALS_LOCK:takeover-ext-1', + 'lock-token', + ); + }); + + it('reads cache builds from the primary, and uncached reads from the connection default', async () => { + const user = await createTestUser({ idOnTheSource: 'primary-ext-1' }); + await Group.create({ + name: 'Primary Team', + source: 'entra', + idOnTheSource: 'primary-grp-1', + memberIds: ['primary-ext-1'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'primary-ext-1', + }; + + const readSpy = jest.spyOn(mongoose.Query.prototype, 'read'); + await cachedMethods.getUserPrincipals(params); + expect(readSpy).toHaveBeenCalledWith('primary'); + + readSpy.mockClear(); + await methods.getUserPrincipals(params); + expect(readSpy).not.toHaveBeenCalledWith('primary'); + readSpy.mockRestore(); + }); + + it('coalesces deferred invalidations into one session listener', async () => { + const user = await createTestUser({ idOnTheSource: 'coalesce-ext-1' }); + const groups = await Group.create( + Array.from({ length: 12 }, (_, index) => ({ + name: `Coalesce Team ${index}`, + source: 'entra' as const, + idOnTheSource: `coalesce-grp-${index}`, + memberIds: [], + })), + ); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + + const session = await mongoose.startSession(); + session.startTransaction(); + const baseListenerCount = session.listenerCount('ended'); + for (const group of groups) { + await cachedMethods.addUserToGroup(user._id, group._id, session); + } + expect(session.listenerCount('ended')).toBe(baseListenerCount + 1); + expect(cache.delete).not.toHaveBeenCalled(); + + await session.commitTransaction(); + await session.endSession(); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(cache.delete).toHaveBeenCalledWith('coalesce-ext-1'); + expect(cache.delete.mock.calls.length).toBeGreaterThanOrEqual(groups.length); + }); + + it('clears the cache for aggregation-pipeline bulk updates touching memberIds', async () => { + await Group.create({ + name: 'Pipeline Team', + source: 'entra', + idOnTheSource: 'pipe-grp-1', + memberIds: ['pipe-ext-1'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const pipelineUpdate = [{ $set: { memberIds: ['pipe-ext-2'] } }] as unknown as Record< + string, + unknown + >; + + await cachedMethods.bulkUpdateGroups({ idOnTheSource: 'pipe-grp-1' }, pipelineUpdate); + + expect(cache.clear).toHaveBeenCalledTimes(1); + }); + it('skips the namespace clear for no-op bulk membership replacements', async () => { await Group.create({ name: 'Noop Team', diff --git a/packages/data-schemas/src/methods/userGroup.ts b/packages/data-schemas/src/methods/userGroup.ts index e78c1985de..cf31eb7298 100644 --- a/packages/data-schemas/src/methods/userGroup.ts +++ b/packages/data-schemas/src/methods/userGroup.ts @@ -20,12 +20,22 @@ const pendingGroupLookups = new Map(); const GROUP_LOCK_POLL_MS = 50; /** Above this many member keys, invalidation clears the namespace instead of fanning out deletes. */ const INVALIDATION_CLEAR_THRESHOLD = 1000; -/** Margin added to the lock wait when scheduling the delayed second invalidation pass. */ -const STALE_WRITE_GRACE_MS = 500; +/** Fallback delay for the second invalidation pass when the store sets none. */ +const DEFAULT_STALE_EVICTION_DELAY_MS = 3000; const isCachedGroupId = (value: unknown): value is string => typeof value === 'string' && isValidObjectIdString(value); +type DeferredInvalidation = { + /** ALS snapshot (tenant scoping) captured where the mutation ran. */ + run: (invalidate: () => Promise) => Promise; + invalidate: () => Promise; +}; + +/** One queue (and one `ended` listener) per session, so many mutations in one transaction + * cannot trip the emitter's max-listeners warning. Weak keys drop abandoned sessions. */ +const sessionInvalidations = new WeakMap(); + /** * Runs cache invalidation immediately, or defers it to session end for transactional * writes. Mid-transaction invalidation would let concurrent readers re-cache pre-commit @@ -36,16 +46,24 @@ function runAfterTransaction( session: ClientSession | undefined, invalidate: () => Promise, ): Promise { - if (session?.inTransaction()) { - /** The `ended` listener runs in the emitter's context; snapshot the caller's ALS - * (tenant scoping) so the deferred invalidation still derives tenant-scoped keys. */ - const runInCallerContext = AsyncLocalStorage.snapshot(); - session.once('ended', () => { - runInCallerContext(invalidate).catch(() => undefined); - }); + if (!session?.inTransaction()) { + return invalidate(); + } + const deferred: DeferredInvalidation = { run: AsyncLocalStorage.snapshot(), invalidate }; + const queue = sessionInvalidations.get(session); + if (queue) { + queue.push(deferred); return Promise.resolve(); } - return invalidate(); + const newQueue: DeferredInvalidation[] = [deferred]; + sessionInvalidations.set(session, newQueue); + session.once('ended', () => { + sessionInvalidations.delete(session); + for (const entry of newQueue) { + entry.run(entry.invalidate).catch(() => undefined); + } + }); + return Promise.resolve(); } /** Extracts member ids referenced by a group update; indeterminate when they cannot be enumerated. */ @@ -53,6 +71,10 @@ function collectMemberIdsFromGroupUpdate(update: Record): { memberIds: string[]; indeterminate: boolean; } { + if (Array.isArray(update)) { + /** Aggregation-pipeline updates: affected members cannot be statically enumerated. */ + return { memberIds: [], indeterminate: true }; + } const memberIds: string[] = []; const collect = (value: unknown): boolean => { @@ -224,12 +246,21 @@ export function createUserGroupMethods( async function queryGroupIds( memberId: string, session?: ClientSession, + readPrimary = false, ): Promise { const Group = mongoose.models.Group as Model; const groupsQuery = Group.find({ memberIds: memberId }, { _id: 1 }); if (session) { groupsQuery.session(session); } + if (readPrimary && !session) { + /** + * Cache builds must not capture a lagging secondary's pre-mutation state for the + * full TTL (`secondaryPreferred` deployments); uncached reads keep the connection + * default, where staleness is bounded by replication lag as before. + */ + groupsQuery.read('primary'); + } const groups = await groupsQuery.lean>>(); return groups.map((group) => group._id); } @@ -292,13 +323,7 @@ export function createUserGroupMethods( } catch { lockFailed = true; } - if (lockToken) { - /** Another process may have filled the key between the fast-path miss and the lock. */ - const cached = await readCachedGroupIds(cache, cacheKey); - if (cached) { - return cached; - } - } else if (!lockFailed) { + if (!lockToken && !lockFailed) { const waitUntil = Date.now() + Math.max(cache.lockWaitMs ?? 0, 0); while (Date.now() < waitUntil) { await new Promise((resolve) => setTimeout(resolve, GROUP_LOCK_POLL_MS)); @@ -306,11 +331,31 @@ export function createUserGroupMethods( if (cached) { return cached; } + /** + * Re-attempt the lock each tick: a holder whose write was invalidated away + * (or that crashed) never fills the key, so take over the build as soon as + * the lock frees instead of sleeping out the full wait budget. + */ + try { + lockToken = await cache.acquireLock(lockKey); + } catch { + break; + } + if (lockToken) { + break; + } + } + } + if (lockToken) { + /** Another process may have filled the key between the fast-path miss and the lock. */ + const cached = await readCachedGroupIds(cache, cacheKey); + if (cached) { + return cached; } } } - const groupIds = await queryGroupIds(memberId); + const groupIds = await queryGroupIds(memberId, undefined, true); if (!stale) { try { await cache.set( @@ -370,14 +415,14 @@ export function createUserGroupMethods( /** * Cross-process builds cannot see this process's stale flags, so a build in another * container that read pre-mutation state can re-cache it after the first delete. - * A delayed second pass (bounded by the lock wait plus one query round-trip) evicts - * such rewrites; only Redis-backed stores (lock-capable) share data across processes. + * A delayed second pass evicts such rewrites on stores shared across processes, + * independent of build locking; the store supplies the delay budget. */ function scheduleSecondInvalidation(cache: CacheStore, secondPass: () => Promise): void { - if (!cache.acquireLock) { + if (!cache.crossProcess) { return; } - const graceMs = Math.max(cache.lockWaitMs ?? 0, 0) + STALE_WRITE_GRACE_MS; + const graceMs = Math.max(cache.staleEvictionDelayMs ?? DEFAULT_STALE_EVICTION_DELAY_MS, 0); const timer = setTimeout(() => { secondPass().catch(() => undefined); }, graceMs); diff --git a/packages/data-schemas/src/types/cache.ts b/packages/data-schemas/src/types/cache.ts index 5db84dbacf..2181f83812 100644 --- a/packages/data-schemas/src/types/cache.ts +++ b/packages/data-schemas/src/types/cache.ts @@ -8,6 +8,10 @@ export interface CacheStore { set: (key: string, value: unknown) => Promise; delete?: (key: string) => Promise; clear?: () => Promise; + /** True when the store is shared across processes (e.g. Redis-backed). */ + crossProcess?: boolean; + /** Delay before the second invalidation pass that evicts cross-process stale rewrites. */ + staleEvictionDelayMs?: number; /** Acquires a cross-process build lock; resolves a release token, or null when already held. */ acquireLock?: (key: string) => Promise; releaseLock?: (key: string, token: string) => Promise;