fix(workouts): serialise record detection with a workout delete

The workout-slot detection reads the best workout and writes its record
inside one transaction under a per-user advisory lock, and the workout
DELETE takes the same lock before removing records and the workout. A
delete can no longer land between the read and the write and leave a
record for a workout that is gone.

The DELETE's match for a legacy record without an external id is scoped
to the slots of the deleted workout's sport, so a record another sport
set at the same instant stays.
This commit is contained in:
Marc-André Bombeck committed 2026-10-03 17:54:04 +02:00
1 parent ab081d2a85
commit 6bb65824bf
4 files changed
+140 -58

No files matched your search

+21 -4
View File
@@ -36,6 +36,10 @@ import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { enqueuePrDetection } from "@/lib/jobs/pr-detection";
import {
lockPersonalRecordsForUser,
workoutPrSlotsForSport,
} from "@/lib/personal-records/pr-detection-worker";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import { requireModuleEnabled } from "@/lib/modules/gate";
import { aiCapabilityToServe } from "@/lib/ai/capabilities/gate";
@@ -380,7 +384,9 @@ export const GET = apiHandler(
* (`module-route-gate-inventory.test.ts`).
*
* The workout table carries no tombstone, so the row is removed outright;
* route, samples, insight and claim rows cascade with it. A personal record
* route, samples and insight rows cascade with it. The insight's generation
* claim stays behind with a null workout id: it is the day's cap ledger, and
* a delete must not hand the slot back. A personal record
* the workout set would otherwise outlive it (the detector only ever raises
* a stored best), so the records keyed on this workout go in the same
* transaction and a silent detection pass re-derives the honest best.
@@ -421,14 +427,25 @@ export const DELETE = apiHandler(
}
const removedRecords = await prisma.$transaction(async (tx) => {
// The detector reads the best workout and writes its record under this
// same lock, so it either wrote before this delete (and the record goes
// below) or reads after it (and never sees the workout).
await lockPersonalRecordsForUser(tx, user.id);
const records = await tx.personalRecord.deleteMany({
where: {
userId: user.id,
metricSlot: { not: null },
source: "MANUAL",
...(row.externalId !== null
? { externalId: row.externalId }
: { externalId: null, achievedAt: row.startedAt }),
? { metricSlot: { not: null }, externalId: row.externalId }
: {
// A legacy record carries no external id, so the start time
// is all that ties it to this workout. Scope it to the slots
// of this workout's sport so a record another sport set at
// the same instant stays.
metricSlot: { in: workoutPrSlotsForSport(row.sportType) },
externalId: null,
achievedAt: row.startedAt,
}),
},
});
await tx.workout.delete({ where: { id: row.id } });
+25 -1
View File
@@ -13,6 +13,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
import { NextRequest } from "next/server";
const tx = {
$queryRaw: vi.fn(async () => []),
personalRecord: { deleteMany: vi.fn() },
workout: { delete: vi.fn() },
};
@@ -129,8 +130,16 @@ describe("DELETE /api/workouts/{id}", () => {
expect(tx.personalRecord.deleteMany).toHaveBeenCalledWith({
where: {
userId: "user-1",
metricSlot: { not: null },
source: "MANUAL",
// Only the running slots: a cycling record set at the same instant
// is not this workout's.
metricSlot: {
in: [
"longest_run_duration",
"longest_distance_run",
"fastest_5km_time",
],
},
externalId: null,
achievedAt: MANUAL_ROW.startedAt,
},
@@ -139,6 +148,21 @@ describe("DELETE /api/workouts/{id}", () => {
expect(enqueuePrDetection).not.toHaveBeenCalled();
});
it("takes the personal-records lock before it deletes anything", async () => {
vi.mocked(prisma.workout.findUnique).mockResolvedValue(MANUAL_ROW as never);
await del();
expect(tx.$queryRaw).toHaveBeenCalledTimes(1);
const lockOrder = tx.$queryRaw.mock.invocationCallOrder[0];
expect(lockOrder).toBeLessThan(
tx.personalRecord.deleteMany.mock.invocationCallOrder[0],
);
expect(lockOrder).toBeLessThan(
tx.workout.delete.mock.invocationCallOrder[0],
);
});
it.each(["APPLE_HEALTH", "WITHINGS", "WHOOP", "EXTERNAL"])(
"refuses a %s workout with 409 and touches nothing",
async (source) => {
@@ -143,7 +143,7 @@ function makeFakePrisma(state: {
return true;
}
return {
const fake = {
// v1.29.6 — the cumulative-kind path no longer runs a `$queryRaw`
// UTC `date_trunc` aggregate; it reads every matching row via
// `measurement.findMany` and does the source-collapse + tz-aware
@@ -277,6 +277,13 @@ function makeFakePrisma(state: {
),
},
} as unknown as PrismaClient;
// The workout-slot leg runs under a per-user advisory lock inside a
// transaction; the fake runs the callback against itself.
Object.assign(fake, {
$queryRaw: vi.fn(async () => []),
$transaction: vi.fn(async (cb: (tx: PrismaClient) => unknown) => cb(fake)),
});
return fake;
}
const USER = "user-1";
+86 -52
View File
@@ -386,58 +386,65 @@ export async function detectPersonalRecordsForUser(
[slot.field]: { gt: 0 } as { gt: number },
} as const;
const sampleCount = await prisma.workout.count({
where: baseWhere,
// Read the best workout and write its record under the same per-user
// lock the workout DELETE takes. Without it a delete could land between
// the read and the write and leave a record for a workout that no longer
// exists (PersonalRecord has no FK to Workout to stop it).
const outcome = await prisma.$transaction(async (tx) => {
await lockPersonalRecordsForUser(tx, userId);
const sampleCount = await tx.workout.count({
where: baseWhere,
});
if (sampleCount < PR_DETECTION_WARMUP_THRESHOLD) return null;
const best = await findBestWorkout(tx, baseWhere, slot);
if (!best) return null;
const value =
slot.field === "durationSec"
? best.durationSec
: (best.totalDistanceM ?? 0);
if (!Number.isFinite(value) || value <= 0) return null;
const currentPR = await tx.personalRecord.findFirst({
where: { userId, metricType: slot.metricType, metricSlot: slot.slot },
orderBy:
slot.direction === PersonalRecordDirection.MAX
? { value: "desc" }
: { value: "asc" },
});
const comparison = compareToCurrentBest(
value,
currentPR?.value,
slot.direction,
);
if (comparison === "no-improvement") return null;
const result = await tx.personalRecord.createMany({
data: [
{
userId,
metricType: slot.metricType,
metricSlot: slot.slot,
direction: slot.direction,
value,
unit: slot.unit,
achievedAt: best.startedAt,
sourceMeasurementId: null,
source: best.source,
externalId: best.externalId,
},
],
skipDuplicates: true,
});
return result.count > 0 ? comparison : null;
});
if (sampleCount < PR_DETECTION_WARMUP_THRESHOLD) continue;
const best = await findBestWorkout(prisma, baseWhere, slot);
if (!best) continue;
const value =
slot.field === "durationSec"
? best.durationSec
: (best.totalDistanceM ?? 0);
if (!Number.isFinite(value) || value <= 0) continue;
const currentPR = await prisma.personalRecord.findFirst({
where: { userId, metricType: slot.metricType, metricSlot: slot.slot },
orderBy:
slot.direction === PersonalRecordDirection.MAX
? { value: "desc" }
: { value: "asc" },
});
const outcome = compareToCurrentBest(
value,
currentPR?.value,
slot.direction,
);
if (outcome === "no-improvement") continue;
const result = await prisma.personalRecord.createMany({
data: [
{
userId,
metricType: slot.metricType,
metricSlot: slot.slot,
direction: slot.direction,
value,
unit: slot.unit,
achievedAt: best.startedAt,
sourceMeasurementId: null,
source: best.source,
externalId: best.externalId,
},
],
skipDuplicates: true,
});
if (result.count > 0) {
if (outcome === "tie") {
ties += 1;
} else {
inserted += 1;
}
if (outcome === "tie") {
ties += 1;
} else if (outcome === "improvement") {
inserted += 1;
}
}
@@ -669,7 +676,7 @@ async function findBestCumulativeDay(
}
async function findBestWorkout(
prisma: PrismaClient,
prisma: Pick<PrismaClient, "workout">,
where: Prisma.WorkoutWhereInput,
slot: WorkoutSlotDefinition,
): Promise<WorkoutCandidate | null> {
@@ -704,6 +711,33 @@ async function findBestWorkout(
};
}
/**
* The per-user lock shared by the workout-slot detection and the workout
* DELETE. Transaction-scoped: it releases on commit or rollback.
*/
export async function lockPersonalRecordsForUser(
tx: Pick<PrismaClient, "$queryRaw">,
userId: string,
): Promise<void> {
await tx.$queryRaw`
SELECT pg_advisory_xact_lock(
hashtext('personal-records'),
hashtext(${userId})
)::text AS locked
`;
}
/**
* The workout-PR slot names the detector derives for one sport. The DELETE
* uses it to scope a legacy record match (no external id) to the sport of the
* workout being removed.
*/
export function workoutPrSlotsForSport(sportType: string): string[] {
return WORKOUT_SLOTS.filter((slot) => slot.sportType === sportType).map(
(slot) => slot.slot,
);
}
/** Exposed for tests + the integration harness. */
export const __test__ = {
WORKOUT_SLOTS,