taskforcesh / bullmq / 436

pending completion

Build # 436

Build Type

Pull #153

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/taskforcesh/bullmq/commit/<a class=hub.com/taskforcesh/bullmq/commit/<a class="double-link" href="https://git"><a class=hub.com/taskforcesh/bullmq/commit/<a class="double-link" href="https://git"><a class=hub.com/taskforcesh/bullmq/commit/<a class="double-link" href="https://git"><a class=hub.com/taskforcesh/bullmq/commit/<a class="double-link" href="https://git"><a class=hub.com/taskforcesh/bullmq/commit/e82c106c730e662a47e4418c6c006de9e687f0a6">e82c106c7">&lt;a href=&quot;https://github.com/taskforcesh/bullmq/commit/</a><a class="double-link" href="https://github.com/taskforcesh/bullmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/taskforcesh/bullmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/taskforcesh/bullmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/taskforcesh/bullmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/taskforcesh/bullmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/taskforcesh/bullmq/commit/e82c106c730e662a47e4418c6c006de9e687f0a6">e82c106c7</a><a href="https://github.com/taskforcesh/bullmq/commit/e82c106c730e662a47e4418c6c006de9e687f0a6">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/taskforcesh/bullmq/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/taskforcesh/bullmq/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/taskforcesh/bullmq/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/taskforcesh/bullmq/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/taskforcesh/bullmq/commit/e82c106c730e662a47e4418c6c006de9e687f0a6&quot;&gt;e82c106c7&lt;/a&gt;&lt;a href=&quot;https://github.com/taskforcesh/bullmq/commit/e82c106c730e662a47e4418c6c006de9e687f0a6&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/taskforcesh/bullmq/commit/e82c106c730e662a47e4418c6c006de9e687f0a6&amp;quot;&amp;gt;&amp;amp;quot;&amp;amp;gt;&amp;amp;amp;lt;a href=&amp;amp;amp;quot;https://github.com/taskforcesh/bullmq/commit/&amp;amp;lt;/... (continued)

Pull Request Pull Request #153: fix(worker): return this.closing when calling close

Run Details

570 of 791 branches covered (72.06%)

Branch coverage included in aggregate %.

14 of 14 new or added lines in 4 files covered. (100.0%)

3773 of 4151 relevant lines covered (90.89%)

70.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.91

/src/classes/queue-scheduler.ts

import { QueueSchedulerOptions } from '../interfaces';
import { array2obj, isRedisInstance } from '../utils';
import { QueueBase } from './';
import { Scripts } from './scripts';
import IORedis = require('ioredis');

const MAX_TIMEOUT_MS = Math.pow(2, 31) - 1; // 32 bit signed

/**
 * This class is just used for some automatic bookkeeping of the queue,
 * such as updating the delay set as well as moving stalled jobs back
 * to the waiting list.
 *
 * Jobs are checked for stallness once every "visibility window" seconds.
 * Jobs are then marked as candidates for being stalled, in the next check,
 * the candidates are marked as stalled and moved to wait.
 * Workers need to clean the candidate list with the jobs that they are working
 * on, failing to update the list results in the job ending being stalled.
 *
 * This class requires a dedicated redis connection, and at least one is needed
 * to be running at a given time, otherwise delays, stalled jobs, retries, repeatable
 * jobs, etc, will not work correctly or at all.
 *
 */
export class QueueScheduler extends QueueBase {
  private nextTimestamp = Number.MAX_VALUE;
  private isBlocked = false;

  constructor(
    name: string,
    { connection, ...opts }: QueueSchedulerOptions = {},
  ) {
    super(name, {
      maxStalledCount: 1,
      stalledInterval: 30000,
      ...opts,
      connection: isRedisInstance(connection)
        ? (<IORedis.Redis>connection).duplicate()
        : connection,
    });

    // tslint:disable: no-floating-promises
    this.run();
  }

  private async run() {
    const client = await this.waitUntilReady();

    const key = this.keys.delay;
    const opts = this.opts as QueueSchedulerOptions;
    const delaySet = await this.updateDelaySet(Date.now());

    const [nextTimestamp] = delaySet;
    let streamLastId = delaySet[1] || '0-0';

    if (nextTimestamp) {
      this.nextTimestamp = nextTimestamp;
    }

    while (!this.closing) {
      // Check if at least the min stalled check time has passed.
      await this.moveStalledJobsToWait();

      // Listen to the delay event stream from lastDelayStreamTimestamp
      // Can we use XGROUPS to reduce redundancy?
      const nextDelay = this.nextTimestamp - Date.now();
      const blockTime = Math.round(
        Math.min(opts.stalledInterval, Math.max(nextDelay, 0)),
      );

      const data = await this.readDelayedData(
        client,
        key,
        streamLastId,
        blockTime,
      );

      if (data && data[0]) {
        const stream = data[0];
        const events = stream[1];

        for (let i = 0; i < events.length; i++) {
          streamLastId = events[i][0];
          const args = array2obj(events[i][1]);
          const nextTimestamp: number = parseInt(args.nextTimestamp);

          if (nextTimestamp < this.nextTimestamp) {
            this.nextTimestamp = nextTimestamp;
          }
        }

        //
        // We trim to a length of 100, which should be a very safe value
        // for all kind of scenarios.
        //
        if (!this.closing) {
          await client.xtrim(key, 'MAXLEN', '~', 100);
        }
      }

      const now = Date.now();
      const delay = this.nextTimestamp - now;

      if (delay <= 0) {
        const [nextTimestamp, id] = await this.updateDelaySet(now);
        if (nextTimestamp) {
          this.nextTimestamp = nextTimestamp / 4096;
          streamLastId = id;
        } else {
          this.nextTimestamp = Number.MAX_VALUE;
        }
      }
    }
  }

  private async readDelayedData(
    client: IORedis.Redis,
    key: string,
    streamLastId: string,
    blockTime: number,
  ) {
    if (!this.closing) {
      let data;
      if (blockTime) {
        try {
          this.isBlocked = true;
          data = await client.xread(
            'BLOCK',
            blockTime,
            'STREAMS',
            key,
            streamLastId,
          );
        } catch (err) {
          // We can ignore closed connection errors
          if (err.message !== 'Connection is closed.') {
            throw err;
          }
        } finally {
          this.isBlocked = false;
        }
      } else {
        data = await client.xread('STREAMS', key, streamLastId);
      }
      return data;
    }
  }

  private async updateDelaySet(timestamp: number) {
    if (!this.closing) {
      return Scripts.updateDelaySet(this, timestamp);
    }
    return [0, 0];
  }

  private async moveStalledJobsToWait() {
    if (!this.closing) {
      const [failed, stalled] = await Scripts.moveStalledJobsToWait(this);

      failed.forEach((jobId: string) =>
        this.emit(
          'failed',
          jobId,
          new Error('job stalled more than allowable limit'),
          'active',
        ),
      );
      stalled.forEach((jobId: string) => this.emit('stalled', jobId, 'active'));
    }
  }

  async close() {
    if (this.closing) {
      return this.closing;
    }
    if (this.isBlocked) {
      this.closing = this.disconnect();
    } else {
      this.closing = super.close();
    }
    return this.closing;
  }
}

1	import { QueueSchedulerOptions } from '../interfaces';
2	import { array2obj, isRedisInstance } from '../utils';	1✔
3	import { QueueBase } from './';	1✔
4	import { Scripts } from './scripts';	1✔
5	import IORedis = require('ioredis');
6
7	const MAX_TIMEOUT_MS = Math.pow(2, 31) - 1; // 32 bit signed	1✔
8
9	/**
10	* This class is just used for some automatic bookkeeping of the queue,
11	* such as updating the delay set as well as moving stalled jobs back
12	* to the waiting list.
13	*
14	* Jobs are checked for stallness once every "visibility window" seconds.
15	* Jobs are then marked as candidates for being stalled, in the next check,
16	* the candidates are marked as stalled and moved to wait.
17	* Workers need to clean the candidate list with the jobs that they are working
18	* on, failing to update the list results in the job ending being stalled.
19	*
20	* This class requires a dedicated redis connection, and at least one is needed
21	* to be running at a given time, otherwise delays, stalled jobs, retries, repeatable
22	* jobs, etc, will not work correctly or at all.
23	*
24	*/
25	export class QueueScheduler extends QueueBase {	1✔
26	private nextTimestamp = Number.MAX_VALUE;	44✔
27	private isBlocked = false;	44✔
28
29	constructor(
30	name: string,
31	{ connection, ...opts }: QueueSchedulerOptions = {},	44✔
32	) {
33	super(name, {	44✔
34	maxStalledCount: 1,
35	stalledInterval: 30000,
36	...opts,
37	connection: isRedisInstance(connection)
38	? (<IORedis.Redis>connection).duplicate()	44!
39	: connection,
40	});
41
42	// tslint:disable: no-floating-promises
43	this.run();	44✔
44	}
45
46	private async run() {
47	const client = await this.waitUntilReady();	44✔
48
49	const key = this.keys.delay;	44✔
50	const opts = this.opts as QueueSchedulerOptions;	44✔
51	const delaySet = await this.updateDelaySet(Date.now());	44✔
52
53	const [nextTimestamp] = delaySet;	44✔
54	let streamLastId = delaySet[1] \|\| '0-0';	44✔
55
56	if (nextTimestamp) {	44!
57	this.nextTimestamp = nextTimestamp;	×
58	}
59
60	while (!this.closing) {	44✔
61	// Check if at least the min stalled check time has passed.
62	await this.moveStalledJobsToWait();	169✔
63
64	// Listen to the delay event stream from lastDelayStreamTimestamp
65	// Can we use XGROUPS to reduce redundancy?
66	const nextDelay = this.nextTimestamp - Date.now();	168✔
67	const blockTime = Math.round(	168✔
68	Math.min(opts.stalledInterval, Math.max(nextDelay, 0)),
69	);
70
71	const data = await this.readDelayedData(	168✔
72	client,
73	key,
74	streamLastId,
75	blockTime,
76	);
77
78	if (data && data[0]) {	167✔
79	const stream = data[0];	80✔
80	const events = stream[1];	80✔
81
82	for (let i = 0; i < events.length; i++) {	80✔
83	streamLastId = events[i][0];	102✔
84	const args = array2obj(events[i][1]);	102✔
85	const nextTimestamp: number = parseInt(args.nextTimestamp);	102✔
86
87	if (nextTimestamp < this.nextTimestamp) {	102✔
88	this.nextTimestamp = nextTimestamp;	63✔
89	}
90	}
91
92	//
93	// We trim to a length of 100, which should be a very safe value
94	// for all kind of scenarios.
95	//
96	if (!this.closing) {	80!
97	await client.xtrim(key, 'MAXLEN', '~', 100);	80✔
98	}
99	}
100
101	const now = Date.now();	167✔
102	const delay = this.nextTimestamp - now;	167✔
103
104	if (delay <= 0) {	167✔
105	const [nextTimestamp, id] = await this.updateDelaySet(now);	73✔
106	if (nextTimestamp) {	73✔
107	this.nextTimestamp = nextTimestamp / 4096;	16✔
108	streamLastId = id;	16✔
109	} else {
110	this.nextTimestamp = Number.MAX_VALUE;	57✔
111	}
112	}
113	}
114	}
115
116	private async readDelayedData(
117	client: IORedis.Redis,
118	key: string,
119	streamLastId: string,
120	blockTime: number,
121	) {
122	if (!this.closing) {	168✔
123	let data;
124	if (blockTime) {	165✔
125	try {	151✔
126	this.isBlocked = true;	151✔
127	data = await client.xread(	151✔
128	'BLOCK',
129	blockTime,
130	'STREAMS',
131	key,
132	streamLastId,
133	);
134	} catch (err) {
135	// We can ignore closed connection errors
136	if (err.message !== 'Connection is closed.') {	37!
137	throw err;	×
138	}
139	} finally {
140	this.isBlocked = false;	150✔
141	}
142	} else {
143	data = await client.xread('STREAMS', key, streamLastId);	14✔
144	}
145	return data;	164✔
146	}
147	}
148
149	private async updateDelaySet(timestamp: number) {
150	if (!this.closing) {	117✔
151	return Scripts.updateDelaySet(this, timestamp);	111✔
152	}
153	return [0, 0];	6✔
154	}
155
156	private async moveStalledJobsToWait() {
157	if (!this.closing) {	169!
158	const [failed, stalled] = await Scripts.moveStalledJobsToWait(this);	169✔
159
160	failed.forEach((jobId: string) =>	168✔
161	this.emit(	4✔
162	'failed',
163	jobId,
164	new Error('job stalled more than allowable limit'),
165	'active',
166	),
167	);
168	stalled.forEach((jobId: string) => this.emit('stalled', jobId, 'active'));	168✔
169	}
170	}
171
172	async close() {
173	if (this.closing) {	43!
174	return this.closing;	×
175	}
176	if (this.isBlocked) {	43✔
177	this.closing = this.disconnect();	37✔
178	} else {
179	this.closing = super.close();	6✔
180	}
181	return this.closing;	43✔
182	}
183	}

taskforcesh / bullmq / 436

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous