Added execution progress updates for one off jobs

refs https://github.com/TryGhost/Toolbox/issues/357

- Job persisted in the database need to track job's execution status such as completion, failure, execution start and end times. This changeset allows to hook into job/bree lifecycle to track job's progress.
- NOTE: only supports "offloaded" jobs at the moment. Support for "inline" jobs will be added once there's a clear usecase for it.
- The "started" status and "started_at" timestamp are assigned to a job at the moment when the worker thread is created inside of bree
- The "finished" status and "finished_at" timestamp are assigned to a job when a "done" event is passed from the job script (NOTE: using process.exit(0) will not trigger the "finished" state")
- The "failed" status is assigned when the job execution is interrupted with an error

ghost/job-manager/lib/job-manager.js

@@ -36,19 +36,71 @@ class JobManager {
      */
     constructor({errorHandler, workerMessageHandler, JobModel}) {
         this.queue = fastq(this, worker, 1);
+        this._jobMessageHandler = this._jobMessageHandler.bind(this);
+        this._jobErrorHandler = this._jobErrorHandler.bind(this);
+
+        const combinedMessageHandler = workerMessageHandler
+            ? ({name, message}) => {
+                workerMessageHandler({name, message});
+                this._jobMessageHandler({name, message});
+            }
+            : this._jobMessageHandler;
+
+        const combinedErrorHandler = errorHandler
+            ? (error, workerMeta) => {
+                errorHandler(error, workerMeta);
+                this._jobErrorHandler(error, workerMeta);
+            }
+            : this._jobErrorHandler;
 
         this.bree = new Bree({
             root: false, // set this to `false` to prevent requiring a root directory of jobs
             hasSeconds: true, // precision is needed to avoid task overlaps after immediate execution
             outputWorkerMetadata: true,
             logger: logging,
-            errorHandler: errorHandler,
-            workerMessageHandler: workerMessageHandler
+            errorHandler: combinedErrorHandler,
+            workerMessageHandler: combinedMessageHandler
         });
 
+        this.bree.on('worker created', (name) => {
+            this._jobMessageHandler({name, message: 'started'});
+          });
+
         this._jobsRepository = new JobsRepository({JobModel});
     }
 
+    async _jobMessageHandler({name, message}) {
+        if (message === 'started') {
+            const job = await this._jobsRepository.read(name);
+
+            if (job) {
+                await this._jobsRepository.update(job.id, {
+                    status: 'started',
+                    started_at: new Date()
+                });
+            }
+        } else if (message === 'done') {
+            const job = await this._jobsRepository.read(name);
+
+            if (job) {
+                await this._jobsRepository.update(job.id, {
+                    status: 'finished',
+                    finished_at: new Date()
+                });
+            }
+        }
+    }
+
+    async _jobErrorHandler(error, workerMeta) {
+        const job = await this._jobsRepository.read(workerMeta.name);
+
+        if (job) {
+            await this._jobsRepository.update(job.id, {
+                status: 'failed'
+            });
+        }
+    }
+
     /**
      * By default schedules an "offloaded" job. If `offloaded: true` parameter is set,
      * puts an "inline" immediate job into the queue.

ghost/job-manager/lib/jobs-repository.js

@@ -14,6 +14,10 @@ class JobsRepository {
 
         return job;
     }
+
+    async update(id, data) {
+        await this._JobModel.edit(data, {id});
+    }
 }
 
 module.exports = JobsRepository;

ghost/job-manager/test/job-manager.test.js

@@ -290,6 +290,84 @@ describe('Job Manager', function () {
                 assert.equal(error.message, 'A "I am the only one" one off job has already been executed.');
             }
         });
+
+        it('sets a finished state on a job', async function () {
+            const JobModel = {
+                findOne: sinon.stub()
+                    .onCall(0)
+                    .resolves(null)
+                    .resolves({id: 'unique', name: 'successful-oneoff'}),
+                add: sinon.stub().resolves({name: 'successful-oneoff'}),
+                edit: sinon.stub().resolves({name: 'successful-oneoff'})
+            };
+
+            const jobManager = new JobManager({JobModel});
+
+            jobManager.addOneOffJob({
+                job: path.resolve(__dirname, './jobs/message.js'),
+                name: 'successful-oneoff'
+            });
+
+            // allow job to get picked up and executed
+            await delay(100);
+
+            jobManager.bree.workers['successful-oneoff'].postMessage('be done!');
+
+            // allow the message to be passed around
+            await delay(100);
+
+            // tracks the job start
+            should(JobModel.edit.args[0][0].status).equal('started');
+            should(JobModel.edit.args[0][0].started_at).not.equal(undefined);
+            should(JobModel.edit.args[0][1].id).equal('unique');
+
+            // tracks the job finish
+            should(JobModel.edit.args[1][0].status).equal('finished');
+            should(JobModel.edit.args[1][0].finished_at).not.equal(undefined);
+            should(JobModel.edit.args[1][1].id).equal('unique');
+        });
+
+        it('sets a failed state on a job', async function () {
+            const JobModel = {
+                findOne: sinon.stub()
+                    .onCall(0)
+                    .resolves(null)
+                    .resolves({id: 'unique', name: 'failed-oneoff'}),
+                add: sinon.stub().resolves({name: 'failed-oneoff'}),
+                edit: sinon.stub().resolves({name: 'failed-oneoff'})
+            };
+
+            let job = function namedJob() {
+                throw new Error('job error');
+            };
+            const spyHandler = sinon.spy();
+            const jobManager = new JobManager({errorHandler: spyHandler, JobModel});
+
+            jobManager.addOneOffJob({
+                job,
+                name: 'failed-oneoff'
+            });
+
+            // give time to execute the job
+            // has to be this long because in Node v10 the communication is
+            // done through processes, which takes longer comparing to worker_threads
+            // can be reduced to 100 when Node v10 support is dropped
+            await delay(100);
+
+            // still calls the original error handler
+            should(spyHandler.called).be.true();
+            should(spyHandler.args[0][0].message).equal('job error');
+            should(spyHandler.args[0][1].name).equal('failed-oneoff');
+
+            // tracks the job start
+            should(JobModel.edit.args[0][0].status).equal('started');
+            should(JobModel.edit.args[0][0].started_at).not.equal(undefined);
+            should(JobModel.edit.args[0][1].id).equal('unique');
+
+            // tracks the job failure
+            should(JobModel.edit.args[1][0].status).equal('failed');
+            should(JobModel.edit.args[1][1].id).equal('unique');
+        });
     });
 
     describe('Remove a job', function () {

ghost/job-manager/test/jobs/message.js

@@ -15,6 +15,6 @@ if (parentPort) {
 
         // post the message back
         parentPort.postMessage(`Worker received: ${message}`);
-        process.exit(0);
+        parentPort.postMessage('done');
     });
 }