fix(scheduled-tasks): prevent dual-server double dispatch of daily tasks (#2713)
* fix(scheduled-tasks): claim schedule occurrences across server instances Two OpenChamber servers sharing project config each armed timers and both dispatched the same daily/weekly/cron/once slot (#2710). Claim the occurrence in shared config under a cross-process write lock before creating a session. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): harden occurrence claim failure and lock ownership Address PR review blockers: release running-slot bookkeeping when claim throws, avoid silently dropping an armed occurrence after a due-slack sync, verify lock-file ownership on release, and cover real on-disk lock behavior. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): always release running slot on state-write failures Wrap runTask bookkeeping in finally so claim, manual-start, and completion lock timeouts cannot stuck-run a task; drop the diskNext claim guard that suppressed later occurrences; recover unparseable locks via mtime age. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): stop re-arming past nextRunAt and clear stuck running Only schedule future nextRunAt values so once-task losers and claim-failed paths cannot spin delay-0 retries. Clear past once nextRunAt on claim, and on completion-write failure retry terminal status so manual runNow still returns the session instead of a hard 500 with lastStatus stuck running. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): release write chain on lock acquire timeout withProjectWriteLock left the in-process promise chain pending when acquireProjectFileLock timed out, wedging every later project write and stranding runTask before finally. Always release the chain; surface persistError on run; record once claim failures in task state. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> --------- Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com>
This commit is contained in:
Serhii Dziupin
committed by
GitHub
GitHub
co-authored by
Serhii Dziupin
Serhii Dziupin
parent
86e6a2ae76
commit
e99f6560be
@@ -254,6 +254,12 @@ const normalizeState = (value, fallback) => {
|
||||
const nextRunAt = typeof source.nextRunAt === 'number' && Number.isFinite(source.nextRunAt)
|
||||
? Math.max(0, Math.round(source.nextRunAt))
|
||||
: undefined;
|
||||
// Absolute ms of the schedule occurrence last claimed for dispatch. Used so
|
||||
// two OpenChamber server instances sharing this config cannot both start a
|
||||
// run for the same daily/weekly/cron/once slot (see issue #2710).
|
||||
const lastScheduledFor = typeof source.lastScheduledFor === 'number' && Number.isFinite(source.lastScheduledFor)
|
||||
? Math.max(0, Math.round(source.lastScheduledFor))
|
||||
: undefined;
|
||||
const lastSessionId = asNonEmptyString(source.lastSessionId);
|
||||
const lastErrorRaw = asNonEmptyString(source.lastError);
|
||||
const lastError = lastErrorRaw ? clampLength(lastErrorRaw, MAX_LAST_ERROR_LENGTH) : undefined;
|
||||
@@ -269,6 +275,7 @@ const normalizeState = (value, fallback) => {
|
||||
...(typeof lastRunAt === 'number' ? { lastRunAt } : {}),
|
||||
...(typeof lastDurationMs === 'number' ? { lastDurationMs } : {}),
|
||||
...(typeof nextRunAt === 'number' ? { nextRunAt } : {}),
|
||||
...(typeof lastScheduledFor === 'number' ? { lastScheduledFor } : {}),
|
||||
...(lastSessionId ? { lastSessionId } : {}),
|
||||
...(lastError ? { lastError } : {}),
|
||||
};
|
||||
@@ -360,6 +367,9 @@ export const createProjectConfigRuntime = (deps) => {
|
||||
});
|
||||
|
||||
const writeLocks = new Map();
|
||||
const PROJECT_FILE_LOCK_WAIT_MS = 10_000;
|
||||
const PROJECT_FILE_LOCK_STALE_MS = 60_000;
|
||||
const PROJECT_FILE_LOCK_RETRY_MS = 20;
|
||||
|
||||
const sanitizeProjectID = (projectID) => {
|
||||
const value = asNonEmptyString(projectID);
|
||||
@@ -377,6 +387,104 @@ export const createProjectConfigRuntime = (deps) => {
|
||||
return path.join(projectsDirPath, `${safeProjectID}.json`);
|
||||
};
|
||||
|
||||
const isProcessAlive = (pid) => {
|
||||
if (!Number.isInteger(pid) || pid <= 0) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch (error) {
|
||||
// EPERM: process exists but belongs to another user — treat as alive.
|
||||
return error?.code === 'EPERM';
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Cross-process exclusive lock for a project config file.
|
||||
* In-process chaining alone cannot serialize Electron (port 57123) and CLI
|
||||
* serve (port 3000) writers that share the same on-disk projects dir.
|
||||
*/
|
||||
const acquireProjectFileLock = async (projectID) => {
|
||||
const configPath = resolveProjectConfigPath(projectID);
|
||||
const lockPath = `${configPath}.lock`;
|
||||
const startedAt = Date.now();
|
||||
|
||||
await fsPromises.mkdir(path.dirname(configPath), { recursive: true });
|
||||
|
||||
while (Date.now() - startedAt < PROJECT_FILE_LOCK_WAIT_MS) {
|
||||
let handle;
|
||||
try {
|
||||
handle = await fsPromises.open(lockPath, 'wx');
|
||||
const lockPayload = {
|
||||
pid: process.pid,
|
||||
at: Date.now(),
|
||||
};
|
||||
await handle.writeFile(JSON.stringify(lockPayload));
|
||||
return {
|
||||
release: async () => {
|
||||
try {
|
||||
await handle.close();
|
||||
} catch {
|
||||
}
|
||||
// Only unlink if we still own the lock. A stale-recovery steal can
|
||||
// replace the file; unlinking blindly would drop the new owner's lock.
|
||||
try {
|
||||
const raw = await fsPromises.readFile(lockPath, 'utf8');
|
||||
const parsed = JSON.parse(raw);
|
||||
if (Number(parsed?.pid) !== process.pid) {
|
||||
return;
|
||||
}
|
||||
await fsPromises.unlink(lockPath);
|
||||
} catch {
|
||||
}
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
if (handle) {
|
||||
try {
|
||||
await handle.close();
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
if (error?.code !== 'EEXIST') {
|
||||
throw error;
|
||||
}
|
||||
|
||||
try {
|
||||
const raw = await fsPromises.readFile(lockPath, 'utf8');
|
||||
const parsed = JSON.parse(raw);
|
||||
const lockPid = Number(parsed?.pid);
|
||||
const lockAt = Number(parsed?.at);
|
||||
const staleByPid = Number.isInteger(lockPid) && lockPid > 0 && !isProcessAlive(lockPid);
|
||||
const staleByAge = Number.isFinite(lockAt) && (Date.now() - lockAt) > PROJECT_FILE_LOCK_STALE_MS;
|
||||
if (staleByPid || staleByAge || !Number.isInteger(lockPid)) {
|
||||
await fsPromises.unlink(lockPath).catch(() => {});
|
||||
continue;
|
||||
}
|
||||
} catch {
|
||||
// Crash between open(wx) and writeFile (or a partial write) leaves an
|
||||
// unparseable lock. Fall back to mtime age so recovery is not wedged.
|
||||
try {
|
||||
const stat = await fsPromises.stat(lockPath);
|
||||
const mtimeMs = Number(stat?.mtimeMs);
|
||||
if (Number.isFinite(mtimeMs) && (Date.now() - mtimeMs) > PROJECT_FILE_LOCK_STALE_MS) {
|
||||
await fsPromises.unlink(lockPath).catch(() => {});
|
||||
continue;
|
||||
}
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
await new Promise((resolve) => {
|
||||
setTimeout(resolve, PROJECT_FILE_LOCK_RETRY_MS);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error(`timeout acquiring project config lock for ${projectID}`);
|
||||
};
|
||||
|
||||
const readRawProjectConfigFromDisk = async (projectID) => {
|
||||
const filePath = resolveProjectConfigPath(projectID);
|
||||
try {
|
||||
@@ -443,8 +551,16 @@ export const createProjectConfigRuntime = (deps) => {
|
||||
writeLocks.set(key, chained);
|
||||
|
||||
await previous;
|
||||
// Acquire sits outside the mutate try — if it throws (10s lock timeout),
|
||||
// we must still release the in-process chain or every later write for this
|
||||
// project hangs forever on await previous.
|
||||
try {
|
||||
return await mutate();
|
||||
const fileLock = await acquireProjectFileLock(projectID);
|
||||
try {
|
||||
return await mutate();
|
||||
} finally {
|
||||
await fileLock.release();
|
||||
}
|
||||
} finally {
|
||||
release();
|
||||
const current = writeLocks.get(key);
|
||||
@@ -533,7 +649,7 @@ export const createProjectConfigRuntime = (deps) => {
|
||||
const current = await readProjectConfigFromDisk(projectID);
|
||||
const taskIndex = current.scheduledTasks.findIndex((task) => task.id === normalizedTaskID);
|
||||
if (taskIndex === -1) {
|
||||
return { task: null, tasks: current.scheduledTasks };
|
||||
return { task: null, tasks: current.scheduledTasks, updated: false };
|
||||
}
|
||||
|
||||
const currentTask = current.scheduledTasks[taskIndex];
|
||||
@@ -561,6 +677,68 @@ export const createProjectConfigRuntime = (deps) => {
|
||||
return {
|
||||
task: nextTask,
|
||||
tasks: nextTasks,
|
||||
updated: true,
|
||||
};
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* Conditionally patch task runtime state under the project write lock.
|
||||
* `predicate(currentTask)` is evaluated after the latest on-disk read; when
|
||||
* it returns false the write is skipped and `{ updated: false }` is returned.
|
||||
* Used by the scheduled-tasks runtime to claim a single schedule occurrence
|
||||
* across concurrent OpenChamber server instances.
|
||||
*/
|
||||
const updateScheduledTaskStateIf = async (projectID, taskID, predicate, statePatch) => {
|
||||
return withProjectWriteLock(projectID, async () => {
|
||||
const normalizedTaskID = asNonEmptyString(taskID);
|
||||
if (!normalizedTaskID) {
|
||||
throw new Error('taskId is required');
|
||||
}
|
||||
if (typeof predicate !== 'function') {
|
||||
throw new Error('predicate is required');
|
||||
}
|
||||
|
||||
const current = await readProjectConfigFromDisk(projectID);
|
||||
const taskIndex = current.scheduledTasks.findIndex((task) => task.id === normalizedTaskID);
|
||||
if (taskIndex === -1) {
|
||||
return { task: null, tasks: current.scheduledTasks, updated: false };
|
||||
}
|
||||
|
||||
const currentTask = current.scheduledTasks[taskIndex];
|
||||
if (!predicate(currentTask)) {
|
||||
return {
|
||||
task: currentTask,
|
||||
tasks: current.scheduledTasks,
|
||||
updated: false,
|
||||
};
|
||||
}
|
||||
|
||||
const patchObject = statePatch && typeof statePatch === 'object' ? statePatch : {};
|
||||
const nextTask = {
|
||||
...currentTask,
|
||||
state: normalizeState(
|
||||
{
|
||||
...currentTask.state,
|
||||
...patchObject,
|
||||
updatedAt: Date.now(),
|
||||
},
|
||||
currentTask.state,
|
||||
),
|
||||
};
|
||||
|
||||
const nextTasks = current.scheduledTasks.slice();
|
||||
nextTasks[taskIndex] = nextTask;
|
||||
|
||||
await writeProjectConfigToDisk(projectID, {
|
||||
version: PROJECT_CONFIG_VERSION,
|
||||
scheduledTasks: nextTasks,
|
||||
});
|
||||
|
||||
return {
|
||||
task: nextTask,
|
||||
tasks: nextTasks,
|
||||
updated: true,
|
||||
};
|
||||
});
|
||||
};
|
||||
@@ -699,6 +877,7 @@ export const createProjectConfigRuntime = (deps) => {
|
||||
upsertScheduledTask,
|
||||
deleteScheduledTask,
|
||||
updateScheduledTaskState,
|
||||
updateScheduledTaskStateIf,
|
||||
reconcileLoopTasks,
|
||||
resolveProjectConfigPath,
|
||||
};
|
||||
|
||||
@@ -471,4 +471,299 @@ describe('project-config loop reconciliation', () => {
|
||||
await cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it('conditionally updates state only when the predicate passes (occurrence claim)', async () => {
|
||||
const { runtime, cleanup } = await createRuntime();
|
||||
try {
|
||||
const created = await runtime.upsertScheduledTask('project-test', {
|
||||
name: 'claim-me',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '15:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run once', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
});
|
||||
|
||||
const scheduledFor = Date.UTC(2026, 0, 1, 15, 0, 0);
|
||||
const first = await runtime.updateScheduledTaskStateIf(
|
||||
'project-test',
|
||||
created.task.id,
|
||||
(task) => !Number.isFinite(task.state?.lastScheduledFor),
|
||||
{
|
||||
lastScheduledFor: scheduledFor,
|
||||
lastStatus: 'running',
|
||||
nextRunAt: scheduledFor + 86_400_000,
|
||||
},
|
||||
);
|
||||
expect(first.updated).toBe(true);
|
||||
expect(first.task.state.lastScheduledFor).toBe(scheduledFor);
|
||||
|
||||
const second = await runtime.updateScheduledTaskStateIf(
|
||||
'project-test',
|
||||
created.task.id,
|
||||
(task) => task.state?.lastScheduledFor !== scheduledFor,
|
||||
{
|
||||
lastScheduledFor: scheduledFor,
|
||||
lastStatus: 'running',
|
||||
},
|
||||
);
|
||||
expect(second.updated).toBe(false);
|
||||
expect(second.task.state.lastScheduledFor).toBe(scheduledFor);
|
||||
|
||||
const reloaded = await runtime.listScheduledTasks('project-test');
|
||||
expect(reloaded[0].state.lastScheduledFor).toBe(scheduledFor);
|
||||
} finally {
|
||||
await cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it('serializes concurrent writes across two runtimes sharing a projects dir', async () => {
|
||||
const tempRoot = await mkdtemp(path.join(os.tmpdir(), 'oc-project-lock-contention-'));
|
||||
const fsPromises = await import('fs/promises');
|
||||
try {
|
||||
const runtimeA = createProjectConfigRuntime({
|
||||
fsPromises,
|
||||
path,
|
||||
projectsDirPath: tempRoot,
|
||||
createTaskID: () => 'shared-task',
|
||||
});
|
||||
const runtimeB = createProjectConfigRuntime({
|
||||
fsPromises,
|
||||
path,
|
||||
projectsDirPath: tempRoot,
|
||||
createTaskID: () => 'shared-task',
|
||||
});
|
||||
|
||||
await runtimeA.upsertScheduledTask('project-lock', {
|
||||
name: 'contended',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '15:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
});
|
||||
|
||||
await Promise.all([
|
||||
runtimeA.updateScheduledTaskState('project-lock', 'shared-task', {
|
||||
lastStatus: 'success',
|
||||
lastRunAt: 100,
|
||||
}),
|
||||
runtimeB.updateScheduledTaskState('project-lock', 'shared-task', {
|
||||
lastStatus: 'error',
|
||||
lastRunAt: 200,
|
||||
}),
|
||||
]);
|
||||
|
||||
const tasks = await runtimeA.listScheduledTasks('project-lock');
|
||||
expect(tasks).toHaveLength(1);
|
||||
expect(['success', 'error']).toContain(tasks[0].state.lastStatus);
|
||||
expect([100, 200]).toContain(tasks[0].state.lastRunAt);
|
||||
|
||||
const lockPath = `${runtimeA.resolveProjectConfigPath('project-lock')}.lock`;
|
||||
await expect(fsPromises.access(lockPath)).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
} finally {
|
||||
await rm(tempRoot, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('recovers from a stale-by-age project config lock and cleans it up', async () => {
|
||||
const { runtime, cleanup } = await createRuntime();
|
||||
const fsPromises = await import('fs/promises');
|
||||
try {
|
||||
const projectID = 'stale-age';
|
||||
const configPath = runtime.resolveProjectConfigPath(projectID);
|
||||
const lockPath = `${configPath}.lock`;
|
||||
await fsPromises.mkdir(path.dirname(configPath), { recursive: true });
|
||||
await writeFile(lockPath, JSON.stringify({
|
||||
pid: process.pid,
|
||||
at: Date.now() - 120_000,
|
||||
}));
|
||||
|
||||
const created = await runtime.upsertScheduledTask(projectID, {
|
||||
name: 'after-stale',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '09:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
});
|
||||
expect(created.created).toBe(true);
|
||||
await expect(fsPromises.access(lockPath)).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
} finally {
|
||||
await cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it('recovers from a stale-by-dead-pid project config lock', async () => {
|
||||
const { runtime, cleanup } = await createRuntime();
|
||||
const fsPromises = await import('fs/promises');
|
||||
try {
|
||||
const projectID = 'stale-pid';
|
||||
const configPath = runtime.resolveProjectConfigPath(projectID);
|
||||
const lockPath = `${configPath}.lock`;
|
||||
await fsPromises.mkdir(path.dirname(configPath), { recursive: true });
|
||||
// PID unlikely to exist; kill(pid, 0) should fail with ESRCH.
|
||||
await writeFile(lockPath, JSON.stringify({
|
||||
pid: 2_147_483_647,
|
||||
at: Date.now(),
|
||||
}));
|
||||
|
||||
const created = await runtime.upsertScheduledTask(projectID, {
|
||||
name: 'after-dead-pid',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '09:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
});
|
||||
expect(created.created).toBe(true);
|
||||
await expect(fsPromises.access(lockPath)).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
} finally {
|
||||
await cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it('release does not unlink a lock stolen by another holder', async () => {
|
||||
const tempRoot = await mkdtemp(path.join(os.tmpdir(), 'oc-project-lock-own-'));
|
||||
const realFs = await import('fs/promises');
|
||||
let enteredCritical;
|
||||
const entered = new Promise((resolve) => {
|
||||
enteredCritical = resolve;
|
||||
});
|
||||
let resumeCritical;
|
||||
const hold = new Promise((resolve) => {
|
||||
resumeCritical = resolve;
|
||||
});
|
||||
|
||||
const fsPromises = {
|
||||
...realFs,
|
||||
rename: async (from, to) => {
|
||||
if (String(from).includes('.tmp-')) {
|
||||
enteredCritical();
|
||||
await hold;
|
||||
}
|
||||
return realFs.rename(from, to);
|
||||
},
|
||||
};
|
||||
|
||||
try {
|
||||
const runtime = createProjectConfigRuntime({
|
||||
fsPromises,
|
||||
path,
|
||||
projectsDirPath: tempRoot,
|
||||
createTaskID: () => 'owned-task',
|
||||
});
|
||||
const projectID = 'lock-own';
|
||||
const lockPath = `${runtime.resolveProjectConfigPath(projectID)}.lock`;
|
||||
const stolenPid = process.pid + 1;
|
||||
|
||||
const upsertPromise = runtime.upsertScheduledTask(projectID, {
|
||||
name: 'ownership',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '10:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
});
|
||||
|
||||
await entered;
|
||||
await realFs.writeFile(lockPath, JSON.stringify({
|
||||
pid: stolenPid,
|
||||
at: Date.now(),
|
||||
}));
|
||||
resumeCritical();
|
||||
await upsertPromise;
|
||||
|
||||
const raw = await realFs.readFile(lockPath, 'utf8');
|
||||
expect(JSON.parse(raw).pid).toBe(stolenPid);
|
||||
} finally {
|
||||
await rm(tempRoot, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('times out when a live lock holder never releases', async () => {
|
||||
const { runtime, cleanup } = await createRuntime();
|
||||
const fsPromises = await import('fs/promises');
|
||||
try {
|
||||
const projectID = 'lock-timeout';
|
||||
const configPath = runtime.resolveProjectConfigPath(projectID);
|
||||
const lockPath = `${configPath}.lock`;
|
||||
await fsPromises.mkdir(path.dirname(configPath), { recursive: true });
|
||||
// Current process is alive — acquire must wait then throw.
|
||||
await writeFile(lockPath, JSON.stringify({
|
||||
pid: process.pid,
|
||||
at: Date.now(),
|
||||
}));
|
||||
|
||||
await expect(runtime.upsertScheduledTask(projectID, {
|
||||
name: 'blocked',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '11:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
})).rejects.toThrow(/timeout acquiring project config lock/);
|
||||
} finally {
|
||||
await cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it('releases the write chain after a lock timeout so a later write can complete', async () => {
|
||||
const { runtime, cleanup } = await createRuntime();
|
||||
const fsPromises = await import('fs/promises');
|
||||
try {
|
||||
const projectID = 'lock-timeout-recover';
|
||||
const configPath = runtime.resolveProjectConfigPath(projectID);
|
||||
const lockPath = `${configPath}.lock`;
|
||||
await fsPromises.mkdir(path.dirname(configPath), { recursive: true });
|
||||
await writeFile(lockPath, JSON.stringify({
|
||||
pid: process.pid,
|
||||
at: Date.now(),
|
||||
}));
|
||||
|
||||
await expect(runtime.upsertScheduledTask(projectID, {
|
||||
name: 'blocked-then-recover',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '11:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
})).rejects.toThrow(/timeout acquiring project config lock/);
|
||||
|
||||
// Remove the hostile lock. A wedged in-process chain would hang forever here.
|
||||
await fsPromises.unlink(lockPath);
|
||||
|
||||
const created = await Promise.race([
|
||||
runtime.upsertScheduledTask(projectID, {
|
||||
name: 'after-timeout',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '12:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run after timeout', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
}),
|
||||
new Promise((_, reject) => {
|
||||
setTimeout(() => reject(new Error('second write hung after lock timeout')), 3_000);
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(created.created).toBe(true);
|
||||
expect(created.task.name).toBe('after-timeout');
|
||||
const listed = await runtime.listScheduledTasks(projectID);
|
||||
expect(listed).toHaveLength(1);
|
||||
expect(listed[0].name).toBe('after-timeout');
|
||||
} finally {
|
||||
await cleanup();
|
||||
}
|
||||
}, 20_000);
|
||||
|
||||
it('recovers from an unparseable lock using mtime age', async () => {
|
||||
const { runtime, cleanup } = await createRuntime();
|
||||
const fsPromises = await import('fs/promises');
|
||||
try {
|
||||
const projectID = 'stale-unparseable';
|
||||
const configPath = runtime.resolveProjectConfigPath(projectID);
|
||||
const lockPath = `${configPath}.lock`;
|
||||
await fsPromises.mkdir(path.dirname(configPath), { recursive: true });
|
||||
// Simulate crash between open(wx) and writeFile / partial payload.
|
||||
await writeFile(lockPath, '{not-json');
|
||||
const staleMtime = new Date(Date.now() - 120_000);
|
||||
await fsPromises.utimes(lockPath, staleMtime, staleMtime);
|
||||
|
||||
const created = await runtime.upsertScheduledTask(projectID, {
|
||||
name: 'after-unparseable',
|
||||
enabled: true,
|
||||
schedule: { kind: 'daily', time: '09:00', timezone: 'UTC' },
|
||||
execution: { prompt: 'Run', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||
});
|
||||
expect(created.created).toBe(true);
|
||||
await expect(fsPromises.access(lockPath)).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
} finally {
|
||||
await cleanup();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -9,6 +9,55 @@ Server-owned scheduled task runtime and routes for OpenChamber-only automation.
|
||||
- Runtime orchestration and execution is owned by `packages/web/server/lib/scheduled-tasks/runtime.js`.
|
||||
- This module is OpenChamber feature logic; it is intentionally separate from OpenCode proxy/runtime internals.
|
||||
|
||||
## Cross-instance occurrence claiming
|
||||
|
||||
Multiple OpenChamber server processes can share the same on-disk project config
|
||||
(for example CLI `serve` on port 3000 and the Electron desktop server on port
|
||||
57123). Each process keeps its own timers, so without coordination a daily (or
|
||||
weekly / cron / once) slot would dispatch twice.
|
||||
|
||||
Before a **scheduled** run creates a session, the runtime claims the occurrence
|
||||
in shared project config under the project write lock:
|
||||
|
||||
- Writes `state.lastScheduledFor` to the armed `nextRunAt` timestamp and advances
|
||||
`state.nextRunAt` to the following occurrence.
|
||||
- A second instance that loses the claim skips session creation and reschedules
|
||||
from the winner's persisted `nextRunAt`.
|
||||
- Project config writes also take a cross-process `.json.lock` file so the
|
||||
read-modify-write is serialized across processes, not only within one process.
|
||||
- Lock timeout / filesystem errors on claim, manual-start, or completion state
|
||||
writes always release the in-process running slot (via `finally`) and best-effort
|
||||
re-arm the **next future** occurrence; they must not leave the task permanently
|
||||
"running" or reject unhandled from the queue pump.
|
||||
- Project write locks release the in-process promise chain even when
|
||||
`acquireProjectFileLock` times out, so a later write for the same project can
|
||||
proceed after the on-disk lock is cleared (a hung chain would permanently wedge
|
||||
every mutating API and strand `runTask` before its `finally`).
|
||||
- Re-arm helpers only schedule a persisted `nextRunAt` when it is still in the
|
||||
future. A past slot (common for `once` after claim, which cannot advance
|
||||
`nextRunAt`) falls back to `computeNextRunAt` — which returns null for a
|
||||
consumed/past once occurrence — so a losing instance stops instead of
|
||||
spinning delay-0 timers against the project lock.
|
||||
- On claim lock/fs failure, best-effort persist `lastStatus: error` + `lastError`
|
||||
when nobody else claimed the occurrence, so a past `once` task is not left
|
||||
enabled-but-inert with only a warn log. Recurring schedules still re-arm the
|
||||
next slot.
|
||||
- On completion-write failure after a session already ran, in-memory status is
|
||||
set to a terminal value and a single persist retry is attempted so
|
||||
`lastStatus` does not stay `running`. Manual `runNow` still returns the
|
||||
`sessionID` as a successful dispatch (`ok` follows run status, with
|
||||
`persistError` set) rather than a hard 500; the run API and Scheduled Tasks
|
||||
UI surface `persistError` as a warning toast.
|
||||
- The claim predicate rejects a duplicate solely via `lastScheduledFor` within
|
||||
slack of this occurrence. It does not consult advanced on-disk `nextRunAt`
|
||||
(that field is routinely overwritten by a second instance syncing inside
|
||||
`TASK_DUE_SLACK_MS`, including on later days when `lastScheduledFor` is already
|
||||
set from a prior claim).
|
||||
- Claiming always writes `nextRunAt` (including `undefined`) so a past once-slot
|
||||
is cleared when there is no following occurrence.
|
||||
|
||||
Manual `runNow` does not claim a schedule occurrence.
|
||||
|
||||
## Files
|
||||
|
||||
- `packages/web/server/lib/scheduled-tasks/runtime.js`
|
||||
|
||||
@@ -0,0 +1,488 @@
|
||||
import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest';
|
||||
|
||||
/**
|
||||
* Regression for https://github.com/openchamber/openchamber/issues/2710
|
||||
* "Scheduled daily task executes twice at the configured time"
|
||||
*
|
||||
* Root cause: each OpenChamber server process keeps its own timers. Two
|
||||
* instances that share the same on-disk project config (CLI serve on port 3000
|
||||
* + Electron on 57123, or a startup login service + desktop) each arm a timer
|
||||
* for the same occurrence and both dispatch.
|
||||
*
|
||||
* Fix: scheduled runs claim the occurrence in shared project config
|
||||
* (`lastScheduledFor` + advanced `nextRunAt`) under a cross-process write lock
|
||||
* before creating a session, so the second instance skips.
|
||||
*/
|
||||
|
||||
const sdk = vi.hoisted(() => ({
|
||||
sessionCreates: [],
|
||||
createOpencodeClient: () => ({
|
||||
session: {
|
||||
create: async () => {
|
||||
sdk.sessionCreates.push(Date.now());
|
||||
return { data: { id: `sess-${sdk.sessionCreates.length}` } };
|
||||
},
|
||||
},
|
||||
command: { list: async () => ({ data: [] }) },
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock('@opencode-ai/sdk/v2', () => ({
|
||||
createOpencodeClient: sdk.createOpencodeClient,
|
||||
}));
|
||||
|
||||
import { createScheduledTasksRuntime } from './runtime.js';
|
||||
|
||||
const UTC = (y, mo, d, h, mi, s = 0) => Date.UTC(y, mo, d, h, mi, s);
|
||||
const MINUTE = 60_000;
|
||||
const HOUR = 3_600_000;
|
||||
|
||||
const makeTask = (schedule) => ({
|
||||
id: 'task-1',
|
||||
name: 'Daily Sync',
|
||||
enabled: true,
|
||||
schedule: { timezone: 'UTC', ...schedule },
|
||||
execution: { prompt: 'Summarize open issues', providerID: 'openai', modelID: 'gpt-4o' },
|
||||
state: { createdAt: UTC(2026, 0, 1, 0, 0, 0), updatedAt: UTC(2026, 0, 1, 0, 0, 0) },
|
||||
});
|
||||
|
||||
/**
|
||||
* Shared on-disk store stand-in. Both runtimes must see the same task state so
|
||||
* occurrence claiming can serialize dispatches the way real project config does.
|
||||
*/
|
||||
const createSharedProjectConfigRuntime = (initialTask) => {
|
||||
let currentTask = structuredClone(initialTask);
|
||||
|
||||
const applyPatch = (patch) => {
|
||||
const nextState = {
|
||||
...(currentTask.state || {}),
|
||||
...patch,
|
||||
updatedAt: Date.now(),
|
||||
};
|
||||
// Mirror normalizeState: explicit undefined clears optional numeric fields.
|
||||
for (const key of ['nextRunAt', 'lastRunAt', 'lastDurationMs', 'lastScheduledFor', 'lastError', 'lastSessionId']) {
|
||||
if (Object.prototype.hasOwnProperty.call(patch, key) && patch[key] === undefined) {
|
||||
delete nextState[key];
|
||||
}
|
||||
}
|
||||
currentTask = {
|
||||
...currentTask,
|
||||
state: nextState,
|
||||
};
|
||||
return currentTask;
|
||||
};
|
||||
|
||||
return {
|
||||
listScheduledTasks: vi.fn(async () => [structuredClone(currentTask)]),
|
||||
reconcileLoopTasks: vi.fn(async () => [structuredClone(currentTask)]),
|
||||
updateScheduledTaskState: vi.fn(async (_pid, _tid, patch) => {
|
||||
const task = applyPatch(patch);
|
||||
return { task: structuredClone(task), updated: true };
|
||||
}),
|
||||
updateScheduledTaskStateIf: vi.fn(async (_pid, _tid, predicate, patch) => {
|
||||
if (!predicate(currentTask)) {
|
||||
return { task: structuredClone(currentTask), updated: false };
|
||||
}
|
||||
const task = applyPatch(patch);
|
||||
return { task: structuredClone(task), updated: true };
|
||||
}),
|
||||
upsertScheduledTask: vi.fn(async (_pid, input) => {
|
||||
currentTask = structuredClone(input);
|
||||
return { task: structuredClone(currentTask) };
|
||||
}),
|
||||
};
|
||||
};
|
||||
|
||||
const createRuntimeDeps = (projectConfigRuntime) => ({
|
||||
projectConfigRuntime,
|
||||
listProjects: vi.fn(async () => [{ id: 'p1', path: '/repo' }]),
|
||||
buildOpenCodeUrl: () => 'http://127.0.0.1:9999/',
|
||||
getOpenCodeAuthHeaders: () => ({}),
|
||||
waitForOpenCodeReady: async () => {},
|
||||
emitTaskRunEvent: vi.fn(),
|
||||
setSessionAutoAccept: async () => {},
|
||||
logger: { info: () => {}, warn: () => {}, error: () => {} },
|
||||
});
|
||||
|
||||
const startInstances = async (count, task) => {
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(task);
|
||||
const runtimes = [];
|
||||
for (let i = 0; i < count; i += 1) {
|
||||
const runtime = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtime.start();
|
||||
runtimes.push(runtime);
|
||||
}
|
||||
return { runtimes, projectConfigRuntime };
|
||||
};
|
||||
|
||||
describe('issue 2710: daily scheduled task double execution at the configured time', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
sdk.sessionCreates.length = 0;
|
||||
globalThis.fetch = vi.fn(async () => ({ ok: true, text: async () => '' }));
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('ONE instance fires a daily 15:00 task exactly once at 15:00', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const { runtimes } = await startInstances(1, makeTask({ kind: 'daily', times: ['15:00'] }));
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
const firedAt = new Date(sdk.sessionCreates[0]);
|
||||
expect(firedAt.getUTCHours()).toBe(15);
|
||||
|
||||
runtimes.forEach((runtime) => runtime.stop());
|
||||
});
|
||||
|
||||
it('ONE instance firing daily 15:00 across 4 days never double-fires', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const { runtimes } = await startInstances(1, makeTask({ kind: 'daily', times: ['15:00'] }));
|
||||
await vi.advanceTimersByTimeAsync((4 * 24 * HOUR) + 3_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(4);
|
||||
const byHourBucket = new Map();
|
||||
for (const timestamp of sdk.sessionCreates) {
|
||||
const date = new Date(timestamp);
|
||||
expect(date.getUTCHours()).toBe(15);
|
||||
expect(date.getUTCMinutes()).toBe(0);
|
||||
const bucket = Math.floor(timestamp / HOUR);
|
||||
byHourBucket.set(bucket, (byHourBucket.get(bucket) || 0) + 1);
|
||||
}
|
||||
for (const count of byHourBucket.values()) {
|
||||
expect(count).toBe(1);
|
||||
}
|
||||
|
||||
runtimes.forEach((runtime) => runtime.stop());
|
||||
});
|
||||
|
||||
it('TWO instances fire a daily 15:00 task exactly ONCE (occurrence claim)', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const { runtimes, projectConfigRuntime } = await startInstances(
|
||||
2,
|
||||
makeTask({ kind: 'daily', times: ['15:00'] }),
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
const firedAt = new Date(sdk.sessionCreates[0]);
|
||||
expect(firedAt.getUTCHours()).toBe(15);
|
||||
expect(firedAt.getUTCMinutes()).toBe(0);
|
||||
expect(projectConfigRuntime.updateScheduledTaskStateIf).toHaveBeenCalled();
|
||||
|
||||
runtimes.forEach((runtime) => runtime.stop());
|
||||
});
|
||||
|
||||
it('TWO instances fire a weekly task exactly once', async () => {
|
||||
// 2026-01-04 is a Sunday. Weekly Mon/Wed/Fri 09:00.
|
||||
vi.setSystemTime(UTC(2026, 0, 4, 8, 0, 0));
|
||||
const { runtimes } = await startInstances(2, makeTask({
|
||||
kind: 'weekly',
|
||||
times: ['09:00'],
|
||||
weekdays: [1, 3, 5],
|
||||
}));
|
||||
await vi.advanceTimersByTimeAsync((25 * HOUR) + 3_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
|
||||
runtimes.forEach((runtime) => runtime.stop());
|
||||
});
|
||||
|
||||
it('TWO instances fire a cron task exactly once', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 3, 0));
|
||||
const { runtimes } = await startInstances(2, makeTask({ kind: 'cron', cron: '*/5 * * * *' }));
|
||||
await vi.advanceTimersByTimeAsync(3 * MINUTE);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
|
||||
runtimes.forEach((runtime) => runtime.stop());
|
||||
});
|
||||
|
||||
it('TWO instances fire a once task exactly once', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 8, 0, 0));
|
||||
const { runtimes } = await startInstances(2, makeTask({
|
||||
kind: 'once',
|
||||
date: '2026-01-01',
|
||||
time: '09:00',
|
||||
}));
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
|
||||
runtimes.forEach((runtime) => runtime.stop());
|
||||
});
|
||||
|
||||
it('claim failure releases the running slot and does not reject unhandled', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(
|
||||
makeTask({ kind: 'daily', times: ['15:00'] }),
|
||||
);
|
||||
projectConfigRuntime.updateScheduledTaskStateIf = vi.fn(async () => {
|
||||
throw new Error('timeout acquiring project config lock for p1');
|
||||
});
|
||||
const runtime = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtime.start();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(0);
|
||||
expect(runtime.getStatus().runningScheduledTasksCount).toBe(0);
|
||||
expect(runtime.getStatus().hasRunningScheduledTasks).toBe(false);
|
||||
|
||||
// Manual runNow must not be stuck behind a permanently "running" claim failure.
|
||||
const manual = await runtime.runNow('p1', 'task-1');
|
||||
expect(manual.ok).toBe(true);
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
|
||||
runtime.stop();
|
||||
});
|
||||
|
||||
it('completion state write failure releases the running slot', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(
|
||||
makeTask({ kind: 'daily', times: ['15:00'] }),
|
||||
);
|
||||
const originalUpdate = projectConfigRuntime.updateScheduledTaskState;
|
||||
let completionWrites = 0;
|
||||
projectConfigRuntime.updateScheduledTaskState = vi.fn(async (pid, tid, patch) => {
|
||||
// syncTaskSchedule + claim path may write; fail the post-run completion write.
|
||||
if (patch?.lastStatus === 'success' || patch?.lastStatus === 'error') {
|
||||
completionWrites += 1;
|
||||
throw new Error('timeout acquiring project config lock for p1');
|
||||
}
|
||||
return originalUpdate(pid, tid, patch);
|
||||
});
|
||||
|
||||
const runtime = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtime.start();
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
|
||||
// Initial completion write + best-effort retry.
|
||||
expect(completionWrites).toBeGreaterThanOrEqual(2);
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
expect(runtime.getStatus().runningScheduledTasksCount).toBe(0);
|
||||
|
||||
const manual = await runtime.runNow('p1', 'task-1');
|
||||
expect(manual.ok).toBe(true);
|
||||
expect(manual.sessionID).toBeTruthy();
|
||||
expect(runtime.getStatus().runningScheduledTasksCount).toBe(0);
|
||||
|
||||
runtime.stop();
|
||||
});
|
||||
|
||||
it('manual run completion write failure returns session and clears running status', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(
|
||||
makeTask({ kind: 'daily', times: ['15:00'] }),
|
||||
);
|
||||
const runtime = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtime.start();
|
||||
|
||||
const originalUpdate = projectConfigRuntime.updateScheduledTaskState;
|
||||
projectConfigRuntime.updateScheduledTaskState = vi.fn(async (pid, tid, patch) => {
|
||||
if (patch?.lastStatus === 'success' || patch?.lastStatus === 'error') {
|
||||
throw new Error('timeout acquiring project config lock for p1');
|
||||
}
|
||||
return originalUpdate(pid, tid, patch);
|
||||
});
|
||||
|
||||
const manual = await runtime.runNow('p1', 'task-1');
|
||||
expect(manual.ok).toBe(true);
|
||||
expect(manual.sessionID).toBeTruthy();
|
||||
expect(manual.reason).toBe('completion-state-failed');
|
||||
expect(manual.persistError).toMatch(/timeout acquiring project config lock/);
|
||||
expect(manual.task?.state?.lastStatus).toBe('success');
|
||||
expect(runtime.getStatus().runningScheduledTasksCount).toBe(0);
|
||||
|
||||
runtime.stop();
|
||||
});
|
||||
|
||||
it('manual start state write failure releases the running slot', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(
|
||||
makeTask({ kind: 'daily', times: ['15:00'] }),
|
||||
);
|
||||
const runtime = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtime.start();
|
||||
|
||||
projectConfigRuntime.updateScheduledTaskState = vi.fn(async () => {
|
||||
throw new Error('timeout acquiring project config lock for p1');
|
||||
});
|
||||
|
||||
const manual = await runtime.runNow('p1', 'task-1');
|
||||
expect(manual.ok).toBe(false);
|
||||
expect(manual.reason).toBe('start-state-failed');
|
||||
expect(sdk.sessionCreates.length).toBe(0);
|
||||
expect(runtime.getStatus().runningScheduledTasksCount).toBe(0);
|
||||
|
||||
runtime.stop();
|
||||
});
|
||||
|
||||
it('armed instance still fires when another instance syncs inside the due-slack window', async () => {
|
||||
// Instance A arms for 15:00 outside the slack window. Instance B then starts
|
||||
// inside TASK_DUE_SLACK_MS and syncTaskSchedule advances disk nextRunAt to
|
||||
// tomorrow. A must still claim today's occurrence.
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const task = makeTask({ kind: 'daily', times: ['15:00'] });
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(task);
|
||||
|
||||
const runtimeA = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtimeA.start();
|
||||
|
||||
// Enter the due-slack window (T-5s .. T) and sync a second instance.
|
||||
await vi.advanceTimersByTimeAsync(HOUR - 2_000);
|
||||
const runtimeB = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtimeB.start();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
|
||||
runtimeA.stop();
|
||||
runtimeB.stop();
|
||||
});
|
||||
|
||||
it('armed instance still fires on a later day when lastScheduledFor is already set', async () => {
|
||||
// After day 1 has claimed, lastScheduledFor is finite. On day 2 a second
|
||||
// instance syncing inside the due-slack window must not suppress the armed fire.
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const task = makeTask({ kind: 'daily', times: ['15:00'] });
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(task);
|
||||
|
||||
const runtimeA = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtimeA.start();
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
|
||||
// Advance to day-2 afternoon, still outside slack, so A re-arms for 15:00.
|
||||
await vi.advanceTimersByTimeAsync((23 * HOUR) - 3_000);
|
||||
// Enter day-2 due-slack window and sync instance B (advances nextRunAt).
|
||||
await vi.advanceTimersByTimeAsync(HOUR - 2_000);
|
||||
const runtimeB = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtimeB.start();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(2);
|
||||
|
||||
runtimeA.stop();
|
||||
runtimeB.stop();
|
||||
});
|
||||
|
||||
it('once-task loser does not spin-rearm a past nextRunAt while the winner runs', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 8, 0, 0));
|
||||
|
||||
let releaseFetch;
|
||||
const fetchGate = new Promise((resolve) => {
|
||||
releaseFetch = resolve;
|
||||
});
|
||||
globalThis.fetch = vi.fn(async () => {
|
||||
await fetchGate;
|
||||
return { ok: true, text: async () => '' };
|
||||
});
|
||||
|
||||
const { runtimes, projectConfigRuntime } = await startInstances(2, makeTask({
|
||||
kind: 'once',
|
||||
date: '2026-01-01',
|
||||
time: '09:00',
|
||||
}));
|
||||
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
|
||||
// Winner claimed and is blocked in prompt_async; exactly one session.
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
const claimsAfterFire = projectConfigRuntime.updateScheduledTaskStateIf.mock.calls.length;
|
||||
expect(claimsAfterFire).toBeGreaterThanOrEqual(1);
|
||||
|
||||
// Advance through many jitter windows. Loser must not keep re-entering claim.
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(projectConfigRuntime.updateScheduledTaskStateIf.mock.calls.length).toBe(claimsAfterFire);
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
|
||||
releaseFetch();
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(1);
|
||||
expect(runtimes.every((runtime) => runtime.getStatus().runningScheduledTasksCount === 0)).toBe(true);
|
||||
|
||||
runtimes.forEach((runtime) => runtime.stop());
|
||||
});
|
||||
|
||||
it('claim-failed re-arms the next occurrence, not an immediate retry of the past slot', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 14, 0, 0));
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(
|
||||
makeTask({ kind: 'daily', times: ['15:00'] }),
|
||||
);
|
||||
|
||||
let claimAttempts = 0;
|
||||
const originalClaim = projectConfigRuntime.updateScheduledTaskStateIf;
|
||||
projectConfigRuntime.updateScheduledTaskStateIf = vi.fn(async (pid, tid, predicate, patch) => {
|
||||
// Claim patches set lastScheduledFor; failure-recording patches set lastStatus error.
|
||||
if (Object.prototype.hasOwnProperty.call(patch || {}, 'lastScheduledFor')) {
|
||||
claimAttempts += 1;
|
||||
throw new Error('timeout acquiring project config lock for p1');
|
||||
}
|
||||
return originalClaim(pid, tid, predicate, patch);
|
||||
});
|
||||
|
||||
const runtime = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtime.start();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
expect(claimAttempts).toBe(1);
|
||||
|
||||
// Must not immediately retry the same past occurrence on a ~jitter cadence.
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(claimAttempts).toBe(1);
|
||||
|
||||
// Next calendar occurrence (tomorrow 15:00) may attempt once more.
|
||||
await vi.advanceTimersByTimeAsync(24 * HOUR);
|
||||
expect(claimAttempts).toBe(2);
|
||||
|
||||
runtime.stop();
|
||||
});
|
||||
|
||||
it('once claim failure records an error status instead of leaving the task silently inert', async () => {
|
||||
vi.setSystemTime(UTC(2026, 0, 1, 8, 0, 0));
|
||||
const projectConfigRuntime = createSharedProjectConfigRuntime(makeTask({
|
||||
kind: 'once',
|
||||
date: '2026-01-01',
|
||||
time: '09:00',
|
||||
}));
|
||||
|
||||
const originalClaim = projectConfigRuntime.updateScheduledTaskStateIf;
|
||||
projectConfigRuntime.updateScheduledTaskStateIf = vi.fn(async (pid, tid, predicate, patch) => {
|
||||
if (Object.prototype.hasOwnProperty.call(patch || {}, 'lastScheduledFor')) {
|
||||
throw new Error('timeout acquiring project config lock for p1');
|
||||
}
|
||||
return originalClaim(pid, tid, predicate, patch);
|
||||
});
|
||||
|
||||
const runtime = createScheduledTasksRuntime(createRuntimeDeps(projectConfigRuntime));
|
||||
await runtime.start();
|
||||
await vi.advanceTimersByTimeAsync(HOUR + 3_000);
|
||||
|
||||
expect(sdk.sessionCreates.length).toBe(0);
|
||||
const tasks = await projectConfigRuntime.listScheduledTasks('p1');
|
||||
expect(tasks[0].state.lastStatus).toBe('error');
|
||||
expect(tasks[0].state.lastError).toMatch(/Scheduled claim failed/);
|
||||
expect(tasks[0].enabled).toBe(true);
|
||||
|
||||
// No silent delay-0 spin after the failed once claim.
|
||||
const errorWrites = projectConfigRuntime.updateScheduledTaskStateIf.mock.calls
|
||||
.filter(([, , , patch]) => patch?.lastStatus === 'error')
|
||||
.length;
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
const errorWritesAfter = projectConfigRuntime.updateScheduledTaskStateIf.mock.calls
|
||||
.filter(([, , , patch]) => patch?.lastStatus === 'error')
|
||||
.length;
|
||||
expect(errorWritesAfter).toBe(errorWrites);
|
||||
|
||||
runtime.stop();
|
||||
});
|
||||
});
|
||||
@@ -327,7 +327,7 @@ export const createScheduledTasksRuntime = (deps) => {
|
||||
if (!task || !task.enabled) {
|
||||
return;
|
||||
}
|
||||
queueTaskRun(projectID, taskID, 'scheduled');
|
||||
queueTaskRun(projectID, taskID, 'scheduled', nextRunAt);
|
||||
pumpQueue();
|
||||
}, boundedDelay);
|
||||
|
||||
@@ -429,13 +429,18 @@ export const createScheduledTasksRuntime = (deps) => {
|
||||
}
|
||||
};
|
||||
|
||||
const queueTaskRun = (projectID, taskID, reason) => {
|
||||
const queueTaskRun = (projectID, taskID, reason, scheduledFor) => {
|
||||
const taskKey = buildTaskKey(projectID, taskID);
|
||||
if (queuedTaskKeys.has(taskKey) || runningTaskKeys.has(taskKey)) {
|
||||
return;
|
||||
}
|
||||
queuedTaskKeys.add(taskKey);
|
||||
queue.push({ projectID, taskID, reason });
|
||||
queue.push({
|
||||
projectID,
|
||||
taskID,
|
||||
reason,
|
||||
...(Number.isFinite(scheduledFor) ? { scheduledFor } : {}),
|
||||
});
|
||||
};
|
||||
|
||||
const canRunTask = (projectID) => {
|
||||
@@ -605,7 +610,51 @@ export const createScheduledTasksRuntime = (deps) => {
|
||||
};
|
||||
};
|
||||
|
||||
const runTask = async (projectID, taskID, reason) => {
|
||||
const releaseRunningSlot = (projectID, taskKey) => {
|
||||
runningTaskKeys.delete(taskKey);
|
||||
runningGlobalCount = Math.max(0, runningGlobalCount - 1);
|
||||
const nextProjectCount = Math.max(0, (runningCountByProject.get(projectID) || 1) - 1);
|
||||
if (nextProjectCount === 0) {
|
||||
runningCountByProject.delete(projectID);
|
||||
} else {
|
||||
runningCountByProject.set(projectID, nextProjectCount);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Arm a timer only for a future occurrence. Scheduling a past nextRunAt
|
||||
* (delay 0 + jitter) re-enters the claim path immediately and can spin —
|
||||
* especially for once tasks where the claim cannot advance nextRunAt.
|
||||
*/
|
||||
const scheduleFutureRun = (projectID, taskID, nextRunAt, fromMs = Date.now()) => {
|
||||
if (!Number.isFinite(nextRunAt)) {
|
||||
return false;
|
||||
}
|
||||
const base = Number.isFinite(fromMs) ? fromMs : Date.now();
|
||||
if (nextRunAt <= base) {
|
||||
return false;
|
||||
}
|
||||
scheduleTask(projectID, taskID, nextRunAt);
|
||||
return true;
|
||||
};
|
||||
|
||||
const rearmFromTaskOrCompute = (projectID, taskID, fallbackTask, fromMs) => {
|
||||
const latest = (tasksByProject.get(projectID)?.get(taskID)) || fallbackTask;
|
||||
if (!latest?.enabled) {
|
||||
return;
|
||||
}
|
||||
const base = Number.isFinite(fromMs) ? fromMs : Date.now();
|
||||
const persistedNext = latest.state?.nextRunAt;
|
||||
// Prefer a still-future persisted slot; never re-arm a past occurrence
|
||||
// (that created silent once-task loser loops and claim-failed retry spam).
|
||||
if (scheduleFutureRun(projectID, taskID, persistedNext, base)) {
|
||||
return;
|
||||
}
|
||||
const computedNext = computeNextRunAt(latest, base);
|
||||
scheduleFutureRun(projectID, taskID, computedNext, base);
|
||||
};
|
||||
|
||||
const runTask = async (projectID, taskID, reason, scheduledFor) => {
|
||||
const taskMap = tasksByProject.get(projectID);
|
||||
const task = taskMap?.get(taskID);
|
||||
if (!task || !task.enabled) {
|
||||
@@ -621,125 +670,328 @@ export const createScheduledTasksRuntime = (deps) => {
|
||||
runningGlobalCount += 1;
|
||||
runningCountByProject.set(projectID, (runningCountByProject.get(projectID) || 0) + 1);
|
||||
|
||||
const runStartedAt = Date.now();
|
||||
await projectConfigRuntime.updateScheduledTaskState(projectID, taskID, {
|
||||
lastRunAt: runStartedAt,
|
||||
lastStatus: 'running',
|
||||
lastError: undefined,
|
||||
updatedAt: runStartedAt,
|
||||
}).then((result) => {
|
||||
if (result.task) {
|
||||
updateInMemoryTask(projectID, result.task);
|
||||
}
|
||||
});
|
||||
|
||||
let status = 'success';
|
||||
let sessionID;
|
||||
let durationMs = 0;
|
||||
let errorMessage;
|
||||
|
||||
// Every path that holds the running slot must exit through this finally so
|
||||
// lock timeouts / fs errors on claim, manual-start, or completion writes
|
||||
// cannot permanently stuck-run the task in this process.
|
||||
try {
|
||||
const runPromise = runTaskWithWatchdog(projectID, task, reason);
|
||||
let timeoutID;
|
||||
const timeoutPromise = new Promise((_, reject) => {
|
||||
timeoutID = setTimeout(() => {
|
||||
reject(new Error('scheduled task run timed out'));
|
||||
}, maxRunDurationMs);
|
||||
});
|
||||
const runStartedAt = Date.now();
|
||||
|
||||
const result = await Promise.race([runPromise, timeoutPromise]).finally(() => {
|
||||
if (timeoutID) {
|
||||
clearTimeout(timeoutID);
|
||||
// Scheduled dispatches must claim the occurrence in shared project config
|
||||
// before creating a session. Two server instances (e.g. CLI serve + desktop)
|
||||
// each arm their own timer; without this claim both would run (#2710).
|
||||
if (reason === 'scheduled') {
|
||||
if (!Number.isFinite(scheduledFor)) {
|
||||
return { ok: false, skipped: true, reason: 'missing-scheduled-for' };
|
||||
}
|
||||
});
|
||||
sessionID = result.sessionID;
|
||||
durationMs = result.durationMs;
|
||||
status = 'success';
|
||||
logger.info?.(
|
||||
'[ScheduledTasks] run completed',
|
||||
{ projectID, taskID, status, reason, sessionID, durationMs }
|
||||
);
|
||||
} catch (error) {
|
||||
status = 'error';
|
||||
errorMessage = safeErrorMessage(error);
|
||||
logger.warn?.('[ScheduledTasks] run failed', {
|
||||
projectID,
|
||||
taskID,
|
||||
reason,
|
||||
status,
|
||||
error: errorMessage,
|
||||
});
|
||||
}
|
||||
|
||||
const finishedAt = Date.now();
|
||||
if (!durationMs) {
|
||||
durationMs = Math.max(0, finishedAt - runStartedAt);
|
||||
}
|
||||
let latestTask = (tasksByProject.get(projectID)?.get(taskID)) || task;
|
||||
const shouldConsumeOneTimeTask = latestTask?.schedule?.kind === 'once' && reason === 'scheduled';
|
||||
if (shouldConsumeOneTimeTask && latestTask?.enabled) {
|
||||
const nextAfterClaim = computeNextRunAt(task, Math.max(runStartedAt, scheduledFor + 1));
|
||||
const claimPatch = {
|
||||
lastScheduledFor: Math.round(scheduledFor),
|
||||
lastRunAt: runStartedAt,
|
||||
lastStatus: 'running',
|
||||
lastError: undefined,
|
||||
updatedAt: runStartedAt,
|
||||
// Always set nextRunAt so a past once-slot is cleared when there is
|
||||
// no following occurrence (omitting the key would leave the past value).
|
||||
nextRunAt: Number.isFinite(nextAfterClaim) ? nextAfterClaim : undefined,
|
||||
};
|
||||
|
||||
// Duplicate protection is solely lastScheduledFor within slack of this
|
||||
// occurrence. Do not reject on advanced disk nextRunAt: lastScheduledFor
|
||||
// persists across days, so a second-instance sync inside TASK_DUE_SLACK_MS
|
||||
// would otherwise suppress every armed occurrence after the first.
|
||||
const canClaimOccurrence = (candidate) => {
|
||||
if (!candidate?.enabled) {
|
||||
return false;
|
||||
}
|
||||
const lastScheduledFor = candidate.state?.lastScheduledFor;
|
||||
if (
|
||||
Number.isFinite(lastScheduledFor)
|
||||
&& Math.abs(lastScheduledFor - scheduledFor) <= TASK_DUE_SLACK_MS
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
};
|
||||
|
||||
let claimResult;
|
||||
try {
|
||||
if (typeof projectConfigRuntime.updateScheduledTaskStateIf === 'function') {
|
||||
claimResult = await projectConfigRuntime.updateScheduledTaskStateIf(
|
||||
projectID,
|
||||
taskID,
|
||||
canClaimOccurrence,
|
||||
claimPatch,
|
||||
);
|
||||
} else {
|
||||
// Fallback for older test doubles: unconditional update (single-instance only).
|
||||
claimResult = await projectConfigRuntime.updateScheduledTaskState(projectID, taskID, claimPatch);
|
||||
claimResult = { ...claimResult, updated: Boolean(claimResult?.task) };
|
||||
}
|
||||
} catch (claimError) {
|
||||
const message = safeErrorMessage(claimError);
|
||||
logger.warn?.('[ScheduledTasks] occurrence claim failed', {
|
||||
projectID,
|
||||
taskID,
|
||||
error: message,
|
||||
});
|
||||
rearmFromTaskOrCompute(projectID, taskID, task, Math.max(runStartedAt, scheduledFor + 1));
|
||||
|
||||
// Best-effort record so once tasks are not left enabled-but-inert with
|
||||
// no UI signal. Do not clobber a winner that claimed this occurrence.
|
||||
const claimFailurePatch = {
|
||||
lastStatus: 'error',
|
||||
lastError: `Scheduled claim failed: ${message}`,
|
||||
updatedAt: Date.now(),
|
||||
};
|
||||
try {
|
||||
if (typeof projectConfigRuntime.updateScheduledTaskStateIf === 'function') {
|
||||
const recorded = await projectConfigRuntime.updateScheduledTaskStateIf(
|
||||
projectID,
|
||||
taskID,
|
||||
(candidate) => {
|
||||
const lastScheduledFor = candidate.state?.lastScheduledFor;
|
||||
if (
|
||||
Number.isFinite(lastScheduledFor)
|
||||
&& Math.abs(lastScheduledFor - scheduledFor) <= TASK_DUE_SLACK_MS
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
},
|
||||
claimFailurePatch,
|
||||
);
|
||||
if (recorded.task) {
|
||||
updateInMemoryTask(projectID, recorded.task);
|
||||
}
|
||||
} else {
|
||||
const recorded = await projectConfigRuntime.updateScheduledTaskState(
|
||||
projectID,
|
||||
taskID,
|
||||
claimFailurePatch,
|
||||
);
|
||||
if (recorded.task) {
|
||||
updateInMemoryTask(projectID, recorded.task);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
updateInMemoryTask(projectID, {
|
||||
...task,
|
||||
state: {
|
||||
...(task.state || {}),
|
||||
...claimFailurePatch,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return { ok: false, skipped: true, reason: 'claim-failed', error: message };
|
||||
}
|
||||
|
||||
if (!claimResult?.updated) {
|
||||
if (claimResult?.task) {
|
||||
updateInMemoryTask(projectID, claimResult.task);
|
||||
// Loser must not schedule a past nextRunAt (once-task spin).
|
||||
rearmFromTaskOrCompute(
|
||||
projectID,
|
||||
taskID,
|
||||
claimResult.task,
|
||||
Math.max(Date.now(), scheduledFor + 1),
|
||||
);
|
||||
}
|
||||
return { ok: false, skipped: true, reason: 'occurrence-claimed' };
|
||||
}
|
||||
|
||||
if (claimResult.task) {
|
||||
updateInMemoryTask(projectID, claimResult.task);
|
||||
}
|
||||
} else {
|
||||
try {
|
||||
const startResult = await projectConfigRuntime.updateScheduledTaskState(projectID, taskID, {
|
||||
lastRunAt: runStartedAt,
|
||||
lastStatus: 'running',
|
||||
lastError: undefined,
|
||||
updatedAt: runStartedAt,
|
||||
});
|
||||
if (startResult.task) {
|
||||
updateInMemoryTask(projectID, startResult.task);
|
||||
}
|
||||
} catch (startError) {
|
||||
const message = safeErrorMessage(startError);
|
||||
logger.warn?.('[ScheduledTasks] manual start state write failed', {
|
||||
projectID,
|
||||
taskID,
|
||||
error: message,
|
||||
});
|
||||
return { ok: false, error: message, reason: 'start-state-failed' };
|
||||
}
|
||||
}
|
||||
|
||||
let status = 'success';
|
||||
let sessionID;
|
||||
let durationMs = 0;
|
||||
let errorMessage;
|
||||
|
||||
try {
|
||||
const consumed = await projectConfigRuntime.upsertScheduledTask(projectID, {
|
||||
...latestTask,
|
||||
enabled: false,
|
||||
const runPromise = runTaskWithWatchdog(projectID, task, reason);
|
||||
let timeoutID;
|
||||
const timeoutPromise = new Promise((_, reject) => {
|
||||
timeoutID = setTimeout(() => {
|
||||
reject(new Error('scheduled task run timed out'));
|
||||
}, maxRunDurationMs);
|
||||
});
|
||||
latestTask = consumed.task || latestTask;
|
||||
updateInMemoryTask(projectID, latestTask);
|
||||
} catch (consumeError) {
|
||||
logger.warn?.('[ScheduledTasks] failed to consume one-time task', {
|
||||
|
||||
const result = await Promise.race([runPromise, timeoutPromise]).finally(() => {
|
||||
if (timeoutID) {
|
||||
clearTimeout(timeoutID);
|
||||
}
|
||||
});
|
||||
sessionID = result.sessionID;
|
||||
durationMs = result.durationMs;
|
||||
status = 'success';
|
||||
logger.info?.(
|
||||
'[ScheduledTasks] run completed',
|
||||
{ projectID, taskID, status, reason, sessionID, durationMs }
|
||||
);
|
||||
} catch (error) {
|
||||
status = 'error';
|
||||
errorMessage = safeErrorMessage(error);
|
||||
logger.warn?.('[ScheduledTasks] run failed', {
|
||||
projectID,
|
||||
taskID,
|
||||
error: safeErrorMessage(consumeError),
|
||||
reason,
|
||||
status,
|
||||
error: errorMessage,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const nextRunAt = computeNextRunAt(latestTask, finishedAt);
|
||||
|
||||
const statePatch = {
|
||||
lastStatus: status,
|
||||
lastDurationMs: durationMs,
|
||||
lastError: status === 'error' ? errorMessage : undefined,
|
||||
lastSessionId: status === 'success' ? sessionID : undefined,
|
||||
nextRunAt: Number.isFinite(nextRunAt) ? nextRunAt : undefined,
|
||||
updatedAt: finishedAt,
|
||||
};
|
||||
|
||||
const stateResult = await projectConfigRuntime.updateScheduledTaskState(projectID, taskID, statePatch);
|
||||
if (stateResult.task) {
|
||||
updateInMemoryTask(projectID, stateResult.task);
|
||||
if (stateResult.task.enabled && Number.isFinite(stateResult.task.state?.nextRunAt)) {
|
||||
scheduleTask(projectID, taskID, stateResult.task.state.nextRunAt);
|
||||
const finishedAt = Date.now();
|
||||
if (!durationMs) {
|
||||
durationMs = Math.max(0, finishedAt - runStartedAt);
|
||||
}
|
||||
let latestTask = (tasksByProject.get(projectID)?.get(taskID)) || task;
|
||||
const shouldConsumeOneTimeTask = latestTask?.schedule?.kind === 'once' && reason === 'scheduled';
|
||||
if (shouldConsumeOneTimeTask && latestTask?.enabled) {
|
||||
try {
|
||||
const consumed = await projectConfigRuntime.upsertScheduledTask(projectID, {
|
||||
...latestTask,
|
||||
enabled: false,
|
||||
});
|
||||
latestTask = consumed.task || latestTask;
|
||||
updateInMemoryTask(projectID, latestTask);
|
||||
} catch (consumeError) {
|
||||
logger.warn?.('[ScheduledTasks] failed to consume one-time task', {
|
||||
projectID,
|
||||
taskID,
|
||||
error: safeErrorMessage(consumeError),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
emitTaskRunEvent?.({
|
||||
projectID,
|
||||
taskID,
|
||||
ranAt: finishedAt,
|
||||
const nextRunAt = computeNextRunAt(latestTask, finishedAt);
|
||||
|
||||
const statePatch = {
|
||||
lastStatus: status,
|
||||
lastDurationMs: durationMs,
|
||||
lastError: status === 'error' ? errorMessage : undefined,
|
||||
lastSessionId: status === 'success' ? sessionID : undefined,
|
||||
nextRunAt: Number.isFinite(nextRunAt) ? nextRunAt : undefined,
|
||||
updatedAt: finishedAt,
|
||||
};
|
||||
|
||||
let stateResult = { task: null };
|
||||
try {
|
||||
stateResult = await projectConfigRuntime.updateScheduledTaskState(projectID, taskID, statePatch);
|
||||
if (stateResult.task) {
|
||||
updateInMemoryTask(projectID, stateResult.task);
|
||||
if (stateResult.task.enabled) {
|
||||
scheduleFutureRun(
|
||||
projectID,
|
||||
taskID,
|
||||
stateResult.task.state?.nextRunAt,
|
||||
finishedAt,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (persistError) {
|
||||
const message = safeErrorMessage(persistError);
|
||||
logger.warn?.('[ScheduledTasks] run completion state write failed', {
|
||||
projectID,
|
||||
taskID,
|
||||
reason,
|
||||
error: message,
|
||||
});
|
||||
|
||||
// Keep in-memory status terminal so this process does not advertise
|
||||
// a stuck "running" task after the session already finished.
|
||||
const recoveredTask = {
|
||||
...latestTask,
|
||||
state: {
|
||||
...(latestTask.state || {}),
|
||||
lastStatus: status,
|
||||
lastDurationMs: durationMs,
|
||||
lastError: status === 'error' ? errorMessage : undefined,
|
||||
lastSessionId: status === 'success' ? sessionID : undefined,
|
||||
nextRunAt: Number.isFinite(nextRunAt) ? nextRunAt : undefined,
|
||||
updatedAt: finishedAt,
|
||||
},
|
||||
};
|
||||
updateInMemoryTask(projectID, recoveredTask);
|
||||
|
||||
// Best-effort single retry so persisted lastStatus does not stay 'running'.
|
||||
try {
|
||||
const retry = await projectConfigRuntime.updateScheduledTaskState(projectID, taskID, statePatch);
|
||||
if (retry.task) {
|
||||
updateInMemoryTask(projectID, retry.task);
|
||||
stateResult = retry;
|
||||
if (retry.task.enabled) {
|
||||
scheduleFutureRun(projectID, taskID, retry.task.state?.nextRunAt, finishedAt);
|
||||
}
|
||||
}
|
||||
} catch (retryError) {
|
||||
logger.warn?.('[ScheduledTasks] run completion state retry failed', {
|
||||
projectID,
|
||||
taskID,
|
||||
reason,
|
||||
error: safeErrorMessage(retryError),
|
||||
});
|
||||
stateResult = { task: recoveredTask };
|
||||
rearmFromTaskOrCompute(projectID, taskID, recoveredTask, finishedAt);
|
||||
}
|
||||
|
||||
// The session already ran — surface persist failure without treating a
|
||||
// successful dispatch as a hard run failure (manual runNow would 500).
|
||||
return {
|
||||
ok: status === 'success',
|
||||
status,
|
||||
sessionID,
|
||||
task: stateResult.task || recoveredTask,
|
||||
error: status === 'error' ? errorMessage : undefined,
|
||||
persistError: message,
|
||||
reason: 'completion-state-failed',
|
||||
};
|
||||
}
|
||||
|
||||
try {
|
||||
emitTaskRunEvent?.({
|
||||
projectID,
|
||||
taskID,
|
||||
ranAt: finishedAt,
|
||||
status,
|
||||
...(sessionID ? { sessionID } : {}),
|
||||
});
|
||||
} catch {
|
||||
}
|
||||
|
||||
return {
|
||||
ok: status === 'success',
|
||||
status,
|
||||
...(sessionID ? { sessionID } : {}),
|
||||
});
|
||||
} catch {
|
||||
sessionID,
|
||||
task: stateResult.task || null,
|
||||
error: errorMessage,
|
||||
};
|
||||
} finally {
|
||||
releaseRunningSlot(projectID, taskKey);
|
||||
}
|
||||
|
||||
runningTaskKeys.delete(taskKey);
|
||||
runningGlobalCount = Math.max(0, runningGlobalCount - 1);
|
||||
const nextProjectCount = Math.max(0, (runningCountByProject.get(projectID) || 1) - 1);
|
||||
if (nextProjectCount === 0) {
|
||||
runningCountByProject.delete(projectID);
|
||||
} else {
|
||||
runningCountByProject.set(projectID, nextProjectCount);
|
||||
}
|
||||
|
||||
return {
|
||||
ok: status === 'success',
|
||||
status,
|
||||
sessionID,
|
||||
task: stateResult.task || null,
|
||||
error: errorMessage,
|
||||
};
|
||||
};
|
||||
|
||||
const pumpQueue = () => {
|
||||
@@ -761,9 +1013,18 @@ export const createScheduledTasksRuntime = (deps) => {
|
||||
queuedTaskKeys.delete(taskKey);
|
||||
consumed = true;
|
||||
|
||||
void runTask(item.projectID, item.taskID, item.reason).finally(() => {
|
||||
pumpQueue();
|
||||
});
|
||||
void runTask(item.projectID, item.taskID, item.reason, item.scheduledFor)
|
||||
.catch((error) => {
|
||||
logger.warn?.('[ScheduledTasks] queued run rejected', {
|
||||
projectID: item.projectID,
|
||||
taskID: item.taskID,
|
||||
reason: item.reason,
|
||||
error: safeErrorMessage(error),
|
||||
});
|
||||
})
|
||||
.finally(() => {
|
||||
pumpQueue();
|
||||
});
|
||||
}
|
||||
|
||||
if (!consumed && queue.length > 0) {
|
||||
|
||||
@@ -152,7 +152,13 @@ export const createScheduledTaskService = (dependencies) => {
|
||||
if (!result.ok) {
|
||||
throw new OpenChamberControlError(result.error || 'Task run failed', 500, { task: result.task });
|
||||
}
|
||||
return { task: result.task, sessionId: result.sessionID };
|
||||
return {
|
||||
task: result.task,
|
||||
sessionId: result.sessionID,
|
||||
...(typeof result.persistError === 'string' && result.persistError.trim()
|
||||
? { persistError: result.persistError.trim() }
|
||||
: {}),
|
||||
};
|
||||
};
|
||||
|
||||
const setEnabled = async (projectID, taskID, enabled) => {
|
||||
|
||||
@@ -262,3 +262,23 @@ describe('scheduled-task service remove', () => {
|
||||
expect(Array.isArray(tasks)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('scheduled-task service run', () => {
|
||||
it('forwards persistError when the runtime reports a completion persist failure', async () => {
|
||||
const { service } = createService({
|
||||
scheduledTasksRuntime: {
|
||||
runNow: vi.fn(async () => ({
|
||||
ok: true,
|
||||
sessionID: 'sess-1',
|
||||
task: { id: 'task-1', state: { lastStatus: 'success' } },
|
||||
persistError: 'timeout acquiring project config lock for project-test',
|
||||
reason: 'completion-state-failed',
|
||||
})),
|
||||
},
|
||||
});
|
||||
|
||||
const result = await service.run('project-test', 'task-1');
|
||||
expect(result.sessionId).toBe('sess-1');
|
||||
expect(result.persistError).toMatch(/timeout acquiring project config lock/);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user