🧯 fix: Harden Principals Cache Invalidation and Lock-Wait Paths (#14120)

* 🧯 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

@@ -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

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<string | null>;
   releaseLock?: (key: string, token: string) => Promise<void>;
 };
@@ -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');

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<string | null> => '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',

packages/data-schemas/src/methods/userGroup.ts

@@ -20,12 +20,22 @@ const pendingGroupLookups = new Map<string, PendingGroupLookup>();
 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<void>) => Promise<void>;
+  invalidate: () => Promise<void>;
+};
+
+/** 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<ClientSession, DeferredInvalidation[]>();
+
 /**
  * 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<void>,
 ): Promise<void> {
-  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<string, unknown>): {
   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<Types.ObjectId[]> {
     const Group = mongoose.models.Group as Model<IGroup>;
     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<Array<Pick<IGroup, '_id'>>>();
     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>): 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);

packages/data-schemas/src/types/cache.ts

@@ -8,6 +8,10 @@ export interface CacheStore {
   set: (key: string, value: unknown) => Promise<unknown>;
   delete?: (key: string) => Promise<unknown>;
   clear?: () => Promise<unknown>;
+  /** 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<string | null>;
   releaseLock?: (key: string, token: string) => Promise<unknown>;