From 70eb6389cad6869cc15b984b855bc37c95f70647 Mon Sep 17 00:00:00 2001 From: roggervalf Date: Fri, 6 Dec 2024 23:06:55 -0600 Subject: [PATCH 1/7] feat(queue): enhance getJobSchedulers method to include template information --- src/classes/job-scheduler.ts | 15 ++++++++++----- src/classes/queue.ts | 10 +++++++--- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/src/classes/job-scheduler.ts b/src/classes/job-scheduler.ts index 1f3493c6e6..85badf6fd3 100644 --- a/src/classes/job-scheduler.ts +++ b/src/classes/job-scheduler.ts @@ -202,11 +202,11 @@ export class JobScheduler extends QueueBase { return this.scripts.removeJobScheduler(jobSchedulerId); } - private async getSchedulerData( + private async getSchedulerData( client: RedisClient, key: string, next?: number, - ): Promise { + ): Promise> { const jobData = await client.hgetall(this.toKey('repeat:' + key)); if (jobData) { @@ -217,6 +217,11 @@ export class JobScheduler extends QueueBase { tz: jobData.tz || null, pattern: jobData.pattern || null, every: jobData.every || null, + ...(jobData.data || jobData.opts + ? { + template: this.getTemplateFromJSON(jobData.data, jobData.opts), + } + : {}), next, }; } @@ -279,11 +284,11 @@ export class JobScheduler extends QueueBase { return template; } - async getJobSchedulers( + async getJobSchedulers( start = 0, end = -1, asc = false, - ): Promise { + ): Promise[]> { const client = await this.client; const jobSchedulersKey = this.keys.repeat; @@ -294,7 +299,7 @@ export class JobScheduler extends QueueBase { const jobs = []; for (let i = 0; i < result.length; i += 2) { jobs.push( - this.getSchedulerData(client, result[i], parseInt(result[i + 1])), + this.getSchedulerData(client, result[i], parseInt(result[i + 1])), ); } return Promise.all(jobs); diff --git a/src/classes/queue.ts b/src/classes/queue.ts index 3afa5691c0..5c61c88e2c 100644 --- a/src/classes/queue.ts +++ b/src/classes/queue.ts @@ -573,7 +573,7 @@ export class Queue< * @param id - identifier of scheduler. */ async getJobScheduler(id: string): Promise> { - return (await this.jobScheduler).getJobScheduler(id); + return (await this.jobScheduler).getJobScheduler(id); } /** @@ -588,8 +588,12 @@ export class Queue< start?: number, end?: number, asc?: boolean, - ): Promise { - return (await this.jobScheduler).getJobSchedulers(start, end, asc); + ): Promise[]> { + return (await this.jobScheduler).getJobSchedulers( + start, + end, + asc, + ); } /** From 980cb29d50ae24b8195363a8b35c62152eba578f Mon Sep 17 00:00:00 2001 From: roggervalf Date: Fri, 6 Dec 2024 23:12:44 -0600 Subject: [PATCH 2/7] test: update test case --- src/interfaces/telemetry.ts | 16 ++++++++-------- tests/test_job_scheduler.ts | 25 ++++++++++++++++++++++++- 2 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/interfaces/telemetry.ts b/src/interfaces/telemetry.ts index e55c0990dc..d39794d34a 100644 --- a/src/interfaces/telemetry.ts +++ b/src/interfaces/telemetry.ts @@ -36,8 +36,8 @@ export interface ContextManager { /** * Creates a new context and sets it as active for the fn passed as last argument * - * @param context - * @param fn + * @param context - + * @param fn - */ with any>( context: Context, @@ -54,7 +54,7 @@ export interface ContextManager { * is the mechanism used to propagate the context across a distributed * application. * - * @param context + * @param context - */ getMetadata(context: Context): string; @@ -62,8 +62,8 @@ export interface ContextManager { * Creates a new context from a serialized version effectively * linking the new context to the parent context. * - * @param activeContext - * @param metadata + * @param activeContext - + * @param metadata - */ fromMetadata(activeContext: Context, metadata: string): Context; } @@ -78,9 +78,9 @@ export interface Tracer { * context. If the context is not provided, the current active context should be * used. * - * @param name - * @param options - * @param context + * @param name - + * @param options - + * @param context - */ startSpan(name: string, options?: SpanOptions, context?: Context): Span; } diff --git a/tests/test_job_scheduler.ts b/tests/test_job_scheduler.ts index ee329207aa..ffd6eee0d8 100644 --- a/tests/test_job_scheduler.ts +++ b/tests/test_job_scheduler.ts @@ -1445,7 +1445,11 @@ describe('Job Scheduler', function () { }); }); - const repeatableJob = await queue.upsertJobScheduler('test', repeatOpts); + const repeatableJob = await queue.upsertJobScheduler('test', repeatOpts, { + name: 'a', + data: { foo: 'bar' }, + opts: { priority: 1 }, + }); const delayedCount = await queue.getDelayedCount(); expect(delayedCount).to.be.equal(1); @@ -1463,6 +1467,25 @@ describe('Job Scheduler', function () { const count = await queue.count(); expect(count).to.be.equal(1); expect(jobSchedulers).to.have.length(1); + + expect(jobSchedulers[0]).to.deep.equal({ + key: 'test', + name: 'a', + endDate: null, + tz: null, + pattern: '0 * 1 * *', + every: null, + next: 25200000, + template: { + data: { + foo: 'bar', + }, + opts: { + priority: 1, + }, + }, + }); + await worker.close(); }); From 9acaf418029037c213d95c48f1bc806daa2a6431 Mon Sep 17 00:00:00 2001 From: roggervalf Date: Sun, 8 Dec 2024 08:23:51 -0600 Subject: [PATCH 3/7] feat(scheduler): expose template data in getJobSchedulers --- src/classes/job-scheduler.ts | 40 ++++++++++++------------------ src/classes/queue.ts | 2 +- src/classes/scripts.ts | 14 +++++++++++ src/commands/getJobScheduler-1.lua | 19 ++++++++++++++ src/commands/updateJobOption-1.lua | 26 ------------------- tests/test_job_scheduler.ts | 6 ++++- tests/test_metrics.ts | 2 +- 7 files changed, 56 insertions(+), 53 deletions(-) create mode 100644 src/commands/getJobScheduler-1.lua delete mode 100644 src/commands/updateJobOption-1.lua diff --git a/src/classes/job-scheduler.ts b/src/classes/job-scheduler.ts index 85badf6fd3..bd47466ce5 100644 --- a/src/classes/job-scheduler.ts +++ b/src/classes/job-scheduler.ts @@ -11,7 +11,7 @@ import { Job } from './job'; import { QueueBase } from './queue-base'; import { RedisConnection } from './redis-connection'; import { SpanKind, TelemetryAttributes } from '../enums'; -import { optsAsJSON, optsFromJSON } from '../utils'; +import { array2obj, optsAsJSON, optsFromJSON } from '../utils'; export class JobScheduler extends QueueBase { private repeatStrategy: RepeatStrategy; @@ -209,6 +209,14 @@ export class JobScheduler extends QueueBase { ): Promise> { const jobData = await client.hgetall(this.toKey('repeat:' + key)); + return this.transformSchedulerData(key, jobData, next); + } + + private async transformSchedulerData( + key: string, + jobData: any, + next?: number, + ): Promise> { if (jobData) { return { key, @@ -244,30 +252,14 @@ export class JobScheduler extends QueueBase { }; } - async getJobScheduler(id: string): Promise> { - const client = await this.client; - const schedulerAttributes = await client.hgetall( - this.toKey('repeat:' + id), - ); + async getScheduler(id: string): Promise> { + const [rawJobData, next] = await this.scripts.getJobScheduler(id); - if (schedulerAttributes) { - return { - key: id, - name: schedulerAttributes.name, - endDate: parseInt(schedulerAttributes.endDate) || null, - tz: schedulerAttributes.tz || null, - pattern: schedulerAttributes.pattern || null, - every: schedulerAttributes.every || null, - ...(schedulerAttributes.data || schedulerAttributes.opts - ? { - template: this.getTemplateFromJSON( - schedulerAttributes.data, - schedulerAttributes.opts, - ), - } - : {}), - }; - } + return this.transformSchedulerData( + id, + rawJobData ? array2obj(rawJobData) : null, + next ? parseInt(next) : null, + ); } private getTemplateFromJSON( diff --git a/src/classes/queue.ts b/src/classes/queue.ts index 5c61c88e2c..1525d13254 100644 --- a/src/classes/queue.ts +++ b/src/classes/queue.ts @@ -573,7 +573,7 @@ export class Queue< * @param id - identifier of scheduler. */ async getJobScheduler(id: string): Promise> { - return (await this.jobScheduler).getJobScheduler(id); + return (await this.jobScheduler).getScheduler(id); } /** diff --git a/src/classes/scripts.ts b/src/classes/scripts.ts index 0897d905a9..dc487b42d1 100644 --- a/src/classes/scripts.ts +++ b/src/classes/scripts.ts @@ -1088,6 +1088,20 @@ export class Scripts { ]); } + getJobSchedulerArgs(id: string): string[] { + const keys: string[] = [this.queue.keys.repeat]; + + return keys.concat([id]); + } + + async getJobScheduler(id: string): Promise<[any, string | null]> { + const client = await this.queue.client; + + const args = this.getJobSchedulerArgs(id); + + return this.execCommand(client, 'getJobScheduler', args); + } + retryJobArgs( jobId: string, lifo: boolean, diff --git a/src/commands/getJobScheduler-1.lua b/src/commands/getJobScheduler-1.lua new file mode 100644 index 0000000000..324bdb58eb --- /dev/null +++ b/src/commands/getJobScheduler-1.lua @@ -0,0 +1,19 @@ +--[[ + Get job scheduler record. + + Input: + KEYS[1] 'repeat' key + + ARGV[1] id +]] + +local rcall = redis.call +local jobSchedulerKey = KEYS[1] .. ":" .. ARGV[1] + +local score = rcall("ZSCORE", KEYS[1], ARGV[1]) + +if score then + return {rcall("HGETALL", jobSchedulerKey), score} -- get job data +end + +return {nil, nil} diff --git a/src/commands/updateJobOption-1.lua b/src/commands/updateJobOption-1.lua deleted file mode 100644 index 03949faf29..0000000000 --- a/src/commands/updateJobOption-1.lua +++ /dev/null @@ -1,26 +0,0 @@ ---[[ - Update a job option - - Input: - KEYS[1] Job id key - - ARGV[1] field - ARGV[2] value - - Output: - 0 - OK - -1 - Missing job. -]] -local rcall = redis.call - -if rcall("EXISTS", KEYS[1]) == 1 then -- // Make sure job exists - - local opts = rcall("HGET", KEYS[1], "opts") - local jsonOpts = cjson.decode(opts) - jsonOpts[ARGV[1]] = ARGV[2] - - rcall("HSET", KEYS[1], "opts", cjson.encode(jsonOpts)) - return 0 -else - return -1 -end diff --git a/tests/test_job_scheduler.ts b/tests/test_job_scheduler.ts index ffd6eee0d8..a44b17549a 100644 --- a/tests/test_job_scheduler.ts +++ b/tests/test_job_scheduler.ts @@ -360,6 +360,7 @@ describe('Job Scheduler', function () { tz: null, pattern: '*/2 * * * * *', every: null, + next: 1486481042000, template: { data: { foo: 'bar', @@ -698,6 +699,7 @@ describe('Job Scheduler', function () { key: 'rrule', name: 'rrule', endDate: null, + next: 1486481042000, tz: null, pattern: 'RRULE:FREQ=SECONDLY;INTERVAL=2;WKST=MO', every: null, @@ -1424,6 +1426,8 @@ describe('Job Scheduler', function () { describe('when repeatable job fails', function () { it('should continue repeating', async function () { + const date = new Date('2024-12-08 9:24:00'); + this.clock.setSystemTime(date); const repeatOpts = { pattern: '0 * 1 * *', }; @@ -1475,7 +1479,7 @@ describe('Job Scheduler', function () { tz: null, pattern: '0 * 1 * *', every: null, - next: 25200000, + next: 1735714800000, template: { data: { foo: 'bar', diff --git a/tests/test_metrics.ts b/tests/test_metrics.ts index 6433afc8b2..f59d4e44d7 100644 --- a/tests/test_metrics.ts +++ b/tests/test_metrics.ts @@ -28,7 +28,7 @@ describe('metrics', function () { }); beforeEach(function () { - this.clock = sinon.useFakeTimers(); + this.clock = sinon.useFakeTimers({ shouldClearNativeTimers: true }); }); beforeEach(async function () { From 3f050864b5ac063eaaaf4c3c922727ce5699c66e Mon Sep 17 00:00:00 2001 From: roggervalf Date: Sun, 8 Dec 2024 08:48:17 -0600 Subject: [PATCH 4/7] test: fix test case --- tests/test_job_scheduler.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_job_scheduler.ts b/tests/test_job_scheduler.ts index a44b17549a..3b8e40822d 100644 --- a/tests/test_job_scheduler.ts +++ b/tests/test_job_scheduler.ts @@ -342,7 +342,7 @@ describe('Job Scheduler', function () { ); const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {}); - const date = new Date('2017-02-07 9:24:00'); + const date = new Date('2017-02-07T15:24:00.000Z'); this.clock.setSystemTime(date); await queue.upsertJobScheduler( From db1ba9038489bbf8290877beb7fe902daba0e611 Mon Sep 17 00:00:00 2001 From: roggervalf Date: Sun, 8 Dec 2024 08:59:01 -0600 Subject: [PATCH 5/7] test: update test case --- tests/test_job_scheduler.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_job_scheduler.ts b/tests/test_job_scheduler.ts index 3b8e40822d..7b402ad903 100644 --- a/tests/test_job_scheduler.ts +++ b/tests/test_job_scheduler.ts @@ -683,7 +683,7 @@ describe('Job Scheduler', function () { ); const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {}); - const date = new Date('2017-02-07 9:24:00'); + const date = new Date('2017-02-07T15:24:00.000Z'); this.clock.setSystemTime(date); const repeat = { @@ -1426,7 +1426,7 @@ describe('Job Scheduler', function () { describe('when repeatable job fails', function () { it('should continue repeating', async function () { - const date = new Date('2024-12-08 9:24:00'); + const date = new Date('2017-02-07T15:24:00.000Z'); this.clock.setSystemTime(date); const repeatOpts = { pattern: '0 * 1 * *', From 29568d467616a8d722c512510da7134432bad6eb Mon Sep 17 00:00:00 2001 From: roggervalf Date: Sun, 8 Dec 2024 09:19:55 -0600 Subject: [PATCH 6/7] test: update test case --- tests/test_job_scheduler.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_job_scheduler.ts b/tests/test_job_scheduler.ts index 7b402ad903..cc7928cbb2 100644 --- a/tests/test_job_scheduler.ts +++ b/tests/test_job_scheduler.ts @@ -1479,7 +1479,7 @@ describe('Job Scheduler', function () { tz: null, pattern: '0 * 1 * *', every: null, - next: 1735714800000, + next: 1488351600000, template: { data: { foo: 'bar', From 093fdaab4a123f4642544ac4078fb1624431a52c Mon Sep 17 00:00:00 2001 From: roggervalf Date: Sun, 8 Dec 2024 09:44:08 -0600 Subject: [PATCH 7/7] test: update test case --- tests/test_job_scheduler.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_job_scheduler.ts b/tests/test_job_scheduler.ts index cc7928cbb2..427cb99cee 100644 --- a/tests/test_job_scheduler.ts +++ b/tests/test_job_scheduler.ts @@ -1430,6 +1430,7 @@ describe('Job Scheduler', function () { this.clock.setSystemTime(date); const repeatOpts = { pattern: '0 * 1 * *', + tz: 'Asia/Calcutta', }; const worker = new Worker( @@ -1476,10 +1477,10 @@ describe('Job Scheduler', function () { key: 'test', name: 'a', endDate: null, - tz: null, + tz: 'Asia/Calcutta', pattern: '0 * 1 * *', every: null, - next: 1488351600000, + next: 1488310200000, template: { data: { foo: 'bar',