feat(backfill): metadata backfill runner — GET dataset → recompute → send to backend (#378 send side)

setup.py

@@ -67,6 +67,10 @@ def _read_requirements(filename):
         # schema/ingest.v1.json, and dispatches to the right ingestor.
         "console_scripts": [
             "tracebloc-ingest = tracebloc_ingestor.cli.run:main",
+            # One-time pre-cutover metadata backfill (backend#1166/#1198): reads
+            # DB + backend creds from the same env as the ingestor (no
+            # ingest.yaml) and sweeps this client's registered datasets.
+            "tracebloc-backfill = tracebloc_ingestor.metadata_backfill_runner:main",
         ],
     },
     classifiers=[

tests/test_api_client_methods.py

@@ -263,3 +263,85 @@ def test_refresh_token_false_when_reauth_raises():
     client.token = "old_token"
     with patch.object(client, "authenticate", side_effect=RuntimeError("auth down")):
         assert client._refresh_token() is False
+
+
+# ---------------------------------------------------------------------------
+# get_dataset_metadata() — read-by-ingestor_id (backend#1198)
+# ---------------------------------------------------------------------------
+
+
+def test_get_dataset_metadata_success():
+    client = _client()
+    body = {"table_name": "tb", "category": "tabular_classification", "meta_data": {}}
+    with patch.object(client.session, "get", return_value=_resp(200, body)):
+        result = client.get_dataset_metadata("ing-1")
+    assert result == body
+
+
+def test_get_dataset_metadata_404_returns_none():
+    """A 404 (no dataset for this ingestor_id owned by this edge) is a skip
+    signal for the sweep, not an error — return None rather than raise."""
+    client = _client()
+    with patch.object(client.session, "get", return_value=_resp(404, text="nope")):
+        assert client.get_dataset_metadata("missing") is None
+
+
+def test_get_dataset_metadata_http_error_raises():
+    client = _client()
+    with patch.object(client.session, "get", return_value=_resp(500, text="boom")):
+        with pytest.raises(requests.exceptions.RequestException):
+            client.get_dataset_metadata("ing-1")
+
+
+def test_get_dataset_metadata_local_mode():
+    client = _client(EDGE_ENV="local")
+    with patch.object(client.session, "get") as get:
+        assert client.get_dataset_metadata("ing-1") is None
+    get.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# send_metadata_backfill() — upsert recomputed metadata (backend#1166/#1198)
+# ---------------------------------------------------------------------------
+
+
+def test_send_metadata_backfill_success():
+    client = _client()
+    body = {"table_name": "tb", "created": True, "competitions_refolded": 2}
+    with patch.object(client.session, "post", return_value=_resp(201, body)):
+        result = client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {})
+    assert result == body
+
+
+def test_send_metadata_backfill_http_error_raises():
+    client = _client()
+    with patch.object(client.session, "post", return_value=_resp(404, text="no table")):
+        with pytest.raises(requests.exceptions.RequestException):
+            client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {})
+
+
+def test_send_metadata_backfill_local_mode():
+    client = _client(EDGE_ENV="local")
+    with patch.object(client.session, "post") as post:
+        result = client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {})
+    post.assert_not_called()
+    assert result["table_name"] == "tb"
+
+
+def test_send_metadata_backfill_payload_shape():
+    import json as _json
+
+    client = _client()
+    captured = {}
+
+    def _capture(url, **kwargs):
+        captured["url"] = url
+        captured["data"] = kwargs.get("data")
+        return _resp(201, {"table_name": "tb", "created": False, "competitions_refolded": 0})
+
+    with patch.object(client.session, "post", side_effect=_capture):
+        client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {"attributes": {}})
+
+    assert captured["url"].endswith("/global_meta/metadata_backfill/tb/")
+    sent = _json.loads(captured["data"])
+    assert sent == {"schema": {"x": {"dtype": "int"}}, "meta_data": {"attributes": {}}}

tests/test_database.py

@@ -1004,3 +1004,28 @@ def test_create_table_rejects_reserved_bookkeeping_tables(reserved):
     db = Database.__new__(Database)
     with pytest.raises(ValueError, match="reserved"):
         db.create_table(reserved, {"feature_0": "FLOAT"})
+
+
+# ---------------------------------------------------------------------------
+# Run journal: enumeration for the metadata backfill sweep
+# ---------------------------------------------------------------------------
+
+
+def test_list_registered_runs_returns_registered_only(db, mock_engine_factory):
+    """The backfill sweep enumerates only REGISTERED runs, one row each, mapped
+    to {ingestor_id, table_name, task}. A NULL task (pre-task-column run) is
+    preserved as None."""
+    _, _, conn = mock_engine_factory
+    conn.execute.return_value.fetchall.return_value = [
+        ("run-a", "ta", "tabular_classification"),
+        ("run-b", "tb", None),
+    ]
+    result = db.list_registered_runs()
+    assert result == [
+        {"ingestor_id": "run-a", "table_name": "ta", "task": "tabular_classification"},
+        {"ingestor_id": "run-b", "table_name": "tb", "task": None},
+    ]
+    select = next(
+        s for s in _executed_sql(conn) if "SELECT ingestor_id, table_name, task" in s
+    )
+    assert "WHERE registered = 1" in select

tests/test_metadata_backfill_runner.py

@@ -0,0 +1,181 @@
+"""Tests for the metadata backfill runner — the GET → recompute → POST sweep
+that wires build_dataset_metadata to the backend metadata-backfill endpoint.
+
+build_dataset_metadata and the APIClient are the units-under-test's collaborators
+and are exercised elsewhere; here they are mocked so the runner's own
+orchestration/skip/guard logic is what's covered.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from tracebloc_ingestor import metadata_backfill_runner as runner
+from tracebloc_ingestor.metadata_backfill_runner import (
+    STATUS_ERROR,
+    STATUS_NOT_FOUND,
+    STATUS_OK,
+    STATUS_SKIPPED_COMPETITION,
+    STATUS_SKIPPED_CURRENT,
+    backfill_dataset,
+    backfill_datasets,
+)
+
+# A pre-cutover plain dataset record as the GET returns it — no new-shape
+# attributes yet, not a competition.
+_PLAIN_RECORD = {
+    "id": 1,
+    "table_name": "tb",
+    "category": "tabular_classification",
+    "data_format": "tabular",
+    "intent": "train",
+    "ingestor_id": "ing-1",
+    "is_competition": False,
+    "source_dataset_ids": [],
+    "schema": {"x": "INT"},
+    "meta_data": {},
+}
+
+_PAYLOAD = {
+    "schema": {"x": {"dtype": "int"}},
+    "meta_data": {"attributes": {"feature_stats": {"x": {"count": 5}}}},
+}
+
+
+def _api(record=None, send_result=None):
+    api = MagicMock()
+    api.get_dataset_metadata.return_value = record
+    api.send_metadata_backfill.return_value = send_result or {
+        "table_name": "tb",
+        "created": True,
+        "competitions_refolded": 2,
+    }
+    return api
+
+
+# ---------------------------------------------------------------------------
+# backfill_dataset — single dataset
+# ---------------------------------------------------------------------------
+
+
+def test_backfill_dataset_happy_path_gets_builds_and_posts():
+    api = _api(record=dict(_PLAIN_RECORD))
+    db = MagicMock()
+    with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD) as build:
+        result = backfill_dataset(db, api, "ing-1")
+
+    # GET by ingestor_id, build keyed on the backend's category/data_format,
+    # then POST the recomputed payload back.
+    api.get_dataset_metadata.assert_called_once_with("ing-1")
+    build.assert_called_once()
+    assert build.call_args.kwargs["category"] == "tabular_classification"
+    assert build.call_args.kwargs["data_format"] == "tabular"
+    api.send_metadata_backfill.assert_called_once_with(
+        "tb", _PAYLOAD["schema"], _PAYLOAD["meta_data"]
+    )
+    assert result.status == STATUS_OK
+    assert result.created is True
+    assert result.competitions_refolded == 2
+
+
+def test_backfill_dataset_not_found_returns_status_and_skips_build():
+    api = _api(record=None)  # GET 404 → None
+    db = MagicMock()
+    with patch.object(runner, "build_dataset_metadata") as build:
+        result = backfill_dataset(db, api, "missing")
+    assert result.status == STATUS_NOT_FOUND
+    build.assert_not_called()
+    api.send_metadata_backfill.assert_not_called()
+
+
+def test_backfill_dataset_skips_competition():
+    record = dict(_PLAIN_RECORD, is_competition=True)
+    api = _api(record=record)
+    db = MagicMock()
+    with patch.object(runner, "build_dataset_metadata") as build:
+        result = backfill_dataset(db, api, "ing-1")
+    assert result.status == STATUS_SKIPPED_COMPETITION
+    build.assert_not_called()
+    api.send_metadata_backfill.assert_not_called()
+
+
+def test_backfill_dataset_skips_merged_with_source_ids():
+    record = dict(_PLAIN_RECORD, source_dataset_ids=[7, 8])
+    api = _api(record=record)
+    with patch.object(runner, "build_dataset_metadata") as build:
+        result = backfill_dataset(MagicMock(), api, "ing-1")
+    assert result.status == STATUS_SKIPPED_COMPETITION
+    build.assert_not_called()
+
+
+def test_backfill_dataset_skips_when_already_new_shape():
+    record = dict(
+        _PLAIN_RECORD,
+        meta_data={"attributes": {"feature_stats": {"x": {"count": 1}}}},
+    )
+    api = _api(record=record)
+    with patch.object(runner, "build_dataset_metadata") as build:
+        result = backfill_dataset(MagicMock(), api, "ing-1")
+    assert result.status == STATUS_SKIPPED_CURRENT
+    build.assert_not_called()
+    api.send_metadata_backfill.assert_not_called()
+
+
+def test_backfill_dataset_reprocesses_current_when_skip_disabled():
+    record = dict(
+        _PLAIN_RECORD,
+        meta_data={"attributes": {"feature_stats": {"x": {"count": 1}}}},
+    )
+    api = _api(record=record)
+    with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD):
+        result = backfill_dataset(MagicMock(), api, "ing-1", skip_if_current=False)
+    assert result.status == STATUS_OK
+    api.send_metadata_backfill.assert_called_once()
+
+
+def test_backfill_dataset_error_when_record_missing_category():
+    record = dict(_PLAIN_RECORD, category=None)
+    api = _api(record=record)
+    with patch.object(runner, "build_dataset_metadata") as build:
+        result = backfill_dataset(MagicMock(), api, "ing-1")
+    assert result.status == STATUS_ERROR
+    build.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# backfill_datasets — the sweep
+# ---------------------------------------------------------------------------
+
+
+def test_backfill_datasets_enumerates_registered_runs_when_ids_omitted():
+    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"},
+    ]
+    api = _api(record=dict(_PLAIN_RECORD))
+    with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD):
+        results = backfill_datasets(db, api)
+    db.list_registered_runs.assert_called_once()
+    assert [r.ingestor_id for r in results] == ["a", "b"]
+    assert all(r.status == STATUS_OK for r in results)
+
+
+def test_backfill_datasets_continues_past_a_failing_dataset():
+    api = MagicMock()
+    # First GET raises (e.g. transient), second succeeds.
+    api.get_dataset_metadata.side_effect = [
+        RuntimeError("boom"),
+        dict(_PLAIN_RECORD, ingestor_id="b"),
+    ]
+    api.send_metadata_backfill.return_value = {
+        "table_name": "tb",
+        "created": False,
+        "competitions_refolded": 0,
+    }
+    with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD):
+        results = backfill_datasets(MagicMock(), api, ["a", "b"])
+    assert results[0].status == STATUS_ERROR
+    # Exception TYPE only — never the raw message (may embed customer cell values).
+    assert results[0].error == "RuntimeError"
+    assert results[1].status == STATUS_OK  # sweep continued