feat(data-warehouse): sync staged rows to person properties via kafka - #70025
Conversation
Gilbert09
added
the
skip-inkeep-docs
|
Reviews (1): Last reviewed commit: "feat(data-warehouse): sync staged rows t..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
🤖 CI report
|
| Root | Eager (shipped) | Δ vs base | Budget |
|---|---|---|---|
entry (logged-out pages, app bootstrap)src/index.tsx |
1.22 MiB · 22 files | no change | ███░░░░░░░ 28.4% of 4.29 MiB |
authenticated shell (every logged-in page)src/scenes/AuthenticatedShell.tsx |
8.15 MiB · 2,978 files | 🔺 +58 B (+0.0%) | █████████░ 88.1% of 9.25 MiB |
🟢 node_modules/monaco-editor/ stays out of src/index.tsx
🟢 src/lib/components/ActivityLog/describers stays out of src/index.tsx
🟢 [object Object] stays out of src/index.tsx
🟢 [object Object] stays out of src/index.tsx
🟢 node_modules/monaco-editor/ stays out of src/scenes/AuthenticatedShell.tsx
🟢 src/lib/components/ActivityLog/describers stays out of src/scenes/AuthenticatedShell.tsx
🟢 [object Object] stays out of src/scenes/AuthenticatedShell.tsx
🟢 [object Object] stays out of src/scenes/AuthenticatedShell.tsx
Largest files eagerly shipped from src/index.tsx
| Size | File |
|---|---|
| 126.8 KiB | ../node_modules/.pnpm/react-dom@18.3.1_react@18.3.1/node_modules/react-dom/cjs/react-dom.production.min.js |
| 24.6 KiB | ../node_modules/.pnpm/buffer@6.0.3/node_modules/buffer/index.js |
| 6.3 KiB | ../node_modules/.pnpm/react@18.3.1/node_modules/react/cjs/react.production.min.js |
| 4.5 KiB | ../node_modules/.pnpm/@jspm+core@2.1.0/node_modules/@jspm/core/nodelibs/browser/process.js |
| 3.9 KiB | ../node_modules/.pnpm/scheduler@0.23.2/node_modules/scheduler/cjs/scheduler.production.min.js |
| 1.4 KiB | ../node_modules/.pnpm/base64-js@1.5.1/node_modules/base64-js/index.js |
| 1.3 KiB | src/RootErrorBoundary.tsx |
| 912 B | ../node_modules/.pnpm/ieee754@1.2.1/node_modules/ieee754/index.js |
| 789 B | src/scenes/ChunkLoadErrorBoundary.tsx |
| 762 B | src/index.tsx |
Largest files eagerly shipped from src/scenes/AuthenticatedShell.tsx
| Size | File |
|---|---|
| 281.1 KiB | ../node_modules/.pnpm/posthog-js@1.401.0/node_modules/posthog-js/dist/rrweb.js |
| 267.7 KiB | ../node_modules/.pnpm/@posthog+icons@0.38.0_react-dom@18.3.1_react@18.3.1__react@18.3.1/node_modules/@posthog/icons/dist/posthog-icons.es.js |
| 235.5 KiB | src/taxonomy/core-filter-definitions-by-group.json |
| 222.7 KiB | ../node_modules/.pnpm/posthog-js@1.401.0/node_modules/posthog-js/dist/module.js |
| 164.0 KiB | src/queries/validators.js |
| 154.3 KiB | ../node_modules/.pnpm/re2js@0.4.1/node_modules/re2js/build/index.esm.js |
| 126.8 KiB | ../node_modules/.pnpm/react-dom@18.3.1_react@18.3.1/node_modules/react-dom/cjs/react-dom.production.min.js |
| 106.1 KiB | src/lib/api.ts |
| 93.3 KiB | ../node_modules/.pnpm/prosemirror-view@1.40.1/node_modules/prosemirror-view/dist/index.js |
| 92.7 KiB | ../packages/quill/packages/quill/dist/index.js |
Posted automatically by check-eager-graph · sizes are eager output bytes (shipped, post-tree-shake) from the esbuild metafile · part of #32479
✅ Dist folder size — 🔺 +2.8 KiB (+0.0%)
Total size of the built frontend/dist folder (all assets), compared against the base branch.
Total: 1307.94 MiB · 🔺 +2.8 KiB (+0.0%)
⚠️ Backend snapshots — 1 updated (1 modified, 0 added, 0 deleted)
Query snapshots: Backend query snapshots updated
Changes: 1 snapshots (1 modified, 0 added, 0 deleted)
What this means:
- Query snapshots have been automatically updated to match current output
- These changes reflect modifications to database queries or schema
Next steps:
- Review the query changes to ensure they're intentional
- If unexpected, investigate what caused the query to change
⚠️ Backend coverage — 74.0% of changed backend lines covered — 92 uncovered
🧪 Backend test coverage
Patch coverage — changed backend lines (products + core): ███████████████░░░░░ 74.0% (262 / 354)
| File | Patch | Uncovered changed lines |
|---|---|---|
products/customer_analytics/backend/presentation/views/views.py |
50.0% | 258 |
products/warehouse_sources/backend/temporal/data_imports/person_property_sync_job.py |
57.1% | 64–69, 72, 75–78, 80–82, 89–90, 92, 101, 108–109, 113 |
products/warehouse_sources/backend/temporal/data_imports/external_data_job.py |
60.0% | 416, 574 |
products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline/person_property_sync.py |
62.9% | 114, 118, 122–132, 135–136, 140, 142, 146, 150–157, 161, 163–168, 172–177, 181–183, 189–192, 203–204, 213, 218, 243, 252, 259 |
products/customer_analytics/backend/logic/person_property_projection.py |
65.4% | 31–36, 45–47 |
products/customer_analytics/backend/apps.py |
75.0% | 33, 37 |
products/warehouse_sources/backend/temporal/data_imports/external_product_hooks.py |
85.7% | 172–174, 191 |
🤖 Agents: add a test covering the lines above, or note why under "How did you test this code?". Machine-readable gap list: the patch-coverage artifact on this run (gh run download 29410301253 -n patch-coverage), or the coverage-data block at the end of this comment.
Per-product line coverage (touched products)
| Product | Coverage | Lines |
|---|---|---|
platform_features |
██░░░░░░░░░░░░░░░░░░ 12.1% |
7 / 58 |
batch_exports |
████████░░░░░░░░░░░░ 39.7% |
8,414 / 21,220 |
demo |
███████████░░░░░░░░░ 56.2% |
1,497 / 2,663 |
warehouse_sources_queue |
████████████░░░░░░░░ 59.2% |
148 / 250 |
tasks |
█████████████░░░░░░░ 67.4% |
25,425 / 37,703 |
data_tools |
██████████████░░░░░░ 70.0% |
63 / 90 |
ai_gateway |
███████████████░░░░░ 75.0% |
9 / 12 |
signals |
████████████████░░░░ 79.1% |
19,072 / 24,106 |
data_modeling |
████████████████░░░░ 80.0% |
4,834 / 6,045 |
cdp |
████████████████░░░░ 80.7% |
3,118 / 3,864 |
wizard |
████████████████░░░░ 82.5% |
772 / 936 |
agent_platform |
█████████████████░░░ 84.2% |
3,112 / 3,695 |
notebooks |
█████████████████░░░ 84.3% |
6,323 / 7,501 |
cohorts |
█████████████████░░░ 86.0% |
3,989 / 4,639 |
actions |
█████████████████░░░ 86.6% |
717 / 828 |
product_tours |
█████████████████░░░ 87.5% |
1,266 / 1,447 |
exports |
██████████████████░░ 88.3% |
6,861 / 7,769 |
visual_review |
██████████████████░░ 88.5% |
5,565 / 6,287 |
business_knowledge |
██████████████████░░ 88.5% |
4,400 / 4,969 |
conversations |
██████████████████░░ 88.9% |
16,028 / 18,034 |
dashboards |
██████████████████░░ 89.1% |
5,648 / 6,337 |
mcp_analytics |
██████████████████░░ 89.1% |
2,502 / 2,807 |
error_tracking |
██████████████████░░ 89.6% |
9,749 / 10,883 |
engineering_analytics |
██████████████████░░ 89.8% |
4,861 / 5,414 |
streamlit_apps |
██████████████████░░ 90.4% |
2,499 / 2,764 |
slack_app |
██████████████████░░ 90.6% |
9,511 / 10,503 |
links |
██████████████████░░ 90.6% |
183 / 202 |
marketing_analytics |
██████████████████░░ 90.8% |
11,514 / 12,684 |
alerts |
██████████████████░░ 90.9% |
3,416 / 3,760 |
product_analytics |
██████████████████░░ 91.1% |
5,527 / 6,068 |
managed_migrations |
██████████████████░░ 91.9% |
908 / 988 |
workflows |
██████████████████░░ 92.0% |
4,795 / 5,210 |
data_warehouse |
██████████████████░░ 92.1% |
17,872 / 19,415 |
mcp_store |
██████████████████░░ 92.1% |
3,665 / 3,981 |
notifications |
███████████████████░ 92.7% |
1,026 / 1,107 |
web_analytics |
███████████████████░ 92.7% |
13,607 / 14,674 |
ai_observability |
███████████████████░ 92.8% |
14,771 / 15,923 |
surveys |
███████████████████░ 92.9% |
5,660 / 6,094 |
posthog_ai |
███████████████████░ 93.2% |
1,312 / 1,408 |
tracing |
███████████████████░ 93.2% |
2,423 / 2,599 |
approvals |
███████████████████░ 93.3% |
3,395 / 3,640 |
reminders |
███████████████████░ 93.4% |
468 / 501 |
early_access_features |
███████████████████░ 93.8% |
848 / 904 |
legal_documents |
███████████████████░ 94.1% |
1,568 / 1,667 |
endpoints |
███████████████████░ 94.1% |
8,606 / 9,143 |
skills |
███████████████████░ 94.4% |
2,827 / 2,995 |
revenue_analytics |
███████████████████░ 94.5% |
3,598 / 3,809 |
messaging |
███████████████████░ 94.5% |
2,530 / 2,677 |
review_hog |
███████████████████░ 94.6% |
6,475 / 6,848 |
logs |
███████████████████░ 95.3% |
9,525 / 9,991 |
growth |
███████████████████░ 95.5% |
2,734 / 2,864 |
experiments |
███████████████████░ 95.6% |
24,138 / 25,244 |
replay_vision |
███████████████████░ 95.7% |
13,300 / 13,896 |
feature_flags |
███████████████████░ 96.1% |
14,750 / 15,354 |
warehouse_sources |
███████████████████░ 96.1% |
219,391 / 228,196 |
annotations |
███████████████████░ 96.2% |
732 / 761 |
user_interviews |
███████████████████░ 96.4% |
2,242 / 2,325 |
access_control |
███████████████████░ 96.8% |
849 / 877 |
customer_analytics |
███████████████████░ 97.2% |
7,480 / 7,698 |
data_catalog |
███████████████████░ 97.2% |
1,642 / 1,689 |
analytics_platform |
████████████████████ 98.0% |
2,102 / 2,145 |
metrics |
████████████████████ 98.3% |
2,363 / 2,405 |
live_debugger |
████████████████████ 99.2% |
613 / 618 |
field_notes |
████████████████████ 99.4% |
158 / 159 |
Report-only. Patch coverage = changed backend lines covered vs origin/master. Sorted lowest first.
Known gaps: lines covered only by Temporal tests show as uncovered; core line numbers may drift if master changed the same file.
![veria-ai[bot]](https://avatars.githubusercontent.com/in/1723776?s=60&v=4){kind=small}
PR overviewAll previously flagged issues have been addressed. No open security concerns remain on this pull request. Security reviewNo open security issues remain on this pull request. Fixed/addressed: 2 · PR risk: 0/10 |
![graphite-app[bot]](https://avatars.githubusercontent.com/in/158384?s=60&v=4){kind=small}
019debe to
0bd415c
Compare
959eb8b to
491eb9d
Compare
|
⏭️ Skipped snapshot commit because branch advanced to The new commit will trigger its own snapshot update workflow. If you expected this workflow to succeed: This can happen due to concurrent commits. To get a fresh workflow run, either:
|
8914b22 to
663cd6d
Compare
dd1cfe9 to
a55c6d3
Compare
Post-sync child workflow that turns the rows a sync staged into person-property $set intents on Kafka (a throttling consumer sends them to capture next). - sync-warehouse-person-properties workflow + activity (customer_analytics), registered on the signals task queue; started fire-and-forget from the import workflow, gated up front via create_job_model (no no-op child per sync) - per-source S3 snapshot diff so unchanged values are skipped even on full refresh; existence check via personhog so only existing persons are touched; provenance stamped on person PropertyDefinitions - KAFKA_WAREHOUSE_PERSON_PROPERTY_UPDATES topic + PersonPropertySyncActivityInputs contract + person_property_sync_enabled_for gate hook - pure-core tests (bundle/hash/diff) + orchestration test with S3/personhog/Kafka mocked; tach interface exposes the temporal registration symbols Consumer that drains the topic to capture lands next. Part of RFC PostHog/requests-for-comments-internal#1195.
… and harden activity - Produce $set intents via posthog.kafka_client.routing.producer_scope instead of constructing _KafkaProducer directly (satisfies the no-direct-kafka-producer semgrep rule and respects topic->cluster routing). - Decode staged/snapshot parquet off the event loop with asyncio.to_thread so a large chunk can't starve the Temporal activity heartbeater and trip its timeout. - Stamp provenance before advancing the snapshot: the snapshot is the checkpoint that makes rows look unchanged next run, so provenance must be written first; the update is idempotent and safe to repeat on retry. - Drop backend:contract-check (package.json) and its now-dead turbo.json override: registering a Temporal workflow means core imports the product's internals, so it is no longer isolated (mirrors the experiments product). Generated-By: PostHog Code Task-Id: 917feee8-bb06-41ca-98a6-8a3328ba51e8
…nc test write_snapshot is bound to an AsyncMock, whose await_args is typed _Call | None, so the direct .args access failed mypy's union-attr check. Assert it is set before indexing. Generated-By: PostHog Code Task-Id: 917feee8-bb06-41ca-98a6-8a3328ba51e8
f6899c2 to
1eb9b44
Compare
Generated-By: PostHog Code Task-Id: a8be4542-9819-4a1e-8255-12bc6ab729f6
|
Warning Finding on
The queue ID only includes the Slack workspace/channel/thread, even though a workspace can route the same thread to different PostHog integrations. A user authorized only for integration A can send a request that reaches the picker/authorship wait in its child workflow (up to 15 minutes); while it waits, requests for integration B in that same Slack thread are queued behind it. Repeating this lets one tenant deny the Slack-agent workflow to another tenant sharing the workspace. Prompt To Fix With AIInclude the resolved integration (or its PostHog team/organization) in derive_slack_app_mention_workflow_id, so each tenant has an independent per-thread queue. Update the corresponding dispatch and workflow tests to assert that messages resolved to different integrations use separate workflow IDs.Severity: medium | Confidence: 91% | React with 👍 if useful or 👎 if not |
… queue, harden observability
🦔 Hogbox preview · ❌ build failedThe preview didn't come up for commit |
…tomer_analytics facade
2 participants
Note
Third PR in the warehouse → person properties stack: config #69979 (merged) → staging #69983 (merged) → this → throttling consumer #70039. Rebased onto master now that its base has merged.
Problem
PR #69983 stages the changed rows of a sync to S3. This PR consumes them: it turns each staged row into a person-property
$set, keyed on the configured distinct_id column, and hands it off to be written. Ingestion asked that we only write when a value actually changed, only touch existing persons, and pace the writes. This PR does the first two and produces to a Kafka topic; the throttling consumer that paces the send to capture lands in the next PR.Changes
A fire-and-forget child workflow
sync-warehouse-person-properties, started from the import workflow after the signals block and gated up front increate_job_model(so we don't spawn a no-op child on every sync). It runs on theDATA_WAREHOUSE_METADATA_TASK_QUEUE(alongside semantic enrichment and column statistics) so post-sync processing never competes with the sync workers. Per enabled person-target source for the schema, the activity:{property: value}bundle keyed by distinct_id,person_property_snapshot/{team}/{source}/{schema}.parquet, one hash per distinct_id over the source's whole bundle) so unchanged values are skipped even on a full refresh,get_persons_mapped_by_distinct_id) — the check runs post-diff so personhog load scales with changed rows, not table size,$setintent per survivor toKAFKA_WAREHOUSE_PERSON_PROPERTY_UPDATES, keyedteam:distinct_id,warehouse_originon the person PropertyDefinitions (UPDATE-only, so it can't race ingestion's propdef upsert), and clears the staged files.flowchart LR classDef phBlue fill:#1d4aff,stroke:#1d4aff,color:#fff; classDef phGray fill:#e5e7eb,stroke:#c7ccd1,color:#000; classDef phRed fill:#f54e00,stroke:#f54e00,color:#fff; Staged[(staged parquet)]:::phGray --> Diff{{diff vs snapshot}}:::phBlue Diff --> Exists{{personhog: existing?}}:::phBlue Exists --> Kafka[(person-property topic)]:::phRed Diff -. skip unchanged .-> Drop[ ]:::phGray class Staged,Kafka,Drop phGrayOwnership: the workflow, activity, and sync logic live in warehouse_sources (
data_imports/person_property_sync_job.py+pipelines/pipeline/person_property_sync.py, registered viafacade/temporal.py). customer_analytics contributes only config resolution, through aPersonPropertySyncSourcehook registered at app-ready — so warehouse_sources never imports the customer_analytics models and customer_analytics keeps its facade-only contract surface (no tach carve-out).Feature flag: the whole feature is gated behind
warehouse-person-properties(org-keyed, fail-closed), enforced at three levels — the customer_analytics resolvers (the pipeline choke point: flag off disables staging, the workflow gate, and the sync at once; only evaluated for schemas that actually have person sources), the definition-create API (person target rejected when off), and the config modal (the "Attach to" toggle is hidden unless the flag is on or an existing person property is being edited).Kafka: the topic (+DLQ) is routed to the
WAREHOUSE_SOURCEScluster profile (Warpstream warehouse-pipelines). No deploy changes needed: the shared temporal-worker and warehouse chart layers already map that cluster's endpoint intoKAFKA_WAREHOUSE_SOURCES_HOSTSfor both the metadata worker and the dedicated consumer.Observability:
warehouse_person_property_sync_total{team_id, outcome},warehouse_person_property_sync_rows_total{team_id, stage}(funnel: read → changed → existing → produced, plus skipped_missing_person, so a stage dropping to zero pinpoints where updates go missing), and a duration histogram; per-source structured logs; activity failures are captured to error tracking before re-raising for Temporal's retries.Recovery layers: staged files are cleared only on success and the snapshot advances only per produced person, so a failed run re-reads and re-diffs on the next sync; the staging sink's abandoned-prefix TTL sweep is the backstop for jobs whose workflow never ran.
Rolling deploy / in-flight executions: safe.
person_property_sync_enableddefaults toFalseon the activity-output dataclass, and Temporal's payload converter applies defaults for missing fields and ignores unknown ones — old executions replay with the gate off, so the new child-start command never appears during replay.How did you test this code?
Automated tests I (Claude) ran locally, all green:
person_property_sync_test.py: the pure core thoroughly —build_bundles(null key / null value / missing column / str-coercion),bundle_hash(order independence, datetimes, value sensitivity),select_changed(skip unchanged, keep changed + new, last-write-wins on dupes). Plus an orchestration test that mocks the S3 / personhog / Kafka boundaries and asserts only changed-and-existing persons are produced, the snapshot advances only for produced ids, provenance is stamped, and staged files are cleared. And a no-sources no-op.person_property_projection_test.py: the resolvers return the right projections/configs per enabled person source, and a flag-off test proving the choke point — configured sources resolve toNonewhen the rollout flag is off, which no other test covers (a regression there silently un-gates the feature).person_property_row_sink_test.py+create_job_modelsuite as regression.tach check --dependencies --interfacespasses (isolation intact).I could not run a real end-to-end sync (needs the temporal worker + object storage + Kafka + a live source). The S3 snapshot I/O and Kafka production mirror the existing
CDPProducer/ backfill patterns but are exercised only via mocks here; they'll be validated in dogfood behind the flag once the consumer lands.Automatic notifications
Docs update
User-facing docs land with the frontend once the feature is out from behind the flag.
🤖 Agent context
Autonomy: Human-driven (agent-assisted)
Built with Claude Code, directed by me (Tom). The workflow originally lived in customer_analytics mirroring the signals
emit-data-import-signalschild; it was moved into warehouse_sources (with a config-resolution hook left behind) since team warehouse owns and operates the sync machinery. Skill invoked:/writing-tests. Review-driven revisions worth noting: moved off the shared post-processing queue onto the data-warehouse metadata queue; added the missing topic→cluster routing (the topic previously defaulted to the shared Kafka cluster rather than the warehouse Warpstream cluster); replaced anapps.get_modelindirection with theposthog.models.PropertyDefinitionre-export and its realTypeenum; reordered the snapshot path to team → source → schema; and added the rollout flag, funnel metrics, and error-tracking capture. Stamping provenance stays in the workflow rather than the Kafka consumer because the messages don't carry source/schema ids and the consumer would stamp per-message instead of once per source.