Gracefully shutdown server (#265)

* Gracefully shutdown server

* Forbid lint warnings

* Avoid setting subgraph path in template contex in codegen

package.json

@@ -16,7 +16,7 @@
     "lerna": "^4.0.0"
   },
   "scripts": {
-    "lint": "lerna run lint --stream",
+    "lint": "lerna run lint --stream -- --max-warnings=0",
     "test:init": "lerna run test:init --stream --ignore @cerc-io/*-watcher",
     "test": "lerna run test --stream --ignore @cerc-io/*-watcher",
     "build": "lerna run build --stream",

packages/codegen/src/client.ts

@@ -71,9 +71,7 @@ export class Client {
     this._exportGql(schemaContent, gqlDir);
 
     const template = Handlebars.compile(this._templateString);
-    const client = template({
+    const client = template({ queries: this._queries });
-      queries: this._queries
-    });
     outStream.write(client);
   }
 

packages/codegen/src/config.ts

@@ -16,14 +16,13 @@ const TEMPLATE_FILE = './templates/config-template.handlebars';
  * @param folderName Watcher folder name to be passed to the template.
  * @param outStream A writable output stream to write the config file to.
  */
-export function exportConfig (watcherKind: string, port: number, folderName: string, outStream: Writable, subgraphPath?: string): void {
+export function exportConfig (watcherKind: string, port: number, folderName: string, outStream: Writable): void {
   const templateString = fs.readFileSync(path.resolve(__dirname, TEMPLATE_FILE)).toString();
   const template = Handlebars.compile(templateString);
   const config = template({
     watcherKind,
     port,
-    folderName,
+    folderName
-    subgraphPath
   });
   outStream.write(config);
 }

packages/codegen/src/fill.ts

@@ -13,9 +13,9 @@ const FILL_TEMPLATE_FILE = './templates/fill-template.handlebars';
  * Writes the fill file generated from a template to a stream.
  * @param fillOutStream A writable output stream to write the fill file to.
  */
-export function exportFill (fillOutStream: Writable, subgraphPath: string): void {
+export function exportFill (fillOutStream: Writable): void {
   const templateString = fs.readFileSync(path.resolve(__dirname, FILL_TEMPLATE_FILE)).toString();
   const template = Handlebars.compile(templateString);
-  const fill = template({ subgraphPath });
+  const fill = template({});
   fillOutStream.write(fill);
 }

packages/codegen/src/generate-code.ts

@@ -193,12 +193,12 @@ function generateWatcher (visitor: Visitor, contracts: any[], config: any) {
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/server.ts'))
     : process.stdout;
-  exportServer(outStream, config.subgraphPath);
+  exportServer(outStream);
 
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'environments/local.toml'))
     : process.stdout;
-  exportConfig(config.kind, config.port, path.basename(outputDir), outStream, config.subgraphPath);
+  exportConfig(config.kind, config.port, path.basename(outputDir), outStream);
 
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/database.ts'))
@@ -228,7 +228,7 @@ function generateWatcher (visitor: Visitor, contracts: any[], config: any) {
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/job-runner.ts'))
     : process.stdout;
-  exportJobRunner(outStream, config.subgraphPath);
+  exportJobRunner(outStream);
 
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/cli/watch-contract.ts'))
@@ -268,7 +268,7 @@ function generateWatcher (visitor: Visitor, contracts: any[], config: any) {
   const fillOutStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/fill.ts'))
     : process.stdout;
-  exportFill(fillOutStream, config.subgraphPath);
+  exportFill(fillOutStream);
 
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/types.ts'))

packages/codegen/src/job-runner.ts

@@ -13,9 +13,9 @@ const TEMPLATE_FILE = './templates/job-runner-template.handlebars';
  * Writes the job-runner file generated from a template to a stream.
  * @param outStream A writable output stream to write the events file to.
  */
-export function exportJobRunner (outStream: Writable, subgraphPath: string): void {
+export function exportJobRunner (outStream: Writable): void {
   const templateString = fs.readFileSync(path.resolve(__dirname, TEMPLATE_FILE)).toString();
   const template = Handlebars.compile(templateString);
-  const events = template({ subgraphPath });
+  const events = template({});
   outStream.write(events);
 }

packages/codegen/src/package.ts

@@ -17,8 +17,6 @@ const TEMPLATE_FILE = './templates/package-template.handlebars';
 export function exportPackage (folderName: string, outStream: Writable): void {
   const templateString = fs.readFileSync(path.resolve(__dirname, TEMPLATE_FILE)).toString();
   const template = Handlebars.compile(templateString);
-  const packageString = template({
+  const packageString = template({ folderName });
-    folderName
-  });
   outStream.write(packageString);
 }

packages/codegen/src/server.ts

@@ -13,9 +13,9 @@ const TEMPLATE_FILE = './templates/server-template.handlebars';
  * Writes the server file generated from a template to a stream.
  * @param outStream A writable output stream to write the server file to.
  */
-export function exportServer (outStream: Writable, subgraphPath: string): void {
+export function exportServer (outStream: Writable): void {
   const templateString = fs.readFileSync(path.resolve(__dirname, TEMPLATE_FILE)).toString();
   const template = Handlebars.compile(templateString);
-  const server = template({ subgraphPath });
+  const server = template({});
   outStream.write(server);
 }

packages/codegen/src/templates/package-template.handlebars

@@ -9,8 +9,8 @@
     "build": "yarn clean && tsc && yarn copy-assets",
     "clean": "rm -rf ./dist",
     "copy-assets": "copyfiles -u 1 src/**/*.gql dist/",
-    "server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
+    "server": "DEBUG=vulcanize:*  YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
-    "server:dev": "DEBUG=vulcanize:* ts-node src/server.ts",
+    "server:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/server.ts",
     "job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
     "job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
     "watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",

packages/codegen/src/templates/server-template.handlebars

@@ -45,8 +45,3 @@ main().then(() => {
 }).catch(err => {
   log(err);
 });
-
-process.on('SIGINT', () => {
-  log(`Exiting process ${process.pid} with code 0`);
-  process.exit(0);
-});

packages/eden-watcher/package.json

@@ -9,8 +9,8 @@
     "build": "yarn clean && tsc && yarn copy-assets",
     "clean": "rm -rf ./dist",
     "copy-assets": "copyfiles -u 1 src/**/*.gql dist/",
-    "server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
+    "server": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
-    "server:dev": "DEBUG=vulcanize:* ts-node src/server.ts",
+    "server:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/server.ts",
     "job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --max-old-space-size=3072 --enable-source-maps dist/job-runner.js",
     "job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
     "watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",

packages/erc20-watcher/package.json

@@ -7,8 +7,8 @@
   "scripts": {
     "lint": "eslint .",
     "build": "tsc",
-    "server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
+    "server": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
-    "server:dev": "DEBUG=vulcanize:* nodemon --watch src src/server.ts",
+    "server:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/server.ts",
     "job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
     "job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
     "watch:contract": "node --enable-source-maps dist/cli/watch-contract.js",

packages/erc721-watcher/package.json

@@ -9,8 +9,8 @@
     "build": "yarn clean && tsc && yarn copy-assets",
     "clean": "rm -rf ./dist",
     "copy-assets": "copyfiles -u 1 src/**/*.gql dist/",
-    "server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
+    "server": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
-    "server:dev": "DEBUG=vulcanize:* ts-node src/server.ts",
+    "server:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/server.ts",
     "job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
     "job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
     "watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",

packages/erc721-watcher/src/server.ts

@@ -29,8 +29,3 @@ main().then(() => {
 }).catch(err => {
   log(err);
 });
-
-process.on('SIGINT', () => {
-  log(`Exiting process ${process.pid} with code 0`);
-  process.exit(0);
-});

packages/graph-test-watcher/package.json

@@ -9,8 +9,8 @@
     "build": "yarn clean && tsc && yarn copy-assets",
     "clean": "rm -rf ./dist",
     "copy-assets": "copyfiles -u 1 src/**/*.gql dist/",
-    "server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
+    "server": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
-    "server:dev": "DEBUG=vulcanize:* ts-node src/server.ts",
+    "server:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/server.ts",
     "job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
     "job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
     "watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",

packages/mobymask-watcher/package.json

@@ -9,8 +9,8 @@
     "build": "yarn clean && tsc && yarn copy-assets",
     "clean": "rm -rf ./dist",
     "copy-assets": "copyfiles -u 1 src/**/*.gql dist/",
-    "server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
+    "server": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
-    "server:dev": "DEBUG=vulcanize:* ts-node src/server.ts",
+    "server:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/server.ts",
     "job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
     "job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
     "watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",

packages/mobymask-watcher/src/server.ts

@@ -29,8 +29,3 @@ main().then(() => {
 }).catch(err => {
   log(err);
 });
-
-process.on('SIGINT', () => {
-  log(`Exiting process ${process.pid} with code 0`);
-  process.exit(0);
-});

packages/util/src/events.ts

@@ -26,6 +26,9 @@ export class EventWatcher {
   _pubsub: PubSub
   _jobQueue: JobQueue
 
+  _shutDown = false
+  _signalCount = 0
+
   constructor (ethClient: EthClient, indexer: IndexerInterface, pubsub: PubSub, jobQueue: JobQueue) {
     this._ethClient = ethClient;
     this._indexer = indexer;
@@ -44,7 +47,10 @@ export class EventWatcher {
   async start (): Promise<void> {
     await this.initBlockProcessingOnCompleteHandler();
     await this.initEventProcessingOnCompleteHandler();
+
     this.startBlockProcessing();
+
+    this.handleShutdown();
   }
 
   async initBlockProcessingOnCompleteHandler (): Promise<void> {
@@ -85,12 +91,33 @@ export class EventWatcher {
     for await (const data of blockProgressEventIterable) {
       const { onBlockProgressEvent: { blockNumber, isComplete } } = data;
 
+      if (this._shutDown) {
+        log(`Graceful shutdown after processing block ${blockNumber}`);
+        process.exit(0);
+      }
+
       if (isComplete) {
         await processBlockByNumber(this._jobQueue, blockNumber + 1);
       }
     }
   }
 
+  handleShutdown (): void {
+    process.on('SIGINT', this._processShutdown.bind(this));
+    process.on('SIGTERM', this._processShutdown.bind(this));
+  }
+
+  async _processShutdown (): Promise<void> {
+    this._shutDown = true;
+    this._signalCount++;
+
+    if (this._signalCount >= 3 || process.env.YARN_CHILD_PROCESS === 'true') {
+      // Forceful exit on receiving signal for the 3rd time or if job-runner is a child process of yarn.
+      log('Forceful shutdown');
+      process.exit(1);
+    }
+  }
+
   async blockProcessingCompleteHandler (job: any): Promise<void> {
     const { id, data: { failed, request: { data } } } = job;
     const { kind } = data;

packages/util/src/job-runner.ts

@@ -39,9 +39,10 @@ export class JobRunner {
   _jobQueueConfig: JobQueueConfig
   _blockProcessStartTime?: Date
   _endBlockProcessTimer?: () => void
+  _blockAndEventsMap: Map<string, PrefetchedBlock> = new Map()
+
   _shutDown = false
   _signalCount = 0
-  _blockAndEventsMap: Map<string, PrefetchedBlock> = new Map()
 
   constructor (jobQueueConfig: JobQueueConfig, indexer: IndexerInterface, jobQueue: JobQueue) {
     this._indexer = indexer;
@@ -220,6 +221,7 @@ export class JobRunner {
 
     if (this._signalCount >= 3 || process.env.YARN_CHILD_PROCESS === 'true') {
       // Forceful exit on receiving signal for the 3rd time or if job-runner is a child process of yarn.
+      log('Forceful shutdown');
       this.jobQueue.stop();
       process.exit(1);
     }