fix(measurements): sum both starts of the repeated fall-back hour into its slot

This commit is contained in:
Marc-André Bombeck committed 2026-10-16 05:57:19 +02:00
1 parent f360d221e1
commit 617b87a583
3 files changed
+123 -31

No files matched your search

+34 -31
View File
@@ -81,8 +81,10 @@ import {
import {
classifyHourBucket,
HOUR_BUCKET_SKIP_REASONS,
combineHourWrites,
parseHourBucketId,
writeHourBuckets,
type HourBucketEntry,
type HourBucketWrite,
} from "@/lib/measurements/intraday-hour-buckets";
import { resolveStoredTimezone } from "@/lib/tz/resolver";
@@ -393,7 +395,7 @@ async function postBatch(request: NextRequest): Promise<Response> {
const unstableShapes: UnstableExternalIdShape[] = [];
// v1.43 — hourly sums for the intraday profile, and the zone they are cut in
// (the account's, the one `/api/auth/me` publishes as `timezone`).
const hourWrites: HourBucketWrite[] = [];
const hourWrites: HourBucketEntry[] = [];
let accountTz: string | undefined;
const batchNow = new Date();
@@ -483,6 +485,7 @@ async function postBatch(request: NextRequest): Promise<Response> {
dateKey: hourId.dateKey,
hour: hourId.hour,
value: hourMapped.value,
startMs: startDate.getTime(),
});
continue;
}
@@ -983,28 +986,24 @@ async function postBatch(request: NextRequest): Promise<Response> {
}
}
// v1.43 — the hours, after the measurements. The last entry for one slot
// wins, as for the day rows; an earlier one in the same batch is a
// superseded duplicate.
// v1.43 — the hours, after the measurements. One write per slot: the last
// entry for one start wins (an earlier one in the same batch is a
// superseded duplicate), and the two starts of the hour a fall-back day
// repeats are summed (`combineHourWrites`).
let hourInsertedCount = 0;
let hourUpdatedCount = 0;
const hourLanded: HourBucketWrite[] = [];
if (hourWrites.length > 0 && accountTz) {
const lastBySlot = new Map<string, HourBucketWrite>();
for (const w of hourWrites) {
lastBySlot.set(`${w.type}::${w.dateKey}::${w.hour}`, w);
const { writes, superseded } = combineHourWrites(hourWrites);
for (const index of superseded) {
results[index] = {
index,
status: "duplicate",
reason: "superseded_in_batch",
};
duplicateCount++;
}
for (const w of hourWrites) {
if (lastBySlot.get(`${w.type}::${w.dateKey}::${w.hour}`) !== w) {
results[w.index] = {
index: w.index,
status: "duplicate",
reason: "superseded_in_batch",
};
duplicateCount++;
}
}
const toWrite = [...lastBySlot.values()];
const toWrite = writes.map(({ write }) => write);
try {
const statuses = await writeHourBuckets(
prisma,
@@ -1012,21 +1011,25 @@ async function postBatch(request: NextRequest): Promise<Response> {
accountTz,
toWrite,
);
for (const w of toWrite) {
const status = statuses.get(w.index) ?? "inserted";
results[w.index] = { index: w.index, status };
if (status === "inserted") hourInsertedCount++;
else hourUpdatedCount++;
hourLanded.push(w);
for (const { write, contributors } of writes) {
const status = statuses.get(write.index) ?? "inserted";
for (const index of contributors) {
results[index] = { index, status };
if (status === "inserted") hourInsertedCount++;
else hourUpdatedCount++;
}
hourLanded.push(write);
}
} catch (err) {
for (const w of toWrite) {
results[w.index] = {
index: w.index,
status: "failed",
reason: "write_failed",
};
failedCount++;
for (const { contributors } of writes) {
for (const index of contributors) {
results[index] = {
index,
status: "failed",
reason: "write_failed",
};
failedCount++;
}
}
annotate({
action: { name: "measurement.batch.hour-bucket.failed" },
@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
import {
classifyHourBucket,
combineHourWrites,
cumulativeFromIncrements,
HOUR_NOT_POSTED,
isStartOfLocalHour,
@@ -158,3 +159,38 @@ describe("cumulativeFromIncrements", () => {
expect(postedHours(increments)).toBe(3);
});
});
describe("combineHourWrites", () => {
const entry = (index: number, startIso: string, value: number) => ({
index,
type: "ACTIVITY_STEPS" as const,
hkIdentifier: "HKQuantityTypeIdentifierStepCount",
dateKey: "2026-10-25",
hour: 2,
value,
startMs: Date.parse(startIso),
});
it("sums the two starts of the hour a fall-back day repeats", () => {
// Berlin, 2026-10-25: 02:00 CEST is 00:00Z, 02:00 CET is 01:00Z.
const { writes, superseded } = combineHourWrites([
entry(0, "2026-10-25T00:00:00Z", 300),
entry(1, "2026-10-25T01:00:00Z", 120),
]);
expect(superseded).toEqual([]);
expect(writes).toHaveLength(1);
expect(writes[0].write.value).toBe(420);
expect(writes[0].contributors.sort()).toEqual([0, 1]);
});
it("lets the last entry of one start win, and marks the earlier one", () => {
const { writes, superseded } = combineHourWrites([
entry(0, "2026-10-25T00:00:00Z", 300),
entry(1, "2026-10-25T00:00:00Z", 310),
entry(2, "2026-10-25T01:00:00Z", 120),
]);
expect(superseded).toEqual([0]);
expect(writes[0].write.value).toBe(430);
expect(writes[0].write.index).toBe(2);
});
});
@@ -177,6 +177,59 @@ export interface HourBucketWrite {
value: number;
}
/** One hour as the batch reads it, with the instant it starts at. */
export interface HourBucketEntry extends HourBucketWrite {
/** `startDate` of the entry, in epoch milliseconds. */
startMs: number;
}
/**
* A batch's hours reduced to one write per slot.
*
* A slot is one local hour of one day. On the autumn day the clock falls
* back, one local hour happens twice, and the phone hands over both under
* the same key (`…T02`), each with its own `startDate`: they are two hours of
* activity and are summed into the slot rather than the second replacing the
* first. Entries that repeat the same start are the same hour posted twice,
* and the last one counts, as for every other slot.
*
* `contributors` are the entries whose values the write carries (they take
* its status); `superseded` are the earlier repeats of one start.
*
* The slot holds one number, so the sum is only as complete as the batch
* that carries it: both occurrences have to arrive together. The phone posts
* a day's hours in one batch, and a later batch that carries only one of
* them replaces the slot with that one, as a re-post of any slot does.
*/
export function combineHourWrites(entries: readonly HourBucketEntry[]): {
writes: Array<{ write: HourBucketWrite; contributors: number[] }>;
superseded: number[];
} {
const slots = new Map<string, Map<number, HourBucketEntry>>();
const superseded: number[] = [];
for (const entry of entries) {
const key = `${entry.type}::${entry.dateKey}::${entry.hour}`;
const byStart = slots.get(key) ?? new Map<number, HourBucketEntry>();
const earlier = byStart.get(entry.startMs);
if (earlier) superseded.push(earlier.index);
byStart.set(entry.startMs, entry);
slots.set(key, byStart);
}
const writes = [...slots.values()].map((byStart) => {
const occurrences = [...byStart.values()];
const last = occurrences.reduce((a, b) => (b.index > a.index ? b : a));
const { startMs: _start, ...base } = last;
return {
write: {
...base,
value: occurrences.reduce((sum, entry) => sum + entry.value, 0),
},
contributors: occurrences.map((entry) => entry.index),
};
});
return { writes, superseded };
}
/**
* Store a batch's hours, one transaction per (type, day). Returns the
* per-entry status: `inserted` for the first value of a slot, `updated` when