mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-20 22:16:26 +00:00
refactor(scheduling): take a cron lease in one conditional upsert
acquire() made sure a row existed and then raced on a conditional UPDATE, two statements the registrar wrapped in a transaction of its own. lint:tx cannot follow a repository reached through a local variable, so it reported the method. One INSERT ... ON CONFLICT DO UPDATE ... WHERE now takes a free lease, renews the holder's and leaves anyone else's alone, atomically on SQLite and on Postgres, and needs no transaction.
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import type { SchedulerLeases } from '../entities/SchedulerLeases.entity';
|
||||
import { TrekRepository } from './_shared/trek-repository';
|
||||
import type { DB } from '../kysely/db';
|
||||
|
||||
/**
|
||||
* `scheduler_leases`: which process may run a cron job's tick, and until when
|
||||
@@ -9,22 +10,23 @@ import { TrekRepository } from './_shared/trek-repository';
|
||||
export class SchedulerLeasesRepository extends TrekRepository<SchedulerLeases> {
|
||||
/**
|
||||
* Take (or renew) the lease on `name` for `owner` until `until`, if it is
|
||||
* free at `now`. The decision is one conditional UPDATE, so of two
|
||||
* processes asking at the same moment exactly one gets a changed row; the
|
||||
* insert before it only makes sure there is a row to race on. True when
|
||||
* `owner` holds the lease afterwards. Two statements, so the caller runs it
|
||||
* inside `UnitOfWork.transactional` (CronRegistrarService does).
|
||||
* free at `now`. One statement: the insert takes a lease nobody has held
|
||||
* yet, and on a conflict the update only applies while the lease has run out
|
||||
* or already belongs to `owner`. Of two processes asking at the same moment
|
||||
* exactly one gets a row written. True when `owner` holds the lease afterwards.
|
||||
*/
|
||||
async acquire(name: string, owner: string, now: number, until: number): Promise<boolean> {
|
||||
await this.upsert(
|
||||
{ name, owner, expires_at: 0 },
|
||||
{ onConflictFields: ['name'], onConflictAction: 'ignore' },
|
||||
);
|
||||
const changed = await this.nativeUpdate(
|
||||
{ name, $or: [{ expires_at: { $lte: now } }, { owner }] },
|
||||
{ owner, expires_at: until },
|
||||
);
|
||||
return changed > 0;
|
||||
const result = await this.kysely<Pick<DB, 'scheduler_leases'>>()
|
||||
.insertInto('scheduler_leases')
|
||||
.values({ name, owner, expires_at: until })
|
||||
.onConflict((oc) =>
|
||||
oc
|
||||
.column('name')
|
||||
.doUpdateSet({ owner, expires_at: until })
|
||||
.where((eb) => eb.or([eb('scheduler_leases.expires_at', '<=', now), eb('scheduler_leases.owner', '=', owner)])),
|
||||
)
|
||||
.executeTakeFirst();
|
||||
return (result.numInsertedOrUpdatedRows ?? 0n) > 0n;
|
||||
}
|
||||
|
||||
/** Move the expiry of a lease `owner` still holds; false when somebody else has it now. */
|
||||
|
||||
@@ -7,7 +7,6 @@ import { MikroORM } from '@mikro-orm/core';
|
||||
import { readEnv } from '../../app-config';
|
||||
import { RuntimeEnvService } from '../app-config/runtime-env.service';
|
||||
import { withRequestContext } from '../database/request-context';
|
||||
import { UnitOfWork } from '../database/unit-of-work';
|
||||
import { logError } from '../audit/audit-log.logger';
|
||||
import { traceEntry, wasTraced } from '../audit/entry-trace.logger';
|
||||
import { SchedulerLeases } from '../../db/entities/SchedulerLeases.entity';
|
||||
@@ -162,14 +161,8 @@ export class CronRegistrarService implements OnApplicationShutdown {
|
||||
onTick: () => void | Promise<void>,
|
||||
): Promise<'ran' | 'skipped'> {
|
||||
const leases = orm.em.getRepository(SchedulerLeases);
|
||||
// acquire() is two statements (make sure the row exists, then the
|
||||
// conditional UPDATE that decides the race), so it runs as one unit of
|
||||
// work like every other multi-statement write. A UnitOfWork over the ORM
|
||||
// this tick already has, rather than an injected one: the partial e2e
|
||||
// graphs build this service without the global OrmModule.
|
||||
const acquired = await new UnitOfWork(orm.em).transactional(() =>
|
||||
leases.acquire(name, LEASE_OWNER, Date.now(), Date.now() + LEASE_TTL_MS),
|
||||
);
|
||||
// acquire() is one conditional upsert, so the race needs no transaction.
|
||||
const acquired = await leases.acquire(name, LEASE_OWNER, Date.now(), Date.now() + LEASE_TTL_MS);
|
||||
if (!acquired) return 'skipped';
|
||||
const leaseFailed = (err: unknown) =>
|
||||
logError(`Cron job "${name}": lease update failed: ${err instanceof Error ? err.message : String(err)}`);
|
||||
|
||||
Reference in New Issue
Block a user