Added Job lifecycle callbacks functionality.

Author: billmalarky

Models/Queue.js

@@ -308,7 +308,9 @@ export class Queue {
 
   /**
    *
-   * Execute a job.
+   * Process a job.
+   *
+   * Job lifecycle callbacks are called as appropriate throughout the job processing lifecycle.
    *
    * Job is deleted upon successful completion.
    *
@@ -321,25 +323,39 @@ export class Queue {
    */
   async processJob(job) {
 
+    // Data must be cloned off the realm job object for several lifecycle callbacks to work correctly.
+    // This is because realm job is deleted before some callbacks are called if job processed successfully.
+    // More info: https://github.com/billmalarky/react-native-queue/issues/2#issuecomment-361418965
+    const jobName = job.name;
+    const jobId = job.id;
+    const jobPayload = JSON.parse(job.payload);
+
+    // Fire onStart job lifecycle callback
+    this.worker.executeJobLifecycleCallback('onStart', jobName, jobId, jobPayload);
+
     try {
 
       await this.worker.executeJob(job);
 
-      // On job completion, remove job
+      // On successful job completion, remove job
       this.realm.write(() => {
 
         this.realm.delete(job);
 
       });
 
+      // Job has processed successfully, fire onSuccess and onComplete job lifecycle callbacks.
+      this.worker.executeJobLifecycleCallback('onSuccess', jobName, jobId, jobPayload);
+      this.worker.executeJobLifecycleCallback('onComplete', jobName, jobId, jobPayload);
+
     } catch (error) {
 
       // Handle job failure logic, including retries.
+      let jobData = JSON.parse(job.data);
+
       this.realm.write(() => {
 
         // Increment failed attempts number
-        let jobData = JSON.parse(job.data);
-
         if (!jobData.failedAttempts) {
           jobData.failedAttempts = 1;
         } else {
@@ -365,6 +381,15 @@ export class Queue {
 
       });
 
+      // Execute job onFailure lifecycle callback.
+      this.worker.executeJobLifecycleCallback('onFailure', jobName, jobId, jobPayload);
+
+      // If job has failed all attempts execute job onFailed and onComplete lifecycle callbacks.
+      if (jobData.failedAttempts >= jobData.attempts) {
+        this.worker.executeJobLifecycleCallback('onFailed', jobName, jobId, jobPayload);
+        this.worker.executeJobLifecycleCallback('onComplete', jobName, jobId, jobPayload);
+      }
+
     }
 
   }

Models/Worker.js

@@ -41,7 +41,12 @@ export default class Worker {
 
     // Attach options to worker
     worker.options = {
-      concurrency: options.concurrency || 1
+      concurrency: options.concurrency || 1,
+      onStart: options.onStart || null,
+      onSuccess: options.onSuccess || null,
+      onFailure: options.onFailure || null,
+      onFailed: options.onFailed || null,
+      onComplete: options.onComplete || null
     };
 
     Worker.workers[jobName] = worker;
@@ -119,4 +124,35 @@ export default class Worker {
 
   }
 
+  /**
+   *
+   * Execute an asynchronous job lifecycle callback associated with related worker.
+   *
+   * @param callbackName {string} - Job lifecycle callback name.
+   * @param jobName {string} - Name associated with jobs assigned to related worker.
+   * @param jobId {string} - Unique id associated with job.
+   * @param jobPayload {object} - Data payload associated with job.
+   */
+  async executeJobLifecycleCallback(callbackName, jobName, jobId, jobPayload) {
+
+    // Validate callback name
+    const validCallbacks = ['onStart', 'onSuccess', 'onFailure', 'onFailed', 'onComplete'];
+    if (!validCallbacks.includes(callbackName)) {
+      throw new Error('Invalid job lifecycle callback name.');
+    }
+
+    // Fire job lifecycle callback if set.
+    // Uses a try catch statement to gracefully degrade errors in production.
+    if (Worker.workers[jobName].options[callbackName]) {
+
+      try {
+        await Worker.workers[jobName].options[callbackName](jobId, jobPayload);
+      } catch (error) {
+        console.error(error); // eslint-disable-line no-console
+      }
+
+    }
+
+  }
+
 }

README.md

@@ -17,7 +17,7 @@ A React Native at-least-once priority job queue / task queue backed by persisten
 * [Example Use Cases](#example-use-cases)
 * [Installation](#installation)
 * [Basic Usage](#basic-usage)
-* [Options](#options)
+* [Options and Job Lifecycle Callbacks](#options-and-job-lifecycle-callbacks)
 * [Testing with Jest](#testing-with-jest)
 * [Caveats](#caveats)
 * [Advanced Usage Examples](#advanced-usage-examples)
@@ -30,7 +30,7 @@ A React Native at-least-once priority job queue / task queue backed by persisten
 * **Simple API:** Set up job workers and begin creating your jobs in minutes with just two basic API calls
   * queue.addWorker(name, workerFunction, options = {})  
   * queue.createJob(name, payload = {}, options = {}, startQueue = true) 
-* **Powerful options:** Easily modify default functionality. Set job timeouts, number of retry attempts, priority, and worker concurrency with an options object. Start queue processing with a lifespan to easily meet OS background task time limits.
+* **Powerful options:** Easily modify default functionality. Set job timeouts, number of retry attempts, priority, job lifecycle callbacks, and worker concurrency with an options object. Start queue processing with a lifespan to easily meet OS background task time limits.
 * **Persistent Jobs:** Jobs are persisted with Realm. Because jobs persist, you can easily continue to process jobs across app restarts or in OS background tasks until completed or failed (or app is uninstalled).
 * **Powerful Integrations:** React Native Queue was designed to play well with others. The queue quickly integrates with a variety of OS background task and Worker packages so  processing your jobs in a background service or dedicated thread have never been easier.
 
@@ -147,19 +147,63 @@ console.log('The above jobs are processing in the background of app now.');
 
 ```
 
-## Options
+## Options and Job Lifecycle Callbacks
 
-#### Worker Options
+#### Worker Options (includes async job lifecycle callbacks)
 
-queue.addWorker() accepts an options object in order to tweak standard functionality.
+queue.addWorker() accepts an options object in order to tweak standard functionality and allow you to hook into asynchronous job lifecycle callbacks.
+
+**IMPORTANT: Job Lifecycle callbacks are called asynchronously.** They do not block job processing or each other. Don't put logic in onStart that you expect to be completed before the actual job process begins executing. Don't put logic in onFailure you expect to be completed before onFailed is called. You can, of course, assume that the job process has completed (or failed) before onSuccess, onFailure, onFailed, or onComplete are asynchonrously called.
 
 ```js
 
-queue.addWorker('job-name-here', (id, payload) => { console.log(id); }, {
+queue.addWorker('job-name-here', async (id, payload) => { console.log(id); }, {
 
   // Set max number of jobs for this worker to process concurrently.
   // Defaults to 1.
-  concurrency: 5
+  concurrency: 5,
+  
+  // JOB LIFECYCLE CALLBACKS
+  
+  // onStart job callback handler is fired when a job begins processing.
+  //
+  // IMPORTANT: Job lifecycle callbacks are executed asynchronously and do not block job processing 
+  // (even if the callback returns a promise it will not be "awaited" on).
+  // As such, do not place any logic in onStart that your actual job worker function will depend on,
+  // this type of logic should of course go inside the job worker function itself.
+  onStart: async (id, payload) => {
+    
+    console.log('Job "job-name-here" with id ' + id + ' has started processing.');  
+    
+  },
+  
+  // onSuccess job callback handler is fired after a job successfully completes processing.
+  onSuccess: async (id, payload) => {
+    
+    console.log('Job "job-name-here" with id ' + id + ' was successful.');
+    
+  },
+  
+  // onFailure job callback handler is fired after each time a job fails (onFailed also fires if job has reached max number of attempts).
+  onFailure: async (id, payload) => {
+    
+    console.log('Job "job-name-here" with id ' + id + ' had an attempt end in failure.');
+    
+  },
+  
+  // onFailed job callback handler is fired if job fails enough times to reach max number of attempts.
+  onFailed: async (id, payload) => {
+    
+    console.log('Job "job-name-here" with id ' + id + ' has failed.');
+    
+  },
+  
+  // onComplete job callback handler fires after job has completed processing successfully or failed entirely.
+  onComplete: async (id, payload) => {
+    
+    console.log('Job "job-name-here" with id ' + id + ' has completed processing.');
+    
+  }
 
 }); 
 

tests/Queue.test.js

@@ -700,13 +700,23 @@ describe('Models/Queue', function() {
   it('#addWorker() and removeWorker() should pass calls through to Worker class', async () => {
 
     const queue = await QueueFactory();
-    const workerOptions = { concurrency: 4 };
+    const workerOptions = {
+      concurrency: 4,
+      onSuccess: async (id, payload) => {}
+    };
 
     queue.addWorker('job-name', () => {}, workerOptions);
 
     // first worker is added with default options.
     Worker.workers['job-name'].should.be.a.Function();
-    Worker.workers['job-name'].options.should.deepEqual(workerOptions);
+    Worker.workers['job-name'].options.should.deepEqual({
+      concurrency: workerOptions.concurrency,
+      onStart: null,
+      onSuccess: workerOptions.onSuccess,
+      onFailure: null,
+      onFailed: null,
+      onComplete: null
+    });
 
     queue.removeWorker('job-name');
 
@@ -1581,4 +1591,489 @@ describe('Models/Queue', function() {
 
   });
 
+  ////
+  //// JOB LIFECYCLE CALLBACK TESTING
+  ////
+
+  it('onStart lifecycle callback fires before job begins processing.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let jobProcessed = false;
+    let testFailed = false;
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      // Timeout needed because onStart runs async so we need to ensure this function gets
+      // executed last.
+      await new Promise((resolve) => {
+        setTimeout(() => {
+          jobProcessed = true;
+          resolve();
+        }, 0);
+      });
+
+    }, {
+      onStart: (id, payload) => {
+
+        // If onStart runs after job has processed, fail test.
+        if (jobProcessed) {
+          testFailed = true;
+          throw new Error('ERROR: onStart fired after job began processing.')
+        }
+
+      }
+    });
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data' }, {}, false);
+
+    jobProcessed.should.equal(false);
+    testFailed.should.equal(false);
+    await queue.start();
+    jobProcessed.should.equal(true);
+    testFailed.should.equal(false);
+
+  });
+
+  it('onSuccess, onComplete lifecycle callbacks fire after job begins processing.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let jobProcessed = false;
+    let testFailed = false;
+    let onSuccessFired = false;
+    let onCompleteFired = false;
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      // Simulate work
+      await new Promise((resolve) => {
+        setTimeout(() => {
+          jobProcessed = true;
+          resolve();
+        }, 300);
+      });
+
+    }, {
+      onSuccess: (id, payload) => {
+
+        onSuccessFired = true;
+
+        // If onSuccess runs before job has processed, fail test.
+        if (!jobProcessed) {
+          testFailed = true;
+          throw new Error('ERROR: onSuccess fired before job began processing.')
+        }
+
+      },
+      onComplete: (id, payload) => {
+
+        onCompleteFired = true;
+
+        // If onComplete runs before job has processed, fail test.
+        if (!jobProcessed) {
+          testFailed = true;
+          throw new Error('ERROR: onComplete fired before job began processing.')
+        }
+
+      }
+    });
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data' }, {}, false);
+
+    jobProcessed.should.equal(false);
+    testFailed.should.equal(false);
+    onSuccessFired.should.equal(false);
+    onCompleteFired.should.equal(false);
+    await queue.start();
+    jobProcessed.should.equal(true);
+    testFailed.should.equal(false);
+    onSuccessFired.should.equal(true);
+    onCompleteFired.should.equal(true);
+
+  });
+
+  it('onFailure, onFailed lifecycle callbacks fire after job begins processing.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let jobProcessStarted = false;
+    let testFailed = false;
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      // Simulate work
+      await new Promise((resolve, reject) => {
+        setTimeout(() => {
+          jobProcessStarted = true;
+          reject(new Error('Job failed.'));
+        }, 300);
+      });
+
+    }, {
+      onFailure: (id, payload) => {
+
+        // If onFailure runs before job has processed, fail test.
+        if (!jobProcessStarted) {
+          testFailed = true;
+          throw new Error('ERROR: onFailure fired before job began processing.')
+        }
+
+      },
+      onFailed: (id, payload) => {
+
+        // If onFailed runs before job has processed, fail test.
+        if (!jobProcessStarted) {
+          testFailed = true;
+          throw new Error('ERROR: onFailed fired before job began processing.')
+        }
+
+      }
+    });
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data' }, {}, false);
+
+    jobProcessStarted.should.equal(false);
+    testFailed.should.equal(false);
+    await queue.start();
+    jobProcessStarted.should.equal(true);
+    testFailed.should.equal(false);
+
+  });
+
+  it('onFailure, onFailed lifecycle callbacks work as expected.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let jobAttemptCounter = 0;
+    let onFailureFiredCounter = 0;
+    let onFailedFiredCounter = 0;
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      // Simulate work
+      await new Promise((resolve, reject) => {
+        setTimeout(() => {
+          jobAttemptCounter++;
+          reject(new Error('Job failed.'));
+        }, 0);
+      });
+
+    }, {
+
+      onFailure: (id, payload) => {
+
+        onFailureFiredCounter++;
+
+      },
+      onFailed: (id, payload) => {
+
+        onFailedFiredCounter++;
+
+      }
+    });
+
+    const attempts = 3;
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data' }, {
+      attempts
+    }, false);
+
+    jobAttemptCounter.should.equal(0);
+    await queue.start();
+    onFailureFiredCounter.should.equal(attempts);
+    onFailedFiredCounter.should.equal(1);
+    jobAttemptCounter.should.equal(attempts);
+
+  });
+
+  it('onComplete fires only once on job with multiple attempts that ends in success.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let jobAttemptCounter = 0;
+    let onFailureFiredCounter = 0;
+    let onFailedFiredCounter = 0;
+    let onCompleteFiredCounter = 0;
+    const attempts = 3;
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      jobAttemptCounter++;
+
+      // Keep failing attempts until last attempt then success.
+      if (jobAttemptCounter < attempts) {
+
+        // Simulate work that fails
+        await new Promise((resolve, reject) => {
+          setTimeout(() => {
+            reject(new Error('Job failed.'));
+          }, 0);
+        });
+
+      } else {
+
+        // Simulate work that succeeds
+        await new Promise((resolve, reject) => {
+          setTimeout(() => {
+            resolve();
+          }, 0);
+        });
+
+      }
+
+    }, {
+
+      onFailure: (id, payload) => {
+
+        onFailureFiredCounter++;
+
+      },
+      onFailed: (id, payload) => {
+
+        onFailedFiredCounter++;
+
+      },
+      onComplete: (id, payload) => {
+
+        onCompleteFiredCounter++;
+
+      }
+    });
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data succes' }, {
+      attempts
+    }, false);
+
+    jobAttemptCounter.should.equal(0);
+    await queue.start();
+    onFailureFiredCounter.should.equal(attempts - 1);
+    onFailedFiredCounter.should.equal(0);
+    jobAttemptCounter.should.equal(attempts);
+    onCompleteFiredCounter.should.equal(1);
+
+  });
+
+  it('onComplete fires only once on job with multiple attempts that ends in failure.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let jobAttemptCounter = 0;
+    let onFailureFiredCounter = 0;
+    let onFailedFiredCounter = 0;
+    let onCompleteFiredCounter = 0;
+    const attempts = 3;
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      jobAttemptCounter++;
+
+      // Simulate work that fails
+      await new Promise((resolve, reject) => {
+        setTimeout(() => {
+          reject(new Error('Job failed.'));
+        }, 0);
+      });
+
+    }, {
+
+      onFailure: (id, payload) => {
+
+        onFailureFiredCounter++;
+
+      },
+      onFailed: (id, payload) => {
+
+        onFailedFiredCounter++;
+
+      },
+      onComplete: (id, payload) => {
+
+        onCompleteFiredCounter++;
+
+      }
+    });
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data' }, {
+      attempts
+    }, false);
+
+    jobAttemptCounter.should.equal(0);
+    await queue.start();
+    onFailureFiredCounter.should.equal(attempts);
+    onFailedFiredCounter.should.equal(1);
+    jobAttemptCounter.should.equal(attempts);
+    onCompleteFiredCounter.should.equal(1);
+
+  });
+
+  it('onStart, onSuccess, onComplete Job lifecycle callbacks do not block job processing.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let workTracker = [];
+    let tracker = [];
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      // Simulate work
+      await new Promise((resolve) => {
+        workTracker.push(payload.random);
+        tracker.push('job processed');
+        setTimeout(resolve, 0);
+      });
+
+    }, {
+
+      onStart: async (id, payload) => {
+
+        // wait a bit
+        await new Promise((resolve) => {
+          setTimeout(() => {
+            tracker.push('onStart completed.');
+          }, 1000);
+        });
+
+      },
+      onSuccess: async (id, payload) => {
+
+        // wait a bit
+        await new Promise((resolve) => {
+          setTimeout(() => {
+            tracker.push('onSuccess completed.');
+          }, 1000);
+        });
+
+      },
+      onComplete: async (id, payload) => {
+
+        // wait a bit
+        await new Promise((resolve) => {
+          setTimeout(() => {
+            tracker.push('onComplete completed.');
+          }, 1000);
+        });
+
+      }
+    });
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 2nd random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 3rd random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 4th random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 5th random data' }, {}, false);
+
+    await queue.start();
+
+    // Ensure all jobs processed.
+    workTracker.should.containDeep([
+      'this is 1st random data',
+      'this is 2nd random data',
+      'this is 4th random data',
+      'this is 3rd random data',
+      'this is 5th random data'
+    ]);
+
+    // Since lifecycle callbacks take a second to process,
+    // queue should churn through all jobs well before any of the lifecycle
+    // callbacks complete.
+    const firstFive = tracker.slice(0, 5);
+    firstFive.should.deepEqual([
+      'job processed',
+      'job processed',
+      'job processed',
+      'job processed',
+      'job processed'
+    ]);
+
+  });
+
+  it('onFailure, onFailed Job lifecycle callbacks do not block job processing.', async () => {
+
+    const queue = await QueueFactory();
+    queue.flushQueue();
+    const jobName = 'job-name';
+    let workTracker = [];
+    let tracker = [];
+
+    queue.addWorker(jobName, async (id, payload) => {
+
+      // Simulate failure
+      await new Promise((resolve, reject) => {
+        workTracker.push(payload.random);
+        setTimeout(() => {
+          tracker.push('job attempted');
+          reject(new Error('job failed'));
+        }, 0);
+      });
+
+    }, {
+      onFailure: async (id, payload) => {
+
+        // wait a bit
+        await new Promise((resolve) => {
+          setTimeout(() => {
+            tracker.push('onFailure completed.');
+          }, 1000);
+        });
+
+      },
+      onFailed: async (id, payload) => {
+
+        // wait a bit
+        await new Promise((resolve) => {
+          setTimeout(() => {
+            tracker.push('onFailed completed.');
+          }, 1000);
+        });
+
+      }
+    });
+
+    // Create a job
+    queue.createJob(jobName, { random: 'this is 1st random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 2nd random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 3rd random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 4th random data' }, {}, false);
+    queue.createJob(jobName, { random: 'this is 5th random data' }, {}, false);
+
+    await queue.start();
+
+    // Ensure all jobs started to process (even though they are failed).
+    workTracker.should.containDeep([
+      'this is 1st random data',
+      'this is 2nd random data',
+      'this is 4th random data',
+      'this is 3rd random data',
+      'this is 5th random data'
+    ]);
+
+    // Since lifecycle callbacks take a second to process,
+    // queue should churn through all jobs well before any of the lifecycle
+    // callbacks complete.
+    const firstFive = tracker.slice(0, 5);
+    firstFive.should.deepEqual([
+      'job attempted',
+      'job attempted',
+      'job attempted',
+      'job attempted',
+      'job attempted'
+    ]);
+
+  });
+
 });

tests/Worker.test.js

@@ -35,19 +35,32 @@ describe('Models/Worker', function() {
     worker.addWorker('test-job-one', async () => {});
 
     const workerOptions = {
-      concurrency: 3
+      concurrency: 3,
+      onStart: async (id, payload) => {}
     };
     worker.addWorker('test-job-two', async () => {}, workerOptions);
 
     // first worker is added with default options.
     Worker.workers['test-job-one'].should.be.a.Function();
     Worker.workers['test-job-one'].options.should.deepEqual({
-      concurrency: 1
+      concurrency: 1,
+      onStart: null,
+      onSuccess: null,
+      onFailure: null,
+      onFailed: null,
+      onComplete: null
     });
 
     // second worker is added with new concurrency option.
     Worker.workers['test-job-two'].should.be.a.Function();
-    Worker.workers['test-job-two'].options.should.deepEqual(workerOptions);
+    Worker.workers['test-job-two'].options.should.deepEqual({
+      concurrency: workerOptions.concurrency,
+      onStart: workerOptions.onStart,
+      onSuccess: null,
+      onFailure: null,
+      onFailed: null,
+      onComplete: null
+    });
 
   });
 
@@ -195,4 +208,153 @@ describe('Models/Worker', function() {
 
   });
 
+  it('#executeJobLifecycleCallback() should execute a job lifecycle method correctly.', async () => {
+
+    let onStartCalled = false;
+    let testPassed = false;
+
+    const job = {
+      id: 'd21dca87-435c-4533-b0af-ed9844e6b827',
+      name: 'test-job-one',
+      payload: JSON.stringify({
+        key: 'value'
+      }),
+      data: JSON.stringify({
+        timeout: 0,
+        attempts: 1
+      }),
+      priority: 0,
+      active: false,
+      created: new Date(),
+      failed: null
+    };
+
+    const worker = new Worker();
+
+    worker.addWorker('test-job-one', async () => {}, {
+      onStart: (id, payload) => {
+
+        onStartCalled = true;
+
+        // Verify params passed correctly off job and payload JSON has been parsed.
+        id.should.equal(job.id);
+        payload.should.deepEqual({
+          key: 'value'
+        });
+
+        // Explicitly mark test as passed because the assertions
+        // directly above will be caught in the try/catch statement within
+        // executeJobLifecycleCallback() if they throw an error. While any thrown errors will be
+        // output to console, the test will still pass so won't be caught by CI testing.
+        testPassed = true;
+
+      }
+    });
+
+    onStartCalled.should.equal(false);
+    const payload = JSON.parse(job.payload); // Payload JSON is always parsed by Queue model before passing to executeJobLifecycleCallback();
+    await worker.executeJobLifecycleCallback('onStart', job.name, job.id, payload);
+    onStartCalled.should.equal(true);
+    testPassed.should.equal(true);
+
+  });
+
+  it('#executeJobLifecycleCallback() should throw an error on invalid job lifecycle name.', async () => {
+
+    let onStartCalled = false;
+    let testPassed = true;
+
+    const job = {
+      id: 'd21dca87-435c-4533-b0af-ed9844e6b827',
+      name: 'test-job-one',
+      payload: JSON.stringify({
+        key: 'value'
+      }),
+      data: JSON.stringify({
+        timeout: 0,
+        attempts: 1
+      }),
+      priority: 0,
+      active: false,
+      created: new Date(),
+      failed: null
+    };
+
+    const worker = new Worker();
+
+    worker.addWorker('test-job-one', async () => {}, {
+      onStart: (id, payload) => {
+
+        testPassed = false;
+        throw new Error('Should not be called.');
+
+      }
+    });
+
+    onStartCalled.should.equal(false);
+    const payload = JSON.parse(job.payload); // Payload JSON is always parsed by Queue model before passing to executeJobLifecycleCallback();
+    try {
+      await worker.executeJobLifecycleCallback('onInvalidLifecycleName', job.name, job.id, payload);
+    } catch (error) {
+      error.should.deepEqual(new Error('Invalid job lifecycle callback name.'));
+    }
+    onStartCalled.should.equal(false);
+    testPassed.should.equal(true);
+
+  });
+
+  it('#executeJobLifecycleCallback() job lifecycle callbacks that error out should gracefully degrade to console error.', async () => {
+
+    let onStartCalled = false;
+    let consoleErrorCalled = false;
+
+    // Cache console error.
+    const consoleErrorCache = console.error; // eslint-disable-line no-console
+
+    // Overwrite console.error to make sure it gets called on job lifecycle
+    // callback error and is passed the error object.
+    console.error = (errorObject) => {
+      consoleErrorCalled = true;
+      errorObject.should.deepEqual(new Error('Something failed catastrophically!'));
+    };
+
+    const job = {
+      id: 'd21dca87-435c-4533-b0af-ed9844e6b827',
+      name: 'test-job-one',
+      payload: JSON.stringify({
+        key: 'value'
+      }),
+      data: JSON.stringify({
+        timeout: 0,
+        attempts: 1
+      }),
+      priority: 0,
+      active: false,
+      created: new Date(),
+      failed: null
+    };
+
+    const worker = new Worker();
+
+    worker.addWorker('test-job-one', async () => {}, {
+      onStart: (id, payload) => {
+
+        onStartCalled = true;
+        throw new Error('Something failed catastrophically!');
+
+      }
+    });
+
+    onStartCalled.should.equal(false);
+    consoleErrorCalled.should.equal(false);
+    const payload = JSON.parse(job.payload); // Payload JSON is always parsed by Queue model before passing to executeJobLifecycleCallback();
+    await worker.executeJobLifecycleCallback('onStart', job.name, job.id, payload);
+    onStartCalled.should.equal(true);
+    consoleErrorCalled.should.equal(true);
+
+    // Re-apply console.error.
+    console.error = consoleErrorCache;
+
+  });
+
 });