feat(identity): sync CLI — copy the persons log into ClickHouse (metrics person_id, iteration 1) - #2065
Conversation
|
Warning Review limit reached
Next review available in: 55 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📥 CommitsReviewing files that changed from the base of the PR and between 710e0ea5f1a1563526b03a38405cfe94d49f6ac1 and b1ee354. ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (21)
📝 WalkthroughWalkthroughAdds a Rust ChangesPersons synchronization
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant Operator
participant IdentityResolution
participant MariaDB
participant ClickHouse
participant JournalAPI
Operator->>IdentityResolution: run sync CLI
IdentityResolution->>MariaDB: acquire lock and read persons log
IdentityResolution->>ClickHouse: stage and exchange snapshot
IdentityResolution->>MariaDB: record operation result
Operator->>JournalAPI: list or fetch persons-sync operation
JournalAPI->>MariaDB: query tenant-scoped operation
JournalAPI-->>Operator: return operation response
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
src/backend/services/identity-resolution/src/main.rs (1)
87-91: 📐 Maintainability & Code Quality | 🔵 TrivialNaming:
EXIT_SEED_*constants are now shared withsync.The doc comment above already documents these as "one shared scheme," but the constant names still read as seed-only, which is a little confusing at both use-sites (124-141 for
Sync). A rename (e.g.,EXIT_LOCK_BUSY/EXIT_GUARD/EXIT_FAILED) would make the sharing explicit; purely cosmetic.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/services/identity-resolution/src/main.rs` around lines 87 - 91, Rename the shared exit-code constants from EXIT_SEED_FAILED, EXIT_SEED_LOCK_BUSY, and EXIT_SEED_GUARD to neutral names such as EXIT_FAILED, EXIT_LOCK_BUSY, and EXIT_GUARD, then update every reference in both the seed and sync subcommand paths, including the Sync use-sites.src/ingestion/tests/e2e/lib/identity.py (1)
289-331: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider a shared helper for
run_sync_cli/run_seed_cli.
run_sync_cliduplicatesrun_seed_cli's capability-gate/env-wiring/subprocess.runshape almost line for line, differing only in the subcommand name and the seed-only--modeflag. Extracting a private_run_gear_cli(subcommand, *, tenant, force, mode=None, timeout_s, extra_env)helper would remove the duplication and keep future CLI subcommands from repeating it again.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/tests/e2e/lib/identity.py` around lines 289 - 331, Extract the duplicated capability check, environment setup, command construction, and subprocess execution from run_sync_cli and run_seed_cli into a private _run_gear_cli helper. Make the helper accept subcommand, tenant, force, mode, timeout_s, and extra_env, applying --mode only when provided, then have both public methods delegate to it while preserving their existing behavior and signatures.src/backend/services/identity-resolution/src/infra/db/mod.rs (1)
148-203: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider extracting a shared advisory-lock helper.
SyncLockGuardduplicatesSeedLockGuard'sGET_LOCK/RELEASE_LOCKsession logic almost verbatim (the doc comment itself calls it "the global sibling ofSeedLockGuard, with identical lifetime semantics"). A small genericAdvisoryLockGuard(parameterized by lock name) would remove the duplication and keep both lock types from drifting independently.♻️ Sketch of a shared helper
-pub struct SyncLockGuard { - conn: DatabaseConnection, -} - -impl SyncLockGuard { - pub async fn try_acquire(database_url: &str) -> anyhow::Result<Option<Self>> { - use sea_orm::{ConnectionTrait, DbBackend, Statement}; - let conn = connect_single(database_url).await?; - let acquired: Option<i8> = conn - .query_one(Statement::from_sql_and_values( - DbBackend::MySql, - "SELECT GET_LOCK(?, 0)", - [SYNC_LOCK.into()], - )) - .await? - .map(|r| r.try_get_by_index::<Option<i8>>(0)) - .transpose()? - .flatten(); - if acquired == Some(1) { - Ok(Some(Self { conn })) - } else { - Ok(None) - } - } - pub async fn release(self) { /* ... */ } -} +async fn try_acquire_named(database_url: &str, name: &str) -> anyhow::Result<Option<DatabaseConnection>> { + use sea_orm::{ConnectionTrait, DbBackend, Statement}; + let conn = connect_single(database_url).await?; + let acquired: Option<i8> = conn + .query_one(Statement::from_sql_and_values( + DbBackend::MySql, + "SELECT GET_LOCK(?, 0)", + [name.into()], + )) + .await? + .map(|r| r.try_get_by_index::<Option<i8>>(0)) + .transpose()? + .flatten(); + Ok((acquired == Some(1)).then_some(conn)) +}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/services/identity-resolution/src/infra/db/mod.rs` around lines 148 - 203, Extract the duplicated GET_LOCK/RELEASE_LOCK session logic from SyncLockGuard and SeedLockGuard into a shared AdvisoryLockGuard parameterized by the advisory lock name. Refactor both guards to reuse this helper while preserving their existing lock constants, non-blocking acquisition behavior, release semantics, and public APIs.src/backend/services/identity-resolution/src/infra/db/persons_log_repo.rs (1)
41-48: 🚀 Performance & Scalability | 🔵 TrivialNote: full-table materialization per run.
read_allloads the entirepersonslog into memory as aVecbefore any row is streamed to ClickHouse (the writer only streams the write side, infill_and_swap). This matches the documented "plain entity scan, full snapshot" design, but as the log grows, memory and read latency scale linearly with total row count on every run. If persons-log volume becomes large, consider a cursor-based/streaming read (e.g., SeaORM'sstream()) so the read and the ClickHouse insert can be pipelined instead of buffered as oneVec.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/services/identity-resolution/src/infra/db/persons_log_repo.rs` around lines 41 - 48, Replace the full-table buffering in read_all with a cursor-based or streaming entity scan, and expose rows incrementally so ClickHouse writes can be pipelined without materializing all PersonsLogRow values in a Vec. Preserve the ascending persons::Column::Id ordering and map each database model through map_row.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/backend/services/identity-resolution/helm/values.yaml`:
- Around line 100-117: Expose configurable CronJob controls by adding
sync.concurrencyPolicy, sync.backoffLimit, and sync.activeDeadlineSeconds to
src/backend/services/identity-resolution/helm/values.yaml with defaults matching
the current behavior, then update the sync CronJob template in
src/backend/services/identity-resolution/helm/templates/sync-cronjob.yaml to
reference those values instead of the hardcoded Forbid, 2, and 600 settings.
In `@src/backend/services/identity-resolution/src/infra/identity_persons.rs`:
- Around line 140-176: Harden identifier validation in drop_stale_stagings
before constructing the DROP TABLE statement: require name to exactly match
STAGING_PREFIX followed by 32 lowercase hexadecimal characters, rather than only
checking starts_with(STAGING_PREFIX). Skip any name that fails this shape
validation, preserving the existing drop and logging behavior for valid
generated staging names.
---
Nitpick comments:
In `@src/backend/services/identity-resolution/src/infra/db/mod.rs`:
- Around line 148-203: Extract the duplicated GET_LOCK/RELEASE_LOCK session
logic from SyncLockGuard and SeedLockGuard into a shared AdvisoryLockGuard
parameterized by the advisory lock name. Refactor both guards to reuse this
helper while preserving their existing lock constants, non-blocking acquisition
behavior, release semantics, and public APIs.
In `@src/backend/services/identity-resolution/src/infra/db/persons_log_repo.rs`:
- Around line 41-48: Replace the full-table buffering in read_all with a
cursor-based or streaming entity scan, and expose rows incrementally so
ClickHouse writes can be pipelined without materializing all PersonsLogRow
values in a Vec. Preserve the ascending persons::Column::Id ordering and map
each database model through map_row.
In `@src/backend/services/identity-resolution/src/main.rs`:
- Around line 87-91: Rename the shared exit-code constants from
EXIT_SEED_FAILED, EXIT_SEED_LOCK_BUSY, and EXIT_SEED_GUARD to neutral names such
as EXIT_FAILED, EXIT_LOCK_BUSY, and EXIT_GUARD, then update every reference in
both the seed and sync subcommand paths, including the Sync use-sites.
In `@src/ingestion/tests/e2e/lib/identity.py`:
- Around line 289-331: Extract the duplicated capability check, environment
setup, command construction, and subprocess execution from run_sync_cli and
run_seed_cli into a private _run_gear_cli helper. Make the helper accept
subcommand, tenant, force, mode, timeout_s, and extra_env, applying --mode only
when provided, then have both public methods delegate to it while preserving
their existing behavior and signatures.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: a3d3fcd6-5d74-4505-83b7-5f2359dabc85
📥 Commits
Reviewing files that changed from the base of the PR and between 7f31597 and 276224b64957732baaaab58903d18a4ea3782e09.
⛔ Files ignored due to path filters (1)
src/backend/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (20)
src/backend/Cargo.tomlsrc/backend/services/identity-resolution/helm/templates/sync-cronjob.yamlsrc/backend/services/identity-resolution/helm/values.yamlsrc/backend/services/identity-resolution/src/api/error.rssrc/backend/services/identity-resolution/src/api/mod.rssrc/backend/services/identity-resolution/src/api/seed.rssrc/backend/services/identity-resolution/src/api/sync.rssrc/backend/services/identity-resolution/src/domain/mod.rssrc/backend/services/identity-resolution/src/domain/sync_service.rssrc/backend/services/identity-resolution/src/gear.rssrc/backend/services/identity-resolution/src/infra/db/mod.rssrc/backend/services/identity-resolution/src/infra/db/ops_repo.rssrc/backend/services/identity-resolution/src/infra/db/persons_log_repo.rssrc/backend/services/identity-resolution/src/infra/identity_persons.rssrc/backend/services/identity-resolution/src/infra/mod.rssrc/backend/services/identity-resolution/src/main.rssrc/backend/services/identity-resolution/src/seed_runner.rssrc/backend/services/identity-resolution/src/sync_runner.rssrc/ingestion/tests/e2e/identity/test_persons_sync.pysrc/ingestion/tests/e2e/lib/identity.py
276224b to
710e0ea
Compare
|
Both bot findings addressed (branch also rebased onto current main):
Checks re-run locally: clippy pedantic 0, fmt clean, 81 unit tests, helm contract suite 17/17. |
710e0ea to
c6799d9
Compare
Iteration 1 of the metrics person_id rework (#1873 consumer side), reworked onto the CLI trigger model the persons-seed established in #1690 — the earlier POST /v1/persons-sync + queue + worker draft never merged; this lands the surface CLI-only from birth. `identity-resolution sync` runs one copy of the MariaDB `persons` observation log into ClickHouse `identity.identity_persons` (next to identity_inputs) and exits — the table dbt gold builds resolve email → person_id against. Execution mirrors the seed subcommand: global MariaDB GET_LOCK advisory lock held for the whole run (RAII guard, dies with the session), zombie sweep, `operations` journal row under the resolved tenant with the SYSTEM_AUTHOR nil UUID, exit codes 0 ok / 1 failed / 2 lock busy / 3 guard. The natural pairing is "sync after seed": the seed rewrites the log, the sync publishes it — the Helm CronJob is scheduled 15 minutes after the seed's (sync-cronjob.yaml, same wiring). Copy semantics: full snapshot per run — per-run staging table, count-verify, atomic EXCHANGE TABLES swap, `_synced_at` watermark guard as a backstop, >1h stale-staging GC. Readers never see an empty or partial table; a failed run leaves the live snapshot untouched. The log is copied verbatim (source provenance columns, NULLs preserved) so future per-source resolution lives entirely in the dbt-side resolve macro. An empty-log run is refused by a guard (publishing it would erase a populated snapshot — the destructive-zero-rows lesson); `--force` overrides deliberately. HTTP keeps only the read-only journal (GET /v1/persons-sync + /{id}), admin-gated, same wire conventions as the seed journal. e2e: rust-only CLI contract module — exit 0 + terminal journal row, snapshot verification in ClickHouse against the fixture dataset (fixed person UUIDs, dup pair stays unresolved, watermark stamped), replace-not-append, journal type-scoping vs the seed surface, status filter, 401/403/404. Full identity suite green (124 passed, 5 skipped). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Sergei Mozhaev <mozhaev.dev@gmail.com>
c6799d9 to
b1ee354
Compare
| # Journal-row tenant (UUID); same override semantics as seed.tenantDefaultId | ||
| # (the copy itself is tenant-agnostic — this only scopes the operations | ||
| # journal the admin GET routes read). | ||
| tenantDefaultId: "" |
There was a problem hiding this comment.
The insight-gitops should be modified for customer cluster or this value should be set from values.
There was a problem hiding this comment.
Already covered on the gitops path: deploy/gitops/scripts/compose-app-secrets.sh (the "First-admin bootstrap inputs" block) writes APP__gears__identity-resolution__config__tenant_default_id into insight-identity-resolution-config whenever TENANT_DEFAULT is configured for the cluster — and the sync CronJob consumes that same Secret via envFrom, so customer clusters need no gitops change. sync.tenantDefaultId here is the standalone-install override, mirroring the seed's pattern from #1690. And the tenant is only journal-row scoping for the sync — a cluster without it still runs via the sole-tenant inference fallback.
mitasovr
left a comment
mitasovr
left a comment
There was a problem hiding this comment.
Reviewed against the #1873 epic. Architecturally sound iteration 1 of the consumer path: the MariaDB persons log is the right publication point per ADR-0002 (any future matching improvements flow through the same pipe), the verbatim copy keeps resolution semantics downstream, and the full-snapshot + EXCHANGE swap with empty-log guard / count-verify / watermark backstop is proportionate to the blast radius of a bad publish. Known trade-offs (time-based seed→sync coupling, lockstep schema copies, PII now in a second store for #719 to account for) are acknowledged in-code and fine as follow-ups. Approving.
| /// (writes) and the `GET /v1/persons-seed*` journal endpoints (filter). | ||
| pub const PERSONS_SEED_OP: &str = "persons-seed"; | ||
| /// Operation type of the persons→ClickHouse sync (`sync` subcommand). | ||
| pub const PERSONS_SYNC_OP: &str = "persons-sync"; |
There was a problem hiding this comment.
Some strings declared twice will drift.
There was a problem hiding this comment.
Within the service each op-type string is declared exactly once: grep '"persons-sync"' over src/ hits only this const, and every consumer (sync_runner journal writes, the GET journal filter) goes through ops_repo::PERSONS_SYNC_OP — same for the seed const. The other spellings live across wire boundaries (the /v1/persons-sync route path, helm component labels, the e2e assertions) which can't share a Rust declaration; the contract e2e pins them together against the live service (test_persons_sync asserts operation_type == "persons-sync" end-to-end), so a drift fails CI rather than going quiet.
Iteration 2 of the metrics person_id rework: the runtime switches from the email entity key to the canonical person UUID the identity pipeline has been resolving into the gold tables since constructorfabric#2065/constructorfabric#2066. Hard switch by decision — no compatibility flag; rollout is gated on the dev coverage numbers, not on code. - validation: `entity.ids` parse as UUIDs into `person_ids: Vec<Uuid>` (parse-don't-validate; the pre-cutover email shape is now a loud 400, never a silent empty result). The email lowercasing normalizer is gone. - compiler: every view (period/timeseries incl. capped, breakdown, histogram, ranking, peer) filters `person_id IN (…)`, groups and orders by person_id, and projects `toString(assumeNotNull(person_id))`. Unresolved rows (person_id NULL) fall out of every query — the epic's "excluded, not guessed" stance, now enforced by the key itself. - peer: the cohort join runs person_id = person_id instead of string email equality — the fragile HR-email↔source-email match this rework exists to kill. The person grain is hardened with LIMIT 1 BY person_id in both cohort CTEs: the view is unique per EMAIL, and two emails resolving to one person must not double-weight a peer or blend pools. - schema validator: person_id joins OBSERVATION_COLUMNS/COHORT_COLUMNS in this same change, as the constants' comment promised — a table without the column now fails the probe loudly instead of answering empty. - naming discipline (review requirement): everything carrying the person UUID is named person_id/person_ids through validation, compiler rows, batch demux and builder; `entity_id` survives ONLY as the wire field name, documented as a seam at the single builder/DTO boundary. 501 unit tests green (peer-guard and ranking tests updated to the new grain), clippy pedantic clean, fmt clean. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Sergei Mozhaev <mozhaev.dev@gmail.com>
3 participants
What
A new
identity-resolution syncCLI subcommand (+ Helm CronJob + read-only journal GETs) that copies the MariaDBpersonsobservation log into ClickHouseidentity.identity_persons— the table the metrics dbt builds will resolveemail → person_idagainst.Consumer-side foundation for the identity mapping engine epic (#1873): however matching improves there, its output reaches metrics through this pipe; the dbt side (resolve macro +
person_idin the observation models + activity-weighted match-rate view) follows in a companion PR.How
Trigger model — CLI-only from birth, mirroring the persons-seed shape (#1690):
identity-resolution syncruns one copy and exits (HelmCronJobscheduled 15 min after the seed's — the seed rewrites the log, the sync publishes it; manualkubectl create job --from=cronjob/...-sync). Exit codes0 ok / 1 failed / 2 lock busy / 3 guard. HTTP keeps only the admin-gated journal:GET /v1/persons-sync+/{id}.Copy semantics — full snapshot per run, atomic publish:
EXCHANGE TABLES→ drop; readers only ever see a complete snapshot, and a failed run leaves the live one untouched;GET_LOCK(RAII guard, released even on process death); the_synced_atwatermark guard before the swap is a backstop for anything bypassing the runner; stale stagings from crashed runs are GC'd after 1h;''preserved) — all resolution semantics stay downstream in the dbt macro, and per-source resolution stays possible;--forceoverrides deliberately.Journal — every run (guard refusals included) lands in
operationsunder the resolved tenant with theSYSTEM_AUTHORnil UUID; the summary carriesrows/max_id/max_created_at/synced_at— the resolution watermark for "why did the numbers change between builds".Testing
cargo test -p identity-resolution— 81 unit tests (incl. run_sync orchestration on fakes, guard decisions, wire-row mapping, live-gated lock mutual-exclusion);E2E_IDENTITY_IMPLEMENTATION=rust): newidentity/test_persons_sync.py— CLI exit 0 + terminal journal row, ClickHouse snapshot assertions against the fixture dataset (fixed person UUIDs, shared-email pair stays unresolved — the copy is a log, not a resolution), replace-not-append semantics, journal type-scoping vs the seed surface, 401/403/404;helm lint+ template render (2 CronJobs) clean.Refs #1873, #1690.
🤖 Generated with Claude Code
Summary by CodeRabbit