|
|
|
@ -11,7 +11,8 @@ import { remove } from 'es-toolkit/array'
|
|
|
|
import {
|
|
|
|
import {
|
|
|
|
jobs as jobsTable,
|
|
|
|
jobs as jobsTable,
|
|
|
|
jobLock as jobLockTable,
|
|
|
|
jobLock as jobLockTable,
|
|
|
|
jobSchedule as jobScheduleTable
|
|
|
|
jobSchedule as jobScheduleTable,
|
|
|
|
|
|
|
|
jobHistory as jobHistoryTable
|
|
|
|
} from '../db/schema.js'
|
|
|
|
} from '../db/schema.js'
|
|
|
|
import { eq, inArray, sql } from 'drizzle-orm'
|
|
|
|
import { eq, inArray, sql } from 'drizzle-orm'
|
|
|
|
|
|
|
|
|
|
|
|
@ -192,8 +193,8 @@ export default {
|
|
|
|
WIKI.logger.info(`Processing new job ${job.id}: ${job.task}...`)
|
|
|
|
WIKI.logger.info(`Processing new job ${job.id}: ${job.task}...`)
|
|
|
|
// -> Add to Job History
|
|
|
|
// -> Add to Job History
|
|
|
|
await WIKI.db
|
|
|
|
await WIKI.db
|
|
|
|
.knex('jobHistory')
|
|
|
|
.insert(jobHistoryTable)
|
|
|
|
.insert({
|
|
|
|
.values({
|
|
|
|
id: job.id,
|
|
|
|
id: job.id,
|
|
|
|
task: job.task,
|
|
|
|
task: job.task,
|
|
|
|
state: 'active',
|
|
|
|
state: 'active',
|
|
|
|
@ -205,10 +206,9 @@ export default {
|
|
|
|
executedBy: WIKI.INSTANCE_ID,
|
|
|
|
executedBy: WIKI.INSTANCE_ID,
|
|
|
|
createdAt: job.createdAt
|
|
|
|
createdAt: job.createdAt
|
|
|
|
})
|
|
|
|
})
|
|
|
|
.onConflict('id')
|
|
|
|
.onConflictDoUpdate({
|
|
|
|
.merge({
|
|
|
|
target: jobHistoryTable.id,
|
|
|
|
executedBy: WIKI.INSTANCE_ID,
|
|
|
|
set: { executedBy: WIKI.INSTANCE_ID, startedAt: sql`now()` }
|
|
|
|
startedAt: new Date()
|
|
|
|
|
|
|
|
})
|
|
|
|
})
|
|
|
|
jobIds.push(job.id)
|
|
|
|
jobIds.push(job.id)
|
|
|
|
|
|
|
|
|
|
|
|
@ -224,14 +224,12 @@ export default {
|
|
|
|
}
|
|
|
|
}
|
|
|
|
// -> Update job history (success)
|
|
|
|
// -> Update job history (success)
|
|
|
|
await WIKI.db
|
|
|
|
await WIKI.db
|
|
|
|
.knex('jobHistory')
|
|
|
|
.update(jobHistoryTable)
|
|
|
|
.where({
|
|
|
|
.set({
|
|
|
|
id: job.id
|
|
|
|
|
|
|
|
})
|
|
|
|
|
|
|
|
.update({
|
|
|
|
|
|
|
|
state: 'completed',
|
|
|
|
state: 'completed',
|
|
|
|
completedAt: new Date()
|
|
|
|
completedAt: sql`now()`
|
|
|
|
})
|
|
|
|
})
|
|
|
|
|
|
|
|
.where(eq(jobHistoryTable.id, job.id))
|
|
|
|
WIKI.logger.info(`Completed job ${job.id}: ${job.task}`)
|
|
|
|
WIKI.logger.info(`Completed job ${job.id}: ${job.task}`)
|
|
|
|
this.pubsubClient.query(`SELECT pg_notify($1, $2)`, [
|
|
|
|
this.pubsubClient.query(`SELECT pg_notify($1, $2)`, [
|
|
|
|
'scheduler',
|
|
|
|
'scheduler',
|
|
|
|
@ -247,15 +245,13 @@ export default {
|
|
|
|
WIKI.logger.warn(err)
|
|
|
|
WIKI.logger.warn(err)
|
|
|
|
// -> Update job history (fail)
|
|
|
|
// -> Update job history (fail)
|
|
|
|
await WIKI.db
|
|
|
|
await WIKI.db
|
|
|
|
.knex('jobHistory')
|
|
|
|
.update(jobHistoryTable)
|
|
|
|
.where({
|
|
|
|
.set({
|
|
|
|
id: job.id
|
|
|
|
|
|
|
|
})
|
|
|
|
|
|
|
|
.update({
|
|
|
|
|
|
|
|
attempt: job.retries + 1,
|
|
|
|
attempt: job.retries + 1,
|
|
|
|
state: 'failed',
|
|
|
|
state: 'failed',
|
|
|
|
lastErrorMessage: err.message
|
|
|
|
lastErrorMessage: err.message
|
|
|
|
})
|
|
|
|
})
|
|
|
|
|
|
|
|
.where(eq(jobHistoryTable.id, job.id))
|
|
|
|
this.pubsubClient.query(`SELECT pg_notify($1, $2)`, [
|
|
|
|
this.pubsubClient.query(`SELECT pg_notify($1, $2)`, [
|
|
|
|
'scheduler',
|
|
|
|
'scheduler',
|
|
|
|
JSON.stringify({
|
|
|
|
JSON.stringify({
|
|
|
|
@ -269,7 +265,7 @@ export default {
|
|
|
|
// -> Reschedule for retry
|
|
|
|
// -> Reschedule for retry
|
|
|
|
if (job.retries < job.maxRetries) {
|
|
|
|
if (job.retries < job.maxRetries) {
|
|
|
|
const backoffDelay = 2 ** job.retries * WIKI.config.scheduler.retryBackoff
|
|
|
|
const backoffDelay = 2 ** job.retries * WIKI.config.scheduler.retryBackoff
|
|
|
|
await trx('jobs').insert({
|
|
|
|
await trx.insert(jobsTable).values({
|
|
|
|
...job,
|
|
|
|
...job,
|
|
|
|
retries: job.retries + 1,
|
|
|
|
retries: job.retries + 1,
|
|
|
|
waitUntil: DateTime.utc().plus({ seconds: backoffDelay }).toJSDate(),
|
|
|
|
waitUntil: DateTime.utc().plus({ seconds: backoffDelay }).toJSDate(),
|
|
|
|
@ -284,10 +280,13 @@ export default {
|
|
|
|
} catch (err) {
|
|
|
|
} catch (err) {
|
|
|
|
WIKI.logger.warn(err)
|
|
|
|
WIKI.logger.warn(err)
|
|
|
|
if (jobIds && jobIds.length > 0) {
|
|
|
|
if (jobIds && jobIds.length > 0) {
|
|
|
|
WIKI.db.knex('jobHistory').whereIn('id', jobIds).update({
|
|
|
|
WIKI.db
|
|
|
|
state: 'interrupted',
|
|
|
|
.update(jobHistoryTable)
|
|
|
|
lastErrorMessage: err.message
|
|
|
|
.set({
|
|
|
|
})
|
|
|
|
state: 'interrupted',
|
|
|
|
|
|
|
|
lastErrorMessage: err.message
|
|
|
|
|
|
|
|
})
|
|
|
|
|
|
|
|
.where(inArray(jobsTable.id, jobIds))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
},
|
|
|
|
},
|
|
|
|
|