From 3ec695b2443b75c103057b424ba50fad51657421 Mon Sep 17 00:00:00 2001 From: Artur Wolny Date: Mon, 28 Aug 2023 14:30:54 +0200 Subject: [PATCH] feature(entitlements): CR fixes --- .../entitlements/entitlements-client.ts | 4 +- .../storage/cache.revision-manager.spec.ts | 4 +- .../storage/cache.revision-manager.ts | 2 +- .../storage/dto-to-cache-sources.mapper.ts | 60 +++++++++---------- .../frontegg.cache-initializer.ts | 4 +- .../leader-election/ioredis.lock-handler.ts | 34 +++++++++-- .../redis-based.lock-handlers.spec.ts | 5 +- .../leader-election/redis.lock-handler.ts | 36 +++++++++-- src/components/leader-election/types.ts | 25 +++++++- 9 files changed, 124 insertions(+), 50 deletions(-) diff --git a/src/clients/entitlements/entitlements-client.ts b/src/clients/entitlements/entitlements-client.ts index 343b2bf..e8179ec 100644 --- a/src/clients/entitlements/entitlements-client.ts +++ b/src/clients/entitlements/entitlements-client.ts @@ -126,7 +126,7 @@ export class EntitlementsClient extends TypedEmitter const entitlementsData = await this.httpClient.get('/api/v1/vendor-entitlements'); const vendorEntitlementsDto = entitlementsData.data; - const { isUpdated, revision } = await this.cacheManager.loadSnapshotAsCurrent(vendorEntitlementsDto); + const { isUpdated, revision } = await this.cacheManager.loadSnapshotAsCurrentRevision(vendorEntitlementsDto); if (isUpdated) { this.emit(EntitlementsClientEventsEnum.SNAPSHOT_UPDATED, revision); @@ -186,4 +186,4 @@ export class EntitlementsClient extends TypedEmitter destroy(): void { this.refreshTimeout && clearTimeout(this.refreshTimeout); } -} \ No newline at end of file +} diff --git a/src/clients/entitlements/storage/cache.revision-manager.spec.ts b/src/clients/entitlements/storage/cache.revision-manager.spec.ts index 9d4fc54..02e782b 100644 --- a/src/clients/entitlements/storage/cache.revision-manager.spec.ts +++ b/src/clients/entitlements/storage/cache.revision-manager.spec.ts @@ -68,7 +68,7 @@ describe(CacheRevisionManager.name, () => { jest.mocked(FronteggEntitlementsCacheInitializer.forLeader).mockResolvedValue(expectedNewEntitlementsCache); // when - loadingSnapshotResult = await cut.loadSnapshotAsCurrent(getDTO(333)); + loadingSnapshotResult = await cut.loadSnapshotAsCurrentRevision(getDTO(333)); }); it('then it resolves to IsUpdatedToRev structure telling with updated revision.', async () => { @@ -97,7 +97,7 @@ describe(CacheRevisionManager.name, () => { jest.mocked(FronteggEntitlementsCacheInitializer.forFollower).mockClear(); // when - loadingSnapshotResult = await cut.loadSnapshotAsCurrent(getDTO(1)); + loadingSnapshotResult = await cut.loadSnapshotAsCurrentRevision(getDTO(1)); }); it('then it resolves to IsUpdatedToRev structure telling nothing got updated and revision (1).', async () => { diff --git a/src/clients/entitlements/storage/cache.revision-manager.ts b/src/clients/entitlements/storage/cache.revision-manager.ts index a357745..2a7d0ab 100644 --- a/src/clients/entitlements/storage/cache.revision-manager.ts +++ b/src/clients/entitlements/storage/cache.revision-manager.ts @@ -15,7 +15,7 @@ export class CacheRevisionManager { constructor(private readonly cache: ICacheManager) {} - async loadSnapshotAsCurrent(dto: VendorEntitlementsDto): Promise { + async loadSnapshotAsCurrentRevision(dto: VendorEntitlementsDto): Promise { const currentRevision = await this.getCurrentCacheRevision(); const givenRevision = dto.snapshotOffset; diff --git a/src/clients/entitlements/storage/dto-to-cache-sources.mapper.ts b/src/clients/entitlements/storage/dto-to-cache-sources.mapper.ts index d5ab0ee..bb2c5d7 100644 --- a/src/clients/entitlements/storage/dto-to-cache-sources.mapper.ts +++ b/src/clients/entitlements/storage/dto-to-cache-sources.mapper.ts @@ -1,8 +1,32 @@ import { FeatureId, VendorEntitlementsDto } from '../types'; import { BundlesSource, ExpirationTime, FeatureSource, NO_EXPIRE, UNBUNDLED_SRC_ID } from './types'; +function ensureMapInMap>(map: Map, mapKey: K): T { + if (!map.has(mapKey)) { + map.set(mapKey, new Map() as T); + } + + return map.get(mapKey)!; +} + +function ensureArrayInMap(map: Map, mapKey: K): T[] { + if (!map.has(mapKey)) { + map.set(mapKey, []); + } + + return map.get(mapKey)!; +} + +function parseExpirationTime(time?: string | null): ExpirationTime { + if (time !== undefined && time !== null) { + return new Date(time).getTime(); + } + + return NO_EXPIRE; +} + export class DtoToCacheSourcesMapper { - map(dto: VendorEntitlementsDto): BundlesSource { + static map(dto: VendorEntitlementsDto): BundlesSource { const { data: { features, entitlements, featureBundles }, } = dto; @@ -56,15 +80,15 @@ export class DtoToCacheSourcesMapper { if (bundle) { if (userId) { // that's user-targeted entitlement - const tenantUserEntitlements = this.ensureMapInMap(bundle.user_entitlements, tenantId); - const usersEntitlements = this.ensureArrayInMap(tenantUserEntitlements, userId); + const tenantUserEntitlements = ensureMapInMap(bundle.user_entitlements, tenantId); + const usersEntitlements = ensureArrayInMap(tenantUserEntitlements, userId); - usersEntitlements.push(this.parseExpirationTime(expirationDate)); + usersEntitlements.push(parseExpirationTime(expirationDate)); } else { // that's tenant-targeted entitlement - const tenantEntitlements = this.ensureArrayInMap(bundle.tenant_entitlements, tenantId); + const tenantEntitlements = ensureArrayInMap(bundle.tenant_entitlements, tenantId); - tenantEntitlements.push(this.parseExpirationTime(expirationDate)); + tenantEntitlements.push(parseExpirationTime(expirationDate)); } } else { // TODO: issue warning here! @@ -87,28 +111,4 @@ export class DtoToCacheSourcesMapper { return bundlesMap; } - - private ensureMapInMap>(map: Map, mapKey: K): T { - if (!map.has(mapKey)) { - map.set(mapKey, new Map() as T); - } - - return map.get(mapKey)!; - } - - private ensureArrayInMap(map: Map, mapKey: K): T[] { - if (!map.has(mapKey)) { - map.set(mapKey, []); - } - - return map.get(mapKey)!; - } - - private parseExpirationTime(time?: string | null): ExpirationTime { - if (time !== undefined && time !== null) { - return new Date(time).getTime(); - } - - return NO_EXPIRE; - } } diff --git a/src/clients/entitlements/storage/frontegg-cache/frontegg.cache-initializer.ts b/src/clients/entitlements/storage/frontegg-cache/frontegg.cache-initializer.ts index 16410eb..1e128dd 100644 --- a/src/clients/entitlements/storage/frontegg-cache/frontegg.cache-initializer.ts +++ b/src/clients/entitlements/storage/frontegg-cache/frontegg.cache-initializer.ts @@ -25,7 +25,7 @@ export class FronteggEntitlementsCacheInitializer { const cacheInitializer = new FronteggEntitlementsCacheInitializer(entitlementsCache); - const sources = new DtoToCacheSourcesMapper().map(dto); + const sources = DtoToCacheSourcesMapper.map(dto); await cacheInitializer.setupPermissionsReadModel(sources); await cacheInitializer.setupEntitlementsReadModel(sources); @@ -96,7 +96,7 @@ export class FronteggEntitlementsCacheInitializer { const allPermissions = await cache.collection(PERMISSIONS_COLLECTION_LIST).getAll(); for (const permission of allPermissions) { - await cache.expire([ getPermissionMappingKey(permission)], FronteggEntitlementsCacheInitializer.CLEAR_TTL); + await cache.expire([getPermissionMappingKey(permission)], FronteggEntitlementsCacheInitializer.CLEAR_TTL); } // clear static fields diff --git a/src/components/leader-election/ioredis.lock-handler.ts b/src/components/leader-election/ioredis.lock-handler.ts index 1b34e4d..f4d1b26 100644 --- a/src/components/leader-election/ioredis.lock-handler.ts +++ b/src/components/leader-election/ioredis.lock-handler.ts @@ -9,19 +9,43 @@ export class IORedisLockHandler implements ILockHandler { private static EXTEND_LEADERSHIP_SCRIPT = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end"; - async tryToMaintainTheLock(key: string, value: string, expirationTimeMs: number): Promise { + /** + * This method calls the Lua script that prolongs the lock on given `leadershipResourceKey` only, when stored value + * equals to given `instanceIdentifier` and then method resolves to `true`. + * + * When `leadershipResourceKey` doesn't exist, or it has a different value, then the leadership is not prolonged and + * method resolves to `false`. + * + * Using Lua script ensures the atomicity of the whole process. Without it there is no guarantee that other Redis + * client doesn't execute operation on `leadershipResourceKey` in-between `GET` and `PEXPIRE` commands. + */ + async tryToMaintainTheLock( + leadershipResourceKey: string, + instanceIdentifier: string, + expirationTimeMs: number, + ): Promise { const extended = await this.redis.eval( IORedisLockHandler.EXTEND_LEADERSHIP_SCRIPT, NUM_OF_KEYS_IN_LUA_SCRIPT, - key, - value, + leadershipResourceKey, + instanceIdentifier, expirationTimeMs, ); return (extended as number) > 0; } - async tryToLockLeaderResource(key: string, value: string, expirationTimeMs: number): Promise { - return (await this.redis.set(key, value, 'PX', expirationTimeMs, 'NX')) !== null; + /** + * This stores the `instanceIdentifier` value into `leadershipResourceKey` only, when the key doesn't exist. If value + * is stored, then TTL is also set to `expirationTimeMs` and method resolves to `true`. + * + * Otherwise method resolved to `false` and no change to `leadershipResourceKey` is introduced. + */ + async tryToLockLeaderResource( + leadershipResourceKey: string, + instanceIdentifier: string, + expirationTimeMs: number, + ): Promise { + return (await this.redis.set(leadershipResourceKey, instanceIdentifier, 'PX', expirationTimeMs, 'NX')) !== null; } } diff --git a/src/components/leader-election/redis-based.lock-handlers.spec.ts b/src/components/leader-election/redis-based.lock-handlers.spec.ts index 6dd649c..23184e7 100644 --- a/src/components/leader-election/redis-based.lock-handlers.spec.ts +++ b/src/components/leader-election/redis-based.lock-handlers.spec.ts @@ -61,7 +61,10 @@ describe.each([ await expect(testRedisConnection.get(RESOURCE_KEY)).resolves.toEqual('bar'); // and: key is about to expire - await expect(testRedisConnection.pttl(RESOURCE_KEY)).resolves.toBeWithin(0, 1000); + const pttl = await testRedisConnection.pttl(RESOURCE_KEY); + + expect(pttl).toBeGreaterThan(0); + expect(pttl).toBeLessThanOrEqual(1000); }); it('when .tryToMaintainTheLock(..) is called, then it resolves to FALSE and no value is written to resource key.', async () => { diff --git a/src/components/leader-election/redis.lock-handler.ts b/src/components/leader-election/redis.lock-handler.ts index 20b3e35..879b734 100644 --- a/src/components/leader-election/redis.lock-handler.ts +++ b/src/components/leader-election/redis.lock-handler.ts @@ -7,16 +7,42 @@ export class RedisLockHandler implements ILockHandler { private static EXTEND_LEADERSHIP_SCRIPT = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end"; - async tryToMaintainTheLock(key: string, value: string, expirationTimeMs: number): Promise { + /** + * This method calls the Lua script that prolongs the lock on given `leadershipResourceKey` only, when stored value + * equals to given `instanceIdentifier` and then method resolves to `true`. + * + * When `leadershipResourceKey` doesn't exist, or it has a different value, then the leadership is not prolonged and + * method resolves to `false`. + * + * Using Lua script ensures the atomicity of the whole process. Without it there is no guarantee that other Redis + * client doesn't execute operation on `leadershipResourceKey` in-between `GET` and `PEXPIRE` commands. + */ + async tryToMaintainTheLock( + leadershipResourceKey: string, + instanceIdentifier: string, + expirationTimeMs: number, + ): Promise { const extended = await this.redis.EVAL(RedisLockHandler.EXTEND_LEADERSHIP_SCRIPT, { - keys: [key], - arguments: [value, expirationTimeMs.toString()], + keys: [leadershipResourceKey], + arguments: [instanceIdentifier, expirationTimeMs.toString()], }); return (extended as number) > 0; } - async tryToLockLeaderResource(key: string, value: string, expirationTimeMs: number): Promise { - return (await this.redis.SET(key, value, { PX: expirationTimeMs, NX: true })) !== null; + /** + * This stores the `instanceIdentifier` value into `leadershipResourceKey` only, when the key doesn't exist. If value + * is stored, then TTL is also set to `expirationTimeMs` and method resolves to `true`. + * + * Otherwise method resolved to `false` and no change to `leadershipResourceKey` is introduced. + */ + async tryToLockLeaderResource( + leadershipResourceKey: string, + instanceIdentifier: string, + expirationTimeMs: number, + ): Promise { + return ( + (await this.redis.SET(leadershipResourceKey, instanceIdentifier, { PX: expirationTimeMs, NX: true })) !== null + ); } } diff --git a/src/components/leader-election/types.ts b/src/components/leader-election/types.ts index 56ac508..6b64690 100644 --- a/src/components/leader-election/types.ts +++ b/src/components/leader-election/types.ts @@ -1,6 +1,27 @@ export interface ILockHandler { - tryToLockLeaderResource(key: string, value: string, expirationTimeMs: number): Promise; - tryToMaintainTheLock(key: string, value: string, expirationTimeMs: number): Promise; + /** + * This method is about to lock the `leadershipResourceKey` by writing its `instanceIdentifier` to it. The lock should + * not be permanent, but limited to given `expirationTimeMs`. Then the lock can be kept (extended) by calling + * `tryToMaintainTheLock` method. + */ + tryToLockLeaderResource( + leadershipResourceKey: string, + instanceIdentifier: string, + expirationTimeMs: number, + ): Promise; + + /** + * This method is about to prolong the `leadershipResourceKey` time-to-live only, when the key contains value equal to + * given `instanceIdentifier`. Each instance competing for a leadership role needs to have a unique identifier. + * + * This way we know, that only the leader process can prolong its leadership. If leader dies, for any reason, no other + * process can extend its leadership. + */ + tryToMaintainTheLock( + leadershipResourceKey: string, + instanceIdentifier: string, + expirationTimeMs: number, + ): Promise; } export interface ILeadershipElectionOptions {