From bd9a9aba35a9a15b238e06603d93b5d0f2782dfe Mon Sep 17 00:00:00 2001 From: Jorge Rivera Date: Mon, 26 Oct 2026 10:09:31 +0100 Subject: [PATCH] feat(capture): pick the long-build specs the server has not built mint-async-build and operation-resume need a 202, which only an unbuilt spec gives, and each live record builds its two specs for good. The tool named their start years as constants, so every record needed new years by hand. Context.async_spec now walks start years 1960 to 2000 for the DAC and OECD donor questions and takes the first spec that is one definite part, sits just above the sync budget, repeats no earlier pick, and has no version on the versions route. It refuses when the range is used up. The party-totals scenarios narrow a built DAC spec and need no unbuilt one, so they keep their own question, PARTY_TOTALS_FLOW. Co-authored-by: Claude --- python/client-contract/README.md | 2 +- tools/capture_client_contract.py | 95 ++++++++++++++------- tools/tests/test_capture_client_contract.py | 72 ++++++++++++++++ 3 files changed, 136 insertions(+), 33 deletions(-) diff --git a/python/client-contract/README.md b/python/client-contract/README.md index 74a5ee3..c57dfb6 100644 --- a/python/client-contract/README.md +++ b/python/client-contract/README.md @@ -209,7 +209,7 @@ Each record names the commit in the responses' `x-dfo-commit` header, and the ru uv run tools/capture_client_contract.py record --base-url http://127.0.0.1:8083/api ``` -A record's `origin.base_url` is the base URL the run was given, a local app's included. A local app's signed URL, which its launcher serves at `/__e2e__/blob/{bucket}/{object}` on the API's origin, is written as `https://storage.googleapis.com/{bucket}/{object}` with its query redacted, as a deployment signs it, because a production download is cross-origin and the client sends no key there; the run, like the validator, refuses a record holding any absolute URL other than such a storage URL, or a `location` under the base URL, so a local app's host under another name or port is refused too. A server answers a version it has built inline, so `mint-async-build` and `operation-resume` each need a spec just above the 100,000-row sync budget that the server has not built: ODA disbursements from the 32 DAC countries to the 133 G77 countries from 1967 to 2024 for the mint, and from the 38 OECD countries to the G77 countries from 1968 to 2024 for the resumed operation, each one variable over some 3,700 donor and recipient pairs, so the done manifest stays small. A local recording therefore starts on a fresh datasets database with `DATASETS_SYNC_BUDGET_ROWS=100000`. `--scenarios-dir` writes records outside the package, unvalidated. +A record's `origin.base_url` is the base URL the run was given, a local app's included. A local app's signed URL, which its launcher serves at `/__e2e__/blob/{bucket}/{object}` on the API's origin, is written as `https://storage.googleapis.com/{bucket}/{object}` with its query redacted, as a deployment signs it, because a production download is cross-origin and the client sends no key there; the run, like the validator, refuses a record holding any absolute URL other than such a storage URL, or a `location` under the base URL, so a local app's host under another name or port is refused too. A server answers a version it has built inline, so `mint-async-build` and `operation-resume` each need a spec just above the 100,000-row sync budget that the server has not built: ODA disbursements from the 32 DAC countries to the 133 G77 countries to 2024 for the mint, and from the 38 OECD countries to the G77 countries to 2024 for the resumed operation, each one variable over some 3,700 donor and recipient pairs, so the done manifest stays small. The tool chooses each start year, the earliest whose spec is in that row band and has no version on the versions route. Each record builds both specs, so the next record takes later start years, and the run fails when no year in the tool's range is left. A local recording runs with `DATASETS_SYNC_BUDGET_ROWS=100000`. `--scenarios-dir` writes records outside the package, unvalidated. ## DatasetRef diff --git a/tools/capture_client_contract.py b/tools/capture_client_contract.py index 0b7d221..1f71bca 100644 --- a/tools/capture_client_contract.py +++ b/tools/capture_client_contract.py @@ -38,9 +38,10 @@ replaced by `X-Goog-Signature=redacted`. A local app's blob URL becomes the storage URL a deployment signs. A record holding any other absolute URL, outside a `location` under the base URL, is refused. The long builds need -two specs the server has not built, so a local recording starts on a fresh -datasets database. `--scenarios-dir` writes outside the package, -unvalidated. The run ends by validating the whole contract with +two specs the server has not built, and the run picks each from the first +start year whose spec the server has not built, so each record moves both +start years up. `--scenarios-dir` writes outside the package, unvalidated. +The run ends by validating the whole contract with `tools/validate_client_contract.py`, and fails when it reports a problem, when the binary bodies pass 5 MB, or when the records and bodies together pass `SCENARIOS_BUDGET`, 16 MB. @@ -1237,24 +1238,29 @@ def _build_body( CANDIDATES_BESIDE_ASSUMED = ( "mortality in Kenya and aid flows to Kenya from 2013 to 2022" ) -# Two specs just above the 100,000-row sync budget, one for the mint and one -# for the operation another process started: a version the server has built -# answers inline, so each long build needs a spec the server has not built. -# A flow side the question leaves unnamed reads as its source's whole total, so -# both sides name groups read as their members: about 3,700 donor and -# recipient pairs of one variable over 60 years. A variable broken down by -# sector would list thousands of variables in the done manifest. -ASYNC = ( - "ODA disbursements from the DAC countries to the G77 countries from 1967 to 2024" -) -ASYNC_RESUMED = ( - "ODA disbursements from the OECD countries to the G77 countries from 1968 to 2024" +# Two specs just above the 100,000-row sync budget, the DAC donors' for the +# mint and the OECD donors' for the operation another process started: a +# version the server has built answers inline, so each long build needs a spec +# the server has not built, and the run takes the first start year in +# LONG_BUILD_YEARS whose spec it has not. Each record builds both for good, so +# the next record takes later start years. A flow side the question leaves unnamed +# reads as its source's whole total, so both sides name groups read as their +# members: about 3,700 donor and recipient pairs of one variable over some 60 +# years. A variable broken down by sector would list thousands of variables in +# the done manifest. +LONG_BUILD = ( + "ODA disbursements from the {donors} countries to the G77 countries" + " from {year} to 2024" ) +LONG_BUILD_YEARS = range(1960, 2001) NARROW_WINDOW = (2020, 2023) REBOUND_ORIGIN = "country/FRA" -# The ASYNC spec narrowed to one recipient and year with its origin open: the -# DAC2a table lists the donor group totals beside their members, about 110 -# rows with them and 95 without. +# The DAC donors' flow to the G77, narrowed to one recipient and year with its +# origin open: the DAC2a table lists the donor group totals beside their +# members, about 110 rows with them and 95 without. +PARTY_TOTALS_FLOW = ( + "ODA disbursements from the DAC countries to the G77 countries from 1967 to 2024" +) PARTY_TOTALS_WINDOW = (2022, 2022) PARTY_TOTALS_ENTITIES = {"origin": None, "destination": ["country/KEN"]} SYNC_ROWS = 100_000 @@ -1322,22 +1328,47 @@ def definite_ref(self) -> dict[str, Any]: self._cache["ref"] = {"format_version": 1, "members": [member.ref()]} return self._cache["ref"] - def async_spec(self, question: str = ASYNC) -> Spec: - """The spec `question` answers, whose estimate is just above the sync - budget.""" - (part,) = self.definite_answer(question).parts - rows = part["spec"]["resolved"].get("download", {}).get("rows") or 0 - require( - SYNC_ROWS < rows < 2 * SYNC_ROWS, - f"{rows} rows is just above the sync budget", - ) - return _part_spec(part["spec"]) + def async_spec(self, donors: str) -> Spec: + """The spec the LONG_BUILD question for `donors` answers at the first + start year in LONG_BUILD_YEARS whose answer is one definite part, whose + estimate is just above the sync budget, and whose spec no earlier year + or other donors answered and the server has not built. No such year + raises ShapeError.""" + picks: dict[str, Spec] = self._cache.setdefault("long builds", {}) + if donors not in picks: + session = self.session("prerequisite") + seen = {spec.spec_id for spec in picks.values()} + for year in LONG_BUILD_YEARS: + answer = session.search(LONG_BUILD.format(donors=donors, year=year)) + if answer.status != "definite" or len(answer.parts) != 1: + continue + (part,) = answer.parts + rows = part["spec"]["resolved"].get("download", {}).get("rows") or 0 + spec = _part_spec(part["spec"]) + # A start year before the source's first year clips to the same + # spec as the years after it, so a spec already seen is built or + # is the other donors' pick. + if not SYNC_ROWS < rows < 2 * SYNC_ROWS or spec.spec_id in seen: + continue + seen.add(spec.spec_id) + # `_newest` and not `newest_version`, which reads a refused + # versions request as no version. + if session._newest(spec.spec_id) is None: + picks[donors] = spec + break + else: + raise ShapeError( + f"no start year from {LONG_BUILD_YEARS[0]} to" + f" {LONG_BUILD_YEARS[-1]} gives the {donors} donors a spec" + " just above the sync budget that the server has not built" + ) + return picks[donors] def running_operation(self) -> str: """The id of a long build another process started.""" if "operation" not in self._cache: self._cache["operation"] = self.session("prerequisite").mint( - self.async_spec(ASYNC_RESUMED) + self.async_spec("OECD") ) return self._cache["operation"] @@ -1428,13 +1459,13 @@ def run(s: Session, ctx: Context) -> None: def _fetch_spec_party_totals(*, keep: bool) -> Callable[[Session, Context], None]: - """Fetches the ASYNC spec narrowed to PARTY_TOTALS_ENTITIES in + """Fetches the PARTY_TOTALS_FLOW spec narrowed to PARTY_TOTALS_ENTITIES in PARTY_TOTALS_WINDOW, keeping party totals when `keep`, and checks against a fetch with the other setting that leaving them out drops rows and answers a version of its own.""" def run(s: Session, ctx: Context) -> None: - spec = ctx.definite_spec(ASYNC) + spec = ctx.definite_spec(PARTY_TOTALS_FLOW) def fetched(session: Session, party_totals: bool) -> Member: (member,) = session.fetch_spec( @@ -1509,7 +1540,7 @@ def _fetch_ref_unknown_version(s: Session, ctx: Context) -> None: def _mint_async(s: Session, ctx: Context) -> None: - s.mint(ctx.async_spec(ASYNC)) + s.mint(ctx.async_spec("DAC")) s.result() diff --git a/tools/tests/test_capture_client_contract.py b/tools/tests/test_capture_client_contract.py index 61dbb91..9954ced 100644 --- a/tools/tests/test_capture_client_contract.py +++ b/tools/tests/test_capture_client_contract.py @@ -551,3 +551,75 @@ def fake_run(args: list[str], **kwargs: Any) -> Any: assert {a for a in command if a.startswith("--extra=")} == { f"--extra={e}" for e in ("eval", "api", "engine", "datasets", "graph") } + + +def _long_build_ctx( + monkeypatch: pytest.MonkeyPatch, + answers: Mapping[tuple[str, int], tuple[str, int]], + built: set[str], +) -> tuple[Any, list[str], list[str]]: + """A context whose searches answer `answers`, (donors, year) to spec id + and row estimate, and whose versions route lists a version of each spec + in `built`, with the questions searched and the spec ids probed.""" + questions: list[str] = [] + probed: list[str] = [] + + def search(self: Any, question: str) -> Any: + questions.append(question) + for (donors, year), (spec_id, rows) in answers.items(): + if question == capture.LONG_BUILD.format(donors=donors, year=year): + spec = { + "spec_id": spec_id, + "definition": {"spec_id": spec_id}, + "resolved": {"download": {"rows": rows}}, + } + return capture.Answer({}, [{"spec": spec}], None, "definite") + return capture.Answer({}, [], None, "declined") + + def newest(self: Any, spec_id: str) -> str | None: + probed.append(spec_id) + return "cv_built" if spec_id in built else None + + monkeypatch.setattr(capture.Session, "search", search) + monkeypatch.setattr(capture.Session, "_newest", newest) + ctx = capture.Context(capture.DEV_BASE_URL, capture._offline, ACTIONS, CLIENT_CODES) + return ctx, questions, probed + + +def test_a_long_build_takes_the_first_unbuilt_spec_in_the_band( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A year outside the row band, a built spec and a spec an earlier year + answered are passed over, and each pick is searched for once.""" + in_band = capture.SYNC_ROWS + 1 + ctx, questions, probed = _long_build_ctx( + monkeypatch, + { + ("DAC", 1960): ("spec_small", capture.SYNC_ROWS), + ("DAC", 1961): ("spec_built", in_band), + ("DAC", 1962): ("spec_built", in_band), + ("DAC", 1963): ("spec_dac", in_band), + ("OECD", 1960): ("spec_dac", in_band), + ("OECD", 1961): ("spec_oecd", in_band), + }, + {"spec_built"}, + ) + assert ctx.async_spec("DAC").spec_id == "spec_dac" + assert ctx.async_spec("OECD").spec_id == "spec_oecd" + assert ctx.async_spec("DAC").spec_id == "spec_dac" + assert probed == ["spec_built", "spec_dac", "spec_oecd"] + assert len(questions) == 6 + + +def test_a_long_build_with_no_unbuilt_year_is_refused( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(capture, "LONG_BUILD_YEARS", range(1960, 1962)) + in_band = capture.SYNC_ROWS + 1 + ctx, _, _ = _long_build_ctx( + monkeypatch, + {("DAC", 1960): ("spec_a", in_band), ("DAC", 1961): ("spec_b", in_band)}, + {"spec_a", "spec_b"}, + ) + with pytest.raises(capture.ShapeError, match="from 1960 to 1961"): + ctx.async_spec("DAC")