From 022dd98095e55978e7c6ba1c53ff71427f01402c Mon Sep 17 00:00:00 2001 From: Sergey <78955917+mozhaev-dev@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:25:43 +0300 Subject: [PATCH] feat(identity): make persons-seed CLI-only with a scheduled CronJob (#1690) (#2046) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cherry-pick of the #2046 merge (fab0089e) from main. The identity org projection froze after bootstrap (nothing re-ran the persons-seed while connectors kept identity_inputs fresh) — Team-view rosters went stale for everyone re-parented since the last run (vz_blocker #1690). The seed is now a `seed` CLI subcommand run by a Helm CronJob (30 6 * * * UTC) with per-tenant advisory locking (RAII), input guards, an operations journal, and sole-tenant inference when tenant_default_id is not configured. POST /v1/persons-seed and its in-process queue are removed; the GET journal routes remain. Verified on dev pre- and post-merge: org_chart unfroze (max valid_from 27.07 → 29.07), tenant inference works with the untouched dev Secret, re-runs are idempotent. Co-Authored-By: Claude Fable 5 --- .../workflows/identity-resolution-helm.yml | 50 ++ charts/insight/templates/secrets.yaml | 3 + charts/insight/values.yaml | 10 + src/backend/Cargo.lock | 1 + .../services/identity-resolution/Cargo.toml | 1 + .../helm/templates/seed-cronjob.yaml | 91 ++++ .../helm/tests/test_seed_cronjob_contract.py | 238 +++++++++ .../identity-resolution/helm/values.yaml | 26 + .../identity-resolution/src/api/mod.rs | 24 +- .../identity-resolution/src/api/seed.rs | 336 ++---------- .../src/domain/seed_service.rs | 59 +-- .../services/identity-resolution/src/gear.rs | 47 +- .../identity-resolution/src/infra/db/mod.rs | 76 +++ .../src/infra/db/ops_repo.rs | 4 + .../src/infra/db/seed_repo.rs | 72 +++ .../src/infra/identity_inputs.rs | 7 +- .../services/identity-resolution/src/main.rs | 58 ++- .../identity-resolution/src/seed_runner.rs | 417 +++++++++++++++ .../e2e/identity/test_error_contracts.py | 20 +- .../tests/e2e/identity/test_meta_gate.py | 17 +- .../tests/e2e/identity/test_persons_seed.py | 493 +++++++++++++++--- src/ingestion/tests/e2e/lib/api_coverage.py | 22 +- src/ingestion/tests/e2e/lib/identity.py | 67 +++ src/ingestion/tests/e2e/lib/identity_seed.py | 2 +- 24 files changed, 1677 insertions(+), 464 deletions(-) create mode 100644 .github/workflows/identity-resolution-helm.yml create mode 100644 src/backend/services/identity-resolution/helm/templates/seed-cronjob.yaml create mode 100644 src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py create mode 100644 src/backend/services/identity-resolution/src/seed_runner.rs diff --git a/.github/workflows/identity-resolution-helm.yml b/.github/workflows/identity-resolution-helm.yml new file mode 100644 index 000000000..260856a39 --- /dev/null +++ b/.github/workflows/identity-resolution-helm.yml @@ -0,0 +1,50 @@ +name: Identity Resolution — Helm contract + +# Render-contract gate for the identity-resolution chart, most importantly the +# persons-seed CronJob (#1690): the original bug was the ABSENCE of scheduling, +# so the schedule wiring is contract, not plumbing. Pure `helm template` + +# assertions — no cluster, no images. The functional-k3s lane does not deploy +# identityResolution, so without this nothing would catch a broken CronJob +# template before an instance rollout. + +on: + pull_request: + branches: [main] + paths: + - "src/backend/services/identity-resolution/helm/**" + - "charts/insight/**" + - ".github/workflows/identity-resolution-helm.yml" + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + helm-contract: + name: render contract + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + # Pinned to full SHAs (mutable tags are repointable — supply-chain + # hardening, same as semgrep.yml / trivy.yml). + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + + - name: Install test deps + run: pip install --quiet pytest pyyaml + + - name: helm lint + run: | + helm lint src/backend/services/identity-resolution/helm + helm dependency update charts/insight + helm lint charts/insight + + - name: Render-contract tests (seed CronJob, umbrella tenant guard) + run: python -m pytest src/backend/services/identity-resolution/helm/tests/ -q diff --git a/charts/insight/templates/secrets.yaml b/charts/insight/templates/secrets.yaml index e61411d2a..c3706f224 100644 --- a/charts/insight/templates/secrets.yaml +++ b/charts/insight/templates/secrets.yaml @@ -297,6 +297,9 @@ stringData: {{- end }} {{- if .Values.identityResolution.deploy }} +{{- if and ((.Values.identityResolution.seed).enabled) (not ((.Values.global | default dict).tenantDefaultId)) (not ((.Values.identityResolution.seed).tenantDefaultId)) }} +{{- fail "identityResolution.seed.enabled=true requires a tenant: set global.tenantDefaultId (it feeds tenant_default_id in insight-identity-resolution-config) or identityResolution.seed.tenantDefaultId — the seed refuses to run without one, so every CronJob run would exit 1. Alternatively disable identityResolution.seed.enabled." }} +{{- end }} --- # Identity Resolution (Rust port, epic #1602) leaf config. Same gears-rust # env-override convention as analytics (`APP__gears____config__*`; the diff --git a/charts/insight/values.yaml b/charts/insight/values.yaml index 7c478b218..b29c3abe7 100644 --- a/charts/insight/values.yaml +++ b/charts/insight/values.yaml @@ -517,6 +517,16 @@ identityResolution: # auto-generated credentials (see templates/secrets.yaml). The subchart # consumes it via envFrom (overriding its mounted config ConfigMap). existingSecret: "insight-identity-resolution-config" + # Scheduled persons-seed (#1690): a CronJob runs `identity-resolution seed` + # daily to rebuild the identity org projection (persons / account_person_map + # / org_chart) from ClickHouse identity_inputs — without it the projection + # freezes at the last manual run and Team-view rosters go stale. Runs + # serialize on a per-tenant MariaDB advisory lock; input guards abort on an + # empty identity_inputs read or a tenant mismatch (`--force` overrides). + # Manual run: kubectl create job --from=cronjob/-identity-resolution-seed ... + seed: + enabled: true + schedule: "30 6 * * *" # daily, after overnight syncs + the 06:00 data-quality run # Gateway-JWT verification (NGINX_BFF R1) — same wiring as analytics: the # oidc-authn-plugin resolves the authenticator's JWKS via OIDC discovery # over https on `issuer` and trusts its CA from the authn-tls cert Secret. diff --git a/src/backend/Cargo.lock b/src/backend/Cargo.lock index b8d7b081b..bac682150 100644 --- a/src/backend/Cargo.lock +++ b/src/backend/Cargo.lock @@ -3045,6 +3045,7 @@ dependencies = [ "serde_json", "tokio", "tracing", + "tracing-subscriber", "utoipa", "uuid", ] diff --git a/src/backend/services/identity-resolution/Cargo.toml b/src/backend/services/identity-resolution/Cargo.toml index 77f98f269..80bd41b3c 100644 --- a/src/backend/services/identity-resolution/Cargo.toml +++ b/src/backend/services/identity-resolution/Cargo.toml @@ -43,6 +43,7 @@ serde_json = { workspace = true } async-trait = { workspace = true } axum = { workspace = true } tracing = { workspace = true } +tracing-subscriber = { workspace = true } # MariaDB via SeaORM — connection pool + entities + raw resolve queries. sea-orm = { workspace = true } diff --git a/src/backend/services/identity-resolution/helm/templates/seed-cronjob.yaml b/src/backend/services/identity-resolution/helm/templates/seed-cronjob.yaml new file mode 100644 index 000000000..df01bcaa4 --- /dev/null +++ b/src/backend/services/identity-resolution/helm/templates/seed-cronjob.yaml @@ -0,0 +1,91 @@ +{{- if .Values.seed.enabled }} +# Scheduled persons-seed (#1690): rebuilds the identity org projection +# (`persons` / `account_person_map` / `org_chart`) from ClickHouse +# `identity.identity_inputs` so ingested manager/org changes reach the Team +# view — the projection is materialized and only a seed run refreshes it. +# +# Same image/config/secret wiring as the deployment; the `seed` subcommand +# runs one seed and exits (exit codes: 0 ok / 1 failed / 2 lock busy / +# 3 input guard). Runs serialize on a per-tenant MariaDB advisory lock, so +# `concurrencyPolicy: Forbid` is belt-and-braces for cron-vs-cron only — +# manual Jobs and other Insight instances are serialized by the lock itself. +# +# Manual run: +# kubectl create job --from=cronjob/{{ include "insight-identity-resolution.fullname" . }}-seed seed-manual-$USER +apiVersion: batch/v1 +kind: CronJob +metadata: + name: {{ include "insight-identity-resolution.fullname" . }}-seed + labels: + {{- include "insight-identity-resolution.labels" . | nindent 4 }} + app.kubernetes.io/component: persons-seed +spec: + schedule: {{ .Values.seed.schedule | quote }} + concurrencyPolicy: Forbid + successfulJobsHistoryLimit: {{ .Values.seed.successfulJobsHistoryLimit }} + failedJobsHistoryLimit: {{ .Values.seed.failedJobsHistoryLimit }} + jobTemplate: + spec: + # A couple of retries for transient connect blips; the advisory lock + + # the operations journal make a repeated run safe. The deadline caps a + # wedged pod well past the in-binary 10-minute seed timeout. + backoffLimit: 2 + activeDeadlineSeconds: 900 + template: + metadata: + # NOT the shared selectorLabels: the Service selects on + # name+instance alone, and a seed pod carrying them would enter the + # Service's endpoints (it listens on nothing). + labels: + app.kubernetes.io/name: identity-resolution-seed + app.kubernetes.io/instance: {{ .Release.Name }} + app.kubernetes.io/component: persons-seed + spec: + {{- with .Values.global }} + {{- with .imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 12 }} + {{- end }} + {{- end }} + enableServiceLinks: false + restartPolicy: Never + securityContext: + runAsNonRoot: true + runAsUser: 1000 + fsGroup: 1000 + containers: + - name: persons-seed + image: "{{ required "image.repository is required" .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: ["/app/identity-resolution"] + args: ["-c", "/app/config/insight.yaml", "seed"] + volumeMounts: + - name: identity-resolution-config + mountPath: /app/config/insight.yaml + subPath: insight.yaml + readOnly: true + envFrom: + # Same Secret as the deployment: database_url, clickhouse_*, + # tenant_default_id — everything the seed needs. + - secretRef: + name: {{ required "existingSecret is required (umbrella provides `insight-identity-resolution-config`; standalone installs supply their own)" .Values.existingSecret | quote }} + {{- with .Values.seed.tenantDefaultId }} + env: + # Explicit tenant for the seed run — the single source when + # set (in Kubernetes `env` entries override `envFrom`, so + # this wins over any tenant_default_id the Secret carries). + # Standalone installs that don't carry the tenant in their + # Secret set this; the umbrella wires the Secret from + # global.tenantDefaultId instead and validates it at render. + - name: APP__gears__identity-resolution__config__tenant_default_id + value: {{ . | quote }} + {{- end }} + securityContext: + allowPrivilegeEscalation: false + resources: + {{- toYaml .Values.seed.resources | nindent 16 }} + volumes: + - name: identity-resolution-config + configMap: + name: {{ include "insight-identity-resolution.fullname" . }}-gears-config +{{- end }} diff --git a/src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py b/src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py new file mode 100644 index 000000000..3011be398 --- /dev/null +++ b/src/backend/services/identity-resolution/helm/tests/test_seed_cronjob_contract.py @@ -0,0 +1,238 @@ +"""Helm render-contract for the persons-seed CronJob (#1690). + +The original bug was the ABSENCE of scheduling — the seed existed but nothing +ever ran it — so the schedule wiring itself is contract, not plumbing: these +tests render the chart(s) with `helm template` and assert the manifests the +cluster would actually get. No cluster involved; runs anywhere helm + PyYAML +exist (CI: .github/workflows/identity-resolution-helm.yml). + +Covered: + * the CronJob exists by default with the documented schedule and the exact + `seed` command/args against the mounted gears config; + * config comes from the SAME Secret/ConfigMap pair the deployment uses; + * `seed.tenantDefaultId` env-overrides the Secret (k8s `env` beats + `envFrom`) — the standalone-install tenant source; + * `seed.enabled=false` removes the CronJob and nothing else; + * the seed pod labels do NOT match the Service selector (a pod that + listens on nothing must never enter the Service's endpoints); + * the umbrella refuses to render when the seed is enabled but no tenant is + configured — only on the path where the umbrella composes the config + Secret itself (`credentials.autoGenerate`). +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest +import yaml + +HERE = Path(__file__).resolve() +SUBCHART = HERE.parents[1] # .../identity-resolution/helm +REPO_ROOT = HERE.parents[6] +UMBRELLA = REPO_ROOT / "charts" / "insight" + +TENANT = "3e1d5a65-434c-95b4-8c1b-eb8f53a39bab" + +# Minimum viable subchart install (mirrors the umbrella's wiring). +SUBCHART_BASE = [ + "--set", + "image.tag=0.0.0-test", + "--set", + "existingSecret=test-secret", + "--set", + "gateway.issuer=https://issuer.test", +] + +# Minimum viable umbrella install on the credentials.autoGenerate path (the +# only path where the umbrella composes the identity-resolution config Secret +# and can therefore vouch for the tenant inside it). +UMBRELLA_BASE = [ + "--set", + "identityResolution.deploy=true", + "--set", + "identityResolution.image.tag=0.0.0-test", + "--set", + "clickhouse.host=ch", + "--set", + "clickhouse.username=u", + "--set", + "clickhouse.password=p", + "--set", + "clickhouse.database=insight", + "--set", + "mariadb.host=m", + "--set", + "mariadb.username=insight", + "--set", + "mariadb.password=pw", + "--set", + "mariadb.database=insight", + "--set", + "redis.host=redis", + "--set", + "redis.password=rp", + "--set", + "redpanda.brokers=rp:9092", + "--set", + "ingestion.reconcile.tenantId=default", + "--set", + "authenticator.oidc.issuerUrl=https://idp", + "--set", + "authenticator.oidc.clientId=c", + "--set", + "authenticator.oidc.clientSecret=s", + "--set", + "authenticator.oidc.redirectUri=https://x/cb", +] + + +def _render(chart: Path, *extra: str) -> tuple[int, str, str]: + proc = subprocess.run( # noqa: S603 — test-controlled argv + ["helm", "template", "contract-test", str(chart), *extra], # noqa: S607 + capture_output=True, + text=True, + timeout=120, + check=False, + ) + return proc.returncode, proc.stdout, proc.stderr + + +def _docs(manifests: str) -> list[dict]: + return [d for d in yaml.safe_load_all(manifests) if isinstance(d, dict)] + + +def _the(docs: list[dict], kind: str) -> dict: + matches = [d for d in docs if d.get("kind") == kind] + assert len(matches) == 1, f"expected exactly one {kind}, got {len(matches)}" + return matches[0] + + +def _subchart_docs(*extra: str) -> list[dict]: + rc, out, err = _render(SUBCHART, *SUBCHART_BASE, *extra) + assert rc == 0, err + return _docs(out) + + +@pytest.fixture(scope="module") +def default_docs() -> list[dict]: + return _subchart_docs() + + +def _seed_container(cronjob: dict) -> dict: + pod = cronjob["spec"]["jobTemplate"]["spec"]["template"]["spec"] + assert pod["restartPolicy"] == "Never", pod + (container,) = pod["containers"] + return container + + +def test_cronjob_exists_by_default_with_documented_schedule(default_docs) -> None: + cj = _the(default_docs, "CronJob") + assert cj["metadata"]["name"] == "contract-test-identity-resolution-seed" + assert cj["spec"]["schedule"] == "30 6 * * *" + assert cj["spec"]["concurrencyPolicy"] == "Forbid" + + +def test_cronjob_runs_the_seed_subcommand_against_the_mounted_config(default_docs) -> None: + container = _seed_container(_the(default_docs, "CronJob")) + assert container["command"] == ["/app/identity-resolution"] + assert container["args"] == ["-c", "/app/config/insight.yaml", "seed"] + # The CronJob must never run forced — --force is a deliberate manual act. + assert "--force" not in container["args"] + + +def test_cronjob_uses_the_deployments_secret_and_configmap(default_docs) -> None: + cj = _the(default_docs, "CronJob") + deploy = _the(default_docs, "Deployment") + container = _seed_container(cj) + + secret_refs = [e["secretRef"]["name"] for e in container["envFrom"]] + deploy_secret_refs = [ + e["secretRef"]["name"] for e in deploy["spec"]["template"]["spec"]["containers"][0]["envFrom"] + ] + assert secret_refs == deploy_secret_refs == ["test-secret"] + + cj_volumes = cj["spec"]["jobTemplate"]["spec"]["template"]["spec"]["volumes"] + deploy_volumes = deploy["spec"]["template"]["spec"]["volumes"] + cj_cm = next(v["configMap"]["name"] for v in cj_volumes if "configMap" in v) + deploy_cm = next(v["configMap"]["name"] for v in deploy_volumes if "configMap" in v) + assert cj_cm == deploy_cm + + +def test_tenant_value_overrides_the_secret_via_env(default_docs) -> None: + # Default: no explicit env — the Secret is the tenant source. + container = _seed_container(_the(default_docs, "CronJob")) + assert "env" not in container, container.get("env") + + docs = _subchart_docs("--set", f"seed.tenantDefaultId={TENANT}") + container = _seed_container(_the(docs, "CronJob")) + env = {e["name"]: e["value"] for e in container["env"]} + assert env == {"APP__gears__identity-resolution__config__tenant_default_id": TENANT} + + +def test_seed_disabled_removes_only_the_cronjob() -> None: + docs = _subchart_docs("--set", "seed.enabled=false") + assert not [d for d in docs if d.get("kind") == "CronJob"] + # The rest of the chart is untouched. + _the(docs, "Deployment") + _the(docs, "Service") + + +def test_seed_pod_labels_never_match_the_service_selector(default_docs) -> None: + """A seed pod listens on nothing: if the Service selector matched it, it + would enter the endpoints and blackhole live traffic during every run.""" + selector = _the(default_docs, "Service")["spec"]["selector"] + pod_labels = _the(default_docs, "CronJob")["spec"]["jobTemplate"]["spec"]["template"][ + "metadata" + ]["labels"] + assert any(pod_labels.get(k) != v for k, v in selector.items()), ( + f"seed pod labels {pod_labels} satisfy the Service selector {selector}" + ) + + +# ── umbrella: the tenant render guard ───────────────────────────────────── + + +@pytest.fixture(scope="module") +def umbrella_deps() -> Path: + subprocess.run( # noqa: S603, S607 — refresh the vendored subcharts + ["helm", "dependency", "update", str(UMBRELLA)], + capture_output=True, + text=True, + timeout=300, + check=True, + ) + return UMBRELLA + + +def test_umbrella_refuses_enabled_seed_without_a_tenant(umbrella_deps) -> None: + rc, _, err = _render(umbrella_deps, *UMBRELLA_BASE) + assert rc != 0 + assert "requires a tenant" in err, err + + +def test_umbrella_renders_the_cronjob_with_a_tenant(umbrella_deps) -> None: + rc, out, err = _render( + umbrella_deps, *UMBRELLA_BASE, "--set", f"global.tenantDefaultId={TENANT}" + ) + assert rc == 0, err + _the(_docs(out), "CronJob") + + +def test_umbrella_accepts_the_explicit_seed_tenant_alone(umbrella_deps) -> None: + rc, out, err = _render( + umbrella_deps, *UMBRELLA_BASE, "--set", f"identityResolution.seed.tenantDefaultId={TENANT}" + ) + assert rc == 0, err + container = _seed_container(_the(_docs(out), "CronJob")) + env = {e["name"]: e["value"] for e in container.get("env", [])} + assert env.get("APP__gears__identity-resolution__config__tenant_default_id") == TENANT + + +def test_umbrella_disabled_seed_needs_no_tenant(umbrella_deps) -> None: + rc, out, err = _render( + umbrella_deps, *UMBRELLA_BASE, "--set", "identityResolution.seed.enabled=false" + ) + assert rc == 0, err + assert not [d for d in _docs(out) if d.get("kind") == "CronJob"] diff --git a/src/backend/services/identity-resolution/helm/values.yaml b/src/backend/services/identity-resolution/helm/values.yaml index cc22d6d35..5f80fd93d 100644 --- a/src/backend/services/identity-resolution/helm/values.yaml +++ b/src/backend/services/identity-resolution/helm/values.yaml @@ -67,6 +67,32 @@ identityResolution: expandSubordinates: true maxDepth: 16 +# Scheduled persons-seed (#1690): a CronJob runs `identity-resolution seed` +# to rebuild the identity org projection from ClickHouse identity_inputs. +# Requires tenant_default_id in the config Secret (the seed refuses to run +# without it). Manual run: +# kubectl create job --from=cronjob/-seed seed-manual- +seed: + enabled: true + # Explicit tenant for the seed run (UUID). Optional: when set it is + # injected as an env var that OVERRIDES any tenant_default_id the + # existingSecret carries (k8s `env` beats `envFrom`) — the single source + # for the CronJob. Leave empty when the Secret already carries the tenant + # (the umbrella wires it from global.tenantDefaultId and validates at + # render). Without a tenant from either source every run exits 1. + tenantDefaultId: "" + # Daily, after the overnight connector syncs and the 06:00 data-quality run. + schedule: "30 6 * * *" + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 3 + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: 500m + memory: 512Mi + # TCP gate before start: the service connects to MariaDB at boot. Same # busybox wait as the .NET identity chart. waitForMariadb: diff --git a/src/backend/services/identity-resolution/src/api/mod.rs b/src/backend/services/identity-resolution/src/api/mod.rs index 54f45ed8c..2b25c955d 100644 --- a/src/backend/services/identity-resolution/src/api/mod.rs +++ b/src/backend/services/identity-resolution/src/api/mod.rs @@ -17,7 +17,6 @@ use axum::Extension; use axum::Router; use axum::http::StatusCode; use sea_orm::DatabaseConnection; -use tokio::sync::mpsc; use toolkit::api::{OpenApiRegistry, OperationBuilder}; use crate::config::GearConfig; @@ -30,8 +29,6 @@ pub struct AppState { pub db: DatabaseConnection, /// Gear config (`org_chart_source_type`, `clickhouse_*`, …). pub config: GearConfig, - /// Sender to the persons-seed worker's job queue (POST enqueues here). - pub seed_tx: mpsc::Sender, } /// Mount the identity-resolution routes onto the host's router. @@ -81,23 +78,10 @@ fn build_operations(router: Router, openapi: &dyn OpenApiRegistry) -> Router { .handler(handlers::resolve_profile) .register(router, openapi); - // Persons-seed (async job): enqueue + poll. Admin-gated: caller = gateway-JWT - // subject, must hold the `admin` role in the tenant. - let router = OperationBuilder::post("/v1/persons-seed") - .operation_id("identity_resolution.persons_seed.create") - .summary("Enqueue a persons-seed run (async)") - .authenticated() - .no_license_required() - .json_request::(openapi, "Seed options") - .json_response_with_schema::( - openapi, - StatusCode::ACCEPTED, - "Queued operation", - ) - .standard_errors(openapi) - .handler(seed::create_persons_seed) - .register(router, openapi); - + // Persons-seed operations journal (read-only; the seed itself runs via the + // `seed` CLI subcommand — CronJob / manual Job, see `crate::seed_runner`). + // Admin-gated: caller = gateway-JWT subject, must hold the `admin` role in + // the tenant. let router = OperationBuilder::get("/v1/persons-seed/{id}") .operation_id("identity_resolution.persons_seed.get") .summary("Get a persons-seed operation") diff --git a/src/backend/services/identity-resolution/src/api/seed.rs b/src/backend/services/identity-resolution/src/api/seed.rs index f39e1dcdc..7d5b0327e 100644 --- a/src/backend/services/identity-resolution/src/api/seed.rs +++ b/src/backend/services/identity-resolution/src/api/seed.rs @@ -1,76 +1,41 @@ -//! Persons-seed HTTP surface + background worker. +//! Persons-seed operations journal — read-only HTTP surface. //! -//! `POST /v1/persons-seed` enqueues an `operations` row and a job on an -//! in-process channel, returning 202; a worker (spawned once in the gear init) -//! drains the channel, runs the seed via [`run_seed`], and marks the operation -//! completed/failed. The GETs poll status. Ported from the .NET -//! `PersonsSeedEndpoints` + `PersonsSeedQueue`. +//! The seed itself is CLI-only (`identity-resolution seed`, run by the Helm +//! `CronJob` or a manual Job — see `crate::seed_runner`); the former +//! `POST /v1/persons-seed` trigger and its in-process queue/worker are gone +//! (#1690). The GETs remain as the observability window over the +//! `operations` rows the CLI runs write: status, summary, error per run. //! //! Admin-gated like the .NET `CallerAdminCheck`: the caller is the gateway-JWT //! subject (`SecurityContext::subject_id`, verified by the host authn pipeline — -//! `NGINX_BFF` R1) and must hold an active `admin` role in the tenant; it is -//! recorded as the seed author. +//! `NGINX_BFF` R1) and must hold an active `admin` role in the tenant. use std::sync::Arc; -use std::time::Duration; use axum::Json; use axum::extract::{Extension, Path, Query}; -use axum::http::StatusCode; -use axum::http::header::LOCATION; use axum::response::IntoResponse; -use sea_orm::DatabaseConnection; use serde::{Deserialize, Serialize}; -use tokio::sync::mpsc; use toolkit_canonical_errors::CanonicalError; use toolkit_security::SecurityContext; use utoipa::ToSchema; use uuid::Uuid; use super::AppState; -use super::canonical_json::CanonicalJson; use super::error::PersonsSeedError; use super::gate::require_admin; -use crate::config::GearConfig; -use crate::domain::seed_service::run_seed; -use crate::infra::db::ops_repo::{self, Operation, OperationStatus}; -use crate::infra::db::seed_repo::MariaDbSeedStore; -use crate::infra::identity_inputs::ClickHouseIdentityInputsReader; - -const LINK_BY_EMAIL_MODE: &str = "link-by-email"; +use crate::infra::db::ops_repo::{self, Operation, OperationStatus, PERSONS_SEED_OP}; /// Default page size / cap for the list endpoint (parity with the .NET /// `PageRequest.DefaultLimit` / `MaxLimit`). const LIST_DEFAULT_LIMIT: u64 = 50; const LIST_MAX_LIMIT: u64 = 500; -/// Upper bound on one seed run in the serial worker; a stall past this fails the -/// job rather than wedging the whole queue. -const SEED_TIMEOUT: Duration = Duration::from_mins(10); -const PERSONS_SEED_OP: &str = "persons-seed"; - -/// A queued persons-seed job handed from the POST handler to the worker. -#[allow(clippy::struct_field_names)] // all three fields are ids by nature -#[derive(Debug, Clone, Copy)] -pub struct PersonsSeedJob { - pub operation_id: Uuid, - pub tenant_id: Uuid, - pub author_person_id: Uuid, -} - -/// Body of `POST /v1/persons-seed`. `mode` defaults to `link-by-email`. -#[derive(Debug, Deserialize, ToSchema)] -pub struct PersonsSeedRequest { - #[serde(default)] - pub mode: Option, -} -impl toolkit::api::api_dto::RequestApiDto for PersonsSeedRequest {} - -/// One operation's status (POST returns the queued row; GETs return current). -/// Wire shape mirrors the .NET `PersonsSeedOperationResponse`: `request` and -/// `summary` are surfaced as parsed JSON (not double-encoded strings), the -/// tenant/author ids are included, timestamps are ISO-8601, and null fields are -/// emitted (the .NET serializer does not drop nulls). +/// One operation's status. Wire shape mirrors the .NET +/// `PersonsSeedOperationResponse`: `request` and `summary` are surfaced as +/// parsed JSON (not double-encoded strings), the tenant/author ids are +/// included, timestamps are ISO-8601, and null fields are emitted (the .NET +/// serializer does not drop nulls). #[derive(Debug, Serialize, ToSchema)] pub struct PersonsSeedOperationResponse { pub operation_id: Uuid, @@ -88,33 +53,6 @@ pub struct PersonsSeedOperationResponse { } impl toolkit::api::api_dto::ResponseApiDto for PersonsSeedOperationResponse {} -impl PersonsSeedOperationResponse { - /// The just-enqueued shape for the `202 Accepted` body, built from the - /// fields the POST handler already holds — avoids a second round-trip to - /// re-read the row, and (unlike a re-read) always reports `queued` even if - /// the worker has already picked the job up. Mirrors the .NET `Queued(...)`. - fn queued( - operation_id: Uuid, - tenant_id: Uuid, - author_person_id: Uuid, - request_json: Option<&str>, - started_at: sea_orm::prelude::DateTime, - ) -> Self { - Self { - operation_id, - operation_type: PERSONS_SEED_OP.to_owned(), - status: OperationStatus::Queued.as_db().to_owned(), - insight_tenant_id: tenant_id, - author_person_id, - request: parse_or_null(request_json), - summary: None, - error_message: None, - started_at: fmt_ts(started_at), - completed_at: None, - } - } -} - impl From for PersonsSeedOperationResponse { fn from(op: Operation) -> Self { Self { @@ -168,111 +106,6 @@ pub struct ListParams { pub limit: Option, } -/// `POST /v1/persons-seed` — enqueue an async persons-seed run. -pub async fn create_persons_seed( - Extension(state): Extension>, - Extension(ctx): Extension, - CanonicalJson(req): CanonicalJson, -) -> Result { - let tenant = ctx.subject_tenant_id(); - // Admin gate first (parity with .NET: the caller/admin check precedes mode - // validation, so an unauthenticated/non-admin caller gets 401/403, not 400). - // The resolved caller is recorded as the author of the job + observations. - let author = require_admin(&state.db, &ctx).await?; - - let mode = req - .mode - .as_deref() - .map(str::trim) - .filter(|m| !m.is_empty()) - .unwrap_or(LINK_BY_EMAIL_MODE); - if mode != LINK_BY_EMAIL_MODE { - return Err(PersonsSeedError::invalid_argument() - .with_field_violation( - "mode", - "unsupported mode; only 'link-by-email' is available", - "INVALID", - ) - .create()); - } - - let operation_id = Uuid::now_v7(); - let started_at = chrono::Utc::now().naive_utc(); - let request_json = serde_json::json!({ "mode": mode }).to_string(); - - ops_repo::enqueue( - &state.db, - operation_id, - PERSONS_SEED_OP, - tenant, - author, - Some(&request_json), - ) - .await - .map_err(|e| { - tracing::error!(error = %e, "enqueue operation failed"); - CanonicalError::internal("failed to enqueue seed").create() - })?; - - let job = PersonsSeedJob { - operation_id, - tenant_id: tenant, - author_person_id: author, - }; - if let Err(err) = try_enqueue_job(&state.seed_tx, job) { - // Channel full/closed — fail the row so it isn't a zombie, and tell the - // caller to retry later (503, not 500 — parity with the .NET queue-full). - // A failed status update is logged, not propagated: 503/retry-later is - // still the right caller signal, and a row left `queued` is reclaimed - // by the startup zombie sweep (`sweep_zombies`). - if let Err(db_err) = - ops_repo::fail(&state.db, operation_id, "seed queue full; retry later").await - { - tracing::error!( - error = %db_err, - %operation_id, - "failed to mark the refused seed operation as failed" - ); - } - return Err(err); - } - - // Audit the enqueue (parity with the .NET `persons_seed.enqueue` audit). - tracing::info!( - %operation_id, - %mode, - author_person_id = %author, - "persons_seed.enqueue" - ); - - // Build the 202 body from the in-memory snapshot (always `queued`) and set - // Location to the status URL — no re-read of the just-inserted row. - let body = PersonsSeedOperationResponse::queued( - operation_id, - tenant, - author, - Some(&request_json), - started_at, - ); - let location = format!("/v1/persons-seed/{operation_id}"); - Ok((StatusCode::ACCEPTED, [(LOCATION, location)], Json(body))) -} - -/// Hand the job to the worker channel, mapping a full/closed channel to the -/// caller-facing 503 (parity with the .NET queue-full path). Split out of the -/// handler so the refusal is unit-testable: the e2e suite cannot fill the -/// channel deterministically from outside. -fn try_enqueue_job( - tx: &mpsc::Sender, - job: PersonsSeedJob, -) -> Result<(), CanonicalError> { - tx.try_send(job).map_err(|_| { - CanonicalError::service_unavailable() - .with_detail("seed queue is full; retry later") - .create() - }) -} - /// `GET /v1/persons-seed/{id}` — poll one operation. pub async fn get_persons_seed( Extension(state): Extension>, @@ -280,7 +113,8 @@ pub async fn get_persons_seed( Path(id): Path, ) -> Result { let tenant = ctx.subject_tenant_id(); - // Same admin gate as POST (parity — the .NET service gates all three routes). + // Same admin gate as the sibling journal route (parity — the .NET service + // gated the whole persons-seed surface). require_admin(&state.db, &ctx).await?; let op = ops_repo::get_by_id(&state.db, tenant, id) .await @@ -305,7 +139,6 @@ pub async fn list_persons_seed( Query(params): Query, ) -> Result { let tenant = ctx.subject_tenant_id(); - // Same admin gate as POST (parity — the .NET service gates all three routes). require_admin(&state.db, &ctx).await?; let status = status_filter(params.status.as_deref()); let limit = params.limit.map_or(LIST_DEFAULT_LIMIT, |l| { @@ -339,138 +172,35 @@ fn status_filter(raw: Option<&str>) -> Option { } } -/// How stale a `queued`/`running` row must be before the startup sweep reclaims -/// it — parity with the .NET `PersonsSeedWorker.ZombieCutoff` (1 hour). -const ZOMBIE_CUTOFF_HOURS: i64 = 1; - -/// Background worker: drain the queue and run each seed to completion, updating -/// the `operations` row. Spawned once from the gear `init`; ends when the -/// channel closes (all senders dropped). -pub async fn run_worker( - mut rx: mpsc::Receiver, - db: DatabaseConnection, - config: GearConfig, -) { - let reader = ClickHouseIdentityInputsReader::connect( - &config.clickhouse_url, - &config.clickhouse_database, - &config.clickhouse_user, - &config.clickhouse_password, - ); - let store = MariaDbSeedStore::new(&db); - - // Startup sweep: a pod restart drops the in-memory queue, so any row left - // `queued`/`running` by the previous process would otherwise never resolve. - // Fail rows older than the cutoff (parity with .NET `SweepZombiesAsync`). - let cutoff = chrono::Utc::now().naive_utc() - chrono::Duration::hours(ZOMBIE_CUTOFF_HOURS); - match ops_repo::sweep_zombies(&db, cutoff).await { - Ok(n) if n > 0 => tracing::warn!(swept = n, "persons-seed: reclaimed zombie operations"), - Ok(_) => {} - Err(e) => tracing::error!(error = %e, "persons-seed: zombie sweep failed"), - } - - while let Some(job) = rx.recv().await { - // Only the worker that wins queued→running proceeds (no double-run). - match ops_repo::try_start(&db, job.operation_id).await { - Ok(true) => {} - Ok(false) => continue, - Err(e) => { - // A transient DB blip on the queued→running transition must not - // strand the (already-consumed) job as a zombie `queued` row — - // mark it failed so it isn't stuck forever, like the queue-full - // path in `create_persons_seed`. - tracing::error!(error = %e, operation_id = %job.operation_id, "try_start failed"); - let _ = - ops_repo::fail(&db, job.operation_id, "try_start failed; retry later").await; - continue; - } - } - - // Bound each run: the worker is single-threaded and serial, so a hung - // ClickHouse/MariaDB call would otherwise block every subsequent job for - // every tenant until the process restarts. Generous ceiling — a healthy - // large-tenant seed is seconds; this only trips on a real stall. - let seed = run_seed( - &reader, - &store, - job.tenant_id, - job.author_person_id, - Uuid::now_v7, - ); - let result = tokio::time::timeout(SEED_TIMEOUT, seed) - .await - .unwrap_or_else(|_| Err(anyhow::anyhow!("persons-seed timed out"))); - - match result { - Ok(summary) => { - let summary_json = - serde_json::to_string(&summary).unwrap_or_else(|_| "{}".to_owned()); - if let Err(e) = ops_repo::complete(&db, job.operation_id, &summary_json).await { - tracing::error!(error = %e, operation_id = %job.operation_id, "complete failed"); - } - } - Err(e) => { - // Log the real error server-side, but persist only a generic - // message: `error_message` is returned verbatim by the GET/list - // endpoints, so raw driver/anyhow text must not leak to callers. - tracing::error!(error = %e, operation_id = %job.operation_id, "persons-seed failed"); - if let Err(e2) = ops_repo::fail( - &db, - job.operation_id, - "persons-seed failed; see server logs", - ) - .await - { - tracing::error!(error = %e2, operation_id = %job.operation_id, "fail update failed"); - } - } - } - } -} - #[cfg(test)] mod tests { - use axum::http::StatusCode; - use axum::response::IntoResponse; - use super::*; - fn job() -> PersonsSeedJob { - PersonsSeedJob { - operation_id: Uuid::from_u128(1), - tenant_id: Uuid::from_u128(2), - author_person_id: Uuid::from_u128(3), - } - } - #[test] - fn enqueue_maps_closed_channel_to_503() -> anyhow::Result<()> { - let (tx, rx) = mpsc::channel(1); - drop(rx); - let Err(err) = try_enqueue_job(&tx, job()) else { - anyhow::bail!("closed channel must refuse the job"); - }; + fn status_filter_maps_known_and_ignores_unknown() { + assert_eq!(status_filter(Some("queued")), Some(OperationStatus::Queued)); assert_eq!( - err.into_response().status(), - StatusCode::SERVICE_UNAVAILABLE + status_filter(Some("running")), + Some(OperationStatus::Running) ); - Ok(()) + assert_eq!( + status_filter(Some("completed")), + Some(OperationStatus::Completed) + ); + assert_eq!(status_filter(Some("failed")), Some(OperationStatus::Failed)); + assert_eq!(status_filter(Some("bogus")), None); + assert_eq!(status_filter(Some("")), None); + assert_eq!(status_filter(None), None); } #[test] - fn enqueue_maps_full_channel_to_503() -> anyhow::Result<()> { - let (tx, _rx) = mpsc::channel(1); - assert!( - try_enqueue_job(&tx, job()).is_ok(), - "first job fits the 1-slot channel" - ); - let Err(err) = try_enqueue_job(&tx, job()) else { - anyhow::bail!("full channel must refuse the job"); - }; + fn parse_or_null_parses_and_tolerates_garbage() { assert_eq!( - err.into_response().status(), - StatusCode::SERVICE_UNAVAILABLE + parse_or_null(Some(r#"{"mode":"link-by-email"}"#)), + Some(serde_json::json!({"mode": "link-by-email"})) ); - Ok(()) + assert_eq!(parse_or_null(Some("")), None); + assert_eq!(parse_or_null(Some("not-json")), None); + assert_eq!(parse_or_null(None), None); } } diff --git a/src/backend/services/identity-resolution/src/domain/seed_service.rs b/src/backend/services/identity-resolution/src/domain/seed_service.rs index 0a21c6c5f..b2d89d25d 100644 --- a/src/backend/services/identity-resolution/src/domain/seed_service.rs +++ b/src/backend/services/identity-resolution/src/domain/seed_service.rs @@ -78,28 +78,27 @@ pub struct SeedSummary { pub known_binding_conflicts: usize, } -/// Run one persons-seed: read the input stream, fold to per-account profiles, -/// group by email, resolve each group to a `person_id`, build the observation -/// rows, and apply them (append + rebuild caches). `mint` is injected so tests -/// are deterministic. +/// Run one persons-seed over an already-read input: fold to per-account +/// profiles, group by email, resolve each group to a `person_id`, build the +/// observation rows, and apply them (append + rebuild caches). The caller +/// (`crate::seed_runner`) reads the input itself so its guards can inspect +/// the rows between the `ClickHouse` read and the MariaDB writes. `mint` is +/// injected so tests are deterministic. /// /// # Errors /// -/// Propagates reader / store errors. -pub async fn run_seed( - reader: &R, +/// Propagates store errors. +pub async fn seed_from_rows( + rows: Vec, store: &S, tenant_id: Uuid, author_person_id: Uuid, mint: impl FnMut() -> Uuid, ) -> anyhow::Result where - R: IdentityInputsReader + ?Sized, S: SeedStore + ?Sized, { // 1. Build per-account profiles from the (latest-first) input stream. - let rows = reader.stream(tenant_id).await?; - tracing::info!(input_rows = rows.len(), "persons-seed: input streamed"); let profiles = build_profiles(rows); let accounts_read = profiles.len(); @@ -149,16 +148,6 @@ mod tests { use super::*; use sea_orm::prelude::DateTime; - struct FakeReader { - rows: Vec, - } - #[async_trait] - impl IdentityInputsReader for FakeReader { - async fn stream(&self, _tenant: Uuid) -> anyhow::Result> { - Ok(self.rows.clone()) - } - } - struct FakeStore { known: HashMap, emails: HashMap, @@ -212,24 +201,22 @@ mod tests { } #[tokio::test] - async fn run_seed_wires_pipeline_end_to_end() -> anyhow::Result<()> { + async fn seed_from_rows_wires_pipeline_end_to_end() -> anyhow::Result<()> { let t: DateTime = "2026-01-01T00:00:00".parse()?; // Anna across two sources (shared email) + Boris; empty store → all mint. - let reader = FakeReader { - rows: vec![ - input("bamboohr", "5001", "email", "anna@corp.com", t), - input("bamboohr", "5001", "display_name", "Anna P", t), - input("slack", "U777", "email", "anna@corp.com", t), - input("bamboohr", "5000", "email", "boris@corp.com", t), - ], - }; + let rows = vec![ + input("bamboohr", "5001", "email", "anna@corp.com", t), + input("bamboohr", "5001", "display_name", "Anna P", t), + input("slack", "U777", "email", "anna@corp.com", t), + input("bamboohr", "5000", "email", "boris@corp.com", t), + ]; let store = FakeStore { known: HashMap::new(), emails: HashMap::new(), }; - let summary = run_seed( - &reader, + let summary = seed_from_rows( + rows, &store, Uuid::from_u128(9), Uuid::from_u128(99), @@ -250,11 +237,9 @@ mod tests { } #[tokio::test] - async fn run_seed_reuses_known_binding() -> anyhow::Result<()> { + async fn seed_from_rows_reuses_known_binding() -> anyhow::Result<()> { let t: DateTime = "2026-01-01T00:00:00".parse()?; - let reader = FakeReader { - rows: vec![input("bamboohr", "5000", "email", "boris@corp.com", t)], - }; + let rows = vec![input("bamboohr", "5000", "email", "boris@corp.com", t)]; let mut known = HashMap::new(); known.insert( SourceAccountKey { @@ -269,8 +254,8 @@ mod tests { emails: HashMap::new(), }; - let summary = run_seed( - &reader, + let summary = seed_from_rows( + rows, &store, Uuid::from_u128(9), Uuid::from_u128(99), diff --git a/src/backend/services/identity-resolution/src/gear.rs b/src/backend/services/identity-resolution/src/gear.rs index da8a95859..a6e5aff7c 100644 --- a/src/backend/services/identity-resolution/src/gear.rs +++ b/src/backend/services/identity-resolution/src/gear.rs @@ -2,8 +2,9 @@ //! //! Runs on the `api-gateway` system gear (the REST host) under //! `toolkit::bootstrap::run_server`. [`IdentityResolutionGear::init`] builds the -//! runtime (MariaDB pool + persons-seed worker); [`register_rest`] mounts the -//! profile-read and persons-seed routes on the host router. +//! runtime (MariaDB pool); [`register_rest`] mounts the profile-read and +//! persons-seed-journal routes on the host router. The seed itself runs via +//! the `seed` CLI subcommand ([`run_seed`]), not in this process. //! //! [`register_rest`]: IdentityResolutionGear::register_rest @@ -32,23 +33,12 @@ impl Gear for IdentityResolutionGear { tracing::info!("starting identity-resolution gear"); // Self-managed MariaDB pool (same approach as the analytics gear). + // No background workers: the persons-seed runs as the `seed` CLI + // subcommand (CronJob / manual Job — see `crate::seed_runner`), so the + // server process only serves reads + the operations journal. let db = crate::infra::db::connect(&config.database_url).await?; - // Persons-seed background worker: drains a job queue and runs each seed. - // A single spawned task (like the analytics validators) owns the queue. - // Capacity matches the .NET `PersonsSeedQueue` bound (100). - let (seed_tx, seed_rx) = tokio::sync::mpsc::channel(100); - let worker_db = db.clone(); - let worker_config = config.clone(); - tokio::spawn(async move { - crate::api::seed::run_worker(seed_rx, worker_db, worker_config).await; - }); - - let state = AppState { - db, - config, - seed_tx, - }; + let state = AppState { db, config }; self.state .set(Arc::new(state)) .map_err(|_| anyhow::anyhow!("{} gear already initialized", Self::MODULE_NAME))?; @@ -94,6 +84,29 @@ pub async fn run_migrate(app: &toolkit::bootstrap::AppConfig) -> anyhow::Result< Ok(()) } +/// `seed` subcommand: run one persons-seed via [`crate::seed_runner`] and +/// exit. Same out-of-lifecycle config extraction as `migrate`. +/// +/// # Errors +/// +/// [`crate::seed_runner::SeedRunError`] — the caller maps each variant to a +/// distinct process exit code. +pub async fn run_seed( + app: &toolkit::bootstrap::AppConfig, + mode: &str, + force: bool, +) -> Result<(), crate::seed_runner::SeedRunError> { + let cfg = extract_gear_config(app).map_err(crate::seed_runner::SeedRunError::Failed)?; + if cfg.database_url.is_empty() { + return Err(crate::seed_runner::SeedRunError::Failed(anyhow::anyhow!( + "`gears.identity-resolution.config.database_url` is required for seed" + ))); + } + let summary = crate::seed_runner::run(&cfg, mode, force).await?; + tracing::info!(?summary, "persons-seed run finished"); + Ok(()) +} + impl RestApiCapability for IdentityResolutionGear { fn register_rest( &self, diff --git a/src/backend/services/identity-resolution/src/infra/db/mod.rs b/src/backend/services/identity-resolution/src/infra/db/mod.rs index 523d6f56f..1c67487af 100644 --- a/src/backend/services/identity-resolution/src/infra/db/mod.rs +++ b/src/backend/services/identity-resolution/src/infra/db/mod.rs @@ -68,6 +68,82 @@ pub async fn connect_single(database_url: &str) -> anyhow::Result anyhow::Result> { + use sea_orm::{ConnectionTrait, DbBackend, Statement}; + let conn = connect_single(database_url).await?; + let acquired: Option = conn + .query_one(Statement::from_sql_and_values( + DbBackend::MySql, + "SELECT GET_LOCK(?, 0)", + [format!("{SEED_LOCK_PREFIX}{tenant_id}").into()], + )) + .await? + .map(|r| r.try_get_by_index::>(0)) + .transpose()? + .flatten(); + if acquired == Some(1) { + Ok(Some(Self { conn, tenant_id })) + } else { + Ok(None) + } + } + + /// Explicit best-effort release (the happy path — hands the lock over + /// without waiting for the session teardown). Consumes the guard; a + /// failure is not worth propagating, dropping the session releases the + /// lock anyway. + pub async fn release(self) { + use sea_orm::{ConnectionTrait, DbBackend, Statement}; + let _ = self + .conn + .execute(Statement::from_sql_and_values( + DbBackend::MySql, + "SELECT RELEASE_LOCK(?)", + [format!("{SEED_LOCK_PREFIX}{}", self.tenant_id).into()], + )) + .await; + } +} + /// Name of the cross-process advisory lock serializing schema migration runs. const MIGRATION_LOCK: &str = "identity_resolution_migrations"; /// How long a second migrator waits for the lock before giving up (seconds). diff --git a/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs b/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs index 6cccff361..f5950052e 100644 --- a/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs +++ b/src/backend/services/identity-resolution/src/infra/db/ops_repo.rs @@ -15,6 +15,10 @@ use sea_orm::prelude::DateTime; use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement}; use uuid::Uuid; +/// `operation_type` value of persons-seed runs — shared by the CLI runner +/// (writes) and the `GET /v1/persons-seed*` journal endpoints (filter). +pub const PERSONS_SEED_OP: &str = "persons-seed"; + /// Lifecycle phase of an operation. DB column is a `VARCHAR(16)`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum OperationStatus { diff --git a/src/backend/services/identity-resolution/src/infra/db/seed_repo.rs b/src/backend/services/identity-resolution/src/infra/db/seed_repo.rs index 6acf5ca01..96f80954e 100644 --- a/src/backend/services/identity-resolution/src/infra/db/seed_repo.rs +++ b/src/backend/services/identity-resolution/src/infra/db/seed_repo.rs @@ -60,6 +60,78 @@ impl SeedStore for MariaDbSeedStore<'_> { } } +/// How the `persons` log is populated relative to one tenant — input to the +/// CLI runner's wrong-tenant guard (see `seed_runner`): rows under OTHER +/// tenants with none under the configured one means the operator is about to +/// mint a parallel person universe (the HOTFIX(#1550) failure mode — the +/// unfiltered reader re-files every row under the configured tenant), not seed a fresh +/// install. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TenantPresence { + /// Whether `persons` holds any row under the given tenant. + pub has_own: bool, + /// Whether `persons` holds any row under any other tenant. + pub has_other: bool, +} + +/// Distinct tenants present in the `persons` log, capped at `cap` rows — +/// the seed runner's tenant inference only needs "zero, one, or more", so a +/// tiny cap keeps this a loose index scan on `idx_tenant_person`. +/// +/// # Errors +/// +/// Returns an error if the query fails or a stored tenant id is not 16 bytes. +pub async fn distinct_tenants(db: &DatabaseConnection, cap: u64) -> anyhow::Result> { + const SQL: &str = "SELECT DISTINCT insight_tenant_id FROM persons LIMIT ?"; + let rows = db + .query_all(Statement::from_sql_and_values( + DbBackend::MySql, + SQL, + [cap.into()], + )) + .await?; + rows.iter() + .map(|r| { + let id: Vec = r.try_get("", "insight_tenant_id")?; + Ok(Uuid::from_slice(&id)?) + }) + .collect() +} + +/// Whether `persons` holds rows under the given tenant / under any other +/// tenant. `EXISTS` probes (short-circuit on `idx_tenant_person`) rather than +/// a whole-table aggregate — the guard only needs zero-vs-non-zero, and this +/// runs ahead of every seed. +/// +/// # Errors +/// +/// Returns an error if the query fails. +pub async fn tenant_presence( + db: &DatabaseConnection, + tenant_id: Uuid, +) -> anyhow::Result { + const SQL: &str = r" + SELECT + EXISTS(SELECT 1 FROM persons WHERE insight_tenant_id = ?) AS has_own, + EXISTS(SELECT 1 FROM persons WHERE insight_tenant_id <> ?) AS has_other + "; + let row = db + .query_one(Statement::from_sql_and_values( + DbBackend::MySql, + SQL, + [ + tenant_id.as_bytes().to_vec().into(), + tenant_id.as_bytes().to_vec().into(), + ], + )) + .await? + .ok_or_else(|| anyhow::anyhow!("tenant_presence query returned no row"))?; + Ok(TenantPresence { + has_own: row.try_get::("", "has_own")? != 0, + has_other: row.try_get::("", "has_other")? != 0, + }) +} + /// Current `source_account_id → person_id` bindings for the tenant — the latest /// `value_type='id'` observation per account. Feeds the known-account branch of /// the resolver. Ported from `SqlPersonsSeed.KnownAccountBindings`. diff --git a/src/backend/services/identity-resolution/src/infra/identity_inputs.rs b/src/backend/services/identity-resolution/src/infra/identity_inputs.rs index 4fc856520..01cf6fc11 100644 --- a/src/backend/services/identity-resolution/src/infra/identity_inputs.rs +++ b/src/backend/services/identity-resolution/src/infra/identity_inputs.rs @@ -33,7 +33,10 @@ use crate::domain::seed_service::IdentityInputsReader; /// the UUID reparse, failing the seed exactly like the .NET reader's /// `Guid.Parse(GetString(...))` throw. /// -/// HOTFIX (#1550) — TEMPORARY, ported from the .NET reader (3256f707). The dbt +/// HOTFIX(#1550) — TEMPORARY, ported from the .NET reader (3256f707). This is +/// the ANCHOR of the hotfix: every piece of code whose behavior exists only +/// because of it carries the literal tag `HOTFIX(#1550)` — grep for it to +/// find the full blast radius when unwinding the hotfix. The dbt /// producer writes `insight_tenant_id` *hashed* — sipHash128 of whatever raw /// string the connector was configured with (`identity_inputs_from_history.sql`, /// documented there as a TEMPORARY cross-source join key) — so the stored tenant @@ -122,7 +125,7 @@ impl ClickHouseIdentityInputsReader { #[async_trait] impl IdentityInputsReader for ClickHouseIdentityInputsReader { async fn stream(&self, tenant_id: Uuid) -> anyhow::Result> { - // tenant_id is intentionally unused while the HOTFIX (#1550) drops the + // tenant_id is intentionally unused while the HOTFIX(#1550) drops the // tenant filter — kept so the `IdentityInputsReader` trait (and the // .NET reader tracking it) stays stable for when the filter comes back. let _ = tenant_id; diff --git a/src/backend/services/identity-resolution/src/main.rs b/src/backend/services/identity-resolution/src/main.rs index 529553105..10710037f 100644 --- a/src/backend/services/identity-resolution/src/main.rs +++ b/src/backend/services/identity-resolution/src/main.rs @@ -14,6 +14,7 @@ mod domain; mod gear; mod infra; mod migration; +mod seed_runner; // System gears — linked via inventory for the REST host and the gateway-JWT auth // pipeline. `use … as _;` is load-bearing: the gears register through `inventory` @@ -58,16 +59,69 @@ enum Commands { /// The Helm chart runs this as an initContainer before the server pod /// (same pattern as the analytics service). Migrate, + /// Run one persons-seed and exit (issue #1690). The Helm chart runs this + /// as a `CronJob`; operators run it manually via `kubectl create job + /// --from=cronjob/...`. Exit codes: 0 ok / 1 failed / 2 another run holds + /// the lock / 3 refused by an input guard. + Seed { + /// Seed mode; only `link-by-email` is implemented. + #[arg(long, default_value = seed_runner::LINK_BY_EMAIL_MODE)] + mode: String, + /// Override the input guards (empty `identity_inputs` / wrong-tenant). + #[arg(long)] + force: bool, + }, } +/// Exit codes of the `seed` subcommand, mirrored in the Job monitoring docs. +const EXIT_SEED_FAILED: i32 = 1; +const EXIT_SEED_LOCK_BUSY: i32 = 2; +const EXIT_SEED_GUARD: i32 = 3; + #[tokio::main] async fn main() -> Result<()> { let cli = Cli::parse(); // Layered config: defaults -> YAML -> env (APP__*). Logging/OTel are - // initialized by the bootstrap runtime, not here. + // initialized by the bootstrap runtime for the server path; subcommands + // run outside it and install their own plain subscriber. let config = AppConfig::load_or_default(cli.config.as_ref())?; match cli.command.unwrap_or(Commands::Run) { Commands::Run => run_server(config).await, - Commands::Migrate => gear::run_migrate(&config).await, + Commands::Migrate => { + init_subcommand_logging(); + gear::run_migrate(&config).await + } + Commands::Seed { mode, force } => { + init_subcommand_logging(); + match gear::run_seed(&config, &mode, force).await { + Ok(()) => Ok(()), + Err(seed_runner::SeedRunError::LockBusy) => { + tracing::warn!("another persons-seed run holds the lock; exiting"); + std::process::exit(EXIT_SEED_LOCK_BUSY); + } + Err(seed_runner::SeedRunError::Guard(msg)) => { + tracing::error!(%msg, "persons-seed refused by input guard"); + std::process::exit(EXIT_SEED_GUARD); + } + Err(seed_runner::SeedRunError::Failed(e)) => { + tracing::error!(error = %format!("{e:#}"), "persons-seed failed"); + std::process::exit(EXIT_SEED_FAILED); + } + } + } } } + +/// Plain stdout logging for the `migrate` / `seed` subcommands. The bootstrap +/// runtime only installs its subscriber inside `run_server`, so without this +/// every `tracing::…` on the subcommand paths is a silent no-op — and the +/// seed Job's logs are half its observability. `try_init` keeps this safe if +/// a future toolkit starts initializing earlier. +fn init_subcommand_logging() { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .try_init(); +} diff --git a/src/backend/services/identity-resolution/src/seed_runner.rs b/src/backend/services/identity-resolution/src/seed_runner.rs new file mode 100644 index 000000000..707e2ec6b --- /dev/null +++ b/src/backend/services/identity-resolution/src/seed_runner.rs @@ -0,0 +1,417 @@ +//! CLI persons-seed runner — the engine behind the `seed` subcommand +//! (issue #1690: the org tree froze because nothing re-ran the seed). +//! +//! The seed is CLI-only: a Helm `CronJob` (and manual `kubectl create job`) +//! runs `identity-resolution seed` inside the cluster — no HTTP trigger, no +//! auth. One run is: advisory lock → zombie sweep → `operations` journal row +//! → input read → guards → the same domain pipeline the removed +//! `POST /v1/persons-seed` used → journal completed/failed. +//! +//! Concurrency: runs serialize on a per-tenant MariaDB `GET_LOCK` owned by an +//! RAII guard for the whole run (see `infra::db::SeedLockGuard` — every exit +//! path releases it) — covers cron-vs-manual overlap and +//! multiple Insight instances sharing one database. A concurrent run fails +//! fast ([`SeedRunError::LockBusy`], exit code 2) instead of queueing a stale +//! re-run behind the active one. +//! +//! Guards (both overridable with `--force`, both recorded as a `failed` +//! operation so the journal explains why nothing was written): +//! * empty input — 0 `identity_inputs` rows means a broken/misconfigured +//! pipeline (wrong ClickHouse URL/database, wiped stand), not "no people"; +//! * wrong tenant — `persons` rows exist under OTHER tenants and none under +//! the configured one: seeding would mint a parallel person universe +//! under the wrong tenant (the HOTFIX(#1550) failure mode), which the append-only +//! log cannot undo. A genuinely fresh install (empty `persons`) passes. + +use std::time::Duration; + +use uuid::Uuid; + +use crate::config::GearConfig; +use crate::domain::seed_service::{IdentityInputsReader, SeedSummary, seed_from_rows}; +use crate::infra::db::{self, ops_repo, seed_repo}; +use crate::infra::identity_inputs::ClickHouseIdentityInputsReader; + +/// Author stamped on CLI-run operations and seed-minted observation rows. +/// Continues the established convention: the legacy Python seed +/// (`seed-persons-from-identity-input.py::SYSTEM_AUTHOR_UUID`) stamped +/// system-written rows with the nil UUID, so existing installs already carry +/// it. No FK constrains `author_person_id`, and the API-layer nil-caller +/// check (`api::gate`) inspects the JWT subject, not data rows. +pub const SYSTEM_AUTHOR: Uuid = Uuid::nil(); + +/// The only seed mode the pipeline implements (parity with the .NET service). +pub const LINK_BY_EMAIL_MODE: &str = "link-by-email"; + +/// Upper bound on the read + pipeline — same ceiling the removed queue +/// worker used; a hung ClickHouse/MariaDB call fails the Job (with the +/// journal row updated to `failed`) instead of wedging it past the +/// `CronJob`'s next tick (the advisory lock would hold that tick off). +const SEED_TIMEOUT: Duration = Duration::from_mins(10); + +/// Backstop over the WHOLE lock-held critical section — the zombie sweep and +/// the journal writes are MariaDB calls outside [`SEED_TIMEOUT`]'s scope, and +/// a hang in any of them would hold the advisory lock just as effectively as +/// a hung pipeline. A run cut off by THIS timeout may leave its `operations` +/// row `running`; the next run's zombie sweep reclaims it. The chart's +/// `activeDeadlineSeconds` (900s) stays the final out-of-process backstop. +const RUN_TIMEOUT: Duration = Duration::from_mins(12); + +/// How stale a `queued`/`running` operation must be before the pre-run sweep +/// reclaims it. A killed Job pod leaves its row `running` forever otherwise — +/// the in-process state is gone, only the next run can clean up. +const ZOMBIE_CUTOFF_HOURS: i64 = 1; + +/// Why a seed run did not complete — the `seed` subcommand maps each variant +/// to a distinct exit code so a `CronJob` failure is diagnosable from the Job +/// status alone (0 ok / 1 failed / 2 lock busy / 3 guard). +#[derive(Debug)] +pub enum SeedRunError { + /// Another run holds the per-tenant advisory lock. + LockBusy, + /// An input guard refused the run (message is operator-facing and is + /// persisted verbatim as the operation's `error_message`). + Guard(String), + /// The run itself failed (connect, read, pipeline, or journal write). + Failed(anyhow::Error), +} + +impl From for SeedRunError { + fn from(e: anyhow::Error) -> Self { + Self::Failed(e) + } +} + +/// Run one CLI persons-seed end to end. See the module docs for the shape. +/// +/// # Errors +/// +/// [`SeedRunError`] — lock busy, guard refusal, or a failed run. +pub async fn run( + config: &GearConfig, + mode: &str, + force: bool, +) -> Result { + if mode != LINK_BY_EMAIL_MODE { + return Err(SeedRunError::Failed(anyhow::anyhow!( + "unsupported mode '{mode}'; only '{LINK_BY_EMAIL_MODE}' is available" + ))); + } + // SINGLE-TENANT BY DESIGN, gated on HOTFIX(#1550): one run seeds exactly + // one tenant — the configured one. A true multi-tenant mode ("enumerate + // tenants, seed each") is NOT possible yet: the identity_inputs reader + // deliberately reads the WHOLE table with no tenant filter (see the + // HOTFIX(#1550) block in `infra::identity_inputs` — the dbt producer + // hashes tenant ids, so there is nothing to filter on), which means every + // tenant's run would ingest every other tenant's rows and mint duplicate + // person universes. Once the producer writes real tenant UUIDs and the + // reader filter returns, this becomes a loop over tenants — the runner is + // already per-tenant everywhere below (lock name, journal row, writes). + let db = db::connect(&config.database_url).await?; + + // Tenant resolution: the configured tenant_default_id wins; when it is + // EMPTY (existing installs whose pre-created config Secret predates the + // seed and cannot be touched right now — dev is one), fall back to the + // single tenant the persons log already holds. Inference is deliberately + // narrow: exactly ONE distinct tenant is the only unambiguous case — + // writing under the sole tenant the data already lives under is what an + // operator would configure anyway. Zero (fresh install) or several + // tenants → refuse and demand explicit config; guessing there recreates + // the HOTFIX(#1550) wrong-tenant hazard the guards exist to prevent. + let distinct = seed_repo::distinct_tenants(&db, 2).await?; + let tenant = match resolve_tenant(&config.tenant_default_id, &distinct) { + Ok(t) => t, + Err(msg) => return Err(SeedRunError::Failed(anyhow::anyhow!(msg))), + }; + if config.tenant_default_id.trim().is_empty() { + tracing::warn!( + %tenant, + "tenant_default_id is not configured — inferred the sole tenant \ + from the persons log; configure it explicitly to silence this" + ); + } + // RAII: the guard owns the lock's dedicated session — every exit path + // (return, cancellation, crash) releases the lock, see `SeedLockGuard`. + let Some(lock) = db::SeedLockGuard::try_acquire(&config.database_url, tenant).await? else { + return Err(SeedRunError::LockBusy); + }; + + let result = tokio::time::timeout(RUN_TIMEOUT, run_locked(&db, config, tenant, mode, force)) + .await + .unwrap_or_else(|_| { + Err(SeedRunError::Failed(anyhow::anyhow!( + "persons-seed run timed out after {}s inside the lock-held critical section", + RUN_TIMEOUT.as_secs() + ))) + }); + + lock.release().await; + result +} + +/// Everything that happens while the advisory lock is held: zombie sweep, +/// journal row, guarded seed, journal resolution. +async fn run_locked( + db: &sea_orm::DatabaseConnection, + config: &GearConfig, + tenant: Uuid, + mode: &str, + force: bool, +) -> Result { + // Reclaim rows a killed run left behind. Log-only failure: a broken sweep + // must not block the seed itself. + let cutoff = chrono::Utc::now().naive_utc() - chrono::Duration::hours(ZOMBIE_CUTOFF_HOURS); + match ops_repo::sweep_zombies(db, cutoff).await { + Ok(n) if n > 0 => tracing::warn!(swept = n, "persons-seed: reclaimed zombie operations"), + Ok(_) => {} + Err(e) => tracing::error!(error = %e, "persons-seed: zombie sweep failed"), + } + + // Journal row first, so every later failure (guard included) is recorded + // and visible over the GET /v1/persons-seed endpoints. + let operation_id = Uuid::now_v7(); + let request_json = + serde_json::json!({ "mode": mode, "trigger": "cli", "force": force }).to_string(); + ops_repo::enqueue( + db, + operation_id, + ops_repo::PERSONS_SEED_OP, + tenant, + SYSTEM_AUTHOR, + Some(&request_json), + ) + .await?; + ops_repo::try_start(db, operation_id).await?; + tracing::info!(%operation_id, %tenant, mode, force, "persons-seed: cli run started"); + + match guarded_seed(db, config, tenant, force).await { + Ok(summary) => { + let summary_json = serde_json::to_string(&summary).unwrap_or_else(|_| "{}".to_owned()); + ops_repo::complete(db, operation_id, &summary_json).await?; + tracing::info!(%operation_id, ?summary, "persons-seed: completed"); + Ok(summary) + } + Err(SeedRunError::Guard(msg)) => { + // Deliberate operator-facing text — safe to persist verbatim. + tracing::warn!(%operation_id, %msg, "persons-seed: refused by input guard"); + if let Err(e) = ops_repo::fail(db, operation_id, &msg).await { + tracing::error!(error = %e, %operation_id, "fail update failed"); + } + Err(SeedRunError::Guard(msg)) + } + Err(e) => { + // Persist only a generic message: `error_message` is returned + // verbatim by the GET endpoints, so raw driver/anyhow text must + // not leak to callers (same rule the queue worker followed). + if let Err(e2) = + ops_repo::fail(db, operation_id, "persons-seed failed; see job logs").await + { + tracing::error!(error = %e2, %operation_id, "fail update failed"); + } + Err(e) + } + } +} + +/// Input read → guards → pipeline, bounded by [`SEED_TIMEOUT`]. +async fn guarded_seed( + db: &sea_orm::DatabaseConnection, + config: &GearConfig, + tenant: Uuid, + force: bool, +) -> Result { + let reader = ClickHouseIdentityInputsReader::connect( + &config.clickhouse_url, + &config.clickhouse_database, + &config.clickhouse_user, + &config.clickhouse_password, + ); + let store = seed_repo::MariaDbSeedStore::new(db); + + let run = async { + let rows = reader.stream(tenant).await?; + tracing::info!(input_rows = rows.len(), "persons-seed: input streamed"); + + let presence = seed_repo::tenant_presence(db, tenant).await?; + if let Err(msg) = input_guards(rows.len(), presence, tenant, force) { + return Err(SeedRunError::Guard(msg)); + } + + seed_from_rows(rows, &store, tenant, SYSTEM_AUTHOR, Uuid::now_v7) + .await + .map_err(SeedRunError::Failed) + }; + + tokio::time::timeout(SEED_TIMEOUT, run) + .await + .unwrap_or_else(|_| { + Err(SeedRunError::Failed(anyhow::anyhow!( + "persons-seed timed out after {}s", + SEED_TIMEOUT.as_secs() + ))) + }) +} + +/// The pure tenant-resolution decision (split out for unit tests): an +/// explicitly configured tenant always wins; an empty config falls back to +/// the SOLE tenant present in the persons log; anything ambiguous refuses +/// with an operator-facing message. +fn resolve_tenant(configured: &str, distinct_in_persons: &[Uuid]) -> Result { + let configured = configured.trim(); + if !configured.is_empty() { + return Uuid::parse_str(configured).map_err(|e| format!("invalid tenant_default_id: {e}")); + } + match distinct_in_persons { + [sole] => Ok(*sole), + [] => Err( + "`gears.identity-resolution.config.tenant_default_id` is required for seed: the \ + persons log is empty, so there is no tenant to infer (fresh install — configure \ + the tenant explicitly)" + .to_owned(), + ), + _ => Err( + "`gears.identity-resolution.config.tenant_default_id` is required for seed: the \ + persons log holds several tenants, so inference is ambiguous — configure the \ + tenant explicitly" + .to_owned(), + ), + } +} + +/// The pure guard decision (see the module docs): refuse an empty input and +/// refuse a wrong-tenant run; `--force` overrides both. The returned message +/// is operator-facing — it lands verbatim in the operation's `error_message` +/// and the Job log. +fn input_guards( + input_rows: usize, + presence: seed_repo::TenantPresence, + tenant: Uuid, + force: bool, +) -> Result<(), String> { + if force { + return Ok(()); + } + if input_rows == 0 { + return Err( + "input guard: identity_inputs returned 0 rows — the ingestion pipeline looks \ + broken or misconfigured (wrong ClickHouse URL/database?); re-run with --force \ + to seed anyway" + .to_owned(), + ); + } + if !presence.has_own && presence.has_other { + return Err(format!( + "tenant guard: persons already holds rows under other tenant(s) and none under \ + the configured tenant {tenant} — seeding would mint a parallel person set under \ + a wrong tenant; fix tenant_default_id or re-run with --force" + )); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::infra::db::seed_repo::TenantPresence; + + fn tenant() -> Uuid { + Uuid::from_u128(9) + } + + fn presence(has_own: bool, has_other: bool) -> TenantPresence { + TenantPresence { has_own, has_other } + } + + #[test] + fn empty_input_refused_and_names_the_table() -> anyhow::Result<()> { + let Err(err) = input_guards(0, presence(false, false), tenant(), false) else { + anyhow::bail!("empty input must be refused"); + }; + assert!(err.contains("identity_inputs"), "{err}"); + Ok(()) + } + + #[test] + fn wrong_tenant_refused_and_names_the_tenant() -> anyhow::Result<()> { + let Err(err) = input_guards(10, presence(false, true), tenant(), false) else { + anyhow::bail!("wrong-tenant run must be refused"); + }; + assert!(err.contains("tenant"), "{err}"); + assert!(err.contains(&tenant().to_string()), "{err}"); + Ok(()) + } + + #[test] + fn force_overrides_both_guards() { + assert!(input_guards(0, presence(false, false), tenant(), true).is_ok()); + assert!(input_guards(10, presence(false, true), tenant(), true).is_ok()); + } + + #[test] + fn fresh_install_passes_unforced() { + // Non-empty input, persons entirely empty — the bootstrap shape. + assert!(input_guards(10, presence(false, false), tenant(), false).is_ok()); + } + + #[test] + fn steady_state_passes_unforced() { + assert!(input_guards(10, presence(true, true), tenant(), false).is_ok()); + } + + #[tokio::test] + async fn unsupported_mode_fails_before_any_connect() { + let cfg = crate::config::GearConfig::default(); + let err = run(&cfg, "no-such-mode", false).await; + assert!(matches!(err, Err(SeedRunError::Failed(_)))); + } + + #[tokio::test] + async fn default_config_fails_cleanly() { + let cfg = crate::config::GearConfig::default(); + let err = run(&cfg, LINK_BY_EMAIL_MODE, false).await; + assert!(matches!(err, Err(SeedRunError::Failed(_)))); + } + + #[test] + fn configured_tenant_wins_over_inference() -> anyhow::Result<()> { + let configured = tenant(); + let resolved = resolve_tenant( + &configured.to_string(), + &[Uuid::from_u128(1), Uuid::from_u128(2)], + ) + .map_err(|e| anyhow::anyhow!(e))?; + assert_eq!(resolved, configured); + Ok(()) + } + + #[test] + fn invalid_configured_tenant_is_refused() { + assert!(resolve_tenant("not-a-uuid", &[]).is_err()); + } + + #[test] + fn empty_config_infers_the_sole_tenant() -> anyhow::Result<()> { + let sole = tenant(); + let resolved = resolve_tenant(" ", &[sole]).map_err(|e| anyhow::anyhow!(e))?; + assert_eq!(resolved, sole); + Ok(()) + } + + #[test] + fn empty_config_with_no_tenants_is_refused_as_fresh_install() -> anyhow::Result<()> { + let Err(msg) = resolve_tenant("", &[]) else { + anyhow::bail!("empty persons log must not infer a tenant"); + }; + assert!(msg.contains("fresh install"), "{msg}"); + Ok(()) + } + + #[test] + fn empty_config_with_several_tenants_is_refused_as_ambiguous() -> anyhow::Result<()> { + let Err(msg) = resolve_tenant("", &[Uuid::from_u128(1), Uuid::from_u128(2)]) else { + anyhow::bail!("several tenants must not infer"); + }; + assert!(msg.contains("ambiguous"), "{msg}"); + Ok(()) + } +} diff --git a/src/ingestion/tests/e2e/identity/test_error_contracts.py b/src/ingestion/tests/e2e/identity/test_error_contracts.py index 30ac9efa3..7a5b7c5a9 100644 --- a/src/ingestion/tests/e2e/identity/test_error_contracts.py +++ b/src/ingestion/tests/e2e/identity/test_error_contracts.py @@ -10,12 +10,13 @@ .NET run stays green. Deliberately absent here: -- 503 on POST /v1/persons-seed (seed queue full): the queue capacity is a - compile-time constant (gear.rs, 100) and the refusal needs the channel - full at the instant of the POST — not deterministically inducible from a +- 503 on POST /v1/persons-seed (seed queue full, .NET-only): the queue + capacity is a compile-time constant and the refusal needs the channel full + at the instant of the POST — not deterministically inducible from a black-box test, the same reason the coverage gate excludes >=500 codes - (SERVER_FAULT_FLOOR). Pinned instead by Rust unit tests on the extracted - refusal path (identity-resolution src/api/seed.rs, `try_enqueue_job`). + (SERVER_FAULT_FLOOR). The Rust successor has no POST and no queue at all + (#1690: the seed is CLI-only; its refusal paths — input guards, advisory + lock — are exit codes, covered in test_persons_seed.py). Nothing here mutates state: the 400s fail validation before any write, the 404s target ids that don't exist, and the 401/403s never pass the gate. @@ -40,12 +41,15 @@ TOO_LONG_REASON = "x" * 501 -# ── POST /v1/persons-seed ───────────────────────────────────────────────── +# ── POST /v1/persons-seed (dotnet-only; removed in the Rust successor) ──── -def test_persons_seed_unsupported_mode_400(api) -> None: +def test_persons_seed_unsupported_mode_400(api, identity_svc) -> None: """Only 'link-by-email' exists; the refusal happens before any enqueue, - so nothing is written.""" + so nothing is written. The Rust CLI's equivalent (`--mode` validation, + exit 1) is covered by its unit tests.""" + if not identity_svc.supports_seed_http_trigger: + pytest.skip("POST /v1/persons-seed removed in the Rust successor (#1690)") r = api.post("/v1/persons-seed", json={"mode": "no-such-mode"}) assert r.status_code == 400, f"status={r.status_code} body={r.text}" problem(r) diff --git a/src/ingestion/tests/e2e/identity/test_meta_gate.py b/src/ingestion/tests/e2e/identity/test_meta_gate.py index 901e4671d..ac843ac46 100644 --- a/src/ingestion/tests/e2e/identity/test_meta_gate.py +++ b/src/ingestion/tests/e2e/identity/test_meta_gate.py @@ -180,15 +180,26 @@ def test_rust_gate_suite_skips_legacy_endpoint_dotnet_requires_it() -> None: """Implementation-aware gate universes: an unexercised legacy endpoint is a legitimate SKIP on identity-rust but a blocking MISSING on identity — so the removal never hides a .NET regression, and the Rust run doesn't - fail on an approved removal (nor count a fallback 404 as coverage).""" - spec = _spec({"/v1/persons/{email}": {"get": [200, 404]}}) - ledger = _ledger({}) # legacy endpoint never touched + fail on an approved removal (nor count a fallback 404 as coverage). + + The synthetic spec must carry EVERY op on the rust SKIP_LIST (the gate + flags a skip absent from the spec as STALE) — so the persons-seed POST + (#1690, second approved removal) is included alongside the legacy lookup. + """ + spec = _spec( + { + "/v1/persons/{email}": {"get": [200, 404]}, + "/v1/persons-seed": {"post": [200, 401, 403]}, + } + ) + ledger = _ledger({}) # removed endpoints never touched try: api_coverage.select_suite("identity-rust") api_coverage.REQUIRED_EXTRA = {} report = api_coverage.build_report(spec, ledger) assert report.passed, api_coverage.gate_violations(report) assert "GET /v1/persons/{email}" in report.skipped + assert "POST /v1/persons-seed" in report.skipped api_coverage.select_suite("identity") api_coverage.REQUIRED_EXTRA = {} diff --git a/src/ingestion/tests/e2e/identity/test_persons_seed.py b/src/ingestion/tests/e2e/identity/test_persons_seed.py index 41d977fd6..6c97fec2a 100644 --- a/src/ingestion/tests/e2e/identity/test_persons_seed.py +++ b/src/ingestion/tests/e2e/identity/test_persons_seed.py @@ -1,5 +1,4 @@ -"""Contract: the persons-seed write path — POST /v1/persons-seed + the -operation-tracking reads. +"""Contract: the persons-seed write path + the operation-tracking reads. The seed streams ClickHouse `identity.identity_inputs` and rebuilds the caller-tenant's persons / account_person_map / org_chart. It runs here under @@ -8,16 +7,22 @@ `identity.identity_inputs` table with a deterministic three-account roster: two accounts sharing an email (one person, two bindings) + one solo account. -The end-to-end case (a COMPLETED seed verified through the read path) runs -only where the implementation's ClickHouse reader works against the -harness's containerized ClickHouse — see +TRIGGER DIVERGENCE (#1690, accepted): the .NET service triggers the seed via +`POST /v1/persons-seed` (async queue + poll); the Rust successor REMOVED the +POST — the seed is CLI-only there (`identity-resolution seed`, run by the +Helm CronJob / a manual Job; synchronous) and only the GET journal routes +remain. Tests select the trigger through `_trigger_seed` and gate the +POST-specific cases on `supports_seed_http_trigger`; the CLI-specific cases +(input guards, advisory lock, exit codes) gate on `supports_seed_cli`. The +POST cases die with the .NET service. + +The end-to-end case runs only where the implementation's ClickHouse reader +works against the harness's containerized ClickHouse — see `lib.identity.supports_containerized_clickhouse`: the frozen .NET service's Octonica native-protocol handshake deadlocks against every containerized CH -tried (works against the dev cluster's), so on `dotnet` that ONE case skips; -the Rust implementation (HTTP ClickHouse client) runs it — and that is the -run that matters as cutover acceptance. The other tests in this module only -need operations to EXIST (queued/running is fine), so they run on both -implementations and keep the coverage gate green. +tried, so on `dotnet` that ONE case skips; the Rust implementation (HTTP +ClickHouse client) runs it — and that is the run that matters as cutover +acceptance. """ from __future__ import annotations @@ -37,9 +42,89 @@ SEED_SOURCE_ID = uuid.UUID("55555555-5555-5555-5555-555555555555") SHARED_EMAIL = "seeded.person@e2e.test" SOLO_EMAIL = "solo.person@e2e.test" +BOSS_EMAIL = "boss.person@e2e.test" +# A parent_email no account ever carries as its email — the seed must skip +# the edge (ADR-0010: no stub persons) and fall back to a NULL-parent +# membership row. +GHOST_EMAIL = "ghost.manager@e2e.test" + +# A tenant nothing ever seeds successfully: `persons` never has rows under it, +# so the wrong-tenant guard must always refuse an unforced run for it. +GUARD_TENANT = uuid.UUID("66666666-6666-6666-6666-666666666666") + +# Author the CLI stamps on its journal rows (no JWT on that path) — the +# SYSTEM_AUTHOR nil-UUID convention shared with the legacy Python seed. +SYSTEM_AUTHOR = "00000000-0000-0000-0000-000000000000" _OPERATION_TIMEOUT_S = 120.0 +_ROSTER: list[tuple[str, str, str]] = [ + # (account, value_type, value) — two accounts share SHARED_EMAIL. + # Connectors emit a source-native `id` observation per account; the + # profile's `ids[]` is built from exactly those. The parent_email rows + # give the org-chart rebuild something to derive edges from: the shared + # person reports to the boss; the solo person's manager is unresolvable + # (GHOST_EMAIL belongs to nobody); the boss reports to nobody. + ("seed-boss", "email", BOSS_EMAIL), + ("seed-boss", "id", "seed-boss"), + ("seed-acc-1", "email", SHARED_EMAIL), + ("seed-acc-1", "id", "seed-acc-1"), + ("seed-acc-1", "display_name", "Seeded Person"), + ("seed-acc-1", "parent_email", BOSS_EMAIL), + ("seed-acc-2", "email", SHARED_EMAIL), + ("seed-acc-2", "id", "seed-acc-2"), + ("seed-acc-3", "email", SOLO_EMAIL), + ("seed-acc-3", "id", "seed-acc-3"), + ("seed-acc-3", "display_name", "Solo Person"), + ("seed-acc-3", "parent_email", GHOST_EMAIL), +] + + +def _insert_inputs( + cfg: SessionConfig, + rows: list[tuple[str, str, str]], + version_start: int, + shift_seconds: int = 0, +) -> None: + """INSERT observation rows into identity_inputs, one distinct _synced_at + per row (production reality): the seed derives observation created_at + from it, and the persons UNIQUE key (…, value_type, created_at) silently + drops same-instant collisions — e.g. two accounts' `id` observations for + the same person. + + `shift_seconds` nudges the batch AFTER earlier same-run batches (the + manager-change rows must outrank the roster). Keep it SMALL (seconds): + the MariaDB `persons` log survives sessions on a kept stack (the session + seed wipes only reason='e2e-seed' rows), so an input stamped far in the + future outranks the NEXT session's fresh roster in the latest-observation + race until the wall clock catches up — an order-of-runs flake. + """ + values = [] + for i, (account, value_type, value) in enumerate(rows): + offset = len(rows) - i + values.append( + "(" + f"'{account}:{value_type}:{version_start + i}', " + f"'{seed.SEED_TENANT}', 'e2e-source', '{SEED_SOURCE_ID}', " + f"'{account}', '{value_type}', '{value}', " + f"'UPSERT', now64(3) + INTERVAL {shift_seconds} SECOND - INTERVAL {offset} SECOND, " + f"{version_start + i}" + ")" + ) + clickhouse.execute( + cfg, + "INSERT INTO identity.identity_inputs " # noqa: S608 — every value is a fixed test literal above, no untrusted input + "(unique_key, insight_tenant_id, insight_source_type, insight_source_id," + " source_account_id, value_type, value, operation_type, _synced_at, _version) VALUES " + + ", ".join(values), + ) + + +def _fill_roster(cfg: SessionConfig) -> None: + """TRUNCATE + INSERT the deterministic roster into identity_inputs.""" + clickhouse.execute(cfg, "TRUNCATE TABLE identity.identity_inputs") + _insert_inputs(cfg, _ROSTER, 1) + @pytest.fixture(scope="module") def identity_inputs(compose_stack: SessionConfig): @@ -64,42 +149,7 @@ def identity_inputs(compose_stack: SessionConfig): ) ENGINE = ReplacingMergeTree(_version) ORDER BY unique_key """, ) - clickhouse.execute(compose_stack, "TRUNCATE TABLE identity.identity_inputs") - - rows: list[tuple[str, str, str]] = [ - # (account, value_type, value) — two accounts share SHARED_EMAIL. - # Connectors emit a source-native `id` observation per account; the - # profile's `ids[]` is built from exactly those. - ("seed-acc-1", "email", SHARED_EMAIL), - ("seed-acc-1", "id", "seed-acc-1"), - ("seed-acc-1", "display_name", "Seeded Person"), - ("seed-acc-2", "email", SHARED_EMAIL), - ("seed-acc-2", "id", "seed-acc-2"), - ("seed-acc-3", "email", SOLO_EMAIL), - ("seed-acc-3", "id", "seed-acc-3"), - ("seed-acc-3", "display_name", "Solo Person"), - ] - values = [] - for i, (account, value_type, value) in enumerate(rows): - # Distinct _synced_at per row (production reality): the seed derives - # observation created_at from it, and the persons UNIQUE key - # (…, value_type, created_at) silently drops same-instant collisions — - # e.g. two accounts' `id` observations for the same person. - values.append( - "(" - f"'{account}:{value_type}', " - f"'{seed.SEED_TENANT}', 'e2e-source', '{SEED_SOURCE_ID}', " - f"'{account}', '{value_type}', '{value}', " - f"'UPSERT', now64(3) - INTERVAL {len(rows) - i} SECOND, {i + 1}" - ")" - ) - clickhouse.execute( - compose_stack, - "INSERT INTO identity.identity_inputs " # noqa: S608 — every value is a fixed test literal above, no untrusted input - "(unique_key, insight_tenant_id, insight_source_type, insight_source_id," - " source_account_id, value_type, value, operation_type, _synced_at, _version) VALUES " - + ", ".join(values), - ) + _fill_roster(compose_stack) return compose_stack @@ -110,13 +160,34 @@ def seed_api(identity_svc): yield c +def _trigger_seed(identity_svc, seed_api) -> str: + """Trigger one seed run through the implementation's trigger and return + its operation id. POST (async) on .NET; the `seed` CLI on Rust — + synchronous, so the returned operation is already terminal there. + + The CLI run is `--force`: the fixture dataset lives under TEST_TENANT_ID + while the seed runs under SEED_TENANT, which is exactly the wrong-tenant + shape the guard exists for — the guard's own contract is proven by the + unforced tests below. + """ + if identity_svc.supports_seed_http_trigger: + r = seed_api.post("/v1/persons-seed", json={"mode": "link-by-email"}) + assert r.status_code == 202, f"status={r.status_code} body={r.text}" + return r.json()["operation_id"] + res = identity_svc.run_seed_cli(tenant=str(seed.SEED_TENANT), force=True) + assert res.returncode == 0, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + r = seed_api.get("/v1/persons-seed?limit=1") + assert r.status_code == 200, f"status={r.status_code} body={r.text}" + rows = items_of(r.json()) + assert rows, "a completed CLI run must be visible in the journal" + return rows[0]["operation_id"] + + @pytest.fixture -def seed_operation(identity_inputs, seed_api) -> str: +def seed_operation(identity_inputs, seed_api, identity_svc) -> str: """A freshly created seed operation's id — each dependent test owns its own operation instead of leaning on another test having run first.""" - r = seed_api.post("/v1/persons-seed", json={"mode": "link-by-email"}) - assert r.status_code == 202, f"status={r.status_code} body={r.text}" - return r.json()["operation_id"] + return _trigger_seed(identity_svc, seed_api) def _wait_completed(client, operation_id: str) -> dict: @@ -133,7 +204,7 @@ def _wait_completed(client, operation_id: str) -> dict: def test_persons_seed_end_to_end(identity_inputs, seed_api, identity_svc) -> None: - """202 + Location → operation completes → the seeded person resolves, + """Seed run → operation completes → the seeded person resolves, with BOTH same-email accounts bound to one person.""" if not identity_svc.supports_containerized_clickhouse: pytest.skip( @@ -141,15 +212,18 @@ def test_persons_seed_end_to_end(identity_inputs, seed_api, identity_svc) -> Non "containerized ClickHouse (see module docstring); the Rust " "implementation runs this case" ) - r = seed_api.post("/v1/persons-seed", json={"mode": "link-by-email"}) - assert r.status_code == 202, f"status={r.status_code} body={r.text}" - operation_id = r.json()["operation_id"] - assert r.headers.get("location"), r.headers + operation_id = _trigger_seed(identity_svc, seed_api) op = _wait_completed(seed_api, operation_id) assert op["status"] == "completed", op summary = op.get("summary") or {} assert summary, op + if identity_svc.supports_seed_cli: + # CLI journal contract: system author (no JWT on that path) and the + # request records the trigger. + assert op["author_person_id"] == SYSTEM_AUTHOR, op + request = op.get("request") or {} + assert request.get("trigger") == "cli", op # Freshly minted person_ids come from the tenant-agnostic internal lookup # (no visibility gate — at this point NOBODY is in the seed admin's @@ -204,12 +278,11 @@ def test_persons_seed_operations_listed(seed_operation, seed_api) -> None: assert matching[0]["insight_tenant_id"] == str(seed.SEED_TENANT), matching[0] -def test_persons_seed_list_limit(seed_operation, seed_api) -> None: +def test_persons_seed_list_limit(seed_operation, seed_api, identity_svc) -> None: """With at least two operations present (the fixture's + one more), limit=1 returns exactly one — an empty list would mean the filter is vacuously 'passing'.""" - second = seed_api.post("/v1/persons-seed", json={"mode": "link-by-email"}) - assert second.status_code == 202, f"status={second.status_code} body={second.text}" + _trigger_seed(identity_svc, seed_api) r = seed_api.get("/v1/persons-seed?limit=1") assert r.status_code == 200, f"status={r.status_code} body={r.text}" rows = items_of(r.json()) @@ -224,11 +297,11 @@ def test_persons_seed_list_status_filter(seed_operation, seed_api) -> None: The lifecycle is one-way (queued → running → completed|failed), so the inclusion check retries until a status read and the filtered list agree (the operation may transition between the two GETs — on a fast CI worker - it can cross two states in milliseconds), and the exclusion check uses - `queued`, which the operation can never re-enter once it was observed - past it. No terminal state is required — the worker may legitimately - still be running (or, on macOS Docker Desktop, stuck — see the module - docstring).""" + it can cross two states in milliseconds; a CLI-triggered run is terminal + already), and the exclusion check uses `queued`, which the operation can + never re-enter once it was observed past it. No terminal state is + required — the .NET worker may legitimately still be running (or, on + macOS Docker Desktop, stuck — see the module docstring).""" deadline = time.monotonic() + 30.0 while True: r = seed_api.get(f"/v1/persons-seed/{seed_operation}") @@ -251,11 +324,301 @@ def test_persons_seed_list_status_filter(seed_operation, seed_api) -> None: assert seed_operation not in {op["operation_id"] for op in excluded}, excluded -def test_persons_seed_403_non_admin(bob_api) -> None: +# ── POST trigger (dotnet-only; dies with the .NET service) ──────────────── + + +def test_persons_seed_403_non_admin(bob_api, identity_svc) -> None: """bob is not an admin anywhere — the seed trigger is refused.""" + if not identity_svc.supports_seed_http_trigger: + pytest.skip("POST /v1/persons-seed removed in the Rust successor (#1690)") r = bob_api.post("/v1/persons-seed", json={"mode": "link-by-email"}) assert r.status_code == 403, f"status={r.status_code} body={r.text}" -def test_persons_seed_401_unauthenticated(anon_api) -> None: +def test_persons_seed_401_unauthenticated(anon_api, identity_svc) -> None: + if not identity_svc.supports_seed_http_trigger: + pytest.skip("POST /v1/persons-seed removed in the Rust successor (#1690)") assert anon_api.post("/v1/persons-seed", json={"mode": "link-by-email"}).status_code == 401 + + +# ── CLI trigger (rust-only): guards, lock, exit codes (#1690) ───────────── + + +def _operation_row(cfg: SessionConfig, tenant: uuid.UUID) -> dict | None: + """Newest `operations` row for a tenant, read straight from MariaDB — + guard-refused tenants have no admin, so the HTTP journal is unreadable + for them by design.""" + with seed._connection(cfg) as conn, conn.cursor() as cur: # noqa: SLF001 — harness-internal helper + cur.execute( + "SELECT status, error_message, HEX(author_person_id) AS author" + " FROM operations WHERE insight_tenant_id = %s" + " ORDER BY started_at DESC LIMIT 1", + (tenant.bytes,), + ) + row = cur.fetchone() + if row is None: + return None + if isinstance(row, dict): + return row + status, error_message, author = row + return {"status": status, "error_message": error_message, "author": author} + + +def test_seed_cli_wrong_tenant_guard(identity_inputs, identity_svc, compose_stack) -> None: + """An unforced run for a tenant `persons` has never seen — while other + tenants' rows exist — must refuse (exit 3) and journal the refusal: + seeding would mint a parallel person set under a wrong tenant + (HOTFIX(#1550): the unfiltered reader re-files every row under the + configured tenant).""" + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + res = identity_svc.run_seed_cli(tenant=str(GUARD_TENANT), force=False) + assert res.returncode == 3, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + + op = _operation_row(compose_stack, GUARD_TENANT) + assert op is not None, "the guard refusal must still write a journal row" + assert op["status"] == "failed", op + assert "tenant" in (op["error_message"] or ""), op + + +def test_seed_cli_empty_input_guard(identity_inputs, identity_svc, compose_stack) -> None: + """An unforced run over an EMPTY identity_inputs must refuse (exit 3) — + an empty read means a broken/misconfigured pipeline, not 'no people'.""" + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + clickhouse.execute(compose_stack, "TRUNCATE TABLE identity.identity_inputs") + try: + res = identity_svc.run_seed_cli(tenant=str(seed.SEED_TENANT), force=False) + assert res.returncode == 3, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + op = _operation_row(compose_stack, seed.SEED_TENANT) + assert op is not None and op["status"] == "failed", op + assert "identity_inputs" in (op["error_message"] or ""), op + finally: + # The module fixture fills once (module scope) — restore for the + # tests that run after this one. + _fill_roster(compose_stack) + + +def test_seed_cli_unconfigured_tenant_refuses_when_ambiguous( + identity_inputs, identity_svc, compose_stack +) -> None: + """With NO tenant configured the binary may only infer a tenant when the + persons log holds exactly one — the fixture dataset spans several + (TEST_TENANT_ID, OTHER_TENANT, ...), so an unconfigured run must refuse + (exit 1) instead of guessing one of them.""" + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + res = identity_svc.run_seed_cli(tenant=None, force=True) + assert res.returncode == 1, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + assert "ambiguous" in (res.stdout + res.stderr), res.stderr + + +def test_seed_cli_failure_exits_1_and_journals(identity_inputs, identity_svc, compose_stack) -> None: + """A run that fails AFTER the journal row exists (here: unreachable + ClickHouse) exits 1 and leaves a failed operation carrying only the + generic message — raw driver/anyhow text must not leak to the journal, + which the GET endpoints return verbatim.""" + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + res = identity_svc.run_seed_cli( + tenant=str(seed.SEED_TENANT), + force=True, + # Closed port → fast connection refusal on the identity_inputs read, + # which happens after the operations row is enqueued. + extra_env={"APP__gears__identity-resolution__config__clickhouse_url": "http://127.0.0.1:1"}, + ) + assert res.returncode == 1, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + + op = _operation_row(compose_stack, seed.SEED_TENANT) + assert op is not None and op["status"] == "failed", op + assert op["error_message"] == "persons-seed failed; see job logs", op + + +def test_seed_cli_sweeps_zombie_operations(identity_inputs, identity_svc, compose_stack) -> None: + """A killed Job pod leaves its operations row `running` with no process + to resolve it — the next run's pre-seed sweep must fail rows older than + the cutoff (1h) and leave fresh ones alone (they may be a live run).""" + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + stale = uuid.uuid4() + fresh = uuid.uuid4() + insert_sql = ( + "INSERT INTO operations (operation_id, operation_type, status," + " insight_tenant_id, author_person_id, started_at)" + " VALUES (%s, 'persons-seed', 'running', %s, %s," + " UTC_TIMESTAMP(6) - INTERVAL %s MINUTE)" + ) + with seed._connection(compose_stack) as conn, conn.cursor() as cur: # noqa: SLF001 — harness-internal helper + cur.execute(insert_sql, (stale.bytes, seed.SEED_TENANT.bytes, uuid.UUID(int=0).bytes, 120)) + cur.execute(insert_sql, (fresh.bytes, seed.SEED_TENANT.bytes, uuid.UUID(int=0).bytes, 5)) + try: + res = identity_svc.run_seed_cli(tenant=str(seed.SEED_TENANT), force=True) + assert res.returncode == 0, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + + with seed._connection(compose_stack) as conn, conn.cursor() as cur: # noqa: SLF001 + cur.execute( + "SELECT LOWER(HEX(operation_id)), status, error_message FROM operations" + " WHERE operation_id IN (%s, %s)", + (stale.bytes, fresh.bytes), + ) + rows = {op_id: (status, message) for op_id, status, message in cur.fetchall()} + assert rows[stale.hex] == ("failed", "aborted by pod restart"), rows + assert rows[fresh.hex][0] == "running", rows + finally: + # The synthetic rows have no process behind them — drop them so the + # fresh one can't confuse later journal assertions. + with seed._connection(compose_stack) as conn, conn.cursor() as cur: # noqa: SLF001 + cur.execute( + "DELETE FROM operations WHERE operation_id IN (%s, %s)", + (stale.bytes, fresh.bytes), + ) + + +# ── input → org_chart correspondence (#1690: the projection itself) ─────── + + +def _person_id_by_email(identity_svc, email: str) -> str: + """Resolve a seeded person's UUID via the tenant-agnostic internal + lookup (freshly minted persons are in nobody's visibility subtree).""" + with identity_svc.client( + sub=str(seed.SEED_ADMIN), tenant=str(seed.SEED_TENANT), sub_type="service", roles="service" + ) as svc: + r = svc.get(f"/internal/persons/by-email/{email}") + assert r.status_code == 200, f"status={r.status_code} body={r.text}" + return r.json()["insight_source_id"] + + +def _org_chart_edges(cfg: SessionConfig, child: str) -> list[tuple[str | None, bool]]: + """(parent_person_id, is_open) org_chart rows for a child, oldest first. + + Straight SQL on purpose: this asserts the seed's WRITE (inputs → + projection), not the read API — the profile projection filters by + `org_chart_source_type` (bamboohr in the rig config), which the + handcrafted fixture tree already covers in the read tests. + """ + with seed._connection(cfg) as conn, conn.cursor() as cur: # noqa: SLF001 — harness-internal helper + cur.execute( + "SELECT LOWER(HEX(parent_person_id)), valid_to IS NULL" + " FROM org_chart" + " WHERE insight_tenant_id = %s AND child_person_id = %s" + " ORDER BY valid_from", + (seed.SEED_TENANT.bytes, uuid.UUID(child).bytes), + ) + return [(row[0], bool(row[1])) for row in cur.fetchall()] + + +def _hex(person: str) -> str: + return uuid.UUID(person).hex + + +def test_seed_org_chart_matches_inputs(identity_inputs, identity_svc, compose_stack) -> None: + """The org_chart the seed rebuilds corresponds to the parent_email + observations in identity_inputs: a resolvable manager becomes the open + edge, an unresolvable one degrades to a NULL-parent membership row + (ADR-0010: no stub persons), and a top-of-tree person gets the Path-B + NULL-parent row.""" + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + res = identity_svc.run_seed_cli(tenant=str(seed.SEED_TENANT), force=True) + assert res.returncode == 0, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + + boss = _person_id_by_email(identity_svc, BOSS_EMAIL) + shared = _person_id_by_email(identity_svc, SHARED_EMAIL) + solo = _person_id_by_email(identity_svc, SOLO_EMAIL) + + # shared reports to boss: exactly one open edge, parent = boss. + open_edges = [p for p, is_open in _org_chart_edges(compose_stack, shared) if is_open] + assert open_edges == [_hex(boss)], open_edges + + # solo's manager is unresolvable (GHOST_EMAIL belongs to nobody) — the + # edge is skipped, membership survives as a NULL-parent row. + open_edges = [p for p, is_open in _org_chart_edges(compose_stack, solo) if is_open] + assert open_edges == [None], open_edges + + # boss reports to nobody — Path-B NULL-parent membership row. + open_edges = [p for p, is_open in _org_chart_edges(compose_stack, boss) if is_open] + assert open_edges == [None], open_edges + + +def test_seed_manager_change_reaches_org_chart(identity_inputs, identity_svc, compose_stack) -> None: + """THE #1690 regression: a manager change lands in identity_inputs → a + RE-RUN of the seed moves the open org_chart edge to the new manager and + closes the old one. Before the fix nothing re-ran the seed, so this + exact transition never reached the Team view. + + The whole cast is PER-RUN unique (uuid-suffixed accounts/emails), never + the shared roster: the MariaDB persons log outlives sessions on a kept + stack, so a re-parenting of a REUSED account would poison the next + session's roster in the latest-observation race whenever runs land + seconds apart (an order-of-runs flake we hit). A fresh child has no + cross-session history by construction. + """ + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + run_tag = uuid.uuid4().hex[:12] + child_acc = f"flip-child-{run_tag}" + child_email = f"flip.child.{run_tag}@e2e.test" + boss_a_acc = f"flip-boss-a-{run_tag}" + boss_a_email = f"flip.boss.a.{run_tag}@e2e.test" + boss_b_acc = f"flip-boss-b-{run_tag}" + boss_b_email = f"flip.boss.b.{run_tag}@e2e.test" + + # Baseline ingest: the child reports to manager A. + _insert_inputs( + compose_stack, + [ + (boss_a_acc, "email", boss_a_email), + (boss_a_acc, "id", boss_a_acc), + (child_acc, "email", child_email), + (child_acc, "id", child_acc), + (child_acc, "parent_email", boss_a_email), + ], + version_start=100, # distinct RMT _version space from the roster + shift_seconds=4, # newer than any roster row (max now-1) + ) + res = identity_svc.run_seed_cli(tenant=str(seed.SEED_TENANT), force=True) + assert res.returncode == 0, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + + child = _person_id_by_email(identity_svc, child_email) + boss_a = _person_id_by_email(identity_svc, boss_a_email) + open_edges = [p for p, is_open in _org_chart_edges(compose_stack, child) if is_open] + assert open_edges == [_hex(boss_a)], f"baseline edge must point at manager A: {open_edges}" + + # The connector ingests a manager change: a NEWER parent_email pointing + # at a new (also newly ingested) manager B. + _insert_inputs( + compose_stack, + [ + (boss_b_acc, "email", boss_b_email), + (boss_b_acc, "id", boss_b_acc), + (child_acc, "parent_email", boss_b_email), + ], + version_start=200, + shift_seconds=8, # newer than the baseline batch above + ) + res = identity_svc.run_seed_cli(tenant=str(seed.SEED_TENANT), force=True) + assert res.returncode == 0, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + + boss_b = _person_id_by_email(identity_svc, boss_b_email) + edges = _org_chart_edges(compose_stack, child) + open_edges = [p for p, is_open in edges if is_open] + assert open_edges == [_hex(boss_b)], f"open edge must move to manager B: {edges}" + # Manager A's edge is closed, not erased — SCD2 history survives. + assert (_hex(boss_a), False) in edges, edges + + +def test_seed_cli_lock_busy(identity_inputs, identity_svc, compose_stack) -> None: + """A run against a held per-tenant advisory lock fails fast with exit 2 — + the serialization that replaced the in-process queue (cron-vs-manual and + multi-instance overlap).""" + if not identity_svc.supports_seed_cli: + pytest.skip("the seed CLI exists only on the Rust implementation (#1690)") + with seed._connection(compose_stack) as conn, conn.cursor() as cur: # noqa: SLF001 — harness-internal helper + cur.execute("SELECT GET_LOCK(%s, 0)", (f"persons-seed:{seed.SEED_TENANT}",)) + got = cur.fetchone() + assert got and next(iter(got if isinstance(got, tuple) else got.values())) == 1, got + try: + res = identity_svc.run_seed_cli(tenant=str(seed.SEED_TENANT), force=True) + assert res.returncode == 2, f"rc={res.returncode}\n{res.stdout}\n{res.stderr}" + finally: + cur.execute("SELECT RELEASE_LOCK(%s)", (f"persons-seed:{seed.SEED_TENANT}",)) diff --git a/src/ingestion/tests/e2e/lib/api_coverage.py b/src/ingestion/tests/e2e/lib/api_coverage.py index 6cd4dfb7b..7c5afa4d7 100755 --- a/src/ingestion/tests/e2e/lib/api_coverage.py +++ b/src/ingestion/tests/e2e/lib/api_coverage.py @@ -140,7 +140,10 @@ "GET /v1/persons/{email}": frozenset({404}), } IDENTITY_RUST_REQUIRED_EXTRA: dict[str, frozenset[int]] = { - **_IDENTITY_COMMON_REQUIRED_EXTRA, + # POST /v1/persons-seed is dropped in the Rust successor (see the SKIP + # entry below) — its inherited requirement must go with it, or the gate + # demands codes no test can ever observe. + **{k: v for k, v in _IDENTITY_COMMON_REQUIRED_EXTRA.items() if k != "POST /v1/persons-seed"}, "POST /v1/profiles": _IDENTITY_COMMON_REQUIRED_EXTRA["POST /v1/profiles"] | {409}, "DELETE /v1/roles/{id}": _IDENTITY_COMMON_REQUIRED_EXTRA["DELETE /v1/roles/{id}"] | {409}, "DELETE /v1/person-roles/{id}": _IDENTITY_COMMON_REQUIRED_EXTRA["DELETE /v1/person-roles/{id}"] @@ -148,12 +151,19 @@ } # The Rust implementation dropped the deprecated persons lookup (approved -# removal, zero callers), but the gate universe is still the committed .NET -# spec until the Rust service publishes its own. The SKIP entry lets the -# operation be legitimately unexercised on a Rust run — while the dotnet -# suite (no such skip) still REQUIRES it, so a .NET regression can't hide. +# removal, zero callers) and the persons-seed POST trigger (#1690: the seed is +# CLI-only — CronJob / manual Job via the `seed` subcommand; the GET journal +# routes remain). The gate universe is still the committed .NET spec until the +# Rust service publishes its own, so SKIP entries let these operations be +# legitimately unexercised on a Rust run — while the dotnet suite (no such +# skips) still REQUIRES them, so a .NET regression can't hide. IDENTITY_RUST_SKIP_LIST: list[tuple[str, str]] = [ - ("GET /v1/persons/{email}", "dropped in the Rust successor (approved removal; tests skip via capabilities)") + ("GET /v1/persons/{email}", "dropped in the Rust successor (approved removal; tests skip via capabilities)"), + ( + "POST /v1/persons-seed", + "removed in the Rust successor (#1690: seed runs via the `seed` CLI " + "subcommand — CronJob/manual Job; the GET journal routes remain)", + ), ] # ── authenticator suite (src/backend/services/authenticator/tests/, run by diff --git a/src/ingestion/tests/e2e/lib/identity.py b/src/ingestion/tests/e2e/lib/identity.py index fceb52adc..06271256b 100644 --- a/src/ingestion/tests/e2e/lib/identity.py +++ b/src/ingestion/tests/e2e/lib/identity.py @@ -113,6 +113,21 @@ def supports_containerized_clickhouse(implementation: str) -> bool: return implementation == "rust" +def supports_seed_http_trigger(implementation: str) -> bool: + """Whether `POST /v1/persons-seed` exists. The Rust successor removed it + (#1690): the seed is CLI-only there — the `seed` subcommand, run by the + Helm CronJob or a manual Job; only the GET journal routes remain. The + .NET service keeps the POST until its deletion. A capability of the + EXPLICIT selection, never probed from runtime behavior.""" + return implementation == "dotnet" + + +def supports_seed_cli(implementation: str) -> bool: + """Whether the binary has the `seed` subcommand (#1690) — the CLI trigger + that replaced the POST on the Rust implementation.""" + return implementation == "rust" + + def supports_strict_input_validation(implementation: str) -> bool: """Validation the Rust port ADDED beyond the .NET behavior (reviewed on epic #1602): a too-long revoke `reason` in DELETE bodies is rejected @@ -255,6 +270,58 @@ def supports_containerized_clickhouse(self) -> bool: def supports_strict_input_validation(self) -> bool: return supports_strict_input_validation(self.implementation) + @property + def supports_seed_http_trigger(self) -> bool: + return supports_seed_http_trigger(self.implementation) + + @property + def supports_seed_cli(self) -> bool: + return supports_seed_cli(self.implementation) + + def run_seed_cli( + self, + *, + tenant: str | None, + force: bool = False, + mode: str | None = None, + timeout_s: float = 300.0, + extra_env: dict[str, str] | None = None, + ) -> subprocess.CompletedProcess[str]: + """Run `identity-resolution seed` — the CLI trigger that replaced + `POST /v1/persons-seed` (#1690). Synchronous: when it returns, the + run's `operations` row is terminal. Exit codes: 0 ok / 1 failed / + 2 lock busy / 3 input guard. + + `tenant` lands as the config `tenant_default_id` — the seed stamps + its writes and journal row with it (there is no JWT on this path). + `None` leaves the config empty, exercising the binary's tenant + inference (sole-tenant fallback / ambiguous refusal). + """ + if not self.supports_seed_cli: + raise ApiSpawnError( + f"the seed CLI exists only on the rust implementation " + f"(selected: {self.implementation})" + ) + cmd = locate_rust_app(self.cfg) + env = self._rust_env() + if tenant is not None: + env["APP__gears__identity-resolution__config__tenant_default_id"] = tenant + if extra_env: + env.update(extra_env) + args = [*cmd, "-c", str(self._rig_config_path), "seed"] + if mode is not None: + args += ["--mode", mode] + if force: + args.append("--force") + return subprocess.run( # noqa: S603 — harness-controlled argv + args, + env=env, + capture_output=True, + text=True, + timeout=timeout_s, + check=False, + ) + def start(self) -> None: create_identity_database(self.cfg) if self.implementation == "rust": diff --git a/src/ingestion/tests/e2e/lib/identity_seed.py b/src/ingestion/tests/e2e/lib/identity_seed.py index d474167d9..148f48ce7 100644 --- a/src/ingestion/tests/e2e/lib/identity_seed.py +++ b/src/ingestion/tests/e2e/lib/identity_seed.py @@ -56,7 +56,7 @@ # persons-seed runs under its OWN tenant so its tenant-scoped rebuild of # account_person_map / org_chart never touches the fixture tree above. -# (The identity_inputs read is deliberately tenant-UNfiltered — hotfix #1550 — +# (The identity_inputs read is deliberately tenant-UNfiltered — HOTFIX(#1550) — # but every WRITE binds the caller's tenant.) SEED_TENANT = uuid.UUID("44444444-4444-4444-4444-444444444444") SEED_ADMIN = uuid.UUID("dddddddd-0000-4000-8000-000000000001")