Bulk test that fails because JobTimeout doesn't kickstart the queue again

queue-bulk.test.js

"use strict";

const assert = require("../utils/assert.js");
const { Queue } = require("../../lib/queue");
const errors = require("../../lib/errors");
const { QueueItem } = require("../../lib/queueItem");
const logger = { log: (level, data) => {} };
const BBPromise = require("bluebird");
BBPromise.config({
  cancellation: true
});

class TestJob extends QueueItem {
  constructor(name, processTime, succesful) {
    super({});
    this.name = name;
    this.time = processTime;
    this.succesful = typeof succesful === "undefined" ? true : succesful;
    this.timer = null;
  }
  get jobId() {
    return this.name;
  }
  cancel() {
    this.timer && clearTimeout(this.timer);
    return BBPromise.resolve();
  }
  process() {
    return new BBPromise((resolve, reject) => {
      if (this.time === undefined) return;
      this.timer = setTimeout(() => {
        if (this.succesful) {
          resolve(this.name);
        } else {
          reject(this.name);
        }
      }, this.time);
    });
  }
}

const metrics = {
  increment: () => {},
  endTiming: () => {},
  gauge: () => {}
};

describe("Queue bulk test", function() {
  this.timeout(100000);

  it("passes the bulk test correctly", done => {
    const tasks = 1000;
    const maxTaskDuration = 1000;
    const concurrency = 40;
    const fails = 0.3;
    const successes = 0.6;
    const timeouts = 0.1;

    function makeJob() {
      const r = Math.random();
      if (r < fails) {
        return new TestJob(
          new Error("Failure"),
          Math.random() * maxTaskDuration,
          false
        );
      } else if (r < fails + successes) {
        return new TestJob(
          "success job",
          Math.random() * maxTaskDuration,
          true
        );
      } else {
        return new TestJob("timeout job", undefined, true);
      }
    }

    const queue = new Queue(
      {
        maxTaskCount: tasks,
        concurrency: concurrency,
        executionTimeout: maxTaskDuration,
        queueTimeout: tasks * maxTaskDuration
      },
      logger,
      metrics
    );

    const stats = {
      success: 0,
      fail: 0,
      timeout: 0,
      other: 0,
      queuetimeout: 0,
      queuefull: 0,
      cancelled: 0
    };

    const promises = Array(tasks)
      .fill(null)
      .map((_, i) => {
        const promise = queue
          .push(makeJob())
          .then(
            _result => {
              stats.success++;
              console.log(i, "success");
            },
            err => {
              if (err instanceof errors.QueueTimeout) {
                stats.queuetimeout++;
                console.log(i, "queuetimeout");
              } else if (err instanceof errors.QueueFull) {
                stats.queuefull++;
                console.log(i, "queuefull");
              } else if (err instanceof errors.ProcessingCancelled) {
                console.log(i, "jobcancelled");
                stats.cancelled++;
              } else if (err instanceof errors.JobTimeout) {
                stats.timeout++;
                console.log(i, "jobtimeout");
              } else if (err.message === "Failure") {
                stats.fail++;
                console.log(i, "failure");
              } else {
                // Something else failed when running the job function
                stats.other++;
                console.log(i, "somethingelse");
              }
              return true;
            }
          )
          .finally(() => {
            console.log({
              processing: queue.countJobsInProcessing(),
              waiting: queue.countJobsInQueue() - queue.countJobsInProcessing(),
              total: queue.countJobsInQueue()
            });
            return true;
          });

        return promise;
      });

    return Promise.all(promises).finally(() => {
      console.log("success: ", (stats.success * 100) / tasks, "%");
      console.log("fail: ", (stats.fail * 100) / tasks, "%");
      console.log("timeout: ", (stats.timeout * 100) / tasks, "%");
      console.log("other: ", (stats.other * 100) / tasks, "%");
      console.log("queuetimeout: ", (stats.queuetimeout * 100) / tasks, "%");
      console.log("queuefull: ", (stats.queuefull * 100) / tasks, "%");
      console.log("cancelled: ", (stats.cancelled * 100) / tasks, "%");

      assert.ok(
        close(fails + successes + timeouts, 1),
        `fails ${fails} successes ${successes} and timeouts ${timeouts} must add to 1`
      );
      assert.ok(
        close(stats.success / tasks, successes),
        `successes ${stats.success / tasks} must be close to ${successes}`
      );
      assert.ok(
        close(stats.fail / tasks, fails),
        `fails ${stats.fail / tasks} must be close to ${fails}`
      );
      assert.ok(
        close(stats.timeout / tasks, timeouts),
        `timeouts ${stats.timeout / tasks} must be close to ${timeouts}`
      );
      done();
    });
  });
});

function close(n1, n2) {
  return Math.abs(n1 - n2) < 0.05;
}