mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-15 15:02:04 +00:00
fix(workouts): keep tracks and climb for accounts that already had them
Migration 0434 turns the route opt-in on for every account that already holds a stored workout track, so its tracks keep arriving after the upgrade; only accounts without one start off. Workouts stored before this release with a track but no elevation get their metres climbed written into Workout.elevationM by a one-time background pass. It runs detached at worker boot, reads one track at a time with a pause between batches of 100, and keeps its position in app_settings, so a restart resumes and a finished pass ends at once. The "starting" health reason is bounded: after five minutes the probe says plain degraded and the process logs one line, and the time the background role took to come up is logged either way. The compose healthcheck's ten-minute start period already keeps those probes from counting; its comment now says so.
This commit is contained in:
1 parent
846b0e5df0
commit
81e1f4cad1
11 files changed
+460
-12
No files matched your search
@@ -323,6 +323,8 @@ describe("workout route geometry never reaches an AI path", () => {
|
||||
"lib/jobs/free-text-encryption-backfill.ts":
|
||||
"Seals legacy readable tracks in place.",
|
||||
"lib/workouts/route-geometry-cipher.ts": "The cipher itself.",
|
||||
"lib/workouts/climb-backfill.ts":
|
||||
"One-time backfill: derives metres climbed into Workout.elevationM.",
|
||||
};
|
||||
|
||||
/** Directories whose code feeds a model. */
|
||||
|
||||
@@ -213,3 +213,19 @@ describe("GET /api/health while the background role starts", () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("GET /api/health — the `starting` window is bounded", () => {
|
||||
it("drops the reason once a start outlasts any real boot", async () => {
|
||||
const { markStartup, STARTING_MAX_MS } =
|
||||
await import("@/lib/boot/startup-state");
|
||||
vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web");
|
||||
markStartup("starting", Date.now() - STARTING_MAX_MS - 1_000);
|
||||
try {
|
||||
const { response, body } = await health();
|
||||
expect(response.status).toBe(503);
|
||||
expect(body).toEqual({ status: "degraded" });
|
||||
} finally {
|
||||
markStartup("ready");
|
||||
}
|
||||
});
|
||||
});
|
||||
+24
-5
@@ -86,11 +86,30 @@ export async function register() {
|
||||
// the boss is not bound yet takes the same path as during a producer
|
||||
// reconnect: a garnish skips its enqueue (the owning cron catches up),
|
||||
// and a route that needs the worker answers 503 until it is up.
|
||||
const [{ shouldRunWorker }, { markStartup }] = await Promise.all([
|
||||
const [
|
||||
{ shouldRunWorker },
|
||||
{ markStartup, startupPhase, startupElapsedMs, STARTING_MAX_MS },
|
||||
] = await Promise.all([
|
||||
import("@/lib/process-type"),
|
||||
import("@/lib/boot/startup-state"),
|
||||
]);
|
||||
markStartup("starting");
|
||||
// One line when the start takes longer than any real boot; from then on
|
||||
// the probe reports plain `degraded` instead of `starting`.
|
||||
setTimeout(() => {
|
||||
if (startupPhase() === "stalled") {
|
||||
console.warn(
|
||||
`[boot] The background role is still not up after ${Math.round(STARTING_MAX_MS / 1000)} s; /api/health now reports degraded without "starting".`,
|
||||
);
|
||||
}
|
||||
}, STARTING_MAX_MS + 1_000).unref();
|
||||
const ready = (up: boolean) => {
|
||||
const elapsed = startupElapsedMs();
|
||||
markStartup(up ? "ready" : "failed");
|
||||
console.info(
|
||||
`[boot] background role ${up ? "ready" : "failed"} after ${elapsed ?? 0} ms`,
|
||||
);
|
||||
};
|
||||
if (!shouldRunWorker()) {
|
||||
// Split deployment: web requests still produce queue jobs, but this
|
||||
// process owns no schedules, supervision, or consumers.
|
||||
@@ -104,9 +123,9 @@ export async function register() {
|
||||
await import("@/lib/jobs/reminder/worker-lifecycle");
|
||||
registerShutdownHandlers(boss);
|
||||
}
|
||||
markStartup(boss ? "ready" : "failed");
|
||||
ready(boss !== null);
|
||||
} catch (err) {
|
||||
markStartup("failed");
|
||||
ready(false);
|
||||
// Queue-backed garnishes degrade independently; never make the web
|
||||
// server unavailable because its producer connection failed.
|
||||
console.error(
|
||||
@@ -130,13 +149,13 @@ export async function register() {
|
||||
const { startReminderWorker } =
|
||||
await import("@/lib/jobs/reminder-worker");
|
||||
await startReminderWorker();
|
||||
markStartup("ready");
|
||||
ready(true);
|
||||
|
||||
startupEvt.addMeta("reminder_worker", "started");
|
||||
startupEvt.finish();
|
||||
emitIfSampled(startupEvt.toJSON());
|
||||
} catch (err) {
|
||||
markStartup("failed");
|
||||
ready(false);
|
||||
startupEvt.setError(err);
|
||||
startupEvt.addMeta("reminder_worker", "failed");
|
||||
startupEvt.finish();
|
||||
|
||||
@@ -8,16 +8,48 @@
|
||||
* The probe reads this to say `starting` while that is under way rather than
|
||||
* hanging, and the operator sees a 503 that names a transient state.
|
||||
*
|
||||
* The window is bounded. A start that has not finished after
|
||||
* `STARTING_MAX_MS` is no longer reported as `starting`: it reads `stalled`,
|
||||
* the probe drops the reason and says plainly `degraded`, and the process
|
||||
* logs one line saying so. An uptime monitor therefore never sees the
|
||||
* "transient" label for longer than a real boot takes.
|
||||
*
|
||||
* `null` means no background role was started in this process at all.
|
||||
*/
|
||||
export type StartupPhase = "starting" | "ready" | "failed";
|
||||
export type StartupPhase = "starting" | "ready" | "failed" | "stalled";
|
||||
|
||||
const KEY = Symbol.for("healthlog.boot.startupPhase");
|
||||
/** Longest a start is still called `starting`. Prod boots take 10–60 s. */
|
||||
export const STARTING_MAX_MS = 5 * 60_000;
|
||||
|
||||
export function markStartup(phase: StartupPhase): void {
|
||||
(globalThis as Record<symbol, unknown>)[KEY] = phase;
|
||||
const KEY = Symbol.for("healthlog.boot.startupState");
|
||||
|
||||
interface State {
|
||||
phase: Exclude<StartupPhase, "stalled">;
|
||||
since: number;
|
||||
}
|
||||
|
||||
export function startupPhase(): StartupPhase | null {
|
||||
return ((globalThis as Record<symbol, unknown>)[KEY] as StartupPhase) ?? null;
|
||||
function state(): State | undefined {
|
||||
return (globalThis as Record<symbol, unknown>)[KEY] as State | undefined;
|
||||
}
|
||||
|
||||
export function markStartup(
|
||||
phase: Exclude<StartupPhase, "stalled">,
|
||||
now: number = Date.now(),
|
||||
): void {
|
||||
(globalThis as Record<symbol, unknown>)[KEY] = { phase, since: now };
|
||||
}
|
||||
|
||||
export function startupPhase(now: number = Date.now()): StartupPhase | null {
|
||||
const current = state();
|
||||
if (!current) return null;
|
||||
if (current.phase === "starting" && now - current.since > STARTING_MAX_MS) {
|
||||
return "stalled";
|
||||
}
|
||||
return current.phase;
|
||||
}
|
||||
|
||||
/** Milliseconds since the current phase began, or null before any start. */
|
||||
export function startupElapsedMs(now: number = Date.now()): number | null {
|
||||
const current = state();
|
||||
return current ? now - current.since : null;
|
||||
}
|
||||
@@ -15,6 +15,7 @@
|
||||
* `intake-slot-dedup-queue`, `tls-pin-monitor-queue`, and `geo-backfill`
|
||||
* guards read THIS module.
|
||||
*/
|
||||
import { trackBackgroundTask } from "@/lib/logging/background-tasks";
|
||||
import { PgBoss } from "pg-boss";
|
||||
import { GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON } from "@/lib/jobs/geo-backfill";
|
||||
import {
|
||||
@@ -2166,4 +2167,31 @@ export async function enqueueMaintenanceBootDiscovery(): Promise<void> {
|
||||
err,
|
||||
);
|
||||
}
|
||||
|
||||
// v1.44 — one-time climb backfill for workouts stored with a track but no
|
||||
// elevation (`climb-backfill.ts`). Detached: it reads one track at a time
|
||||
// with a pause between batches and resumes from its cursor on the next
|
||||
// boot, so it never holds the boot or the heap. A finished pass returns at
|
||||
// once.
|
||||
trackBackgroundTask(
|
||||
(async () => {
|
||||
const [{ prisma }, { runWorkoutClimbBackfill }] = await Promise.all([
|
||||
import("@/lib/db"),
|
||||
import("@/lib/workouts/climb-backfill"),
|
||||
]);
|
||||
const result = await runWorkoutClimbBackfill({ prisma });
|
||||
if (result.scanned > 0) {
|
||||
workerLog(
|
||||
"info",
|
||||
`[workout-climb-backfill] scanned=${result.scanned} filled=${result.filled} done=${result.done}`,
|
||||
);
|
||||
}
|
||||
})().catch((err) => {
|
||||
workerLog(
|
||||
"error",
|
||||
"[workout-climb-backfill] failed; the next boot resumes from the cursor",
|
||||
err,
|
||||
);
|
||||
}),
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import {
|
||||
CLIMB_BACKFILL_DONE,
|
||||
runWorkoutClimbBackfill,
|
||||
} from "../climb-backfill";
|
||||
|
||||
/**
|
||||
* An in-memory stand-in: workouts with an optional readable track (the
|
||||
* legacy column, which `readRouteGeometry` falls back to) and the cursor.
|
||||
*/
|
||||
function fake(
|
||||
workouts: Array<{
|
||||
id: string;
|
||||
elevationM: number | null;
|
||||
coords?: number[][];
|
||||
}>,
|
||||
cursor: string | null = null,
|
||||
) {
|
||||
const state = { cursor, reads: 0 };
|
||||
const prisma = {
|
||||
appSettings: {
|
||||
findUnique: async () => ({ workoutClimbBackfillCursor: state.cursor }),
|
||||
upsert: async (args: {
|
||||
update: { workoutClimbBackfillCursor: string };
|
||||
}) => {
|
||||
state.cursor = args.update.workoutClimbBackfillCursor;
|
||||
return {};
|
||||
},
|
||||
},
|
||||
workout: {
|
||||
findMany: async (args: {
|
||||
where: { id?: { gt: string } };
|
||||
take: number;
|
||||
}) =>
|
||||
workouts
|
||||
.filter((w) => w.elevationM === null && w.coords)
|
||||
.filter((w) => !args.where.id || w.id > args.where.id.gt)
|
||||
.sort((a, b) => a.id.localeCompare(b.id))
|
||||
.slice(0, args.take)
|
||||
.map((w) => ({ id: w.id })),
|
||||
updateMany: async (args: {
|
||||
where: { id: string };
|
||||
data: { elevationM: number };
|
||||
}) => {
|
||||
const w = workouts.find((x) => x.id === args.where.id)!;
|
||||
if (w.elevationM !== null) return { count: 0 };
|
||||
w.elevationM = args.data.elevationM;
|
||||
return { count: 1 };
|
||||
},
|
||||
},
|
||||
workoutRoute: {
|
||||
findUnique: async (args: { where: { workoutId: string } }) => {
|
||||
state.reads += 1;
|
||||
const w = workouts.find((x) => x.id === args.where.workoutId);
|
||||
return w?.coords
|
||||
? {
|
||||
geometry: { type: "LineString", coordinates: w.coords },
|
||||
geometryEncrypted: null,
|
||||
}
|
||||
: null;
|
||||
},
|
||||
},
|
||||
};
|
||||
return { prisma: prisma as never, state, workouts };
|
||||
}
|
||||
|
||||
describe("runWorkoutClimbBackfill", () => {
|
||||
it("fills the climb from the track and marks the pass done", async () => {
|
||||
const f = fake([
|
||||
{
|
||||
id: "a",
|
||||
elevationM: null,
|
||||
coords: [
|
||||
[8, 50, 100],
|
||||
[8, 50, 160],
|
||||
[8, 50, 140],
|
||||
],
|
||||
},
|
||||
{
|
||||
id: "b",
|
||||
elevationM: 12,
|
||||
coords: [
|
||||
[8, 50, 0],
|
||||
[8, 50, 500],
|
||||
],
|
||||
},
|
||||
{
|
||||
id: "c",
|
||||
elevationM: null,
|
||||
coords: [
|
||||
[8, 50],
|
||||
[8.1, 50.1],
|
||||
],
|
||||
},
|
||||
{ id: "d", elevationM: null },
|
||||
]);
|
||||
const result = await runWorkoutClimbBackfill({
|
||||
prisma: f.prisma,
|
||||
pauseMs: 0,
|
||||
});
|
||||
expect(result).toEqual({ scanned: 2, filled: 1, done: true });
|
||||
expect(f.workouts.find((w) => w.id === "a")?.elevationM).toBe(60);
|
||||
// A stored elevation is never overwritten; no altitude stays NULL.
|
||||
expect(f.workouts.find((w) => w.id === "b")?.elevationM).toBe(12);
|
||||
expect(f.workouts.find((w) => w.id === "c")?.elevationM).toBeNull();
|
||||
expect(f.state.cursor).toBe(CLIMB_BACKFILL_DONE);
|
||||
});
|
||||
|
||||
it("resumes after the cursor and never re-reads what it passed", async () => {
|
||||
const many = Array.from({ length: 250 }, (_, i) => ({
|
||||
id: `w${String(i).padStart(4, "0")}`,
|
||||
elevationM: null as number | null,
|
||||
coords: [[8, 50]] as number[][],
|
||||
}));
|
||||
const f = fake(many);
|
||||
const first = await runWorkoutClimbBackfill({
|
||||
prisma: f.prisma,
|
||||
pauseMs: 0,
|
||||
maxBatches: 1,
|
||||
});
|
||||
expect(first).toEqual({ scanned: 100, filled: 0, done: false });
|
||||
expect(f.state.cursor).toBe("w0099");
|
||||
const rest = await runWorkoutClimbBackfill({
|
||||
prisma: f.prisma,
|
||||
pauseMs: 0,
|
||||
});
|
||||
expect(rest).toEqual({ scanned: 150, filled: 0, done: true });
|
||||
expect(f.state.reads).toBe(250);
|
||||
const again = await runWorkoutClimbBackfill({
|
||||
prisma: f.prisma,
|
||||
pauseMs: 0,
|
||||
});
|
||||
expect(again).toEqual({ scanned: 0, filled: 0, done: true });
|
||||
expect(f.state.reads).toBe(250);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,99 @@
|
||||
/**
|
||||
* One-time backfill: metres climbed for workouts stored before v1.44.
|
||||
*
|
||||
* Since v1.44 a kept GPS track fills `Workout.elevationM` when the workout is
|
||||
* written (`route-climb.ts`), and the workout insight reads only that column.
|
||||
* Workouts stored earlier with a track but no elevation used to get their
|
||||
* climb from the track when the insight was built; this pass writes it into
|
||||
* the column once so they keep it.
|
||||
*
|
||||
* Bounded and resumable: workouts are taken in id order, `BATCH` ids at a
|
||||
* time, and each track is opened, reduced to one number and dropped before
|
||||
* the next is read, so memory holds one track at most. The last id is stored
|
||||
* in `AppSettings.workoutClimbBackfillCursor` after every batch; a restart
|
||||
* continues from there, and `done` ends it for good. A track without an
|
||||
* altitude channel leaves the column NULL, as before; the cursor has moved
|
||||
* past it, so it is never read again. Workouts written after the upgrade
|
||||
* already carry their climb, so one pass in id order is complete.
|
||||
*
|
||||
* Not an AI path: the number lands on the workout row, nothing else.
|
||||
*/
|
||||
import type { PrismaClient } from "@/generated/prisma/client";
|
||||
import { readRouteGeometry } from "@/lib/workouts/route-geometry-cipher";
|
||||
import { routeClimbM } from "@/lib/workouts/route-climb";
|
||||
|
||||
export const CLIMB_BACKFILL_DONE = "done";
|
||||
const BATCH = 100;
|
||||
const SETTINGS_ID = "singleton";
|
||||
|
||||
export interface ClimbBackfillResult {
|
||||
scanned: number;
|
||||
filled: number;
|
||||
done: boolean;
|
||||
}
|
||||
|
||||
type Client = Pick<PrismaClient, "appSettings" | "workout" | "workoutRoute">;
|
||||
|
||||
export async function runWorkoutClimbBackfill(input: {
|
||||
prisma: Client;
|
||||
/** Pause between batches so the pass never crowds out request traffic. */
|
||||
pauseMs?: number;
|
||||
/** Upper bound on batches in one run; the next run resumes. */
|
||||
maxBatches?: number;
|
||||
}): Promise<ClimbBackfillResult> {
|
||||
const {
|
||||
prisma,
|
||||
pauseMs = 250,
|
||||
maxBatches = Number.POSITIVE_INFINITY,
|
||||
} = input;
|
||||
const settings = await prisma.appSettings.findUnique({
|
||||
where: { id: SETTINGS_ID },
|
||||
select: { workoutClimbBackfillCursor: true },
|
||||
});
|
||||
let cursor = settings?.workoutClimbBackfillCursor ?? null;
|
||||
const result: ClimbBackfillResult = { scanned: 0, filled: 0, done: false };
|
||||
if (cursor === CLIMB_BACKFILL_DONE) return { ...result, done: true };
|
||||
|
||||
for (let batch = 0; batch < maxBatches; batch += 1) {
|
||||
const ids = await prisma.workout.findMany({
|
||||
where: {
|
||||
elevationM: null,
|
||||
route: { isNot: null },
|
||||
...(cursor ? { id: { gt: cursor } } : {}),
|
||||
},
|
||||
orderBy: { id: "asc" },
|
||||
take: BATCH,
|
||||
select: { id: true },
|
||||
});
|
||||
for (const { id } of ids) {
|
||||
result.scanned += 1;
|
||||
const route = await prisma.workoutRoute.findUnique({
|
||||
where: { workoutId: id },
|
||||
select: { geometry: true, geometryEncrypted: true },
|
||||
});
|
||||
const climb = route ? routeClimbM(readRouteGeometry(route)) : null;
|
||||
if (climb !== null) {
|
||||
// Only where the column is still empty: a sync that wrote an
|
||||
// elevation in the meantime wins.
|
||||
const { count } = await prisma.workout.updateMany({
|
||||
where: { id, elevationM: null },
|
||||
data: { elevationM: climb },
|
||||
});
|
||||
result.filled += count;
|
||||
}
|
||||
}
|
||||
const finished = ids.length < BATCH;
|
||||
cursor = finished ? CLIMB_BACKFILL_DONE : ids[ids.length - 1].id;
|
||||
await prisma.appSettings.upsert({
|
||||
where: { id: SETTINGS_ID },
|
||||
create: { id: SETTINGS_ID, workoutClimbBackfillCursor: cursor },
|
||||
update: { workoutClimbBackfillCursor: cursor },
|
||||
});
|
||||
if (finished) {
|
||||
result.done = true;
|
||||
return result;
|
||||
}
|
||||
if (pauseMs > 0) await new Promise((r) => setTimeout(r, pauseMs));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
Reference in new issue
Block a user