-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge remote-tracking branch 'origin/master' into feat/achievements
- Loading branch information
Showing
14 changed files
with
304 additions
and
173 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| import { JobName, allJobs } from './list'; | ||
| import tracer from '../common/logger/tracing'; | ||
| import { getLogger } from '../common/logger/logger'; | ||
| import { metrics, metricsRouter } from '../common/logger/metrics'; | ||
| import { prisma } from '../common/prisma'; | ||
| import { Prisma, job_run } from '@prisma/client'; | ||
| import assert from 'assert'; | ||
|
|
||
| const logger = getLogger('Job Execution'); | ||
|
|
||
| enum LockStatus { | ||
| // We failed to aquire the lock as the transaction was rolled back by the database | ||
| // (due to a conflict), but we don't yet know whether another job is currently running | ||
| // Best to retry soon | ||
| ROLLBACK = 1, | ||
| // Failed to aquire a lock because the same job is already running | ||
| CONFLICT = 2, | ||
| AQUIRED = 3, | ||
| } | ||
|
|
||
| export async function runJob(jobName: JobName): Promise<boolean> { | ||
| let success = false; | ||
|
|
||
| try { | ||
| logger.info(`Starting to run Job '${jobName}'`, { jobName }); | ||
|
|
||
| // ---------- AQUIRE -------------- | ||
| // Prevent Job Runs running concurrently (across dynos), as jobs usually lack synchronization internally | ||
| // To synchronize we use the 'job_run' table in our Postgres | ||
| // During insert we need transaction level SERIALIZABLE to prevent two jobs from inserting a new job run | ||
| // at the same time | ||
|
|
||
| let jobRun: job_run; | ||
| let lockStatus: LockStatus = LockStatus.ROLLBACK as LockStatus; | ||
| let lockRetries = 5; | ||
|
|
||
| do { | ||
| try { | ||
| // Wait between 0 and 1000ms to reduce the likelihood of transaction deadlocks | ||
| // (as a lot of Cron Jobs fire at exactly the same time) | ||
| await new Promise((resolve) => setTimeout(resolve, Math.floor(Math.random() * 1000))); | ||
|
|
||
| jobRun = await prisma.$transaction( | ||
| async (jobPrisma) => { | ||
| const runningJob = await jobPrisma.job_run.findFirst({ | ||
| where: { | ||
| job_name: jobName, | ||
| endedAt: { equals: null }, | ||
| }, | ||
| }); | ||
|
|
||
| if (runningJob) { | ||
| logger.error( | ||
| `Cannot concurrently execute Job '${jobName}' as it is already running on '${runningJob.worker}' since ${runningJob.startedAt}`, | ||
| undefined, | ||
| { jobName, runningJob } | ||
| ); | ||
| lockStatus = LockStatus.CONFLICT; | ||
| return undefined; | ||
| } | ||
|
|
||
| lockStatus = LockStatus.AQUIRED; | ||
|
|
||
| return await jobPrisma.job_run.create({ | ||
| data: { job_name: jobName, worker: process.env.DYNO ?? '?' }, | ||
| }); | ||
| // It is important that the transaction ends here and the INSERT above is commited | ||
| // Otherwise we would continue execution, and the commit would be rolled back after the job actually executed | ||
| }, | ||
| { isolationLevel: Prisma.TransactionIsolationLevel.Serializable } | ||
| ); | ||
| } catch (error) { | ||
| logger.warn(`Aquiring Lock failed - ${error.message}`, { jobName, error }); | ||
| // The transaction was aborted, likely because the DB rolled back the deadlock | ||
| lockStatus = LockStatus.ROLLBACK; | ||
| lockRetries -= 1; | ||
| } | ||
| } while (lockStatus === LockStatus.ROLLBACK && lockRetries > 0); | ||
|
|
||
| if (lockStatus === LockStatus.CONFLICT) { | ||
| return false; | ||
| } | ||
|
|
||
| if (lockStatus === LockStatus.ROLLBACK) { | ||
| logger.error(`Failed to aquire Lock after at most 5 retries - This might leave the system in a locked state requiring manual cleanup!`, undefined, { | ||
| jobName, | ||
| }); | ||
| return false; | ||
| } | ||
|
|
||
| assert.ok(lockStatus === LockStatus.AQUIRED); | ||
| assert.ok(runJob != null); | ||
|
|
||
| logger.info(`Aquired Table Lock to run Job '${jobName}'`, { jobName }); | ||
|
|
||
| // ---------- RUN ---------------- | ||
|
|
||
| const span = tracer.startSpan(jobName); | ||
| await tracer.scope().activate(span, async () => { | ||
| let hasError = false; | ||
| try { | ||
| const job = allJobs[jobName]; | ||
| await job(); | ||
| success = true; | ||
| } catch (e) { | ||
| logger.error(`Can't execute job: ${jobName} due to error`, e); | ||
| logger.debug(e); | ||
| hasError = true; | ||
| } | ||
|
|
||
| metrics.JobCountExecuted.inc({ hasError: `${hasError}`, name: jobName }); | ||
|
|
||
| span.finish(); | ||
| }); | ||
|
|
||
| logger.info(`Finished Running Job '${jobName}', releasing table lock`, { jobName }); | ||
|
|
||
| // ---------- RELEASE ------------- | ||
| await prisma.job_run.update({ | ||
| where: { job_name_startedAt: { startedAt: jobRun.startedAt, job_name: jobRun.job_name } }, | ||
| data: { endedAt: new Date() }, | ||
| }); | ||
|
|
||
| logger.info(`Finished Job '${jobName}'`, { jobName }); | ||
| } catch (error) { | ||
| logger.error(error.message); | ||
| logger.error(`Failure during Job Scheduling - This might leave the system in a locked state requiring manual cleanup!`, error, { jobName }); | ||
| success = false; | ||
| // Eventually we now have a job run in the job_run table that has no endedAt, | ||
| // but which will never finish. To unlock this again, simply delete this entry | ||
| // (This should only happen in the rare case that the Dyno is killed (!) during execution) | ||
| } | ||
|
|
||
| return success; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,39 +1,61 @@ | ||
| import { CSCronJob } from './types'; | ||
|
|
||
| //import the jobs | ||
| import * as Notification from '../common/notification'; | ||
| import { cleanupSecrets } from '../common/secret'; | ||
| import dropOldNotificationContexts from './periodic/drop-old-notification-contexts'; | ||
| import { runInterestConfirmations } from '../common/match/pool'; | ||
| import { checkReminders } from '../common/notification'; | ||
| import { cleanupSecrets } from '../common/secret'; | ||
| import anonymiseAttendanceLog from './periodic/anonymise-attendance-log'; | ||
| import syncToWebflow from './periodic/sync-to-webflow'; | ||
| import { postStatisticsToSlack } from './slack-statistics'; | ||
| import dropOldNotificationContexts from './periodic/drop-old-notification-contexts'; | ||
| import flagInactiveConversationsAsReadonly from './periodic/flag-old-conversations'; | ||
| import redactInactiveAccounts from './periodic/redact-inactive-accounts'; | ||
| import { sendInactivityNotification } from './periodic/redact-inactive-accounts/send-inactivity-notification'; | ||
| import { deactivateInactiveAccounts } from './periodic/redact-inactive-accounts/deactivate-inactive-accounts'; | ||
| import { sendInactivityNotification } from './periodic/redact-inactive-accounts/send-inactivity-notification'; | ||
| import syncToWebflow from './periodic/sync-to-webflow'; | ||
| import { postStatisticsToSlack } from './slack-statistics'; | ||
| import notificationsEndedYesterday from './periodic/notification-courses-ended-yesterday'; | ||
|
|
||
| export const allJobs = { | ||
| cleanupSecrets, | ||
| dropOldNotificationContexts, | ||
| runInterestConfirmations, | ||
| anonymiseAttendanceLog, | ||
| syncToWebflow, | ||
| postStatisticsToSlack, | ||
| redactInactiveAccounts, | ||
| sendInactivityNotification, | ||
| deactivateInactiveAccounts, | ||
| flagInactiveConversationsAsReadonly, | ||
| notificationsEndedYesterday, | ||
| checkReminders, | ||
|
|
||
| // For Integration Tests only: | ||
| NOTHING_DO_NOT_USE: async () => { | ||
| await new Promise((resolve) => setTimeout(resolve, 1000)); | ||
| }, | ||
| } as const; | ||
|
|
||
| export type JobName = keyof typeof allJobs; | ||
| export const jobExists = (name: string): name is JobName => name in allJobs; | ||
|
|
||
| // A list of all jobs that should be scheduled at the moment | ||
| export const allJobs: CSCronJob[] = [ | ||
| export type ScheduledJob = { cronTime: string; name: JobName }; | ||
| export const regularJobs: ScheduledJob[] = [ | ||
| // every morning, quite early (but only on Monday and Thursday) | ||
| // { cronTime: "00 55 07 * * 1,4", jobFunction: initialInterestConfirmationRequests}, | ||
| { cronTime: '00 55 07 * * 1,4', jobFunction: runInterestConfirmations, name: 'runInterestConfirmations' }, | ||
| // { cronTime: "00 56 08 * * *", jobFunction: tutoringMatchMaking}, // only scheduled manually, at the moment | ||
| { cronTime: '00 55 07 * * 1,4', name: 'runInterestConfirmations' }, | ||
| // every morning, but a little bit later | ||
| // every 10 minutes during the day (to distribute load and send out notifications faster) | ||
| { cronTime: '00 */10 * * * *', jobFunction: Notification.checkReminders, name: 'checkReminders' }, | ||
| { cronTime: '00 */10 * * * *', name: 'checkReminders' }, | ||
| // each night - database cleanups | ||
| { cronTime: '00 00 05 * * *', jobFunction: anonymiseAttendanceLog, name: 'anonymiseAttendanceLog' }, | ||
| { cronTime: '00 00 04 * * *', jobFunction: cleanupSecrets, name: 'cleanupSecrets' }, | ||
| { cronTime: '00 00 01 * * *', jobFunction: dropOldNotificationContexts, name: 'dropOldNotificationContexts' }, | ||
| { cronTime: '00 00 05 * * *', name: 'anonymiseAttendanceLog' }, | ||
| { cronTime: '00 00 04 * * *', name: 'cleanupSecrets' }, | ||
| { cronTime: '00 00 01 * * *', name: 'dropOldNotificationContexts' }, | ||
| // Account redaction | ||
| { cronTime: '00 00 01 * * *', jobFunction: deactivateInactiveAccounts, name: 'deactivateInactiveAccounts' }, | ||
| { cronTime: '00 00 02 * * *', jobFunction: redactInactiveAccounts, name: 'redactInactiveAccounts' }, | ||
| { cronTime: '00 00 02 * * *', jobFunction: sendInactivityNotification, name: 'sendInactivityNotification' }, | ||
| { cronTime: '00 00 01 * * *', name: 'deactivateInactiveAccounts' }, | ||
| { cronTime: '00 00 02 * * *', name: 'redactInactiveAccounts' }, | ||
| { cronTime: '00 00 02 * * *', name: 'sendInactivityNotification' }, | ||
| // Synch DB data to webflow CMS | ||
| { cronTime: '00 */15 * * * *', jobFunction: syncToWebflow, name: 'syncToWebflow' }, | ||
| { cronTime: '00 */15 * * * *', name: 'syncToWebflow' }, | ||
| // Send Slack Messages monthly: | ||
| { cronTime: '00 00 10 01 * *', jobFunction: postStatisticsToSlack, name: 'postStatisticsToSlack' }, | ||
| { cronTime: '00 00 10 01 * *', name: 'postStatisticsToSlack' }, | ||
| // Disable old chats on a daily basis: | ||
| { cronTime: '00 00 10 * * *', name: 'flagInactiveConversationsAsReadonly' }, | ||
| // Every night, trigger actions for courses that ended yesterday | ||
| { cronTime: '00 00 10 * * *', jobFunction: notificationsEndedYesterday, name: 'notificationsEndedYesterday' }, | ||
| { cronTime: '00 00 10 * * *', name: 'notificationsEndedYesterday' }, | ||
| ]; |
Oops, something went wrong.