hansjone/dsh-ops-cron
1
0
Fork 0
mirror of https://github.com/hansjone/dsh-ops-cron.git synced 2026-10-12 12:53:32 +08:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add cron_reschedule so pending one-shots can be moved without delete/recreate.

Browse source 
Keeps retrigger for consumed jobs only; agents can advance or delay nextRunAt safely.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-11 19:57:22 +08:00
parent b22048d377
commit 89ab5f59af
6 changed files with 382 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

113
lib/host.js
View file

@ -766,7 +766,7 @@ export function createHostService(options = {}) {
// one-shot
if (job.nextRunAt != null) {
const error = new Error('one-shot still has a pending next run; wait or edit the schedule instead of retrigger')
const error = new Error('one-shot still has a pending next run; use cron_reschedule (or PATCH schedule) instead of retrigger')
error.code = 'INVALID_RETRIGGER'
throw error
}
@ -822,6 +822,106 @@ export function createHostService(options = {}) {
}
}
/**
* Move a pending one-shot's next fire time (does not start a second concurrent run).
* Prefer this over delete+create when nextRunAt is still set.
* Does not change enabled (paused stays paused). Does not dispatch.
*/
async function rescheduleJob(jobId, opts = {}, identity = null) {
const afterRaw = opts.afterMinutes ?? opts.after_minutes
const hasAfter = afterRaw !== undefined && afterRaw !== null && afterRaw !== ''
const afterMinutes = hasAfter ? Number(afterRaw) : null
const atRaw = typeof opts.at === 'string' ? opts.at.trim() : ''
const tzOpt = typeof opts.timezone === 'string' && opts.timezone.trim()
? opts.timezone.trim()
: (typeof opts.time_zone === 'string' && opts.time_zone.trim() ? opts.time_zone.trim() : '')
if (hasAfter && (!Number.isFinite(afterMinutes) || afterMinutes < 0)) {
const error = new Error('after_minutes must be >= 0')
error.code = 'INVALID_RESCHEDULE'
throw error
}
if (!hasAfter && !atRaw) {
const error = new Error('provide after_minutes or at')
error.code = 'INVALID_RESCHEDULE'
throw error
}
const t = now()
let previousNextRunAt = null
await withState((current) => {
const job = getJob(current, jobId)
if (!job) {
const error = new Error('job not found')
error.code = 'NOT_FOUND'
throw error
}
if (identity) assertCanAccessJob(job, identity)
if (job.schedule?.kind !== 'at') {
const error = new Error('cron_reschedule only applies to one-shot (at) jobs')
error.code = 'INVALID_RESCHEDULE'
throw error
}
if (job.nextRunAt == null) {
const error = new Error('one-shot has no pending next run; use cron_retrigger after it finishes')
error.code = 'INVALID_RESCHEDULE'
throw error
}
const runs = (current.runs || []).filter((run) => run?.jobId === job.id)
if (runs.some((run) => ACTIVE_RUN_STATUSES.has(run.status))) {
const error = new Error('job already has a pending or running instance')
error.code = 'ALREADY_RUNNING'
throw error
}
const timezone = tzOpt
|| job.schedule?.timezone
|| current.settings?.timezone
|| 'Asia/Shanghai'
let atMs
if (hasAfter) {
atMs = t + Math.round(afterMinutes * 60_000)
} else {
const validated = validateSchedule({ kind: 'at', at: atRaw, timezone }, timezone)
atMs = validated.at
}
if (!Number.isFinite(atMs)) {
const error = new Error('invalid reschedule time')
error.code = 'INVALID_RESCHEDULE'
throw error
}
// Align with createJobRecord: slight past (≤60s) → fire ASAP; older → reject.
if (atMs < t - 60_000) {
const error = new Error(`reschedule time is already in the past (now is ${new Date(t).toISOString()})`)
error.code = 'INVALID_RESCHEDULE'
throw error
}
if (atMs < t) atMs = t
previousNextRunAt = job.nextRunAt
return upsertJob(current, {
...job,
schedule: {
kind: 'at',
at: new Date(atMs).toISOString(),
timezone,
},
nextRunAt: atMs,
updatedAt: t,
})
})
const state = await snapshot()
const job = getJob(state, jobId)
return {
ok: true,
job: jobView(job, state.runs),
nextRunAt: job?.nextRunAt ?? null,
previousNextRunAt,
}
}
async function deleteJob(jobId) {
await withState((current) => {
if (!getJob(current, jobId)) {
@ -1135,7 +1235,7 @@ export function createHostService(options = {}) {
return
}
const jobMatch = path.match(new RegExp(`^${API_PREFIX}/jobs/([^/]+)(/run|/pause|/resume|/retrigger)?$`))
const jobMatch = path.match(new RegExp(`^${API_PREFIX}/jobs/([^/]+)(/run|/pause|/resume|/retrigger|/reschedule)?$`))
if (jobMatch) {
const jobId = decodeURIComponent(jobMatch[1])
const rest = jobMatch[2] || ''
@ -1170,6 +1270,12 @@ export function createHostService(options = {}) {
write(200, result)
return
}
if (method === 'POST' && rest === '/reschedule') {
const body = await readJsonBody(req).catch(() => ({}))
const result = await rescheduleJob(jobId, body || {}, identity)
write(200, result)
return
}
if (method === 'POST' && rest === '/pause') {
const job = await pauseJob(jobId, false, identity)
write(200, { ok: true, job })
@ -1325,7 +1431,7 @@ export function createHostService(options = {}) {
if (code === 'ALREADY_RUNNING') {
return write(409, { ok: false, error: error.message, code, message: error.message })
}
if (code === 'INVALID_CRON' || code === 'INVALID_AT' || code === 'INVALID_SCHEDULE' || code === 'INVALID_JOB' || code === 'INVALID_TIMEZONE' || code === 'INVALID_CWD' || code === 'INVALID_DELIVERY' || code === 'IM_DELIVERY_FORBIDDEN' || code === 'IM_TARGET_FORBIDDEN' || code === 'IM_JOB_MISSING_CWD' || code === 'INVALID_WATCH' || code === 'INVALID_PROGRESS' || code === 'INVALID_REPORT' || code === 'INVALID_PERSIST_HISTORY' || code === 'INVALID_PROGRESS_PATH' || code === 'INVALID_ARCHIVE' || code === 'ARCHIVE_FAILED' || code === 'ARCHIVE_UNAVAILABLE' || code === 'INVALID_RETRIGGER' || code === 'INVALID_PERMISSION_PRESET') {
if (code === 'INVALID_CRON' || code === 'INVALID_AT' || code === 'INVALID_SCHEDULE' || code === 'INVALID_JOB' || code === 'INVALID_TIMEZONE' || code === 'INVALID_CWD' || code === 'INVALID_DELIVERY' || code === 'IM_DELIVERY_FORBIDDEN' || code === 'IM_TARGET_FORBIDDEN' || code === 'IM_JOB_MISSING_CWD' || code === 'INVALID_WATCH' || code === 'INVALID_PROGRESS' || code === 'INVALID_REPORT' || code === 'INVALID_PERSIST_HISTORY' || code === 'INVALID_PROGRESS_PATH' || code === 'INVALID_ARCHIVE' || code === 'ARCHIVE_FAILED' || code === 'ARCHIVE_UNAVAILABLE' || code === 'INVALID_RETRIGGER' || code === 'INVALID_RESCHEDULE' || code === 'INVALID_PERMISSION_PRESET') {
return write(400, { ok: false, error: error.message, code, message: error.message })
}
if (code === 'PAYLOAD_TOO_LARGE') return write(413, apiError('payload_too_large', locale))
@ -1498,6 +1604,7 @@ export function createHostService(options = {}) {
updateJob,
pauseJob,
retriggerJob,
rescheduleJob,
deleteJob,
updateSettings,
dispatchRun,