Feat/speed up chainfollower (#899)

* feat: speed up

* feat: speeded up sync

* feat: introduced caching

* chore: removed logs and wrapped chainfollower in try-catch to continue even when there are errors

* fix: fixed a bug, where the worker weren't able to read the assetId, since wrong data was passed into the queue

* fix: removed hex prefix of assetID string, otherwise the tokenregistry won't find it

packages/api-cardano-db-hasura/src/ChainFollower.ts

@@ -19,13 +19,17 @@ export class ChainFollower {
   private chainSyncClient: ChainSynchronizationClient
   private queue: PgBoss
   private state: RunnableModuleState
+  private cacheAssets : { assetId: string; assetName: string; firstAppearedInSlot: number; fingerprint: string; policyId: string; }[]
+  private cacheTimer : number
 
   constructor (
     readonly hasuraClient: HasuraBackgroundClient,
     private logger: Logger = dummyLogger,
     private queueConfig: DbConfig
   ) {
     this.state = null
+    this.cacheAssets = []
+    this.cacheTimer = Date.now()
   }
 
   public async initialize (ogmiosConfig: Config['ogmios'], getMostRecentPoint: () => Promise<PointOrOrigin[]>) {
@@ -60,31 +64,36 @@ export class ChainFollower {
             requestNext()
           },
           rollForward: async ({ block }, requestNext) => {
-            let b
-            switch (block.type) {
-              case 'praos':
-                b = block as BlockPraos
-                break
-              case 'bft':
-                b = block as BlockBFT
-                break
-              case 'ebb': // No transaction in there
-                return
-            }
-            if (b !== undefined && b.transactions !== undefined) {
-              for (const tx of b.transactions) {
-                if (tx.mint !== undefined) {
-                  for (const entry of Object.entries(tx.mint)) {
-                    const policyId = entry[0]
-                    const assetNames = Object.keys(entry[1])
-                    for (const assetName of assetNames) {
-                      await this.saveAsset(policyId, assetName, b)
+            try {
+              let b
+              switch (block.type) {
+                case 'praos':
+                  b = block as BlockPraos
+                  break
+                case 'bft':
+                  b = block as BlockBFT
+                  break
+                case 'ebb': // No transaction in there
+                  return
+              }
+              if (b !== undefined && b.transactions !== undefined) {
+                for (const tx of b.transactions) {
+                  if (tx.mint !== undefined) {
+                    for (const entry of Object.entries(tx.mint)) {
+                      const policyId = entry[0]
+                      const assetNames = Object.keys(entry[1])
+                      for (const assetName of assetNames) {
+                        await this.saveAsset(policyId, assetName, b)
+                      }
                     }
                   }
                 }
               }
+            } catch (e) {
+              console.log(e)
+            } finally {
+              requestNext()
             }
-            requestNext()
           }
         }
       )
@@ -102,20 +111,27 @@ export class ChainFollower {
 
   async saveAsset (policyId: string, assetName: string | undefined, b: BlockPraos | BlockBFT) {
     const assetId = `${policyId}${assetName !== undefined ? assetName : ''}`
-    if (!(await this.hasuraClient.hasAsset(assetId))) {
-      const asset = {
-        assetId,
-        assetName,
-        firstAppearedInSlot: b.slot,
-        fingerprint: assetFingerprint(policyId, assetName),
-        policyId
-      }
-      await this.hasuraClient.insertAssets([asset])
-      const SIX_HOURS = 21600
-      const THREE_MONTHS = 365
-      await this.queue.publish('asset-metadata-fetch-initial', { assetId }, {
-        retryDelay: SIX_HOURS,
-        retryLimit: THREE_MONTHS
+    const asset = {
+      assetId,
+      assetName,
+      firstAppearedInSlot: b.slot,
+      fingerprint: assetFingerprint(policyId, assetName),
+      policyId
+    }
+    // introducing a caching to speed things up. The GraphQL insertAssets takes a lot of time.
+    // Saving when > 1000 assets in the cache or every minute
+    this.cacheAssets.push(asset)
+    if (this.cacheAssets.length > 1000 || (Date.now() - this.cacheTimer) / 1000 > 60) {
+      this.cacheTimer = Date.now() // resetting the timer
+      const response = await this.hasuraClient.insertAssets(this.cacheAssets)
+      this.cacheAssets = []
+      response.insert_assets.returning.forEach((asset: { assetId: string }) => {
+        const SIX_HOURS = 21600
+        const THREE_MONTHS = 365
+        this.queue.publish('asset-metadata-fetch-initial', { assetId: asset.assetId.replace('\\x', '') }, {
+          retryDelay: SIX_HOURS,
+          retryLimit: THREE_MONTHS
+        })
       })
     }
   }

packages/api-cardano-db-hasura/src/HasuraBackgroundClient.ts

@@ -258,7 +258,13 @@ export class HasuraBackgroundClient {
     )
     const result = await this.client.request(
       gql`mutation InsertAssets($assets: [Asset_insert_input!]!) {
-          insert_assets(objects: $assets) {
+          insert_assets(
+              objects: $assets,
+              on_conflict: {
+                  constraint: Asset_pkey,
+                  update_columns: []
+              }
+          ) {
               returning {
                   name
                   policyId