feat(backend): performance and caching poc

localenv/telemetry/otel-collector-config.yaml

@@ -8,6 +8,8 @@ processors:
   batch:
 
 exporters:
+  logging:
+    loglevel: info
   debug:
     verbosity: detailed
   prometheus:
@@ -18,6 +20,9 @@ exporters:
       insecure: true
 
 service:
+  telemetry:
+    logs:
+      level: warn
   pipelines:
     metrics:
       receivers: [otlp]

packages/backend/src/accounting/psql/balance.ts

@@ -14,6 +14,9 @@ export async function getAccountBalances(
   account: LedgerAccount,
   trx?: TransactionOrKnex
 ): Promise<AccountBalance> {
+  const stopTimer = deps.telemetry?.startTimer('psql_getAccountBalances', {
+    callName: 'psql_getAccountBalances'
+  })
   try {
     const queryResult = await (trx ?? deps.knex).raw(
       `
@@ -38,6 +41,8 @@ export async function getAccountBalances(
     const debitsPosted = BigInt(queryResult.rows[0].debitsPosted)
     const debitsPending = BigInt(queryResult.rows[0].debitsPending)
 
+    stopTimer && stopTimer()
+
     return {
       creditsPosted,
       creditsPending,

packages/backend/src/accounting/psql/service.ts

@@ -34,10 +34,14 @@ import {
   getAccountTransfers
 } from './ledger-transfer'
 import { LedgerTransfer, LedgerTransferType } from './ledger-transfer/model'
+import { TelemetryService } from '../../telemetry/service'
+import { AssetService } from '../../asset/service'
 
 export interface ServiceDependencies extends BaseService {
+  fetchAssetService?: () => Promise<AssetService>
   knex: TransactionOrKnex
   withdrawalThrottleDelay?: number
+  telemetry?: TelemetryService
 }
 
 export function createAccountingService(
@@ -145,13 +149,21 @@ export async function getAccountTotalSent(
   deps: ServiceDependencies,
   accountRef: string
 ): Promise<bigint | undefined> {
+  const stopTimer = deps.telemetry?.startTimer('psql_getAccountTotalSent', {
+    callName: 'psql_getAccountTotalSent'
+  })
   const account = await getLiquidityAccount(deps, accountRef)
 
   if (!account) {
+    stopTimer && stopTimer()
     return
   }
 
-  return (await getAccountBalances(deps, account)).debitsPosted
+  const totalsSent = (await getAccountBalances(deps, account)).debitsPosted
+
+  stopTimer && stopTimer()
+
+  return totalsSent
 }
 
 export async function getAccountsTotalSent(

packages/backend/src/accounting/service.ts

@@ -1,6 +1,7 @@
 import { TransactionOrKnex } from 'objection'
 import { BaseService } from '../shared/baseService'
 import { TransferError, isTransferError } from './errors'
+import { AssetService } from '../asset/service'
 
 export enum LiquidityAccountType {
   ASSET = 'ASSET',
@@ -40,6 +41,8 @@ export interface LiquidityAccount {
 export interface OnCreditOptions {
   totalReceived: bigint
   withdrawalThrottleDelay?: number
+  // TODO: make this required. should only need updating the tests to do this.
+  fetchAssetService?: () => Promise<AssetService>
 }
 
 export interface OnDebitOptions {
@@ -133,6 +136,7 @@ export interface TransferToCreate {
 }
 
 export interface BaseAccountingServiceDependencies extends BaseService {
+  fetchAssetService?: () => Promise<AssetService>
   withdrawalThrottleDelay?: number
 }
 
@@ -160,7 +164,7 @@ export async function createAccountToAccountTransfer(
     getAccountBalance
   } = args
 
-  const { withdrawalThrottleDelay } = deps
+  const { withdrawalThrottleDelay, fetchAssetService } = deps
 
   const { sourceAccount, destinationAccount, sourceAmount, destinationAmount } =
     transferArgs
@@ -226,7 +230,8 @@ export async function createAccountToAccountTransfer(
 
         await destinationAccount.onCredit({
           totalReceived,
-          withdrawalThrottleDelay
+          withdrawalThrottleDelay,
+          fetchAssetService
         })
       }
     },

packages/backend/src/accounting/tigerbeetle/service.ts

@@ -31,6 +31,8 @@ import {
   getAccountTransfers
 } from './transfers'
 import { toTigerBeetleId } from './utils'
+import { TelemetryService } from '../../telemetry/service'
+import { AssetService } from '../../asset/service'
 
 export enum TigerBeetleAccountCode {
   LIQUIDITY_WEB_MONETIZATION = 1,
@@ -67,8 +69,10 @@ export const convertToTigerBeetleTransferCode: {
 }
 
 export interface ServiceDependencies extends BaseService {
+  fetchAssetService?: () => Promise<AssetService>
   tigerBeetle: Client
   withdrawalThrottleDelay?: number
+  telemetry?: TelemetryService
 }
 
 export function createAccountingService(
@@ -218,10 +222,15 @@ export async function getAccountTotalSent(
   deps: ServiceDependencies,
   id: string
 ): Promise<bigint | undefined> {
+  const stopTimer = deps.telemetry?.startTimer('tb_getAccountTotalSent', {
+    callName: 'tb_getAccountTotalSent'
+  })
   const account = (await getAccounts(deps, [id]))[0]
   if (account) {
+    stopTimer && stopTimer()
     return account.debits_posted
   }
+  stopTimer && stopTimer()
 }
 
 export async function getAccountsTotalSent(

packages/backend/src/asset/service.ts

@@ -7,6 +7,9 @@ import { BaseService } from '../shared/baseService'
 import { AccountingService, LiquidityAccountType } from '../accounting/service'
 import { WalletAddress } from '../open_payments/wallet_address/model'
 import { Peer } from '../payment-method/ilp/peer/model'
+import { Quote } from '../open_payments/quote/model'
+import { IncomingPayment } from '../open_payments/payment/incoming/model'
+import { CacheDataStore } from '../cache'
 
 export interface AssetOptions {
   code: string
@@ -28,37 +31,46 @@ export interface DeleteOptions {
   deletedAt: Date
 }
 
+export type ToSetOn = Quote | IncomingPayment | WalletAddress | Peer | undefined
+
 export interface AssetService {
   create(options: CreateOptions): Promise<Asset | AssetError>
   update(options: UpdateOptions): Promise<Asset | AssetError>
   delete(options: DeleteOptions): Promise<Asset | AssetError>
   get(id: string): Promise<void | Asset>
+  setOn(obj: ToSetOn): Promise<void | Asset>
   getPage(pagination?: Pagination, sortOrder?: SortOrder): Promise<Asset[]>
   getAll(): Promise<Asset[]>
 }
 
 interface ServiceDependencies extends BaseService {
   accountingService: AccountingService
+  cacheDataStore: CacheDataStore
 }
 
 export async function createAssetService({
   logger,
   knex,
-  accountingService
+  accountingService,
+  cacheDataStore
 }: ServiceDependencies): Promise<AssetService> {
   const log = logger.child({
     service: 'AssetService'
   })
+
   const deps: ServiceDependencies = {
     logger: log,
     knex,
-    accountingService
+    accountingService,
+    cacheDataStore
   }
+
   return {
     create: (options) => createAsset(deps, options),
     update: (options) => updateAsset(deps, options),
     delete: (options) => deleteAsset(deps, options),
     get: (id) => getAsset(deps, id),
+    setOn: (toSetOn) => setAssetOn(deps, toSetOn),
     getPage: (pagination?, sortOrder?) =>
       getAssetsPage(deps, pagination, sortOrder),
     getAll: () => getAll(deps)
@@ -78,6 +90,8 @@ async function createAsset(
       .first()
 
     if (deletedAsset) {
+      await deps.cacheDataStore.delete(deletedAsset.id)
+
       // if found, enable
       return await Asset.query(deps.knex)
         .patchAndFetchById(deletedAsset.id, { deletedAt: null })
@@ -120,9 +134,12 @@ async function updateAsset(
     throw new Error('Knex undefined')
   }
   try {
-    return await Asset.query(deps.knex)
+    const returnVal = await Asset.query(deps.knex)
       .patchAndFetchById(id, { withdrawalThreshold, liquidityThreshold })
       .throwIfNotFound()
+
+    await deps.cacheDataStore.delete(id)
+    return returnVal
   } catch (err) {
     if (err instanceof NotFoundError) {
       return AssetError.UnknownAsset
@@ -139,6 +156,8 @@ async function deleteAsset(
   if (!deps.knex) {
     throw new Error('Knex undefined')
   }
+
+  await deps.cacheDataStore.delete(id)
   try {
     // return error in case there is a peer or wallet address using the asset
     const peer = await Peer.query(deps.knex).where('assetId', id).first()
@@ -152,7 +171,6 @@ async function deleteAsset(
     if (walletAddress) {
       return AssetError.CannotDeleteInUseAsset
     }
-
     return await Asset.query(deps.knex)
       .patchAndFetchById(id, { deletedAt: deletedAt.toISOString() })
       .throwIfNotFound()
@@ -168,19 +186,35 @@ async function getAsset(
   deps: ServiceDependencies,
   id: string
 ): Promise<void | Asset> {
-  return await Asset.query(deps.knex).whereNull('deletedAt').findById(id)
+  const inMem = (await deps.cacheDataStore.get(id)) as Asset
+  if (inMem) return inMem
+
+  const asset = await Asset.query(deps.knex).whereNull('deletedAt').findById(id)
+  if (asset) await deps.cacheDataStore.set(asset.id, asset)
+
+  return asset
 }
 
 async function getAssetsPage(
   deps: ServiceDependencies,
   pagination?: Pagination,
   sortOrder?: SortOrder
 ): Promise<Asset[]> {
-  return await Asset.query(deps.knex)
+  return Asset.query(deps.knex)
     .whereNull('deletedAt')
     .getPage(pagination, sortOrder)
 }
 
 async function getAll(deps: ServiceDependencies): Promise<Asset[]> {
-  return await Asset.query(deps.knex).whereNull('deletedAt')
+  return Asset.query(deps.knex).whereNull('deletedAt')
+}
+
+async function setAssetOn(
+  deps: ServiceDependencies,
+  obj: ToSetOn
+): Promise<void | Asset> {
+  if (!obj) return
+  const asset = await getAsset(deps, obj.assetId)
+  if (asset) obj.asset = asset
+  return asset
 }

packages/backend/src/cache/cache.ts

@@ -0,0 +1,46 @@
+import { CacheDataStore, CacheSupport } from './index'
+
+export interface Cached {
+  expiry: number
+  data: CacheSupport
+}
+
+export function createInMemoryDataStore(keyTtlMs: number): CacheDataStore {
+  const map = new Map<string, Cached>()
+
+  const getAndCheckExpiry = (key: string): Cached | undefined => {
+    const cached = map.get(key)
+    if (cached?.expiry && Date.now() >= cached.expiry) {
+      deleteKey(key)
+      return undefined
+    }
+    return cached
+  }
+
+  const deleteKey = (key: string) => map.delete(key)
+
+  return {
+    async get(key: string): Promise<CacheSupport> {
+      if (keyTtlMs < 1) return undefined
+      const cached = getAndCheckExpiry(key)
+      return cached?.data
+    },
+    async getKeyExpiry(key: string): Promise<Date | undefined> {
+      const cached = getAndCheckExpiry(key)
+
+      return cached ? new Date(cached.expiry) : undefined
+    },
+    async delete(key: string): Promise<void> {
+      deleteKey(key)
+    },
+    async set(key: string, value: CacheSupport): Promise<boolean> {
+      if (keyTtlMs < 1) return true
+      map.set(key, { expiry: Date.now() + keyTtlMs, data: value })
+      return true
+    },
+    async deleteAll(): Promise<void> {
+      map.clear()
+    }
+  }
+}
+export { CacheDataStore }

packages/backend/src/cache/index.ts

@@ -0,0 +1,19 @@
+import { Quote } from '../open_payments/quote/model'
+import { WalletAddress } from '../open_payments/wallet_address/model'
+import { OutgoingPayment } from '../open_payments/payment/outgoing/model'
+import { Asset } from '../asset/model'
+
+export type CacheSupport =
+  | Quote
+  | WalletAddress
+  | OutgoingPayment
+  | Asset
+  | undefined
+
+export interface CacheDataStore {
+  get(key: string): Promise<CacheSupport>
+  getKeyExpiry(key: string): Promise<Date | undefined>
+  set(key: string, value: CacheSupport): Promise<boolean>
+  delete(key: string): Promise<void>
+  deleteAll(): Promise<void>
+}

packages/backend/src/config/app.ts

@@ -127,7 +127,7 @@ export const Config = {
   authServerGrantUrl: envString('AUTH_SERVER_GRANT_URL'),
   authServerIntrospectionUrl: envString('AUTH_SERVER_INTROSPECTION_URL'),
 
-  outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 4),
+  outgoingPaymentWorkers: envInt('OUTGOING_PAYMENT_WORKERS', 1),
   outgoingPaymentWorkerIdle: envInt('OUTGOING_PAYMENT_WORKER_IDLE', 200), // milliseconds
 
   incomingPaymentWorkers: envInt('INCOMING_PAYMENT_WORKERS', 1),
@@ -187,7 +187,8 @@ export const Config = {
     'INCOMING_PAYMENT_EXPIRY_MAX_MS',
     2592000000
   ), // 30 days
-  enableSpspPaymentPointers: envBool('ENABLE_SPSP_PAYMENT_POINTERS', true)
+  enableSpspPaymentPointers: envBool('ENABLE_SPSP_PAYMENT_POINTERS', true),
+  localCacheDuration: envInt('LOCAL_CACHE_DURATION', 5000)
 }
 
 function parseRedisTlsConfig(

packages/backend/src/fee/service.ts

@@ -66,6 +66,7 @@ async function getLatestFee(
   type: FeeType
 ): Promise<Fee | undefined> {
   return await Fee.query(deps.knex)
+    // TODO: index on assetId/type
     .where({ assetId, type })
     .orderBy('createdAt', 'desc')
     .first()

packages/backend/src/graphql/resolvers/liquidity.test.ts

@@ -53,7 +53,10 @@ describe('Liquidity Resolvers', (): void => {
   const timeoutTwoPhase = 10
 
   beforeAll(async (): Promise<void> => {
-    deps = await initIocContainer(Config)
+    deps = initIocContainer({
+      ...Config,
+      localCacheDuration: 0
+    })
     appContainer = await createTestApp(deps)
     knex = appContainer.knex
     accountingService = await deps.use('accountingService')