diff --git a/docs/metric-stream-timescaledb-runbook.md b/docs/metric-stream-timescaledb-runbook.md index b89260d08f..185b827ff9 100644 --- a/docs/metric-stream-timescaledb-runbook.md +++ b/docs/metric-stream-timescaledb-runbook.md @@ -30,6 +30,19 @@ SELECT hypertable_schema, hypertable_name FROM timescaledb_information.hypertables WHERE hypertable_schema = 'fitness' AND hypertable_name = 'metric_stream'; + +SELECT hypertable_schema, hypertable_name, dimension_number, column_name, time_interval +FROM timescaledb_information.dimensions +WHERE hypertable_schema = 'fitness' + AND hypertable_name = 'metric_stream'; + +SELECT chunk_schema, chunk_name, range_start, range_end, is_compressed, + pg_size_pretty(pg_total_relation_size(format('%I.%I', chunk_schema, chunk_name)::regclass)) AS total_size +FROM timescaledb_information.chunks +WHERE hypertable_schema = 'fitness' + AND hypertable_name = 'metric_stream' +ORDER BY pg_total_relation_size(format('%I.%I', chunk_schema, chunk_name)::regclass) DESC +LIMIT 20; ``` ## 2) Pause writers @@ -70,6 +83,16 @@ That migration configures: - order by: `recorded_at DESC` - compression policy: compress chunks older than `7 days` +For existing environments, forward migration +`drizzle/0011_metric_stream_storage_controls.sql` enforces the `1 day` chunk +interval and removes storage-heavy indexes that are not part of the current +read model. If production reports a larger `time_interval`, run pending +migrations before compression work. + +Timescale expands hypertable indexes across chunks, so `DROP INDEX CONCURRENTLY` +is not reliable for these index removals. Drop obsolete hypertable indexes +during the maintenance window after app services are scaled down. + ## 5) Backfill compression for old chunks Do not run this while materialized-view refresh is rebuilding `deduped_sensor`, @@ -175,6 +198,21 @@ SELECT pg_size_pretty(hypertable_size('fitness.metric_stream')) AS hypertable_si Record resistant chunks in `.context/metric-stream-compression-YYYY-MM-DD.md` with the error, elapsed time, and whether inserts were blocked. +If online compression repeatedly fails with lock timeouts while app traffic or +materialized-view refreshes are active, use a maintenance window: + +1. Confirm the latest `dofek-db-backups` object or Databasus backup metadata is + less than 24 hours old. +2. Scale down writers and read-heavy app services: + ```bash + docker service scale dofek_web=0 dofek_worker=0 dofek_training-export-worker=0 + ``` +3. Cancel active `metric_stream`, `deduped_sensor`, `activity_summary`, or + `provider_stats` maintenance queries with `pg_cancel_backend`. +4. Run `compress_chunk` for closed chunks only. +5. Reconcile chunk compression metadata and disk usage before resuming + services. + ## 6) Resume services ```bash @@ -201,6 +239,19 @@ Check app health: curl -fsS https://dofek.asherlc.com/healthz ``` +Check for future-dated chunks after any incident involving unexpected chunk +fanout: + +```sql +SELECT chunk_name, range_start, range_end, is_compressed, + pg_size_pretty(pg_total_relation_size(format('%I.%I', chunk_schema, chunk_name)::regclass)) AS total_size +FROM timescaledb_information.chunks +WHERE hypertable_schema = 'fitness' + AND hypertable_name = 'metric_stream' + AND range_start > now() + INTERVAL '30 days' +ORDER BY range_start; +``` + ## Optional next step: retention Retention is intentionally not auto-enabled in migrations because it is a product/data-retention decision. diff --git a/docs/production-incident-baseline.md b/docs/production-incident-baseline.md index 58d6e66c41..95271b899d 100644 --- a/docs/production-incident-baseline.md +++ b/docs/production-incident-baseline.md @@ -7,6 +7,66 @@ full incident log or a replacement for runbooks. Use it to build shared memory about the kinds of issues this system encounters, the signals that identified them, and the durability work they suggest. +## 2026-05-02: metric_stream Storage Pressure from Oversized Chunk and Duplicate Indexes + +### Impact + +Production entered a planned maintenance window while `web`, `worker`, and +`training-export-worker` were scaled down to compress the largest +`fitness.metric_stream` chunk and remove obsolete indexes. The app was restored +after maintenance and `/healthz` returned OK. + +### What Happened + +The production data volume was 64% used, with Postgres accounting for roughly +49GB. `fitness.metric_stream` dominated logical database size at about 35GB, +including about 18GB of indexes. The largest closed Timescale chunk, +`_hyper_1_186_chunk`, covered 2026-04-23 through 2026-04-30 and was still +uncompressed at about 28GB because production had a 7-day chunk interval and +the compression policy only acts on chunks older than 7 days. + +### Evidence That Mattered + +- Data volume before maintenance: `99G` total, `60G` used, `35G` available. +- `fitness.metric_stream`: about `35GB` total, `16GB` heap, `18GB` indexes. +- Largest chunk: `_hyper_1_186_chunk`, uncompressed, about `28GB`, about + `51.6M` rows. +- Online compression failed with: + `ERROR: canceling statement due to lock timeout`. +- Active blockers included recurring materialized-view refreshes for + `fitness.deduped_sensor`, `fitness.activity_summary`, and + `fitness.provider_stats`. +- Future-dated chunks were tiny, but present: `apple_motion` / `api` / `imu` + rows from `WHOOP Strap` plus one `whoop_ble` / `ble` / `orientation` row. + +### Root Cause + +`metric_stream` storage pressure came from a one-week active chunk that +accumulated tens of millions of high-frequency sensor rows before compression +could apply, plus non-primary-key indexes that were no longer needed for the +current read model. + +### Fix Or Mitigation + +- Verified a fresh Databasus backup existed before the window. +- Set the `metric_stream` chunk interval to `1 day`. +- Scaled down app services, canceled active materialized-view refresh work, and + compressed `_hyper_1_186_chunk`. +- Dropped `metric_stream_provider_time_idx` and the obsolete recorded-at index + during the maintenance window. +- Added migration `0011_metric_stream_storage_controls.sql` to enforce the + one-day chunk interval and index removals for future environments. +- Added server-side guards that reject IMU and WHOOP BLE realtime samples more + than five minutes in the future before inserting into `metric_stream`. + +### Remaining Risk + +The current open chunk can still grow until it closes, but future chunks should +be one day wide. The tiny existing future-dated rows were not deleted during +this maintenance window; remove them only after an explicit data-retention / +cleanup decision. The timestamp guard must be deployed before it protects live +ingest traffic. + ## 2026-04-29: PR 1075 CI Blocked by ClickHouse Bootstrap and Web E2E Drift ### Impact diff --git a/docs/schema.dbml b/docs/schema.dbml index 63d51d7c00..bf47233477 100644 --- a/docs/schema.dbml +++ b/docs/schema.dbml @@ -756,7 +756,6 @@ table fitness.metric_stream { (activity_id, channel, recorded_at) [name: 'metric_stream_activity_channel_time_idx'] (id, recorded_at) [pk] (user_id, channel, recorded_at) [name: 'metric_stream_user_channel_time_idx'] - (provider_id, recorded_at) [name: 'metric_stream_provider_time_idx'] } } diff --git a/drizzle/0011_metric_stream_storage_controls.sql b/drizzle/0011_metric_stream_storage_controls.sql new file mode 100644 index 0000000000..f0965945f5 --- /dev/null +++ b/drizzle/0011_metric_stream_storage_controls.sql @@ -0,0 +1,7 @@ +SET lock_timeout = '10s'; +--> statement-breakpoint +SELECT public.set_chunk_time_interval('fitness.metric_stream', INTERVAL '1 day'); +--> statement-breakpoint +DROP INDEX CONCURRENTLY IF EXISTS fitness.metric_stream_provider_time_idx; +--> statement-breakpoint +DROP INDEX CONCURRENTLY IF EXISTS fitness.metric_stream_recorded_at_idx; diff --git a/packages/server/src/repositories/activity-sensor-store.test-helpers.test.ts b/packages/server/src/repositories/activity-sensor-store.test-helpers.test.ts new file mode 100644 index 0000000000..1b4c23954d --- /dev/null +++ b/packages/server/src/repositories/activity-sensor-store.test-helpers.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, it, vi } from "vitest"; +import { createPostgresTestActivitySensorStore } from "./activity-sensor-store.test-helpers.ts"; + +describe("PostgresTestActivitySensorStore", () => { + it("falls back to metric stream when deduped sensor values are missing", async () => { + const execute = vi + .fn() + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ scalar: 110 }, { scalar: 120 }]); + const store = createPostgresTestActivitySensorStore({ execute }); + + const result = await store.getHeartRateZoneSeconds( + { + activityId: "activity-id", + userId: "user-id", + startedAt: "2026-01-01T00:00:00Z", + endedAt: null, + memberActivityIds: ["linked-activity-id"], + }, + 160, + 60, + ); + + expect(execute).toHaveBeenCalledTimes(2); + expect(result).toEqual([ + { zone: 1, seconds: 1 }, + { zone: 2, seconds: 1 }, + { zone: 3, seconds: 0 }, + { zone: 4, seconds: 0 }, + { zone: 5, seconds: 0 }, + ]); + }); + + it("falls back to metric stream when primary zone values are all zero seconds", async () => { + const execute = vi + .fn() + .mockResolvedValueOnce([{ scalar: 40 }]) + .mockResolvedValueOnce([{ scalar: 110 }, { scalar: 140 }]); + const store = createPostgresTestActivitySensorStore({ execute }); + + const result = await store.getHeartRateZoneSeconds( + { + activityId: "activity-id", + userId: "user-id", + startedAt: "2026-01-01T00:00:00Z", + endedAt: null, + memberActivityIds: ["linked-activity-id", "linked-activity-id"], + }, + 160, + 60, + ); + + expect(execute).toHaveBeenCalledTimes(2); + expect(result).toEqual([ + { zone: 1, seconds: 1 }, + { zone: 2, seconds: 0 }, + { zone: 3, seconds: 0 }, + { zone: 4, seconds: 1 }, + { zone: 5, seconds: 0 }, + ]); + }); + + it("uses the current activity as fallback source when member activities are empty", async () => { + const execute = vi + .fn() + .mockResolvedValueOnce([{ scalar: 110 }]) + .mockResolvedValueOnce([{ scalar: 120 }]); + const store = createPostgresTestActivitySensorStore({ execute }); + + const result = await store.getHeartRateZoneSeconds( + { + activityId: "activity-id", + userId: "user-id", + startedAt: "2026-01-01T00:00:00Z", + endedAt: null, + memberActivityIds: [], + }, + 170, + 70, + ); + + expect(execute).toHaveBeenCalledTimes(2); + expect(result).toEqual([ + { zone: 1, seconds: 1 }, + { zone: 2, seconds: 0 }, + { zone: 3, seconds: 0 }, + { zone: 4, seconds: 0 }, + { zone: 5, seconds: 0 }, + ]); + }); +}); diff --git a/packages/server/src/repositories/activity-sensor-store.test-helpers.ts b/packages/server/src/repositories/activity-sensor-store.test-helpers.ts index 9026e5d8ec..81d29bde5b 100644 --- a/packages/server/src/repositories/activity-sensor-store.test-helpers.ts +++ b/packages/server/src/repositories/activity-sensor-store.test-helpers.ts @@ -369,11 +369,33 @@ class PostgresTestActivitySensorStore implements ActivitySensorStore { maxHr: number, restingHr: number, ): Promise { - const values = await this.#activityChannelValues(window, "heart_rate"); - return [1, 2, 3, 4, 5].map((zone) => ({ - zone, - seconds: values.filter((value) => valueInHeartRateZone(value, zone, maxHr, restingHr)).length, - })); + const primaryValues = await this.#activityChannelValues(window, "heart_rate"); + let values = primaryValues; + if (values.length === 0) { + values = await this.#activityChannelValuesFromMetricStream(window, "heart_rate"); + } + + const calculateZoneRows = (zoneValues: number[]) => + [1, 2, 3, 4, 5].map((zone) => ({ + zone, + seconds: zoneValues.filter((value) => valueInHeartRateZone(value, zone, maxHr, restingHr)) + .length, + })); + + const zoneRows = calculateZoneRows(values); + + if (zoneRows.every((row) => row.seconds === 0)) { + const fallbackValues = await this.#activityChannelValuesFromMetricStream( + window, + "heart_rate", + ); + const fallbackRows = calculateZoneRows(fallbackValues); + if (fallbackRows.some((row) => row.seconds > 0)) { + return fallbackRows; + } + } + + return zoneRows; } async getPowerZoneSeconds(window: ActivitySensorWindow, ftp: number): Promise { @@ -397,6 +419,32 @@ class PostgresTestActivitySensorStore implements ActivitySensorStore { return rows.map((row) => row.scalar); } + async #activityChannelValuesFromMetricStream( + window: ActivitySensorWindow, + channel: string, + ): Promise { + const activityIds = + window.memberActivityIds.length === 0 + ? [window.activityId] + : [window.activityId, ...window.memberActivityIds]; + + const deduplicatedActivityIds = [...new Set(activityIds)]; + const activityIdClauses = deduplicatedActivityIds.map( + (memberActivityId) => sql`${memberActivityId}::uuid`, + ); + const rows = await this.#db.execute<{ scalar: number }>( + sql`SELECT MAX(scalar)::real AS scalar + FROM fitness.metric_stream + WHERE user_id = ${window.userId}::uuid + AND activity_id IN (${sql.join(activityIdClauses, sql`, `)}) + AND channel = ${channel} + AND scalar IS NOT NULL + GROUP BY activity_id, recorded_at + ORDER BY recorded_at`, + ); + return rows.map((row) => row.scalar); + } + async #samplesForChannel( days: number, userId: string, diff --git a/packages/server/src/routers/inertial-measurement-unit-sync.test.ts b/packages/server/src/routers/inertial-measurement-unit-sync.test.ts index d29c0838f5..c5415c0401 100644 --- a/packages/server/src/routers/inertial-measurement-unit-sync.test.ts +++ b/packages/server/src/routers/inertial-measurement-unit-sync.test.ts @@ -135,13 +135,51 @@ describe("inertialMeasurementUnitSyncRouter", () => { ).rejects.toThrow(); }); + it("rejects samples more than five minutes in the future", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-03-25T10:00:00.000Z")); + + const execute = makeExecute(); + const caller = createCaller({ db: { execute }, userId: "user-1" }); + + try { + await expect( + caller.pushSamples({ + deviceId: "WHOOP Strap", + deviceType: "whoop", + samples: [makeSample({ timestamp: "2026-03-25T10:06:00.001Z" })], + }), + ).rejects.toThrow("IMU sample timestamp is too far in the future"); + + expect(execute).toHaveBeenCalledTimes(0); + } finally { + vi.useRealTimers(); + } + }); + + it("rejects malformed sample timestamps", async () => { + const execute = makeExecute(); + const caller = createCaller({ db: { execute }, userId: "user-1" }); + + await expect( + caller.pushSamples({ + deviceId: "WHOOP Strap", + deviceType: "whoop", + samples: [makeSample({ timestamp: "not-a-timestamp" })], + }), + ).rejects.toThrow("IMU sample timestamp is invalid"); + + expect(execute).toHaveBeenCalledTimes(0); + }); + it("batches large sample arrays into multiple INSERT statements", async () => { + const startTimestamp = new Date("2026-03-25T10:00:00.020Z"); const execute = makeExecute(); const caller = createCaller({ db: { execute }, userId: "user-1" }); const samples = Array.from({ length: 7500 }, (_, index) => makeSample({ - timestamp: `2026-03-25T10:00:${String(Math.floor(index / 50)).padStart(2, "0")}.${String((index % 50) * 20).padStart(3, "0")}Z`, + timestamp: new Date(startTimestamp.getTime() + index * 20).toISOString(), }), ); diff --git a/packages/server/src/routers/inertial-measurement-unit-sync.ts b/packages/server/src/routers/inertial-measurement-unit-sync.ts index 77fe034d04..da5e828067 100644 --- a/packages/server/src/routers/inertial-measurement-unit-sync.ts +++ b/packages/server/src/routers/inertial-measurement-unit-sync.ts @@ -3,6 +3,7 @@ import { z } from "zod"; import { SOURCE_TYPE_API } from "../../../../src/db/sensor-channels.ts"; import { logger } from "../logger.ts"; import { protectedProcedure, router } from "../trpc.ts"; +import { rejectFutureSamples } from "./sample-validation.ts"; const PROVIDER_ID = "apple_motion"; const INSERT_BATCH_SIZE = 5000; @@ -83,9 +84,8 @@ async function insertBatch( export const inertialMeasurementUnitSyncRouter = router({ pushSamples: protectedProcedure.input(pushSamplesInput).mutation(async ({ ctx, input }) => { - await ensureProvider(ctx.db, ctx.userId); - if (input.samples.length === 0) { + await ensureProvider(ctx.db, ctx.userId); logger.info("IMU push with 0 samples", { userId: ctx.userId, deviceId: input.deviceId, @@ -94,10 +94,14 @@ export const inertialMeasurementUnitSyncRouter = router({ return { inserted: 0 }; } + const now = new Date(); + rejectFutureSamples(input.samples, now, "IMU"); + await ensureProvider(ctx.db, ctx.userId); + // Log timestamp range to detect stale/future data const firstTimestamp = input.samples[0]?.timestamp; const lastTimestamp = input.samples[input.samples.length - 1]?.timestamp; - const nowIso = new Date().toISOString(); + const nowIso = now.toISOString(); const inserted = await insertBatch( ctx.db, diff --git a/packages/server/src/routers/sample-validation.ts b/packages/server/src/routers/sample-validation.ts new file mode 100644 index 0000000000..47b9457032 --- /dev/null +++ b/packages/server/src/routers/sample-validation.ts @@ -0,0 +1,36 @@ +import { TRPCError } from "@trpc/server"; + +export const MAX_FUTURE_SAMPLE_SKEW_MS = 5 * 60 * 1000; + +type TimestampedSample = { + readonly timestamp: string; +}; + +export function rejectFutureSamples( + samples: readonly TimestampedSample[], + now: Date, + source: string, +) { + const futureLimitMs = now.getTime() + MAX_FUTURE_SAMPLE_SKEW_MS; + + const invalidSample = samples.find((sample) => { + const sampleTimeMs = Date.parse(sample.timestamp); + if (!Number.isFinite(sampleTimeMs)) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: `${source} sample timestamp is invalid: ${sample.timestamp}`, + }); + } + + return sampleTimeMs > futureLimitMs; + }); + + if (!invalidSample) { + return; + } + + throw new TRPCError({ + code: "BAD_REQUEST", + message: `${source} sample timestamp is too far in the future: ${invalidSample.timestamp}`, + }); +} diff --git a/packages/server/src/routers/whoop-ble-sync.test.ts b/packages/server/src/routers/whoop-ble-sync.test.ts index 48abb7283d..e4859e2be2 100644 --- a/packages/server/src/routers/whoop-ble-sync.test.ts +++ b/packages/server/src/routers/whoop-ble-sync.test.ts @@ -391,6 +391,57 @@ describe("whoopBleSyncRouter", () => { ).rejects.toThrow(); }); + it("rejects samples more than five minutes in the future", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-03-30T12:00:00.000Z")); + + const trpcCaller = caller(ctx); + + try { + await expect( + trpcCaller.pushRealtimeData({ + deviceId: "WHOOP Strap", + samples: [ + { + timestamp: "2026-03-30T12:06:00.001Z", + rrIntervalMs: 812, + quaternionW: 1.0, + quaternionX: 0.0, + quaternionY: 0.0, + quaternionZ: 0.0, + }, + ], + }), + ).rejects.toThrow("WHOOP BLE sample timestamp is too far in the future"); + + expect(mockDb.execute).toHaveBeenCalledTimes(0); + } finally { + vi.useRealTimers(); + } + }); + + it("rejects malformed sample timestamps", async () => { + const trpcCaller = caller(ctx); + + await expect( + trpcCaller.pushRealtimeData({ + deviceId: "WHOOP Strap", + samples: [ + { + timestamp: "not-a-timestamp", + rrIntervalMs: 812, + quaternionW: 1.0, + quaternionX: 0.0, + quaternionY: 0.0, + quaternionZ: 0.0, + }, + ], + }), + ).rejects.toThrow("WHOOP BLE sample timestamp is invalid"); + + expect(mockDb.execute).toHaveBeenCalledTimes(0); + }); + it("handles batch splitting for large sample arrays", async () => { const trpcCaller = caller(ctx); // Create 2500 samples (exceeds INSERT_BATCH_SIZE of 2000) diff --git a/packages/server/src/routers/whoop-ble-sync.ts b/packages/server/src/routers/whoop-ble-sync.ts index 80424da10e..4ad7552a99 100644 --- a/packages/server/src/routers/whoop-ble-sync.ts +++ b/packages/server/src/routers/whoop-ble-sync.ts @@ -3,6 +3,7 @@ import { z } from "zod"; import { RR_INTERVAL_MS } from "../../../../src/db/sensor-channels.ts"; import { logger } from "../logger.ts"; import { protectedProcedure, router } from "../trpc.ts"; +import { rejectFutureSamples } from "./sample-validation.ts"; const PROVIDER_ID = "whoop_ble"; const INSERT_BATCH_SIZE = 2000; @@ -101,13 +102,16 @@ export const whoopBleSyncRouter = router({ pushRealtimeData: protectedProcedure .input(pushRealtimeDataInput) .mutation(async ({ ctx, input }) => { - await ensureProvider(ctx.db, ctx.userId); - if (input.samples.length === 0) { + await ensureProvider(ctx.db, ctx.userId); logger.info("WHOOP BLE realtime push with 0 samples", { userId: ctx.userId }); return { inserted: 0 }; } + const now = new Date(); + rejectFutureSamples(input.samples, now, "WHOOP BLE"); + await ensureProvider(ctx.db, ctx.userId); + const firstTimestamp = input.samples[0]?.timestamp; const lastTimestamp = input.samples[input.samples.length - 1]?.timestamp; @@ -124,7 +128,7 @@ export const whoopBleSyncRouter = router({ sampleCount: inserted, firstTimestamp, lastTimestamp, - serverTime: new Date().toISOString(), + serverTime: now.toISOString(), }); return { inserted }; diff --git a/src/db/schema.ts b/src/db/schema.ts index 724b000e71..2c735c6e18 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -588,7 +588,6 @@ export const metricStream = fitness.table( ), primaryKey({ columns: [table.id, table.recordedAt] }), index("metric_stream_user_channel_time_idx").on(table.userId, table.channel, table.recordedAt), - index("metric_stream_provider_time_idx").on(table.providerId, table.recordedAt), ], );