Merge branch 'fix/1438/optimize_sources_by_pubkey' into 'dev'

[fix] Optimize access to sources by pubkey, using specific index

Closes #1438

See merge request !1420

app/lib/common-libs/constants.ts

@@ -53,7 +53,7 @@ const CONDITIONS =
   "\\)|CSV\\(" +
   CSV_INTEGER +
   "\\))))*";
-
+const CONDITION_SIG_PUBKEY = "SIG\\((" + PUBKEY + ")\\)";
 const BMA_REGEXP = /^BASIC_MERKLED_API( ([a-z_][a-z0-9-_.]*))?( ([0-9.]+))?( ([0-9a-f:]+))?( ([0-9]+))$/;
 const BMAS_REGEXP = /^BMAS( ([a-z_][a-z0-9-_.]*))?( ([0-9.]+))?( ([0-9a-f:]+))?( ([0-9]+))( (\/.+))?$/;
 const BMATOR_REGEXP = /^BMATOR( ([a-z0-9]{16})\.onion)( ([0-9.]+))?( ([0-9a-f:]+))?( ([0-9]+))$/;
@@ -535,6 +535,8 @@ export const CommonConstants = {
     LOCKTIME: find("Locktime: (" + INTEGER + ")"),
     INLINE_COMMENT: exact(COMMENT),
     OUTPUT_CONDITION: exact(CONDITIONS),
+    OUTPUT_CONDITION_SIG_PUBKEY: find(CONDITION_SIG_PUBKEY),
+    OUTPUT_CONDITION_SIG_PUBKEY_UNIQUE: exact(CONDITION_SIG_PUBKEY),
   },
   PEER: {
     BLOCK: find("Block: (" + INTEGER + "-" + FINGERPRINT + ")"),

app/lib/dal/fileDAL.ts

@@ -470,8 +470,10 @@ export class FileDAL implements ServerDAO {
   }
 
   async getAvailableSourcesByPubkey(pubkey: string): Promise<HttpSource[]> {
-    const txAvailable = await this.sindexDAL.getAvailableForPubkey(pubkey);
-    const sources: UDSource[] = await this.dividendDAL.getUDSources(pubkey);
+    const [txAvailable, sources] = await Promise.all([
+      this.sindexDAL.getAvailableForPubkey(pubkey),
+      this.dividendDAL.getUDSources(pubkey),
+    ]);
     return sources
       .map((d) => {
         return {

app/lib/dal/indexDAL/leveldb/LevelDBSindex.ts

@@ -12,12 +12,14 @@ import { SIndexDAO } from "../abstract/SIndexDAO";
 import { Underscore } from "../../../common-libs/underscore";
 import { pint } from "../../../common-libs/pint";
 import { arrayPruneAllCopy } from "../../../common-libs/array-prune";
+import { CommonConstants } from "../../../common-libs/constants";
 
 export class LevelDBSindex extends LevelDBTable<SindexEntry>
   implements SIndexDAO {
   private indexForTrimming: LevelDBTable<string[]>;
   private indexForConsumed: LevelDBTable<string[]>;
   private indexForConditions: LevelDBTable<string[]>;
+  private indexOfComplexeConditionForPubkeys: LevelDBTable<string[]>;
 
   constructor(protected getLevelDB: (dbName: string) => Promise<LevelUp>) {
     super("level_sindex", getLevelDB);
@@ -41,9 +43,14 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
       "level_sindex/conditions",
       this.getLevelDB
     );
+    this.indexOfComplexeConditionForPubkeys = new LevelDBTable<string[]>(
+      "level_sindex/complex_condition_pubkeys",
+      this.getLevelDB
+    );
     await this.indexForTrimming.init();
     await this.indexForConsumed.init();
     await this.indexForConditions.init();
+    await this.indexOfComplexeConditionForPubkeys.init();
   }
 
   async close(): Promise<void> {
@@ -51,6 +58,7 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
     await this.indexForTrimming.close();
     await this.indexForConsumed.close();
     await this.indexForConditions.close();
+    await this.indexOfComplexeConditionForPubkeys.close();
   }
 
   /**
@@ -127,14 +135,14 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
       pos: number;
     }[]
   > {
-    // TODO: very costly: needs a full scan, would be better to change this implementatio
-    const entries = await this.findWhere((e) =>
-      e.conditions.includes(`SIG(${pubkey})`)
+    const forSimpleConditions = await this.getForConditions(`SIG(${pubkey})`);
+    const forComplexConditions = await this.getForComplexeConditionPubkey(
+      pubkey
+    );
+    const reduced = Indexer.DUP_HELPERS.reduceBy(
+      forSimpleConditions.concat(forComplexConditions),
+      ["identifier", "pos"]
     );
-    const reduced = Indexer.DUP_HELPERS.reduceBy(entries, [
-      "identifier",
-      "pos",
-    ]);
     return reduced.filter((r) => !r.consumed);
   }
 
@@ -269,6 +277,20 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
     return found;
   }
 
+  async getForComplexeConditionPubkey(pubkey: string): Promise<SindexEntry[]> {
+    const ids =
+      (await this.indexOfComplexeConditionForPubkeys.getOrNull(pubkey)) || [];
+    const found: SindexEntry[] = [];
+    for (const id of ids) {
+      const entries = await this.findByIdentifierAndPos(
+        id.split("-")[0],
+        pint(id.split("-")[1])
+      );
+      entries.forEach((e) => found.push(e));
+    }
+    return found;
+  }
+
   async removeBlock(blockstamp: string): Promise<void> {
     const writtenOn = pint(blockstamp);
     // We look at records written on this blockstamp: `indexForTrimming` allows to get them
@@ -316,24 +338,25 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
   }
 
   private async trimConditions(condition: string, id: string) {
-    // Get all the account's TX sources
+    // Get all the condition's sources
     const existing = (await this.indexForConditions.getOrNull(condition)) || [];
-    // Prune the source from the account
+    // Prune the source from the condition
     const trimmed = arrayPruneAllCopy(existing, id);
     if (trimmed.length) {
-      // If some sources are left for this "account", persist what remains
+      // If some sources are left for this "condition", persist what remains
       await this.indexForConditions.put(condition, trimmed);
     } else {
       // Otherwise just delete the "account"
       await this.indexForConditions.del(condition);
     }
+
+    // If complex conditions
+    if (this.isComplexCondition(condition)) {
+      const pubkeys = this.getDistinctPubkeysFromCondition(condition);
+      await this.trimComplexeConditionPubkeys(pubkeys, id);
+    }
   }
 
-  /**
-   * Duplicate with trimConditions?!
-   * @param writtenOn
-   * @param id
-   */
   private async trimWrittenOn(writtenOn: number, id: string) {
     const k = LevelDBSindex.trimWrittenOnKey(writtenOn);
     const existing = await this.getWrittenOnSourceIds(writtenOn);
@@ -356,6 +379,28 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
     }
   }
 
+  private async trimComplexeConditionPubkeys(pubkeys: string[], id: string) {
+    if (!pubkeys || !pubkeys.length) return;
+    for (const p of pubkeys) {
+      await this.trimComplexeConditionPubkey(p, id);
+    }
+  }
+
+  private async trimComplexeConditionPubkey(pubkey: string, id: string) {
+    // Get all the condition's sources
+    const existing =
+      (await this.indexOfComplexeConditionForPubkeys.getOrNull(pubkey)) || [];
+    // Prune the source from the condition
+    const trimmed = arrayPruneAllCopy(existing, id);
+    if (trimmed.length) {
+      // If some sources are left for this "condition", persist what remains
+      await this.indexOfComplexeConditionForPubkeys.put(pubkey, trimmed);
+    } else {
+      // Otherwise just delete the "account"
+      await this.indexOfComplexeConditionForPubkeys.del(pubkey);
+    }
+  }
+
   private async getWrittenOnSourceIds(writtenOn: number) {
     const indexForTrimmingId = LevelDBSindex.trimWrittenOnKey(writtenOn);
     return (await this.indexForTrimming.getOrNull(indexForTrimmingId)) || [];
@@ -393,6 +438,7 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
     const byConsumed: { [k: number]: SindexEntry[] } = {};
     const byWrittenOn: { [k: number]: SindexEntry[] } = {};
     const byConditions: { [k: string]: SindexEntry[] } = {};
+    const byPubkeys: { [k: string]: SindexEntry[] } = {};
     records
       .filter((r) => r.consumed)
       .forEach((r) => {
@@ -410,12 +456,24 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
         arrWO = byWrittenOn[r.writtenOn] = [];
       }
       arrWO.push(r);
-      // Conditiosn
+      // Conditions
       let arrCN = byConditions[r.conditions];
       if (!arrCN) {
         arrCN = byConditions[r.conditions] = [];
       }
       arrCN.push(r);
+
+      // If complex condition
+      if (this.isComplexCondition(r.conditions)) {
+        const pubkeys = this.getDistinctPubkeysFromCondition(r.conditions);
+        pubkeys.forEach((pub) => {
+          let arrPub = byPubkeys[pub];
+          if (!arrPub) {
+            arrPub = byPubkeys[pub] = [];
+          }
+          arrPub.push(r);
+        });
+      }
     });
     // Index consumed => (identifier + pos)[]
     for (const k of Underscore.keys(byConsumed)) {
@@ -446,5 +504,47 @@ export class LevelDBSindex extends LevelDBTable<SindexEntry>
         Underscore.uniq(existing.concat(newSources))
       );
     }
+    // Index pubkeys => (identifier + pos)[]
+    for (const k of Underscore.keys(byPubkeys).map(String)) {
+      const existing =
+        (await this.indexOfComplexeConditionForPubkeys.getOrNull(k)) || [];
+      const newSources = byPubkeys[k].map((r) =>
+        LevelDBSindex.trimPartialKey(r.identifier, r.pos)
+      );
+      await this.indexOfComplexeConditionForPubkeys.put(
+        k,
+        Underscore.uniq(existing.concat(newSources))
+      );
+    }
+  }
+
+  private isComplexCondition(condition: string): boolean {
+    return (
+      (condition &&
+        !CommonConstants.TRANSACTION.OUTPUT_CONDITION_SIG_PUBKEY_UNIQUE.test(
+          condition
+        )) ||
+      false
+    );
+  }
+  /**
+   * Get all pubkeys used by an output condition (e.g. 'SIG(A) && SIG(B)' will return ['A', 'B']
+   * @param condition
+   * @private
+   */
+  private getDistinctPubkeysFromCondition(condition: string): string[] {
+    const pubKeys: string[] = [];
+    if (!condition) return pubKeys;
+    let match: RegExpExecArray | null;
+    while (
+      (match = CommonConstants.TRANSACTION.OUTPUT_CONDITION_SIG_PUBKEY.exec(
+        condition
+      )) !== null
+    ) {
+      pubKeys.push(match[1]);
+      condition = condition.substring(match.index + match[0].length);
+    }
+
+    return Underscore.uniq(pubKeys);
   }
 }

test/dal/sources-dal.ts

@@ -18,30 +18,70 @@ const should = require('should');
 
 let dal:FileDAL
 
-describe("Source DAL", function(){
+describe("Source DAL", function() {
+  const pubkeyA = 'BYfWYFrsyjpvpFysgu19rGK3VHBkz4MqmQbNyEuVU64g';
+  const pubkeyB = 'DSz4rgncXCytsUMW2JU2yhLquZECD2XpEkpP9gG5HyAx';
 
   before(async () => {
     dal = new FileDAL(await Directory.getHomeParams(true, 'db0'), async (name: string) => Directory.getHomeDB(true, name), async (name: string) => Directory.getHomeLevelDB(true, name))
     await dal.init({} as any)
   })
 
-  it('should be able to feed the sindex with unordered rows', async () => {
+  it('should be able to fill the sindex with unordered rows', async () => {
     await dal.sindexDAL.insertBatch([
-      { op: 'UPDATE', tx: null, identifier: 'SOURCE_1', pos: 4, written_on: '139-H', writtenOn: 139, written_time: 4500, consumed: true,  conditions: 'SIG(ABC)' },
-      { op: 'CREATE', tx: null, identifier: 'SOURCE_1', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: 'SIG(ABC)' },
-      { op: 'CREATE', tx: null, identifier: 'SOURCE_2', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: 'SIG(ABC)' },
-      { op: 'CREATE', tx: null, identifier: 'SOURCE_3', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: 'SIG(DEF)' }
+      { op: 'UPDATE', tx: null, identifier: 'SOURCE_1', pos: 4, written_on: '139-H', writtenOn: 139, written_time: 4500, consumed: true,  conditions: `SIG(${pubkeyA})` },
+      { op: 'CREATE', tx: null, identifier: 'SOURCE_1', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: `SIG(${pubkeyA})` },
+      { op: 'CREATE', tx: null, identifier: 'SOURCE_2', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: `SIG(${pubkeyA})` },
+      { op: 'CREATE', tx: null, identifier: 'SOURCE_3', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: `SIG(${pubkeyB})` }
     ] as any);
     (await dal.sindexDAL.findByIdentifier('SOURCE_1')).should.have.length(2);
     (await dal.sindexDAL.findByPos(4)).should.have.length(4);
     // Source availability
-    const sourcesOfDEF = await dal.sindexDAL.getAvailableForPubkey('DEF');
-    sourcesOfDEF.should.have.length(1);
-    const sourcesOfABC = await dal.sindexDAL.getAvailableForPubkey('ABC');
-    sourcesOfABC.should.have.length(1);
+    const sourcesOfA = await dal.sindexDAL.getAvailableForPubkey(pubkeyA);
+    sourcesOfA.should.have.length(1);
+    const sourcesOfB = await dal.sindexDAL.getAvailableForPubkey(pubkeyB);
+    sourcesOfB.should.have.length(1);
     const source1 = await dal.sindexDAL.getTxSource('SOURCE_1', 4) as any
     source1.should.have.property('consumed').equal(true);
     const source2 = await dal.sindexDAL.getTxSource('SOURCE_2', 4) as any
     source2.should.have.property('consumed').equal(false);
+
+    // Check sources not available after block deletion
+    await dal.sindexDAL.removeBlock('126-H');
+    (await dal.sindexDAL.findByIdentifier('SOURCE_1')).should.have.length(1);
+    should(await dal.sindexDAL.getTxSource('SOURCE_2', 4) as any).be.null();
+    should(await dal.sindexDAL.getTxSource('SOURCE_3', 4) as any).be.null();
+    (await dal.sindexDAL.findByPos(4)).should.have.length(1);
+    await dal.sindexDAL.removeBlock('139-H');
+    (await dal.sindexDAL.findByIdentifier('SOURCE_1')).should.have.length(0);
+    (await dal.sindexDAL.findByPos(4)).should.have.length(0);
+    (await dal.sindexDAL.getAvailableForPubkey(pubkeyA)).should.have.length(0);
+    (await dal.sindexDAL.getAvailableForPubkey(pubkeyB)).should.have.length(0);
+    should(await dal.sindexDAL.getTxSource('SOURCE_1', 4) as any).be.null();
+    should(await dal.sindexDAL.getTxSource('SOURCE_2', 4) as any).be.null();
+    should(await dal.sindexDAL.getTxSource('SOURCE_3', 4) as any).be.null();
+  })
+
+  it('should be able to read sindex by pubkey', async () => {
+    // Test insertion, using complex condition
+    await dal.sindexDAL.insertBatch([
+      { op: 'CREATE', tx: null, identifier: 'SOURCE_4', pos: 4, written_on: '139-H', writtenOn: 139, written_time: 2000, consumed: false, conditions: `SIG(${pubkeyA})` },
+      { op: 'CREATE', tx: null, identifier: 'SOURCE_5', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: `(SIG(${pubkeyA}) && SIG(${pubkeyB}))` },
+      { op: 'CREATE', tx: null, identifier: 'SOURCE_6', pos: 4, written_on: '126-H', writtenOn: 126, written_time: 2000, consumed: false, conditions: `(XHX(3EB4702F2AC2FD3FA4FDC46A4FC05AE8CDEE1A85F2AC2FD3FA4FDC46A4FC01CA) || SIG(${pubkeyB}))` }
+    ] as any);
+
+    // Check sources availability by pubkey
+    let sourcesOfA = await dal.sindexDAL.getAvailableForPubkey(pubkeyA);
+    sourcesOfA.should.have.length(2);
+    let sourcesOfB = await dal.sindexDAL.getAvailableForPubkey(pubkeyB);
+    sourcesOfB.should.have.length(2);
+
+    // Check sources not available after block deletion
+    await dal.sindexDAL.removeBlock('126-H');
+    await dal.sindexDAL.removeBlock('139-H');
+    sourcesOfA = await dal.sindexDAL.getAvailableForPubkey(pubkeyA);
+    sourcesOfA.should.have.length(0);
+    sourcesOfB = await dal.sindexDAL.getAvailableForPubkey(pubkeyB);
+    sourcesOfB.should.have.length(0);
   })
 })