paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 37 Packages Projects Releases Wiki Activity

Merge pull request #20283 from alecjacobs5401/ajacobs/task-schedule-timing-upsert

Browse Source 
fix(backend-tasks): prevent immediately triggering tasks out of band
This commit is contained in:
Fredrik Adelöw
2023-10-03 09:10:59 +02:00
committed by GitHub
parent 0507326be6 06451c88d1
commit 2fe23ff546
3 changed files with 157 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/nine-garlics-attend.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/backend-tasks': patch
---
Fix bug where backend tasks that are defined with HumanDuration are immediately triggered on application startup
packages/backend-tasks/src/tasks/TaskWorker.test.ts
+125 -2
View File
@@ -16,7 +16,7 @@
import { getVoidLogger } from '@backstage/backend-common';
import { TestDatabases } from '@backstage/backend-test-utils';
import { Duration } from 'luxon';
import { Duration, DateTime } from 'luxon';
import waitForExpect from 'wait-for-expect';
import { migrateBackendTasks } from '../database/migrateBackendTasks';
import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';
@@ -338,7 +338,7 @@ describe('TaskWorker', () => {
);
it.each(databases.eachSupportedId())(
'next_run_start_at is always the min between schedule changes, %p',
'next_run_start_at is always the min between schedule changes from cron frequency, %p',
async databaseId => {
const knex = await databases.init(databaseId);
await migrateBackendTasks(knex);
@@ -376,4 +376,127 @@ describe('TaskWorker', () => {
await knex.destroy();
},
);
it.each(databases.eachSupportedId())(
'next_run_start_at is always the min between schedule changes when using human duration frequency, %p',
async databaseId => {
const knex = await databases.init(databaseId);
await migrateBackendTasks(knex);
const fn = jest.fn(
async () => new Promise<void>(resolve => setTimeout(resolve, 50)),
);
const initialSettings: TaskSettingsV2 = {
version: 2,
cadence: 'PT120M',
timeoutAfterDuration: 'PT1M',
};
const worker = new TaskWorker('task99', fn, knex, logger);
await worker.persistTask(initialSettings);
// replicate task running, sets next_run_start_at based on cadence
await worker.tryClaimTask('ticket', initialSettings);
await worker.tryReleaseTask('ticket', initialSettings);
// grab initial row for comparisons later
const rowAfterClaimAndRelease = (
await knex<DbTasksRow>(DB_TASKS_TABLE)
)[0];
const settings: TaskSettingsV2 = {
...initialSettings,
cadence: 'PT60M',
};
await worker.persistTask(settings);
const row1 = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
const rowAfterClaimAndReleaseNextStartAt = DateTime.fromJSDate(
new Date(rowAfterClaimAndRelease.next_run_start_at),
);
const row1NextStartAt = DateTime.fromJSDate(
new Date(row1.next_run_start_at),
);
const now = DateTime.now();
expect(
rowAfterClaimAndReleaseNextStartAt.diff(row1NextStartAt).as('minutes'),
).toBeCloseTo(60, 1); // ensure that next start at is sooner than initial by one hour
expect(row1NextStartAt.diff(now).as('minutes')).toBeCloseTo(60, 1); // ensure that next start at is later than now by one hour
expect(
rowAfterClaimAndReleaseNextStartAt.diff(now).as('minutes'),
).toBeCloseTo(120, 1);
const settings2 = {
...settings,
};
await worker.persistTask(settings2);
const row2 = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
expect(row2.next_run_start_at).toStrictEqual(row1.next_run_start_at);
await knex.destroy();
},
);
it.each(databases.eachSupportedId())(
'next_run_start_at is always the min between schedule changes when using human duration frequency with initial start delay, %p',
async databaseId => {
const knex = await databases.init(databaseId);
await migrateBackendTasks(knex);
const fn = jest.fn(
async () => new Promise<void>(resolve => setTimeout(resolve, 50)),
);
const initialSettings: TaskSettingsV2 = {
version: 2,
cadence: 'PT120M',
initialDelayDuration: 'PT2M',
timeoutAfterDuration: 'PT1M',
};
const worker = new TaskWorker('task99', fn, knex, logger);
await worker.persistTask(initialSettings);
// replicate task running, sets next_run_start_at based on cadence
await worker.tryClaimTask('ticket', initialSettings);
await worker.tryReleaseTask('ticket', initialSettings);
// grab initial row for comparisons later
const rowAfterClaimAndRelease = (
await knex<DbTasksRow>(DB_TASKS_TABLE)
)[0];
const settings: TaskSettingsV2 = {
...initialSettings,
cadence: 'PT60M',
};
await worker.persistTask(settings);
const row1 = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
const rowAfterClaimAndReleaseNextStartAt = DateTime.fromJSDate(
new Date(rowAfterClaimAndRelease.next_run_start_at),
);
const row1NextStartAt = DateTime.fromJSDate(
new Date(row1.next_run_start_at),
);
const now = DateTime.now();
expect(
rowAfterClaimAndReleaseNextStartAt.diff(row1NextStartAt).as('minutes'),
).toBeCloseTo(62, 1); // ensure that next start at is sooner than initial by one hour, plus the 2 minute delay (set my tryReleaseTask)
expect(row1NextStartAt.diff(now).as('minutes')).toBeCloseTo(60, 1); // ensure that next start at is later than now by one hour (2 minute delay doesn't take effect here)
expect(
rowAfterClaimAndReleaseNextStartAt.diff(now).as('minutes'),
).toBeCloseTo(122, 1); // includes 2 minute start delay (which is persisted from tryReleaseTask)
const settings2 = {
...settings,
};
await worker.persistTask(settings2);
const row2 = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
expect(row2.next_run_start_at).toStrictEqual(row1.next_run_start_at);
await knex.destroy();
},
);
});
packages/backend-tasks/src/tasks/TaskWorker.ts
+27 -20
View File
@@ -165,27 +165,26 @@ export class TaskWorker {
const isCron = !settings?.cadence.startsWith('P');
let startAt: Knex.Raw;
let startAt: Knex.Raw | undefined;
let nextStartAt: Knex.Raw | undefined;
if (settings.initialDelayDuration) {
startAt = nowPlus(
Duration.fromISO(settings.initialDelayDuration),
this.knex,
);
} else if (isCron) {
}
if (isCron) {
const time = new CronTime(settings.cadence)
.sendAt()
.minus({ seconds: 1 }) // immediately, if "* * * * * *"
.toUTC();
if (this.knex.client.config.client.includes('sqlite3')) {
startAt = this.knex.raw('datetime(?)', [time.toISO()]);
} else if (this.knex.client.config.client.includes('mysql')) {
startAt = this.knex.raw(`?`, [time.toSQL({ includeOffset: false })]);
} else {
startAt = this.knex.raw(`?`, [time.toISO()]);
}
nextStartAt = this.nextRunAtRaw(time);
startAt ||= nextStartAt;
} else {
startAt = this.knex.fn.now();
startAt ||= this.knex.fn.now();
nextStartAt = nowPlus(Duration.fromISO(settings.cadence), this.knex);
}
this.logger.debug(`task: ${this.taskId} configured to run at: ${startAt}`);
@@ -206,7 +205,12 @@ export class TaskWorker {
settings_json: settingsJson,
next_run_start_at: this.knex.raw(
`CASE WHEN ?? < ?? THEN ?? ELSE ?? END`,
[startAt, 'next_run_start_at', startAt, 'next_run_start_at'],
[
nextStartAt,
'next_run_start_at',
nextStartAt,
'next_run_start_at',
],
),
}
: {
@@ -214,9 +218,9 @@ export class TaskWorker {
next_run_start_at: this.knex.raw(
`CASE WHEN ?? < ?? THEN ?? ELSE ?? END`,
[
'excluded.next_run_start_at',
nextStartAt,
`${DB_TASKS_TABLE}.next_run_start_at`,
'excluded.next_run_start_at',
nextStartAt,
`${DB_TASKS_TABLE}.next_run_start_at`,
],
),
@@ -308,13 +312,7 @@ export class TaskWorker {
const time = new CronTime(settings.cadence).sendAt().toUTC();
this.logger.debug(`task: ${this.taskId} will next occur around ${time}`);
if (this.knex.client.config.client.includes('sqlite3')) {
nextRun = this.knex.raw('datetime(?)', [time.toISO()]);
} else if (this.knex.client.config.client.includes('mysql')) {
nextRun = this.knex.raw(`?`, [time.toSQL({ includeOffset: false })]);
} else {
nextRun = this.knex.raw(`?`, [time.toISO()]);
}
nextRun = this.nextRunAtRaw(time);
} else {
const dt = Duration.fromISO(settings.cadence).as('seconds');
this.logger.debug(
@@ -351,4 +349,13 @@ export class TaskWorker {
return rows === 1;
}
private nextRunAtRaw(time: DateTime): Knex.Raw {
if (this.knex.client.config.client.includes('sqlite3')) {
return this.knex.raw('datetime(?)', [time.toISO()]);
} else if (this.knex.client.config.client.includes('mysql')) {
return this.knex.raw(`?`, [time.toSQL({ includeOffset: false })]);
}
return this.knex.raw(`?`, [time.toISO()]);
}
}