Call afterPerform in performInline on error (#773)

This PR makes it so `worker.performInline` calls Plugins' `afterPerform`
method when the `perform` function throws an error.

__tests__/core/worker.ts

@@ -1,6 +1,22 @@
-import { ParsedFailedJobPayload, Job, Queue, Worker } from "../../src";
+import { ParsedFailedJobPayload, Job, Queue, Worker, Plugin } from "../../src";
 import specHelper from "../utils/specHelper";
 
+class MyPlugin extends Plugin {
+  async beforeEnqueue() {
+    return true;
+  }
+  async afterEnqueue() {
+    return true;
+  }
+  async beforePerform() {
+    return true;
+  }
+  async afterPerform() {
+    this.options.afterPerform(this);
+    return true;
+  }
+}
+
 const jobs: { [key: string]: Job<any> } = {
   add: {
     perform: (a, b) => {
@@ -88,6 +104,30 @@ describe("worker", () => {
         expect(String(error)).toBe("Error: Blue Smoke");
       }
     });
+
+    test("can call a Plugin's afterPerform given the job throws an error", async () => {
+      let actual = null;
+      let expected = new TypeError("John");
+
+      let failingJob = {
+        plugins: [MyPlugin],
+        pluginOptions: {
+          MyPlugin: {
+            afterPerform: (plugin: Plugin) => {
+              actual = plugin.worker.error;
+              delete plugin.worker.error;
+            },
+          },
+        },
+        perform: (x: string) => {
+          throw new TypeError(x);
+        },
+      };
+      let worker = new Worker({}, { failingJob });
+      await worker.performInline("failingJob", ["John"]);
+
+      expect(actual).toEqual(expected);
+    });
   });
 
   describe("[with connection]", () => {

src/core/worker.ts

@@ -337,6 +337,7 @@ export class Worker extends EventEmitter {
       throw new Error(`Missing Job: "${func}"`);
     }
 
+    let triedAfterPerform = false;
     try {
       toRun = await RunPlugins(
         this,
@@ -350,6 +351,7 @@ export class Worker extends EventEmitter {
         return;
       }
       this.result = await this.jobs[func].perform.apply(this, args);
+      triedAfterPerform = true;
       toRun = await RunPlugins(
         this,
         "afterPerform",
@@ -361,7 +363,24 @@ export class Worker extends EventEmitter {
       return this.result;
     } catch (error) {
       this.error = error;
-      throw error;
+      if (!triedAfterPerform) {
+        try {
+          await RunPlugins(
+            this,
+            "afterPerform",
+            func,
+            this.queue,
+            this.jobs[func],
+            args
+          );
+        } catch (error) {
+          if (error && !this.error) {
+            this.error = error;
+          }
+        }
+      }
+      // Allow afterPerform to clear the error
+      if (this.error) throw this.error;
     }
   }