better handling for stuck tasks: clear hanging tasks and avoid running tasks that were scheduled for a long time ago

.github/workflows/lint.yml

@@ -0,0 +1,18 @@
+name: Lint
+on:
+  pull_request:
+  push:
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Setting up the node version
+        uses: actions/setup-node@v3
+        with:
+          node-version: 20.19.0
+      - name: setup project
+        run: npm i
+      - name: run lint
+        run: |
+          npm run lint

.github/workflows/test.yml

@@ -11,9 +11,9 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        node: [18, 20]
-        os: [ubuntu-20.04]
-        mongo: [5.0.8]
+        node: [22]
+        os: [ubuntu-22.04]
+        mongo: [8.2.0]
     name: Node ${{ matrix.node }} MongoDB ${{ matrix.mongo }}
     steps:
       - uses: actions/checkout@a12a3943b4bdde767164f792f33f40b04645d846 # v3
@@ -27,11 +27,10 @@ jobs:
 
       - name: Setup
         run: |
-          wget -q https://fastdl.mongodb.org/linux/mongodb-linux-x86_64-ubuntu2004-${{ matrix.mongo }}.tgz
-          tar xf mongodb-linux-x86_64-ubuntu2004-${{ matrix.mongo }}.tgz
+          wget -q https://fastdl.mongodb.org/linux/mongodb-linux-x86_64-ubuntu2204-${{ matrix.mongo }}.tgz
+          tar xf mongodb-linux-x86_64-ubuntu2204-${{ matrix.mongo }}.tgz
           mkdir -p ./data/db/27017 ./data/db/27000
-          ./mongodb-linux-x86_64-ubuntu2004-${{ matrix.mongo }}/bin/mongod --setParameter ttlMonitorSleepSecs=1 --fork --dbpath ./data/db/27017 --syslog --port 27017
+          ./mongodb-linux-x86_64-ubuntu2204-${{ matrix.mongo }}/bin/mongod --setParameter ttlMonitorSleepSecs=1 --fork --dbpath ./data/db/27017 --syslog --port 27017
           sleep 2
-          mongod --version
-          echo `pwd`/mongodb-linux-x86_64-ubuntu2004-${{ matrix.mongo }}/bin >> $GITHUB_PATH
+          echo `pwd`/mongodb-linux-x86_64-ubuntu2204-${{ matrix.mongo }}/bin >> $GITHUB_PATH
       - run: npm test

eslint.config.js

@@ -0,0 +1,21 @@
+'use strict';
+
+const config = require('@masteringjs/eslint-config');
+const { defineConfig } = require('eslint/config');
+
+module.exports = defineConfig([
+  {
+    files: ['src/*.js'],
+    languageOptions: {
+      sourceType: 'commonjs',
+      globals: {
+        fetch: true,
+        setTimeout: true,
+        process: true,
+        console: true,
+        clearTimeout: true
+      }
+    },
+    extends: [config]
+  }
+]);

package.json

@@ -3,6 +3,7 @@
   "version": "0.3.0",
   "private": false,
   "scripts": {
+    "lint": "eslint .",
     "test": "mocha test/*.test.js"
   },
   "repository": {
@@ -16,6 +17,8 @@
     "mongoose": "^6.7.0 || 7.x || 8.x"
   },
   "devDependencies": {
+    "@masteringjs/eslint-config": "0.1.1",
+    "eslint": "9.30.0",
     "mocha": "10.1.0",
     "mongoose": "8.x",
     "sinon": "15.2.0"

src/taskSchema.js

@@ -8,13 +8,21 @@ const taskSchema = new mongoose.Schema({
     type: String,
     required: true
   },
+  // The time at which the task was scheduled to run. The task will start running at or after this time.
   scheduledAt: {
     type: Date,
     required: true
   },
+  // If the task has not started running by this time, it will be marked as `scheduling_timed_out`
+  // and the next scheduled task will be created.
+  schedulingTimeoutAt: {
+    type: Date
+  },
+  // The next time this task will be scheduled to run.
   nextScheduledAt: {
     type: Date
   },
+  // When this task is done, automatically schedule the next task for scheduledAt + repeatAfterMS
   repeatAfterMS: {
     type: Number
   },
@@ -30,6 +38,10 @@ const taskSchema = new mongoose.Schema({
   finishedRunningAt: {
     type: Date
   },
+  // If this task is still running after this time, it will be marked as `timed_out`.
+  timeoutAt: {
+    type: Date
+  },
   previousTaskId: {
     type: mongoose.ObjectId
   },
@@ -42,7 +54,22 @@ const taskSchema = new mongoose.Schema({
   status: {
     type: String,
     default: 'pending',
-    enum: ['pending', 'in_progress', 'succeeded', 'failed', 'cancelled']
+    enum: [
+      // Waiting to run
+      'pending',
+      // Currently running
+      'in_progress',
+      // Completed successfully
+      'succeeded',
+      // Error occurred while executing the task
+      'failed',
+      // Cancelled by user
+      'cancelled',
+      // Task execution timed out
+      'timed_out',
+      // Timed out waiting for a worker to pick up the task
+      'scheduling_timed_out'
+    ]
   },
   result: 'Mixed',
   error: {
@@ -63,8 +90,8 @@ taskSchema.methods.log = function log(message, extra) {
 
 taskSchema.statics.cancelTask = async function cancelTask(filter) {
   if (filter != null) {
-    filter = { $and: [{ status: 'pending' }, filter] }
-  };
+    filter = { $and: [{ status: 'pending' }, filter] };
+  }
   const task = await this.findOneAndUpdate(filter, { status: 'cancelled', cancelledAt: new Date() }, { returnDocument: 'after' });
   return task;
 };
@@ -92,7 +119,7 @@ taskSchema.statics.startPolling = function startPolling(options) {
     doPoll.call(this);
     this._cancel = () => {
       cancelled = true;
-      clearTimeout(timeout)
+      clearTimeout(timeout);
     };
   }
   return this._cancel;
@@ -101,6 +128,12 @@ taskSchema.statics.startPolling = function startPolling(options) {
     if (cancelled) {
       return;
     }
+
+    const Task = this;
+
+    // Expire tasks that have timed out (refactored to separate function)
+    await Task.expireTimedOutTasks();
+
     this._currentPoll = this.poll(pollOptions);
     await this._currentPoll.then(
       () => {
@@ -114,12 +147,68 @@ taskSchema.statics.startPolling = function startPolling(options) {
   }
 };
 
+// Refactor logic for expiring timed out tasks here
+taskSchema.statics.expireTimedOutTasks = async function expireTimedOutTasks() {
+  const now = time.now();
+  const Task = this;
+  while (true) {
+    const task = await Task.findOneAndUpdate(
+      {
+        status: 'in_progress',
+        startedRunningAt: { $exists: true },
+        timeoutAt: { $exists: true, $lte: now }
+      },
+      {
+        $set: {
+          status: 'timed_out',
+          finishedRunningAt: now
+        }
+      },
+      { new: true }
+    );
+
+    if (!task) {
+      break;
+    }
+
+    await _handleRepeatingTask(Task, task);
+  }
+};
+
 taskSchema.statics.registerHandler = async function registerHandler(name, fn) {
   this._handlers = this._handlers || new Map();
   this._handlers.set(name, fn);
   return this;
 };
 
+async function _handleRepeatingTask(Task, task) {
+  if (task.nextScheduledAt != null) {
+    const scheduledAt = new Date(task.nextScheduledAt);
+    return Task.create({
+      name: task.name,
+      scheduledAt,
+      repeatAfterMS: task.repeatAfterMS,
+      params: task.params,
+      previousTaskId: task._id,
+      originalTaskId: task.originalTaskId || task._id,
+      timeoutMS: task.timeoutMS,
+      schedulingTimeoutAt: scheduledAt.valueOf() + 10 * 60 * 1000
+    });
+  } else if (task.repeatAfterMS != null) {
+    const scheduledAt = new Date(task.scheduledAt.valueOf() + task.repeatAfterMS);
+    return Task.create({
+      name: task.name,
+      scheduledAt,
+      repeatAfterMS: task.repeatAfterMS,
+      params: task.params,
+      previousTaskId: task._id,
+      originalTaskId: task.originalTaskId || task._id,
+      timeoutMS: task.timeoutMS,
+      schedulingTimeoutAt: scheduledAt.valueOf() + 10 * 60 * 1000
+    });
+  }
+}
+
 taskSchema.statics.registerHandlers = async function registerHandlers(obj, prefix) {
   this._handlers = this._handlers || new Map();
   for (const key of Object.keys(obj)) {
@@ -145,21 +234,23 @@ taskSchema.statics.poll = async function poll(opts) {
   const additionalParams = workerName ? { workerName } : {};
 
   while (true) {
-    let tasksInProgress = [];
+    const tasksInProgress = [];
     for (let i = 0; i < parallel; ++i) {
       const now = time.now();
       const task = await this.findOneAndUpdate(
         { status: 'pending', scheduledAt: { $lte: now } },
-        { status: 'in_progress', startedRunningAt: now, ...additionalParams },
+        {
+          status: 'in_progress',
+          timeoutAt: new Date(now.valueOf() + 10 * 60 * 1000), // 10 minutes from startedRunningAt
+          ...additionalParams
+        },
         { new: false }
       );
 
       if (task == null || task.status !== 'pending') {
         break;
       }
 
-      task.status = 'in_progress';
-
       tasksInProgress.push(this.execute(task));
     }
 
@@ -176,6 +267,18 @@ taskSchema.statics.execute = async function(task) {
     return null;
   }
 
+  task.status = 'in_progress';
+  const now = time.now();
+  task.startedRunningAt = now;
+
+  if (task.schedulingTimeoutAt && task.schedulingTimeoutAt < now) {
+    task.status = 'scheduling_timed_out';
+    task.finishedRunningAt = now;
+    await task.save();
+    await _handleRepeatingTask(this, task);
+    return task;
+  }
+
   try {
     let result = null;
     if (typeof task.timeoutMS === 'number') {
@@ -204,27 +307,7 @@ taskSchema.statics.execute = async function(task) {
     await task.save();
   }
 
-  if (task.nextScheduledAt != null) {
-    await this.create({
-      name: task.name,
-      scheduledAt: new Date(task.nextScheduledAt),
-      repeatAfterMS: task.repeatAfterMS,
-      params: task.params,
-      previousTaskId: task._id,
-      originalTaskId: task.originalTaskId || task._id,
-      timeoutMS: task.timeoutMS
-    });
-  } else if (task.repeatAfterMS != null) {
-    await this.create({
-      name: task.name,
-      scheduledAt: new Date(task.scheduledAt.valueOf() + task.repeatAfterMS),
-      repeatAfterMS: task.repeatAfterMS,
-      params: task.params,
-      previousTaskId: task._id,
-      originalTaskId: task.originalTaskId || task._id,
-      timeoutMS: task.timeoutMS
-    });
-  }
+  await _handleRepeatingTask(this, task);
 
   return task;
 };
@@ -241,6 +324,7 @@ taskSchema.statics.schedule = async function schedule(name, scheduledAt, params,
     scheduledAt,
     params,
     repeatAfterMS,
+    schedulingTimeoutAt: scheduledAt.valueOf() + 10 * 60 * 1000,
     ...options
   });
 };