Skip to content

Commit 1be4fcb

Browse files
authored
Fixes the jitter option for the scanConcurrencySets cron job (#2220)
1 parent 124435e commit 1be4fcb

File tree

3 files changed

+11
-5
lines changed

3 files changed

+11
-5
lines changed

internal-packages/run-engine/src/run-queue/index.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -212,7 +212,7 @@ export class RunQueue {
212212
scanConcurrencySets: {
213213
...workerCatalog.scanConcurrencySets,
214214
cron: options.concurrencySweeper?.scanSchedule ?? workerCatalog.scanConcurrencySets.cron,
215-
jitter:
215+
jitterInMs:
216216
options.concurrencySweeper?.scanJitterInMs ??
217217
workerCatalog.scanConcurrencySets.jitterInMs,
218218
},

internal-packages/run-engine/src/run-queue/tests/concurrencySweeper.test.ts

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ const testOptions = {
1313
tracer: trace.getTracer("rq"),
1414
workers: 1,
1515
defaultEnvConcurrency: 25,
16-
logger: new Logger("RunQueue", "warn"),
16+
logger: new Logger("RunQueue", "debug"),
1717
retryOptions: {
1818
maxAttempts: 5,
1919
factor: 1.1,
@@ -59,6 +59,7 @@ describe("RunQueue Concurrency Sweeper", () => {
5959

6060
const queue = new RunQueue({
6161
...testOptions,
62+
logLevel: "debug",
6263
queueSelectionStrategy: new FairQueueSelectionStrategy({
6364
redis: {
6465
keyPrefix: "runqueue:test:",
@@ -67,6 +68,10 @@ describe("RunQueue Concurrency Sweeper", () => {
6768
},
6869
keys: testOptions.keys,
6970
}),
71+
workerOptions: {
72+
pollIntervalMs: 100,
73+
immediatePollIntervalMs: 100,
74+
},
7075
redis: {
7176
keyPrefix: "runqueue:test:",
7277
host: redisContainer.getHost(),

packages/redis-worker/src/worker.ts

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -693,9 +693,8 @@ class Worker<TCatalog extends WorkerCatalog> {
693693
const scheduledAt = this.calculateNextScheduledAt(cron, lastTimestamp);
694694
const identifier = [job, this.timestampIdentifier(scheduledAt)].join(":");
695695
// Calculate the availableAt date by calculating a random number between -jitter/2 and jitter/2 and adding it to the scheduledAt
696-
const availableAt = jitter
697-
? new Date(scheduledAt.getTime() + Math.random() * jitter - jitter / 2)
698-
: scheduledAt;
696+
const appliedJitter = typeof jitter === "number" ? Math.random() * jitter - jitter / 2 : 0;
697+
const availableAt = new Date(scheduledAt.getTime() + appliedJitter);
699698

700699
const enqueued = await this.enqueueOnce({
701700
id: identifier,
@@ -715,6 +714,8 @@ class Worker<TCatalog extends WorkerCatalog> {
715714
scheduledAt,
716715
enqueued,
717716
availableAt,
717+
appliedJitter,
718+
jitter,
718719
});
719720

720721
return {

0 commit comments

Comments
 (0)