Merge pull request #1410 from input-output-hk/fix/no-migrations-on-worker-and-provider-startup

fix: no migrations on worker and provider startup

Author: mkazlauskas

Dockerfile

@@ -61,7 +61,7 @@ CMD ["bash", "-c", "../../node_modules/.bin/tsx watch --clear-screen=false --con
 
 FROM cardano-services as worker
 WORKDIR /app/packages/cardano-services
-CMD ["bash", "-c", "../../node_modules/.bin/tsx watch --clear-screen=false --conditions=development src/cli start-worker"]
+CMD ["bash", "-c", "../../node_modules/.bin/tsx watch --clear-screen=false --conditions=development src/cli start-pg-boss-worker"]
 
 FROM cardano-services as blockfrost-worker
 ENV \

packages/cardano-services/.env.test

@@ -3,6 +3,7 @@ POSTGRES_CONNECTION_STRING_HANDLE=postgresql://postgres:[email protected]
 POSTGRES_CONNECTION_STRING_PROJECTION=postgresql://postgres:[email protected]:5433/projection
 POSTGRES_CONNECTION_STRING_STAKE_POOL=postgresql://postgres:[email protected]:5433/stake_pool
 POSTGRES_CONNECTION_STRING_ASSET=postgresql://postgres:[email protected]:5433/asset
+POSTGRES_CONNECTION_STRING_EMPTY=postgresql://postgres:[email protected]:5433/empty
 CARDANO_NODE_CONFIG_PATH=./config/network/mainnet/cardano-node/config.json
 DB_CACHE_TTL=120
 EPOCH_POLL_INTERVAL=10000

packages/cardano-services/README.md

@@ -99,19 +99,16 @@ OGMIOS_SRV_SERVICE_NAME=some-domain-for-ogmios \
 ./dist/cjs/cli.js start-provider-server
 ```
 
-**`start-worker` using CLI options:**
+**`start-pg-boss-worker` using CLI options:**
 
 ```bash
-./dist/cjs/cli.js \
-  start-worker \
-    --ogmios-srv-service-name  some-domain-for-ogmios
+./dist/cjs/cli.js start-pg-boss-worker --queues=pool-metadata --postgres-connection-string-stake-pool "postgresql://postgres:doNoUseThisSecret\!@localhost:5432/projection" --postgres-connection-string-db-sync "postgresql://postgres:doNoUseThisSecret\!@localhost:5432/cexplorer"
 ```
 
-**`start-worker` using env variables:**
+**`start-pg-boss-worker` using env variables:**
 
 ```bash
-OGMIOS_SRV_SERVICE_NAME=some-domain-for-ogmios \
-./dist/cjs/cli.js start-worker
+QUEUES=pool-metadata POSTGRES_CONNECTION_STRING_STAKE_POOL=postgresql://postgres:doNoUseThisSecret\!@localhost:5432/projection POSTGRES_CONNECTION_STRING_DB_SYNC=postgresql://postgres:doNoUseThisSecret\!@localhost:5432/cexplorer ./dist/cjs/cli.js start-pg-boss-worker
 ```
 
 **`start-projector` using CLI options with Ogmios and PostgreSQL running on localhost:**

packages/cardano-services/package.json

@@ -86,6 +86,7 @@
     "npm-run-all": "^4.1.5",
     "ts-jest": "^28.0.7",
     "ts-node": "^10.0.0",
+    "typeorm-extension": "^2.7.0",
     "typescript": "^4.7.4",
     "wait-on": "^6.0.1"
   },

packages/cardano-services/src/Asset/TypeOrmNftMetadataService.ts

@@ -1,12 +1,12 @@
-import { Asset, Cardano } from '@cardano-sdk/core';
+import { Asset, Cardano, Milliseconds } from '@cardano-sdk/core';
 import { AssetPolicyIdAndName, NftMetadataService } from './types';
 import { NftMetadataEntity } from '@cardano-sdk/projection-typeorm';
 import { QueryRunner } from 'typeorm';
 import { TypeormProviderDependencies, TypeormService } from '../util';
 
 export class TypeOrmNftMetadataService extends TypeormService implements NftMetadataService {
   constructor({ connectionConfig$, logger, entities }: TypeormProviderDependencies) {
-    super('TypeOrmNftMetadataService', { connectionConfig$, entities, logger });
+    super('TypeOrmNftMetadataService', { connectionConfig$, connectionTimeout: Milliseconds(1000), entities, logger });
   }
 
   async getNftMetadata(assetInfo: AssetPolicyIdAndName): Promise<Asset.NftMetadata | null> {

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

@@ -30,6 +30,10 @@ export class PgBossWorkerHttpServer extends HttpServer {
     const { apiUrl } = cfg;
     const { logger } = deps;
     const pgBossService = new PgBossHttpService(cfg, deps);
+    pgBossService.onUnrecoverableError$.subscribe(() => {
+      // eslint-disable-next-line unicorn/no-process-exit
+      process.exit(1);
+    });
 
     super(
       { listen: getListen(apiUrl), name: pgBossWorker },

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

@@ -19,11 +19,11 @@ import { HttpService } from '../../Http/HttpService';
 import { Logger } from 'ts-log';
 import {
   Observable,
+  Subject,
   Subscription,
   catchError,
   concat,
   finalize,
-  firstValueFrom,
   from,
   merge,
   share,
@@ -66,9 +66,18 @@ export const createPgBossDataSource = (connectionConfig$: Observable<PgConnectio
           const dataSource = createDataSource({
             connectionConfig,
             entities: pgBossEntities,
-            logger
+            extensions: { pgBoss: true },
+            logger,
+            options: { migrationsRun: false }
           });
           await dataSource.initialize();
+          const pgbossSchema = await dataSource.query(
+            "SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'pgboss'"
+          );
+          if (pgbossSchema.length === 0) {
+            await dataSource.destroy();
+            throw new Error('Database schema is not ready. Please make sure projector is running.');
+          }
           return dataSource;
         })()
       )
@@ -105,6 +114,7 @@ export class PgBossHttpService extends HttpService {
   #db: Pool;
   #subscription?: Subscription;
   #health: HealthCheckResponse = { ok: false, reason: 'PgBossHttpService not started' };
+  onUnrecoverableError$ = new Subject<unknown>();
 
   constructor(cfg: PgBossWorkerArgs, deps: PgBossServiceDependencies) {
     const { connectionConfig$, db, logger } = deps;
@@ -127,11 +137,16 @@ export class PgBossHttpService extends HttpService {
     // Used for later use of firstValueFrom() to avoid it subscribes again
     const sharedWork$ = this.work().pipe(share());
 
-    // Subscribe to work() to create the first DataSource and start pg-boss
-    this.#subscription = sharedWork$.subscribe();
-
-    // Used to make startImpl actually await for a first emitted value from work()
-    await firstValueFrom(sharedWork$);
+    return new Promise<void>((resolve, reject) => {
+      // Subscribe to work() to create the first DataSource and start pg-boss
+      this.#subscription = sharedWork$.subscribe({
+        error: (error) => {
+          this.onUnrecoverableError$.next(error);
+          reject(error);
+        },
+        next: () => resolve()
+      });
+    });
   }
 
   protected async shutdownImpl() {
@@ -174,8 +189,7 @@ export class PgBossHttpService extends HttpService {
       }),
       catchError((error) => {
         this.logger.error('Fatal worker error', error);
-        // eslint-disable-next-line unicorn/no-process-exit
-        process.exit(1);
+        throw error;
       })
     );
   }

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

@@ -109,6 +109,31 @@ const mergeTlsOptions = (
       }
     : ssl || !!conn.ssl;
 
+export const connStringToPgConnectionConfig = (
+  postgresConnectionString: string,
+  {
+    poolSize,
+    ssl
+  }: {
+    poolSize?: number;
+    ssl?: { ca: string };
+  } = {}
+): PgConnectionConfig => {
+  const conn = connString.parse(postgresConnectionString);
+  if (!conn.database || !conn.host) {
+    throw new InvalidProgramOption('postgresConnectionString');
+  }
+  return {
+    database: conn.database,
+    host: conn.host,
+    password: conn.password,
+    poolSize,
+    port: conn.port ? Number.parseInt(conn.port) : undefined,
+    ssl: mergeTlsOptions(conn, ssl),
+    username: conn.user
+  };
+};
+
 export const getConnectionConfig = <Suffix extends ConnectionNames>(
   dnsResolver: DnsResolver,
   program: string,
@@ -121,19 +146,7 @@ export const getConnectionConfig = <Suffix extends ConnectionNames>(
   const ssl = postgresSslCaFile ? { ca: loadSecret(postgresSslCaFile) } : undefined;
 
   if (postgresConnectionString) {
-    const conn = connString.parse(postgresConnectionString);
-    if (!conn.database || !conn.host) {
-      throw new InvalidProgramOption('postgresConnectionString');
-    }
-    return of({
-      database: conn.database,
-      host: conn.host,
-      max,
-      password: conn.password,
-      port: conn.port ? Number.parseInt(conn.port) : undefined,
-      ssl: mergeTlsOptions(conn, ssl),
-      username: conn.user
-    });
+    return of(connStringToPgConnectionConfig(postgresConnectionString, { poolSize: max, ssl }));
   }
 
   const postgresDb = getPostgresOption(suffix, 'postgresDb', options);

packages/cardano-services/src/util/TypeormProvider/TypeormProvider.ts

@@ -1,22 +1,16 @@
-import { HealthCheckResponse, Provider } from '@cardano-sdk/core';
-import { Logger } from 'ts-log';
-import { Observable, skip } from 'rxjs';
-import { PgConnectionConfig } from '@cardano-sdk/projection-typeorm';
-import { TypeormService } from '../TypeormService';
+import { HealthCheckResponse, Milliseconds, Provider } from '@cardano-sdk/core';
+import { TypeormService, TypeormServiceDependencies } from '../TypeormService';
+import { skip } from 'rxjs';
 
-export interface TypeormProviderDependencies {
-  logger: Logger;
-  entities: Function[];
-  connectionConfig$: Observable<PgConnectionConfig>;
-}
+export type TypeormProviderDependencies = Omit<TypeormServiceDependencies, 'connectionTimeout'>;
 
 const unhealthy = { ok: false, reason: 'Provider error' };
 
 export abstract class TypeormProvider extends TypeormService implements Provider {
   health: HealthCheckResponse = { ok: false, reason: 'not started' };
 
-  constructor(name: string, { connectionConfig$, logger, entities }: TypeormProviderDependencies) {
-    super(name, { connectionConfig$, entities, logger });
+  constructor(name: string, dependencies: TypeormProviderDependencies) {
+    super(name, { ...dependencies, connectionTimeout: Milliseconds(1000) });
     // We skip 1 to omit the initial null value of the subject
     this.dataSource$.pipe(skip(1)).subscribe((dataSource) => {
       this.health = dataSource ? { ok: true } : unhealthy;

packages/cardano-services/src/util/TypeormService/TypeormService.ts

@@ -1,32 +1,47 @@
-import { BehaviorSubject, Observable, Subscription, filter, firstValueFrom } from 'rxjs';
+import { BehaviorSubject, Observable, Subscription, filter, firstValueFrom, timeout } from 'rxjs';
 import { DataSource, QueryRunner } from 'typeorm';
 import { Logger } from 'ts-log';
+import { Milliseconds } from '@cardano-sdk/core';
 import { PgConnectionConfig } from '@cardano-sdk/projection-typeorm';
 import { RunnableModule, isNotNil } from '@cardano-sdk/util';
 import { createTypeormDataSource } from '../createTypeormDataSource';
 
-interface TypeormServiceDependencies {
+export interface TypeormServiceDependencies {
   logger: Logger;
   entities: Function[];
   connectionConfig$: Observable<PgConnectionConfig>;
+  connectionTimeout: Milliseconds;
 }
 
 export abstract class TypeormService extends RunnableModule {
   #entities: Function[];
   #connectionConfig$: Observable<PgConnectionConfig>;
   protected dataSource$ = new BehaviorSubject<DataSource | null>(null);
   #subscription: Subscription | undefined;
+  #connectionTimeout: Milliseconds;
 
-  constructor(name: string, { connectionConfig$, logger, entities }: TypeormServiceDependencies) {
+  constructor(name: string, { connectionConfig$, logger, entities, connectionTimeout }: TypeormServiceDependencies) {
     super(name, logger);
     this.#entities = entities;
     this.#connectionConfig$ = connectionConfig$;
+    this.#connectionTimeout = connectionTimeout;
   }
 
-  #subscribeToDataSource() {
-    this.#subscription = createTypeormDataSource(this.#connectionConfig$, this.#entities, this.logger).subscribe(
-      (dataSource) => this.dataSource$.next(dataSource)
-    );
+  async #subscribeToDataSource() {
+    return new Promise((resolve, reject) => {
+      this.#subscription = createTypeormDataSource(this.#connectionConfig$, this.#entities, this.logger).subscribe(
+        (dataSource) => {
+          if (dataSource !== this.dataSource$.value) {
+            this.dataSource$.next(dataSource);
+          }
+          if (dataSource) {
+            resolve(dataSource);
+          } else {
+            reject(new Error('Failed to initialize data source'));
+          }
+        }
+      );
+    });
   }
 
   #reset() {
@@ -37,12 +52,16 @@ export abstract class TypeormService extends RunnableModule {
 
   onError(_: unknown) {
     this.#reset();
-    this.#subscribeToDataSource();
+    void this.#subscribeToDataSource().catch(() => void 0);
   }
 
   async withDataSource<T>(callback: (dataSource: DataSource) => Promise<T>): Promise<T> {
+    // eslint-disable-next-line @typescript-eslint/no-unused-expressions
+    this.#connectionTimeout;
     try {
-      return await callback(await firstValueFrom(this.dataSource$.pipe(filter(isNotNil))));
+      return await callback(
+        await firstValueFrom(this.dataSource$.pipe(filter(isNotNil), timeout({ first: this.#connectionTimeout })))
+      );
     } catch (error) {
       this.onError(error);
       throw error;
@@ -72,7 +91,7 @@ export abstract class TypeormService extends RunnableModule {
   }
 
   async startImpl() {
-    this.#subscribeToDataSource();
+    await this.#subscribeToDataSource();
   }
 
   async shutdownImpl() {

packages/cardano-services/src/util/createTypeormDataSource.ts

@@ -12,7 +12,11 @@ export const createTypeormDataSource = (
       from(
         (async () => {
           try {
-            const dataSource = createDataSource({ connectionConfig, entities, logger });
+            const dataSource = createDataSource({
+              connectionConfig,
+              entities,
+              logger
+            });
             await dataSource.initialize();
             return dataSource;
           } catch (error) {