mirror of
https://github.com/civitai/civitai.git
synced 2026-09-20 22:08:18 +08:00
fix(jobs): stop delete-old-training-data running four concurrent passes a day
`delete-old-training-data` registered with no options object, so it inherited `createJob`'s five-minute `lockExpiration` and the default release-the-lock-on- disconnect. It walks the whole outstanding set of expired training-data files serially — one object delete plus one row update per file, each awaited — and no pass has been observed to finish inside the caller's timeout. The consequence, observed in production over several consecutive days: the daily trigger is abandoned just under an hour in, the route's close handler releases the lock there and then, and each of the scheduler's three retries acquires the freed lock and starts another concurrent pass over the same rows. One once-daily cron, four full-length attempts, all re-issuing the same work. This applies the pattern `process-csam.ts` and `process-huggingface-imports.ts` already use: a real `lockExpiration` plus `keepLockOnDisconnect: true`. Neither half works alone — a longer lock is discarded at the disconnect before it can govern anything, and the opt-in alone leaves the lock expiring five minutes in, long before the first retry. The lock is six hours: 1.5x the retry ladder's full wall-clock span, which is the floor the sizing argument rests on, and a quarter of the 24h cron period, so a run that is alive but wedged can never delay the next scheduled run. Both operands are named constants and the reasoning — including what the held lock costs, and the residual case a lock cannot close — is on them. Deliberately NOT changed: the handler still takes no `jobContext` and still does not poll `checkIfCanceled`. A pass that outlives the caller's timeout finishes only because it keeps running past the disconnect; making it cancellable would kill it at every attempt's timeout and complete nothing. Same reasoning as `process-csam`. Tests: `delete-old-training-data-lock.test.ts` pins both options, both directions of the sizing argument against their own pinned yardsticks, drives the real `createDisconnectHandler` with this job's own options against a non-opted control job, and pins the seam neither of those owns — that the job is still in the run-jobs route's dispatch list, without which every other assertion stays green while the options reach nothing. Verified: 8 mutants, each killed by its own named assertion — including the pre-fix state (no options object), dropping either option alone, shrinking the lock to just over the caller's timeout, shrinking the yardstick to re-satisfy the ratio, growing the lock toward the cron period, changing the cron it is compared against, and unregistering the job from the route. Typecheck error set byte- identical to the base branch's.
This commit is contained in:
ZacxDev
@@ -0,0 +1,146 @@
|
||||
import { readFileSync } from 'fs';
|
||||
import path from 'path';
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
/**
|
||||
* The run lock on `delete-old-training-data`.
|
||||
*
|
||||
* THE DEFECT THIS PINS. The job registered with no options object at all, so it inherited
|
||||
* `createJob`'s five-minute `lockExpiration` and the default release-on-disconnect. It walks the
|
||||
* whole outstanding set of expired training-data files serially and no pass has been observed to
|
||||
* finish inside the caller's timeout — so in production the caller hung up, the route's close
|
||||
* handler released the lock, and each of the scheduler's retries acquired it and started another
|
||||
* concurrent pass over the same rows. Measured over several consecutive days: one daily trigger,
|
||||
* four full-length attempts.
|
||||
*
|
||||
* WHY BOTH OPTIONS OR NEITHER. A longer `lockExpiration` alone is inert — the disconnect throws
|
||||
* the budget away before it can govern anything. `keepLockOnDisconnect` alone leaves the lock
|
||||
* expiring five minutes in, long before the first retry. Every case below therefore exists to fail
|
||||
* when EITHER half is reverted.
|
||||
*
|
||||
* The generic both-arms contract for the disconnect handler, and the check that the route still
|
||||
* installs it in a position where it can fire, live in `job-disconnect-lock.test.ts`.
|
||||
*/
|
||||
|
||||
// The job module reaches for the S3 client at call time only, but stubbing the module keeps the
|
||||
// AWS SDK out of a suite that never runs the handler — none of the cases below invoke it.
|
||||
vi.mock('~/utils/s3-utils', () => ({
|
||||
deleteObject: vi.fn(),
|
||||
parseKey: vi.fn(() => ({ key: 'k', bucket: 'b' })),
|
||||
}));
|
||||
|
||||
import {
|
||||
DELETE_OLD_TRAINING_DATA_LOCK_SECONDS,
|
||||
DELETE_OLD_TRAINING_DATA_RETRY_LADDER_SECONDS,
|
||||
deleteOldTrainingData,
|
||||
} from '~/server/jobs/delete-old-training-data';
|
||||
import { createDisconnectHandler, createJob } from '~/server/jobs/job';
|
||||
|
||||
const RUN_JOBS_ROUTE = path.resolve(
|
||||
__dirname,
|
||||
'../../../pages/api/webhooks/run-jobs/[[...run]].ts'
|
||||
);
|
||||
|
||||
function harness(options: Parameters<typeof createDisconnectHandler>[0]) {
|
||||
const cancel = vi.fn(async () => undefined);
|
||||
const release = vi.fn(async () => undefined);
|
||||
return { cancel, release, handler: createDisconnectHandler(options, { cancel }, { release }) };
|
||||
}
|
||||
|
||||
describe('delete-old-training-data asks for a lock that outlasts the caller’s retries', () => {
|
||||
it('uses the named constant, not createJob’s inherited default', () => {
|
||||
const inherited = createJob('probe-delete-old-training-data', '5 11 * * *', async () => void 0);
|
||||
|
||||
expect(deleteOldTrainingData.options.lockExpiration).toBe(
|
||||
DELETE_OLD_TRAINING_DATA_LOCK_SECONDS
|
||||
);
|
||||
// The non-vacuous half. An "override" that is not actually longer than the inherited default
|
||||
// leaves the duplicate-pass hazard exactly where it was, and reads in review like a fix.
|
||||
expect(DELETE_OLD_TRAINING_DATA_LOCK_SECONDS).toBeGreaterThan(inherited.options.lockExpiration);
|
||||
});
|
||||
|
||||
it('🔴 clears the RETRY LADDER it is sized against, with headroom', () => {
|
||||
// 🔴 The yardstick is pinned to its own literal FIRST, and that is what makes the ratio below
|
||||
// a guard at all. Both constants live in the same module, so with only the lock pinned,
|
||||
// shrinking the ladder satisfies the ratio for any lock value — and the exact mutant this case
|
||||
// exists to kill, a lock cut back to just over the caller's one-hour timeout, would survive a
|
||||
// one-token edit to the constant it is supposedly measured against. Re-measuring the ladder
|
||||
// must land here and force the lock to be re-argued.
|
||||
expect(DELETE_OLD_TRAINING_DATA_RETRY_LADDER_SECONDS).toBe(4 * 60 * 60);
|
||||
expect(DELETE_OLD_TRAINING_DATA_LOCK_SECONDS).toBeGreaterThanOrEqual(
|
||||
1.5 * DELETE_OLD_TRAINING_DATA_RETRY_LADDER_SECONDS
|
||||
);
|
||||
});
|
||||
|
||||
it('stays far enough below the cron period that a wedged run cannot delay a scheduled run', () => {
|
||||
// Asserting the SCHEDULE as well as the number is what makes this a relationship rather than
|
||||
// two unrelated literals: if the cron is ever made faster, this fails instead of silently
|
||||
// comparing the lock against a period the job no longer runs at.
|
||||
expect(deleteOldTrainingData.cron).toBe('5 11 * * *'); // once daily
|
||||
const cronPeriodSeconds = 24 * 60 * 60;
|
||||
|
||||
// The direction `keepLockOnDisconnect` turned into a real cost: an alive-but-wedged run now
|
||||
// holds this lock for its full duration instead of losing it at the disconnect. Growing the
|
||||
// value toward the cron period should have to be argued here, not discovered in production.
|
||||
expect(DELETE_OLD_TRAINING_DATA_LOCK_SECONDS).toBeLessThanOrEqual(cronPeriodSeconds / 4);
|
||||
});
|
||||
});
|
||||
|
||||
describe('delete-old-training-data’s lock survives the caller hanging up', () => {
|
||||
// 🔴 THE HALF THAT MAKES THE CONSTANT ABOVE MEAN ANYTHING. These drive the REAL factory the
|
||||
// route installs, using this job's own options object, so they fail if the opt-in is dropped
|
||||
// from the job OR broken in the factory.
|
||||
|
||||
it('a disconnect cancels the context but leaves the lock held', () => {
|
||||
expect(deleteOldTrainingData.options.keepLockOnDisconnect).toBe(true);
|
||||
});
|
||||
|
||||
it('drives the real handler: cancel fires, release does not', async () => {
|
||||
const { cancel, release, handler } = harness(deleteOldTrainingData.options);
|
||||
|
||||
await handler();
|
||||
|
||||
// Cancel still fires. The flag is about the LOCK, not about whether the context is told to
|
||||
// stop — dropping the cancel would change every cancellation-aware job that adopts this.
|
||||
expect(cancel).toHaveBeenCalledTimes(1);
|
||||
expect(release).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('CONTROL: a job that does not opt in still releases on a disconnect', async () => {
|
||||
// Same factory, same harness, a job built the ordinary way — so a green above is a fact about
|
||||
// THIS job's options rather than about a harness that never calls `release` at all.
|
||||
const control = createJob(
|
||||
'probe-delete-old-training-data-control',
|
||||
'5 11 * * *',
|
||||
async () => undefined
|
||||
);
|
||||
const { cancel, release, handler } = harness(control.options);
|
||||
|
||||
await handler();
|
||||
|
||||
expect(cancel).toHaveBeenCalledTimes(1);
|
||||
expect(release).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('the options above can actually reach a run', () => {
|
||||
it('the job is still registered in the run-jobs route’s dispatch list', () => {
|
||||
// 🔴 WHAT THIS IS: a source read, because importing that route pulls in every job in the
|
||||
// application. WHAT IT IS WORTH: the route dispatches by looking the requested name up in its
|
||||
// `jobs` array and reads `options` off whatever it finds. Drop this job from that array and
|
||||
// every assertion above stays green while the job never runs at all — the seam neither the
|
||||
// options tests nor the handler tests own. It cannot tell you the lookup behaves correctly.
|
||||
const source = readFileSync(RUN_JOBS_ROUTE, 'utf8');
|
||||
|
||||
expect(source).toContain(
|
||||
"import { deleteOldTrainingData } from '~/server/jobs/delete-old-training-data';"
|
||||
);
|
||||
// Membership in the exported array, not merely the import — an unused import type-checks.
|
||||
const jobsArray = source.slice(
|
||||
source.indexOf('export const jobs: Job[] = ['),
|
||||
source.indexOf('const log = createLogger')
|
||||
);
|
||||
expect(jobsArray).not.toHaveLength(0);
|
||||
expect(jobsArray).toContain('deleteOldTrainingData,');
|
||||
});
|
||||
});
|
||||
@@ -13,6 +13,64 @@ type OldTrainingRow = {
|
||||
url: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Wall-clock span over which the caller's retries of ONE day's trigger can still arrive.
|
||||
*
|
||||
* The external scheduler holds the trigger request open under a client-side timeout and retries a
|
||||
* fixed number of times when that fires. Observed in production over several consecutive days: the
|
||||
* request is abandoned just under an hour in, three further attempts follow at roughly hourly
|
||||
* spacing, and the ladder is finished about four hours after the day's first trigger. The last
|
||||
* retry is therefore POSTed about three hours in — that POST is the latest moment at which a freed
|
||||
* lock can be claimed by a competing run of this job.
|
||||
*
|
||||
* Exported so the lock below is sized against something checkable instead of being a bare literal.
|
||||
*
|
||||
* 🔴 THIS IS A PROPERTY OF THE SCHEDULER'S CONFIGURATION, NOT OF THIS JOB. Raising that
|
||||
* client-side timeout or its retry count lengthens this ladder, and the lock has to be re-argued
|
||||
* rather than silently left behind.
|
||||
*/
|
||||
export const DELETE_OLD_TRAINING_DATA_RETRY_LADDER_SECONDS = 4 * 60 * 60;
|
||||
|
||||
/**
|
||||
* How long this job may hold its run lock, overriding `createJob`'s five-minute default.
|
||||
*
|
||||
* WHY AN OVERRIDE AT ALL. This job walks the whole outstanding set of expired training-data files
|
||||
* serially — one object delete plus one row update per file, each awaited — and no pass has been
|
||||
* observed to finish inside the caller's timeout. At the inherited 300s the lock lapses a few
|
||||
* minutes in, so each of the retries above acquires it and starts a second, third and fourth
|
||||
* concurrent pass over the same rows, re-issuing deletes the earlier passes are still working
|
||||
* through. That is not hypothetical: it is the steady state this constant was added to end.
|
||||
*
|
||||
* WHY IT DOES NOT WORK WITHOUT `keepLockOnDisconnect`. The run-jobs route's close handler releases
|
||||
* the lock the moment the caller hangs up (`createDisconnectHandler`). A pass that outlives the
|
||||
* caller's timeout loses its socket first and its lock with it, so the whole budget below is
|
||||
* discarded precisely in the case it was sized for. The two options are one mitigation; either one
|
||||
* alone leaves the duplicate passes exactly where they were.
|
||||
*
|
||||
* 🔴 AND DO NOT "COMPLETE" THIS BY WIRING `checkIfCanceled` INTO THE LOOP. The handler takes no
|
||||
* `jobContext` deliberately. A pass that outlives the caller's timeout finishes only because it
|
||||
* keeps running past the disconnect; a cancelling version would be killed at every attempt's
|
||||
* timeout and — on the evidence above, where no attempt has been seen to finish inside it — would
|
||||
* never complete a pass at all. Same reasoning as `process-csam.ts`.
|
||||
*
|
||||
* WHY THIS VALUE — BOTH DIRECTIONS.
|
||||
* The floor is the retry ladder: the lock has to still be held when the last retry POSTs, or that
|
||||
* retry starts the competing pass this exists to prevent. Six hours is 1.5× the ladder's full
|
||||
* wall-clock span, which is the least headroom the argument can be read as claiming.
|
||||
* The ceiling is the cron period. `keepLockOnDisconnect` makes a long hold a real cost — a run
|
||||
* that is alive but wedged now holds this lock for its full duration instead of losing it at the
|
||||
* disconnect — so the value stays a quarter of the 24h period, far enough below it that a wedged
|
||||
* run can never delay the next SCHEDULED run. (A pod that dies pays nothing either way: the redis
|
||||
* key carries a ~10s TTL refreshed by an in-process interval, so a dead pod's lock lapses within
|
||||
* seconds. See `JobOptions.keepLockOnDisconnect`.)
|
||||
*
|
||||
* 🔴 RESIDUAL, STATED SO IT IS NOT MISTAKEN FOR COVERED: a pass that outran the cron period would
|
||||
* overlap the NEXT day's run, and no value below that period can close that — a lock long enough
|
||||
* to cover it would also block legitimate daily runs. The fix for that case is bounding the work
|
||||
* done per pass, not growing this number.
|
||||
*/
|
||||
export const DELETE_OLD_TRAINING_DATA_LOCK_SECONDS = 6 * 60 * 60;
|
||||
|
||||
export const deleteOldTrainingData = createJob(
|
||||
'delete-old-training-data',
|
||||
'5 11 * * *',
|
||||
@@ -108,5 +166,12 @@ export const deleteOldTrainingData = createJob(
|
||||
});
|
||||
|
||||
return { status: 'ok' };
|
||||
},
|
||||
// Both of these, or neither: the lock has to be long enough to outlast the caller's retries AND
|
||||
// has to survive the disconnect that precedes them. See the two constants above for the sizing
|
||||
// argument and for what the held lock costs.
|
||||
{
|
||||
lockExpiration: DELETE_OLD_TRAINING_DATA_LOCK_SECONDS,
|
||||
keepLockOnDisconnect: true,
|
||||
}
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user