fix: use keyv to persist Ethereum metadata cache (#431)

Author: lumtis

packages/ethereum-storage/package.json

@@ -55,6 +55,7 @@
     "bn.js": "4.11.8",
     "form-data": "2.3.3",
     "ipfs-unixfs": "0.1.16",
+    "keyv": "3.1.0",
     "sinon": "7.3.2",
     "web3-eth": "1.0.0-beta.37",
     "web3-utils": "1.0.0-beta.37"

packages/ethereum-storage/src/lib/ethereum-metadata-cache.ts

@@ -1,6 +1,8 @@
 import { StorageTypes } from '@requestnetwork/types';
 import SmartContractManager from './smart-contract-manager';
 
+import * as Keyv from 'keyv';
+
 /**
  * Allows to save and retrieve ethereum metadata associated to a data id
  * Metadata represents general information about the Ethereum network used like network name and smart contract address
@@ -12,7 +14,7 @@ export default class EthereumMetadataCache {
    * Store the ethereum metadata for a data id in a dictionary
    * This attribute is left public for mocking purpose to facilitate tests on the module
    */
-  public metadataCache: { [dataId: string]: StorageTypes.IEthereumMetadata } = {};
+  public metadataCache: Keyv<StorageTypes.IEthereumMetadata>;
 
   /**
    * Manager for the storage smart contract
@@ -23,24 +25,30 @@ export default class EthereumMetadataCache {
   /**
    * Constructor
    * @param smartContractManager Instance of SmartContractManager used to get metadata in case they're not registered yet
+   * @param store a Keyv store to persist the metadata
    */
-  public constructor(smartContractManager: SmartContractManager) {
+  public constructor(smartContractManager: SmartContractManager, store?: Keyv.Store<any>) {
     this.smartContractManager = smartContractManager;
+
+    this.metadataCache = new Keyv<StorageTypes.IEthereumMetadata>({
+      namespace: 'ethereumMetadata',
+      store,
+    });
   }
 
   /**
    * Saves in the cache the Ethereum metadata related to a dataId
    * @param dataId dataId to index ethereum metadata
    * @param meta Ethereum metadata related to the dataId
    */
-  public saveDataIdMeta(dataId: string, meta: StorageTypes.IEthereumMetadata): void {
+  public async saveDataIdMeta(dataId: string, meta: StorageTypes.IEthereumMetadata): Promise<void> {
     // We save the metadata only if it doesn't exist yet
     // A user can add the same dataId into the smart contract indefinitely
     // Therefore, only the first occurrence of the dataId has valid metadata
     // Finding several occurrences of the same dataId is not abnormal and we don't throw an error in this case
     // PROT-503: We should ensure the corresponding metadata is the metadata of the first occurrence of the dataId
-    if (!this.metadataCache[dataId]) {
-      this.metadataCache[dataId] = meta;
+    if (!(await this.metadataCache.get(dataId))) {
+      await this.metadataCache.set(dataId, meta);
     }
   }
 
@@ -54,10 +62,12 @@ export default class EthereumMetadataCache {
   public async getDataIdMeta(dataId: string): Promise<StorageTypes.IEthereumMetadata> {
     // If the metadata has not been saved in the cache yet
     // we get them with smartContractManager and save them
-    if (!this.metadataCache[dataId]) {
-      this.metadataCache[dataId] = await this.smartContractManager.getMetaFromEthereum(dataId);
+    let metadata: StorageTypes.IEthereumMetadata | undefined = await this.metadataCache.get(dataId);
+    if (!metadata) {
+      metadata = await this.smartContractManager.getMetaFromEthereum(dataId);
+      await this.metadataCache.set(dataId, metadata);
     }
 
-    return this.metadataCache[dataId];
+    return metadata;
   }
 }

packages/ethereum-storage/src/lib/ethereum-storage.ts

@@ -6,6 +6,8 @@ import EthereumMetadataCache from './ethereum-metadata-cache';
 import IpfsManager from './ipfs-manager';
 import SmartContractManager from './smart-contract-manager';
 
+import * as Keyv from 'keyv';
+
 // rate of the size of the Header of a ipfs file regarding its content size
 // used to estimate the size of a ipfs file from the content size
 const SAFE_RATE_HEADER_SIZE: number = 0.3;
@@ -51,6 +53,7 @@ export default class EthereumStorage implements StorageTypes.IStorage {
    * @param ipfsGatewayConnection Information structure to connect to the ipfs gateway
    * @param web3Connection Information structure to connect to the Ethereum network
    * @param [options.getLastBlockNumberDelay] the minimum delay to wait between fetches of lastBlockNumber
+   * @param metadataStore a Keyv store to persist the metadata in ethereumMetadataCache
    */
   public constructor(
     ipfsGatewayConnection?: StorageTypes.IIpfsGatewayConnection,
@@ -68,6 +71,7 @@ export default class EthereumStorage implements StorageTypes.IStorage {
       maxRetries?: number;
       retryDelay?: number;
     } = {},
+    metadataStore?: Keyv.Store<any>,
   ) {
     this.maxConcurrency = maxConcurrency || getMaxConcurrency();
     this.logger = logger || new Utils.SimpleLogger();
@@ -79,7 +83,7 @@ export default class EthereumStorage implements StorageTypes.IStorage {
       maxRetries,
       retryDelay,
     });
-    this.ethereumMetadataCache = new EthereumMetadataCache(this.smartContractManager);
+    this.ethereumMetadataCache = new EthereumMetadataCache(this.smartContractManager, metadataStore);
   }
 
   /**
@@ -220,7 +224,7 @@ export default class EthereumStorage implements StorageTypes.IStorage {
     }
 
     // Save the metadata of the new dataId into the Ethereum metadata cache
-    this.ethereumMetadataCache.saveDataIdMeta(dataId, ethereumMetadata);
+    await this.ethereumMetadataCache.saveDataIdMeta(dataId, ethereumMetadata);
 
     return {
       meta: {
@@ -393,7 +397,7 @@ export default class EthereumStorage implements StorageTypes.IStorage {
       const ethereumMetadata = contentDataIdAndMeta.meta.metaData[i].ethereum;
       if (ethereumMetadata) {
         // PROT-504: The saving of dataId's metadata should be encapsulated when retrieving dataId inside smart contract (getPastEvents)
-        this.ethereumMetadataCache.saveDataIdMeta(dataIds[i], ethereumMetadata);
+        await this.ethereumMetadataCache.saveDataIdMeta(dataIds[i], ethereumMetadata);
       }
     }
 

packages/ethereum-storage/test/lib/ethereum-metadata-cache.ts

@@ -1,13 +1,15 @@
 import 'mocha';
 
 import * as chai from 'chai';
+import * as chaiAsPromised from 'chai-as-promised';
 
 import { StorageTypes } from '@requestnetwork/types';
 import EthereumMetadataCache from '../../src/lib/ethereum-metadata-cache';
 import SmartContractManager from '../../src/lib/smart-contract-manager';
 
 const spies = require('chai-spies');
 chai.use(spies);
+chai.use(chaiAsPromised);
 const expect = chai.expect;
 const sandbox = chai.spy.sandbox();
 
@@ -56,36 +58,38 @@ const getMetaFromEthereumMock = async (hash: string): Promise<StorageTypes.IEthe
 };
 
 describe('EthereumMetadataCache', () => {
-
   beforeEach(() => {
     sandbox.restore();
     smartContractManager = new SmartContractManager();
     ethereumMetadataCache = new EthereumMetadataCache(smartContractManager);
   });
 
   it('allows to save metadata', async () => {
-    await expect(ethereumMetadataCache.metadataCache[hashExample1]).to.be.undefined;
-    await expect(ethereumMetadataCache.metadataCache[hashExample2]).to.be.undefined;
-    await expect(ethereumMetadataCache.metadataCache[hashExample3]).to.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample1)).to.eventually.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample2)).to.eventually.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample3)).to.eventually.be.undefined;
 
-    ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample1);
-    ethereumMetadataCache.saveDataIdMeta(hashExample2, ethereumMetadataExample2);
-    ethereumMetadataCache.saveDataIdMeta(hashExample3, ethereumMetadataExample3);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample1);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample2, ethereumMetadataExample2);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample3, ethereumMetadataExample3);
 
-    await expect(ethereumMetadataCache.metadataCache[hashExample1]).to.deep.equal(ethereumMetadataExample1);
-    await expect(ethereumMetadataCache.metadataCache[hashExample2]).to.deep.equal(ethereumMetadataExample2);
-    await expect(ethereumMetadataCache.metadataCache[hashExample3]).to.deep.equal(ethereumMetadataExample3);
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample1)).to.eventually.deep.equal(
+      ethereumMetadataExample1,
+    );
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample2)).to.eventually.deep.equal(
+      ethereumMetadataExample2,
+    );
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample3)).to.eventually.deep.equal(
+      ethereumMetadataExample3,
+    );
   });
 
   it('allows to retrieve saved metadata', async () => {
-    const spy = sandbox.on(
-      smartContractManager,
-      'getMetaFromEthereum',
-    );
+    const spy = sandbox.on(smartContractManager, 'getMetaFromEthereum');
 
-    ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample1);
-    ethereumMetadataCache.saveDataIdMeta(hashExample2, ethereumMetadataExample2);
-    ethereumMetadataCache.saveDataIdMeta(hashExample3, ethereumMetadataExample3);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample1);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample2, ethereumMetadataExample2);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample3, ethereumMetadataExample3);
 
     const readReturn1 = await ethereumMetadataCache.getDataIdMeta(hashExample1);
     const readReturn2 = await ethereumMetadataCache.getDataIdMeta(hashExample2);
@@ -101,35 +105,34 @@ describe('EthereumMetadataCache', () => {
   it('allows to save when trying to read new metadata', async () => {
     smartContractManager.getMetaFromEthereum = getMetaFromEthereumMock;
 
-    const spy = sandbox.on(
-      smartContractManager,
-      'getMetaFromEthereum',
-    );
+    const spy = sandbox.on(smartContractManager, 'getMetaFromEthereum');
 
-    await expect(ethereumMetadataCache.metadataCache[hashExample1]).to.be.undefined;
-    await expect(ethereumMetadataCache.metadataCache[hashExample2]).to.be.undefined;
-    await expect(ethereumMetadataCache.metadataCache[hashExample3]).to.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample1)).to.eventually.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample2)).to.eventually.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample3)).to.eventually.be.undefined;
 
     const readReturn1 = await ethereumMetadataCache.getDataIdMeta(hashExample1);
     await expect(readReturn1).to.deep.equal(ethereumMetadataExample1);
     await expect(spy).to.have.been.called.once;
-    await expect(ethereumMetadataCache.metadataCache[hashExample2]).to.be.undefined;
-    await expect(ethereumMetadataCache.metadataCache[hashExample3]).to.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample2)).to.eventually.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample3)).to.eventually.be.undefined;
 
     const readReturn2 = await ethereumMetadataCache.getDataIdMeta(hashExample2);
     await expect(readReturn2).to.deep.equal(ethereumMetadataExample2);
     await expect(spy).to.have.been.called.twice;
-    await expect(ethereumMetadataCache.metadataCache[hashExample3]).to.be.undefined;
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample3)).to.eventually.be.undefined;
 
     const readReturn3 = await ethereumMetadataCache.getDataIdMeta(hashExample3);
     await expect(readReturn3).to.deep.equal(ethereumMetadataExample3);
     await expect(spy).to.have.been.called.exactly(3);
   });
 
   it('cannot erase metadata of dataId with new metadata', async () => {
-    ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample1);
-    ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample2);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample1);
+    await ethereumMetadataCache.saveDataIdMeta(hashExample1, ethereumMetadataExample2);
 
-    await expect(ethereumMetadataCache.metadataCache[hashExample1]).to.deep.equal(ethereumMetadataExample1);
+    await expect(ethereumMetadataCache.metadataCache.get(hashExample1)).to.eventually.deep.equal(
+      ethereumMetadataExample1,
+    );
   });
 });

packages/ethereum-storage/test/lib/ethereum-storage.test.ts

@@ -292,17 +292,14 @@ describe('EthereumStorage', () => {
       }
     });
     it(`allows to save dataId's Ethereum metadata into the metadata cache when append is called`, async () => {
-      await assert.isUndefined(ethereumStorage.ethereumMetadataCache.metadataCache[hash1]);
+      await expect(ethereumStorage.ethereumMetadataCache.metadataCache.get(hash1)).to.eventually.be.undefined;
 
       const result = await ethereumStorage.append(content1);
-      await assert.deepEqual(
-        result.meta.ethereum,
-        ethereumStorage.ethereumMetadataCache.metadataCache[hash1],
-      );
+      await expect(ethereumStorage.ethereumMetadataCache.metadataCache.get(hash1)).to.eventually.deep.equal(result.meta.ethereum);
     });
 
     it(`prevents already saved dataId's Ethereum metadata to be erased in the metadata cache when append is called`, async () => {
-      await assert.isUndefined(ethereumStorage.ethereumMetadataCache.metadataCache[hash1]);
+      await expect(ethereumStorage.ethereumMetadataCache.metadataCache.get(hash1)).to.eventually.be.undefined;
 
       const result1 = await ethereumStorage.append(content1);
 
@@ -327,17 +324,13 @@ describe('EthereumStorage', () => {
       const result2 = await ethereumStorage.append(content1);
 
       await assert.notDeepEqual(result1, result2);
-
-      await assert.deepEqual(
-        result1.meta.ethereum,
-        ethereumStorage.ethereumMetadataCache.metadataCache[hash1],
-      );
+      await expect(ethereumStorage.ethereumMetadataCache.metadataCache.get(hash1)).to.eventually.deep.equal(result1.meta.ethereum);
     });
 
     it('allows to read a file', async () => {
       // For this test, we don't want to use the ethereum metadata cache
       // We want to force the retrieval of metadata with getPastEvents function
-      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = (_dataId, _meta) => {};
+      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = async (_dataId, _meta) => { };
 
       await ethereumStorage.append(content1);
       const result = await ethereumStorage.read(hash1);
@@ -374,7 +367,7 @@ describe('EthereumStorage', () => {
     });
 
     it('cannot read if ethereumMetadataCache.getDataIdMeta fail', async () => {
-      ethereumStorage.ethereumMetadataCache.getDataIdMeta = () => {
+      ethereumStorage.ethereumMetadataCache.getDataIdMeta = async () => {
         throw Error('expected error');
       };
       await expect(ethereumStorage.read(content1)).to.eventually.rejectedWith(
@@ -450,7 +443,7 @@ describe('EthereumStorage', () => {
     it('allows to retrieve all data', async () => {
       // For this test, we don't want to use the ethereum metadata cache
       // We want to force the retrieval of metadata with getPastEvents function
-      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = (_dataId, _meta) => {};
+      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = async (_dataId, _meta) => { };
 
       // These contents have to be appended in order to check their size
       await ethereumStorage.append(content1);
@@ -517,7 +510,7 @@ describe('EthereumStorage', () => {
     it('doest get meta data if the fees are too low', async () => {
       // For this test, we don't want to use the ethereum metadata cache
       // We want to force the retrieval of metadata with getPastEvents function
-      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = (_dataId, _meta) => {};
+      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = async (_dataId, _meta) => {};
       ethereumStorage.smartContractManager.getHashesAndSizesFromEthereum = async (): Promise<
         any[]
       > => {
@@ -611,7 +604,7 @@ describe('EthereumStorage', () => {
     });
 
     it('allows to read a file', async () => {
-      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = (_dataId, _meta) => {};
+      ethereumStorage.ethereumMetadataCache.saveDataIdMeta = async (_dataId, _meta) => { };
 
       const content = [content1, content2];
       const realSizes = [realSize1, realSize2];

packages/request-node/README.md

@@ -227,8 +227,8 @@ Default values correspond to the basic configuration used to run a server in a t
 - `--storageConcurrency` Maximum number of concurrent calls to Ethereum or IPFS
   - Default value: `'200'`
   - Environment variable name: `$STORAGE_MAX_CONCURRENCY`
-- `--transactionIndexFilePath` Path to a file to persist the transaction index for faster intialization
-  - Environment variable name: `$TRANSACTION_INDEX_FILE_PATH`
+- `--initializationStorageFilePath` Path to a file to persist the ethereum metadata and transaction index for faster initialization
+  - Environment variable name: `$INITIALIZATION_STORAGE_FILE_PATH`
 - `--logLevel` The maximum level of messages we will log
   - Environment variable name: `$LOG_LEVEL`
   - Available levels: ERROR, WARN, INFO and DEBUG

packages/request-node/src/config.ts

@@ -209,12 +209,12 @@ export function getEthereumRetryDelay(): number {
 }
 
 /**
- * Get the transaction index storage (a json-like file) path.
- * @returns the path to the json-like file that stores the transaction index.
+ * Get the initialization storage (a json-like file) path.
+ * @returns the path to the json-like file that stores the initialization data (ethereum metadata and transaction index).
  */
-export function getTransactionIndexFilePath(): string | null {
+export function getInitializationStorageFilePath(): string | null {
   return (
-    (argv.transactionIndexFilePath as string) || process.env.TRANSACTION_INDEX_FILE_PATH || null
+    (argv.initializationStorageFilePath as string) || process.env.INITIALIZATION_STORAGE_FILE_PATH || null
   );
 }
 

packages/request-node/src/requestNode.ts

@@ -7,7 +7,7 @@ import * as httpStatus from 'http-status-codes';
 import KeyvFile from 'keyv-file';
 
 import Utils from '@requestnetwork/utils';
-import { getCustomHeaders, getMnemonic, getTransactionIndexFilePath } from './config';
+import { getCustomHeaders, getInitializationStorageFilePath, getMnemonic } from './config';
 import getChannelsByTopic from './request/getChannelsByTopic';
 import getTransactionsByChannelId from './request/getTransactionsByChannelId';
 import persistTransaction from './request/persistTransaction';
@@ -43,17 +43,17 @@ class RequestNode {
 
     this.logger = logger || new Utils.SimpleLogger();
 
-    // Use ethereum storage for the storage layer
-    const ethereumStorage: StorageTypes.IStorage = getEthereumStorage(getMnemonic(), this.logger);
-
-    const transactionIndexStoragePath = getTransactionIndexFilePath();
+    const initializationStoragePath = getInitializationStorageFilePath();
 
-    const store = transactionIndexStoragePath
+    const store = initializationStoragePath
       ? new KeyvFile({
-          filename: transactionIndexStoragePath,
+          filename: initializationStoragePath,
         })
       : undefined;
 
+    // Use ethereum storage for the storage layer
+    const ethereumStorage: StorageTypes.IStorage = getEthereumStorage(getMnemonic(), this.logger, store);
+
     // Use an in-file Transaction index if a path is specified, an in-memory otherwise
     const transactionIndex = new TransactionIndex(store);
 

packages/request-node/src/server.ts

@@ -20,7 +20,7 @@ const startNode = async (): Promise<void> => {
   IPFS protocol: ${config.getIpfsProtocol()}
   IPFS timeout: ${config.getIpfsTimeout()}
   Storage concurrency: ${config.getStorageConcurrency()}
-  Transaction Index path: ${config.getTransactionIndexFilePath()}
+  Initialization storage path: ${config.getInitializationStorageFilePath()}
 `;
 
   logger.info(serverMessage);