aboviq/emigrate
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(cli): add graceful process abort

Browse source 
Using an AbortSignal and Promise.race we abandon running migrations that take longer to complete after the process is aborted than the given abortRespite period
This commit is contained in:
Joakim Carlstein 2024-01-22 10:53:01 +01:00 • committed by Joakim Carlstein
parent ce15648251
commit a4da353d5a
17 changed files with 378 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

31
packages/cli/src/migration-runner.ts
View file

@ -18,6 +18,8 @@ type MigrationRunnerParameters = {
limit?: number;
from?: string;
to?: string;
abortSignal?: AbortSignal;
abortRespite?: number;
reporter: EmigrateReporter;
storage: Storage;
migrations: Array<MigrationMetadata | MigrationMetadataFinished>;
@ -30,6 +32,8 @@ export const migrationRunner = async ({
limit,
from,
to,
abortSignal,
abortRespite,
reporter,
storage,
migrations,
@ -43,6 +47,22 @@ export const migrationRunner = async ({
let skip = false;
abortSignal?.addEventListener(
'abort',
() => {
skip = true;
reporter.onAbort?.(toError(abortSignal.reason))?.then(
() => {
/* noop */
},
() => {
/* noop */
},
);
},
{ once: true },
);
for await (const migration of migrations) {
if (isFinishedMigration(migration)) {
skip ||= migration.status === 'failed' || migration.status === 'skipped';
@ -89,7 +109,7 @@ export const migrationRunner = async ({
const [lockedMigrations, lockError] = dry
? [migrationsToLock]
: await exec(async () => storage.lock(migrationsToLock));
: await exec(async () => storage.lock(migrationsToLock), { abortSignal, abortRespite });
if (lockError) {
for (const migration of migrationsToLock) {
@ -167,7 +187,7 @@ export const migrationRunner = async ({
const start = hrtime();
const [, migrationError] = await exec(async () => execute(migration));
const [, migrationError] = await exec(async () => execute(migration), { abortSignal, abortRespite });
const duration = getDuration(start);
@ -194,7 +214,9 @@ export const migrationRunner = async ({
}
}
const [, unlockError] = dry ? [] : await exec(async () => storage.unlock(lockedMigrations ?? []));
const [, unlockError] = dry
? []
: await exec(async () => storage.unlock(lockedMigrations ?? []), { abortSignal, abortRespite });
// eslint-disable-next-line unicorn/no-array-callback-reference
const firstFailed = finishedMigrations.find(isFailedMigration);
@ -204,7 +226,8 @@ export const migrationRunner = async ({
: firstFailed
? MigrationRunError.fromMetadata(firstFailed)
: undefined;
const error = unlockError ?? firstError ?? lockError;
const error =
unlockError ?? firstError ?? lockError ?? (abortSignal?.aborted ? toError(abortSignal.reason) : undefined);
await reporter.onFinished?.(finishedMigrations, error);