diff --git a/analytics/macros/bounded_activity_graph.sql b/analytics/macros/bounded_activity_graph.sql new file mode 100644 index 0000000000..bff2e7f210 --- /dev/null +++ b/analytics/macros/bounded_activity_graph.sql @@ -0,0 +1,206 @@ +{% macro bounded_activity_graph() %} +{% set batch_start_value = config.get("__dbt_internal_microbatch_event_time_start") %} +{% set batch_end_value = config.get("__dbt_internal_microbatch_event_time_end") %} +{% if batch_start_value and batch_end_value %} +{% set batch_start = batch_start_value.strftime("%Y-%m-%d %H:%M:%S") %} +{% set batch_end = batch_end_value.strftime("%Y-%m-%d %H:%M:%S") %} +{% endif %} +active_activity AS ( + SELECT * + FROM {{ source('postgres_fitness', 'activity') }} FINAL + WHERE _peerdb_is_deleted = 0 + {% if batch_start_value and batch_end_value %} + AND started_at < toDateTime64('{{ batch_end }}', 6, 'UTC') + AND coalesce(ended_at, started_at + INTERVAL 12 HOUR) >= toDateTime64('{{ batch_start }}', 6, 'UTC') + {% endif %} +), + +active_provider_priority AS ( + SELECT * + FROM {{ source('postgres_fitness', 'provider_priority') }} FINAL + WHERE _peerdb_is_deleted = 0 +), + +active_device_priority AS ( + SELECT * + FROM {{ source('postgres_fitness', 'device_priority') }} FINAL + WHERE _peerdb_is_deleted = 0 +), + +device_priority_match AS ( + SELECT + activity_id, + priority + FROM ( + SELECT + active_activity.id AS activity_id, + active_device_priority.priority AS priority, + row_number() OVER ( + PARTITION BY active_activity.id + ORDER BY length(active_device_priority.source_name_pattern) DESC + ) AS row_number + FROM active_activity + INNER JOIN active_device_priority + ON active_device_priority.provider_id = active_activity.provider_id + AND active_activity.source_name LIKE active_device_priority.source_name_pattern + ) + WHERE row_number = 1 +), + +ranked AS ( + SELECT + active_activity.id AS id, + active_activity.provider_id AS provider_id, + active_activity.user_id AS user_id, + active_activity.external_id AS external_id, + active_activity.activity_type AS activity_type, + active_activity.started_at AS started_at, + active_activity.ended_at AS ended_at, + active_activity.source_name AS source_name, + active_activity.name AS name, + active_activity.notes AS notes, + active_activity.timezone AS timezone, + active_activity.raw AS raw, + coalesce(device_priority_match.priority, active_provider_priority.priority, 100) AS priority + FROM active_activity + LEFT JOIN active_provider_priority + ON active_provider_priority.provider_id = active_activity.provider_id + LEFT JOIN device_priority_match + ON device_priority_match.activity_id = active_activity.id +), + +pairs AS ( + SELECT + left_activity.id AS id1, + right_activity.id AS id2 + FROM ranked AS left_activity + INNER JOIN ranked AS right_activity + ON left_activity.user_id = right_activity.user_id + AND toString(left_activity.id) < toString(right_activity.id) + AND dateDiff( + 'second', + greatest(left_activity.started_at, right_activity.started_at), + least( + coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 12 HOUR), + coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 12 HOUR) + ) + ) / nullIf(dateDiff( + 'second', + least(left_activity.started_at, right_activity.started_at), + greatest( + coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 12 HOUR), + coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 12 HOUR) + ) + ), 0) > 0.8 +), + +graph_edges AS ( + SELECT + id1 AS from_id, + id2 AS to_id + FROM pairs + UNION ALL + SELECT + id2 AS from_id, + id1 AS to_id + FROM pairs +), + +connected_components AS ( + SELECT + id AS activity_id, + id AS connected_activity_id, + [toString(id)] AS visited_activity_ids + FROM ranked + UNION ALL + SELECT + connected_components.activity_id AS activity_id, + graph_edges.to_id AS connected_activity_id, + arrayConcat(connected_components.visited_activity_ids, [toString(graph_edges.to_id)]) AS visited_activity_ids + FROM connected_components + INNER JOIN graph_edges + ON graph_edges.from_id = connected_components.connected_activity_id + WHERE NOT has(connected_components.visited_activity_ids, toString(graph_edges.to_id)) +), + +final_groups AS ( + SELECT + activity_id, + min(toString(connected_activity_id)) AS group_id + FROM connected_components + GROUP BY activity_id +), + +best AS ( + SELECT * + FROM ( + SELECT + final_groups.group_id AS group_id, + ranked.id AS canonical_id, + ranked.provider_id AS provider_id, + ranked.user_id AS user_id, + ranked.activity_type AS activity_type, + ranked.started_at AS started_at, + ranked.ended_at AS ended_at, + ranked.source_name AS source_name, + ranked.priority AS priority, + row_number() OVER ( + PARTITION BY final_groups.group_id + ORDER BY ranked.priority ASC, toString(ranked.id) ASC + ) AS row_number + FROM final_groups + INNER JOIN ranked + ON ranked.id = final_groups.activity_id + ) + WHERE row_number = 1 +), + +merged AS ( + SELECT + best.group_id AS group_id, + best.canonical_id AS id, + any(best.provider_id) AS provider_id, + any(best.user_id) AS user_id, + any(best.activity_type) AS activity_type, + min(ranked.started_at) AS started_at, + max(coalesce(ranked.ended_at, ranked.started_at + INTERVAL 12 HOUR)) AS ended_at, + any(best.source_name) AS source_name, + argMinIf(ranked.name, ranked.priority, ranked.name IS NOT NULL) AS name, + argMinIf(ranked.notes, ranked.priority, ranked.notes IS NOT NULL) AS notes, + argMinIf(ranked.timezone, ranked.priority, ranked.timezone IS NOT NULL) AS timezone, + argMinIf(ranked.raw, ranked.priority, ranked.raw IS NOT NULL) AS raw, + arraySort(groupUniqArray(ranked.provider_id)) AS source_providers, + groupArrayIf( + map('providerId', ranked.provider_id, 'externalId', ranked.external_id), + ranked.external_id IS NOT NULL AND ranked.external_id != '' + ) AS source_external_ids, + groupArray(ranked.id) AS member_activity_ids + FROM best + INNER JOIN final_groups + ON final_groups.group_id = best.group_id + INNER JOIN ranked + ON ranked.id = final_groups.activity_id + GROUP BY best.group_id, best.canonical_id +), + +current_activity AS ( + SELECT + id AS activity_id, + user_id, + activity_type, + name, + started_at, + ended_at + FROM merged +), + +activity_members AS ( + SELECT + id AS activity_id, + user_id, + started_at, + ended_at, + arrayJoin(member_activity_ids) AS member_activity_id + FROM merged +) +{% endmacro %} diff --git a/analytics/models/read_models/activity_location_sample.sql b/analytics/models/read_models/activity_location_sample.sql index bec1c41f27..90c7e15cef 100644 --- a/analytics/models/read_models/activity_location_sample.sql +++ b/analytics/models/read_models/activity_location_sample.sql @@ -15,13 +15,7 @@ } ) }} -WITH activity_members AS ( - SELECT - activity_id, - user_id, - member_activity_id - FROM {{ source('analytics', 'v_activity_members') }} -), +WITH RECURSIVE {{ bounded_activity_graph() }}, location_versions AS ( SELECT * diff --git a/analytics/models/read_models/activity_sensor_sample.sql b/analytics/models/read_models/activity_sensor_sample.sql index f3959062a0..f8480d7cec 100644 --- a/analytics/models/read_models/activity_sensor_sample.sql +++ b/analytics/models/read_models/activity_sensor_sample.sql @@ -15,14 +15,7 @@ } ) }} -WITH current_activity AS ( - SELECT - id AS activity_id, - user_id, - started_at, - coalesce(ended_at, started_at + INTERVAL 12 HOUR) AS ended_at - FROM {{ source('analytics', 'v_activity') }} -), +WITH RECURSIVE {{ bounded_activity_graph() }}, activity_samples AS ( SELECT @@ -37,7 +30,7 @@ activity_samples AS ( INNER JOIN current_activity ON current_activity.user_id = samples.user_id AND samples.recorded_at >= current_activity.started_at - AND samples.recorded_at <= current_activity.ended_at + AND samples.recorded_at <= coalesce(current_activity.ended_at, current_activity.started_at + INTERVAL 12 HOUR) ) SELECT diff --git a/analytics/models/read_models/read_model_microbatch.sql.test.ts b/analytics/models/read_models/read_model_microbatch.sql.test.ts index 7dde67dc18..38df221ba8 100644 --- a/analytics/models/read_models/read_model_microbatch.sql.test.ts +++ b/analytics/models/read_models/read_model_microbatch.sql.test.ts @@ -14,6 +14,14 @@ function compactWhitespace(value: string): string { } describe("production analytics read-model build", () => { + it("does not block the BullMQ worker on analytics dbt builds", () => { + const entrypoint = readProjectFile("entrypoint.sh"); + const workerBlockMatch = entrypoint.match(/ worker\)\n(?[\s\S]*?)\n ;;/); + + expect(workerBlockMatch?.groups?.body).toContain("exec $NODE src/jobs/worker.ts"); + expect(workerBlockMatch?.groups?.body).not.toContain("dbt build"); + }); + it("runs every bounded intermediary and final read model in dependency order", () => { const entrypoint = readProjectFile("entrypoint.sh"); const safeModelMatch = entrypoint.match(/^DBT_SAFE_MODELS="([^"]+)"$/m); @@ -61,7 +69,9 @@ describe("production analytics read-model build", () => { expect(sql).toContain("incremental_strategy='microbatch'"); expect(sql).toContain("event_time='recorded_at'"); expect(sql).toContain("ref('deduped_sensor')"); - expect(sql).toContain("source('analytics', 'v_activity')"); + expect(sql).toContain("WITH RECURSIVE {{ bounded_activity_graph() }}"); + expect(sql).toContain("bounded_activity_graph()"); + expect(sql).not.toContain("source('analytics', 'v_activity')"); expect(sql).toContain("activity_id"); }); @@ -72,10 +82,40 @@ describe("production analytics read-model build", () => { expect(sql).toContain("incremental_strategy='microbatch'"); expect(sql).toContain("event_time='recorded_at'"); expect(sql).toContain("source('postgres_fitness', 'metric_stream')"); - expect(sql).toContain("source('analytics', 'v_activity_members')"); + expect(sql).toContain("WITH RECURSIVE {{ bounded_activity_graph() }}"); + expect(sql).toContain("bounded_activity_graph()"); + expect(sql).not.toContain("source('analytics', 'v_activity_members')"); expect(sql).toContain("channel = 'location'"); }); + it("bounds activity graph construction to the active microbatch window", () => { + const sql = readProjectFile("analytics/macros/bounded_activity_graph.sql"); + + expect(sql).toContain("__dbt_internal_microbatch_event_time_start"); + expect(sql).toContain("__dbt_internal_microbatch_event_time_end"); + expect(sql).toContain("started_at < toDateTime64('{{ batch_end }}', 6, 'UTC')"); + expect(sql).toContain( + "coalesce(ended_at, started_at + INTERVAL 12 HOUR) >= toDateTime64('{{ batch_start }}', 6, 'UTC')", + ); + }); + + it("uses the same null-ended activity window for overlap matching", () => { + const sql = readProjectFile("analytics/macros/bounded_activity_graph.sql"); + + expect(sql).toContain("coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 12 HOUR)"); + expect(sql).toContain("coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 12 HOUR)"); + expect(sql).not.toContain("INTERVAL 1 HOUR"); + }); + + it("uses merged group time bounds for current activity sensor membership", () => { + const sql = readProjectFile("analytics/macros/bounded_activity_graph.sql"); + + expect(sql).toContain("min(ranked.started_at) AS started_at"); + expect(sql).toContain("max(coalesce(ranked.ended_at, ranked.started_at + INTERVAL 12 HOUR)) AS ended_at"); + expect(sql).not.toContain("any(best.started_at) AS started_at"); + expect(sql).not.toContain("any(best.ended_at) AS ended_at"); + }); + it("aggregates activity sensor summary from the bounded sensor intermediary", () => { expect(existsSync(new URL("./activity_sensor_summary_rows.sql", import.meta.url))).toBe(true); const sql = readModel("activity_sensor_summary_rows"); diff --git a/analytics/models/sources.yml b/analytics/models/sources.yml index 5ee4f8e7d1..817b42c831 100644 --- a/analytics/models/sources.yml +++ b/analytics/models/sources.yml @@ -10,9 +10,11 @@ sources: schema: postgres_fitness tables: - name: activity + - name: device_priority - name: metric_stream config: event_time: recorded_at + - name: provider_priority - name: sensor_device_priority - name: sensor_provider_priority - name: sleep_session diff --git a/docs/production-incident-baseline.md b/docs/production-incident-baseline.md index 0605568ce3..9ec6f281b6 100644 --- a/docs/production-incident-baseline.md +++ b/docs/production-incident-baseline.md @@ -8827,6 +8827,47 @@ new incremental tables are populated. update; add freshness monitoring comparing latest sleep/heart-rate inputs to latest active RHR output date. +### Production worker restart loop from analytics dbt OOM + +- Date: 2026-05-26 PT / 2026-05-27 UTC. +- Symptoms: Production was reported down. Public `/healthz` and the SPA shell + still returned HTTP 200, but background processing was impaired: the + `dofek_worker` service was repeatedly failing with `task: non-zero exit (1)`, + and `dofek_analytics-worker` was retrying dbt builds. +- User Impact: Background sync and import jobs were unavailable while the + worker was in a restart loop. ClickHouse-backed dashboard data was at risk of + stale or failed reads while ClickHouse was under memory pressure. +- Evidence: The first fatal analytics-worker log lines were ClickHouse + `MEMORY_LIMIT_EXCEEDED` exceptions in dbt models including + `activity_location_sample` and `activity_sensor_sample`, with ClickHouse at + its configured 3 GiB memory limit while executing recursive activity graph + work. The failing batch for `2026-05-20` had only 458 heart-rate sample rows + but used `analytics.v_activity`/`analytics.v_activity_members`, which forced + ClickHouse to traverse the unbounded 1,029-activity user graph before the + microbatch could finish. The worker logs also showed the `worker` entrypoint + running Postgres migrations, ClickHouse migrations, and then `dbt build` + before the BullMQ worker process could start. +- Root Cause: The activity sample dbt models used global recursive activity + views inside each microbatch. ClickHouse did not push the day filter inside + those views, so each batch recomputed the full dedupe graph and OOMed. The + blast radius was larger because `entrypoint.sh` also coupled the BullMQ + `worker` service startup to the same analytics dbt build. +- Fix/Mitigation: A short-lived manual mitigation stopped the failing dbt retry + loop and temporarily ran the Node BullMQ worker directly so background jobs + could resume. The durable code change removes `dbt build` from the `worker` + entrypoint and bounds the activity graph inside the affected analytics models + to the active microbatch window so `analytics-worker` remains enabled. +- Validation: After the current production image was redeployed, + `dofek_web` was `2/2`, `dofek_worker` was `1/1`, and + `dofek_analytics-worker` was `1/1`. `/healthz` returned HTTP 200. The local + fix compiles the affected dbt models with batch-local activity graph bounds + and passes the changed test suite. +- Remaining Risk: Medium until the bounded activity graph fix is deployed and + the next scheduled analytics-worker run completes cleanly on production-scale + data. +- Follow-Up Work: Add freshness monitoring for analytics read models and keep + the BullMQ worker independent from analytics rebuilds. + ### Healthspan activity and steps undercount - Date: 2026-05-26. diff --git a/entrypoint.sh b/entrypoint.sh index 53d60dc4d9..8fdf34e624 100755 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -31,7 +31,6 @@ case "${1:-sync}" in ;; worker) $NODE src/db/run-migrate.ts - dbt build --project-dir analytics --profiles-dir analytics --threads 1 --select $DBT_SAFE_MODELS exec $NODE src/jobs/worker.ts ;; migrate) diff --git a/src/providers/ultrahuman-sync.integration.test.ts b/src/providers/ultrahuman-sync.integration.test.ts index 3cbb373307..6d17dcd00d 100644 --- a/src/providers/ultrahuman-sync.integration.test.ts +++ b/src/providers/ultrahuman-sync.integration.test.ts @@ -1,7 +1,7 @@ import { eq } from "drizzle-orm"; import { HttpResponse, http } from "msw"; import { setupServer } from "msw/node"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { dailyMetrics, sleepSession } from "../db/schema.ts"; import { setupTestDatabase, type TestContext } from "../db/test-helpers.ts"; import { ensureProvider, saveTokens } from "../db/tokens.ts"; @@ -124,6 +124,7 @@ describe("UltrahumanProvider.sync() (integration)", () => { "Ultrahuman", "https://partner.ultrahuman.com/api/v1", ); + vi.useFakeTimers({ now: new Date("2026-03-15T12:00:00Z"), toFake: ["Date"] }); }, 60_000); afterEach(() => { @@ -132,6 +133,7 @@ describe("UltrahumanProvider.sync() (integration)", () => { afterAll(async () => { server.close(); + vi.useRealTimers(); if (ctx) await ctx.cleanup(); });