fix: performance update unlocking entityIndexerStateDAL DAL between different accounts

Author: amalcaraz

packages/core/src/storage/entityIndexStorage.ts

@@ -30,7 +30,7 @@ export type EntityIndexStorageOptions<Entity> = LevelStorageOptions<
 }
 
 export type EntityIndexStorageInvokeOptions = StorageCommonOptions & {
-  atomic?: boolean
+  atomic?: string | boolean
 }
 export type EntityIndexStorageCallOptions = EntityIndexStorageInvokeOptions
 export type EntityIndexStorageSaveOptions<V> = StoragePutOptions<string, V> &
@@ -67,7 +67,7 @@ export class EntityIndexStorage<
 
   constructor(
     protected options: EntityIndexStorageOptions<Entity>,
-    protected atomicOpMutex: Mutex = new Mutex(),
+    protected atomicOpMutex: Record<string, Mutex> = {},
     protected storage: LevelStorage<string | Entity> = new LevelStorage({
       ...options,
       path: path.join(options.path, options.name),
@@ -93,8 +93,8 @@ export class EntityIndexStorage<
     }
   }
 
-  async acquire(): Promise<() => void> {
-    return this.getAtomicOpMutex(true)
+  async acquire(id?: string): Promise<() => void> {
+    return this.getAtomicOpMutex(id || true)
   }
 
   async getCount(options?: EntityIndexStorageCallOptions): Promise<number> {
@@ -664,10 +664,39 @@ export class EntityIndexStorage<
     return item.value
   }
 
-  protected getAtomicOpMutex(
-    atomic = !!this.options.entityStore,
+  protected async getAtomicOpMutex(
+    atomic: string | boolean = !!this.options.entityStore,
   ): Promise<() => void> {
-    return atomic ? this.atomicOpMutex.acquire() : this.noopMutex
+    if (!atomic) return this.noopMutex
+
+    const mainMutex = this.getMutexById('main')
+    const mainPromise = mainMutex.acquire()
+    if (atomic === true) return mainPromise
+
+    const idMutex = this.getMutexById(atomic)
+    const idPromise = idMutex.acquire()
+    const promises = [mainPromise, idPromise]
+
+    // @note: Only wait for id mutex but lock and release both: main and id mutexs
+    await idPromise
+
+    return async () => {
+      const releaseAll = await Promise.all(promises)
+      for (const release of releaseAll) {
+        release()
+      }
+    }
+  }
+
+  protected getMutexById(id: string): Mutex {
+    let mutex = this.atomicOpMutex[id]
+
+    if (!mutex) {
+      mutex = new Mutex()
+      this.atomicOpMutex[id] = mutex
+    }
+
+    return mutex
   }
 
   protected async getCountDelta(

packages/core/src/storage/entityStorage.ts

@@ -1,7 +1,6 @@
 /* eslint-disable @typescript-eslint/no-empty-function */
 import path from 'node:path'
 
-import { Mutex } from '../utils/index.js'
 import { StorageGetOptions } from './baseStorage.js'
 import {
   EntityIndexStorage,
@@ -32,18 +31,17 @@ export type EntityStorageOptions<Entity> = EntityIndexStorageOptions<Entity> & {
   count?: boolean
 }
 
-export type EntityStorageInvokeOptions = { atomic?: boolean }
+export type EntityStorageInvokeOptions = { atomic?: string | boolean }
 export type EntityStorageCallOptions = EntityStorageInvokeOptions
 export type EntityStorageGetStreamOptions<K, V> = StorageGetOptions<K, V> &
   EntityStorageInvokeOptions
-
+export type EntityStorageSaveOptions = EntityStorageInvokeOptions
+export type EntityStorageRemoveOptions = EntityStorageInvokeOptions
 /**
  * Defines the storage handler class for different entities.
  */
 export class EntityStorage<Entity> extends EntityIndexStorage<Entity, Entity> {
   protected byIndex: Record<string, EntityIndexStorage<Entity, Entity>> = {}
-  protected atomicOpMutex: Mutex
-  protected noopMutex = Promise.resolve((): void => {})
 
   static AddressLength = EntityIndexStorage.AddressLength
   static TimestampLength = EntityIndexStorage.TimestampLength
@@ -54,15 +52,14 @@ export class EntityStorage<Entity> extends EntityIndexStorage<Entity, Entity> {
     protected options: EntityStorageOptions<Entity>,
     protected StorageClass: typeof LevelStorage = LevelStorage,
   ) {
-    const atomicOpMutex = new Mutex()
+    const atomicOpMutex = {}
     const basePath = path.join(options.path, options.name)
     const storage = new LevelStorage({ ...options, path: basePath })
 
     const opts = { ...options, sublevel: 'main' }
     super(opts, atomicOpMutex, storage)
 
     this.options = opts
-    this.atomicOpMutex = atomicOpMutex
 
     const { indexes } = this.options
     if (!indexes) return
@@ -79,7 +76,7 @@ export class EntityStorage<Entity> extends EntityIndexStorage<Entity, Entity> {
           entityStore: this,
           count: false,
         },
-        this.atomicOpMutex,
+        atomicOpMutex,
         storage,
       )
     }
@@ -96,8 +93,13 @@ export class EntityStorage<Entity> extends EntityIndexStorage<Entity, Entity> {
     )
   }
 
-  async save(entities: Entity | Entity[]): Promise<void> {
-    const release = await this.getAtomicOpMutex(true)
+  async save(
+    entities: Entity | Entity[],
+    options?: EntityStorageSaveOptions,
+  ): Promise<void> {
+    const atomicTop = options?.atomic !== undefined ? options.atomic : true
+    const release = await this.getAtomicOpMutex(atomicTop)
+
     const atomic = false
     const batch = this.getBatch()
 
@@ -144,8 +146,13 @@ export class EntityStorage<Entity> extends EntityIndexStorage<Entity, Entity> {
     }
   }
 
-  async remove(entities: Entity | Entity[]): Promise<void> {
-    const release = await this.getAtomicOpMutex(true)
+  async remove(
+    entities: Entity | Entity[],
+    options?: EntityStorageRemoveOptions,
+  ): Promise<void> {
+    const atomicTop = options?.atomic !== undefined ? options.atomic : true
+    const release = await this.getAtomicOpMutex(atomicTop)
+
     const atomic = false
     const batch = this.getBatch()
 

packages/framework/src/services/indexer/src/accountEntityIndexer.ts

@@ -184,7 +184,7 @@ export class BaseAccountEntityIndexer<T extends ParsedEntity<unknown>> {
 
     // @note: Update the state of the request to ready (mark for processing)
     // @note: Bulk write for boosting performance (more RAM comsumption)
-    await this.entityIndexerStateDAL.save(readyRanges)
+    await this.entityIndexerStateDAL.save(readyRanges, { atomic: account })
   }
 
   protected async onEntityResponse(request: EntityRequest): Promise<void> {
@@ -203,11 +203,14 @@ export class BaseAccountEntityIndexer<T extends ParsedEntity<unknown>> {
     if (!pendingRange) return
 
     // @note: Update the state of the request to ready (mark for processing)
-    await this.entityIndexerStateDAL.save({
-      ...pendingRange,
-      requestNonce,
-      state: Ready,
-    })
+    await this.entityIndexerStateDAL.save(
+      {
+        ...pendingRange,
+        requestNonce,
+        state: Ready,
+      },
+      { atomic: account },
+    )
   }
 
   protected async fetchAllRanges({
@@ -309,8 +312,8 @@ export class BaseAccountEntityIndexer<T extends ParsedEntity<unknown>> {
     // @note: Ordering is important for not causing
     // race conditions issues on pending ranges calculation due
     // to empty processed entries in db
-    await this.entityIndexerStateDAL.save(newStates)
-    await this.entityIndexerStateDAL.remove(oldStates)
+    await this.entityIndexerStateDAL.save(newStates, { atomic: account })
+    await this.entityIndexerStateDAL.remove(oldStates, { atomic: account })
 
     return mergedRanges
   }
@@ -349,11 +352,14 @@ export class BaseAccountEntityIndexer<T extends ParsedEntity<unknown>> {
       })
 
       // @note: Update the state of the request to processed (mark for compaction)
-      await this.entityIndexerStateDAL.save({
-        ...range,
-        requestNonce: undefined,
-        state: Processed,
-      })
+      await this.entityIndexerStateDAL.save(
+        {
+          ...range,
+          requestNonce: undefined,
+          state: Processed,
+        },
+        { atomic: account },
+      )
 
       // @note: Remove the request state on the entity fetcher
       await remove()
@@ -367,28 +373,35 @@ export class BaseAccountEntityIndexer<T extends ParsedEntity<unknown>> {
   }
 
   protected async fetchRangeByDate(dateRange: AccountDateRange): Promise<void> {
+    const { account } = this.config
     const { Pending, Ready } = EntityIndexerStateCode
 
     // @note: Do the request and get the nonce
     const nonce = await this.entityFetcher.fetchAccountEntitiesByDate(dateRange)
 
     // @note: Save the pending state of the request
-    await this.entityIndexerStateDAL.save({
-      ...dateRange,
-      requestNonce: nonce,
-      state: Pending,
-    })
+    await this.entityIndexerStateDAL.save(
+      {
+        ...dateRange,
+        requestNonce: nonce,
+        state: Pending,
+      },
+      { atomic: account },
+    )
 
     // @note: Wait till the request is complete
     await this.entityFetcher.awaitRequestComplete(nonce)
 
     // @note: Update the state to ready
     // (in some cases, the response comes before saving the pending state, so we must always check it here too)
-    await this.entityIndexerStateDAL.save({
-      ...dateRange,
-      requestNonce: nonce,
-      state: Ready,
-    })
+    await this.entityIndexerStateDAL.save(
+      {
+        ...dateRange,
+        requestNonce: nonce,
+        state: Ready,
+      },
+      { atomic: account },
+    )
   }
 
   protected async calculateRangesToFetch(

packages/indexer-example/src/domain/main.ts

@@ -48,7 +48,7 @@ export default class MainDomain
     if (this.context.supportedBlockchains.includes(BlockchainChain.Solana))
       accountIndexerConfigs.push({
         blockchainId: BlockchainChain.Solana,
-        account: 'DkxNXPCuJDpYA4rKKpqm1Jwv4tD9SNRcpEH2rf9juicd', // alephTokenSol,
+        account: alephTokenSol,
         meta: 2,
         index: {
           transactions: {