HARMONY-2016: Fix race condition in work reaper.

ygliuvt · ygliuvt · commit 11df0f508c1f · 2025-03-04T17:47:59.000-05:00
diff --git a/services/cron-service/app/cronjobs/work-reaper.ts b/services/cron-service/app/cronjobs/work-reaper.ts
@@ -1,3 +1,4 @@
+import { subMinutes } from 'date-fns';
 import { JobStatus, terminalStates } from '../../../harmony/app/models/job';
 import {
   deleteWorkItemsById, getWorkItemIdsByJobUpdateAgeAndStatus,
@@ -10,12 +11,12 @@ import env from '../util/env';
 import { CronJob } from './cronjob';
 
 /**
- * Find work items that are older than notUpdatedForMinutes and delete them.
- * @param notUpdatedForMinutes - upper limit on the duration since the last update
+ * Find work items that are older than updatedAtCutoff and delete them.
+ * @param updatedAtCutoff - updatedAt cutoff time since the last update
  * @param jobStatus - a list of terminal job statuses
  * @returns Resolves when the request is complete
  */
-async function deleteTerminalWorkItems(ctx: Context, notUpdatedForMinutes: number, jobStatus: JobStatus[]): Promise < void> {
+async function deleteTerminalWorkItems(ctx: Context, updatedAtCutoff: Date, jobStatus: JobStatus[]): Promise < void> {
   let done = false;
   let startingId = 0;
   let totalDeleted = 0;
@@ -26,7 +27,7 @@ async function deleteTerminalWorkItems(ctx: Context, notUpdatedForMinutes: numbe
   while (!done) {
     try {
       const workItemIds = await getWorkItemIdsByJobUpdateAgeAndStatus(
-        db, notUpdatedForMinutes, jobStatus, startingId, batchSize,
+        db, updatedAtCutoff, jobStatus, startingId, batchSize,
       );
       if (workItemIds.length > 0) {
         const numItemsDeleted = await deleteWorkItemsById(db, workItemIds);
@@ -51,12 +52,12 @@ async function deleteTerminalWorkItems(ctx: Context, notUpdatedForMinutes: numbe
 
 
 /**
- * Find workflow steps that are older than notUpdatedForMinutes and delete them.
- * @param notUpdatedForMinutes - upper limit on the duration since the last update
+ * Find workflow steps that are older than updatedAtCutoff and delete them.
+ * @param updatedAtCutoff - updatedAt cutoff time since the last update
  * @param jobStatus - a list of terminal job statuses
  * @returns Resolves when the request is complete
  */
-async function deleteTerminalWorkflowSteps(ctx: Context, notUpdatedForMinutes: number, jobStatus: JobStatus[]): Promise < void> {
+async function deleteTerminalWorkflowSteps(ctx: Context, updatedAtCutoff: Date, jobStatus: JobStatus[]): Promise < void> {
   let done = false;
   let startingId = 0;
   let totalDeleted = 0;
@@ -67,7 +68,7 @@ async function deleteTerminalWorkflowSteps(ctx: Context, notUpdatedForMinutes: n
   while (!done) {
     try {
       const workflowSteps = await getWorkflowStepIdsByJobUpdateAgeAndStatus(
-        db, notUpdatedForMinutes, jobStatus, startingId, batchSize,
+        db, updatedAtCutoff, jobStatus, startingId, batchSize,
       );
       if (workflowSteps.length > 0) {
         const numItemsDeleted = await deleteWorkflowStepsById(db, workflowSteps);
@@ -99,14 +100,15 @@ export class WorkReaper extends CronJob {
     const { logger } = ctx;
     logger.debug('Running');
     try {
+      const updatedAtCutoff = subMinutes(new Date(), env.reapableWorkAgeMinutes);
       await deleteTerminalWorkItems(
         ctx,
-        env.reapableWorkAgeMinutes,
+        updatedAtCutoff,
         terminalStates,
       );
       await deleteTerminalWorkflowSteps(
         ctx,
-        env.reapableWorkAgeMinutes,
+        updatedAtCutoff,
         terminalStates,
       );
     } catch (e) {
diff --git a/services/cron-service/test/work-reaper.ts b/services/cron-service/test/work-reaper.ts
@@ -1,5 +1,6 @@
 import { describe, it } from 'mocha';
 import { expect } from 'chai';
+import { subMinutes } from 'date-fns';
 import MockDate from 'mockdate';
 import { buildJob } from './helpers/jobs';
 import { buildWorkflowStep } from './helpers/workflow-steps';
@@ -53,18 +54,19 @@ describe('WorkReaper-related functions', function () {
   describe('.getWorkItemIdsByJobUpdateAgeAndStatus', function () {
     it('returns the work items and steps of jobs that have not been updated for n minutes', async function () {
       MockDate.set(newDate);
+      const updatedAtCutoff = subMinutes(new Date(), 60);
 
       const itemIds = await getWorkItemIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL, JobStatus.FAILED],
       );
       expect(itemIds.length).to.eql(3);
       expect(itemIds).to.have.same.members([1, 2, 3]);
 
       const stepIds = await getWorkflowStepIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL, JobStatus.FAILED],
       );
       expect(stepIds.length).to.eql(3);
@@ -75,17 +77,18 @@ describe('WorkReaper-related functions', function () {
 
     it('returns no work items / steps when they were created recently', async function () {
       MockDate.set(oldDate);
+      const updatedAtCutoff = subMinutes(new Date(), 60);
 
       const itemIds = await getWorkItemIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL],
       );
       expect(itemIds.length).to.eql(0);
 
       const stepIds = await getWorkflowStepIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL],
       );
       expect(stepIds.length).to.eql(0);
@@ -96,11 +99,12 @@ describe('WorkReaper-related functions', function () {
   describe('.deleteWorkItemsById', function () {
     it('deletes work items and steps by id', async function () {
       MockDate.set(newDate);
+      const updatedAtCutoff = subMinutes(new Date(), 60);
 
       // get the old work items
       const beforeDeletionItemIds = await getWorkItemIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL, JobStatus.FAILED],
       );
       expect(beforeDeletionItemIds.length).to.eql(3);
@@ -110,15 +114,15 @@ describe('WorkReaper-related functions', function () {
       await deleteWorkItemsById(this.trx, beforeDeletionItemIds);
       const afterDeletionItemIds = await getWorkItemIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL, JobStatus.FAILED],
       );
       expect(afterDeletionItemIds.length).to.eql(0);
 
       // get the old steps
       const beforeDeletionStepIds = await getWorkflowStepIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL, JobStatus.FAILED],
       );
       expect(beforeDeletionStepIds.length).to.eql(3);
@@ -128,7 +132,7 @@ describe('WorkReaper-related functions', function () {
       await deleteWorkflowStepsById(this.trx, beforeDeletionItemIds);
       const afterDeletionStepIds = await getWorkflowStepIdsByJobUpdateAgeAndStatus(
         this.trx,
-        60,
+        updatedAtCutoff,
         [JobStatus.CANCELED, JobStatus.SUCCESSFUL, JobStatus.FAILED],
       );
       expect(afterDeletionStepIds.length).to.eql(0);
diff --git a/services/harmony/app/models/work-item.ts b/services/harmony/app/models/work-item.ts
@@ -525,11 +525,11 @@ export async function getWorkItemsByJobIdAndStepIndex(
 }
 
 /**
- * Get all work item ids associated with jobs that haven't been updated for a
- * certain amount of minutes and that have a particular JobStatus
+ * Get all work item ids associated with jobs that haven't been updated
+ * since updatedAtCutoff and that have a particular JobStatus
  * @param tx - the transaction to use for querying
- * @param notUpdatedForMinutes - jobs with updateAt older than notUpdatedForMinutes ago
- * will be joined with the returned work items
+ * @param updatedAtCutoff - jobs with updatedAt older than updatedAtCutoff will be
+ * joined with the returned work items
  * @param jobStatus - only jobs with this status will be joined
  * @param startingId - the work item id to begin the query with, i.e. query work items with id greater than startingId
  * @param batchSize - the batch size
@@ -538,16 +538,15 @@ export async function getWorkItemsByJobIdAndStepIndex(
  */
 export async function getWorkItemIdsByJobUpdateAgeAndStatus(
   tx: Transaction,
-  notUpdatedForMinutes: number,
+  updatedAtCutoff: Date,
   jobStatus: JobStatus[],
   startingId = 0,
   batchSize = 2000,
 ): Promise<number[]> {
-  const pastDate = subMinutes(new Date(), notUpdatedForMinutes);
   const workItemIds = (await tx(`${WorkItem.table} as w`)
     .innerJoin(Job.table, 'w.jobID', '=', `${Job.table}.jobID`)
     .select(['w.id'])
-    .where(`${Job.table}.updatedAt`, '<', pastDate)
+    .where(`${Job.table}.updatedAt`, '<', updatedAtCutoff)
     .whereIn(`${Job.table}.status`, jobStatus)
     .where('w.id', '>', startingId)
     .orderBy('w.id', 'asc')
diff --git a/services/harmony/app/models/workflow-steps.ts b/services/harmony/app/models/workflow-steps.ts
@@ -1,6 +1,5 @@
 /* eslint-disable @typescript-eslint/dot-notation */
 import env from '../util/env';
-import { subMinutes } from 'date-fns';
 import _ from 'lodash';
 import { Transaction } from '../util/db';
 import { Job, JobStatus } from './job';
@@ -234,7 +233,7 @@ export async function getWorkflowStepByJobIdServiceId(
  * Get all workflow step ids associated with jobs that haven't been updated for a
  * certain amount of minutes and that have a particular JobStatus
  * @param tx - the transaction to use for querying
- * @param notUpdatedForMinutes - jobs with updateAt older than notUpdatedForMinutes ago will be
+ * @param updatedAtCutoff - jobs with updatedAt older than updatedAtCutoff will be
  * joined with the returned workflow steps
  * @param jobStatus - only jobs with this status will be joined
  * @param startingId - the workflow step id to begin the query with, i.e. query workflow steps
@@ -245,16 +244,15 @@ export async function getWorkflowStepByJobIdServiceId(
  */
 export async function getWorkflowStepIdsByJobUpdateAgeAndStatus(
   tx: Transaction,
-  notUpdatedForMinutes: number,
+  updatedAtCutoff: Date,
   jobStatus: JobStatus[],
   startingId = 0,
   batchSize = 2000,
 ): Promise<number[]> {
-  const pastDate = subMinutes(new Date(), notUpdatedForMinutes);
   const workflowStepIds = (await tx(WorkflowStep.table)
     .innerJoin(Job.table, `${WorkflowStep.table}.jobID`, '=', `${Job.table}.jobID`)
     .select([`${WorkflowStep.table}.id`])
-    .where(`${Job.table}.updatedAt`, '<', pastDate)
+    .where(`${Job.table}.updatedAt`, '<', updatedAtCutoff)
     .whereIn(`${Job.table}.status`, jobStatus)
     .where(`${WorkflowStep.table}.id`, '>', startingId)
     .orderBy(`${WorkflowStep.table}.id`, 'asc')
