diff --git a/lib/AbstractCache.ts b/lib/AbstractCache.ts index 4be3a33..358032c 100644 --- a/lib/AbstractCache.ts +++ b/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) => { diff --git a/lib/AbstractFlatCache.ts b/lib/AbstractFlatCache.ts index d46ecf7..251603d 100644 --- a/lib/AbstractFlatCache.ts +++ b/lib/AbstractFlatCache.ts @@ -48,13 +48,21 @@ export abstract class AbstractFlatCache { + // 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 { 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 { /* v8 ignore next -- @preserve */ diff --git a/lib/AbstractGroupCache.ts b/lib/AbstractGroupCache.ts index 80026c2..90f5e14 100644 --- a/lib/AbstractGroupCache.ts +++ b/lib/AbstractGroupCache.ts @@ -19,14 +19,20 @@ export abstract class AbstractGroupCache { 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 { + // 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 { 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, 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) } } diff --git a/lib/GroupLoader.ts b/lib/GroupLoader.ts index 055a890..d7b4cb4 100644 --- a/lib/GroupLoader.ts +++ b/lib/GroupLoader.ts @@ -51,31 +51,38 @@ export class GroupLoader { - 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() - 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() + 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 diff --git a/lib/Loader.ts b/lib/Loader.ts index 0fda48e..76fa4ff 100644 --- a/lib/Loader.ts +++ b/lib/Loader.ts @@ -120,21 +120,28 @@ export class Loader { - 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) + }) } } diff --git a/lib/memory/InMemoryCache.ts b/lib/memory/InMemoryCache.ts index e29ae1b..3b07ae2 100644 --- a/lib/memory/InMemoryCache.ts +++ b/lib/memory/InMemoryCache.ts @@ -60,8 +60,9 @@ export class InMemoryCache implements SynchronousCache { 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]) } diff --git a/lib/memory/InMemoryGroupCache.ts b/lib/memory/InMemoryGroupCache.ts index e32ada7..2c52f80 100644 --- a/lib/memory/InMemoryGroupCache.ts +++ b/lib/memory/InMemoryGroupCache.ts @@ -77,19 +77,28 @@ export class InMemoryGroupCache implements SynchronousGroupCache { } 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 { 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,8 +116,8 @@ export class InMemoryGroupCache implements SynchronousGroupCache { } 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 { @@ -116,7 +125,7 @@ export class InMemoryGroupCache implements SynchronousGroupCache { } 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) } } diff --git a/lib/redis/RedisGroupCache.ts b/lib/redis/RedisGroupCache.ts index fcd0d41..71b26b1 100644 --- a/lib/redis/RedisGroupCache.ts +++ b/lib/redis/RedisGroupCache.ts @@ -50,8 +50,11 @@ export class RedisGroupCache extends AbstractRedisCache extends AbstractNotificationConsumer< LoadedValue, @@ -39,30 +40,45 @@ export class RedisGroupNotificationConsumer extends AbstractNotific } subscribe(): Promise { - return this.redis.subscribe(this.channel).then(() => { - this.messageHandler = (channel, message) => { - const parsedMessage: GroupNotificationCommand = JSON.parse(message) - // this is a local message, ignore - if (parsedMessage.originUuid === this.serverUuid) { - return - } + // The listener is registered before subscribing so that no message can slip through + // between the SUBSCRIBE ack and the listener attachment. + this.messageHandler = (channel, message) => { + // ioredis emits 'message' for every channel the connection is subscribed to; + // without this check a shared consumer connection would cross-apply commands + // from other channels against this target cache + if (channel !== this.channel) { + return + } - if (parsedMessage.actionId === 'DELETE_FROM_GROUP') { - return this.targetCache.deleteFromGroup( - (parsedMessage as DeleteFromGroupNotificationCommand).key, - (parsedMessage as DeleteFromGroupNotificationCommand).group, - ) - } + let parsedMessage: GroupNotificationCommand + try { + parsedMessage = JSON.parse(message) + } catch (err) { + this.errorHandler(err as Error, this.channel, defaultLogger) + return + } + // this is a local message, ignore + if (parsedMessage.originUuid === this.serverUuid) { + return + } - if (parsedMessage.actionId === 'DELETE_GROUP') { - return this.targetCache.deleteGroup((parsedMessage as DeleteGroupNotificationCommand).group) - } + if (parsedMessage.actionId === 'DELETE_FROM_GROUP') { + return this.targetCache.deleteFromGroup( + (parsedMessage as DeleteFromGroupNotificationCommand).key, + (parsedMessage as DeleteFromGroupNotificationCommand).group, + ) + } - if (parsedMessage.actionId === 'CLEAR') { - return this.targetCache.clear() - } + if (parsedMessage.actionId === 'DELETE_GROUP') { + return this.targetCache.deleteGroup((parsedMessage as DeleteGroupNotificationCommand).group) } - this.redis.on('message', this.messageHandler) - }) + + if (parsedMessage.actionId === 'CLEAR') { + return this.targetCache.clear() + } + } + this.redis.on('message', this.messageHandler) + + return this.redis.subscribe(this.channel).then(() => {}) } } diff --git a/lib/redis/RedisNotificationConsumer.ts b/lib/redis/RedisNotificationConsumer.ts index a41f8ab..5492d91 100644 --- a/lib/redis/RedisNotificationConsumer.ts +++ b/lib/redis/RedisNotificationConsumer.ts @@ -1,6 +1,7 @@ import type { Redis } from 'ioredis' import { AbstractNotificationConsumer } from '../notifications/AbstractNotificationConsumer' import type { SynchronousCache } from '../types/SyncDataSources' +import { defaultLogger } from '../util/Logger' import type { DeleteManyNotificationCommand, DeleteNotificationCommand, @@ -44,34 +45,49 @@ export class RedisNotificationConsumer extends AbstractNotification } subscribe(): Promise { - return this.redis.subscribe(this.channel).then(() => { - this.messageHandler = (channel, message) => { - const parsedMessage: NotificationCommand = JSON.parse(message) - // this is a local message, ignore - if (parsedMessage.originUuid === this.serverUuid) { - return - } + // The listener is registered before subscribing so that no message can slip through + // between the SUBSCRIBE ack and the listener attachment. + this.messageHandler = (channel, message) => { + // ioredis emits 'message' for every channel the connection is subscribed to; + // without this check a shared consumer connection would cross-apply commands + // from other channels against this target cache + if (channel !== this.channel) { + return + } - if (parsedMessage.actionId === 'DELETE') { - return this.targetCache.delete((parsedMessage as DeleteNotificationCommand).key) - } + let parsedMessage: NotificationCommand + try { + parsedMessage = JSON.parse(message) + } catch (err) { + this.errorHandler(err as Error, this.channel, defaultLogger) + return + } + // this is a local message, ignore + if (parsedMessage.originUuid === this.serverUuid) { + return + } + + if (parsedMessage.actionId === 'DELETE') { + return this.targetCache.delete((parsedMessage as DeleteNotificationCommand).key) + } - if (parsedMessage.actionId === 'DELETE_MANY') { - return this.targetCache.deleteMany((parsedMessage as DeleteManyNotificationCommand).keys) - } + if (parsedMessage.actionId === 'DELETE_MANY') { + return this.targetCache.deleteMany((parsedMessage as DeleteManyNotificationCommand).keys) + } - if (parsedMessage.actionId === 'CLEAR') { - return this.targetCache.clear() - } + if (parsedMessage.actionId === 'CLEAR') { + return this.targetCache.clear() + } - if (parsedMessage.actionId === 'SET') { - return this.targetCache.set( - (parsedMessage as SetNotificationCommand).key, - (parsedMessage as SetNotificationCommand).value, - ) - } + if (parsedMessage.actionId === 'SET') { + return this.targetCache.set( + (parsedMessage as SetNotificationCommand).key, + (parsedMessage as SetNotificationCommand).value, + ) } - this.redis.on('message', this.messageHandler) - }) + } + this.redis.on('message', this.messageHandler) + + return this.redis.subscribe(this.channel).then(() => {}) } } diff --git a/test/GroupLoader-async-refresh.spec.ts b/test/GroupLoader-async-refresh.spec.ts index c0c4e65..9e983fc 100644 --- a/test/GroupLoader-async-refresh.spec.ts +++ b/test/GroupLoader-async-refresh.spec.ts @@ -5,8 +5,10 @@ import { GroupLoader } from '../lib/GroupLoader' import { RedisGroupCache } from '../lib/redis/RedisGroupCache' import { CountingGroupedLoader } from './fakes/CountingGroupedLoader' import { DelayedCountingGroupedLoader } from './fakes/DelayedCountingGroupedLoader' +import { DummyGroupedCache } from './fakes/DummyGroupedCache' import { redisOptions } from './fakes/TestRedisConfig' import type { User } from './types/testTypes' +import { waitAndRetry } from './utils/waitUtils' const user1: User = { companyId: '1', @@ -242,6 +244,55 @@ describe('GroupLoader Async Refresh', () => { expect(groupRefreshFlags.has(user1.companyId)).toBe(false) }) + it('two-layer cache propagates fresh value to in-memory after background refresh', async () => { + const loader = new CountingGroupedLoader(userValues) + const asyncCache = new RedisGroupCache(redis, { + ttlInMsecs: 5000, + json: true, + ttlLeftBeforeRefreshInMsecs: 4500, + }) + + const operation = new GroupLoader({ + inMemoryCache: { + cacheId: 'two-layer-group-refresh', + cacheType: 'lru-object', + ttlInMsecs: 5000, + ttlLeftBeforeRefreshInMsecs: 4500, + }, + asyncCache, + dataSources: [loader], + }) + + // Populates both layers with the initial value + expect(await operation.get(user1.userId, user1.companyId)).toEqual(user1) + expect(loader.counter).toBe(1) + + // Source value changes + const updatedUser1: User = { ...user1, parametrized: 'updated' } + loader.groupValues![user1.companyId][user1.userId] = updatedUser1 + + // Wait until both layers' entries are within the refresh window + await setTimeout(600) + + // SWR: returns stale value and triggers a background refresh + expect(await operation.get(user1.userId, user1.companyId)).toEqual(user1) + + // Wait for the background refresh to actually call the data source + for (let attempt = 0; attempt < 20 && loader.counter < 2; attempt++) { + await setTimeout(10) + } + expect(loader.counter).toBe(2) + + // Sanity check: Redis is now fresh + // @ts-expect-error + expect(await operation.asyncCache.getFromGroup(user1.userId, user1.companyId)).toEqual( + updatedUser1, + ) + + // The next read should observe the fresh value that the background refresh fetched + expect(await operation.get(user1.userId, user1.companyId)).toEqual(updatedUser1) + }) + it('async background refresh errors do not crash app', async () => { const loader = new CountingGroupedLoader(userValues) const asyncCache = new RedisGroupCache(redis, { @@ -275,5 +326,27 @@ describe('GroupLoader Async Refresh', () => { await Promise.resolve() expect(loader.counter).toBe(3) }) + + it('does not crash when async expiration lookup rejects', async () => { + const asyncCache = new DummyGroupedCache(userValues) + // Force the fire-and-forget expiration lookup to reject + ;(asyncCache as any).ttlLeftBeforeRefreshInMsecs = 999999 + ;(asyncCache as any).expirationTimeLoadingGroupedOperation = { + get: () => Promise.reject(new Error('expiration lookup failed')), + } + + const errors: unknown[] = [] + const operation = new GroupLoader({ + asyncCache, + logger: { error: (msg) => errors.push(msg) }, + }) + + expect(await operation.get(user1.userId, user1.companyId)).toEqual(user1) + + await waitAndRetry(() => errors.length > 0, 5, 20) + expect(errors).toContain('expiration lookup failed') + + await asyncCache.close() + }) }) }) diff --git a/test/GroupLoader-main.spec.ts b/test/GroupLoader-main.spec.ts index 7b678c3..98b2afe 100644 --- a/test/GroupLoader-main.spec.ts +++ b/test/GroupLoader-main.spec.ts @@ -4,6 +4,7 @@ import type { CacheKeyResolver } from '../lib/AbstractCache' import { GroupLoader } from '../lib/GroupLoader' import type { InMemoryGroupCacheConfiguration } from '../lib/memory/InMemoryGroupCache' import { CountingGroupedLoader } from './fakes/CountingGroupedLoader' +import { DelayedCountingGroupedLoader } from './fakes/DelayedCountingGroupedLoader' import type { DummyLoaderManyParams, DummyLoaderParams } from './fakes/DummyDataSourceWithParams' import { DummyGroupedCache } from './fakes/DummyGroupedCache' import { DummyGroupedDataSourceWithParams } from './fakes/DummyGroupedDataSourceWithParams' @@ -11,6 +12,7 @@ import { DummyGroupedLoader } from './fakes/DummyGroupedLoader' import { DummyGroupNotificationConsumer } from './fakes/DummyGroupNotificationConsumer' import { DummyGroupNotificationConsumerMultiplexer } from './fakes/DummyGroupNotificationConsumerMultiplexer' import { DummyGroupNotificationPublisher } from './fakes/DummyGroupNotificationPublisher' +import { MultiDelayedCountingGroupedLoader } from './fakes/MultiDelayedCountingGroupedLoader' import { TemporaryThrowingGroupedLoader } from './fakes/TemporaryThrowingGroupedLoader' import { ThrowingGroupedCache } from './fakes/ThrowingGroupedCache' import { ThrowingGroupedLoader } from './fakes/ThrowingGroupedLoader' @@ -919,6 +921,86 @@ describe('GroupLoader Main', () => { }) }) + describe('invalidation racing with in-flight loads', () => { + it('does not cache a load that was in flight when invalidateCacheForGroup was called', async () => { + const loader = new DelayedCountingGroupedLoader(userValues) + + const operation = new GroupLoader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + dataSources: [loader], + }) + + const loadPromise = operation.get(user1.userId, user1.companyId) + await setTimeout(2) + expect(loader.counter).toBe(1) + + await operation.invalidateCacheForGroup(user1.companyId) + await loader.finishLoading() + + // the caller that was already awaiting the load still receives the value + expect(await loadPromise).toEqual(user1) + // but the invalidated group is not resurrected in the cache + expect(operation.getInMemoryOnly(user1.userId, user1.companyId)).toBeUndefined() + }) + + it('does not cache a load that was in flight when invalidateCacheFor was called', async () => { + const loader = new DelayedCountingGroupedLoader(userValues) + + const operation = new GroupLoader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + dataSources: [loader], + }) + + const loadPromise = operation.get(user1.userId, user1.companyId) + await setTimeout(2) + expect(loader.counter).toBe(1) + + await operation.invalidateCacheFor(user1.userId, user1.companyId) + await loader.finishLoading() + + expect(await loadPromise).toEqual(user1) + expect(operation.getInMemoryOnly(user1.userId, user1.companyId)).toBeUndefined() + }) + + it('load settling after group invalidation does not break deduplication of newer loads', async () => { + const loader = new MultiDelayedCountingGroupedLoader(userValues) + + const operation = new GroupLoader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + dataSources: [loader], + }) + + // load A starts and registers the running-loads map for the group + const loadPromiseA = operation.get(user1.userId, user1.companyId) + await setTimeout(2) + expect(loader.counter).toBe(1) + + // invalidation detaches the running-loads map for the group + await operation.invalidateCacheForGroup(user1.companyId) + + // load B starts after the invalidation and registers a fresh running-loads map + const loadPromiseB = operation.get(user1.userId, user1.companyId) + await setTimeout(2) + expect(loader.counter).toBe(2) + + // load A settles against the detached map - it must not evict B's map + await loader.finishLoading() + expect(await loadPromiseA).toEqual(user1) + // A started before the invalidation, so its result must not be cached + expect(operation.getInMemoryOnly(user1.userId, user1.companyId)).toBeUndefined() + + // load C must be deduplicated into the still-running load B + const loadPromiseC = operation.getAsyncOnly(user1.userId, user1.companyId) + await setTimeout(2) + expect(loader.counter).toBe(2) + + await loader.finishLoading() + expect(await loadPromiseB).toEqual(user1) + expect(await loadPromiseC).toEqual(user1) + expect(operation.getInMemoryOnly(user1.userId, user1.companyId)).toEqual(user1) + }) + }) + describe('invalidateCache', () => { it('correctly invalidates cache', async () => { const cache2 = new DummyGroupedCache(userValuesUndefined) diff --git a/test/Loader-async-refresh.spec.ts b/test/Loader-async-refresh.spec.ts index d333baf..78d76a6 100644 --- a/test/Loader-async-refresh.spec.ts +++ b/test/Loader-async-refresh.spec.ts @@ -5,7 +5,9 @@ import { Loader } from '../lib/Loader' import { RedisCache } from '../lib/redis' import { CountingDataSource } from './fakes/CountingDataSource' import { DelayedCountingLoader } from './fakes/DelayedCountingLoader' +import { DummyCache } from './fakes/DummyCache' import { redisOptions } from './fakes/TestRedisConfig' +import { waitAndRetry } from './utils/waitUtils' describe('Loader Async', () => { let redis: Redis @@ -185,5 +187,26 @@ describe('Loader Async', () => { await Promise.resolve() expect(loader.counter).toBe(3) }) + + it('does not crash when async expiration lookup rejects', async () => { + const asyncCache = new DummyCache('value') + // Force the fire-and-forget expiration lookup to reject + ;(asyncCache as any).expirationTimeLoadingOperation = { + get: () => Promise.reject(new Error('expiration lookup failed')), + } + + const errors: unknown[] = [] + const operation = new Loader({ + asyncCache, + logger: { error: (msg) => errors.push(msg) }, + }) + + expect(await operation.get('key')).toBe('value') + + await waitAndRetry(() => errors.length > 0, 5, 20) + expect(errors).toContain('expiration lookup failed') + + await asyncCache.close() + }) }) }) diff --git a/test/Loader-main.spec.ts b/test/Loader-main.spec.ts index 4b0fb28..4fa6639 100644 --- a/test/Loader-main.spec.ts +++ b/test/Loader-main.spec.ts @@ -8,6 +8,8 @@ import type { InMemoryCacheConfiguration } from '../lib/memory/InMemoryCache' import { CountingDataSource } from './fakes/CountingDataSource' import { CountingRecordLoader } from './fakes/CountingRecordLoader' import { CountingTimedCache } from './fakes/CountingTimedCache' +import { DelayedCountingLoader } from './fakes/DelayedCountingLoader' +import { DelayedDeleteCache } from './fakes/DelayedDeleteCache' import { DummyCache } from './fakes/DummyCache' import { DummyDataSource } from './fakes/DummyDataSource' import { @@ -1102,6 +1104,103 @@ describe('Loader Main', () => { }) }) + describe('invalidation racing with in-flight loads', () => { + it('does not cache a load that was in flight when invalidateCacheFor was called', async () => { + const loader = new DelayedCountingLoader('value') + + const operation = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + dataSources: [loader], + }) + + const loadPromise = operation.get('key') + await setTimeout(2) + expect(loader.counter).toBe(1) + + await operation.invalidateCacheFor('key') + await loader.finishLoading() + + // the caller that was already awaiting the load still receives the value + expect(await loadPromise).toBe('value') + // but the invalidated key is not resurrected in the cache + expect(operation.getInMemoryOnly('key')).toBeUndefined() + + // the next get starts a fresh load instead of serving the fenced-out result + const freshPromise = operation.get('key') + await setTimeout(2) + expect(loader.counter).toBe(2) + await loader.finishLoading() + expect(await freshPromise).toBe('value') + }) + + it('does not cache a load that was in flight when invalidateCacheForMany was called', async () => { + const loader = new DelayedCountingLoader('value') + + const operation = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + dataSources: [loader], + }) + + const loadPromise = operation.get('key') + await setTimeout(2) + expect(loader.counter).toBe(1) + + await operation.invalidateCacheForMany(['key', 'key2']) + await loader.finishLoading() + + expect(await loadPromise).toBe('value') + expect(operation.getInMemoryOnly('key')).toBeUndefined() + }) + + it('does not let an in-flight load overwrite forceSetValue', async () => { + const loader = new DelayedCountingLoader('value') + + const operation = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + dataSources: [loader], + }) + + const loadPromise = operation.get('key') + await setTimeout(2) + expect(loader.counter).toBe(1) + + await operation.forceSetValue('key', 'value2') + await loader.finishLoading() + + // the caller that was already awaiting the load still receives the loaded value + expect(await loadPromise).toBe('value') + // but the forced value is not clobbered by the older in-flight load + expect(operation.getInMemoryOnly('key')).toBe('value2') + expect(await operation.get('key')).toBe('value2') + expect(loader.counter).toBe(1) + }) + + it('does not repopulate the in-memory cache from a get that raced a slow async delete', async () => { + const asyncCache = new DelayedDeleteCache('value') + const loader = new CountingDataSource('value') + + const operation = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache, + dataSources: [loader], + }) + + // populate both tiers + expect(await operation.get('key')).toBe('value') + + const invalidationPromise = operation.invalidateCacheFor('key') + // concurrent read while the async delete is still in flight + const racingGetPromise = operation.get('key') + asyncCache.finishDelete() + await invalidationPromise + await racingGetPromise + + // after the invalidation resolves, neither tier holds the old value + expect(operation.getInMemoryOnly('key')).toBeUndefined() + expect(asyncCache.value).toBeUndefined() + }) + }) + describe('invalidateCacheForMany', () => { it('invalidates multiple entries', async () => { const cache2 = new DummyRecordCache({}) diff --git a/test/fakes/DelayedDeleteCache.ts b/test/fakes/DelayedDeleteCache.ts new file mode 100644 index 0000000..49536a3 --- /dev/null +++ b/test/fakes/DelayedDeleteCache.ts @@ -0,0 +1,70 @@ +import { Loader } from '../../lib/Loader' +import type { Cache } from '../../lib/types/DataSources' +import type { GetManyResult } from '../../lib/types/SyncDataSources' + +export class DelayedDeleteCache implements Cache { + value: string | undefined | null + name = 'Delayed delete cache' + isCache = true + readonly expirationTimeLoadingOperation = new Loader({}) + readonly ttlLeftBeforeRefreshInMsecs = undefined + private deleteResolver?: () => void + + constructor(returnedValue: string | undefined) { + this.value = returnedValue + } + + get(_key: string): Promise { + return Promise.resolve(this.value) + } + + getMany(keys: string[]): Promise> { + const resolvedValues = keys.map(() => this.value as string).filter((entry) => entry != null) + + return Promise.resolve({ + resolvedValues, + unresolvedKeys: resolvedValues.length === 0 ? keys : [], + }) + } + + clear(): Promise { + this.value = undefined + return Promise.resolve(undefined) + } + + delete(_key: string): Promise { + return new Promise((resolve) => { + this.deleteResolver = () => { + this.value = undefined + resolve(undefined) + } + }) + } + + finishDelete() { + this.deleteResolver?.() + this.deleteResolver = undefined + } + + deleteMany(_keys: string[]): Promise { + this.value = undefined + return Promise.resolve(undefined) + } + + set(_key: string, value: string | null): Promise { + this.value = value ?? undefined + return Promise.resolve(undefined) + } + + getExpirationTime(): Promise { + return Promise.resolve(99999) + } + + close(): Promise { + return Promise.resolve(undefined) + } + + setMany(): Promise { + throw new Error('Not implemented') + } +} diff --git a/test/fakes/MultiDelayedCountingGroupedLoader.ts b/test/fakes/MultiDelayedCountingGroupedLoader.ts new file mode 100644 index 0000000..de803dc --- /dev/null +++ b/test/fakes/MultiDelayedCountingGroupedLoader.ts @@ -0,0 +1,43 @@ +import type { GroupDataSource } from '../../lib/types/DataSources' +import type { GroupValues, User } from '../types/testTypes' +import { cloneDeep } from '../utils/cloneUtils' + +/** + * Unlike DelayedCountingGroupedLoader, this fake supports several concurrent pending loads: + * each getFromGroup call queues its own resolver and finishLoading() settles them in FIFO order. + */ +export class MultiDelayedCountingGroupedLoader implements GroupDataSource { + public counter = 0 + public groupValues: GroupValues | null | undefined + + name = 'Multi delayed counting grouped loader' + private pendingLoads: { resolve: (value: any) => void; value: any; promise: Promise }[] = [] + + constructor(returnedValues: GroupValues) { + this.groupValues = cloneDeep(returnedValues) + } + + getFromGroup(key: string, group: string) { + this.counter++ + const value = this.groupValues?.[group]?.[key] + let resolve!: (value: any) => void + const promise = new Promise((_resolve) => { + resolve = _resolve + }) + this.pendingLoads.push({ resolve, value, promise }) + return promise + } + + getManyFromGroup(): Promise { + throw new Error('Method not implemented.') + } + + finishLoading() { + const pendingLoad = this.pendingLoads.shift() + if (!pendingLoad) { + throw new Error('No pending load to finish') + } + pendingLoad.resolve(pendingLoad.value) + return pendingLoad.promise + } +} diff --git a/test/memory/InMemoryCache.spec.ts b/test/memory/InMemoryCache.spec.ts index 3a931d5..7af115f 100644 --- a/test/memory/InMemoryCache.spec.ts +++ b/test/memory/InMemoryCache.spec.ts @@ -147,6 +147,25 @@ describe('InMemoryCache', () => { resolvedValues: ['value', 'value2'], }) }) + + it('resolves falsy and null cached values', () => { + const cache = new InMemoryCache({ + cacheId: 'dummy', + cacheType: 'lru-map', + maxItems: 5, + ttlInMsecs: 9999, + }) + cache.set('zero', 0) + cache.set('emptyString', '') + cache.set('false', false) + cache.set('null', null) + + const values = cache.getMany(['zero', 'emptyString', 'false', 'null', 'missing']) + expect(values).toEqual({ + unresolvedKeys: ['missing'], + resolvedValues: [0, '', false, null], + }) + }) }) describe('getExpirationTime', () => { diff --git a/test/memory/InMemoryGroupCache.spec.ts b/test/memory/InMemoryGroupCache.spec.ts index b4f0ac1..fa19d07 100644 --- a/test/memory/InMemoryGroupCache.spec.ts +++ b/test/memory/InMemoryGroupCache.spec.ts @@ -130,6 +130,62 @@ describe('InMemoryCache', () => { resolvedValues: ['value', 'value2'], }) }) + + it('resolves falsy and null cached values', () => { + const cache = new InMemoryGroupCache({ + maxGroups: 2, + ttlInMsecs: 9999, + }) + cache.setForGroup('zero', 0, 'group') + cache.setForGroup('emptyString', '', 'group') + cache.setForGroup('false', false, 'group') + cache.setForGroup('null', null, 'group') + + const values = cache.getManyFromGroup( + ['zero', 'emptyString', 'false', 'null', 'missing'], + 'group', + ) + expect(values).toEqual({ + unresolvedKeys: ['missing'], + resolvedValues: [0, '', false, null], + }) + }) + + it('returns all keys as unresolved for a nonexistent group', () => { + const cache = new InMemoryGroupCache({ + maxGroups: 2, + ttlInMsecs: 9999, + }) + + const values = cache.getManyFromGroup(['key', 'key2'], 'no-such-group') + expect(values).toEqual({ + unresolvedKeys: ['key', 'key2'], + resolvedValues: [], + }) + }) + }) + + describe('read operations on nonexistent groups', () => { + it('do not evict existing groups from the groups cache', () => { + const cache = new InMemoryGroupCache({ + maxGroups: 2, + ttlInMsecs: 9999, + }) + cache.setForGroup('key', 'value1', 'group1') + cache.setForGroup('key', 'value2', 'group2') + + // each of these reads used to insert an empty group, evicting a populated one + expect(cache.getFromGroup('key', 'group3')).toBeUndefined() + expect(cache.getManyFromGroup(['key'], 'group4')).toEqual({ + resolvedValues: [], + unresolvedKeys: ['key'], + }) + expect(cache.getExpirationTimeFromGroup('key', 'group5')).toBeUndefined() + cache.deleteFromGroup('key', 'group6') + + expect(cache.getFromGroup('key', 'group1')).toBe('value1') + expect(cache.getFromGroup('key', 'group2')).toBe('value2') + }) }) describe('getExpirationTimeFromGroup', () => { diff --git a/test/redis/RedisGroupCache.spec.ts b/test/redis/RedisGroupCache.spec.ts index f763cf9..3c44a38 100644 --- a/test/redis/RedisGroupCache.spec.ts +++ b/test/redis/RedisGroupCache.spec.ts @@ -475,6 +475,32 @@ describe('RedisGroupCache', () => { expect(value).toBeUndefined() }) + it('expires the group index with groupTtlInMsecs, not the entry ttl', async () => { + const cache = new RedisGroupCache(redis, { + ttlInMsecs: 500, + groupTtlInMsecs: 60000, + }) + await cache.setForGroup('key', 'value', 'group') + + await cache.deleteGroup('group') + + const indexTtl = await redis.pttl(cache.resolveGroupIndexPrefix('group')) + // the group index counter must live as long as groupTtlInMsecs dictates; + // scoping it to the entry ttl would let the counter expire and reset early + expect(indexTtl).toBeGreaterThan(500) + expect(indexTtl).toBeLessThanOrEqual(60000) + }) + + it('does not expire the group index when groupTtlInMsecs is not set', async () => { + const cache = new RedisGroupCache(redis, { ttlInMsecs: 500 }) + await cache.setForGroup('key', 'value', 'group') + + await cache.deleteGroup('group') + + const indexTtl = await redis.pttl(cache.resolveGroupIndexPrefix('group')) + expect(indexTtl).toBe(-1) // no TTL, consistent with setForGroup + }) + it('deletes values matching the group pattern', async () => { const cache = new RedisGroupCache(redis) diff --git a/test/redis/RedisGroupNotificationPublisher.spec.ts b/test/redis/RedisGroupNotificationPublisher.spec.ts index 70750c0..7685423 100644 --- a/test/redis/RedisGroupNotificationPublisher.spec.ts +++ b/test/redis/RedisGroupNotificationPublisher.spec.ts @@ -407,6 +407,101 @@ describe('RedisGroupNotificationPublisher', () => { await notificationPublisher1.close() }) + it('Survives malformed messages', async () => { + const { publisher: notificationPublisher1, consumer: notificationConsumer1 } = + createGroupNotificationPair({ + channel: CHANNEL_ID, + consumerRedis: redisConsumer, + publisherRedis: redisPublisher, + }) + + const operation = new GroupLoader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache: new DummyGroupedCache(userValues), + notificationConsumer: notificationConsumer1, + notificationPublisher: notificationPublisher1, + }) + await operation.init() + + await operation.getAsyncOnly('key', 'group') + expect(operation.getInMemoryOnly('key', 'group')).toEqual(user1) + + // A non-JSON payload must not crash the consumer + await redisPublisher.publish(CHANNEL_ID, 'this is not json') + + // A valid message afterwards must still be processed + await redisPublisher.publish( + CHANNEL_ID, + JSON.stringify({ + actionId: 'DELETE_GROUP', + group: 'group', + originUuid: 'different-uuid', + }), + ) + + await waitAndRetry(() => operation.getInMemoryOnly('key', 'group') === undefined, 50, 100) + expect(operation.getInMemoryOnly('key', 'group')).toBeUndefined() + + await notificationConsumer1.close() + await notificationPublisher1.close() + }) + + it('Ignores messages from other channels on a shared consumer connection', async () => { + const { publisher: notificationPublisherA, consumer: notificationConsumerA } = + createGroupNotificationPair({ + channel: 'channel_a', + consumerRedis: redisConsumer, + publisherRedis: redisPublisher, + }) + + const { publisher: notificationPublisherB, consumer: notificationConsumerB } = + createGroupNotificationPair({ + channel: 'channel_b', + consumerRedis: redisConsumer, + publisherRedis: redisPublisher, + }) + + const operationA = new GroupLoader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache: new DummyGroupedCache(userValues), + notificationConsumer: notificationConsumerA, + notificationPublisher: notificationPublisherA, + }) + const operationB = new GroupLoader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache: new DummyGroupedCache(userValues), + notificationConsumer: notificationConsumerB, + notificationPublisher: notificationPublisherB, + }) + await operationA.init() + await operationB.init() + + await operationA.getAsyncOnly('key', 'group') + await operationB.getAsyncOnly('key', 'group') + expect(operationA.getInMemoryOnly('key', 'group')).toEqual(user1) + expect(operationB.getInMemoryOnly('key', 'group')).toEqual(user1) + + // Publish a DELETE_GROUP on channel_a only, from a foreign origin + await redisPublisher.publish( + 'channel_a', + JSON.stringify({ + actionId: 'DELETE_GROUP', + group: 'group', + originUuid: 'different-uuid', + }), + ) + + // consumer A must apply it... + await waitAndRetry(() => operationA.getInMemoryOnly('key', 'group') === undefined, 50, 100) + expect(operationA.getInMemoryOnly('key', 'group')).toBeUndefined() + + // ...but consumer B, sharing the same Redis connection on another channel, must not + expect(operationB.getInMemoryOnly('key', 'group')).toEqual(user1) + + await notificationConsumerA.close() + await notificationPublisherA.close() + }) + it('Removes message listeners on close', async () => { const { publisher: notificationPublisher1, consumer: notificationConsumer1 } = createGroupNotificationPair({ diff --git a/test/redis/RedisNotificationPublisher.spec.ts b/test/redis/RedisNotificationPublisher.spec.ts index 3aacf8c..794f286 100644 --- a/test/redis/RedisNotificationPublisher.spec.ts +++ b/test/redis/RedisNotificationPublisher.spec.ts @@ -407,6 +407,101 @@ describe('RedisNotificationPublisher', () => { await setTimeout(1) }) + it('Ignores messages from other channels on a shared consumer connection', async () => { + const { publisher: notificationPublisherA, consumer: notificationConsumerA } = + createNotificationPair({ + channel: 'channel_a', + consumerRedis: redisConsumer, + publisherRedis: redisPublisher, + }) + + const { publisher: notificationPublisherB, consumer: notificationConsumerB } = + createNotificationPair({ + channel: 'channel_b', + consumerRedis: redisConsumer, + publisherRedis: redisPublisher, + }) + + const operationA = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache: new DummyCache('value'), + notificationConsumer: notificationConsumerA, + notificationPublisher: notificationPublisherA, + }) + const operationB = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache: new DummyCache('value'), + notificationConsumer: notificationConsumerB, + notificationPublisher: notificationPublisherB, + }) + await operationA.init() + await operationB.init() + + await operationA.getAsyncOnly('key') + await operationB.getAsyncOnly('key') + expect(operationA.getInMemoryOnly('key')).toBe('value') + expect(operationB.getInMemoryOnly('key')).toBe('value') + + // Publish a DELETE on channel_a only, from a foreign origin + await redisPublisher.publish( + 'channel_a', + JSON.stringify({ + actionId: 'DELETE', + key: 'key', + originUuid: 'different-uuid', + }), + ) + + // consumer A must apply it... + await waitAndRetry(() => operationA.getInMemoryOnly('key') === undefined, 50, 100) + expect(operationA.getInMemoryOnly('key')).toBeUndefined() + + // ...but consumer B, sharing the same Redis connection on another channel, must not + expect(operationB.getInMemoryOnly('key')).toBe('value') + + await notificationConsumerA.close() + await notificationPublisherA.close() + }) + + it('Survives malformed messages', async () => { + const { publisher: notificationPublisher1, consumer: notificationConsumer1 } = + createNotificationPair({ + channel: CHANNEL_ID, + consumerRedis: redisConsumer, + publisherRedis: redisPublisher, + }) + + const operation = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache: new DummyCache('value'), + notificationConsumer: notificationConsumer1, + notificationPublisher: notificationPublisher1, + }) + await operation.init() + + const resultPre = await operation.get('key') + expect(resultPre).toBe('value') + + // A non-JSON payload must not crash the consumer + await redisPublisher.publish(CHANNEL_ID, 'this is not json') + + // A valid message afterwards must still be processed + await redisPublisher.publish( + CHANNEL_ID, + JSON.stringify({ + actionId: 'DELETE', + key: 'key', + originUuid: 'different-uuid', + }), + ) + + await waitAndRetry(() => operation.getInMemoryOnly('key') === undefined, 50, 100) + expect(operation.getInMemoryOnly('key')).toBeUndefined() + + await notificationConsumer1.close() + await notificationPublisher1.close() + }) + it('Ignores unknown action IDs', async () => { const { publisher: notificationPublisher1, consumer: notificationConsumer1 } = createNotificationPair({