<commit_before>import {describe} from 'ava-spec';

import Queue from '../lib/queue';
import helpers from '../lib/helpers';
import sinon from 'sinon';

import {promisify} from 'promise-callbacks';

import redis from '../lib/redis';
import {createClient} from 'redis';

// A promise-based barrier.
function reef(n = 1) {
  const done = helpers.deferred(), end = done.defer();
  return {
    done,
    next() {
      --n;
      if (n < 0) return false;
      if (n === 0) end();
      return true;
    }
  };
}

async function recordUntil(emitter, trackedEvents, lastEvent) {
  const recordedEvents = [];

  const done = helpers.waitOn(emitter, lastEvent);
  for (let event of trackedEvents) {
    const handler = (...values) => {
      recordedEvents.push([event, ...values]);
    };
    emitter.on(event, handler);
    done.then(() => emitter.removeListener(event, handler));
  }

  await done;
  return recordedEvents;
}

function delKeys(client, pattern) {
  const promise = helpers.deferred(), done = promise.defer();
  client.keys(pattern, (err, keys) => {
    if (err) return done(err);
    if (keys.length) {
      client.del(keys, done);
    } else {
      done();
    }
  });
  return promise;
}

function spitter() {
  const values = [], resume = [];

  function push(value) {
    if (resume.length) {
      resume.shift()(value);
    } else {
      values.push(value);
    }
  }

  return {
    push,
    pushSuspend(value) {
      return new Promise((resolve) => push([value, resolve]));
    },
    count() {
      return values.length;
    },
    shift() {
      if (values.length) {
        return Promise.resolve(values.shift());
      }
      return new Promise((resolve) => resume.push(resolve));
    }
  };
}

describe('Queue', (it) => {
  const gclient = redis.createClient();

  it.before(() => gclient);

  let uid = 0;
  it.beforeEach((t) => {
    const ctx = t.context;

    Object.assign(ctx, {
      queueName: `test-queue-${uid++}`,
      queues: [],
      queueErrors: [],
      makeQueue,
      handleErrors,
    });

    function makeQueue(...args) {
      const queue = new Queue(ctx.queueName, ...args);
      queue.on('error', (err) => ctx.queueErrors.push(err));
      ctx.queues.push(queue);
      return queue;
    }

    function handleErrors(t) {
      if (t) return t.notThrows(handleErrors);
      if (ctx.queueErrors && ctx.queueErrors.length) {
        throw ctx.queueErrors[0];
      }
    }
  });

  it.afterEach((t) => {
    const errs = t.context.queueErrors;
    if (errs && errs.length) {
      t.fail('errors were not cleaned up');
    }

    // Close all the queues that were created during the test, and wait for them to close before
    // ending the test.
    if (t.context.queues) {
      return Promise.all(t.context.queues.map((queue) => {
        if (!queue.paused) {
          return queue.close();
        }
      }));
    }
  });

  it.beforeEach(async (t) => delKeys(await gclient, `bq:${t.context.queueName}:*`));
  it.afterEach(async (t) => delKeys(await gclient, `bq:${t.context.queueName}:*`));

  it('should initialize without ensuring scripts', async (t) => {
    const queue = t.context.makeQueue({
      ensureScripts: false
    });

    await queue.ready();

    t.context.handleErrors(t);
  });

  it.cb('should support a ready callback', (t) => {
    const queue = t.context.makeQueue();
    queue.ready(t.end);
  });

  it('should indicate whether it is running', async (t) => {
    const queue = t.context.makeQueue();

    // The queue should be "running" immediately - different from ready because it can accept jobs
    // immediately.
    t.true(queue.isRunning());
    await queue.ready();
    t.true(queue.isRunning());
    await queue.close();
    t.false(queue.isRunning());
  });

  it.describe('Connection', (it) => {
    it.describe('Close', (it) => {
      it('should close the redis clients', async (t) => {
        const queue = t.context.makeQueue();

        await queue.ready();

        t.true(redis.isReady(queue.client));
        t.true(redis.isReady(queue.bclient));
        t.true(redis.isReady(queue.eclient));

        await queue.close();

        // ioredis closes the connection some time after the quit response has
        // been received.
        await Promise.all([
          redis.isReady(queue.client) && helpers.waitOn(queue.client, 'close'),
          redis.isReady(queue.bclient) && helpers.waitOn(queue.bclient, 'close'),
          redis.isReady(queue.eclient) && helpers.waitOn(queue.eclient, 'close'),
        ]);

        t.false(redis.isReady(queue.client));
        t.false(redis.isReady(queue.eclient));
      });

      it.cb('should support callbacks', (t) => {
        const queue = t.context.makeQueue();

        queue.ready().then(() => {
          queue.close(t.end);
        }).catch(t.end);
      });

      it('should not fail after a second call', async (t) => {
        const queue = t.context.makeQueue();

        await queue.ready();

        await queue.close();
        await t.notThrows(queue.close());
      });

      it('should stop processing even with a redis retry strategy', async (t) => {
        const queue = t.context.makeQueue({
          redis: {
            // Retry after 1 millisecond.
            retryStrategy: () => 1
          }
        });

        const processSpy = sinon.spy(async () => {});
        queue.process(processSpy);

        await queue.createJob({is: 'first'}).save();
        await helpers.waitOn(queue, 'succeeded', true);
        t.true(processSpy.calledOnce);
        processSpy.reset();

        // Close the queue so queue3 will process later jobs.
        queue.close();

        const queue2 = t.context.makeQueue({
          // If the other queue is still somehow running, ensure that we can't take work from it.
          isWorker: false
        });

        await queue2.createJob({is: 'second'}).save();
        await helpers.delay(50);

        const queue3 = t.context.makeQueue();

        const newSpy = sinon.spy(async () => {});
        queue3.process(newSpy);

        await helpers.waitOn(queue3, 'succeeded', true);
        await queue3.close();

        t.true(newSpy.calledOnce);
        t.false(processSpy.called);
      });

      it('should gracefully shut down', async (t) => {
        const queue = t.context.makeQueue();

        const started = helpers.deferred(), resume = helpers.deferred();
        queue.process(() => {
          setImmediate(started.defer(), null);
          return resume;
        });

        const successSpy = sinon.spy();
        queue.on('succeeded', successSpy);

        await queue.createJob({}).save();
        await started;

        // Asynchronously wait 20 seconds, then complete the process handler.
        setTimeout(resume.defer(), 20, null);

        // Meanwhile, close the queue, and verify that success is called before close completes.
        t.false(successSpy.called);
        await queue.close();
        t.true(successSpy.calledOnce);
      });

      it('should not process new jobs while shutting down', async (t) => {
        const queue = t.context.makeQueue();

        const started = helpers.deferred(),
              resumed = helpers.deferred(), resume = resumed.defer();
        const processSpy = sinon.spy(() => {
          setImmediate(started.defer(), null);
          return resumed;
        });
        queue.process(processSpy);

        const successSpy = sinon.spy();
        queue.on('succeeded', successSpy);

        await queue.createJob({is: 'first'}).save();
        await started;

        // Close the queue, save a new job, and then complete the first job.
        const closed = queue.close();
        await queue.createJob({is: 'second'}).save();
        resume(null);
        await closed;

        // Verify that the second job wasn't picked up for processing.
        t.true(processSpy.calledOnce);
        t.true(successSpy.calledOnce);
        t.deepEqual(processSpy.firstCall.args[0].data, {is: 'first'});
      });

      it('should stop the check timer', async (t) => {
        const queue = t.context.makeQueue({
          stallInterval: 100
        });

        queue.checkStalledJobs(50);

        await helpers.delay(25);

        const spy = sinon.spy(queue, 'checkStalledJobs');
        await queue.close();

        await helpers.delay(50);

        t.false(spy.called);
      });

      it('should time out', async (t) => {
        const queue = t.context.makeQueue();

        // Intentionally stall the job.
        const jobs = spitter();
        queue.process((job) => jobs.pushSuspend(job));

        await queue.createJob({}).save();
        await jobs.shift();

        await t.throws(queue.close(10), 'Operation timed out.');
      });

      it('should not time out when a job fails', async (t) => {
        const queue = t.context.makeQueue();

        const jobs = spitter();
        queue.process((job) => jobs.pushSuspend(job));

        await queue.createJob({}).save();
        const [, finishJob] = await jobs.shift();

        process.nextTick(finishJob, new Error('fails the job'));
        await t.notThrows(queue.close(1000));
      });

      it('should error if a job completes after the timeout', async (t) => {
        const queue = t.context.makeQueue();

        const jobs = spitter();
        queue.process((job) => jobs.pushSuspend(job));

        await queue.createJob({}).save();
        const [, finishJob] = await jobs.shift();

        await t.throws(queue.close(10));
        finishJob(null);

        await helpers.delay(5);

        const errors = t.context.queueErrors, count = errors.length;
        t.context.queueErrors = errors.filter((err) => {
          return err.message !== 'unable to update the status of succeeded job 1';
        });
        t.is(t.context.queueErrors.length, count - 1);
        t.context.handleErrors(t);
      });

      it('should not error on close', async (t) => {
        const queue = t.context.makeQueue();

        await queue.close();

        await helpers.delay(30);

        t.context.handleErrors(t);
      });

      it('should not interfere with checkStalledJobs', async (t) => {
        const queue = t.context.makeQueue();

        await queue.checkStalledJobs(10);
        await queue.close();

        await helpers.delay(50);

        t.context.handleErrors(t);
      });

      it('should not quit the command client by default if given in settings', async (t) => {
        const client = await redis.createClient();

        sinon.spy(client, 'quit');

        let queue = t.context.makeQueue({
          redis: client
        });

        await queue.close();

        t.true(client.ready);
        t.false(client.quit.called);

        queue = t.context.makeQueue({
          redis: client,
          quitCommandClient: true
        });

        await queue.close();

        t.false(client.ready);
        t.true(client.quit.called);
      });
    });

    it('should recover from a connection loss', async (t) => {
      const queue = t.context.makeQueue({
        redis: {
          // Retry after 1 millisecond.
          retryStrategy: () => 1
        }
      });

      const jobSpy = sinon.spy(queue, '_getNextJob');

      queue.process(async (job) => {
        t.is(job.data.foo, 'bar');

        // First getNextJob fails on the disconnect, second should succeed.
        t.is(jobSpy.callCount, 2);
      });<commit_msg>Expand tests for shared client support<commit_after>import {describe} from 'ava-spec';

import Queue from '../lib/queue';
import helpers from '../lib/helpers';
import sinon from 'sinon';

import {promisify} from 'promise-callbacks';

import redis from '../lib/redis';
import {createClient} from 'redis';

// A promise-based barrier.
function reef(n = 1) {
  const done = helpers.deferred(), end = done.defer();
  return {
    done,
    next() {
      --n;
      if (n < 0) return false;
      if (n === 0) end();
      return true;
    }
  };
}

async function recordUntil(emitter, trackedEvents, lastEvent) {
  const recordedEvents = [];

  const done = helpers.waitOn(emitter, lastEvent);
  for (let event of trackedEvents) {
    const handler = (...values) => {
      recordedEvents.push([event, ...values]);
    };
    emitter.on(event, handler);
    done.then(() => emitter.removeListener(event, handler));
  }

  await done;
  return recordedEvents;
}

function delKeys(client, pattern) {
  const promise = helpers.deferred(), done = promise.defer();
  client.keys(pattern, (err, keys) => {
    if (err) return done(err);
    if (keys.length) {
      client.del(keys, done);
    } else {
      done();
    }
  });
  return promise;
}

function spitter() {
  const values = [], resume = [];

  function push(value) {
    if (resume.length) {
      resume.shift()(value);
    } else {
      values.push(value);
    }
  }

  return {
    push,
    pushSuspend(value) {
      return new Promise((resolve) => push([value, resolve]));
    },
    count() {
      return values.length;
    },
    shift() {
      if (values.length) {
        return Promise.resolve(values.shift());
      }
      return new Promise((resolve) => resume.push(resolve));
    }
  };
}

describe('Queue', (it) => {
  const gclient = redis.createClient();

  it.before(() => gclient);

  let uid = 0;
  it.beforeEach((t) => {
    const ctx = t.context;

    Object.assign(ctx, {
      queueName: `test-queue-${uid++}`,
      queues: [],
      queueErrors: [],
      makeQueue,
      handleErrors,
    });

    function makeQueue(...args) {
      const queue = new Queue(ctx.queueName, ...args);
      queue.on('error', (err) => ctx.queueErrors.push(err));
      ctx.queues.push(queue);
      return queue;
    }

    function handleErrors(t) {
      if (t) return t.notThrows(handleErrors);
      if (ctx.queueErrors && ctx.queueErrors.length) {
        throw ctx.queueErrors[0];
      }
    }
  });

  it.afterEach((t) => {
    const errs = t.context.queueErrors;
    if (errs && errs.length) {
      t.fail('errors were not cleaned up');
    }

    // Close all the queues that were created during the test, and wait for them to close before
    // ending the test.
    if (t.context.queues) {
      return Promise.all(t.context.queues.map((queue) => {
        if (!queue.paused) {
          return queue.close();
        }
      }));
    }
  });

  it.beforeEach(async (t) => delKeys(await gclient, `bq:${t.context.queueName}:*`));
  it.afterEach(async (t) => delKeys(await gclient, `bq:${t.context.queueName}:*`));

  it('should initialize without ensuring scripts', async (t) => {
    const queue = t.context.makeQueue({
      ensureScripts: false
    });

    await queue.ready();

    t.context.handleErrors(t);
  });

  it.cb('should support a ready callback', (t) => {
    const queue = t.context.makeQueue();
    queue.ready(t.end);
  });

  it('should indicate whether it is running', async (t) => {
    const queue = t.context.makeQueue();

    // The queue should be "running" immediately - different from ready because it can accept jobs
    // immediately.
    t.true(queue.isRunning());
    await queue.ready();
    t.true(queue.isRunning());
    await queue.close();
    t.false(queue.isRunning());
  });

  it.describe('Connection', (it) => {
    it.describe('Close', (it) => {
      it('should close the redis clients', async (t) => {
        const queue = t.context.makeQueue();

        await queue.ready();

        t.true(redis.isReady(queue.client));
        t.true(redis.isReady(queue.bclient));
        t.true(redis.isReady(queue.eclient));

        await queue.close();

        // ioredis closes the connection some time after the quit response has
        // been received.
        await Promise.all([
          redis.isReady(queue.client) && helpers.waitOn(queue.client, 'close'),
          redis.isReady(queue.bclient) && helpers.waitOn(queue.bclient, 'close'),
          redis.isReady(queue.eclient) && helpers.waitOn(queue.eclient, 'close'),
        ]);

        t.false(redis.isReady(queue.client));
        t.false(redis.isReady(queue.eclient));
      });

      it.cb('should support callbacks', (t) => {
        const queue = t.context.makeQueue();

        queue.ready().then(() => {
          queue.close(t.end);
        }).catch(t.end);
      });

      it('should not fail after a second call', async (t) => {
        const queue = t.context.makeQueue();

        await queue.ready();

        await queue.close();
        await t.notThrows(queue.close());
      });

      it('should stop processing even with a redis retry strategy', async (t) => {
        const queue = t.context.makeQueue({
          redis: {
            // Retry after 1 millisecond.
            retryStrategy: () => 1
          }
        });

        const processSpy = sinon.spy(async () => {});
        queue.process(processSpy);

        await queue.createJob({is: 'first'}).save();
        await helpers.waitOn(queue, 'succeeded', true);
        t.true(processSpy.calledOnce);
        processSpy.reset();

        // Close the queue so queue3 will process later jobs.
        queue.close();

        const queue2 = t.context.makeQueue({
          // If the other queue is still somehow running, ensure that we can't take work from it.
          isWorker: false
        });

        await queue2.createJob({is: 'second'}).save();
        await helpers.delay(50);

        const queue3 = t.context.makeQueue();

        const newSpy = sinon.spy(async () => {});
        queue3.process(newSpy);

        await helpers.waitOn(queue3, 'succeeded', true);
        await queue3.close();

        t.true(newSpy.calledOnce);
        t.false(processSpy.called);
      });

      it('should gracefully shut down', async (t) => {
        const queue = t.context.makeQueue();

        const started = helpers.deferred(), resume = helpers.deferred();
        queue.process(() => {
          setImmediate(started.defer(), null);
          return resume;
        });

        const successSpy = sinon.spy();
        queue.on('succeeded', successSpy);

        await queue.createJob({}).save();
        await started;

        // Asynchronously wait 20 seconds, then complete the process handler.
        setTimeout(resume.defer(), 20, null);

        // Meanwhile, close the queue, and verify that success is called before close completes.
        t.false(successSpy.called);
        await queue.close();
        t.true(successSpy.calledOnce);
      });

      it('should not process new jobs while shutting down', async (t) => {
        const queue = t.context.makeQueue();

        const started = helpers.deferred(),
              resumed = helpers.deferred(), resume = resumed.defer();
        const processSpy = sinon.spy(() => {
          setImmediate(started.defer(), null);
          return resumed;
        });
        queue.process(processSpy);

        const successSpy = sinon.spy();
        queue.on('succeeded', successSpy);

        await queue.createJob({is: 'first'}).save();
        await started;

        // Close the queue, save a new job, and then complete the first job.
        const closed = queue.close();
        await queue.createJob({is: 'second'}).save();
        resume(null);
        await closed;

        // Verify that the second job wasn't picked up for processing.
        t.true(processSpy.calledOnce);
        t.true(successSpy.calledOnce);
        t.deepEqual(processSpy.firstCall.args[0].data, {is: 'first'});
      });

      it('should stop the check timer', async (t) => {
        const queue = t.context.makeQueue({
          stallInterval: 100
        });

        queue.checkStalledJobs(50);

        await helpers.delay(25);

        const spy = sinon.spy(queue, 'checkStalledJobs');
        await queue.close();

        await helpers.delay(50);

        t.false(spy.called);
      });

      it('should time out', async (t) => {
        const queue = t.context.makeQueue();

        // Intentionally stall the job.
        const jobs = spitter();
        queue.process((job) => jobs.pushSuspend(job));

        await queue.createJob({}).save();
        await jobs.shift();

        await t.throws(queue.close(10), 'Operation timed out.');
      });

      it('should not time out when a job fails', async (t) => {
        const queue = t.context.makeQueue();

        const jobs = spitter();
        queue.process((job) => jobs.pushSuspend(job));

        await queue.createJob({}).save();
        const [, finishJob] = await jobs.shift();

        process.nextTick(finishJob, new Error('fails the job'));
        await t.notThrows(queue.close(1000));
      });

      it('should error if a job completes after the timeout', async (t) => {
        const queue = t.context.makeQueue();

        const jobs = spitter();
        queue.process((job) => jobs.pushSuspend(job));

        await queue.createJob({}).save();
        const [, finishJob] = await jobs.shift();

        await t.throws(queue.close(10));
        finishJob(null);

        await helpers.delay(5);

        const errors = t.context.queueErrors, count = errors.length;
        t.context.queueErrors = errors.filter((err) => {
          return err.message !== 'unable to update the status of succeeded job 1';
        });
        t.is(t.context.queueErrors.length, count - 1);
        t.context.handleErrors(t);
      });

      it('should not error on close', async (t) => {
        const queue = t.context.makeQueue();

        await queue.close();

        await helpers.delay(30);

        t.context.handleErrors(t);
      });

      it('should not interfere with checkStalledJobs', async (t) => {
        const queue = t.context.makeQueue();

        await queue.checkStalledJobs(10);
        await queue.close();

        await helpers.delay(50);

        t.context.handleErrors(t);
      });

      it('should not quit the command client by default if given in settings', async (t) => {
        const client = await redis.createClient();

        sinon.spy(client, 'quit');

        let queue = t.context.makeQueue({
          redis: client
        });

        await queue.close();

        t.true(client.ready);
        t.false(client.quit.called);

        let promise = helpers.deferred();
        client.ping(promise.defer());
        await t.notThrows(promise);

        queue = t.context.makeQueue({
          redis: client,
          quitCommandClient: true
        });

        await queue.close();

        t.false(client.ready);
        t.true(client.quit.called);

        promise = helpers.deferred();
        client.ping(promise.defer());
        await t.throws(promise, (err) => redis.isAbortError(err));
      });

      it('should not quit the command client when quitCommandClient=false', async (t) => {
        const queue = t.context.makeQueue({
          quitCommandClient: false
        });

        await queue.ready();

        const client = queue.client;
        sinon.spy(client, 'quit');

        await queue.close();

        t.true(client.ready);
        t.false(client.quit.called);

        let promise = helpers.deferred();
        client.ping(promise.defer());
        await t.notThrows(promise);

        promise = helpers.deferred();
        client.quit(promise.defer());
        await promise;
      });
    });

    it('should recover from a connection loss', async (t) => {
      const queue = t.context.makeQueue({
        redis: {
          // Retry after 1 millisecond.
          retryStrategy: () => 1
        }
      });

      const jobSpy = sinon.spy(queue, '_getNextJob');

      queue.process(async (job) => {
        t.is(job.data.foo, 'bar');

        // First getNextJob fails on the disconnect, second should succeed.
        t.is(jobSpy.callCount, 2);
      });
{'fraction_non_alphanumeric': 0.1495090809489697, 'fraction_numerical': 0.004213897433736463, 'mean_word_length': 2.241633656604289, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5678894', 'n_tokens_mistral': 7310, 'n_tokens_neox': 7102, 'n_words': 1705}