Fix race conditions and correctness issues in caching layers

lib/AbstractCache.ts

@@ -225,14 +225,20 @@ export abstract class AbstractCache<
   }
 
   public async invalidateCache() {
-    this.inMemoryCache.clear()
+    // Evict the running loads first so in-flight results are fenced out of the caches.
+    this.runningLoads.clear()
     if (this.asyncCache) {
       await this.asyncCache.clear().catch((err) => {
         this.cacheUpdateErrorHandler(err, undefined, this.asyncCache!, this.logger)
       })
     }
 
+    // Evict again: a load that started while the async clear was in flight may have
+    // read a not-yet-deleted async value; fencing it out here stops it from
+    // repopulating the caches after this invalidation resolves. The in-memory clear
+    // comes last for the same reason.
     this.runningLoads.clear()
+    this.inMemoryCache.clear()
 
     if (this.notificationPublisher) {
       this.notificationPublisher.clear().catch((err) => {

lib/AbstractFlatCache.ts

@@ -48,13 +48,21 @@ export abstract class AbstractFlatCache<LoadedValue, LoadParams = string, LoadMa
 
     loadingPromise
       .then((resolvedValue) => {
+        // If the running load was evicted (invalidation, forceSetValue) while this load
+        // was in flight, its result reflects a snapshot taken before that point and must
+        // not be persisted - callers awaiting the promise still receive the value.
+        if (this.runningLoads.get(key) !== loadingPromise) {
+          return
+        }
         if (resolvedValue !== undefined) {
           this.inMemoryCache.set(key, resolvedValue)
         }
         this.runningLoads.delete(key)
       })
       .catch(() => {
-        this.runningLoads.delete(key)
+        if (this.runningLoads.get(key) === loadingPromise) {
+          this.runningLoads.delete(key)
+        }
       })
 
     return loadingPromise
@@ -131,14 +139,20 @@ export abstract class AbstractFlatCache<LoadedValue, LoadParams = string, LoadMa
   }
 
   public async invalidateCacheFor(key: string) {
-    this.inMemoryCache.delete(key)
+    // Evict the running load first so an in-flight result is fenced out of the caches.
+    this.runningLoads.delete(key)
     if (this.asyncCache) {
       await this.asyncCache.delete(key).catch((err) => {
         this.cacheUpdateErrorHandler(err, undefined, this.asyncCache!, this.logger)
       })
     }
 
+    // Evict again: a load that started while the async delete was in flight may have
+    // read the not-yet-deleted async value; fencing it out here stops it from
+    // repopulating the caches after this invalidation resolves. The in-memory delete
+    // comes last for the same reason.
     this.runningLoads.delete(key)
+    this.inMemoryCache.delete(key)
     if (this.notificationPublisher) {
       this.notificationPublisher.delete(key).catch((err) => {
         this.notificationPublisher!.errorHandler(err, this.notificationPublisher!.channel, this.logger)
@@ -147,6 +161,10 @@ export abstract class AbstractFlatCache<LoadedValue, LoadParams = string, LoadMa
   }
 
   public async invalidateCacheForMany(keys: string[]) {
+    // Evict the running loads first so in-flight results are fenced out of the caches.
+    for (let i = 0; i < keys.length; i++) {
+      this.runningLoads.delete(keys[i])
+    }
     if (this.asyncCache) {
       await this.asyncCache.deleteMany(keys).catch((err) => {
         /* v8 ignore next -- @preserve */

lib/AbstractGroupCache.ts

@@ -19,14 +19,20 @@ export abstract class AbstractGroupCache<LoadedValue, LoadParams = string, LoadM
   }
 
   public async invalidateCacheForGroup(group: string) {
+    // Evict the running loads first so in-flight results are fenced out of the caches.
+    this.runningLoads.delete(group)
     if (this.asyncCache) {
       await this.asyncCache.deleteGroup(group).catch((err) => {
         this.cacheUpdateErrorHandler(err, `group: ${group}`, this.asyncCache!, this.logger)
       })
     }
 
-    this.inMemoryCache.deleteGroup(group)
+    // Evict again: a load that started while the async delete was in flight may have
+    // read the not-yet-deleted async value; fencing it out here stops it from
+    // repopulating the caches after this invalidation resolves. The in-memory delete
+    // comes last for the same reason.
     this.runningLoads.delete(group)
+    this.inMemoryCache.deleteGroup(group)
 
     if (this.notificationPublisher) {
       void this.notificationPublisher.deleteGroup(group).catch((err) => {
@@ -67,13 +73,21 @@ export abstract class AbstractGroupCache<LoadedValue, LoadParams = string, LoadM
 
     loadingPromise
       .then((resolvedValue) => {
+        // If the running load was evicted (group or key invalidation) while this load
+        // was in flight, its result reflects a snapshot taken before that point and must
+        // not be persisted - callers awaiting the promise still receive the value.
+        if (this.runningLoads.get(group) !== groupLoads || groupLoads.get(key) !== loadingPromise) {
+          return
+        }
         if (resolvedValue !== undefined) {
           this.inMemoryCache.setForGroup(key, resolvedValue, group)
         }
         this.deleteGroupRunningLoad(groupLoads, group, key)
       })
       .catch(() => {
-        this.deleteGroupRunningLoad(groupLoads, group, key)
+        if (this.runningLoads.get(group) === groupLoads && groupLoads.get(key) === loadingPromise) {
+          this.deleteGroupRunningLoad(groupLoads, group, key)
+        }
       })
 
     return loadingPromise
@@ -125,15 +139,20 @@ export abstract class AbstractGroupCache<LoadedValue, LoadParams = string, LoadM
   }
 
   public async invalidateCacheFor(key: string, group: string) {
-    this.inMemoryCache.deleteFromGroup(key, group)
+    // Evict the running load first so an in-flight result is fenced out of the caches.
+    this.evictGroupRunningLoad(group, key)
     if (this.asyncCache) {
       await this.asyncCache.deleteFromGroup(key, group).catch((err) => {
         this.cacheUpdateErrorHandler(err, undefined, this.asyncCache!, this.logger)
       })
     }
 
-    const groupLoads = this.resolveGroupLoads(group)
-    this.deleteGroupRunningLoad(groupLoads, group, key)
+    // Evict again: a load that started while the async delete was in flight may have
+    // read the not-yet-deleted async value; fencing it out here stops it from
+    // repopulating the caches after this invalidation resolves. The in-memory delete
+    // comes last for the same reason.
+    this.evictGroupRunningLoad(group, key)
+    this.inMemoryCache.deleteFromGroup(key, group)
 
     if (this.notificationPublisher) {
       void this.notificationPublisher.deleteFromGroup(key, group).catch((err) => {
@@ -178,6 +197,13 @@ export abstract class AbstractGroupCache<LoadedValue, LoadParams = string, LoadM
     }
   }
 
+  private evictGroupRunningLoad(group: string, key: string) {
+    const groupLoads = this.runningLoads.get(group)
+    if (groupLoads) {
+      this.deleteGroupRunningLoad(groupLoads, group, key)
+    }
+  }
+
   protected resolveGroupLoads(group: string) {
     const load = this.runningLoads.get(group)
     if (load) {
@@ -191,7 +217,10 @@ export abstract class AbstractGroupCache<LoadedValue, LoadParams = string, LoadM
 
   protected deleteGroupRunningLoad(groupLoads: Map<string, unknown>, group: string, key: string) {
     groupLoads.delete(key)
-    if (groupLoads.size === 0) {
+    // Only drop the group entry if it still points at this map - a group invalidation
+    // may have detached it and a newer load registered a fresh map under the same group,
+    // which must not be evicted by a load settling against the stale map.
+    if (groupLoads.size === 0 && this.runningLoads.get(group) === groupLoads) {
       this.runningLoads.delete(group)
     }
   }

lib/GroupLoader.ts

@@ -51,31 +51,38 @@ export class GroupLoader<LoadedValue, LoadParams = string, LoadManyParams = Load
           let isAlreadyRefreshing = groupSet?.has(key)
 
           if (!isAlreadyRefreshing) {
-            this.asyncCache.expirationTimeLoadingGroupedOperation.get(key, group).then((expirationTime) => {
-              if (expirationTime && expirationTime - Date.now() < this.asyncCache!.ttlLeftBeforeRefreshInMsecs!) {
-                // Check if someone else didn't start refreshing while we were checking expiration time
-                groupSet = this.groupRefreshFlags.get(group)
-                isAlreadyRefreshing = groupSet?.has(key)
-                if (!isAlreadyRefreshing) {
-                  if (!groupSet) {
-                    groupSet = new Set<string>()
-                    this.groupRefreshFlags.set(group, groupSet)
+            this.asyncCache.expirationTimeLoadingGroupedOperation
+              .get(key, group)
+              .then((expirationTime) => {
+                if (expirationTime && expirationTime - Date.now() < this.asyncCache!.ttlLeftBeforeRefreshInMsecs!) {
+                  // Check if someone else didn't start refreshing while we were checking expiration time
+                  groupSet = this.groupRefreshFlags.get(group)
+                  isAlreadyRefreshing = groupSet?.has(key)
+                  if (!isAlreadyRefreshing) {
+                    if (!groupSet) {
+                      groupSet = new Set<string>()
+                      this.groupRefreshFlags.set(group, groupSet)
+                    }
+                    groupSet.add(key)
+
+                    this.refreshOrBumpTtl(key, group, loadParams, cachedValue)
+                      .catch((err) => {
+                        this.logger.error(err.message)
+                      })
+                      .finally(() => {
+                        groupSet!.delete(key)
+                        if (groupSet!.size === 0) {
+                          this.groupRefreshFlags.delete(group)
+                        }
+                      })
                   }
-                  groupSet.add(key)
-
-                  this.refreshOrBumpTtl(key, group, loadParams, cachedValue)
-                    .catch((err) => {
-                      this.logger.error(err.message)
-                    })
-                    .finally(() => {
-                      groupSet!.delete(key)
-                      if (groupSet!.size === 0) {
-                        this.groupRefreshFlags.delete(group)
-                      }
-                    })
                 }
-              }
-            })
+              })
+              .catch((err) => {
+                // expiration lookup is fire-and-forget; a rejection here must not become
+                // an unhandled promise rejection
+                this.logger.error(err.message)
+              })
           }
         }
         return cachedValue

lib/Loader.ts

@@ -120,21 +120,28 @@ export class Loader<LoadedValue, LoadParams = string, LoadManyParams = LoadParam
       if (cachedValue !== undefined) {
         if (this.asyncCache?.ttlLeftBeforeRefreshInMsecs) {
           if (!this.isKeyRefreshing.has(key)) {
-            this.asyncCache.expirationTimeLoadingOperation.get(key).then((expirationTime) => {
-              if (expirationTime && expirationTime - Date.now() < this.asyncCache!.ttlLeftBeforeRefreshInMsecs!) {
-                // check second time, maybe someone obtained the lock while we were checking the expiration date
-                if (!this.isKeyRefreshing.has(key)) {
-                  this.isKeyRefreshing.add(key)
-                  this.refreshOrBumpTtl(key, loadParams, cachedValue)
-                    .catch((err) => {
-                      this.logger.error(err.message)
-                    })
-                    .finally(() => {
-                      this.isKeyRefreshing.delete(key)
-                    })
+            this.asyncCache.expirationTimeLoadingOperation
+              .get(key)
+              .then((expirationTime) => {
+                if (expirationTime && expirationTime - Date.now() < this.asyncCache!.ttlLeftBeforeRefreshInMsecs!) {
+                  // check second time, maybe someone obtained the lock while we were checking the expiration date
+                  if (!this.isKeyRefreshing.has(key)) {
+                    this.isKeyRefreshing.add(key)
+                    this.refreshOrBumpTtl(key, loadParams, cachedValue)
+                      .catch((err) => {
+                        this.logger.error(err.message)
+                      })
+                      .finally(() => {
+                        this.isKeyRefreshing.delete(key)
+                      })
+                  }
                 }
-              }
-            })
+              })
+              .catch((err) => {
+                // expiration lookup is fire-and-forget; a rejection here must not become
+                // an unhandled promise rejection
+                this.logger.error(err.message)
+              })
           }
         }
 

lib/memory/InMemoryCache.ts

@@ -60,8 +60,9 @@ export class InMemoryCache<T> implements SynchronousCache<T> {
 
     for (let i = 0; i < keys.length; i++) {
       const resolvedValue = this.cache.get(keys[i])
-      if (resolvedValue) {
-        resolvedValues.push(resolvedValue)
+      // null is a valid cached value ("resolved to empty"), only undefined means a miss
+      if (resolvedValue !== undefined) {
+        resolvedValues.push(resolvedValue as T)
       } else {
         unresolvedKeys.push(keys[i])
       }

lib/memory/InMemoryGroupCache.ts

@@ -77,19 +77,28 @@ export class InMemoryGroupCache<T> implements SynchronousGroupCache<T> {
   }
 
   getFromGroup(key: string, groupId: string) {
-    const group = this.resolveGroup(groupId)
-    return group.get(key)
+    // do not use resolveGroup here - reads must not insert empty groups into the LRU,
+    // or a miss on a nonexistent group could evict a real group's still-valid data
+    return this.groups.get(groupId)?.get(key)
   }
 
   getManyFromGroup(keys: string[], group: string): GetManyResult<T> {
     const resolvedValues: T[] = []
     const unresolvedKeys: string[] = []
-    const groupCache = this.resolveGroup(group)
+    const groupCache = this.groups.get(group)
+
+    if (!groupCache) {
+      return {
+        resolvedValues,
+        unresolvedKeys: [...keys],
+      }
+    }
 
     for (let i = 0; i < keys.length; i++) {
       const resolvedValue = groupCache.get(keys[i])
-      if (resolvedValue) {
-        resolvedValues.push(resolvedValue)
+      // null is a valid cached value ("resolved to empty"), only undefined means a miss
+      if (resolvedValue !== undefined) {
+        resolvedValues.push(resolvedValue as T)
       } else {
         unresolvedKeys.push(keys[i])
       }
@@ -107,16 +116,16 @@ export class InMemoryGroupCache<T> implements SynchronousGroupCache<T> {
   }
 
   deleteFromGroup(key: string, groupId: string): void {
-    const group = this.resolveGroup(groupId)
-    group.delete(key)
+    // do not use resolveGroup here - reads must not insert empty groups into the LRU
+    this.groups.get(groupId)?.delete(key)
   }
 
   clear(): void {
     this.groups.clear()
   }
 
   getExpirationTimeFromGroup(key: string, groupId: string): number | undefined {
-    const group = this.resolveGroup(groupId)
-    return group.expiresAt(key)
+    // do not use resolveGroup here - reads must not insert empty groups into the LRU
+    return this.groups.get(groupId)?.expiresAt(key)
   }
 }

lib/redis/RedisGroupCache.ts

@@ -50,8 +50,11 @@ export class RedisGroupCache<T> extends AbstractRedisCache<RedisGroupCacheConfig
 
   async deleteGroup(group: string) {
     const key = this.resolveGroupIndexPrefix(group)
-    if (this.config.ttlInMsecs) {
-      await this.redis.multi().incr(key).pexpire(key, this.config.ttlInMsecs).exec()
+    // the group index key's lifetime is governed by groupTtlInMsecs everywhere else
+    // (see setForGroup/setManyForGroup); using entry ttlInMsecs here would silently
+    // re-scope the generation counter to the entry TTL
+    if (this.config.groupTtlInMsecs) {
+      await this.redis.multi().incr(key).pexpire(key, this.config.groupTtlInMsecs).exec()
       return
     }