feat(discovery): configurable table_field for persons pipelines

README.md

@@ -361,7 +361,7 @@ A destination paused while it has never been seeded (cursor 0) skips seeding at
 
 ## CP-Driven Destination Discovery (additive)
 
-With `discovery.enabled`, viaduck polls the duckgres control plane's read-only endpoint (`GET /api/v1/warehouses`, authenticated with the scoped read-only secret via `discovery.auth_header_name` + `discovery.auth_token_env`) at startup and extends the static destination set: one destination per (warehouse, team), id `org-<org_id>-team-<team_id>`, routing value = team id, **table = the payload's `events_table` verbatim** (the CP owns naming; renames are not allowed upstream). Metadata-store passwords resolve via a direct Kubernetes Secret read with viaduck's ServiceAccount (`password_secret_ref` — RBAC into the tenant namespace, no secret copies, no plaintext payloads).
+With `discovery.enabled`, viaduck polls the duckgres control plane's read-only endpoint (`GET /api/v1/warehouses`, authenticated with the scoped read-only secret via `discovery.auth_header_name` + `discovery.auth_token_env`) at startup and extends the static destination set: one destination per (warehouse, team), id `org-<org_id>-team-<team_id>`, routing value = team id, **table = a payload table field verbatim** — `events_table` by default; a persons pipeline sets `discovery.table_field` to `persons_table` or `persons_distinct_ids_table` (the CP owns naming; renames are not allowed upstream). Metadata-store passwords resolve via a direct Kubernetes Secret read with viaduck's ServiceAccount (`password_secret_ref` — RBAC into the tenant namespace, no secret copies, no plaintext payloads).
 
 Semantics (M4): **additive, static wins, fixed set.** A static destination beats a discovered one on routing-value collision (cutover = delete the static entry). Discovered destinations initialize at the source head (`seed_mode`-independent: discovery starts the stream, never backfills) with convention defaults — schema projection on, `captured_at` dropped, `memory_limit` from `discovery.defaults`. The destination set is fixed at startup; a background poller detects drift (`viaduck_discovery_drift_destinations{kind}`) and a restart applies it. Fail-open at startup (CP unreachable → static-only + `viaduck_discovery_synced` 0); fail-safe per entry (`viaduck_discovery_broken_entries_total{reason}` — a broken tenant never takes down the rest; non-writable/resharding warehouses are skipped).
 
@@ -373,6 +373,7 @@ discovery:
   url: "http://duckgres-admin.duckgres.svc.cluster.local:8080/api/v1/warehouses"
   auth_header_name: "X-Duckgres-Internal-Secret"
   auth_token_env: "VIADUCK_DISCOVERY_TOKEN"   # the read-only secret, never the internal secret
+  table_field: "events_table"                 # persons pipeline: persons_table / persons_distinct_ids_table
   poll_interval_s: 60
   defaults:
     memory_limit: "8GB"

tests/unit/test_config.py

@@ -1043,3 +1043,65 @@ def test_destination_buffer_max_bytes_loader(tmp_path: Path):
 def test_destination_buffer_max_bytes_default_zero(config_file: Path):
     cfg = load(config_file)
     assert cfg.destinations[0].buffer_max_bytes == 0
+
+
+def test_discovery_table_field_loads_from_yaml(tmp_path):
+    import yaml
+
+    from viaduck.config import load
+
+    cfg_yaml = {
+        "source": {"name": "s", "postgres_uri_env": "SRC", "data_path": "/d", "table": "t"},
+        "routing": {"field": "company", "mode": "append_only"},
+        "destinations": [
+            {
+                "id": "d1",
+                "routing_value": "a",
+                "name": "n",
+                "postgres_uri_env": "DST",
+                "data_path": "/d",
+                "table": "t",
+            }
+        ],
+        "discovery": {"enabled": True, "url": "http://cp/api/v1/warehouses", "table_field": "persons_table"},
+    }
+    p = tmp_path / "viaduck.yaml"
+    p.write_text(yaml.safe_dump(cfg_yaml))
+    cfg = load(str(p))
+    assert cfg.discovery.table_field == "persons_table"
+
+
+def test_discovery_table_field_defaults_to_events_table(config_file):
+    from viaduck.config import load
+
+    cfg = load(config_file)
+    assert cfg.discovery.table_field == "events_table"
+
+
+def test_discovery_table_field_rejects_unknown_field(tmp_path):
+    import pytest
+    import yaml
+
+    from viaduck.config import ConfigError, load
+
+    cfg_yaml = {
+        "source": {"name": "s", "postgres_uri_env": "SRC", "data_path": "/d", "table": "t"},
+        "routing": {"field": "company", "mode": "append_only"},
+        "destinations": [
+            {
+                "id": "d1",
+                "routing_value": "a",
+                "name": "n",
+                "postgres_uri_env": "DST",
+                "data_path": "/d",
+                "table": "t",
+            }
+        ],
+        # A typo here must fail fast: silently falling back to
+        # events_table would route person rows into the events tables.
+        "discovery": {"table_field": "person_table"},
+    }
+    p = tmp_path / "viaduck.yaml"
+    p.write_text(yaml.safe_dump(cfg_yaml))
+    with pytest.raises(ConfigError, match="table_field"):
+        load(str(p))

tests/unit/test_discovery.py

@@ -69,6 +69,47 @@ def test_happy_path_events_table_verbatim(self):
         assert m.pg_endpoint == "cnpg-shard-1-rw.ducklings.svc"
         assert m.secret_namespace == "ducklings"
 
+    def test_table_field_selects_persons_table(self):
+        # A persons pipeline routes into the payload's persons_table,
+        # still verbatim — the CP owns naming for every streamed table.
+        wh = _warehouse(
+            teams=[
+                {
+                    "team_id": 7,
+                    "schema_name": "t7",
+                    "enabled": True,
+                    "events_table": "t7.events",
+                    "persons_table": "t7.persons_ab12",
+                }
+            ]
+        )
+        mapped = discovery.map_payload(_payload([wh]), table_field="persons_table")
+        assert [m.table for m in mapped] == ["t7.persons_ab12"]
+
+    def test_table_field_selects_persons_distinct_ids_table(self):
+        wh = _warehouse(
+            teams=[
+                {
+                    "team_id": 7,
+                    "schema_name": "t7",
+                    "enabled": True,
+                    "events_table": "t7.events",
+                    "persons_distinct_ids_table": "t7.persons_distinct_ids_ab12",
+                }
+            ]
+        )
+        mapped = discovery.map_payload(_payload([wh]), table_field="persons_distinct_ids_table")
+        assert [m.table for m in mapped] == ["t7.persons_distinct_ids_ab12"]
+
+    def test_table_field_missing_is_mentioned_not_poison(self):
+        # Same contract as a missing events_table: the id is nameable, so
+        # the row is degraded, never absent and never view-poisoning.
+        wh = _warehouse()
+        view = discovery.classify_payload(_payload([wh]), table_field="persons_table")
+        e = view.entries["org-acme-team-666"]
+        assert not e.startable
+        assert not view.parse_poisoned
+
     def test_disabled_team_still_included(self):
         # `enabled` is the QUERY-SERVING switch (duckgres migration
         # 000024): deriving ingestion-stop from it turns a serving hold

viaduck/config.py

@@ -586,6 +586,12 @@ class DiscoveryConfig:
     url: str | None = None
     auth_header_name: str | None = None
     auth_token_env: str | None = None
+    # Which team-payload field discovery reads as the destination table.
+    # events_table for the events pipeline; persons_table /
+    # persons_distinct_ids_table for the two streamed-persons pipelines.
+    # Validated against the known payload fields: silently defaulting a
+    # typo to events_table would route person rows into events tables.
+    table_field: str = "events_table"
     poll_interval_s: float = 60.0
     request_timeout_s: float = 10.0
     # C3 reconciler (viaduck/reconciler.py). Default OFF: the classified
@@ -636,9 +642,15 @@ class DiscoveryConfig:
     secret_cache_ttl_s: float = 300.0
     defaults: dict = field(default_factory=dict)
 
+    # Team-payload fields that name a destination table. Kept in sync
+    # with the duckgres discovery payload (resolveTeamTables).
+    TABLE_FIELDS = ("events_table", "persons_table", "persons_distinct_ids_table")
+
     def __post_init__(self):
         if self.enabled and not self.url:
             raise ConfigError("discovery.enabled requires discovery.url")
+        if self.table_field not in self.TABLE_FIELDS:
+            raise ConfigError(f"discovery.table_field must be one of {self.TABLE_FIELDS}, got {self.table_field!r}")
         if (self.auth_header_name is None) != (self.auth_token_env is None):
             raise ConfigError("discovery.auth_header_name and discovery.auth_token_env must be set together")
         if self.poll_interval_s <= 0 or self.request_timeout_s <= 0 or self.materialize_deadline_s <= 0:
@@ -1006,6 +1018,7 @@ def load(path: str | Path) -> ViaduckConfig:
             url=disc_raw.get("url"),
             auth_header_name=disc_raw.get("auth_header_name"),
             auth_token_env=disc_raw.get("auth_token_env"),
+            table_field=str(disc_raw.get("table_field", "events_table")),
             poll_interval_s=float(disc_raw.get("poll_interval_s", 60.0)),
             request_timeout_s=float(disc_raw.get("request_timeout_s", 10.0)),
             min_destinations=int(disc_raw.get("min_destinations", 1)),

viaduck/discovery.py

@@ -5,10 +5,12 @@
 secret) and maps (warehouse, team) pairs onto destination configs:
 
 - id ``org-<org_id>-team-<team_id>``; routing value = the team id.
-- **Table = the payload's ``events_table`` VERBATIM** — the CP owns
-  naming (schema-per-team + legacy bare-name overrides resolve there);
-  viaduck never derives a table name. Renames are not allowed upstream,
-  so the table is immutable for a destination's lifetime.
+- **Table = a payload table field VERBATIM** — ``events_table`` by
+  default; a persons pipeline sets ``discovery.table_field`` to
+  ``persons_table``/``persons_distinct_ids_table``. The CP owns naming
+  (schema-per-team + legacy bare-name overrides resolve there); viaduck
+  never derives a table name. Renames are not allowed upstream, so the
+  table is immutable for a destination's lifetime.
 - Metadata-store credentials come from the payload's connection fields
   plus a Kubernetes Secret reference (``password_secret_ref``) resolved
   by reading the Secret directly with viaduck's ServiceAccount (RBAC
@@ -244,18 +246,19 @@ def __post_init__(self) -> None:
         object.__setattr__(self, "entries", MappingProxyType(first))
 
 
-def classify_payload(payload: dict, *, count_broken: bool = True) -> ClassifiedView:
+def classify_payload(payload: dict, *, count_broken: bool = True, table_field: str = "events_table") -> ClassifiedView:
     """Classify every enumerable destination id in the raw payload as
     startable (mapped config) or merely mentioned (fenced, degraded —
     the reason is counted, not carried). Never raises on data problems:
     un-enumerable content poisons the view (see ClassifiedView).
     `count_broken` keeps the startup/loud vs drift-poll/quiet split of
-    map_payload."""
+    map_payload. `table_field` names the team-payload field read as the
+    destination table (see DiscoveryConfig.table_field)."""
     entries: list[ClassifiedEntry] = []
     poisoned = False
     for wh in payload.get("warehouses", []):
         try:
-            if not _classify_warehouse(wh, entries, count_broken):
+            if not _classify_warehouse(wh, entries, count_broken, table_field):
                 poisoned = True
         except Exception as e:
             _broken("malformed", f"warehouse entry unparseable: {e!r}", count=count_broken)
@@ -290,16 +293,18 @@ def derive_absent(view: ClassifiedView | None, registry_snapshot) -> frozenset[s
     return frozenset(d for d in routable_discovered if d not in view.entries)
 
 
-def map_payload(payload: dict, *, count_broken: bool = True) -> list[MappedDestination]:
+def map_payload(
+    payload: dict, *, count_broken: bool = True, table_field: str = "events_table"
+) -> list[MappedDestination]:
     """STARTABLE entries of the classified view, in payload order with
     duplicates preserved (materialize() owns dedupe + its counter).
     Startup-compatible shape; the classification is the single parsing
     path so the two can never drift."""
-    view = classify_payload(payload, count_broken=count_broken)
+    view = classify_payload(payload, count_broken=count_broken, table_field=table_field)
     return [e.mapped for e in view.entry_list if e.mapped is not None]
 
 
-def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken: bool) -> bool:
+def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken: bool, table_field: str) -> bool:
     """Classify one warehouse's teams into `entries`. Returns False when
     any content was UN-ENUMERABLE (a team id we cannot even name — that
     id would falsely read as ABSENT, so the caller poisons the view).
@@ -352,7 +357,7 @@ def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken:
 
     for team in teams:
         team_id = team.get("team_id")
-        events_table = team.get("events_table")
+        table = team.get(table_field)
         if team_id is None:
             # An id we cannot name — the poison case. Deliberate cadence
             # change vs the pre-v6 parser: this fires for unstartable
@@ -363,8 +368,8 @@ def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken:
             enumerable = False
             continue
         dest_id = f"org-{org}-team-{team_id}"
-        if startable and not events_table:
-            _broken("bad_team_row", f"org {org} team row missing events_table", count=count_broken)
+        if startable and not table:
+            _broken("bad_team_row", f"org {org} team row missing {table_field}", count=count_broken)
             entries.append(ClassifiedEntry(dest_id=dest_id, mapped=None))
             continue
         if not startable:
@@ -381,7 +386,7 @@ def _classify_warehouse(wh: dict, entries: list[ClassifiedEntry], count_broken:
                     dest_id=dest_id,
                     org_id=org,
                     team_id=team_id,
-                    table=events_table,
+                    table=table,
                     data_path=f"s3://{bucket}/",
                     pg_endpoint=ms["endpoint"],
                     pg_port=ms.get("port") or 5432,
@@ -574,6 +579,10 @@ class DriftWatcher:
     poll_interval_s: float
     baseline: dict[str, MappedDestination]
     startup_generation: int
+    # Team-payload field read as the destination table — must match the
+    # field the startup map_payload used, or every discovered entry reads
+    # as changed drift.
+    table_field: str = "events_table"
     # True when the C3 reconciler is applying views (discovery.
     # apply_enabled): the vs-STARTUP drift comparison below is then
     # permanently wrong after the first applied change (an applied add
@@ -636,7 +645,7 @@ def _poll_once(self) -> None:
         # static-only alert (round-2 review).
         metrics.discovery_config_generation.set(payload["config_generation"])
         metrics.discovery_last_success_timestamp_seconds.set_to_current_time()
-        view = classify_payload(payload, count_broken=False)
+        view = classify_payload(payload, count_broken=False, table_field=self.table_field)
         with self._view_lock:
             self._view = view
         startable = sum(1 for e in view.entries.values() if e.startable)

viaduck/main.py

@@ -748,7 +748,7 @@ def run(cfg: config.ViaduckConfig) -> None:
                         time.sleep(min(2**attempt, 5))
             if payload is None:
                 raise last_err  # type: ignore[misc]
-            mapped = disco.map_payload(payload)
+            mapped = disco.map_payload(payload, table_field=cfg.discovery.table_field)
             if len(mapped) < cfg.discovery.min_destinations:
                 raise disco.DiscoveryError(
                     f"payload mapped {len(mapped)} destination(s) < discovery.min_destinations "
@@ -821,6 +821,7 @@ def run(cfg: config.ViaduckConfig) -> None:
             baseline=baseline,
             startup_generation=generation,
             apply_mode=cfg.discovery.apply_enabled,
+            table_field=cfg.discovery.table_field,
         )
     # The registry is built from the post-merge cfg and is THE runtime
     # resolution path for destination configs — the pool, the poll cycle,