Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Gracefully shutdown watcher job-runner #417

Merged
merged 1 commit into from
Nov 25, 2022
Merged

Gracefully shutdown watcher job-runner #417

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions packages/erc20-watcher/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,8 @@
"build": "tsc",
"server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
"server:dev": "DEBUG=vulcanize:* nodemon --watch src src/server.ts",
"job-runner": "DEBUG=vulcanize:* node --enable-source-maps dist/job-runner.js",
"job-runner:dev": "DEBUG=vulcanize:* nodemon --watch src src/job-runner.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",
"watch:contract:dev": "ts-node src/cli/watch-contract.ts",
"fill": "DEBUG=vulcanize:* node dist/fill.js",
Expand Down
6 changes: 3 additions & 3 deletions packages/uni-info-watcher/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,9 +36,9 @@
"server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
"server:prof": "DEBUG=vulcanize:* node --require pprof --enable-source-maps dist/server.js",
"server:dev": "DEBUG=vulcanize:* nodemon --watch src src/server.ts",
"job-runner": "DEBUG=vulcanize:* node --enable-source-maps dist/job-runner.js",
"job-runner:prof": "DEBUG=vulcanize:* node --require pprof --enable-source-maps dist/job-runner.js",
"job-runner:dev": "DEBUG=vulcanize:* nodemon --watch src src/job-runner.ts",
"job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
"job-runner:prof": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --require pprof --enable-source-maps dist/job-runner.js",
"job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
"smoke-test": "yarn test:init && mocha src/smoke.test.ts",
"fill": "DEBUG=vulcanize:* node --enable-source-maps dist/fill.js",
"fill:prof": "DEBUG=vulcanize:* node --require pprof --enable-source-maps dist/fill.js",
Expand Down
5 changes: 0 additions & 5 deletions packages/uni-info-watcher/src/job-runner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,3 @@ main().then(() => {
process.on('uncaughtException', err => {
log('uncaughtException', err);
});

process.on('SIGINT', () => {
log(`Exiting process ${process.pid} with code 0`);
process.exit(0);
});
6 changes: 3 additions & 3 deletions packages/uni-watcher/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,9 @@
"server": "DEBUG=vulcanize:* node --enable-source-maps dist/server.js",
"server:prof": "DEBUG=vulcanize:* node --require pprof --enable-source-maps dist/server.js",
"server:dev": "DEBUG=vulcanize:* nodemon --watch src src/server.ts",
"job-runner": "DEBUG=vulcanize:* node --enable-source-maps dist/job-runner.js",
"job-runner:prof": "DEBUG=vulcanize:* node --require pprof --enable-source-maps dist/job-runner.js",
"job-runner:dev": "DEBUG=vulcanize:* nodemon --watch src src/job-runner.ts",
"job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
"job-runner:prof": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --require pprof --enable-source-maps dist/job-runner.js",
"job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
"smoke-test": "yarn test:init && mocha src/smoke.test.ts",
"fill": "DEBUG=vulcanize:* node --enable-source-maps dist/fill.js",
"fill:prof": "DEBUG=vulcanize:* node --require pprof --enable-source-maps dist/fill.js",
Expand Down
5 changes: 0 additions & 5 deletions packages/uni-watcher/src/job-runner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,3 @@ main().then(() => {
process.on('uncaughtException', err => {
log('uncaughtException', err);
});

process.on('SIGINT', () => {
log(`Exiting process ${process.pid} with code 0`);
process.exit(0);
});
29 changes: 29 additions & 0 deletions packages/util/src/job-runner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,9 @@ export class JobRunner {

_blockAndEventsMap: Map<string, PrefetchedBlock> = new Map()

_shutDown = false
_signalCount = 0

constructor (jobQueueConfig: JobQueueConfig, indexer: IndexerInterface, jobQueue: JobQueue) {
this._jobQueueConfig = jobQueueConfig;
this._indexer = indexer;
Expand All @@ -50,8 +53,11 @@ export class JobRunner {
async start (): Promise<void> {
await this._jobQueue.deleteAllJobs();
await this.resetToPrevIndexedBlock();

await this.subscribeBlockProcessingQueue();
await this.subscribeEventProcessingQueue();

this.handleShutdown();
}

async subscribeBlockProcessingQueue (): Promise<void> {
Expand Down Expand Up @@ -124,6 +130,23 @@ export class JobRunner {
}
}

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');
this._jobQueue.stop();
process.exit(1);
}
}

async _pruneChain (job: any): Promise<void> {
console.time('time:job-runner#_pruneChain');

Expand Down Expand Up @@ -449,6 +472,12 @@ export class JobRunner {
log('size:job-runner#_processEvents-_blockAndEventsMap:', this._blockAndEventsMap.size);

console.timeEnd('time:job-runner#_processEvents-events');

if (this._shutDown) {
log(`Graceful shutdown after processing block ${block.blockNumber}`);
this._jobQueue.stop();
process.exit(0);
}
}

_updateWatchedContracts (job: any): void {
Expand Down