mirror of
https://github.com/hansjone/dsh-ops-cron.git
synced 2026-10-19 17:10:46 +08:00
Fix one-shot jobs that never auto-fire after restart or slow ticks.
recover was clearing past-due at nextRunAt via nextFire→null; keep and re-arm them, and evaluate schedule misfire at tick start so a slow prior job cannot age peers out of grace. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
8cf624ce7f
commit
b22048d377
2 changed files with 82 additions and 3 deletions
17
lib/host.js
17
lib/host.js
|
|
@ -842,9 +842,9 @@ export function createHostService(options = {}) {
|
||||||
return state.settings
|
return state.settings
|
||||||
}
|
}
|
||||||
|
|
||||||
async function dispatchRun(jobId, trigger) {
|
async function dispatchRun(jobId, trigger, opts = {}) {
|
||||||
let claimed
|
let claimed
|
||||||
const t = now()
|
const t = Number.isFinite(opts.at) ? opts.at : now()
|
||||||
await withState((current) => {
|
await withState((current) => {
|
||||||
claimed = claimOccurrence(current, jobId, t, trigger, current.settings)
|
claimed = claimOccurrence(current, jobId, t, trigger, current.settings)
|
||||||
if (claimed.decision.action === 'wait') return current
|
if (claimed.decision.action === 'wait') return current
|
||||||
|
|
@ -897,7 +897,9 @@ export function createHostService(options = {}) {
|
||||||
misfirePolicy: state.settings.misfirePolicy,
|
misfirePolicy: state.settings.misfirePolicy,
|
||||||
})
|
})
|
||||||
if (decision.action === 'wait') continue
|
if (decision.action === 'wait') continue
|
||||||
const result = await dispatchRun(job.id, 'schedule')
|
// Use tick start time for claim/misfire so a slow earlier job cannot
|
||||||
|
// push later oneshots past the grace window.
|
||||||
|
const result = await dispatchRun(job.id, 'schedule', { at: t })
|
||||||
if (result.run) fired.push(result)
|
if (result.run) fired.push(result)
|
||||||
}
|
}
|
||||||
await tickWatcherReports()
|
await tickWatcherReports()
|
||||||
|
|
@ -967,6 +969,15 @@ export function createHostService(options = {}) {
|
||||||
try {
|
try {
|
||||||
if (job.nextRunAt === null) return job
|
if (job.nextRunAt === null) return job
|
||||||
if (Number.isFinite(job.nextRunAt) && job.nextRunAt > t) return job
|
if (Number.isFinite(job.nextRunAt) && job.nextRunAt > t) return job
|
||||||
|
// Past-due one-shot: do NOT call nextFire (that returns null and
|
||||||
|
// permanently kills the job). Re-arm to now so the next tick fires
|
||||||
|
// a catch-up run instead of wiping or grace-misfiring after restart.
|
||||||
|
if (job.schedule?.kind === 'at') {
|
||||||
|
if (Number.isFinite(job.nextRunAt) && job.nextRunAt <= t) {
|
||||||
|
return { ...job, nextRunAt: t }
|
||||||
|
}
|
||||||
|
return job
|
||||||
|
}
|
||||||
const nextRunAt = nextFire(job.schedule, t, job.schedule?.timezone || next.settings.timezone)
|
const nextRunAt = nextFire(job.schedule, t, job.schedule?.timezone || next.settings.timezone)
|
||||||
return { ...job, nextRunAt }
|
return { ...job, nextRunAt }
|
||||||
} catch {
|
} catch {
|
||||||
|
|
|
||||||
|
|
@ -718,6 +718,74 @@ test('one-shot at job fires once then later ticks wait instead of retriggering',
|
||||||
assert.equal(misfires.length, 0)
|
assert.equal(misfires.length, 0)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test('recover re-arms past-due oneshot instead of clearing nextRunAt', async (t) => {
|
||||||
|
const dir = await mkdtemp(join(tmpdir(), 'dsh-ops-cron-'))
|
||||||
|
t.after(() => rm(dir, { recursive: true, force: true }))
|
||||||
|
const at = Date.parse('2026-09-11T00:00:00.000Z')
|
||||||
|
let clock = at - 30_000
|
||||||
|
let sessions = 0
|
||||||
|
const service = createTestHost({
|
||||||
|
filePath: join(dir, 'store.json'),
|
||||||
|
now: () => clock,
|
||||||
|
sessionPort: {
|
||||||
|
async createAndPrompt() {
|
||||||
|
sessions += 1
|
||||||
|
return { sessionId: `catchup-${sessions}`, status: 'succeeded', summary: 'catch-up' }
|
||||||
|
},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
const job = await service.createJob({
|
||||||
|
name: 'catchup-at',
|
||||||
|
prompt: 'ping',
|
||||||
|
schedule: { kind: 'at', at: new Date(at).toISOString(), timezone: 'UTC' },
|
||||||
|
})
|
||||||
|
assert.equal(job.nextRunAt, at)
|
||||||
|
|
||||||
|
// Simulate Host restart after the due time (previously wiped nextRunAt via nextFire→null).
|
||||||
|
clock = at + 90_000
|
||||||
|
await service.recover()
|
||||||
|
const after = await service.getJob(job.id)
|
||||||
|
assert.equal(after.nextRunAt, clock)
|
||||||
|
|
||||||
|
const fired = await service.tick()
|
||||||
|
assert.equal(fired.length, 1)
|
||||||
|
assert.equal(fired[0].run.status, 'succeeded')
|
||||||
|
assert.equal(sessions, 1)
|
||||||
|
})
|
||||||
|
|
||||||
|
test('tick evaluates misfire against tick-start time so a slow prior job cannot age out oneshots', async (t) => {
|
||||||
|
const dir = await mkdtemp(join(tmpdir(), 'dsh-ops-cron-'))
|
||||||
|
t.after(() => rm(dir, { recursive: true, force: true }))
|
||||||
|
const due = Date.parse('2026-09-11T01:00:00.000Z')
|
||||||
|
let clock = due
|
||||||
|
let sessions = 0
|
||||||
|
const service = createTestHost({
|
||||||
|
filePath: join(dir, 'store.json'),
|
||||||
|
now: () => clock,
|
||||||
|
sessionPort: {
|
||||||
|
async createAndPrompt({ job }) {
|
||||||
|
sessions += 1
|
||||||
|
// First job stretches wall clock past oneshot grace (60s).
|
||||||
|
if (job.name === 'slow') clock = due + 90_000
|
||||||
|
return { sessionId: `s-${sessions}`, status: 'succeeded', summary: 'ok' }
|
||||||
|
},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
await service.createJob({
|
||||||
|
name: 'slow',
|
||||||
|
prompt: 'first',
|
||||||
|
schedule: { kind: 'at', at: new Date(due).toISOString(), timezone: 'UTC' },
|
||||||
|
})
|
||||||
|
await service.createJob({
|
||||||
|
name: 'peer',
|
||||||
|
prompt: 'second',
|
||||||
|
schedule: { kind: 'at', at: new Date(due).toISOString(), timezone: 'UTC' },
|
||||||
|
})
|
||||||
|
const fired = await service.tick()
|
||||||
|
assert.equal(fired.filter((row) => row.run?.status === 'succeeded').length, 2)
|
||||||
|
assert.equal(sessions, 2)
|
||||||
|
})
|
||||||
|
|
||||||
test('unarchiveSession drops the id from archivedSessionIds', async () => {
|
test('unarchiveSession drops the id from archivedSessionIds', async () => {
|
||||||
const state = { workspaceIds: ['w'], archivedSessionIds: ['s1', 's2'], initialized: true }
|
const state = { workspaceIds: ['w'], archivedSessionIds: ['s1', 's2'], initialized: true }
|
||||||
const ctx = {
|
const ctx = {
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue