feat: add a buffer after reading blocks from ogmios

input-output-hk · Aug 14, 2023 · 0095c80 · 0095c80
1 parent 0c988e3
commit 0095c80
Show file tree

Hide file tree

Showing 20 changed files with 109 additions and 60 deletions.
diff --git a/compose/common.yml b/compose/common.yml
@@ -46,6 +46,7 @@ x-with-postgres: &with-postgres
 
 x-projector-environment: &projector-environment
   API_URL: http://0.0.0.0:3000
+  BLOCKS_BUFFER_LENGTH: ${BLOCKS_BUFFER_LENGTH:-10}
   DROP_SCHEMA: ${DROP_PROJECTOR_SCHEMA:-false}
   POSTGRES_HOST: postgres
   POSTGRES_POOL_MAX: ${POSTGRES_POOL_MAX:-10}

diff --git a/packages/cardano-services/.env.test b/packages/cardano-services/.env.test
@@ -1,5 +1,6 @@
 POSTGRES_CONNECTION_STRING_DB_SYNC=postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5433/cexplorer
 POSTGRES_CONNECTION_STRING_HANDLE=postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5433/handle
+POSTGRES_CONNECTION_STRING_PROJECTION=postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5433/projection
 POSTGRES_CONNECTION_STRING_STAKE_POOL=postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5433/stake_pool
 CARDANO_NODE_CONFIG_PATH=./config/network/testnet/cardano-node/config.json
 DB_CACHE_TTL=120

diff --git a/packages/cardano-services/src/Program/programs/projector.ts b/packages/cardano-services/src/Program/programs/projector.ts
@@ -17,12 +17,14 @@ import { URL } from 'url';
 import { createLogger } from 'bunyan';
 import { getConnectionConfig, getOgmiosObservableCardanoNode } from '../services';
 
+export const BLOCKS_BUFFER_LENGTH_DEFAULT = 10;
 export const PROJECTOR_API_URL_DEFAULT = new URL('http://localhost:3002');
 
 export type ProjectorArgs = CommonProgramOptions &
   PosgresProgramOptions<''> &
   HandlePolicyIdsProgramOptions &
   OgmiosProgramOptions & {
+    blocksBufferLength: number;
     dropSchema: boolean;
     dryRun: boolean;
     exitAtBlockNo: Cardano.BlockNo;
@@ -50,8 +52,9 @@ const createProjectionHttpService = async (options: ProjectionMapFactoryOptions)
   });
   const connectionConfig$ = getConnectionConfig(dnsResolver, 'projector', '', args);
   const buffer = new TypeormStabilityWindowBuffer({ logger });
-  const { dropSchema, dryRun, exitAtBlockNo, projectionNames, synchronize, handlePolicyIds } = args;
+  const { blocksBufferLength, dropSchema, dryRun, exitAtBlockNo, handlePolicyIds, projectionNames, synchronize } = args;
   const projection$ = createTypeormProjection({
+    blocksBufferLength,
     buffer,
     connectionConfig$,
     devOptions: { dropSchema, synchronize },
@@ -61,6 +64,7 @@ const createProjectionHttpService = async (options: ProjectionMapFactoryOptions)
       handlePolicyIds
     },
     projectionSource$: Bootstrap.fromCardanoNode({
+      blocksBufferLength,
       buffer,
       cardanoNode,
       logger

diff --git a/packages/cardano-services/src/Program/programs/types.ts b/packages/cardano-services/src/Program/programs/types.ts
@@ -33,6 +33,7 @@ export enum ServiceNames {
 export const POOLS_METRICS_INTERVAL_DEFAULT = 1000;
 
 export enum ProjectorOptionDescriptions {
+  BlocksBufferLength = 'Chain sync event (blocks) buffer length',
   DropSchema = 'Drop and recreate database schema to project from origin',
   DryRun = 'Initialize the projection, but do not start it',
   ExitAtBlockNo = 'Exit after processing this block. Intended for benchmark testing',

diff --git a/packages/cardano-services/src/Projection/createTypeormProjection.ts b/packages/cardano-services/src/Projection/createTypeormProjection.ts
@@ -25,6 +25,7 @@ import { passthrough, shareRetryBackoff } from '@cardano-sdk/util-rxjs';
 
 export interface CreateTypeormProjectionProps {
   projections: ProjectionName[];
+  blocksBufferLength: number;
   buffer?: TypeormStabilityWindowBuffer;
   projectionSource$: Observable<ProjectionEvent>;
   connectionConfig$: Observable<PgConnectionConfig>;
@@ -51,7 +52,10 @@ export const createObservableDataSource = ({
   entities,
   extensions,
   migrationsRun
-}: Omit<CreateTypeormProjectionProps, 'exitAtBlockNo' | 'projections' | 'projectionSource$' | 'projectionOptions'> &
+}: Omit<
+  CreateTypeormProjectionProps,
+  'blocksBufferLength' | 'exitAtBlockNo' | 'projections' | 'projectionSource$' | 'projectionOptions'
+> &
   Pick<PreparedProjection, 'entities' | 'extensions'> & { migrationsRun: boolean }) =>
   connectionConfig$.pipe(
     switchMap((connectionConfig) =>
@@ -89,6 +93,7 @@ export const createObservableDataSource = ({
  * Dependencies of each projection are defined in ./prepareTypeormProjection.ts
  */
 export const createTypeormProjection = ({
+  blocksBufferLength,
   projections,
   projectionSource$,
   connectionConfig$,
@@ -101,6 +106,7 @@ export const createTypeormProjection = ({
   const { handlePolicyIds } = { handlePolicyIds: [], ...projectionOptions };
 
   logger.debug(`Creating projection with policyIds ${JSON.stringify(handlePolicyIds)}`);
+  logger.debug(`Using a ${blocksBufferLength} blocks buffer`);
 
   const { mappers, entities, stores, extensions } = prepareTypeormProjection({
     buffer,

diff --git a/packages/cardano-services/src/cli.ts b/packages/cardano-services/src/cli.ts
@@ -42,6 +42,12 @@ import {
   loadProviderServer,
   stringOptionToBoolean
 } from './Program';
+import {
+  BLOCKS_BUFFER_LENGTH_DEFAULT,
+  PROJECTOR_API_URL_DEFAULT,
+  ProjectorArgs,
+  loadProjector
+} from './Program/programs/projector';
 import { Command, Option } from 'commander';
 import { DB_CACHE_TTL_DEFAULT } from './InMemoryCache';
 import {
@@ -58,7 +64,6 @@ import {
   PgBossWorkerOptionDescriptions,
   loadPgBossWorker
 } from './Program/programs/pgBossWorker';
-import { PROJECTOR_API_URL_DEFAULT, ProjectorArgs, loadProjector } from './Program/programs/projector';
 import { PgBossQueue, isValidQueue } from './PgBoss';
 import { ProjectionName } from './Projection';
 import { URL } from 'url';
@@ -124,6 +129,12 @@ withCommonOptions(
   ),
   { apiUrl: PROJECTOR_API_URL_DEFAULT }
 )
+  .addOption(
+    new Option('--blocks-buffer-length <blocksBufferLength>', ProjectorOptionDescriptions.BlocksBufferLength)
+      .default(BLOCKS_BUFFER_LENGTH_DEFAULT)
+      .env('BLOCKS_BUFFER_LENGTH')
+      .argParser((blocksBufferLength) => Number.parseInt(blocksBufferLength, 10))
+  )
   .addOption(
     new Option('--drop-schema <true/false>', ProjectorOptionDescriptions.DropSchema)
       .default(false)

diff --git a/packages/cardano-services/test/Projection/createTypeormProjection.test.ts b/packages/cardano-services/test/Projection/createTypeormProjection.test.ts
@@ -18,11 +18,13 @@ describe('createTypeormProjection', () => {
     const buffer = new TypeormStabilityWindowBuffer({ allowNonSequentialBlockHeights: true, logger });
     const projections = [ProjectionName.UTXO];
     const projection$ = createTypeormProjection({
+      blocksBufferLength: 10,
       buffer,
       connectionConfig$: projectorConnectionConfig$,
       devOptions: { dropSchema: true, synchronize: true },
       logger,
       projectionSource$: Bootstrap.fromCardanoNode({
+        blocksBufferLength: 10,
         buffer,
         cardanoNode: data.cardanoNode,
         logger

diff --git a/packages/cardano-services/test/cli.test.ts b/packages/cardano-services/test/cli.test.ts
@@ -177,11 +177,13 @@ describe('CLI', () => {
   let lastBlock: LedgerTipModel;
   let postgresConnectionString: string;
   let postgresConnectionStringHandle: string;
+  let postgresConnectionStringProjection: string;
   let postgresConnectionStringStakePool: string;
 
   beforeAll(() => {
     postgresConnectionString = process.env.POSTGRES_CONNECTION_STRING_DB_SYNC!;
     postgresConnectionStringHandle = process.env.POSTGRES_CONNECTION_STRING_HANDLE!;
+    postgresConnectionStringProjection = process.env.POSTGRES_CONNECTION_STRING_PROJECTION!;
     postgresConnectionStringStakePool = process.env.POSTGRES_CONNECTION_STRING_STAKE_POOL!;
   });
 
@@ -2176,7 +2178,7 @@ describe('CLI', () => {
             '--ogmios-url',
             'ws://localhost:1234',
             '--postgres-connection-string',
-            'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+            postgresConnectionStringProjection,
             ProjectionName.UTXO
           ]);
           await assertServerAlive();
@@ -2187,7 +2189,7 @@ describe('CLI', () => {
             '--ogmios-url',
             'ws://localhost:1234',
             '--postgres-connection-string',
-            'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+            postgresConnectionStringProjection,
             `${ProjectionName.UTXO},${ProjectionName.StakePool}`
           ]);
           await assertServerAlive();
@@ -2200,7 +2202,7 @@ describe('CLI', () => {
             '--ogmios-url',
             'ws://localhost:1234',
             '--postgres-connection-string',
-            'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection'
+            postgresConnectionStringProjection
           ]);
           await assertServerAlive();
         });
@@ -2210,7 +2212,7 @@ describe('CLI', () => {
             '--ogmios-url',
             'ws://localhost:1234',
             '--postgres-connection-string',
-            'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+            postgresConnectionStringProjection,
             '--drop-schema',
             'true',
             ProjectionName.UTXO
@@ -2243,7 +2245,7 @@ describe('CLI', () => {
           '--ogmios-url',
           'ws://localhost:1234',
           '--postgres-connection-string',
-          'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+          postgresConnectionStringProjection,
           '--handle-policy-ids',
           'f0ff48bbb7bbe9d59a40f1ce90e9e9d0ff5002ec48f232b49ca0fb9a',
           ProjectionName.Handle
@@ -2258,7 +2260,7 @@ describe('CLI', () => {
           '--ogmios-url',
           'ws://localhost:1234',
           '--postgres-connection-string',
-          'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+          postgresConnectionStringProjection,
           '--logger-min-severity',
           'debug',
           '--handle-policy-ids-file',
@@ -2285,7 +2287,7 @@ describe('CLI', () => {
               '--ogmios-url',
               'ws://localhost:1234',
               '--postgres-connection-string',
-              'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+              postgresConnectionStringProjection,
               ProjectionName.Handle
             ],
             { env: {}, stdio: 'pipe' }
@@ -2313,7 +2315,7 @@ describe('CLI', () => {
               '--ogmios-url',
               'ws://localhost:1234',
               '--postgres-connection-string',
-              'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+              postgresConnectionStringProjection,
               '--handle-policy-ids',
               'policyId',
               ProjectionName.Handle
@@ -2330,6 +2332,28 @@ describe('CLI', () => {
           done();
         });
       });
+
+      test('uses the configured blocks buffer length', async () => {
+        const chunks: string[] = [];
+
+        startProjector([
+          '--ogmios-url',
+          'ws://localhost:1234',
+          '--postgres-connection-string',
+          postgresConnectionStringProjection,
+          '--logger-min-severity',
+          'debug',
+          '--blocks-buffer-length',
+          '23',
+          ProjectionName.StakePool
+        ]);
+
+        proc.stdout?.on('data', (data: Buffer) => chunks.push(data.toString('utf8')));
+
+        await assertServerAlive();
+
+        expect(chunks.join('')).toMatch('Using a 23 blocks buffer');
+      });
     });
 
     describe('with environment variables', () => {
@@ -2355,7 +2379,7 @@ describe('CLI', () => {
           startProjector(
             {
               OGMIOS_URL: 'ws://localhost:1234',
-              POSTGRES_CONNECTION_STRING: 'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection'
+              POSTGRES_CONNECTION_STRING: postgresConnectionStringProjection
             },
             [ProjectionName.UTXO]
           );
@@ -2366,7 +2390,7 @@ describe('CLI', () => {
           startProjector(
             {
               OGMIOS_URL: 'ws://localhost:1234',
-              POSTGRES_CONNECTION_STRING: 'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection'
+              POSTGRES_CONNECTION_STRING: postgresConnectionStringProjection
             },
             [`${ProjectionName.UTXO},${ProjectionName.StakePool}`]
           );
@@ -2376,7 +2400,7 @@ describe('CLI', () => {
         test('with multiple projections as --projection-names argument', async () => {
           startProjector({
             OGMIOS_URL: 'ws://localhost:1234',
-            POSTGRES_CONNECTION_STRING: 'postgresql://postgres:doNoUseThisSecret!@127.0.0.1:5432/projection',
+            POSTGRES_CONNECTION_STRING: postgresConnectionStringProjection,
             PROJECTION_NAMES: `${ProjectionName.UTXO},${ProjectionName.StakePool}`
           });
           await assertServerAlive();

diff --git a/packages/core/test/CardanoNode/util/bufferChainSyncEvent.test.ts b/packages/core/test/CardanoNode/util/bufferChainSyncEvent.test.ts
@@ -158,7 +158,7 @@ describe('bufferChainSyncEvent', () => {
     consumer.consumeTill(3);
     await sleep(10);
     producer.produceTill(16);
-    await sleep(10);
+    await sleep(20);
     consumer.consumeTill(4);
     await sleep(10);
   };

diff --git a/packages/e2e/test/projection/offline-fork.test.ts b/packages/e2e/test/projection/offline-fork.test.ts
@@ -121,7 +121,7 @@ describe('resuming projection when intersection is not local tip', () => {
     buffer: StabilityWindowBuffer,
     into: ProjectionOperator<Mappers.WithStakeKeys>
   ) =>
-    Bootstrap.fromCardanoNode({ buffer, cardanoNode, logger }).pipe(
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode, logger }).pipe(
       Mappers.withCertificates(),
       Mappers.withStakeKeys(),
       into,

diff --git a/packages/e2e/test/projection/single-tenant-utxo.test.ts b/packages/e2e/test/projection/single-tenant-utxo.test.ts
@@ -100,7 +100,7 @@ describe('single-tenant utxo projection', () => {
   afterEach(async () => cleanup());
 
   const projectMultiTenant = () =>
-    Bootstrap.fromCardanoNode({ buffer, cardanoNode, logger }).pipe(
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode, logger }).pipe(
       Mappers.withMint(),
       Mappers.withUtxo(),
       Postgres.withTypeormTransaction({ dataSource$: of(dataSource), logger }),
@@ -123,7 +123,7 @@ describe('single-tenant utxo projection', () => {
     );
 
   const projectSingleTenant = (addresses: Cardano.PaymentAddress[]) =>
-    Bootstrap.fromCardanoNode({ buffer, cardanoNode, logger }).pipe(
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode, logger }).pipe(
       Mappers.withMint(),
       Mappers.withUtxo(),
       Mappers.filterProducedUtxoByAddresses({ addresses }),

diff --git a/packages/projection-typeorm/test/TypeormStabilityWindowBuffer.test.ts b/packages/projection-typeorm/test/TypeormStabilityWindowBuffer.test.ts
@@ -51,6 +51,7 @@ describe('TypeormStabilityWindowBuffer', () => {
     await buffer.initialize(queryRunner);
     project$ = defer(() =>
       Bootstrap.fromCardanoNode({
+        blocksBufferLength: 10,
         buffer,
         cardanoNode: {
           ...cardanoNode,

diff --git a/packages/projection-typeorm/test/operators/storeAssets.test.ts b/packages/projection-typeorm/test/operators/storeAssets.test.ts
@@ -24,11 +24,7 @@ describe('storeAssets', () => {
   const entities = [BlockEntity, BlockDataEntity, AssetEntity];
 
   const project$ = () =>
-    Bootstrap.fromCardanoNode({
-      buffer,
-      cardanoNode: stubEvents.cardanoNode,
-      logger
-    }).pipe(
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode: stubEvents.cardanoNode, logger }).pipe(
       Mappers.withMint(),
       withTypeormTransaction({
         dataSource$,

diff --git a/packages/projection-typeorm/test/operators/storeHandles.test.ts b/packages/projection-typeorm/test/operators/storeHandles.test.ts
@@ -56,11 +56,9 @@ describe('storeHandles', () => {
     );
 
   const project$ = () =>
-    Bootstrap.fromCardanoNode({
-      buffer,
-      cardanoNode: stubEvents.cardanoNode,
-      logger
-    }).pipe(applyOperators());
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode: stubEvents.cardanoNode, logger }).pipe(
+      applyOperators()
+    );
 
   const projectTilFirst = createProjectorTilFirst(project$);
 

diff --git a/packages/projection-typeorm/test/operators/storeStakePoolMetadataJob.test.ts b/packages/projection-typeorm/test/operators/storeStakePoolMetadataJob.test.ts
@@ -28,11 +28,7 @@ describe('storeStakePoolMetadataJob', () => {
   let queryRunner: QueryRunner;
   let buffer: TypeormStabilityWindowBuffer;
   const project$ = () =>
-    Bootstrap.fromCardanoNode({
-      buffer,
-      cardanoNode: stubEvents.cardanoNode,
-      logger
-    }).pipe(
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode: stubEvents.cardanoNode, logger }).pipe(
       // skipping 1st event because it's not rolled back
       filter((evt) => {
         const SKIP = 32_159;

diff --git a/packages/projection-typeorm/test/operators/storeStakePools.test.ts b/packages/projection-typeorm/test/operators/storeStakePools.test.ts
@@ -27,11 +27,7 @@ describe('storeStakePools', () => {
   let queryRunner: QueryRunner;
   let buffer: TypeormStabilityWindowBuffer;
   const project = () =>
-    Bootstrap.fromCardanoNode({
-      buffer,
-      cardanoNode: data.cardanoNode,
-      logger
-    }).pipe(
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode: data.cardanoNode, logger }).pipe(
       Mappers.withCertificates(),
       Mappers.withStakePools(),
       withTypeormTransaction({ dataSource$: of(dataSource), logger }),

diff --git a/packages/projection-typeorm/test/operators/storeUtxo.test.ts b/packages/projection-typeorm/test/operators/storeUtxo.test.ts
@@ -26,11 +26,7 @@ describe('storeUtxo', () => {
   const entities = [BlockEntity, BlockDataEntity, AssetEntity, TokensEntity, OutputEntity];
 
   const project$ = () =>
-    Bootstrap.fromCardanoNode({
-      buffer,
-      cardanoNode: stubEvents.cardanoNode,
-      logger
-    }).pipe(
+    Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode: stubEvents.cardanoNode, logger }).pipe(
       Mappers.withMint(),
       Mappers.withUtxo(),
       withTypeormTransaction({