Fix metric_stream CI failures and future-sample guards

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.

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

docs/schema.dbml

@@ -758,7 +758,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']
   }
 }
 

drizzle/0011_metric_stream_storage_controls.sql

@@ -0,0 +1,7 @@
+SELECT public.set_chunk_time_interval('fitness.metric_stream', INTERVAL '1 day');
+--> statement-breakpoint
+SET lock_timeout = '10s';
+--> statement-breakpoint
+DROP INDEX IF EXISTS fitness.metric_stream_provider_time_idx;
+--> statement-breakpoint
+DROP INDEX IF EXISTS fitness.metric_stream_recorded_at_idx;

packages/server/src/routers/inertial-measurement-unit-sync.test.ts

@@ -135,6 +135,25 @@ 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" });
+
+      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);
+      vi.useRealTimers();
+    });
+
     it("batches large sample arrays into multiple INSERT statements", async () => {
       const execute = makeExecute();
       const caller = createCaller({ db: { execute }, userId: "user-1" });

packages/server/src/routers/inertial-measurement-unit-sync.ts

@@ -1,3 +1,4 @@
+import { TRPCError } from "@trpc/server";
 import { sql } from "drizzle-orm";
 import { z } from "zod";
 import { SOURCE_TYPE_API } from "../../../../src/db/sensor-channels.ts";
@@ -6,6 +7,7 @@ import { protectedProcedure, router } from "../trpc.ts";
 
 const PROVIDER_ID = "apple_motion";
 const INSERT_BATCH_SIZE = 5000;
+const MAX_FUTURE_SAMPLE_SKEW_MS = 5 * 60 * 1000;
 
 // ── Zod schemas ──
 
@@ -38,6 +40,21 @@ async function ensureProvider(db: Database, userId: string) {
   );
 }
 
+function rejectFutureSamples(samples: InertialMeasurementUnitSample[], now: Date) {
+  const futureLimitMs = now.getTime() + MAX_FUTURE_SAMPLE_SKEW_MS;
+  const futureSample = samples.find((sample) => {
+    const sampleTimeMs = Date.parse(sample.timestamp);
+    return Number.isFinite(sampleTimeMs) && sampleTimeMs > futureLimitMs;
+  });
+
+  if (!futureSample) return;
+
+  throw new TRPCError({
+    code: "BAD_REQUEST",
+    message: `IMU sample timestamp is too far in the future: ${futureSample.timestamp}`,
+  });
+}
+
 /**
  * Bulk-insert IMU samples using multi-row VALUES.
  * At 50 Hz, a 12-hour sync produces ~2.16M samples.
@@ -83,9 +100,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 +110,14 @@ export const inertialMeasurementUnitSyncRouter = router({
       return { inserted: 0 };
     }
 
+    const now = new Date();
+    rejectFutureSamples(input.samples, now);
+    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,

packages/server/src/routers/whoop-ble-sync.test.ts

@@ -391,6 +391,32 @@ 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);
+
+      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);
+      vi.useRealTimers();
+    });
+
     it("handles batch splitting for large sample arrays", async () => {
       const trpcCaller = caller(ctx);
       // Create 2500 samples (exceeds INSERT_BATCH_SIZE of 2000)

packages/server/src/routers/whoop-ble-sync.ts

@@ -1,3 +1,4 @@
+import { TRPCError } from "@trpc/server";
 import { sql } from "drizzle-orm";
 import { z } from "zod";
 import { RR_INTERVAL_MS } from "../../../../src/db/sensor-channels.ts";
@@ -6,6 +7,7 @@ import { protectedProcedure, router } from "../trpc.ts";
 
 const PROVIDER_ID = "whoop_ble";
 const INSERT_BATCH_SIZE = 2000;
+const MAX_FUTURE_SAMPLE_SKEW_MS = 5 * 60 * 1000;
 
 // ── Zod schemas ──
 
@@ -42,6 +44,21 @@ async function ensureProvider(database: Database, userId: string) {
   );
 }
 
+function rejectFutureSamples(samples: WhoopBleRealtimeDataSample[], now: Date) {
+  const futureLimitMs = now.getTime() + MAX_FUTURE_SAMPLE_SKEW_MS;
+  const futureSample = samples.find((sample) => {
+    const sampleTimeMs = Date.parse(sample.timestamp);
+    return Number.isFinite(sampleTimeMs) && sampleTimeMs > futureLimitMs;
+  });
+
+  if (!futureSample) return;
+
+  throw new TRPCError({
+    code: "BAD_REQUEST",
+    message: `WHOOP BLE sample timestamp is too far in the future: ${futureSample.timestamp}`,
+  });
+}
+
 /** Insert WHOOP BLE realtime samples into metric_stream. */
 async function insertRealtimeDataBatch(
   database: Database,
@@ -101,13 +118,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);
+      await ensureProvider(ctx.db, ctx.userId);
+
       const firstTimestamp = input.samples[0]?.timestamp;
       const lastTimestamp = input.samples[input.samples.length - 1]?.timestamp;
 
@@ -124,7 +144,7 @@ export const whoopBleSyncRouter = router({
         sampleCount: inserted,
         firstTimestamp,
         lastTimestamp,
-        serverTime: new Date().toISOString(),
+        serverTime: now.toISOString(),
       });
 
       return { inserted };

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),
   ],
 );