feat(backfill): discover dataset tables directly, not just the runs journal

tests/test_database.py

@@ -1029,3 +1029,65 @@ def test_list_registered_runs_returns_registered_only(db, mock_engine_factory):
         s for s in _executed_sql(conn) if "SELECT ingestor_id, table_name, task" in s
     )
     assert "WHERE registered = 1" in select
+
+
+def test_list_dataset_ingestor_ids_scans_tables_excluding_framework(
+    db, mock_engine_factory
+):
+    """Discovers dataset tables via information_schema (tables carrying an
+    ingestor_id column), skips the framework tables (run journal + salt store),
+    and returns each table's distinct non-null ingestor_id."""
+    _, _, conn = mock_engine_factory
+
+    def _result(rows):
+        r = MagicMock()
+        r.fetchall.return_value = rows
+        return r
+
+    # 1st execute: information_schema table list (includes a framework table).
+    # then one execute per NON-framework table, in returned order.
+    conn.execute.side_effect = [
+        _result([("ds_one",), ("tracebloc_ingest_runs",), ("ds_two",)]),
+        _result([("i1",), ("i2",)]),  # ds_one distinct ids
+        _result([("i3",)]),  # ds_two distinct ids
+    ]
+
+    result = db.list_dataset_ingestor_ids()
+    assert result == [
+        {"ingestor_id": "i1", "table_name": "ds_one"},
+        {"ingestor_id": "i2", "table_name": "ds_one"},
+        {"ingestor_id": "i3", "table_name": "ds_two"},
+    ]
+
+    executed = _executed_sql(conn)
+    # Discovery query hit information_schema for the ingestor_id column.
+    assert any("information_schema.columns" in s and "ingestor_id" in s for s in executed)
+    # The framework run-journal table was skipped — never SELECTed as a dataset.
+    assert not any("`tracebloc_ingest_runs`" in s for s in executed)
+    # Per-dataset-table id scans excluded null/empty ids.
+    assert any("FROM `ds_one`" in s and "IS NOT NULL" in s for s in executed)
+
+
+def test_list_dataset_ingestor_ids_skips_a_table_that_errors(db, mock_engine_factory):
+    """A single unreadable table (scan timeout / dropped mid-sweep / perms) is
+    skipped and discovery continues — the sweep's 'one bad table can't abort the
+    rollout' guarantee must hold at enumeration time too. (bugbot)"""
+    _, _, conn = mock_engine_factory
+
+    def _result(rows):
+        r = MagicMock()
+        r.fetchall.return_value = rows
+        return r
+
+    conn.execute.side_effect = [
+        _result([("ds_one",), ("ds_bad",), ("ds_two",)]),  # information_schema
+        _result([("i1",)]),  # ds_one
+        RuntimeError("scan timeout"),  # ds_bad — per-table query fails
+        _result([("i2",)]),  # ds_two still scanned
+    ]
+
+    result = db.list_dataset_ingestor_ids()
+    assert result == [
+        {"ingestor_id": "i1", "table_name": "ds_one"},
+        {"ingestor_id": "i2", "table_name": "ds_two"},
+    ]

tests/test_metadata_backfill_runner.py

@@ -147,20 +147,70 @@ def test_backfill_dataset_error_when_record_missing_category():
 # ---------------------------------------------------------------------------
 
 
-def test_backfill_datasets_enumerates_registered_runs_when_ids_omitted():
+def test_backfill_datasets_discovers_tables_when_ids_omitted():
+    # Default enumeration scans dataset tables (covers the pre-cutover backlog),
+    # NOT the runs journal.
     db = MagicMock()
-    db.list_registered_runs.return_value = [
-        {"ingestor_id": "a", "table_name": "ta", "task": "tabular_classification"},
-        {"ingestor_id": "b", "table_name": "tb", "task": "tabular_classification"},
+    db.list_dataset_ingestor_ids.return_value = [
+        {"ingestor_id": "a", "table_name": "ta"},
+        {"ingestor_id": "b", "table_name": "tb"},
     ]
-    api = _api(record=dict(_PLAIN_RECORD))
+    api = MagicMock()
+    api.get_dataset_metadata.side_effect = lambda iid: dict(
+        _PLAIN_RECORD, ingestor_id=iid, table_name={"a": "ta", "b": "tb"}[iid]
+    )
+    api.send_metadata_backfill.side_effect = lambda table, schema, meta=None: {
+        "table_name": table,
+        "created": True,
+        "competitions_refolded": 0,
+    }
     with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD):
         results = backfill_datasets(db, api)
-    db.list_registered_runs.assert_called_once()
+    db.list_dataset_ingestor_ids.assert_called_once()
+    db.list_registered_runs.assert_not_called()
     assert [r.ingestor_id for r in results] == ["a", "b"]
     assert all(r.status == STATUS_OK for r in results)
 
 
+def test_backfill_datasets_backfills_a_multi_id_table_once():
+    # A table appended by several ingest runs has multiple ingestor_ids; its
+    # metadata is table-level, so it is backfilled ONCE and the sibling id is
+    # skipped (before any backend call).
+    db = MagicMock()
+    db.list_dataset_ingestor_ids.return_value = [
+        {"ingestor_id": "run1", "table_name": "shared"},
+        {"ingestor_id": "run2", "table_name": "shared"},
+    ]
+    api = _api(record=dict(_PLAIN_RECORD, table_name="shared"))
+    with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD):
+        results = backfill_datasets(db, api)
+    assert len(results) == 1  # second id skipped by per-table dedup
+    assert results[0].table_name == "shared"
+    api.send_metadata_backfill.assert_called_once()
+
+
+def test_backfill_datasets_retries_sibling_id_when_first_not_found():
+    # If a table's first ingestor_id 404s, a sibling id is still tried (the table
+    # isn't marked done on not_found).
+    db = MagicMock()
+    db.list_dataset_ingestor_ids.return_value = [
+        {"ingestor_id": "missing", "table_name": "shared"},
+        {"ingestor_id": "good", "table_name": "shared"},
+    ]
+    api = MagicMock()
+    api.get_dataset_metadata.side_effect = lambda iid: (
+        None if iid == "missing" else dict(_PLAIN_RECORD, table_name="shared")
+    )
+    api.send_metadata_backfill.return_value = {
+        "table_name": "shared",
+        "created": True,
+        "competitions_refolded": 0,
+    }
+    with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD):
+        results = backfill_datasets(db, api)
+    assert [r.status for r in results] == [STATUS_NOT_FOUND, STATUS_OK]
+
+
 def test_backfill_datasets_continues_past_a_failing_dataset():
     api = MagicMock()
     # First GET raises (e.g. transient), second succeeds.

tracebloc_ingestor/__init__.py

@@ -28,7 +28,7 @@
 # Single source of truth for the package version. setup.py parses this literal
 # (see _read_version in setup.py) so the two can't drift again (#175). Bump here
 # only — setup.py picks it up automatically.
-__version__ = "0.7.6"
+__version__ = "0.7.7"
 
 __all__ = [
     "Config",

tracebloc_ingestor/database.py

@@ -890,6 +890,68 @@ def list_registered_runs(self) -> List[Dict[str, str]]:
             for (iid, tname, task) in rows
         ]
 
+    def list_dataset_ingestor_ids(self) -> List[Dict[str, str]]:
+        """Discover every dataset table directly and return its distinct
+        ``ingestor_id``\\ s: ``[{"ingestor_id", "table_name"}, ...]``.
+
+        This is the enumeration the metadata backfill sweeps by default, in
+        preference to :meth:`list_registered_runs`. The pre-cutover backlog the
+        backfill exists for was ingested BEFORE the run journal existed, so those
+        datasets have no journal row — but their tables still carry the framework
+        ``ingestor_id`` column. Scanning for that column finds them; the journal
+        would miss them entirely.
+
+        A dataset table is any table in this schema with an ``ingestor_id``
+        column, minus the framework bookkeeping tables (the run journal — which
+        also has the column — and the salt store). Only non-null, non-empty
+        ``ingestor_id`` values are returned; a table appended by several runs
+        yields one pair per distinct id (the caller backfills the table once).
+        """
+        framework_tables = {self.SALT_TABLE, self.RUNS_TABLE}
+        with self.engine.connect() as connection:
+            table_rows = _execute_with_retry(
+                connection,
+                text(
+                    "SELECT DISTINCT table_name FROM information_schema.columns "
+                    "WHERE table_schema = DATABASE() AND column_name = 'ingestor_id'"
+                ),
+            ).fetchall()
+
+        pairs: List[Dict[str, str]] = []
+        for (table_name,) in table_rows:
+            if table_name in framework_tables:
+                continue
+            # Identifier interpolated (bound params can't name a table); the name
+            # comes from information_schema on our own DB, and backticks are
+            # escaped, so it is not attacker-controlled free text.
+            safe_table = table_name.replace("`", "``")
+            # Per-table guard: a single unreadable table (timeout on a large
+            # unindexed scan, a table dropped mid-sweep, a permissions gap) must
+            # NOT abort discovery — otherwise every other table is skipped and the
+            # sweep's "one bad table can't stop the rollout" guarantee is lost.
+            # Log the table name + exception TYPE only (never str(exc): a driver
+            # message can embed cell values, and this feeds install logs).
+            try:
+                with self.engine.connect() as connection:
+                    id_rows = _execute_with_retry(
+                        connection,
+                        text(
+                            f"SELECT DISTINCT ingestor_id FROM `{safe_table}` "
+                            f"WHERE ingestor_id IS NOT NULL AND ingestor_id <> ''"
+                        ),
+                    ).fetchall()
+            except Exception as exc:  # noqa: BLE001 — skip the table, continue the sweep
+                logger.warning(
+                    "list_dataset_ingestor_ids: skipping table %s — could not read "
+                    "ingestor_ids (%s)",
+                    table_name,
+                    type(exc).__name__,
+                )
+                continue
+            for (ingestor_id,) in id_rows:
+                pairs.append({"ingestor_id": ingestor_id, "table_name": table_name})
+        return pairs
+
     def get_label_counts(self, table_name: str, ingestor_id: str) -> Dict[str, int]:
         """
         Return ``{label: row_count}`` for every label inserted by *ingestor_id*.

tracebloc_ingestor/metadata_backfill_runner.py

@@ -143,21 +143,46 @@ def backfill_datasets(
     *,
     skip_if_current: bool = True,
 ) -> List[BackfillResult]:
-    """Backfill every dataset in ``ingestor_ids``, or — when ``None`` — every
-    REGISTERED run in this client's ingest journal.
-
-    Per-dataset guarded: an unexpected failure on one dataset is logged and
-    recorded as an ``error`` result, and the sweep continues to the next (a
-    single bad table can't abort the rollout). Returns one
-    :class:`BackfillResult` per dataset.
+    """Backfill datasets. With an explicit ``ingestor_ids`` list, one attempt per
+    id; otherwise **discover every dataset table** in the client DB and backfill
+    each — see :meth:`Database.list_dataset_ingestor_ids`.
+
+    Table discovery (not the runs journal) is the default because the whole point
+    of the backfill is the PRE-CUTOVER backlog — datasets ingested before the
+    runs journal existed, so they have no journal entry — yet their tables still
+    carry the framework ``ingestor_id`` column. Enumerating tables therefore
+    covers both journalled and pre-journal datasets; the journal is a strict
+    subset.
+
+    A table can carry several ingestor_ids (multiple ingest runs into one table);
+    its metadata is table-level, so a table is backfilled once — the first of its
+    ids the backend resolves wins and the rest are skipped.
+
+    Per-dataset guarded: an unexpected failure on one dataset is logged (type
+    only) and recorded as an ``error`` result; the sweep continues (a single bad
+    table can't abort the rollout). Returns one :class:`BackfillResult` per
+    attempt made.
     """
-    if ingestor_ids is None:
-        ingestor_ids = [run["ingestor_id"] for run in database.list_registered_runs()]
+    if ingestor_ids is not None:
+        # Caller-supplied ids: one attempt each; the table is learned from the
+        # backend GET inside backfill_dataset.
+        items = [(ingestor_id, None) for ingestor_id in ingestor_ids]
+    else:
+        items = [
+            (row["ingestor_id"], row["table_name"])
+            for row in database.list_dataset_ingestor_ids()
+        ]
 
-    logger.info("Metadata backfill: %d dataset(s) to process.", len(ingestor_ids))
+    logger.info("Metadata backfill: %d ingestor_id(s) to process.", len(items))
 
     results: List[BackfillResult] = []
-    for ingestor_id in ingestor_ids:
+    done_tables: set = set()
+    for ingestor_id, table_name in items:
+        # A table's metadata is backfilled once. Skip the remaining ids of a
+        # table already resolved — checked by the ENUMERATED name (no backend
+        # call) so multi-run tables don't re-POST.
+        if table_name is not None and table_name in done_tables:
+            continue
         try:
             result = backfill_dataset(
                 database, api_client, ingestor_id, skip_if_current=skip_if_current
@@ -178,6 +203,15 @@ def backfill_datasets(
             )
         results.append(result)
         _log_result(result)
+        # Mark the table done only on a DEFINITIVE backend resolution (backfilled,
+        # already-current, or a competition we intentionally skip). not_found /
+        # error leave it open so a sibling ingestor_id can still resolve it.
+        if (
+            result.status
+            in (STATUS_OK, STATUS_SKIPPED_CURRENT, STATUS_SKIPPED_COMPETITION)
+            and result.table_name
+        ):
+            done_tables.add(result.table_name)
 
     _log_summary(results)
     return results