|
|
|
|
@ -16,6 +16,43 @@ export interface JobHistoryPage {
|
|
|
|
|
jobs: (typeof jobHistoryTable.$inferSelect)[]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/** One entry of the system's own cron schedule. */
|
|
|
|
|
interface SystemScheduleEntry {
|
|
|
|
|
task: string
|
|
|
|
|
cron: string
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* The system's cron schedule, in full.
|
|
|
|
|
*
|
|
|
|
|
* This is the definition, not a seed: `reconcileSchedule()` makes the `jobSchedule` rows of type
|
|
|
|
|
* `system` match it on every boot. Adding a task here is therefore all it takes to have every
|
|
|
|
|
* instance start running it — a list that was only inserted on first run would leave a task nobody's
|
|
|
|
|
* database had ever heard of unscheduled forever, which is exactly what happened to
|
|
|
|
|
* `syncStorageTargets`.
|
|
|
|
|
*
|
|
|
|
|
* A `type: 'user'` row is nothing to do with this and is never touched.
|
|
|
|
|
*/
|
|
|
|
|
export const SYSTEM_SCHEDULE: SystemScheduleEntry[] = [
|
|
|
|
|
{ task: 'checkVersion', cron: '0 0 * * *' },
|
|
|
|
|
{ task: 'cleanJobHistory', cron: '5 0 * * *' },
|
|
|
|
|
// { task: 'refreshAutocomplete', cron: '0 */6 * * *' },
|
|
|
|
|
{ task: 'purgeRateLimits', cron: '10 * * * *' },
|
|
|
|
|
{ task: 'updateLocales', cron: '0 0 * * *' },
|
|
|
|
|
// -> Every minute, and the task decides which sites are actually due: the interval is a per-site
|
|
|
|
|
// setting, so the tick has to be as fine as the shortest one anybody can ask for
|
|
|
|
|
{ task: 'syncStorageTargets', cron: '* * * * *' }
|
|
|
|
|
]
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Advisory lock held for the length of `reconcileSchedule()`'s transaction.
|
|
|
|
|
*
|
|
|
|
|
* Every instance reconciles as it boots, and a restarted HA set boots them together — without this
|
|
|
|
|
* they all read the same missing row and all insert it. Transaction-scoped, so it is released with
|
|
|
|
|
* the commit whatever happens.
|
|
|
|
|
*/
|
|
|
|
|
const SCHEDULE_LOCK_KEY = 4210001
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Jobs model
|
|
|
|
|
*
|
|
|
|
|
@ -26,45 +63,11 @@ export interface JobHistoryPage {
|
|
|
|
|
class Jobs {
|
|
|
|
|
/**
|
|
|
|
|
* Initialize jobs table
|
|
|
|
|
*
|
|
|
|
|
* Only the cron lock: the schedule itself is `reconcileSchedule()`'s, on this boot and every one
|
|
|
|
|
* after it.
|
|
|
|
|
*/
|
|
|
|
|
async init(): Promise<void> {
|
|
|
|
|
WIKI.logger.info('Inserting scheduled jobs...')
|
|
|
|
|
|
|
|
|
|
await WIKI.db.insert(jobScheduleTable).values([
|
|
|
|
|
{
|
|
|
|
|
task: 'checkVersion',
|
|
|
|
|
cron: '0 0 * * *',
|
|
|
|
|
type: 'system'
|
|
|
|
|
},
|
|
|
|
|
{
|
|
|
|
|
task: 'cleanJobHistory',
|
|
|
|
|
cron: '5 0 * * *',
|
|
|
|
|
type: 'system'
|
|
|
|
|
},
|
|
|
|
|
// {
|
|
|
|
|
// task: 'refreshAutocomplete',
|
|
|
|
|
// cron: '0 */6 * * *',
|
|
|
|
|
// type: 'system'
|
|
|
|
|
// },
|
|
|
|
|
{
|
|
|
|
|
task: 'purgeRateLimits',
|
|
|
|
|
cron: '10 * * * *',
|
|
|
|
|
type: 'system'
|
|
|
|
|
},
|
|
|
|
|
{
|
|
|
|
|
task: 'updateLocales',
|
|
|
|
|
cron: '0 0 * * *',
|
|
|
|
|
type: 'system'
|
|
|
|
|
},
|
|
|
|
|
{
|
|
|
|
|
// -> Every minute, and the task decides which sites are actually due: the interval is a
|
|
|
|
|
// per-site setting, so the tick has to be as fine as the shortest one anybody can ask for
|
|
|
|
|
task: 'syncStorageTargets',
|
|
|
|
|
cron: '* * * * *',
|
|
|
|
|
type: 'system'
|
|
|
|
|
}
|
|
|
|
|
])
|
|
|
|
|
|
|
|
|
|
await WIKI.db.insert(jobLockTable).values({
|
|
|
|
|
key: 'cron',
|
|
|
|
|
lastCheckedBy: 'init',
|
|
|
|
|
@ -77,6 +80,83 @@ class Jobs {
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Bring the `system` cron entries in line with `SYSTEM_SCHEDULE`.
|
|
|
|
|
*
|
|
|
|
|
* Runs on every boot, which is what makes the code the authority on what the wiki runs on a
|
|
|
|
|
* schedule: a task added to the list appears on an existing instance, a task removed from it stops
|
|
|
|
|
* being queued, and a changed cron takes effect. A row deleted by hand comes back, too.
|
|
|
|
|
*
|
|
|
|
|
* Pending iterations of anything that changed are dropped, since `scheduler.addScheduled()` only
|
|
|
|
|
* ever adds: left alone, a task whose cron just changed would keep running at its old times for
|
|
|
|
|
* the next 24 hours, and one that no longer exists would be queued for a task the scheduler cannot
|
|
|
|
|
* load.
|
|
|
|
|
*/
|
|
|
|
|
async reconcileSchedule(): Promise<void> {
|
|
|
|
|
await WIKI.db.transaction(async (trx: any) => {
|
|
|
|
|
await trx.execute(sql`SELECT pg_advisory_xact_lock(${SCHEDULE_LOCK_KEY}::bigint)`)
|
|
|
|
|
|
|
|
|
|
const existing = await trx
|
|
|
|
|
.select()
|
|
|
|
|
.from(jobScheduleTable)
|
|
|
|
|
.where(eq(jobScheduleTable.type, 'system'))
|
|
|
|
|
|
|
|
|
|
const wanted = new Map(SYSTEM_SCHEDULE.map((entry) => [entry.task, entry]))
|
|
|
|
|
const kept = new Set<string>()
|
|
|
|
|
const staleIds: string[] = []
|
|
|
|
|
// -> Tasks whose queued iterations no longer match what is scheduled for them
|
|
|
|
|
const dirtyTasks = new Set<string>()
|
|
|
|
|
|
|
|
|
|
for (const row of existing) {
|
|
|
|
|
const entry = wanted.get(row.task)
|
|
|
|
|
// -> A task that is gone, or a duplicate of one already kept: either way this row goes
|
|
|
|
|
if (!entry || kept.has(row.task)) {
|
|
|
|
|
staleIds.push(row.id)
|
|
|
|
|
dirtyTasks.add(row.task)
|
|
|
|
|
continue
|
|
|
|
|
}
|
|
|
|
|
kept.add(row.task)
|
|
|
|
|
if (row.cron !== entry.cron) {
|
|
|
|
|
await trx
|
|
|
|
|
.update(jobScheduleTable)
|
|
|
|
|
.set({
|
|
|
|
|
cron: entry.cron,
|
|
|
|
|
updatedAt: new Date(Temporal.Now.instant().epochMilliseconds)
|
|
|
|
|
})
|
|
|
|
|
.where(eq(jobScheduleTable.id, row.id))
|
|
|
|
|
dirtyTasks.add(row.task)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (staleIds.length > 0) {
|
|
|
|
|
await trx.delete(jobScheduleTable).where(inArray(jobScheduleTable.id, staleIds))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const missing = SYSTEM_SCHEDULE.filter((entry) => !kept.has(entry.task))
|
|
|
|
|
if (missing.length > 0) {
|
|
|
|
|
await trx.insert(jobScheduleTable).values(
|
|
|
|
|
missing.map((entry) => ({
|
|
|
|
|
task: entry.task,
|
|
|
|
|
cron: entry.cron,
|
|
|
|
|
type: 'system'
|
|
|
|
|
}))
|
|
|
|
|
)
|
|
|
|
|
for (const entry of missing) {
|
|
|
|
|
dirtyTasks.add(entry.task)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (dirtyTasks.size > 0) {
|
|
|
|
|
await trx
|
|
|
|
|
.delete(jobsTable)
|
|
|
|
|
.where(and(eq(jobsTable.isScheduled, true), inArray(jobsTable.task, [...dirtyTasks])))
|
|
|
|
|
WIKI.logger.info(
|
|
|
|
|
`Scheduled tasks reconciled: ${[...dirtyTasks].sort().join(', ')} [ UPDATED ]`
|
|
|
|
|
)
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Whether the scheduler is keeping up with its cron duties.
|
|
|
|
|
*
|
|
|
|
|
|