refactor: hoist createObservableDataSource to projection-typeorm

restructure observable DataSource utils for easier re-use
as preparation for new StabilityWindowBuffer interface that
will require it's own connection during Bootstrap

Author: mkazlauskas

packages/cardano-services/src/Program/services/pgboss.ts

@@ -6,6 +6,7 @@ import {
   PoolRegistrationEntity,
   PoolRetirementEntity,
   StakePoolEntity,
+  createDataSource,
   createPgBoss,
   isRecoverableTypeormError
 } from '@cardano-sdk/projection-typeorm';
@@ -32,14 +33,13 @@ import { Pool } from 'pg';
 import { Router } from 'express';
 import { StakePoolMetadataProgramOptions } from '../options/stakePoolMetadata';
 import { contextLogger } from '@cardano-sdk/util';
-import { createObservableDataSource } from '../../Projection/createTypeormProjection';
 import { retryBackoff } from 'backoff-rxjs';
 import PgBoss from 'pg-boss';
 
 /**
  * The entities required by the job handlers
  */
-export const pgBossEntities = [
+export const pgBossEntities: Function[] = [
   CurrentPoolMetricsEntity,
   BlockEntity,
   PoolMetadataEntity,
@@ -49,13 +49,28 @@ export const pgBossEntities = [
 ];
 
 export const createPgBossDataSource = (connectionConfig$: Observable<PgConnectionConfig>, logger: Logger) =>
-  createObservableDataSource({
-    connectionConfig$,
-    entities: pgBossEntities,
-    extensions: {},
-    logger,
-    migrationsRun: false
-  });
+  // TODO: use createObservableDataSource from projection-typeorm package.
+  // A challenge in doing that is that we call subscriber.error on retryable errors in order to reconnect.
+  // Doing that with createObservableDataSource will 'destroy' the data source that's currently used,
+  // so pg-boss is then unable to update job status and it stays 'active', not available for the newly
+  // recreated worker to be picked up.
+  // TODO: this raises another question - what happens when database connection drops while working on a job?
+  // Will it stay 'active' forever, or will pg-boss eventually update it due to some sort of timeout?
+  connectionConfig$.pipe(
+    switchMap((connectionConfig) =>
+      from(
+        (async () => {
+          const dataSource = createDataSource({
+            connectionConfig,
+            entities: pgBossEntities,
+            logger
+          });
+          await dataSource.initialize();
+          return dataSource;
+        })()
+      )
+    )
+  );
 
 export type PgBossWorkerArgs = CommonProgramOptions &
   StakePoolMetadataProgramOptions &
@@ -140,6 +155,7 @@ export class PgBossHttpService extends HttpService {
         // This ensures that if an error which can't be retried arrives here is handled as a FATAL error
         shouldRetry: (error: unknown) => {
           const retry = isRecoverableError(error);
+          this.logger.debug('work() shouldRetry', retry, error);
 
           this.#health = {
             ok: false,

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

@@ -2,13 +2,13 @@
 /* eslint-disable prefer-spread */
 import { Cardano } from '@cardano-sdk/core';
 import { Logger } from 'ts-log';
-import { Observable, from, switchMap, takeWhile } from 'rxjs';
+import { Observable, takeWhile } from 'rxjs';
 import {
   PgConnectionConfig,
   TypeormDevOptions,
   TypeormStabilityWindowBuffer,
   WithTypeormContext,
-  createDataSource,
+  createObservableConnection,
   isRecoverableTypeormError,
   typeormTransactionCommit,
   withTypeormTransaction
@@ -44,47 +44,6 @@ const applyStores =
   (evt$: Observable<T>) =>
     evt$.pipe.apply(evt$, selectedStores as any) as Observable<T>;
 
-export const createObservableDataSource = ({
-  connectionConfig$,
-  logger,
-  buffer,
-  devOptions,
-  entities,
-  extensions,
-  migrationsRun
-}: Omit<
-  CreateTypeormProjectionProps,
-  'blocksBufferLength' | 'exitAtBlockNo' | 'projections' | 'projectionSource$' | 'projectionOptions'
-> &
-  Pick<PreparedProjection, 'entities' | 'extensions'> & { migrationsRun: boolean }) =>
-  connectionConfig$.pipe(
-    switchMap((connectionConfig) =>
-      from(
-        (async () => {
-          const dataSource = createDataSource({
-            connectionConfig,
-            devOptions,
-            entities,
-            extensions,
-            logger,
-            options: {
-              installExtensions: true,
-              migrations: migrations.filter(({ entity }) => entities.includes(entity as any)),
-              migrationsRun: migrationsRun && !devOptions?.synchronize
-            }
-          });
-          await dataSource.initialize();
-          if (buffer) {
-            const queryRunner = dataSource.createQueryRunner('master');
-            await buffer.initialize(queryRunner);
-            await queryRunner.release();
-          }
-          return dataSource;
-        })()
-      )
-    )
-  );
-
 /**
  * Creates a projection observable that applies a sequence of operators
  * required to project requested `projections` into a postgres database.
@@ -116,24 +75,22 @@ export const createTypeormProjection = ({
     },
     { logger }
   );
-  const dataSource$ = createObservableDataSource({
-    buffer,
+  const connection$ = createObservableConnection({
     connectionConfig$,
     devOptions,
     entities,
     extensions,
     logger,
-    migrationsRun: true
+    options: {
+      installExtensions: true,
+      migrations: migrations.filter(({ entity }) => entities.includes(entity as any)),
+      migrationsRun: !devOptions?.synchronize
+    }
   });
   return projectionSource$.pipe(
     applyMappers(mappers),
     shareRetryBackoff(
-      (evt$) =>
-        evt$.pipe(
-          withTypeormTransaction({ dataSource$, logger }, extensions),
-          applyStores(stores),
-          typeormTransactionCommit()
-        ),
+      (evt$) => evt$.pipe(withTypeormTransaction({ connection$ }), applyStores(stores), typeormTransactionCommit()),
       { shouldRetry: isRecoverableTypeormError }
     ),
     requestNext(),

packages/cardano-services/test/Program/services/pgboss.test.ts

@@ -2,6 +2,7 @@ import {
   BlockEntity,
   PgConnectionConfig,
   STAKE_POOL_METADATA_QUEUE,
+  createDataSource,
   createPgBossExtension
 } from '@cardano-sdk/projection-typeorm';
 import { Cardano } from '@cardano-sdk/core';
@@ -11,7 +12,7 @@ import { PgBossHttpService, pgBossEntities } from '../../../src/Program/services
 import { Pool } from 'pg';
 import { StakePoolMetadataFetchMode } from '../../../src/Program/options';
 import { WorkerHandlerFactoryOptions } from '../../../src/PgBoss';
-import { createObservableDataSource, getConnectionConfig, getPool } from '../../../src';
+import { getConnectionConfig, getPool } from '../../../src/Program/services/postgres';
 import { logger } from '@cardano-sdk/util-dev';
 
 const dnsResolver = () => Promise.resolve({ name: 'localhost', port: 5433, priority: 6, weight: 5 });
@@ -70,6 +71,7 @@ jest.mock('@cardano-sdk/projection-typeorm', () => {
 
 describe('PgBossHttpService', () => {
   let connectionConfig$: Observable<PgConnectionConfig>;
+  let connectionConfig: PgConnectionConfig;
   let dataSource: DataSource;
   let db: Pool;
   let service: PgBossHttpService | undefined;
@@ -87,15 +89,7 @@ describe('PgBossHttpService', () => {
     };
 
     connectionConfig$ = getConnectionConfig(dnsResolver, 'test', 'StakePool', args);
-    const dataSource$ = createObservableDataSource({
-      connectionConfig$,
-      devOptions: { dropSchema: true, synchronize: true },
-      entities: pgBossEntities,
-      extensions: { pgBoss: true },
-      logger,
-      migrationsRun: false
-    });
-    dataSource = await firstValueFrom(dataSource$);
+    connectionConfig = await firstValueFrom(connectionConfig$);
 
     const pool = await getPool(dnsResolver, logger, args);
 
@@ -104,8 +98,20 @@ describe('PgBossHttpService', () => {
     db = pool;
   });
 
+  beforeEach(async () => {
+    dataSource = createDataSource({
+      connectionConfig,
+      devOptions: { dropSchema: true, synchronize: true },
+      entities: pgBossEntities,
+      extensions: { pgBoss: true },
+      logger
+    });
+    await dataSource.initialize();
+  });
+
   afterEach(async () => {
     await service?.shutdown();
+    await dataSource.destroy().catch(() => void 0);
   });
 
   it('health check is ok after start with a valid db connection', async () => {
@@ -116,7 +122,6 @@ describe('PgBossHttpService', () => {
         parallelJobs: 3,
         queues: []
       },
-
       { connectionConfig$, db, logger }
     );
     expect(await service.healthCheck()).toEqual({ ok: false, reason: 'PgBossHttpService not started' });
@@ -132,7 +137,6 @@ describe('PgBossHttpService', () => {
     let observableResolver = () => {};
     let subscriptions = 0;
 
-    const connectionConfig = await firstValueFrom(connectionConfig$);
     const config$ = new Observable<PgConnectionConfig>((subscriber) => {
       subscriptions++;
 
@@ -155,6 +159,8 @@ describe('PgBossHttpService', () => {
     await service.start();
 
     // Insert test block with slot 1
+    const queryRunner = dataSource.createQueryRunner();
+    await queryRunner.connect();
     const blockRepos = dataSource.getRepository(BlockEntity);
     const block = { hash: 'test', height: 1, slot: 1 };
     await blockRepos.insert(block);
@@ -165,7 +171,6 @@ describe('PgBossHttpService', () => {
     expect(await collectStatus()).toEqual({ calls: 0, health: { ok: true }, subscriptions: 1 });
 
     // Schedule a job
-    const queryRunner = dataSource.createQueryRunner();
     const pgboss = createPgBossExtension(queryRunner, logger);
     await pgboss.send(STAKE_POOL_METADATA_QUEUE, {}, { retryDelay: 1, retryLimit: 100, slot: Cardano.Slot(1) });
     await queryRunner.release();
@@ -225,5 +230,7 @@ describe('PgBossHttpService', () => {
     await testPromises[3];
 
     expect(await collectStatus()).toEqual({ calls: 4, health: { ok: true }, subscriptions: 3 });
+
+    await dataSource.destroy();
   });
 });

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

@@ -14,9 +14,21 @@ import { projectorConnectionConfig, projectorConnectionConfig$ } from '../util';
 describe('createTypeormProjection', () => {
   it('creates a projection to PostgreSQL based on requested projection names', async () => {
     // Setup
-    const data = chainSyncData(ChainSyncDataSet.WithMint);
-    const buffer = new TypeormStabilityWindowBuffer({ allowNonSequentialBlockHeights: true, logger });
     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,
@@ -36,11 +48,6 @@ describe('createTypeormProjection', () => {
     await lastValueFrom(projection$);
 
     // Check data in the database
-    const { entities } = prepareTypeormProjection({ buffer, projections }, { logger });
-    const dataSource = createDataSource({ connectionConfig: projectorConnectionConfig, entities, logger });
-    await dataSource.initialize();
-    const queryRunner = dataSource.createQueryRunner();
-    await queryRunner.connect();
     expect(await queryRunner.manager.count(AssetEntity)).toBeGreaterThan(0);
     expect(await queryRunner.manager.count(TokensEntity)).toBeGreaterThan(0);
     expect(await queryRunner.manager.count(OutputEntity)).toBeGreaterThan(0);

packages/e2e/test/projection/offline-fork.test.ts

@@ -21,8 +21,9 @@ import {
 } from '@cardano-sdk/core';
 import { ChainSyncDataSet, chainSyncData, logger } from '@cardano-sdk/util-dev';
 import { ConnectionConfig } from '@cardano-ogmios/client';
-import { Observable, filter, firstValueFrom, lastValueFrom, of, take, takeWhile, toArray } from 'rxjs';
+import { Observable, defer, filter, firstValueFrom, lastValueFrom, of, take, takeWhile, toArray } from 'rxjs';
 import { OgmiosObservableCardanoNode } from '@cardano-sdk/ogmios';
+import { QueryRunner } from 'typeorm';
 import { createDatabase } from 'typeorm-extension';
 import { getEnv } from '../../src';
 
@@ -202,6 +203,7 @@ describe('resuming projection when intersection is not local tip', () => {
         installExtensions: true
       }
     });
+    let queryRunner: QueryRunner;
 
     const getNumberOfLocalStakeKeys = async () => {
       const repository = dataSource.getRepository(Postgres.StakeKeyEntity);
@@ -217,15 +219,16 @@ describe('resuming projection when intersection is not local tip', () => {
         }
       });
       await dataSource.initialize();
-      await buffer.initialize(dataSource.createQueryRunner());
+      queryRunner = dataSource.createQueryRunner();
+      await buffer.initialize(queryRunner);
     });
     afterAll(() => dataSource.destroy());
 
     testRollbackAndContinue(
       buffer,
       (evt$) =>
         evt$.pipe(
-          Postgres.withTypeormTransaction({ dataSource$: of(dataSource), logger }),
+          Postgres.withTypeormTransaction({ connection$: defer(() => of({ queryRunner })) }),
           Postgres.storeBlock(),
           Postgres.storeStakeKeys(),
           buffer.storeBlockData(),

packages/e2e/test/projection/single-tenant-utxo.test.ts

@@ -4,7 +4,7 @@ import { Bootstrap, Mappers, ProjectionEvent, requestNext } from '@cardano-sdk/p
 import { Cardano, ObservableCardanoNode } from '@cardano-sdk/core';
 import { ConnectionConfig } from '@cardano-ogmios/client';
 import { DataSource, QueryRunner } from 'typeorm';
-import { Observable, filter, firstValueFrom, lastValueFrom, of, scan, takeWhile } from 'rxjs';
+import { Observable, defer, filter, firstValueFrom, lastValueFrom, of, scan, takeWhile } from 'rxjs';
 import { OgmiosObservableCardanoNode } from '@cardano-sdk/ogmios';
 import { createDatabase, dropDatabase } from 'typeorm-extension';
 import { getEnv } from '../../src';
@@ -104,7 +104,7 @@ describe('single-tenant utxo projection', () => {
     Bootstrap.fromCardanoNode({ blocksBufferLength: 10, buffer, cardanoNode, logger }).pipe(
       Mappers.withMint(),
       Mappers.withUtxo(),
-      Postgres.withTypeormTransaction({ dataSource$: of(dataSource), logger }),
+      Postgres.withTypeormTransaction({ connection$: defer(() => of({ queryRunner })) }),
       Postgres.storeBlock(),
       Postgres.storeAssets(),
       Postgres.storeUtxo(),
@@ -115,7 +115,7 @@ describe('single-tenant utxo projection', () => {
 
   const storeUtxo = (evt$: Observable<ProjectionEvent<Mappers.WithMint & Mappers.WithUtxo>>) =>
     evt$.pipe(
-      Postgres.withTypeormTransaction({ dataSource$: of(dataSource), logger }),
+      Postgres.withTypeormTransaction({ connection$: defer(() => of({ queryRunner })) }),
       Postgres.storeBlock(),
       Postgres.storeAssets(),
       Postgres.storeUtxo(),

packages/golden-test-generator/src/ChainSyncEvents/chainSyncEvents.ts

@@ -96,6 +96,7 @@ export const getChainSyncEvents = async (
             const header = ogmiosToCore.blockHeader(block);
             if (!header) return;
             currentBlock = header.blockNo;
+
             if (onBlock !== undefined) {
               onBlock(currentBlock);
             }