fix: remove the old expired jobs

restorecommerce · Mar 15, 2024 · 5f91e22 · 5f91e22
1 parent f7be702
commit 5f91e22
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 12 deletions.
diff --git a/cfg/config.json b/cfg/config.json
@@ -14,7 +14,7 @@
   "logger": {
     "console": {
       "handleExceptions": false,
-      "level": "error",
+      "level": "silly",
       "colorize": true,
       "prettyPrint": true
     },
@@ -147,7 +147,7 @@
       "configUpdateResponse": {
         "messageObject": "io.restorecommerce.commandinterface.CommandResponse"
       },
-      "setApiKeyCommand": {  
+      "setApiKeyCommand": {
         "messageObject": "io.restorecommerce.commandinterface.CommandRequest"
       },
       "setApiKeyResponse": {
@@ -308,6 +308,7 @@
   },
   "queueCleanup": {
     "ttlAfterFinished": 86400000,
-    "cleanInterval": 86400000
+    "cleanInterval": 86400000,
+    "maxJobsToCleanLimit": 10000
   }
 }
diff --git a/src/schedulingService.ts b/src/schedulingService.ts
@@ -1096,11 +1096,11 @@ export class SchedulingService implements SchedulingServiceServiceImplementation
    * Clean up queues - removes complted and failed jobs from queue
    * @param {any} job clean up job
    */
-  async cleanupJobs(ttlAfterFinished) {
+  async cleanupJobs(ttlAfterFinished: number, maxJobsToCleanLimit: number) {
     for (let queue of this.queuesList) {
       try {
-        await queue.clean(ttlAfterFinished, 0, COMPLETED_JOB_STATE);
-        await queue.clean(ttlAfterFinished, 0, FAILED_JOB_STATE);
+        await queue.clean(ttlAfterFinished, maxJobsToCleanLimit, COMPLETED_JOB_STATE);
+        await queue.clean(ttlAfterFinished, maxJobsToCleanLimit, FAILED_JOB_STATE);
       } catch (err) {
         this.logger.error('Error cleaning up jobs', err);
       }
@@ -1110,7 +1110,7 @@ export class SchedulingService implements SchedulingServiceServiceImplementation
     await this.repeatJobIdRedisClient.set(QUEUE_CLEANUP, JSON.stringify(lastExecutedInterval));
   }
 
-  async setupCleanInterval(cleanInterval: number, ttlAfterFinished: number) {
+  async setupCleanInterval(cleanInterval: number, ttlAfterFinished: number, maxJobsToCleanLimit: number) {
     if (!ttlAfterFinished) {
       ttlAfterFinished = DEFAULT_CLEANUP_COMPLETED_JOBS;
     }
@@ -1128,11 +1128,11 @@ export class SchedulingService implements SchedulingServiceServiceImplementation
       // use setTimeout and then create interval on setTimeout
       this.logger.info('Restoring previous execution interval with set timeout', { time: cleanInterval - delta });
       setTimeout(async () => {
-        await this.cleanupJobs(ttlAfterFinished);
-        setInterval(this.cleanupJobs.bind(this), cleanInterval, ttlAfterFinished);
+        await this.cleanupJobs(ttlAfterFinished, maxJobsToCleanLimit);
+        setInterval(this.cleanupJobs.bind(this), cleanInterval, ttlAfterFinished, maxJobsToCleanLimit);
       }, cleanInterval - delta);
     } else {
-      setInterval(this.cleanupJobs.bind(this), cleanInterval, ttlAfterFinished);
+      setInterval(this.cleanupJobs.bind(this), cleanInterval, ttlAfterFinished, maxJobsToCleanLimit);
       this.logger.info('Clean up job interval set successfully');
     }
   }

diff --git a/src/worker.ts b/src/worker.ts
@@ -258,7 +258,7 @@ export class Worker {
     // cleanup job
     const queueCleanup = cfg.get('queueCleanup');
     if (queueCleanup?.cleanInterval && typeof queueCleanup.cleanInterval === 'number') {
-      await schedulingService.setupCleanInterval(queueCleanup.cleanInterval, queueCleanup.ttlAfterFinished);
+      await schedulingService.setupCleanInterval(queueCleanup.cleanInterval, queueCleanup.ttlAfterFinished, queueCleanup.maxJobsToCleanLimit);
     }
 
     const cis = new JobsCommandInterface(server, cfg,
@@ -354,7 +354,7 @@ export class Worker {
           }
           // check for double default
           let fileImport = await import(require_dir + externalFile);
-          if(fileImport?.default?.default) {
+          if (fileImport?.default?.default) {
             (async () => (await import(require_dir + externalFile)).default.default(cfg, logger, events, runWorker))().catch(err => {
               this.logger.error(`Error scheduling external job ${externalFile}`, { err: err.message });
             });