intervalCap not respected when using large intervalCaps or small intervals

[thejamespower]

We have been trying to debug an issue with our PQueue in our application for a few weeks now. We keep getting throttled by our API provider for going over their 2000/minute request limit. Even with setting our intervalCap to 1900/minute, we were still breaching the limit regularly.

I modified one of the current tests to prove that somethings starts to go wrong when you increase the intervalCap to numbers above 1. The current tests files use a cap of one which always passes. When you increase the intervalCap to above ~100, the test starts failing randomly. Sometimes it passes, sometimes it fails. The higher the intervalCap, the more likely it is to fail.

test('.add() - throttled, carryoverConcurrencyCount false large', async t => {
	const result: number[] = [];

	const testIntervals = 5;
        const intervalCap = 10000;
	const interval = 1000;

	const queue = new PQueue({
		intervalCap,
		carryoverConcurrencyCount: false,
		interval,
		autoStart: false
	});

	const values = [...new Array(testIntervals * intervalCap).keys()];

	values.forEach(async value => queue.add(async () => {
		await delay(500);
		result.push(value);
	}));

	queue.start();

	t.is(queue.pending, intervalCap);
	t.is(queue.size, (testIntervals - 1) * intervalCap);
	t.deepEqual(result, []);

	(async () => {
		await delay(1.1 * interval);
		t.is(queue.pending, intervalCap);
		t.is(queue.size, (testIntervals - 2) * Number(intervalCap));
		t.deepEqual(result, [...new Array(1 * Number(intervalCap)).keys()]);
	})();

	(async () => {
		await delay(2.1 * interval);
		t.is(queue.pending, intervalCap);
		t.is(queue.size, (testIntervals - 3) * intervalCap);
		t.deepEqual(result, [...new Array(2 * Number(intervalCap)).keys()]);
	})();

	(async () => {
		await delay(3.1 * interval);
		t.is(queue.pending, intervalCap);
		t.is(queue.size, (testIntervals - 4) * intervalCap);
		t.deepEqual(result, [...new Array(3 * Number(intervalCap)).keys()]);
	})();

	(async () => {
		await delay(4.1 * interval);
		t.is(queue.pending, intervalCap);
		t.is(queue.size, (testIntervals - 5) * intervalCap);
		t.deepEqual(result, [...new Array(4 * Number(intervalCap)).keys()]);
	})();

	await delay(5.1 * interval);
	t.deepEqual(result, values);
});

It's worth noting the same behaviour is demonstrated if you use small intervals (<100).

[sindresorhus]

// @Rafael09ED

[Rafael09ED]

Interesting. Just adding that 1 test causes failures in multiple other tests.
Before:

  39 tests passed
  2 known failures

After:

  5 tests failed
  2 known failures

in tests

  .add() - throttled
  .add() - concurrency: 1
  .add() - throttled, carryoverConcurrencyCount false
  .add() - throttled, carryoverConcurrencyCount true
  .add() - throttled, carryoverConcurrencyCount false large

Since the tests, to my understanding, run in parallel perhaps they are interfering with each other. I don't know how this is possible but I have some far fetched guesses if someone has any ideas. I will have to look into this more after work.

[Rafael09ED]

@thejamespower, I think intervalCap is still being respected. I think that what is happening is at a larger number of tasks, the awaiting no longer sufficient to meet the time window needed to do an accurate check.

I shortened the await delay() to get a time closer to what you would want for the test but it is not a reliable way of getting inside the time window needed. By shortening it and disabling t.is(queue.pending, intervalCap) I was able to get it past the first x.1 * interval block. I did not try to get past the subsequent ones.

From the log I don't see any unexpected behavior based on the timestamps. Every 1000ms 10000 jobs are started. I could be missing something but right now this seems to me like we need to change how we handle tests with larger numbers of jobs so we can run tests at this size.

Log w/ timestamps

[T+00000] QUEUE START Time:1613016599237
[T+00001] job 0 starting
[T+00078] job 1000 starting
[T+00143] job 2000 starting
[T+00203] job 3000 starting
[T+00272] job 4000 starting
[T+00327] job 5000 starting
[T+00379] job 6000 starting
[T+00430] job 7000 starting
[T+00480] job 8000 starting
[T+00545] job 9000 starting
[T+00616] result size:0
[T+00616] pending:10000
[T+00616] queue size:40000
[T+00616] job 0 finishing
[T+00628] job 1000 finishing
[T+00643] job 2000 finishing
[T+00704] job 3000 finishing
[T+00717] result size:3214
[T+00717] pending:6786
[T+00717] queue size:40000
[T+00772] job 4000 finishing
[T+00817] result size:4802
[T+00817] pending:5198
[T+00817] queue size:40000
[T+00827] job 5000 finishing
[T+00879] job 6000 finishing
[T+00917] result size:6726
[T+00917] pending:3274
[T+00917] queue size:40000
[T+00930] job 7000 finishing
[T+00980] job 8000 finishing
[T+01002] job 10000 starting
[T+01057] job 11000 starting
[T+01110] job 12000 starting
[T+01156] job 13000 starting
[T+01206] job 14000 starting
[T+01247] job 15000 starting
[T+01288] job 16000 starting
[T+01329] job 17000 starting
[T+01367] job 18000 starting
[T+01403] job 19000 starting
[T+01457] job 9000 finishing
[T+01459] await delay(1.1 * interval) check
[T+01459] pending:10000
[T+01459] result size:10000
[T+01487] result size:10000
[T+01487] pending:10000
[T+01487] queue size:30000
[T+01502] job 10000 finishing
[T+01558] job 11000 finishing
[T+01587] result size:11563
[T+01587] pending:8437
[T+01587] queue size:30000
[T+01610] job 12000 finishing
[T+01656] job 13000 finishing
[T+01687] result size:13634
[T+01687] pending:6366
[T+01687] queue size:30000
[T+01706] job 14000 finishing
[T+01748] job 15000 finishing
[T+01787] result size:15950
[T+01787] pending:4050
[T+01787] queue size:30000
[T+01789] job 16000 finishing
[T+01829] job 17000 finishing
[T+01867] job 18000 finishing
[T+01887] result size:18528
[T+01887] pending:1472
[T+01887] queue size:30000
[T+01904] job 19000 finishing
[T+01987] result size:20000
[T+01987] pending:0
[T+01987] queue size:30000
[T+02002] job 20000 starting
[T+02048] job 21000 starting
[T+02086] job 22000 starting
[T+02124] job 23000 starting
[T+02157] job 24000 starting
[T+02187] job 25000 starting
[T+02219] job 26000 starting
[T+02254] job 27000 starting
[T+02290] job 28000 starting
[T+02315] job 29000 starting
[T+02368] result size:20000
[T+02368] pending:10000
[T+02368] queue size:20000
[T+02469] result size:20000
[T+02469] pending:10000
[T+02469] queue size:20000
[T+02502] job 20000 finishing
[T+02548] job 21000 finishing
[T+02568] result size:21473
[T+02568] pending:8527
[T+02568] queue size:20000
[T+02587] job 22000 finishing
[T+02624] job 23000 finishing
[T+02657] job 24000 finishing
[T+02668] result size:24417
[T+02668] pending:5583
[T+02668] queue size:20000
[T+02687] job 25000 finishing
[T+02838] job 26000 finishing
[T+02839] job 27000 finishing
[T+02840] job 28000 finishing
[T+02842] job 29000 finishing
[T+02844] result size:29476
[T+02844] pending:524
[T+02844] queue size:20000
[T+02944] result size:30000
[T+02944] pending:0
[T+02944] queue size:20000
[T+03003] job 30000 starting
[T+03027] job 31000 starting
[T+03050] job 32000 starting
[T+03082] job 33000 starting
[T+03101] job 34000 starting
[T+03120] job 35000 starting
[T+03137] job 36000 starting
[T+03153] job 37000 starting
[T+03181] job 38000 starting
[T+03216] job 39000 starting
[T+03230] result size:30000
[T+03230] pending:10000
[T+03230] queue size:10000
[T+03330] result size:30000
[T+03330] pending:10000
[T+03330] queue size:10000
[T+03430] result size:30000
[T+03430] pending:10000
[T+03430] queue size:10000
[T+03503] job 30000 finishing
[T+03528] job 31000 finishing
[T+03530] result size:31093
[T+03530] pending:8907
[T+03530] queue size:10000
[T+03550] job 32000 finishing
[T+03582] job 33000 finishing
[T+03601] job 34000 finishing
[T+03620] job 35000 finishing
[T+03629] result size:35554
[T+03629] pending:4446
[T+03629] queue size:10000
[T+03636] job 36000 finishing
[T+03654] job 37000 finishing
[T+03682] job 38000 finishing
[T+03852] job 39000 finishing
[T+03857] result size:40000
[T+03857] pending:0
[T+03857] queue size:10000
[T+03957] result size:40000
[T+03957] pending:0
[T+03957] queue size:10000
[T+04003] job 40000 starting
[T+04017] job 41000 starting
[T+04027] job 42000 starting
[T+04038] job 43000 starting
[T+04047] job 44000 starting
[T+04054] job 45000 starting
[T+04060] job 46000 starting
[T+04064] job 47000 starting
[T+04068] job 48000 starting
[T+04071] job 49000 starting
[T+04109] result size:40000
[T+04109] pending:10000
[T+04109] queue size:0
[T+04209] result size:40000
[T+04209] pending:10000
[T+04209] queue size:0
[T+04308] result size:40000
[T+04309] pending:10000
[T+04309] queue size:0
[T+04409] result size:40000
[T+04409] pending:10000
[T+04409] queue size:0
[T+04503] job 40000 finishing
[T+04509] result size:40341
[T+04509] pending:9659
[T+04509] queue size:0
[T+04517] job 41000 finishing
[T+04532] job 42000 finishing
[T+04543] job 43000 finishing
[T+04547] job 44000 finishing
[T+04554] job 45000 finishing
[T+04561] job 46000 finishing
[T+04566] job 47000 finishing
[T+04569] job 48000 finishing
[T+04571] job 49000 finishing
[T+04610] result size:50000
[T+04610] pending:0
[T+04610] queue size:0
[T+04866] result size:50000
[T+04866] pending:0
[T+04866] queue size:0
[T+04967] result size:50000
[T+04967] pending:0
[T+04967] queue size:0
[T+05067] result size:50000
[T+05067] pending:0
[T+05067] queue size:0
[T+05168] result size:50000
[T+05168] pending:0
[T+05168] queue size:0

[thejamespower]

Thanks for investigating @Rafael09ED. Yes, I noticed adding this test result in other tests breaking, unfortunately, I have no experience in your chosen test runner to be able to investigate any further. I would love to be able to spend some time on this, unfortunately, we do not have the spare capacity right now, so will have to find another library/solution.

[sandinmyjoints]

@thejamespower Did you find a workable solution? I have run into this as well.

With rate limits 50 per second, max 10 concurrency requests, this configuration breaches the limits very quickly. It even happens if I add some padding.

  const queue = new PQueue({
    concurrency: 10,
    intervalCap: 50, 
    interval: 1000, 
    carryoverConcurrencyCount: true
  });

[Aervue]

Experiencing the same issue