Skip to content

BREAKING CHANGE: track nextTaskId to prevent timeout race with long-running task scheduling duplicate repeat tasks, remove retryOnTimeoutCount #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vkarpov15 merged 1 commit into from
Oct 29, 2026
Merged

BREAKING CHANGE: track nextTaskId to prevent timeout race with long-running task scheduling duplicate repeat tasks, remove retryOnTimeoutCount #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 33 additions & 45 deletions src/taskSchema.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,17 +38,16 @@ 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
},
previousTaskId: {
type: mongoose.ObjectId
},
nextTaskId: {
type: mongoose.ObjectId
},
originalTaskId: {
type: mongoose.ObjectId
},
Expand Down Expand Up @@ -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);
}
};

Expand All @@ -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' }
Comment on lines +217 to +220

@chatgpt-codex-connector chatgpt-codex-connector Bot Oct 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Recover successor claims when task creation fails

If timeout expiry successfully sets nextTaskId but the subsequent Task.create() fails, the parent permanently points to a nonexistent successor. When the still-running handler later finishes, its scheduling attempt fails this claim check and silently returns, stopping the repeating task chain. Before this change, that handler completion could retry creating the successor. Make successor creation recoverable, for example by allowing subsequent callers to idempotently create the already-claimed successor.

Useful? React with 👍 / 👎.

);
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) {
Expand Down
76 changes: 35 additions & 41 deletions test/task.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
Loading