fix: do not run schema migrations when starting pgboss worker

Author: mkazlauskas

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/test/Program/services/pgboss.test.ts

@@ -70,6 +70,7 @@ jest.mock('@cardano-sdk/projection-typeorm', () => {
 });
 
 describe('PgBossHttpService', () => {
+  const apiUrl = new URL('http://unused/');
   let connectionConfig$: Observable<PgConnectionConfig>;
   let connectionConfig: PgConnectionConfig;
   let dataSource: DataSource;
@@ -98,9 +99,37 @@ describe('PgBossHttpService', () => {
     db = pool;
   });
 
-  afterEach(async () => {
-    await service?.shutdown();
-    await dataSource.destroy().catch(() => void 0);
+  describe('without existing database', () => {
+    describe('initialize', () => {
+      it('throws an error and does not initialize pgboss schema', async () => {
+        service = new PgBossHttpService(
+          {
+            apiUrl,
+            dbCacheTtl: 0,
+            lastRosEpochs: 10,
+            metadataFetchMode: StakePoolMetadataFetchMode.DIRECT,
+            parallelJobs: 3,
+            queues: [],
+            schedules: []
+          },
+          { connectionConfig$, db, logger }
+        );
+        await expect(async () => {
+          await service!.initialize();
+          await service!.start();
+        }).rejects.toThrowError();
+        const pool = new Pool({
+          // most of the props are the same as for typeorm
+          ...connectionConfig,
+          ssl: undefined,
+          user: connectionConfig.username
+        });
+        const pgbossSchema = await pool.query(
+          "SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'pgboss'"
+        );
+        expect(pgbossSchema.rowCount).toBe(0);
+      });
+    });
   });
 
   describe('with existing database', () => {
@@ -115,10 +144,15 @@ describe('PgBossHttpService', () => {
       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 () => {
       service = new PgBossHttpService(
         {
-          apiUrl: new URL('http://unused/'),
+          apiUrl,
           dbCacheTtl: 0,
           lastRosEpochs: 10,
           metadataFetchMode: StakePoolMetadataFetchMode.DIRECT,
@@ -152,7 +186,7 @@ describe('PgBossHttpService', () => {
 
       service = new PgBossHttpService(
         {
-          apiUrl: new URL('http://unused/'),
+          apiUrl,
           dbCacheTtl: 0,
           lastRosEpochs: 10,
           metadataFetchMode: StakePoolMetadataFetchMode.DIRECT,

packages/projection-typeorm/src/createDataSource.ts

@@ -149,12 +149,14 @@ export const createDataSource = ({
   return patchObject(dataSource, {
     async initialize() {
       await dataSource.initialize();
-      await initializePgBoss(
-        dataSource,
-        contextLogger(logger, 'createDataSource'),
-        extensions?.pgBoss,
-        devOptions?.dropSchema
-      );
+      if (extensions?.pgBoss && (options?.migrationsRun || devOptions?.synchronize)) {
+        await initializePgBoss(
+          dataSource,
+          contextLogger(logger, 'createDataSource'),
+          extensions?.pgBoss,
+          devOptions?.dropSchema
+        );
+      }
       return dataSource;
     }
   });

packages/projection-typeorm/test/createDataSource.test.ts

@@ -16,7 +16,7 @@ describe('createDataSource', () => {
   });
 
   describe('pg-boss schema', () => {
-    it('initialize() creates and drops pg-boss schema when pgBoss extension is enabled', async () => {
+    it('initialize() creates pg-boss schema when pgBoss extension is enabled', async () => {
       const dataSourceWithBoss = await initializeDataSource({
         entities: [BlockEntity],
         extensions: {
@@ -26,11 +26,6 @@ describe('createDataSource', () => {
       const queryRunnerWithBoss = dataSourceWithBoss.createQueryRunner();
       expect(await pgBossSchemaExists(queryRunnerWithBoss)).toBe(true);
       await queryRunnerWithBoss.release();
-
-      const dataSourceWithoutBoss = await initializeDataSource({ entities: [] });
-      const queryRunnerWithoutBoss = dataSourceWithoutBoss.createQueryRunner();
-      expect(await pgBossSchemaExists(queryRunnerWithoutBoss)).toBe(false);
-      await queryRunnerWithBoss.release();
     });
   });
 });