From 3c98fd214de1b8a18163c9469a1ff159217f4eb9 Mon Sep 17 00:00:00 2001 From: Valeri Karpov Date: Thu, 29 Oct 2026 17:31:44 -0400 Subject: [PATCH] BREAKING CHANGE: track nextTaskId to prevent timeout race with long-running task scheduling duplicate repeat tasks, remove retryOnTimeoutCount --- src/taskSchema.js | 78 ++++++++++++++++++++--------------------------- test/task.test.js | 76 +++++++++++++++++++++------------------------ 2 files changed, 68 insertions(+), 86 deletions(-) diff --git a/src/taskSchema.js b/src/taskSchema.js index 87c0c27..287d2cf 100644 --- a/src/taskSchema.js +++ b/src/taskSchema.js @@ -38,10 +38,6 @@ const taskSchema = new mongoose.Schema({ finishedRunningAt: { type: Date }, - retryOnTimeoutCount: { - type: Number, - default: 0 - }, // If this task is still running after this time, it will be marked as `timed_out`. timeoutAt: { type: Date @@ -49,6 +45,9 @@ const taskSchema = new mongoose.Schema({ previousTaskId: { type: mongoose.ObjectId }, + nextTaskId: { + type: mongoose.ObjectId + }, originalTaskId: { type: mongoose.ObjectId }, @@ -192,25 +191,7 @@ taskSchema.statics.expireTimedOutTasks = async function expireTimedOutTasks(opti break; } - if (task.retryOnTimeoutCount > 0) { - // Copy task data but remove _id so MongoDB generates a new one - const taskData = task.toObject({ virtuals: false }); - delete taskData._id; - await Task.create({ - ...taskData, - status: 'pending', - retryOnTimeoutCount: task.retryOnTimeoutCount - 1, - startedRunningAt: null, - finishedRunningAt: null, - workerName: null, - error: null, - result: null, - timeoutAt: null, - schedulingTimeoutAt: now.valueOf() + 10 * 60 * 1000 - }); - } else { - await _handleRepeatingTask(Task, task); - } + await _handleRepeatingTask(Task, task); } }; @@ -221,31 +202,38 @@ taskSchema.statics.registerHandler = async function registerHandler(name, fn) { }; async function _handleRepeatingTask(Task, task) { + let scheduledAt; 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 - }); + scheduledAt = new Date(task.nextScheduledAt); } 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 - }); + scheduledAt = new Date(task.scheduledAt.valueOf() + task.repeatAfterMS); + } else { + return; } + + const nextTaskId = new mongoose.Types.ObjectId(); + // Timeout expiry and handler completion may both try to schedule the next task. + // Claim the successor on the parent so only one caller can create it. + const claimedTask = await Task.findOneAndUpdate( + { _id: task._id, nextTaskId: null }, + { $set: { nextTaskId } }, + { returnDocument: 'before' } + ); + if (claimedTask == null) { + return; + } + + return Task.create({ + _id: nextTaskId, + 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) { diff --git a/test/task.test.js b/test/task.test.js index de8e565..928e93f 100644 --- a/test/task.test.js +++ b/test/task.test.js @@ -377,51 +377,45 @@ describe('Task', function() { assert.ok(repeated.scheduledAt.valueOf() === repeatTaskObj.scheduledAt.valueOf() + 60000); }); - it('creates a retry task when a timed out task has retryOnTimeoutCount', async function() { - Task.registerHandler('timeoutRetry', async () => { - // handler intentionally does nothing (we'll simulate a timeout) + it('does not duplicate a repeat when a handler finishes after timeout expiry', async function() { + let resolve; + let started; + const handlerStarted = new Promise(resolve => { started = resolve; }); + const handlerFinished = new Promise(_resolve => { resolve = _resolve; }); + Task.registerHandler('lateRepeatingJob', async () => { + started(); + await handlerFinished; }); - const scheduledAt = new Date(now.valueOf() - 11 * 60 * 1000); - const startedRunningAt = new Date(now.valueOf() - 20 * 60 * 1000); - const timeoutAt = new Date(now.valueOf() - 10 * 60 * 1000 - 1000); - - let timedOutTask = await Task.create({ - name: 'timeoutRetry', - scheduledAt, - startedRunningAt, - timeoutAt, + const task = await Task.create({ + name: 'lateRepeatingJob', status: 'in_progress', - timeoutMS: 10000, - retryOnTimeoutCount: 2, - params: { foo: 'bar' } - }); - - await Task.expireTimedOutTasks(); - - timedOutTask = await Task.findById(timedOutTask._id); - assert.ok(timedOutTask); - assert.equal(timedOutTask.status, 'timed_out'); - assert.equal(timedOutTask.finishedRunningAt.valueOf(), now.valueOf()); + scheduledAt: time.now(), + repeatAfterMS: 24 * 60 * 60 * 1000, + timeoutAt: new Date(now.valueOf() - 20 * 60 * 1000) + }); + const execution = Task.execute(task); + await handlerStarted; + try { + await Task.expireTimedOutTasks(); + const expiredTask = await Task.findById(task._id); + assert.equal(expiredTask.status, 'timed_out'); + } finally { + resolve(); + await execution; + } - const retryTask = await Task.findOne({ - name: 'timeoutRetry', - status: 'pending', - retryOnTimeoutCount: 1 - }); - assert.ok(retryTask, 'Retry task should be created'); - assert.equal(retryTask.scheduledAt.valueOf(), scheduledAt.valueOf()); - assert.strictEqual(retryTask.startedRunningAt, null); - assert.strictEqual(retryTask.finishedRunningAt, null); - assert.strictEqual(retryTask.workerName, null); - assert.strictEqual(retryTask.timeoutAt, null); - assert.deepStrictEqual(retryTask.toObject().params, { foo: 'bar' }); - assert.ok(retryTask.error.$isEmpty()); - assert.strictEqual(retryTask.result, null); - assert.equal( - retryTask.schedulingTimeoutAt.valueOf(), - now.valueOf() + 10 * 60 * 1000 - ); + const repeats = await Task.find({ previousTaskId: task._id }); + assert.equal(repeats.length, 1); + const reloadedTask = await Task.findById(task._id); + assert.equal(reloadedTask.nextTaskId.toString(), repeats[0]._id.toString()); + assert.equal(repeats[0].originalTaskId.toString(), task._id.toString()); + + // The successor must be able to schedule its own successor. + await Task.execute(repeats[0]); + const nextRepeat = await Task.findOne({ previousTaskId: repeats[0]._id }); + assert.ok(nextRepeat); + assert.equal(nextRepeat.originalTaskId.toString(), task._id.toString()); }); it('handles scheduling_timed_out tasks and schedules next repeat if needed', async function() {