cache for sqlite method calls to prevent duplicate writes

src/database/standalone-sqlite.ts

@@ -29,6 +29,7 @@ import * as R from 'ramda';
 import sql from 'sql-bricks';
 import * as winston from 'winston';
 import CircuitBreaker from 'opossum';
+import NodeCache from 'node-cache';
 
 // TODO enable eslint
 /* eslint-disable */
@@ -405,6 +406,8 @@ export class StandaloneSqliteDatabaseWorker {
   private bundleFormatIds: { [filter: string]: number } = {};
   private filterIds: { [filter: string]: number } = {};
 
+  private insertDataHashCache: NodeCache;
+
   // Transactions
   resetBundlesToHeightFn: Sqlite.Transaction;
   resetCoreToHeightFn: Sqlite.Transaction;
@@ -761,6 +764,12 @@ export class StandaloneSqliteDatabaseWorker {
         });
       },
     );
+
+    this.insertDataHashCache = new NodeCache({
+      stdTTL: 60 * 7, // 7 minutes
+      checkperiod: 60, // 1 minute
+      useClones: false,
+    });
   }
 
   getMaxHeight() {
@@ -1132,13 +1141,7 @@ export class StandaloneSqliteDatabaseWorker {
     cachedAt?: number;
   }) {
     const hashBuffer = fromB64Url(hash);
-    this.stmts.data.insertDataHash.run({
-      hash: hashBuffer,
-      data_size: dataSize,
-      original_source_content_type: contentType,
-      indexed_at: currentUnixTimestamp(),
-      cached_at: cachedAt,
-    });
+
     this.stmts.data.insertDataId.run({
       id: fromB64Url(id),
       contiguous_data_hash: hashBuffer,
@@ -1151,6 +1154,19 @@ export class StandaloneSqliteDatabaseWorker {
         indexed_at: currentUnixTimestamp(),
       });
     }
+
+    if (this.insertDataHashCache.get(hash)) {
+      return;
+    }
+    this.insertDataHashCache.set(hash, true);
+
+    this.stmts.data.insertDataHash.run({
+      hash: hashBuffer,
+      data_size: dataSize,
+      original_source_content_type: contentType,
+      indexed_at: currentUnixTimestamp(),
+      cached_at: cachedAt,
+    });
   }
 
   getGqlNewTransactionTags(txId: Buffer) {
@@ -2432,6 +2448,8 @@ export class StandaloneSqliteDatabase
     Awaited<ReturnType<StandaloneSqliteDatabase['getTransactionAttributes']>>
   >;
 
+  private saveDataContentAttributesCache: NodeCache;
+
   constructor({
     log,
     coreDbPath,
@@ -2505,6 +2523,16 @@ export class StandaloneSqliteDatabase
       this.getTransactionAttributesCircuitBreaker,
     ]);
 
+    //
+    // Initialize method caches
+    //
+
+    this.saveDataContentAttributesCache = new NodeCache({
+      stdTTL: 60 * 7, // 7 minutes
+      checkperiod: 60, // 1 minute
+      useClones: false,
+    });
+
     //
     // Initialize workers
     //
@@ -2780,6 +2808,15 @@ export class StandaloneSqliteDatabase
     dataSize: number;
     contentType?: string;
   }) {
+    if (this.saveDataContentAttributesCache.get(id)) {
+      metrics.saveMethodsDuplicateCounter.inc({
+        method: 'saveDataContentAttributes',
+      });
+      return Promise.resolve();
+    }
+
+    this.saveDataContentAttributesCache.set(id, true);
+
     return this.queueWrite('data', 'saveDataContentAttributes', [
       {
         id,

src/metrics.ts

@@ -124,6 +124,12 @@ export const methodDurationSummary = new promClient.Summary({
   labelNames: ['worker', 'role', 'method'],
 });
 
+export const saveMethodsDuplicateCounter = new promClient.Counter({
+  name: 'save_methods_duplicate_total',
+  help: 'Count of duplicate calls to save methods',
+  labelNames: ['method'],
+});
+
 //
 // Block importer metrics
 //