Merge pull request #1200 from input-output-hk/feat/projector-no-pg-transaction

Feat/projector no pg transaction

Author: AngelCastilloB

packages/cardano-services/src/Projection/createTypeormProjection.ts

@@ -3,7 +3,7 @@
 import { Bootstrap, ProjectionEvent, logProjectionProgress, requestNext } from '@cardano-sdk/projection';
 import { Cardano, ObservableCardanoNode } from '@cardano-sdk/core';
 import { Logger } from 'ts-log';
-import { Observable, concat, defer, take, takeWhile } from 'rxjs';
+import { Observable, concat, defer, groupBy, mergeMap, take, takeWhile } from 'rxjs';
 import {
   PgConnectionConfig,
   TypeormDevOptions,
@@ -72,7 +72,7 @@ export const createTypeormProjection = ({
   logger.debug(`Creating projection with policyIds ${JSON.stringify(handlePolicyIds)}`);
   logger.debug(`Using a ${blocksBufferLength} blocks buffer`);
 
-  const { mappers, entities, stores, extensions } = prepareTypeormProjection(
+  const { mappers, entities, stores, extensions, willStore } = prepareTypeormProjection(
     {
       options: projectionOptions,
       projections
@@ -118,15 +118,23 @@ export const createTypeormProjection = ({
     defer(() =>
       projectionSource$.pipe(
         applyMappers(mappers),
-        shareRetryBackoff(
-          (evt$) =>
-            evt$.pipe(
-              withTypeormTransaction({ connection$: connect() }),
-              applyStores(stores),
-              buffer.storeBlockData(),
-              typeormTransactionCommit()
-            ),
-          { shouldRetry: isRecoverableTypeormError }
+        // if there are any relevant data to write into db
+        groupBy((evt) => willStore(evt)),
+        mergeMap((group$) =>
+          group$.key
+            ? group$.pipe(
+                shareRetryBackoff(
+                  (evt$) =>
+                    evt$.pipe(
+                      withTypeormTransaction({ connection$: connect() }),
+                      applyStores(stores),
+                      buffer.storeBlockData(),
+                      typeormTransactionCommit()
+                    ),
+                  { shouldRetry: isRecoverableTypeormError }
+                )
+              )
+            : group$
         ),
         tipTracker.trackProjectedTip(),
         requestNext(),

packages/cardano-services/src/Projection/prepareTypeormProjection.ts

@@ -28,14 +28,24 @@ import {
   storeStakeKeyRegistrations,
   storeStakePoolRewardsJob,
   storeStakePools,
-  storeUtxo
+  storeUtxo,
+  willStoreAddresses,
+  willStoreAssets,
+  willStoreHandleMetadata,
+  willStoreHandles,
+  willStoreNftMetadata,
+  willStoreStakeKeyRegistrations,
+  willStoreStakePoolMetadataJob,
+  willStoreStakePoolRewardsJob,
+  willStoreStakePools,
+  willStoreUtxo
 } from '@cardano-sdk/projection-typeorm';
-import { Cardano } from '@cardano-sdk/core';
-import { Mappers as Mapper } from '@cardano-sdk/projection';
+import { Cardano, ChainSyncEventType } from '@cardano-sdk/core';
+import { Mappers as Mapper, ProjectionEvent } from '@cardano-sdk/projection';
+import { ObservableType, passthrough } from '@cardano-sdk/util-rxjs';
 import { POOLS_METRICS_INTERVAL_DEFAULT, POOLS_METRICS_OUTDATED_INTERVAL_DEFAULT } from '../Program/programs/types';
 import { Sorter } from '@hapi/topo';
-import { WithLogger } from '@cardano-sdk/util';
-import { passthrough } from '@cardano-sdk/util-rxjs';
+import { WithLogger, isNotNil } from '@cardano-sdk/util';
 
 /** Used as mount segments, so must be URL-friendly */
 export enum ProjectionName {
@@ -118,6 +128,25 @@ type StoreOperators = typeof storeOperators;
 type StoreName = keyof StoreOperators;
 type StoreOperator = StoreOperators[StoreName];
 
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+type InferArg0<F extends Function> = F extends (arg0: infer Args) => any ? Args : never;
+type WillStore = {
+  [k in keyof StoreOperators]: (args: ObservableType<InferArg0<StoreOperators[k]>>) => boolean;
+};
+
+const willStore: Partial<WillStore> = {
+  storeAddresses: willStoreAddresses,
+  storeAssets: willStoreAssets,
+  storeHandleMetadata: willStoreHandleMetadata,
+  storeHandles: willStoreHandles,
+  storeNftMetadata: willStoreNftMetadata,
+  storeStakeKeyRegistrations: willStoreStakeKeyRegistrations,
+  storeStakePoolMetadataJob: willStoreStakePoolMetadataJob,
+  storeStakePoolRewardsJob: willStoreStakePoolRewardsJob,
+  storeStakePools: willStoreStakePools,
+  storeUtxo: willStoreUtxo
+};
+
 const entities = {
   address: AddressEntity,
   asset: AssetEntity,
@@ -221,7 +250,7 @@ const storeMapperDependencies: Partial<Record<StoreName, MapperName[]>> = {
 };
 
 const storeInterDependencies: Partial<Record<StoreName, StoreName[]>> = {
-  storeAddresses: ['storeStakeKeyRegistrations'],
+  storeAddresses: ['storeBlock', 'storeStakeKeyRegistrations'],
   storeAssets: ['storeBlock'],
   storeHandleMetadata: ['storeUtxo'],
   storeHandles: ['storeUtxo', 'storeAddresses', 'storeHandleMetadata'],
@@ -338,16 +367,24 @@ export const prepareTypeormProjection = (
   const selectedMappers = mapperSorter.nodes;
   const selectedStores = storeSorter.nodes;
   const extensions = requiredExtensions(projections);
+  const willStoreCheckers = selectedStores
+    .map((store) => Object.entries(storeOperators).find(([_, operator]) => store === operator)!)
+    .map(([storeName]) => willStore[storeName as keyof StoreOperators])
+    .filter(isNotNil);
   return {
     __debug: {
       entities: selectedEntities.map((Entity) => keyOf(entities, Entity)),
       mappers: selectedMappers.map((mapper) => keyOf(mapperOperators, mapper)),
-      stores: selectedStores.map((store) => keyOf(storeOperators, store))
+      stores: selectedStores.map((store) => keyOf(storeOperators, store)),
+      willStoreCheckers: willStoreCheckers.map((checker) => checker.name)
     },
     entities: selectedEntities,
     extensions,
     mappers: selectedMappers,
-    stores: selectedStores
+    stores: selectedStores,
+    willStore: <T extends ProjectionEvent>(evt: T) =>
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      evt.eventType === ChainSyncEventType.RollBackward || willStoreCheckers.some((check) => check(evt as any))
   };
 };
 

packages/cardano-services/test/Projection/createTypeormProjection.test.ts

@@ -1,4 +1,11 @@
-import { AssetEntity, OutputEntity, TokensEntity, createDataSource } from '@cardano-sdk/projection-typeorm';
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import {
+  AssetEntity,
+  BlockEntity,
+  OutputEntity,
+  TokensEntity,
+  createDataSource
+} from '@cardano-sdk/projection-typeorm';
 import { ChainSyncDataSet, chainSyncData, logger } from '@cardano-sdk/util-dev';
 import { ProjectionName, createTypeormProjection, prepareTypeormProjection } from '../../src';
 import { lastValueFrom } from 'rxjs';
@@ -41,5 +48,48 @@ describe('createTypeormProjection', () => {
     await dataSource.destroy();
   });
 
+  it('only store blocks which have relevant information to the configured stores', async () => {
+    // Setup projector
+    const projections = [ProjectionName.Asset];
+    const data = chainSyncData(ChainSyncDataSet.WithMint);
+
+    const emptyBlocksHashes = data.allEvents
+      .filter((evt) => (evt as any).block && (evt as any).block.body.length === 0)
+      .map((evt) => (evt as any).block.header.hash);
+
+    const projection$ = createTypeormProjection({
+      blocksBufferLength: 10,
+      cardanoNode: data.cardanoNode,
+      connectionConfig$: projectorConnectionConfig$,
+      devOptions: { dropSchema: true },
+      logger,
+      projections
+    });
+
+    // Project
+    await lastValueFrom(projection$);
+
+    // Setup query runner for assertions
+    const { entities } = prepareTypeormProjection({ projections }, { logger });
+    const dataSource = createDataSource({
+      connectionConfig: projectorConnectionConfig,
+      entities,
+      logger
+    });
+    await dataSource.initialize();
+    const queryRunner = dataSource.createQueryRunner();
+    await queryRunner.connect();
+
+    // Check data in the database
+    for (const emptyBlockHash of emptyBlocksHashes)
+      expect(
+        await queryRunner.manager.getRepository(BlockEntity).findOne({ where: { hash: emptyBlockHash } })
+      ).toBeNull();
+
+    expect.hasAssertions();
+
+    await queryRunner.release();
+    await dataSource.destroy();
+  });
   // PostgreSQL transaction retries are tested in projection-typeorm package
 });

packages/cardano-services/test/Projection/prepareTypeormProjection.test.ts

@@ -5,16 +5,20 @@ const prepare = (projections: ProjectionName[]) =>
   prepareTypeormProjection({ projections }, { logger: dummyLogger }).__debug;
 
 describe('prepareTypeormProjection', () => {
-  describe('computes required entities, mappers and stores based on selected projections and presence of a buffer', () => {
+  describe('computes required entities, mappers, predicates, and stores based on selected projections and presence of a buffer', () => {
     test('utxo', () => {
-      const { entities, mappers, stores } = prepare([ProjectionName.UTXO]);
+      const { entities, mappers, stores, willStoreCheckers } = prepare([ProjectionName.UTXO]);
       expect(new Set(entities)).toEqual(new Set(['tokens', 'block', 'asset', 'nftMetadata', 'output', 'blockData']));
       expect(mappers).toEqual(['withMint', 'withUtxo']);
       expect(stores).toEqual(['storeBlock', 'storeAssets', 'storeUtxo']);
+      expect(willStoreCheckers).toEqual(['willStoreAssets', 'willStoreUtxo']);
     });
 
     test('stake-pool,stake-pool-metadata', () => {
-      const { entities, mappers, stores } = prepare([ProjectionName.StakePool, ProjectionName.StakePoolMetadataJob]);
+      const { entities, mappers, stores, willStoreCheckers } = prepare([
+        ProjectionName.StakePool,
+        ProjectionName.StakePoolMetadataJob
+      ]);
       expect(new Set(entities)).toEqual(
         new Set([
           'block',
@@ -29,6 +33,7 @@ describe('prepareTypeormProjection', () => {
       );
       expect(mappers).toEqual(['withCertificates', 'withStakePools']);
       expect(stores).toEqual(['storeBlock', 'storeStakePools', 'storeStakePoolMetadataJob']);
+      expect(willStoreCheckers).toEqual(['willStoreStakePools', 'willStoreStakePoolMetadataJob']);
     });
   });
 });

packages/projection-typeorm/src/operators/storeAddresses.ts

@@ -16,6 +16,8 @@ const lookupStakeKeyRegistration = async (pointer: Cardano.Pointer | undefined,
   return Hash28ByteBase16.fromEd25519KeyHashHex(stakeKeyRegistration.stakeKeyHash);
 };
 
+export const willStoreAddresses = ({ addresses }: Mappers.WithAddresses) => addresses.length > 0;
+
 export const storeAddresses = typeormOperator<Mappers.WithAddresses>(async (evt) => {
   const { addresses, eventType, queryRunner } = evt;
   if (addresses.length === 0 || eventType !== ChainSyncEventType.RollForward) return;

packages/projection-typeorm/src/operators/storeAssets.ts

@@ -53,6 +53,8 @@ const rollBackward = async ({ mint, queryRunner }: StoreAssetEventParams): Promi
   return mintedAssetTotalSupplies;
 };
 
+export const willStoreAssets = ({ mint }: Mappers.WithMint) => mint.length > 0;
+
 export const storeAssets = typeormOperator<Mappers.WithMint, WithMintedAssetSupplies>(
   async ({ mint, block: { header }, eventType, queryRunner }) => {
     const storeAssetEventParams: StoreAssetEventParams = { header, mint, queryRunner };

packages/projection-typeorm/src/operators/storeHandleMetadata.ts

@@ -4,6 +4,8 @@ import { Mappers } from '@cardano-sdk/projection';
 import { WithStoredProducedUtxo } from './storeUtxo';
 import { typeormOperator } from './util';
 
+export const willStoreHandleMetadata = ({ handleMetadata }: Mappers.WithHandleMetadata) => handleMetadata.length > 0;
+
 export const storeHandleMetadata = typeormOperator<Mappers.WithHandleMetadata & WithStoredProducedUtxo>(
   async ({
     eventType,

packages/projection-typeorm/src/operators/storeHandles.ts

@@ -230,6 +230,8 @@ const withTotalSupplies = (
     )
   );
 
+export const willStoreHandles = ({ handles }: Mappers.WithHandles) => handles.length > 0;
+
 export const storeHandles = typeormOperator<Mappers.WithHandles & WithMintedAssetSupplies & Mappers.WithHandleMetadata>(
   async ({ handles, queryRunner, eventType, block, mintedAssetTotalSupplies, handleMetadata }) => {
     const handleEventParams: HandleEventParams = {

packages/projection-typeorm/src/operators/storeNftMetadata.ts

@@ -104,6 +104,9 @@ const handleRollBackwardEvent = async (
   }
 };
 
+export const willStoreNftMetadata = ({ nftMetadata, cip67 }: Mappers.WithCIP67 & Mappers.WithNftMetadata) =>
+  nftMetadata.length > 0 || Object.keys(cip67.byLabel).length > 0;
+
 export const storeNftMetadata = typeormOperator<Mappers.WithCIP67 & Mappers.WithMint & Mappers.WithNftMetadata>(
   async (evt) => {
     const nftMetadataRepository = evt.queryRunner.manager.getRepository(NftMetadataEntity);

packages/projection-typeorm/src/operators/storePoolMetricsUpdateJob.ts

@@ -5,11 +5,13 @@ import { typeormOperator } from './util';
 
 export const createStorePoolMetricsUpdateJob = (jobFrequency = 1000, jobOutdatedFrequency?: number) => {
   // Remember the blockNo of last sent job in order to no resend another job in case of rollback
-  let lastSentBlock: Cardano.BlockNo | undefined;
+  let lastAllSentBlock: Cardano.BlockNo | undefined;
+  let lastOutdatedSentBlock: Cardano.BlockNo | undefined;
   // Metrics updated before this slot is considered outdated
   let outdatedSlot: Cardano.Slot;
   let reachedTheTip = false;
 
+  // eslint-disable-next-line complexity
   return typeormOperator<WithPgBoss>(async ({ eventType, pgBoss, block: { header }, tip }) => {
     let insertFirstJob = false;
 
@@ -20,23 +22,31 @@ export const createStorePoolMetricsUpdateJob = (jobFrequency = 1000, jobOutdated
 
     const sendForAll = async () => {
       // run the update for all pools
-      lastSentBlock = blockNo;
+      lastAllSentBlock = blockNo;
+
+      if (lastOutdatedSentBlock === undefined) lastOutdatedSentBlock = blockNo;
+
       outdatedSlot = slot;
       await pgBoss.send(STAKE_POOL_METRICS_UPDATE, { slot }, { slot });
     };
 
     const sendForOutdated = async () => {
       // run the update for only pools with outdated metrics
-      lastSentBlock = blockNo;
+      lastOutdatedSentBlock = blockNo;
       await pgBoss.send(STAKE_POOL_METRICS_UPDATE, { outdatedSlot, slot }, { slot });
     };
 
     if (insertFirstJob) {
       await sendForAll();
-    } else if (blockNo !== lastSentBlock && reachedTheTip) {
-      if (blockNo % jobFrequency === 0) {
+    } else if ((blockNo !== lastAllSentBlock || blockNo !== lastOutdatedSentBlock) && reachedTheTip) {
+      if (lastAllSentBlock && blockNo - lastAllSentBlock >= jobFrequency) {
         await sendForAll();
-      } else if (jobOutdatedFrequency && outdatedSlot && blockNo % jobOutdatedFrequency === 0) {
+      } else if (
+        jobOutdatedFrequency &&
+        lastOutdatedSentBlock &&
+        outdatedSlot &&
+        blockNo - lastOutdatedSentBlock >= jobOutdatedFrequency
+      ) {
         await sendForOutdated();
       }
     }