Fix issue with db sync and workers enabled

packages/node/src/indexer/store.service.ts

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import assert from 'assert';
+import { isMainThread } from 'worker_threads';
 import { Injectable } from '@nestjs/common';
 import { hexToU8a, u8aToBuffer } from '@polkadot/util';
 import { blake2AsHex } from '@polkadot/util-crypto';
@@ -305,7 +306,8 @@ export class StoreService {
 
     // this will allow alter current entity, including fields
     // TODO, add rules for changes, eg only allow add nullable field
-    await this.sequelize.sync({ alter: { drop: true } });
+    // Only allow altering the tables on the main thread
+    await this.sequelize.sync({ alter: { drop: isMainThread } });
     await this.setMetadata('historicalStateEnabled', this.historical);
     for (const query of extraQueries) {
       await this.sequelize.query(query);

packages/node/src/indexer/worker/block-dispatcher.service.ts

@@ -338,7 +338,9 @@ export class WorkerBlockDispatcherService
     this.queue.abort();
 
     // Stop all workers
-    await Promise.all(this.workers.map((w) => w.terminate()));
+    if (this.workers) {
+      await Promise.all(this.workers.map((w) => w.terminate()));
+    }
   }
 
   enqueueBlocks(heights: number[]): void {

packages/node/src/indexer/worker/worker.builder.ts

@@ -5,6 +5,11 @@ import * as workers from 'worker_threads';
 import { Logger } from 'pino';
 import { getLogger } from '../../utils/logger';
 
+export type SerializableError = {
+  message: string;
+  stack?: string;
+};
+
 export type Request = {
   id: number | string;
   name: string;
@@ -13,7 +18,7 @@ export type Request = {
 
 export type Response<T = any> = {
   id: number | string;
-  error?: string;
+  error?: SerializableError;
   result?: T;
 };
 
@@ -39,7 +44,9 @@ export function registerWorker(fns: AsyncMethods): void {
     if (!fn) {
       workers.parentPort.postMessage(<Response>{
         id: req.id,
-        error: `handleRequest: Function "${req.name}" not found`,
+        error: {
+          message: `handleRequest: Function "${req.name}" not found`,
+        },
       });
       return;
     }
@@ -54,7 +61,7 @@ export function registerWorker(fns: AsyncMethods): void {
     } catch (e) {
       workers.parentPort.postMessage(<Response>{
         id: req.id,
-        error: e.message,
+        error: e,
       });
     }
   }
@@ -73,7 +80,7 @@ export class Worker<T extends AsyncMethods> {
 
   private responseListeners: Record<
     number | string,
-    (data?: any, error?: string) => void
+    (data?: any, error?: SerializableError) => void
   > = {};
 
   private _reqCounter = 0;
@@ -140,7 +147,9 @@ export class Worker<T extends AsyncMethods> {
     return new Promise<T>((resolve, reject) => {
       this.responseListeners[id] = (data, error) => {
         if (error) {
-          reject(new Error(error));
+          const e = new Error(error.message);
+          e.stack = error.stack ?? e.stack;
+          reject(e);
         } else {
           resolve(data);
         }

packages/node/src/indexer/worker/worker.ts

@@ -22,21 +22,28 @@ let workerService: WorkerService;
 const logger = getLogger(`worker #${threadId}`);
 
 async function initWorker(): Promise<void> {
-  if (app) {
-    logger.warn('Worker already initialised');
-    return;
+  try {
+    if (app) {
+      logger.warn('Worker already initialised');
+      return;
+    }
+
+    app = await NestFactory.create(WorkerModule, {
+      logger: new NestLogger(),
+    });
+
+    await app.init();
+
+    const indexerManager = app.get(IndexerManager);
+    // Initialise async services, we do this here rather than in factories so we can capture one off events
+    await indexerManager.start();
+
+    workerService = app.get(WorkerService);
+  } catch (e) {
+    console.log('Failed to start worker', e);
+    logger.error(e, 'Failed to start worker');
+    throw e;
   }
-
-  app = await NestFactory.create(WorkerModule, {
-    logger: new NestLogger(),
-  });
-
-  await app.init();
-
-  const indexerManager = app.get(IndexerManager);
-  await indexerManager.start();
-
-  workerService = app.get(WorkerService);
 }
 
 async function fetchBlock(height: number): Promise<FetchBlockResponse> {
@@ -97,3 +104,8 @@ export type NumFetchedBlocks = typeof numFetchedBlocks;
 export type NumFetchingBlocks = typeof numFetchingBlocks;
 export type SetCurrentRuntimeVersion = typeof setCurrentRuntimeVersion;
 export type GetWorkerStatus = typeof getStatus;
+
+process.on('uncaughtException', (e) => {
+  logger.error(e, 'Uncaught Exception');
+  throw e;
+});