feat: do not write to stability window buffer til volatile

It's implemented by splitting StabilityWindowBuffer into 2 pieces:
- Tip observable, which is used to determine local tip
- StabilityWindowBuffer is now simplified to just 'getBlock' method

BREAKING CHANGE: simplify StabilityWindowBuffer interface to just 'getBlock'
- Bootstrap.fromCardanoNode now requires Tip observable parameter

Author: mkazlauskas

packages/cardano-services/src/Program/programs/projector.ts

@@ -1,4 +1,3 @@
-import { Bootstrap } from '@cardano-sdk/projection';
 import { Cardano } from '@cardano-sdk/core';
 import { CommonProgramOptions, OgmiosProgramOptions, PosgresProgramOptions } from '../options';
 import { DnsResolver, createDnsResolver } from '../utils';
@@ -12,8 +11,8 @@ import { Logger } from 'ts-log';
 import { MissingProgramOption, UnknownServiceName } from '../errors';
 import { ProjectionHttpService, ProjectionName, createTypeormProjection, storeOperators } from '../../Projection';
 import { SrvRecord } from 'dns';
-import { TypeormStabilityWindowBuffer, createStorePoolMetricsUpdateJob } from '@cardano-sdk/projection-typeorm';
 import { createLogger } from 'bunyan';
+import { createStorePoolMetricsUpdateJob } from '@cardano-sdk/projection-typeorm';
 import { getConnectionConfig, getOgmiosObservableCardanoNode } from '../services';
 
 export const BLOCKS_BUFFER_LENGTH_DEFAULT = 10;
@@ -50,24 +49,17 @@ const createProjectionHttpService = async (options: ProjectionMapFactoryOptions)
     ogmiosUrl: args.ogmiosUrl
   });
   const connectionConfig$ = getConnectionConfig(dnsResolver, 'projector', '', args);
-  const buffer = new TypeormStabilityWindowBuffer({ logger });
   const { blocksBufferLength, dropSchema, dryRun, exitAtBlockNo, handlePolicyIds, projectionNames, synchronize } = args;
   const projection$ = createTypeormProjection({
     blocksBufferLength,
-    buffer,
+    cardanoNode,
     connectionConfig$,
     devOptions: { dropSchema, synchronize },
     exitAtBlockNo,
     logger,
     projectionOptions: {
       handlePolicyIds
     },
-    projectionSource$: Bootstrap.fromCardanoNode({
-      blocksBufferLength,
-      buffer,
-      cardanoNode,
-      logger
-    }),
     projections: projectionNames
   });
   return new ProjectionHttpService({ dryRun, projection$, projectionNames }, { logger });

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

@@ -1,14 +1,17 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 /* eslint-disable prefer-spread */
-import { Cardano } from '@cardano-sdk/core';
+import { Bootstrap, ProjectionEvent, logProjectionProgress, requestNext } from '@cardano-sdk/projection';
+import { Cardano, ObservableCardanoNode } from '@cardano-sdk/core';
 import { Logger } from 'ts-log';
-import { Observable, takeWhile } from 'rxjs';
+import { Observable, concat, defer, take, takeWhile } from 'rxjs';
 import {
   PgConnectionConfig,
   TypeormDevOptions,
+  TypeormOptions,
   TypeormStabilityWindowBuffer,
   WithTypeormContext,
   createObservableConnection,
+  createTypeormTipTracker,
   isRecoverableTypeormError,
   typeormTransactionCommit,
   withTypeormTransaction
@@ -19,19 +22,22 @@ import {
   ProjectionOptions,
   prepareTypeormProjection
 } from './prepareTypeormProjection';
-import { ProjectionEvent, logProjectionProgress, requestNext } from '@cardano-sdk/projection';
+import { ReconnectionConfig, passthrough, shareRetryBackoff, toEmpty } from '@cardano-sdk/util-rxjs';
 import { migrations } from './migrations';
-import { passthrough, shareRetryBackoff } from '@cardano-sdk/util-rxjs';
+
+const reconnectionConfig: ReconnectionConfig = {
+  initialInterval: 50,
+  maxInterval: 5000
+};
 
 export interface CreateTypeormProjectionProps {
   projections: ProjectionName[];
   blocksBufferLength: number;
-  buffer?: TypeormStabilityWindowBuffer;
-  projectionSource$: Observable<ProjectionEvent>;
   connectionConfig$: Observable<PgConnectionConfig>;
   devOptions?: TypeormDevOptions;
   exitAtBlockNo?: Cardano.BlockNo;
   logger: Logger;
+  cardanoNode: ObservableCardanoNode;
   projectionOptions?: ProjectionOptions;
 }
 
@@ -54,12 +60,11 @@ const applyStores =
 export const createTypeormProjection = ({
   blocksBufferLength,
   projections,
-  projectionSource$,
   connectionConfig$,
   logger,
-  devOptions,
+  devOptions: requestedDevOptions,
+  cardanoNode,
   exitAtBlockNo,
-  buffer,
   projectionOptions
 }: CreateTypeormProjectionProps) => {
   const { handlePolicyIds } = { handlePolicyIds: [], ...projectionOptions };
@@ -69,32 +74,65 @@ export const createTypeormProjection = ({
 
   const { mappers, entities, stores, extensions } = prepareTypeormProjection(
     {
-      buffer,
       options: projectionOptions,
       projections
     },
     { logger }
   );
-  const connection$ = createObservableConnection({
-    connectionConfig$,
-    devOptions,
-    entities,
-    extensions,
+  const connect = (options?: TypeormOptions, devOptions?: TypeormDevOptions) =>
+    createObservableConnection({
+      connectionConfig$,
+      devOptions,
+      entities,
+      extensions,
+      logger,
+      options
+    });
+
+  const tipTracker = createTypeormTipTracker({
+    connection$: connect(),
+    reconnectionConfig
+  });
+  const buffer = new TypeormStabilityWindowBuffer({
+    connection$: connect(),
+    logger,
+    reconnectionConfig
+  });
+  const projectionSource$ = Bootstrap.fromCardanoNode({
+    blocksBufferLength,
+    buffer,
+    cardanoNode,
     logger,
-    options: {
-      installExtensions: true,
-      migrations: migrations.filter(({ entity }) => entities.includes(entity as any)),
-      migrationsRun: !devOptions?.synchronize
-    }
+    projectedTip$: tipTracker.tip$
   });
-  return projectionSource$.pipe(
-    applyMappers(mappers),
-    shareRetryBackoff(
-      (evt$) => evt$.pipe(withTypeormTransaction({ connection$ }), applyStores(stores), typeormTransactionCommit()),
-      { shouldRetry: isRecoverableTypeormError }
-    ),
-    requestNext(),
-    logProjectionProgress(logger),
-    exitAtBlockNo ? takeWhile((event) => event.block.header.blockNo < exitAtBlockNo) : passthrough()
+  return concat(
+    // initialize database before starting the projector
+    connect(
+      {
+        installExtensions: true,
+        migrations: migrations.filter(({ entity }) => entities.includes(entity as any)),
+        migrationsRun: !requestedDevOptions?.synchronize
+      },
+      requestedDevOptions
+    ).pipe(take(1), toEmpty),
+    defer(() =>
+      projectionSource$.pipe(
+        applyMappers(mappers),
+        shareRetryBackoff(
+          (evt$) =>
+            evt$.pipe(
+              withTypeormTransaction({ connection$: connect() }),
+              applyStores(stores),
+              buffer.storeBlockData(),
+              typeormTransactionCommit()
+            ),
+          { shouldRetry: isRecoverableTypeormError }
+        ),
+        tipTracker.trackProjectedTip(),
+        requestNext(),
+        logProjectionProgress(logger),
+        exitAtBlockNo ? takeWhile((event) => event.block.header.blockNo < exitAtBlockNo) : passthrough()
+      )
+    )
   );
 };

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

@@ -15,7 +15,6 @@ import {
   StakeKeyRegistrationEntity,
   StakePoolEntity,
   TokensEntity,
-  TypeormStabilityWindowBuffer,
   createStorePoolMetricsUpdateJob,
   storeAddresses,
   storeAssets,
@@ -139,7 +138,7 @@ type Entity = Entities[EntityName];
 const storeEntities: Partial<Record<StoreName, EntityName[]>> = {
   storeAddresses: ['address'],
   storeAssets: ['asset'],
-  storeBlock: ['block'],
+  storeBlock: ['block', 'blockData'],
   storeHandleMetadata: ['handleMetadata', 'output'],
   storeHandles: ['handle', 'asset', 'tokens', 'output'],
   storeNftMetadata: ['asset'],
@@ -303,7 +302,6 @@ const keyOf = <T extends {}>(obj: T, value: unknown): keyof T | null => {
 
 export interface PrepareTypeormProjectionProps {
   projections: ProjectionName[];
-  buffer?: TypeormStabilityWindowBuffer;
   options?: ProjectionOptions;
 }
 
@@ -312,7 +310,7 @@ export interface PrepareTypeormProjectionProps {
  * based on 'projections' and presence of 'buffer':
  */
 export const prepareTypeormProjection = (
-  { projections, buffer, options = {} }: PrepareTypeormProjectionProps,
+  { projections, options = {} }: PrepareTypeormProjectionProps,
   dependencies: WithLogger
 ) => {
   const mapperSorter = new Sorter<MapperOperator>();
@@ -329,10 +327,6 @@ export const prepareTypeormProjection = (
   const selectedEntities = entitySorter.nodes;
   const selectedMappers = mapperSorter.nodes;
   const selectedStores = storeSorter.nodes;
-  if (buffer) {
-    selectedEntities.push(BlockDataEntity);
-    selectedStores.push(buffer.storeBlockData());
-  }
   const extensions = requiredExtensions(projections);
   return {
     __debug: {

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

@@ -1,56 +1,42 @@
-import {
-  AssetEntity,
-  OutputEntity,
-  TokensEntity,
-  TypeormStabilityWindowBuffer,
-  createDataSource
-} from '@cardano-sdk/projection-typeorm';
-import { Bootstrap } from '@cardano-sdk/projection';
+import { AssetEntity, 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';
 import { projectorConnectionConfig, projectorConnectionConfig$ } from '../util';
 
 describe('createTypeormProjection', () => {
   it('creates a projection to PostgreSQL based on requested projection names', async () => {
-    // Setup
+    // Setup projector
     const projections = [ProjectionName.UTXO];
-    const buffer = new TypeormStabilityWindowBuffer({ allowNonSequentialBlockHeights: true, logger });
-    const { entities } = prepareTypeormProjection({ buffer, projections }, { logger });
-    const dataSource = createDataSource({
-      connectionConfig: projectorConnectionConfig,
-      devOptions: { dropSchema: true, synchronize: true },
-      entities,
-      logger
-    });
-    await dataSource.initialize();
-    const queryRunner = dataSource.createQueryRunner();
-    await queryRunner.connect();
     const data = chainSyncData(ChainSyncDataSet.WithMint);
-    await buffer.initialize(queryRunner);
-
     const projection$ = createTypeormProjection({
       blocksBufferLength: 10,
-      buffer,
+      cardanoNode: data.cardanoNode,
       connectionConfig$: projectorConnectionConfig$,
-      devOptions: { dropSchema: true, synchronize: true },
+      devOptions: { dropSchema: true },
       logger,
-      projectionSource$: Bootstrap.fromCardanoNode({
-        blocksBufferLength: 10,
-        buffer,
-        cardanoNode: data.cardanoNode,
-        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
     expect(await queryRunner.manager.count(AssetEntity)).toBeGreaterThan(0);
     expect(await queryRunner.manager.count(TokensEntity)).toBeGreaterThan(0);
     expect(await queryRunner.manager.count(OutputEntity)).toBeGreaterThan(0);
+
     await queryRunner.release();
     await dataSource.destroy();
   });

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

@@ -1,40 +1,30 @@
 import { ProjectionName, prepareTypeormProjection } from '../../src';
-import { TypeormStabilityWindowBuffer } from '@cardano-sdk/projection-typeorm';
 import { dummyLogger } from 'ts-log';
 
-const prepare = (projections: ProjectionName[], useBuffer?: boolean) => {
-  const { __debug } = prepareTypeormProjection(
-    {
-      buffer: useBuffer ? new TypeormStabilityWindowBuffer({ logger: dummyLogger }) : undefined,
-      projections
-    },
-    { logger: dummyLogger }
-  );
-  return __debug;
-};
+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', () => {
-    test('utxo (without buffer)', () => {
+    test('utxo', () => {
       const { entities, mappers, stores } = prepare([ProjectionName.UTXO]);
-      expect(new Set(entities)).toEqual(new Set(['tokens', 'block', 'asset', 'nftMetadata', 'output']));
-      expect(mappers).toEqual(['withMint', 'withUtxo']);
-      expect(stores).toEqual(['storeBlock', 'storeAssets', 'storeUtxo']);
-    });
-
-    test('utxo (with buffer)', () => {
-      const { entities, mappers, stores } = prepare([ProjectionName.UTXO], true);
       expect(new Set(entities)).toEqual(new Set(['tokens', 'block', 'asset', 'nftMetadata', 'output', 'blockData']));
       expect(mappers).toEqual(['withMint', 'withUtxo']);
-      // 'null' is expected here because buffer.storeBlockData is not a common operator,
-      // but is a method of the buffer. As a result it's not part of the predefined operators object.
-      expect(stores).toEqual(['storeBlock', 'storeAssets', 'storeUtxo', null]);
+      expect(stores).toEqual(['storeBlock', 'storeAssets', 'storeUtxo']);
     });
 
     test('stake-pool,stake-pool-metadata', () => {
       const { entities, mappers, stores } = prepare([ProjectionName.StakePool, ProjectionName.StakePoolMetadataJob]);
       expect(new Set(entities)).toEqual(
-        new Set(['block', 'stakePool', 'poolRegistration', 'poolRetirement', 'poolMetadata', 'currentPoolMetrics'])
+        new Set([
+          'block',
+          'blockData',
+          'stakePool',
+          'poolRegistration',
+          'poolRetirement',
+          'poolMetadata',
+          'currentPoolMetrics'
+        ])
       );
       expect(mappers).toEqual(['withCertificates', 'withStakePools']);
       expect(stores).toEqual(['storeBlock', 'storeStakePools', 'storeStakePoolMetadataJob']);