feat: sliding expiration cache (#18)

common/lib/plugins/failover/failover_plugin.ts

@@ -26,7 +26,7 @@ import { PluginService } from "../../plugin_service";
 import { ConnectionPlugin } from "../../connection_plugin";
 import { HostListProviderService } from "../../host_list_provider_service";
 import { ClusterAwareReaderFailoverHandler } from "./reader_failover_handler";
-import { SubscribedMethodHelper } from "../../utils/subsribed_method_helper";
+import { SubscribedMethodHelper } from "../../utils/subscribed_method_helper";
 import { HostChangeOptions } from "../../host_change_options";
 import { ClusterAwareWriterFailoverHandler } from "./writer_failover_handler";
 

common/lib/utils/map_utils.ts

@@ -0,0 +1,111 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ 
+  Licensed under the Apache License, Version 2.0 (the "License").
+  You may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+ 
+  http://www.apache.org/licenses/LICENSE-2.0
+ 
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+export class MapUtils<K, V> {
+  protected map: Map<K, V> = new Map<K, V>();
+
+  get size(): number {
+    return this.map.size;
+  }
+
+  get keys() {
+    return this.map.keys();
+  }
+
+  get entries() {
+    return this.map.entries();
+  }
+
+  get(key: K): V | undefined {
+    return this.map.get(key);
+  }
+
+  clear() {
+    this.map.clear();
+  }
+
+  computeIfPresent(key: K, remappingFunc: (key: K, existingValue: V) => V | null): V | undefined {
+    const existingValue: V | undefined = this.map.get(key);
+    if (existingValue === undefined) {
+      return undefined;
+    }
+    const newValue: any = remappingFunc(key, existingValue);
+    if (newValue !== null) {
+      this.map.set(key, newValue);
+      return newValue;
+    } else {
+      this.map.delete(key);
+      return undefined;
+    }
+  }
+
+  computeIfAbsent(key: K, mappingFunc: (key: K) => V | null): V | undefined {
+    const value: V | undefined = this.map.get(key);
+    if (value == undefined) {
+      const newValue: V | null = mappingFunc(key);
+      if (newValue !== null) {
+        this.map.set(key, newValue);
+        return newValue;
+      }
+      return undefined;
+    }
+    return value;
+  }
+
+  putIfAbsent(key: K, newValue: V): V | undefined {
+    const existingValue: V | undefined = this.map.get(key);
+    if (existingValue === undefined) {
+      this.map.set(key, newValue);
+      return newValue;
+    }
+    return existingValue;
+  }
+
+  remove(key: K): V | undefined {
+    const value = this.map.get(key);
+    this.map.delete(key);
+    return value;
+  }
+
+  removeIf(predicate: (v: any, k: any) => V): boolean {
+    const originalSize = this.size;
+    this.map.forEach((v, k) => {
+      if (predicate(v, k)) {
+        this.remove(k);
+      }
+    });
+    return this.size < originalSize;
+  }
+
+  removeMatchingValues(removalValues: any[]): boolean {
+    const originalSize = this.size;
+    this.map.forEach((v, k) => {
+      if (removalValues.includes(v)) {
+        this.remove(k);
+      }
+    });
+    return this.size < originalSize;
+  }
+
+  applyIf(predicate: (v: any, k: any) => V, apply: (v: any, k: any) => V): void {
+    const originalSize = this.size;
+    this.map.forEach((v, k) => {
+      if (predicate(v, k)) {
+        apply(v, k);
+      }
+    });
+  }
+}

common/lib/utils/sliding_expiration_cache.ts

@@ -0,0 +1,122 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ 
+  Licensed under the Apache License, Version 2.0 (the "License").
+  You may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+ 
+  http://www.apache.org/licenses/LICENSE-2.0
+ 
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+import { MapUtils } from "./map_utils";
+import { getTimeInNanos } from "aws-wrapper-common-lib/lib/utils/utils";
+
+class CacheItem<V> {
+  private _item: V;
+  private _expirationTimeNanos: bigint;
+
+  constructor(item: V, expirationTimeNanos: bigint) {
+    this._item = item;
+    this._expirationTimeNanos = expirationTimeNanos;
+  }
+
+  get item(): V {
+    return this._item;
+  }
+
+  get expirationTimeNs(): bigint {
+    return this._expirationTimeNanos;
+  }
+
+  updateExpiration(expirationIntervalNanos: bigint): CacheItem<V> {
+    this._expirationTimeNanos = getTimeInNanos() + expirationIntervalNanos;
+    return this;
+  }
+}
+
+export class SlidingExpirationCache<K, V> {
+  private _cleanupIntervalNanos: bigint = BigInt(10 * 60_000_000_000); // 10 minutes
+  private readonly _shouldDisposeFunc?: (item: V) => boolean;
+  private readonly _itemDisposalFunc?: (item: V) => void;
+  private _map: MapUtils<K, CacheItem<V>> = new MapUtils<K, CacheItem<V>>();
+  private _cleanupTimeNanos: bigint;
+
+  constructor(cleanupIntervalNanos: bigint, shouldDisposeFunc?: (item: V) => boolean, itemDisposalFunc?: (item: V) => void) {
+    this._cleanupIntervalNanos = cleanupIntervalNanos;
+    this._shouldDisposeFunc = shouldDisposeFunc;
+    this._itemDisposalFunc = itemDisposalFunc;
+    this._cleanupTimeNanos = getTimeInNanos() + this._cleanupIntervalNanos;
+  }
+
+  get size(): number {
+    return this._map.size;
+  }
+
+  set cleanupIntervalNs(value: bigint) {
+    this._cleanupIntervalNanos = value;
+  }
+
+  computeIfAbsent(key: K, mappingFunc: (key: K) => V, itemExpirationNanos: bigint): V | null {
+    this.cleanUp();
+    const cacheItem = this._map.computeIfAbsent(key, (k) => new CacheItem(mappingFunc(k), getTimeInNanos() + itemExpirationNanos));
+    return cacheItem?.updateExpiration(itemExpirationNanos).item ?? null;
+  }
+
+  get(key: K): V | undefined {
+    this.cleanUp();
+    const cacheItem = this._map.get(key);
+    return cacheItem?.item ?? undefined;
+  }
+
+  remove(key: K): void {
+    this.removeAndDispose(key);
+    this.cleanUp();
+  }
+
+  removeAndDispose(key: K): void {
+    const cacheItem = this._map.remove(key);
+    if (cacheItem != null && this._itemDisposalFunc != null) {
+      this._itemDisposalFunc(cacheItem.item);
+    }
+  }
+
+  removeIfExpired(key: K): void {
+    const cacheItem = this._map.get(key);
+    if (cacheItem == null || this.shouldCleanupItem(cacheItem)) {
+      this.removeAndDispose(key);
+    }
+  }
+
+  shouldCleanupItem(cacheItem: CacheItem<V>): boolean {
+    if (this._shouldDisposeFunc != null) {
+      return getTimeInNanos() > cacheItem.expirationTimeNs && this._shouldDisposeFunc(cacheItem.item);
+    }
+    return getTimeInNanos() > cacheItem.expirationTimeNs;
+  }
+
+  clear(): void {
+    for (const [key, val] of this._map.entries) {
+      if (val !== undefined && this._itemDisposalFunc !== undefined) {
+        this._itemDisposalFunc(val.item);
+      }
+    }
+    this._map.clear();
+  }
+
+  protected cleanUp() {
+    const currentTime = getTimeInNanos();
+    if (this._cleanupTimeNanos > currentTime) {
+      return;
+    }
+    this._cleanupTimeNanos = currentTime + this._cleanupIntervalNanos;
+    for (const k of this._map.keys) {
+      this.removeIfExpired(k);
+    }
+  }
+}

common/lib/utils/utils.ts

@@ -38,7 +38,7 @@ export function logTopology(hosts: HostInfo[], msgPrefix: string) {
 }
 
 export function getTimeInNanos() {
-  return performance.now();
+  return process.hrtime.bigint();
 }
 
 export function maskProperties(props: Map<string, any>) {

common/lib/wrapper_property.ts

@@ -98,7 +98,12 @@ export class WrapperProperties {
 
   static removeWrapperProperties<T>(config: T): T {
     const copy = Object.assign({}, config);
-    const persistingProperties = [WrapperProperties.USER.name, WrapperProperties.PASSWORD.name, WrapperProperties.DATABASE.name, WrapperProperties.PORT.name];
+    const persistingProperties = [
+      WrapperProperties.USER.name,
+      WrapperProperties.PASSWORD.name,
+      WrapperProperties.DATABASE.name,
+      WrapperProperties.PORT.name
+    ];
 
     Object.values(WrapperProperties).forEach((prop) => {
       if (prop instanceof WrapperProperty) {

mysql/lib/client.ts

@@ -43,7 +43,7 @@ export class AwsMySQLClient extends AwsClient {
 
   async connect(): Promise<Connection> {
     await this.internalConnect();
-    let conn: Promise<Connection> = this.pluginManager.connect(this.pluginService.getCurrentHostInfo(), this.properties, true, async () => {
+    const conn: Promise<Connection> = this.pluginManager.connect(this.pluginService.getCurrentHostInfo(), this.properties, true, async () => {
       return this.targetClient.promise().connect();
     });
     this.isConnected = true;

pg/lib/client.ts

@@ -39,7 +39,7 @@ export class AwsPGClient extends AwsClient {
 
   async connect(): Promise<void> {
     await this.internalConnect();
-    let res: Promise<void> = this.pluginManager.connect(this.pluginService.getCurrentHostInfo(), this.properties, true, () =>
+    const res: Promise<void> = this.pluginManager.connect(this.pluginService.getCurrentHostInfo(), this.properties, true, () =>
       this.targetClient.connect()
     );
     this.isConnected = true;